eoc-cascade: a hit reads no meter and is charged a calibrated estimate; the span opens at the neural stage #233
10 changed files with 2217 additions and 80 deletions
|
|
@ -11,6 +11,17 @@ EOC is a four-stage memoizing cascade. Every query is tried at each stage in tur
|
|||
|
||||
Joule cost is the unit of accounting. The `eoc-meter` crate reads cumulative micro-joules from whatever whole-work hardware counter is available and live (Linux RAPL, Apple Silicon IOReport; NVIDIA NVML behind the `cuda` feature is GPU telemetry, never the cascade's counter) and falls back to a `StubCounter` everywhere else — including WASM.
|
||||
|
||||
## What a resolve is charged
|
||||
|
||||
A cost is `Measured` only when every micro-joule of it was read from a live counter over the work it is booked to; anything else is `Estimated`. The cascade reads its meter where the meter can resolve the work, and nowhere else.
|
||||
|
||||
- **A resolve the neural stage answers is read.** Two reads of the meter bracket the stage, the first taken once the cache, the KV stage and the graph have missed. The span is combined with the cost the stage reported: measured when the stage estimated nothing, the larger of the two readings when the stage metered itself, an estimate when the stage estimated. The three misses are spent before the first read and are not in the figure; the meter's own closing cost is. On an M3 Ultra the misses together took 220 ns and 0.95 µJ, and a span over no work read 12 to 29 µJ, so the measured figure is not under the neural stage's energy.
|
||||
- **A resolve the cache, the KV stage or the graph answers reads no meter**, and is never measured. It is charged the stage's own figure plus what a hit on that tier costs on this host. The first time a tier answers, the cascade repeats that lookup inside one span of the meter until the span reads fifty times what a span over no work does (or 40 ms have gone), and keeps the reading per lookup, in nanojoules. Every later hit on the tier is charged that, in whole micro-joules with the remainder carried to the next hit, so the micro-joules charged over any number of hits sum to the calibrated total. `Cascade::calibration` returns the figure, the batch, the span and the empty span, or why a tier has none; `Cascade::calibrate` calibrates again; `Cascade::without_hit_calibration` switches a tier off.
|
||||
|
||||
A hit is not bracketed because a read of a counter costs more than a hit. The in-process counter on macOS takes each read only once the reading thread is seen to have left its core, about 80 µs. With a span around every resolve, 10,000 cache hits took 1,647 ms on that host, 164 µs each, and each was sealed as a measured 13 to 28 µJ: the meter's own cost, the same as a span around a neural stage that does nothing. Without it they take 4.2 ms, 0.38 µs each, and are charged the 1.2 to 1.7 µJ a cache hit was calibrated at there (2.6 to 3.4 µJ for a KV or graph hit; the calibration takes 1 to 2.4 ms, once a tier), as an estimate: read by this counter on this host, but over a batch, and the mean of a batch is not a reading of any one hit.
|
||||
|
||||
With no live counter there is no calibration and no span, and every resolve keeps the cost its stage reported.
|
||||
|
||||
## Workspace layout
|
||||
|
||||
| Crate | Purpose |
|
||||
|
|
@ -34,16 +45,30 @@ cargo run --bin eoc -- query "what is 2+2"
|
|||
cargo run --bin eoc -- meter
|
||||
```
|
||||
|
||||
Example output:
|
||||
Example output, on a host with no live counter, where the cost is the stage's own figure:
|
||||
|
||||
```
|
||||
$ cargo run --bin eoc -- query "what is 2+2"
|
||||
stage : kv
|
||||
payload: 4
|
||||
cost : 50 µJ (estimated)
|
||||
hit : the kv stage's own figure; no calibration by stub (the counter did not advance)
|
||||
receipt: <64-char hex>
|
||||
```
|
||||
|
||||
and on a Mac whose in-process counter is live (`--release`; the stage's 50 µJ plus the calibrated figure):
|
||||
|
||||
```
|
||||
$ cargo run --release --bin eoc -- query "what is 2+2"
|
||||
stage : kv
|
||||
payload: 4
|
||||
cost : 53 µJ (estimated)
|
||||
hit : a kv hit is charged 3357 nJ on this host, on top of the stage's own figure: process-energy read 6876 µJ over 2048 lookups in one span (24 µJ over none)
|
||||
receipt: <64-char hex>
|
||||
```
|
||||
|
||||
`cargo run --release -p eoc-cascade --example hit_cost` prints what each tier's hit takes and is charged on the host it runs on.
|
||||
|
||||
## CUDA / NVML
|
||||
|
||||
NVIDIA NVML support is feature-gated and **off by default** so the workspace builds without `libnvidia-ml`. NVML's total-energy counter reads the GPU alone, so `detect()` never hands it to the cascade, whose costs carry no field to say a figure covers part of the work: `NvmlCounter` is GPU telemetry, read beside the whole-work counter, and it is constructed only while its counter is live and counts no faster than the board can draw. To enable:
|
||||
|
|
|
|||
|
|
@ -2,9 +2,11 @@
|
|||
//!
|
||||
//! The cheapest stage: an LRU keyed by `QueryId` returning a previously
|
||||
//! computed `Response`. Cache hits are reported with `JouleCost::zero()`:
|
||||
//! zero micro-joules, tagged `Estimated`. The energy of a hashmap lookup is
|
||||
//! below the noise floor of any contemporary hardware energy counter, and no
|
||||
//! counter is read for a hit, so a hit never claims a measurement.
|
||||
//! zero micro-joules, tagged `Estimated`. One lookup spends about a
|
||||
//! micro-joule, less than a read of an energy counter around it would, so no
|
||||
//! counter is read for a hit and a hit never claims a measurement. What a hit
|
||||
//! costs on a host is readable as the mean of a batch in one span, and
|
||||
//! eoc-cascade charges hits that, as an estimate.
|
||||
|
||||
#![forbid(unsafe_code)]
|
||||
|
||||
|
|
|
|||
167
eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs
Normal file
167
eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs
Normal file
|
|
@ -0,0 +1,167 @@
|
|||
//! What a resolve costs in wall time, and what figure it is charged, tier by
|
||||
//! tier, on this host's in-process counter.
|
||||
//!
|
||||
//! ```text
|
||||
//! cargo run --release -p eoc-cascade --example hit_cost
|
||||
//! ```
|
||||
//!
|
||||
//! It prints; it asserts nothing. The figures follow the core the work ran on
|
||||
//! and the load on the host.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use eoc_cache::{LruCache, Stage as _};
|
||||
use eoc_cascade::Cascade;
|
||||
use eoc_core::{JouleCost, JouleSource, Query, Response, Stage};
|
||||
use eoc_graph::{GraphStage, Triple};
|
||||
use eoc_kv::{KvBackend, KvStage, MemoryKvBackend};
|
||||
use eoc_meter::detect_in_process;
|
||||
use eoc_neural::{NeuralBackend, NeuralStage};
|
||||
|
||||
/// A neural stage that does nothing and claims no figure, so a resolve it
|
||||
/// answers costs what the cascade itself adds around it.
|
||||
struct Nothing;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl NeuralBackend for Nothing {
|
||||
async fn infer(&self, q: &Query) -> Response {
|
||||
Response::new(q.id, String::new(), Stage::Neural, JouleCost::zero())
|
||||
}
|
||||
}
|
||||
|
||||
/// The stages the `eoc` CLI wires, without memoization so a KV or graph hit
|
||||
/// stays one; metered by this process's own counter where it is live, or
|
||||
/// given no meter at all.
|
||||
fn cascade(metered: bool) -> Cascade {
|
||||
let kv = Box::new(MemoryKvBackend::new());
|
||||
kv.put("ping", b"pong".to_vec());
|
||||
let graph = GraphStage::new();
|
||||
graph.extend([
|
||||
Triple::new("Paris", "capital of", "France"),
|
||||
Triple::new("Tokyo", "capital of", "Japan"),
|
||||
Triple::new("Mars", "fourth planet from", "the Sun"),
|
||||
]);
|
||||
let cascade = Cascade::new(
|
||||
Arc::new(LruCache::new(1024)),
|
||||
Arc::new(KvStage::new(kv)),
|
||||
Arc::new(graph),
|
||||
Arc::new(NeuralStage::new(Box::new(Nothing))),
|
||||
)
|
||||
.without_memoization();
|
||||
if metered { cascade.with_meter(Arc::from(detect_in_process())) } else { cascade }
|
||||
}
|
||||
|
||||
/// Resolve `q` `n` times and print the wall time of a resolve and the cost
|
||||
/// each was charged.
|
||||
async fn timed(cascade: &Cascade, what: &str, q: &Query, n: usize) {
|
||||
let mut micros = Vec::with_capacity(n);
|
||||
let (mut measured, mut sum_uj, mut stage) = (0usize, 0u64, None);
|
||||
let (mut first, mut last) = (None, None);
|
||||
let all = Instant::now();
|
||||
for _ in 0..n {
|
||||
let started = Instant::now();
|
||||
let r = cascade.resolve(q.clone()).await;
|
||||
micros.push(started.elapsed().as_nanos() as f64 / 1_000.0);
|
||||
measured += usize::from(r.joule_cost.source == JouleSource::Measured);
|
||||
sum_uj += r.joule_cost.microjoules;
|
||||
first.get_or_insert(r.joule_cost);
|
||||
last = Some(r.joule_cost);
|
||||
stage = Some(r.stage);
|
||||
}
|
||||
let all = all.elapsed();
|
||||
micros.sort_by(f64::total_cmp);
|
||||
println!(
|
||||
"{what}: {n} resolves answered by {} in {:.3} ms ({:.3} µs each at the median, {:.3} µs at the 95th \
|
||||
percentile, {:.3} µs at most); charged {sum_uj} µJ in all, {measured} of {n} tagged measured; first {}, \
|
||||
last {}",
|
||||
stage.map(|s| s.to_string()).unwrap_or_default(),
|
||||
all.as_secs_f64() * 1_000.0,
|
||||
micros[n / 2],
|
||||
micros[n * 95 / 100],
|
||||
micros[n - 1],
|
||||
first.map(|c| c.to_string()).unwrap_or_default(),
|
||||
last.map(|c| c.to_string()).unwrap_or_default(),
|
||||
);
|
||||
}
|
||||
|
||||
/// The first hit on a tier, which calibrates it, then `n` more; then the
|
||||
/// tier calibrated five times over, to show how the figure moves.
|
||||
async fn tier(cascade: &Cascade, q: &Query, n: usize) {
|
||||
let started = Instant::now();
|
||||
let first = cascade.resolve(q.clone()).await;
|
||||
let took = started.elapsed();
|
||||
let stage = first.stage;
|
||||
println!("first {stage} hit: {:.3} ms, {}; {}", took.as_secs_f64() * 1_000.0, first.joule_cost, held(cascade, stage));
|
||||
timed(cascade, &stage.to_string(), q, n).await;
|
||||
for _ in 0..5 {
|
||||
cascade.calibrate(q).await;
|
||||
println!(" {stage} calibrated again: {}", held(cascade, stage));
|
||||
}
|
||||
}
|
||||
|
||||
/// The calibration the cascade holds for `stage`, in words.
|
||||
fn held(cascade: &Cascade, stage: Stage) -> String {
|
||||
match cascade.calibration(stage) {
|
||||
Ok(c) => format!(
|
||||
"{} nJ a hit: {} lookups in one span of {} µJ, {} times the {} µJ an empty span read, in {:.3} ms",
|
||||
c.per_hit_nanojoules,
|
||||
c.batch,
|
||||
c.span_microjoules,
|
||||
c.span_microjoules / c.empty_span_microjoules.max(1),
|
||||
c.empty_span_microjoules,
|
||||
c.took.as_secs_f64() * 1_000.0
|
||||
),
|
||||
Err(why) => format!("not calibrated ({why})"),
|
||||
}
|
||||
}
|
||||
|
||||
/// What the three deterministic misses before a neural resolve spend: one
|
||||
/// span of the meter around `n` walks that every tier misses.
|
||||
async fn misses(cascade: &Cascade, n: u64) {
|
||||
let q = Query::new("nothing knows this");
|
||||
let started = Instant::now();
|
||||
let open = cascade.meter().read_microjoules();
|
||||
for _ in 0..n {
|
||||
let missed = cascade.cache().try_resolve(&q).await.is_none()
|
||||
&& cascade.kv().try_resolve(&q).await.is_none()
|
||||
&& cascade.graph().try_resolve(&q).await.is_none();
|
||||
assert!(std::hint::black_box(missed));
|
||||
}
|
||||
let close = cascade.meter().read_microjoules();
|
||||
let took = started.elapsed();
|
||||
match (open, close) {
|
||||
(Ok(open), Ok(close)) if close > open => println!(
|
||||
"the three misses: {} nJ and {:.0} ns a walk ({n} walks in one span of {} µJ)",
|
||||
(close - open) * 1_000 / n,
|
||||
took.as_nanos() as f64 / n as f64,
|
||||
close - open
|
||||
),
|
||||
_ => println!("the three misses: not read ({:.0} ns a walk)", took.as_nanos() as f64 / n as f64),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::main(flavor = "current_thread")]
|
||||
async fn main() {
|
||||
let c = cascade(true);
|
||||
println!("meter: {}", c.meter().name());
|
||||
|
||||
let cached = Query::new("seen before");
|
||||
c.cache().insert(&cached, "answer");
|
||||
tier(&c, &cached, 10_000).await;
|
||||
tier(&c, &Query::new("ping"), 10_000).await;
|
||||
tier(&c, &Query::new("Mars is the fourth planet from what?"), 10_000).await;
|
||||
|
||||
// A neural stage that does nothing: what the cascade adds around it,
|
||||
// its two reads of the meter.
|
||||
timed(&c, "neural (a stage that does nothing)", &Query::new("nothing knows this"), 2_000).await;
|
||||
misses(&c, 100_000).await;
|
||||
|
||||
// The same with no meter at all: what a resolve takes when nothing is
|
||||
// read or calibrated.
|
||||
let c = cascade(false);
|
||||
c.cache().insert(&cached, "answer");
|
||||
timed(&c, "no meter: cache", &cached, 10_000).await;
|
||||
timed(&c, "no meter: neural (a stage that does nothing)", &Query::new("nothing knows this"), 2_000).await;
|
||||
std::thread::sleep(Duration::from_millis(1));
|
||||
}
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -0,0 +1,304 @@
|
|||
//! What a hit is charged, against the hardware: the cascade on this
|
||||
//! process's own CPU energy counter, each tier's calibrated per-hit figure
|
||||
//! held against an oracle that reads the same count its own way.
|
||||
//!
|
||||
//! A hit on a deterministic tier reads no meter. It is charged a figure the
|
||||
//! cascade calibrated on this host: the counter's reading over one span
|
||||
//! around a batch of the tier's lookup, divided by the batch. Here every
|
||||
//! read the cascade takes of its counter has the oracle's own read on either
|
||||
//! side of it, so the span a calibration kept lies between two windows of
|
||||
//! the oracle's around the same batch, and the figure is held against both.
|
||||
//!
|
||||
//! One test binary with one test: the counter's domain is every thread of
|
||||
//! the process, and the oracle's check is the process's context-switch
|
||||
//! count, so nothing else may run here. No joule figure and no timing is
|
||||
//! asserted: what a lookup costs follows the core that ran it, and a loaded
|
||||
//! host stretches everything's wall time.
|
||||
|
||||
#![cfg(target_os = "macos")]
|
||||
|
||||
mod oracle;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use eoc_cache::{LruCache, Stage as _};
|
||||
use eoc_cascade::{Cascade, HitCalibration, NoCalibration};
|
||||
use eoc_core::{JouleSource, Query, Response, Stage};
|
||||
use eoc_graph::{GraphStage, Triple};
|
||||
use eoc_kv::{KvBackend, KvStage, MemoryKvBackend};
|
||||
use eoc_meter::{CounterStatus, JouleCounter, MacosProcessEnergyCounter};
|
||||
use eoc_neural::{EchoBackend, NeuralStage};
|
||||
|
||||
use oracle::{Counted, Span, Watched};
|
||||
|
||||
const TIERS: [Stage; 3] = [Stage::Cache, Stage::Kv, Stage::Graph];
|
||||
|
||||
/// A cascade whose three deterministic tiers each answer one query, metered
|
||||
/// by `meter`. It does not memoize, so a KV or graph hit stays one.
|
||||
fn cascade(meter: Arc<dyn JouleCounter>) -> Cascade {
|
||||
let kv = Box::new(MemoryKvBackend::new());
|
||||
kv.put("ping", b"pong".to_vec());
|
||||
let graph = GraphStage::new();
|
||||
graph.extend([
|
||||
Triple::new("Paris", "capital of", "France"),
|
||||
Triple::new("Tokyo", "capital of", "Japan"),
|
||||
Triple::new("Mars", "fourth planet from", "the Sun"),
|
||||
]);
|
||||
let cascade = Cascade::new(
|
||||
Arc::new(LruCache::new(1024)),
|
||||
Arc::new(KvStage::new(kv)),
|
||||
Arc::new(graph),
|
||||
Arc::new(NeuralStage::new(Box::new(EchoBackend::new()))),
|
||||
)
|
||||
.with_meter(meter)
|
||||
.without_memoization();
|
||||
cascade.cache().insert(&answered_by(Stage::Cache), "answer");
|
||||
cascade
|
||||
}
|
||||
|
||||
/// The query `stage` answers in [`cascade`].
|
||||
fn answered_by(stage: Stage) -> Query {
|
||||
Query::new(match stage {
|
||||
Stage::Cache => "seen before",
|
||||
Stage::Kv => "ping",
|
||||
Stage::Graph => "Mars is the fourth planet from what?",
|
||||
Stage::Neural => "nothing knows this",
|
||||
})
|
||||
}
|
||||
|
||||
/// What `stage` reports for a hit of its own, in micro-joules.
|
||||
fn own_cost(stage: Stage) -> u64 {
|
||||
match stage {
|
||||
Stage::Cache => 0,
|
||||
Stage::Kv => 50,
|
||||
Stage::Graph => 500,
|
||||
Stage::Neural => unreachable!("not a deterministic tier"),
|
||||
}
|
||||
}
|
||||
|
||||
/// `stage`'s own lookup of `q`, and the response minted for it: what a
|
||||
/// calibration repeats.
|
||||
async fn lookup(cascade: &Cascade, stage: Stage, q: &Query) {
|
||||
let r = match stage {
|
||||
Stage::Cache => cascade.cache().try_resolve(q).await,
|
||||
Stage::Kv => cascade.kv().try_resolve(q).await,
|
||||
Stage::Graph => cascade.graph().try_resolve(q).await,
|
||||
Stage::Neural => unreachable!("not a deterministic tier"),
|
||||
}
|
||||
.expect("the tier answers its query");
|
||||
std::hint::black_box(Response::new(r.query_id, r.payload, r.stage, r.joule_cost));
|
||||
}
|
||||
|
||||
fn in_words(c: &HitCalibration) -> String {
|
||||
format!(
|
||||
"{} nJ a hit ({} lookups in one span of {} µJ, {} times the {} µJ an empty span read, in {:?})",
|
||||
c.per_hit_nanojoules,
|
||||
c.batch,
|
||||
c.span_microjoules,
|
||||
c.span_microjoules / c.empty_span_microjoules.max(1),
|
||||
c.empty_span_microjoules,
|
||||
c.took
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_hit_is_charged_what_an_oracle_reads_around_the_same_batch() {
|
||||
const TRIES: usize = 20;
|
||||
const HITS: u64 = 10_000;
|
||||
let status = eoc_meter::process_energy_status();
|
||||
eprintln!("hit calibration: status={status:?}");
|
||||
assert_eq!(status, CounterStatus::Live, "this process's CPU energy counter is not live here");
|
||||
|
||||
// 1. Each tier calibrated again and again through a counter the oracle
|
||||
// reads on either side of. The span a calibration kept is its last
|
||||
// two reads, and it lies between the oracle's window inside it and
|
||||
// the oracle's window around it and both reads: that much holds for
|
||||
// any count that only grows, so a figure outside the two windows is
|
||||
// not this process's count over that batch. What the windows leave
|
||||
// open is their own distance apart, the two readers' reads and
|
||||
// wake-ups: the figure is pinned within a fifth when the window
|
||||
// inside is at least four fifths of the one around (it was 96% at
|
||||
// least on a quiet M3 Ultra).
|
||||
let watched = Arc::new(Watched::new(MacosProcessEnergyCounter::try_new().expect("the process counter is live here")));
|
||||
let through_oracle = cascade(watched.clone());
|
||||
let (mut not_calibrated, mut loose, mut tight) = (Vec::new(), Vec::new(), Vec::new());
|
||||
for stage in TIERS {
|
||||
let q = answered_by(stage);
|
||||
let (mut per_hit, mut short) = (Vec::new(), 0usize);
|
||||
for attempt in 0..TRIES {
|
||||
watched.take();
|
||||
let (answered, calibration) = through_oracle.calibrate(&q).await.expect("a deterministic tier answers");
|
||||
assert_eq!(answered, stage);
|
||||
let reads = watched.take();
|
||||
let c = match calibration {
|
||||
Ok(c) => c,
|
||||
Err(why) => {
|
||||
not_calibrated.push((stage, attempt, why));
|
||||
continue;
|
||||
}
|
||||
};
|
||||
assert_eq!(through_oracle.calibration(stage), Ok(c));
|
||||
|
||||
// The empty span the calibration reports is the counter's first
|
||||
// advance over no work.
|
||||
let base = reads[0].microjoules.expect("a calibration's first read");
|
||||
let advanced = reads[1..].iter().filter_map(|r| r.microjoules).find(|uj| *uj > base);
|
||||
assert_eq!(Some(c.empty_span_microjoules), advanced.map(|uj| uj - base), "{stage} #{attempt}: {c:?}");
|
||||
assert!(c.empty_span_microjoules > 0 && c.batch > 0, "{stage} #{attempt}: {c:?}");
|
||||
|
||||
// The span it kept, and the oracle's two windows around the
|
||||
// same batch.
|
||||
let span = Span::between(&reads[reads.len() - 2], &reads[reads.len() - 1]);
|
||||
assert_eq!(span.read, Some(c.span_microjoules * 1_000), "{stage} #{attempt}: {c:?} is not its last span");
|
||||
assert!(
|
||||
span.holds(c.span_microjoules * 1_000),
|
||||
"{stage} #{attempt}: a span of {} µJ outside the oracle's windows around it, {} nJ inside and {} nJ \
|
||||
around",
|
||||
c.span_microjoules,
|
||||
span.inside,
|
||||
span.around
|
||||
);
|
||||
// The per-hit figure times the batch is that span, to the half
|
||||
// nanojoule a lookup the figure is rounded by.
|
||||
let figure = c.per_hit_nanojoules * c.batch;
|
||||
assert!(figure.abs_diff(c.span_microjoules * 1_000) <= c.batch / 2 + 1, "{stage} #{attempt}: {c:?}");
|
||||
if span.tightness() < 800 {
|
||||
loose.push((stage, attempt, figure, span.inside, span.around));
|
||||
} else {
|
||||
tight.push(span.tightness());
|
||||
}
|
||||
short += usize::from(c.span_microjoules < 50 * c.empty_span_microjoules);
|
||||
per_hit.push(c.per_hit_nanojoules);
|
||||
}
|
||||
per_hit.sort_unstable();
|
||||
eprintln!(
|
||||
"hit calibration: {stage} through the oracle x {TRIES}: {} to {} nJ a hit, {} at the median; {short} \
|
||||
stopped by the time allowed short of fifty empty spans; last: {}",
|
||||
per_hit.first().copied().unwrap_or(0),
|
||||
per_hit.last().copied().unwrap_or(0),
|
||||
per_hit.get(per_hit.len() / 2).copied().unwrap_or(0),
|
||||
through_oracle.calibration(stage).map(|c| in_words(&c)).unwrap_or_else(|why| why.to_string()),
|
||||
);
|
||||
}
|
||||
eprintln!(
|
||||
"hit calibration: {} of {} figures within the oracle's two windows with the inner at least {}/1000 of the \
|
||||
outer; {} looser {loose:?} (tier, try, figure nJ, inside nJ, around nJ); {} not calibrated {not_calibrated:?}",
|
||||
tight.len(),
|
||||
3 * TRIES,
|
||||
tight.iter().min().copied().unwrap_or(0),
|
||||
loose.len(),
|
||||
not_calibrated.len()
|
||||
);
|
||||
|
||||
// 2. The cascade as it is used, on the bare counter: after a tier's
|
||||
// first hit no hit reads the meter, every hit is an estimate, and the
|
||||
// micro-joules charged over ten thousand hits are the calibrated
|
||||
// nanojoules, to the one micro-joule still carried.
|
||||
let counted = Arc::new(Counted::new(MacosProcessEnergyCounter::try_new().expect("the process counter is live here")));
|
||||
let bare = cascade(counted.clone());
|
||||
let (mut over, mut not_over, mut unread, mut medians) = (0usize, Vec::new(), 0usize, Vec::new());
|
||||
for stage in TIERS {
|
||||
let q = answered_by(stage);
|
||||
let first = bare.resolve(q.clone()).await;
|
||||
assert_eq!((first.stage, first.joule_cost.source), (stage, JouleSource::Estimated), "{}", first.joule_cost);
|
||||
// A read whose thread was not seen to leave its core is no reading
|
||||
// by design, and leaves the tier uncalibrated: ask again.
|
||||
let mut held = bare.calibration(stage);
|
||||
for _ in 0..3 {
|
||||
if held.is_err() {
|
||||
held = bare.calibrate(&q).await.expect("a deterministic tier answers").1;
|
||||
}
|
||||
}
|
||||
let c = held.unwrap_or_else(|why| panic!("{stage} was not calibrated on a live counter: {why}"));
|
||||
let reads = counted.reads();
|
||||
let mut charged = 0u64;
|
||||
for _ in 0..HITS {
|
||||
let r = bare.resolve(q.clone()).await;
|
||||
assert_eq!((r.stage, r.joule_cost.source), (stage, JouleSource::Estimated), "{}", r.joule_cost);
|
||||
charged += r.joule_cost.microjoules - own_cost(stage);
|
||||
}
|
||||
assert_eq!(counted.reads(), reads, "{stage}: a hit read the meter");
|
||||
let calibrated = (HITS * c.per_hit_nanojoules + 500) / 1_000;
|
||||
assert!(charged.abs_diff(calibrated) <= 1, "{stage}: {charged} µJ charged over {HITS} hits of {c:?}");
|
||||
eprintln!(
|
||||
"hit calibration: {stage} on the bare counter: {}; {HITS} hits charged {charged} µJ beside the stage's \
|
||||
own {} µJ each, with no read",
|
||||
in_words(&c),
|
||||
own_cost(stage)
|
||||
);
|
||||
|
||||
// 3. What this change removes: a span of the meter around one hit,
|
||||
// two reads about one lookup, as every resolve took before. Such
|
||||
// a span reads the hit and the meter's own cost, so it reads more
|
||||
// than the hit is charged now. In an optimized build, where the
|
||||
// lookup is a fraction of a microsecond, it reads several times
|
||||
// as much: at least three times, at the median of the tries. An
|
||||
// unoptimized lookup is itself about as dear as the meter, and
|
||||
// there only the first is asked. Each try calibrates the tier and
|
||||
// then reads one such span, so both ran on the host as it was
|
||||
// then; the median, because the meter's own cost in one span
|
||||
// moves severalfold from span to span on a loaded host.
|
||||
let mut tenths = Vec::new();
|
||||
for attempt in 0..TRIES {
|
||||
let Ok(c) = bare.calibrate(&q).await.expect("a deterministic tier answers").1 else {
|
||||
unread += 1;
|
||||
continue;
|
||||
};
|
||||
let open = counted.read_microjoules();
|
||||
lookup(&bare, stage, &q).await;
|
||||
let close = counted.read_microjoules();
|
||||
let (Ok(open), Ok(close)) = (open, close) else {
|
||||
unread += 1;
|
||||
continue;
|
||||
};
|
||||
let around_one_hit = close.saturating_sub(open) * 1_000;
|
||||
tenths.push(around_one_hit * 10 / c.per_hit_nanojoules.max(1));
|
||||
if around_one_hit > c.per_hit_nanojoules {
|
||||
over += 1;
|
||||
} else {
|
||||
not_over.push((stage, attempt, c.per_hit_nanojoules, around_one_hit));
|
||||
}
|
||||
}
|
||||
tenths.sort_unstable();
|
||||
let times = |tenths: Option<&u64>| tenths.map_or(0.0, |t| *t as f64 / 10.0);
|
||||
eprintln!(
|
||||
"hit calibration: {stage}: a span around one hit read {:.1} to {:.1} times what the hit is charged, {:.1} \
|
||||
at the median ({TRIES} tries)",
|
||||
times(tenths.first()),
|
||||
times(tenths.last()),
|
||||
times(tenths.get(tenths.len() / 2))
|
||||
);
|
||||
medians.push((stage, times(tenths.get(tenths.len() / 2))));
|
||||
}
|
||||
eprintln!(
|
||||
"hit calibration: {over} of {} spans around one hit read more than the hit's calibrated figure; {} did not \
|
||||
{not_over:?} (tier, try, per-hit nJ, span nJ); {unread} not read; at the median {medians:?} times the figure{}",
|
||||
3 * TRIES,
|
||||
not_over.len(),
|
||||
if cfg!(debug_assertions) { " (an unoptimized build: no multiple is asked)" } else { "" }
|
||||
);
|
||||
|
||||
// A tier that cannot be calibrated says why, and is not charged.
|
||||
assert_eq!(bare.calibration(Stage::Neural), Err(NoCalibration::NotADeterministicTier));
|
||||
|
||||
// A tenth may miss: on a loaded host one reader's wake-up can land on a
|
||||
// costlier core than the batch did, and a read whose thread was not
|
||||
// seen to leave its core is no reading by design.
|
||||
assert!(
|
||||
(not_calibrated.len() + loose.len()) * 10 <= 3 * TRIES,
|
||||
"{} calibrations failed {not_calibrated:?} and {} were not pinned within a fifth by the oracle {loose:?}",
|
||||
not_calibrated.len(),
|
||||
loose.len()
|
||||
);
|
||||
assert!(
|
||||
(not_over.len() + unread) * 10 <= 3 * TRIES,
|
||||
"{} spans around one hit read no more than the hit's figure {not_over:?}, and {unread} were not read",
|
||||
not_over.len()
|
||||
);
|
||||
if !cfg!(debug_assertions) {
|
||||
assert!(
|
||||
medians.iter().all(|(_, times)| *times >= 3.0),
|
||||
"a span around one hit read under three times the hit's figure at the median: {medians:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,201 @@
|
|||
//! What a resolve the neural stage answers is charged, against the
|
||||
//! hardware: the cascade on this process's own CPU energy counter, its span
|
||||
//! held against an oracle that reads the same count its own way.
|
||||
//!
|
||||
//! The cascade reads its meter only once the three deterministic tiers have
|
||||
//! missed, and again after the neural stage. So the figure is the neural
|
||||
//! stage's, and work done before the walk reached it is not in it. Here the
|
||||
//! KV stage burns 8 ms of this thread's CPU on its way to missing, every
|
||||
//! resolve, straight before the cascade opens its span.
|
||||
//!
|
||||
//! One test binary with one test: the counter's domain is every thread of
|
||||
//! the process, and the oracle's check is the process's context-switch
|
||||
//! count, so nothing else may run here. No joule figure and no timing is
|
||||
//! asserted: what a millisecond of CPU costs follows the core that ran it,
|
||||
//! and a loaded host stretches everything's wall time.
|
||||
|
||||
#![cfg(target_os = "macos")]
|
||||
|
||||
mod oracle;
|
||||
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use eoc_cache::LruCache;
|
||||
use eoc_cascade::Cascade;
|
||||
use eoc_core::{JouleCost, JouleSource, Query, Response, Stage};
|
||||
use eoc_graph::GraphStage;
|
||||
use eoc_kv::{KvBackend, KvStage};
|
||||
use eoc_meter::{CounterStatus, JouleCounter, MacosProcessEnergyCounter};
|
||||
use eoc_neural::{NeuralBackend, NeuralStage};
|
||||
|
||||
use oracle::{Counted, Span, Watched, burn, energy_nj_off_core};
|
||||
|
||||
/// What the KV stage burns before it misses, and what the neural stage
|
||||
/// then burns or sleeps.
|
||||
const BEFORE_THE_SPAN: Duration = Duration::from_millis(8);
|
||||
const NEURAL: Duration = Duration::from_millis(2);
|
||||
|
||||
/// A KV backend that knows nothing, and burns this thread's CPU finding
|
||||
/// that out. It counts its lookups, so the test can see the walk ran.
|
||||
struct BurnsThenMisses(Arc<AtomicU64>);
|
||||
|
||||
impl KvBackend for BurnsThenMisses {
|
||||
fn get(&self, _key: &str) -> Option<Vec<u8>> {
|
||||
burn(BEFORE_THE_SPAN);
|
||||
self.0.fetch_add(1, Ordering::SeqCst);
|
||||
None
|
||||
}
|
||||
fn put(&self, _key: &str, _value: Vec<u8>) {}
|
||||
fn embeddings(&self) -> Vec<(Vec<f32>, String)> {
|
||||
Vec::new()
|
||||
}
|
||||
fn put_embedding(&self, _embedding: Vec<f32>, _key: String) {}
|
||||
}
|
||||
|
||||
/// A neural stage that works, or only sleeps, and claims no figure for it:
|
||||
/// the resolve's cost is then the cascade's span alone, measured.
|
||||
enum Neural {
|
||||
Burns,
|
||||
Sleeps,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl NeuralBackend for Neural {
|
||||
async fn infer(&self, q: &Query) -> Response {
|
||||
match self {
|
||||
Self::Burns => burn(NEURAL),
|
||||
Self::Sleeps => std::thread::sleep(NEURAL),
|
||||
}
|
||||
Response::new(q.id, "inferred".to_string(), Stage::Neural, JouleCost::zero())
|
||||
}
|
||||
}
|
||||
|
||||
/// A cascade whose KV stage burns and misses, around `neural`, metered by
|
||||
/// `meter`. Nothing is memoized, so every resolve walks to the neural stage.
|
||||
fn cascade(meter: Arc<dyn JouleCounter>, neural: Neural, lookups: &Arc<AtomicU64>) -> Cascade {
|
||||
Cascade::new(
|
||||
Arc::new(LruCache::new(8)),
|
||||
Arc::new(KvStage::new(Box::new(BurnsThenMisses(lookups.clone())))),
|
||||
Arc::new(GraphStage::new()),
|
||||
Arc::new(NeuralStage::new(Box::new(neural))),
|
||||
)
|
||||
.with_meter(meter)
|
||||
.without_memoization()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_neural_resolve_reads_the_neural_stage_and_nothing_before_it() {
|
||||
const TRIES: usize = 40;
|
||||
let status = eoc_meter::process_energy_status();
|
||||
eprintln!("neural span: status={status:?}");
|
||||
assert_eq!(status, CounterStatus::Live, "this process's CPU energy counter is not live here");
|
||||
let q = Query::new("nothing knows this");
|
||||
let lookups = Arc::new(AtomicU64::new(0));
|
||||
|
||||
// 1. A neural stage that burns 2 ms, through a counter the oracle reads
|
||||
// on either side of. The resolve takes exactly two reads, both after
|
||||
// the KV stage's burn. Its figure lies between the oracle's window
|
||||
// inside the span, which holds the neural stage, and the oracle's
|
||||
// window around the span and both reads, which opens after the burn:
|
||||
// that much holds for any count that only grows. What the windows
|
||||
// leave open is their own distance apart, the two readers' reads and
|
||||
// wake-ups: the figure is pinned within a third when the window
|
||||
// inside is at least two thirds of the one around.
|
||||
let counter = MacosProcessEnergyCounter::try_new().expect("the process counter is live here");
|
||||
let watched = Arc::new(Watched::beside(counter, lookups.clone()));
|
||||
let through_oracle = cascade(watched.clone(), Neural::Burns, &lookups);
|
||||
let (mut tight, mut loose, mut unread, mut burned) = (Vec::new(), Vec::new(), 0usize, Vec::new());
|
||||
for attempt in 0..TRIES {
|
||||
watched.take();
|
||||
let (walked, before) = (lookups.load(Ordering::SeqCst), energy_nj_off_core());
|
||||
let r = through_oracle.resolve(q.clone()).await;
|
||||
let reads = watched.take();
|
||||
assert_eq!(r.stage, Stage::Neural);
|
||||
assert_eq!(lookups.load(Ordering::SeqCst), walked + 1, "the walk went through the KV stage");
|
||||
assert_eq!(reads.len(), 2, "#{attempt}: a neural resolve reads the meter exactly twice: {reads:?}");
|
||||
assert_eq!(reads[0].count, walked + 1, "#{attempt}: the span opened before the KV stage had missed");
|
||||
let span = Span::between(&reads[0], &reads[1]);
|
||||
// The burn is in the oracle's count before the cascade's first read.
|
||||
burned.push(reads[0].before_nanojoules - before);
|
||||
let Some(read) = span.read else {
|
||||
// A read failed: the stage's own cost stands, an estimated zero.
|
||||
assert_eq!(r.joule_cost, JouleCost::zero(), "#{attempt}");
|
||||
unread += 1;
|
||||
continue;
|
||||
};
|
||||
assert_eq!(r.joule_cost, JouleCost::measured(read / 1_000), "#{attempt}: the span, measured");
|
||||
assert!(
|
||||
span.holds(read),
|
||||
"#{attempt}: {} µJ measured outside the oracle's windows, {} nJ inside the span and {} nJ around it",
|
||||
read / 1_000,
|
||||
span.inside,
|
||||
span.around
|
||||
);
|
||||
if span.tightness() * 3 < 2_000 {
|
||||
loose.push((attempt, read, span.inside, span.around));
|
||||
} else {
|
||||
tight.push(span.tightness());
|
||||
}
|
||||
}
|
||||
burned.sort_unstable();
|
||||
eprintln!(
|
||||
"neural span: a 2 ms neural stage x {TRIES}: {} figures within the oracle's two windows with the inner at \
|
||||
least {}/1000 of the outer, {} looser {loose:?} (try, measured nJ, inside nJ, around nJ), {unread} not read; \
|
||||
the 8 ms before each span cost {} to {} µJ and none of it is in a figure",
|
||||
tight.len(),
|
||||
tight.iter().min().copied().unwrap_or(0),
|
||||
loose.len(),
|
||||
burned.first().copied().unwrap_or(0) / 1_000,
|
||||
burned.last().copied().unwrap_or(0) / 1_000,
|
||||
);
|
||||
|
||||
// 2. A neural stage that only sleeps 2 ms, on the bare counter, as the
|
||||
// cascade is used: nothing takes this thread off its core between the
|
||||
// KV stage's 8 ms and the cascade's first read. The oracle's window
|
||||
// is the whole resolve, burn and sleep. A span opened before the
|
||||
// walk would read the burn; opened after it, the figure for the
|
||||
// sleep is under a quarter of that window, or no reading at all.
|
||||
let counted = Arc::new(Counted::new(MacosProcessEnergyCounter::try_new().expect("the process counter is live here")));
|
||||
let bare = cascade(counted.clone(), Neural::Sleeps, &lookups);
|
||||
let (mut slept, mut slept_unread, mut leaked) = (Vec::new(), 0usize, Vec::new());
|
||||
for attempt in 0..TRIES {
|
||||
let (walked, reads, before) = (lookups.load(Ordering::SeqCst), counted.reads(), energy_nj_off_core());
|
||||
let r = bare.resolve(q.clone()).await;
|
||||
let whole = energy_nj_off_core() - before;
|
||||
assert_eq!(lookups.load(Ordering::SeqCst), walked + 1, "the walk went through the KV stage");
|
||||
assert_eq!(counted.reads(), reads + 2, "#{attempt}: a neural resolve reads the meter exactly twice");
|
||||
match r.joule_cost.source {
|
||||
// No span: the stage's own cost stands, an estimated zero.
|
||||
JouleSource::Estimated => {
|
||||
assert_eq!(r.joule_cost, JouleCost::zero(), "#{attempt}");
|
||||
slept_unread += 1;
|
||||
}
|
||||
JouleSource::Measured if r.joule_cost.microjoules * 1_000 * 4 >= whole => {
|
||||
leaked.push((attempt, r.joule_cost.microjoules, whole));
|
||||
}
|
||||
JouleSource::Measured => slept.push(r.joule_cost.microjoules * 1_000_000 / whole.max(1)),
|
||||
}
|
||||
}
|
||||
eprintln!(
|
||||
"neural span: a 2 ms sleep after the KV stage's 8 ms burn x {TRIES}: {} read under a quarter of the oracle's \
|
||||
window over the whole resolve (at most {}/1000 of it), {slept_unread} not read, {} read the burn before the \
|
||||
span {leaked:?} (try, µJ, oracle nJ)",
|
||||
slept.len(),
|
||||
slept.iter().max().copied().unwrap_or(0),
|
||||
leaked.len()
|
||||
);
|
||||
|
||||
// A tenth may miss: on a loaded host a reader's wake-up can land on a
|
||||
// costlier core than the work did, and a read whose thread was not seen
|
||||
// to leave its core is no reading by design. A span opened before the
|
||||
// walk misses every time.
|
||||
assert!(
|
||||
(loose.len() + unread) * 10 <= TRIES,
|
||||
"{} of {TRIES} neural spans were not pinned within a third by the oracle {loose:?} and {unread} were not read",
|
||||
loose.len()
|
||||
);
|
||||
assert!(leaked.len() * 10 <= TRIES, "{} of {TRIES} sleeps read the burn before them: {leaked:?}", leaked.len());
|
||||
assert!(slept_unread * 10 <= TRIES, "{slept_unread} of {TRIES} sleeping neural stages were not read");
|
||||
}
|
||||
213
eoc/eoc-rs/crates/eoc-cascade/tests/oracle/mod.rs
Normal file
213
eoc/eoc-rs/crates/eoc-cascade/tests/oracle/mod.rs
Normal file
|
|
@ -0,0 +1,213 @@
|
|||
//! The tests' own reading of this process's CPU energy, shared by the test
|
||||
//! binaries of this directory that hold the cascade's figures against it.
|
||||
//!
|
||||
//! It shares no call and no number with the counter under test:
|
||||
//! `proc_pid_rusage` is declared here and read into raw words at the SDK's
|
||||
//! offset, and a read is taken only once this thread has left its core by a
|
||||
//! rule of the oracle's own — the process's context-switch count rose across
|
||||
//! a sleep — where the counter watches the reading thread's own energy count.
|
||||
//! That rule holds only while one thread of the process runs, so each binary
|
||||
//! here has one test.
|
||||
|
||||
#![allow(dead_code)]
|
||||
|
||||
use std::os::raw::{c_int, c_void};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use eoc_meter::JouleCounter;
|
||||
|
||||
unsafe extern "C" {
|
||||
fn proc_pid_rusage(pid: c_int, flavor: c_int, buffer: *mut c_void) -> c_int;
|
||||
fn proc_pidinfo(pid: c_int, flavor: c_int, arg: u64, buffer: *mut c_void, buffersize: c_int) -> c_int;
|
||||
fn clock_gettime_nsec_np(clock_id: u32) -> u64;
|
||||
}
|
||||
|
||||
/// `CLOCK_THREAD_CPUTIME_ID` in macOS `<time.h>`.
|
||||
const CLOCK_THREAD_CPUTIME_ID: u32 = 16;
|
||||
|
||||
/// This thread's CPU time.
|
||||
fn thread_cpu() -> Duration {
|
||||
Duration::from_nanos(unsafe { clock_gettime_nsec_np(CLOCK_THREAD_CPUTIME_ID) })
|
||||
}
|
||||
|
||||
/// Spin this thread until it has used `d` of its own CPU time (or 20 s of
|
||||
/// wall time have passed), however loaded the host is.
|
||||
pub fn burn(d: Duration) {
|
||||
let (started, wall) = (thread_cpu(), Instant::now());
|
||||
let mut acc: u64 = 0;
|
||||
while thread_cpu() - started < d && wall.elapsed() < Duration::from_secs(20) {
|
||||
for _ in 0..1_000 {
|
||||
acc = std::hint::black_box(acc.wrapping_add(1));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn me() -> c_int {
|
||||
std::process::id() as c_int
|
||||
}
|
||||
|
||||
/// This process's `ri_energy_nj` as it stands: `proc_pid_rusage(pid,
|
||||
/// RUSAGE_INFO_V6)` into 464 raw bytes, the count at byte 336.
|
||||
fn energy_nj() -> u64 {
|
||||
let mut words = [0u64; 464 / 8];
|
||||
let rc = unsafe { proc_pid_rusage(me(), 6, words.as_mut_ptr() as *mut c_void) };
|
||||
assert_eq!(rc, 0, "the oracle's proc_pid_rusage");
|
||||
words[336 / 8]
|
||||
}
|
||||
|
||||
/// How many times this process's threads have been switched off a core:
|
||||
/// `pti_csw`, the `int32_t` at byte 80 of the 96-byte `proc_taskinfo` that
|
||||
/// `proc_pidinfo(pid, PROC_PIDTASKINFO = 4)` fills.
|
||||
fn context_switches() -> i32 {
|
||||
let mut info = [0i32; 96 / 4];
|
||||
let wrote = unsafe { proc_pidinfo(me(), 4, 0, info.as_mut_ptr() as *mut c_void, 96) };
|
||||
assert_eq!(wrote, 96, "the oracle's proc_pidinfo(PROC_PIDTASKINFO)");
|
||||
info[80 / 4]
|
||||
}
|
||||
|
||||
/// The oracle's read of this process's energy, in nanojoules, taken once
|
||||
/// this thread has left its core: a sleep of 200 µs, then 1, 2, 4 and 8 ms,
|
||||
/// until the process's context-switch count rose across one. Only the test's
|
||||
/// thread runs in the process, so a rise is this thread's switch, and the
|
||||
/// read after it holds all the thread spent before it.
|
||||
pub fn energy_nj_off_core() -> u64 {
|
||||
let left = [200, 1_000, 2_000, 4_000, 8_000].into_iter().any(|micros| {
|
||||
let before = context_switches();
|
||||
std::thread::sleep(Duration::from_micros(micros));
|
||||
context_switches() > before
|
||||
});
|
||||
assert!(left, "the oracle never saw this thread leave its core across sleeps of 0.2, 1, 2, 4 and 8 ms");
|
||||
energy_nj()
|
||||
}
|
||||
|
||||
/// One read of a [`Watched`] counter: the oracle's reading before it, what
|
||||
/// the counter read (`None` for a failed read), the oracle's after it, and
|
||||
/// where the test's own count stood when the read was asked for.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct Read {
|
||||
pub before_nanojoules: u64,
|
||||
pub microjoules: Option<u64>,
|
||||
pub after_nanojoules: u64,
|
||||
pub count: u64,
|
||||
}
|
||||
|
||||
/// A counter with the oracle's own read on either side of every read of it,
|
||||
/// kept in order. A span between two of its reads therefore lies between two
|
||||
/// windows of the oracle's: the one inside the span, from the oracle's read
|
||||
/// after the opening read to its read before the closing one, and the one
|
||||
/// around the span and both reads.
|
||||
///
|
||||
/// The oracle's reads between the two readings of a span are the span's too:
|
||||
/// its sleep after the opening read and before the closing one. So a span
|
||||
/// through this counter reads a little more than one through the bare
|
||||
/// counter would, and both windows hold it.
|
||||
pub struct Watched<C> {
|
||||
counter: C,
|
||||
reads: Mutex<Vec<Read>>,
|
||||
count: Arc<AtomicU64>,
|
||||
}
|
||||
|
||||
impl<C: JouleCounter> Watched<C> {
|
||||
pub fn new(counter: C) -> Self {
|
||||
Self::beside(counter, Arc::new(AtomicU64::new(0)))
|
||||
}
|
||||
|
||||
/// A watched counter that also notes, at each read, where `count`
|
||||
/// stood: a count the test moves, to place a read among its own steps.
|
||||
pub fn beside(counter: C, count: Arc<AtomicU64>) -> Self {
|
||||
Self { counter, reads: Mutex::new(Vec::new()), count }
|
||||
}
|
||||
|
||||
/// The reads taken since the last call, in order.
|
||||
pub fn take(&self) -> Vec<Read> {
|
||||
std::mem::take(&mut *self.reads.lock().unwrap())
|
||||
}
|
||||
}
|
||||
|
||||
impl<C: JouleCounter> JouleCounter for Watched<C> {
|
||||
fn read_microjoules(&self) -> eoc_core::Result<u64> {
|
||||
let count = self.count.load(Ordering::SeqCst);
|
||||
let before_nanojoules = energy_nj_off_core();
|
||||
let read = self.counter.read_microjoules();
|
||||
let after_nanojoules = energy_nj_off_core();
|
||||
self.reads.lock().unwrap().push(Read {
|
||||
before_nanojoules,
|
||||
microjoules: read.as_ref().ok().copied(),
|
||||
after_nanojoules,
|
||||
count,
|
||||
});
|
||||
read
|
||||
}
|
||||
fn name(&self) -> &'static str {
|
||||
self.counter.name()
|
||||
}
|
||||
}
|
||||
|
||||
/// A span between two reads of a [`Watched`] counter, in nanojoules: what
|
||||
/// the counter read, and the oracle's two windows.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct Span {
|
||||
/// The counter's two reads' delta; `None` when either failed or the
|
||||
/// counter did not advance.
|
||||
pub read: Option<u64>,
|
||||
/// The oracle's window inside the span.
|
||||
pub inside: u64,
|
||||
/// The oracle's window around the span and both of its reads.
|
||||
pub around: u64,
|
||||
}
|
||||
|
||||
impl Span {
|
||||
pub fn between(open: &Read, close: &Read) -> Self {
|
||||
let read = match (open.microjoules, close.microjoules) {
|
||||
(Some(open), Some(close)) if close > open => Some((close - open) * 1_000),
|
||||
_ => None,
|
||||
};
|
||||
Self {
|
||||
read,
|
||||
inside: close.before_nanojoules - open.after_nanojoules,
|
||||
around: close.after_nanojoules - open.before_nanojoules,
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether `nanojoules`, a figure for this span, lies between the
|
||||
/// oracle's two windows, give or take the micro-joule the counter's two
|
||||
/// reads round away. The count only grows, so a reading of the span that
|
||||
/// does not is not this process's count.
|
||||
pub fn holds(&self, nanojoules: u64) -> bool {
|
||||
self.inside < nanojoules + 1_000 && nanojoules < self.around + 1_000
|
||||
}
|
||||
|
||||
/// The window inside the span as thousandths of the one around it: how
|
||||
/// closely the oracle pins the span.
|
||||
pub fn tightness(&self) -> u64 {
|
||||
self.inside * 1_000 / self.around.max(1)
|
||||
}
|
||||
}
|
||||
|
||||
/// A counter that counts its reads.
|
||||
pub struct Counted<C> {
|
||||
counter: C,
|
||||
reads: Mutex<u64>,
|
||||
}
|
||||
|
||||
impl<C: JouleCounter> Counted<C> {
|
||||
pub fn new(counter: C) -> Self {
|
||||
Self { counter, reads: Mutex::new(0) }
|
||||
}
|
||||
|
||||
pub fn reads(&self) -> u64 {
|
||||
*self.reads.lock().unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
impl<C: JouleCounter> JouleCounter for Counted<C> {
|
||||
fn read_microjoules(&self) -> eoc_core::Result<u64> {
|
||||
*self.reads.lock().unwrap() += 1;
|
||||
self.counter.read_microjoules()
|
||||
}
|
||||
fn name(&self) -> &'static str {
|
||||
self.counter.name()
|
||||
}
|
||||
}
|
||||
|
|
@ -27,9 +27,10 @@ pub struct JouleCost {
|
|||
|
||||
impl JouleCost {
|
||||
/// Zero micro-joules, tagged [`JouleSource::Estimated`]: a figure no
|
||||
/// counter read. It is what a cache hit costs by assumption, since the
|
||||
/// energy of an in-process lookup is below any counter's resolution and
|
||||
/// no meter is read for it.
|
||||
/// counter read. It is what the cache stage reports for a hit: no meter
|
||||
/// is read for one, since a read of a counter costs more than the lookup
|
||||
/// it would bracket. (eoc-cascade adds what a hit costs on the host, as
|
||||
/// its meter read it over a batch of lookups; that is an estimate too.)
|
||||
///
|
||||
/// It is never [`JouleSource::Measured`]. Measured means a live hardware
|
||||
/// counter read the span and advanced over it; a zero that no counter read
|
||||
|
|
|
|||
|
|
@ -13,6 +13,10 @@
|
|||
//! which never moves — is no reading, and yields no measured figure:
|
||||
//! eoc-streaming reports none, eoc-cascade keeps the answering stage's own
|
||||
//! cost as the stage reported it, and those backends report an estimate.
|
||||
//! eoc-cascade reads a span only around its neural stage: a hit on its
|
||||
//! cache, KV stage or graph is shorter than a read of a counter, so it is
|
||||
//! charged what the counter read over a batch of such hits in one span, per
|
||||
//! hit, and that is an estimate.
|
||||
//!
|
||||
//! Hardware counter dependencies are **platform-conditional and
|
||||
//! feature-gated**:
|
||||
|
|
@ -5077,7 +5081,11 @@ pub fn process_energy_status() -> CounterStatus {
|
|||
/// core ([`MacosProcessEnergyCounter`]). A span bracketed by two reads
|
||||
/// therefore costs two such sleeps and holds its closing read's own wake-up
|
||||
/// (tens of microjoules): bracket a batch of short operations with one span,
|
||||
/// not each operation with its own.
|
||||
/// not each operation with its own. eoc-cascade does so: it reads this
|
||||
/// counter around its neural stage, and for a cache, KV or graph hit, which
|
||||
/// is under a microsecond, it reads one span around a batch of the lookup the
|
||||
/// first time the tier answers and charges every hit that span's reading per
|
||||
/// lookup, with no read around the hit.
|
||||
pub fn detect_in_process() -> Box<dyn JouleCounter> {
|
||||
detect_in_process_with(process_energy_status, detect)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,7 +3,14 @@
|
|||
//! Subcommands:
|
||||
//!
|
||||
//! - `eoc query "<prompt>"` — run a prompt through the cascade and print
|
||||
//! the stage that resolved it plus the attributed joule cost.
|
||||
//! the stage that resolved it plus the attributed joule cost. A prompt the
|
||||
//! cache, the KV store or the graph answers is charged an estimate: the
|
||||
//! stage's own figure plus what a hit on that tier costs on this host, as
|
||||
//! the meter read it over a batch of the lookup (the `hit` line says how
|
||||
//! many, and what the meter read). No meter is read around the hit itself:
|
||||
//! a read costs more than the hit. A prompt that falls through to the
|
||||
//! neural stage is read by the meter around that stage, and its figure is
|
||||
//! measured when the counter is live and the stage estimated nothing.
|
||||
//! - `eoc meter` — detect available hardware energy counters and report
|
||||
//! the micro-joules consumed during one second of idle, or why nothing
|
||||
//! was metered.
|
||||
|
|
@ -16,8 +23,8 @@ use std::time::Duration;
|
|||
use anyhow::Result;
|
||||
use clap::{Parser, Subcommand};
|
||||
use eoc_cache::LruCache;
|
||||
use eoc_cascade::Cascade;
|
||||
use eoc_core::Query;
|
||||
use eoc_cascade::{Cascade, HitCalibration, NoCalibration};
|
||||
use eoc_core::{Query, Stage};
|
||||
use eoc_graph::{GraphStage, Triple};
|
||||
use eoc_kv::{KvBackend, KvStage, MemoryKvBackend};
|
||||
use eoc_meter::{detect, detect_in_process};
|
||||
|
|
@ -69,6 +76,12 @@ fn build_cascade() -> Cascade {
|
|||
/// else than a whole-machine counter would. A backend passed here must keep
|
||||
/// to that too: work it ran in a child process, or on the GPU, would be
|
||||
/// outside the counter's domain.
|
||||
///
|
||||
/// The cascade reads that counter around the neural stage only. Each read of
|
||||
/// this process's counter sleeps the reading thread off its core first, about
|
||||
/// 80 µs, where a hit on the cache, the KV store or the graph is under a
|
||||
/// microsecond; so a hit is not bracketed, and is charged the figure its
|
||||
/// tier was calibrated at the first time it answered.
|
||||
fn build_cascade_with(neural: Box<dyn NeuralBackend>) -> Cascade {
|
||||
let cache = Arc::new(LruCache::new(1024));
|
||||
|
||||
|
|
@ -97,17 +110,35 @@ async fn run_query(prompt: String) -> Result<()> {
|
|||
println!("stage : {}", r.stage);
|
||||
println!("payload: {}", r.payload);
|
||||
println!("cost : {}", r.joule_cost);
|
||||
if r.stage != Stage::Neural {
|
||||
println!("hit : {}", hit_line(r.stage, cascade.meter().name(), cascade.calibration(r.stage)));
|
||||
}
|
||||
println!("receipt: {}", r.receipt);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Where a hit's estimated cost came from: the figure its tier was
|
||||
/// calibrated at on this host and the span it was read over, or why the
|
||||
/// tier has none and the cost is the stage's own figure alone.
|
||||
fn hit_line(stage: Stage, meter: &str, calibration: Result<HitCalibration, NoCalibration>) -> String {
|
||||
match calibration {
|
||||
Ok(c) => format!(
|
||||
"a {stage} hit is charged {} nJ on this host, on top of the stage's own figure: {meter} read {} µJ over \
|
||||
{} lookups in one span ({} µJ over none)",
|
||||
c.per_hit_nanojoules, c.span_microjoules, c.batch, c.empty_span_microjoules
|
||||
),
|
||||
Err(why) => format!("the {stage} stage's own figure; no calibration by {meter} ({why})"),
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_meter() -> Result<()> {
|
||||
let counter = detect();
|
||||
println!("counter: {}", counter.name());
|
||||
#[cfg(target_os = "macos")]
|
||||
println!("ioreport: {}", ioreport_line(eoc_meter::ioreport_status()));
|
||||
println!(
|
||||
"query : metered by {} (its stages all run on the CPU in this process)",
|
||||
"query : metered by {} (its stages all run on the CPU in this process): read around the neural stage; a \
|
||||
cache, kv or graph hit is charged a figure calibrated over a batch, with no read",
|
||||
detect_in_process().name()
|
||||
);
|
||||
let start = counter.read_microjoules();
|
||||
|
|
@ -151,8 +182,9 @@ fn idle_line(counter: &str, start: eoc_core::Result<u64>, end: eoc_core::Result<
|
|||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{build_cascade_with, idle_line};
|
||||
use super::{build_cascade, build_cascade_with, hit_line, idle_line};
|
||||
use async_trait::async_trait;
|
||||
use eoc_cascade::{HitCalibration, NoCalibration};
|
||||
use eoc_core::{Error, JouleCost, JouleSource, Query, Response, Stage};
|
||||
use eoc_neural::NeuralBackend;
|
||||
use std::time::{Duration, Instant};
|
||||
|
|
@ -186,9 +218,9 @@ mod tests {
|
|||
/// the stub, or the powermetrics stand-in — the cost is the backend's own
|
||||
/// estimate: the same non-zero figure, tagged estimated. Where it is a live
|
||||
/// counter (this process's own, on a Mac where it is live), the cost is
|
||||
/// that estimate plus what the counter read, still tagged estimated,
|
||||
/// because part of it is. Never a zero for the work, and never tagged
|
||||
/// measured.
|
||||
/// that estimate plus what the counter read around the neural stage, still
|
||||
/// tagged estimated, because part of it is. Never a zero for the work, and
|
||||
/// never tagged measured.
|
||||
#[tokio::test]
|
||||
async fn work_on_an_unmetered_host_keeps_its_estimate() {
|
||||
let _one = BURNING.lock().await;
|
||||
|
|
@ -242,8 +274,9 @@ mod tests {
|
|||
}
|
||||
|
||||
/// On a Mac whose process counter is live, the CLI's cascade measures
|
||||
/// in-process CPU work: ~100 ms of one thread's work reads as a measured
|
||||
/// figure — the counter read all of it, and nothing was estimated —
|
||||
/// in-process CPU work: ~100 ms of one thread's work in the neural stage
|
||||
/// reads as a measured figure — the counter read all of it, between the
|
||||
/// cascade's two reads around that stage, and nothing was estimated —
|
||||
/// between 1 mJ and 10 J. The floor is low on purpose: a wall-time burn
|
||||
/// on a loaded host, or on efficiency cores, can draw little (an M3
|
||||
/// Ultra's efficiency cores read 0.013 J per 200 ms of CPU); the size of
|
||||
|
|
@ -261,6 +294,74 @@ mod tests {
|
|||
assert!((1_000..=10_000_000).contains(&r.joule_cost.microjoules), "{}", r.joule_cost);
|
||||
}
|
||||
|
||||
/// The CLI's own path for a prompt a deterministic tier answers: `ping`
|
||||
/// is in the KV store, and asked again it is in the cache. Neither hit
|
||||
/// is measured, on any host. Where the tier was calibrated — the meter
|
||||
/// is a live counter — the cost is the stage's own figure plus the whole
|
||||
/// micro-joules of the calibrated figure (a first hit carries nothing
|
||||
/// over); where it was not, the stage's own figure alone, and the reason
|
||||
/// is one this host's meter can give.
|
||||
#[tokio::test]
|
||||
async fn a_hit_is_charged_an_estimate_calibrated_on_this_host() {
|
||||
// The meter's domain is the whole process: no other test's burn
|
||||
// may land in a calibration's span.
|
||||
let _one = BURNING.lock().await;
|
||||
let cascade = build_cascade();
|
||||
let meter = cascade.meter().name();
|
||||
for (stage, own) in [(Stage::Kv, 50), (Stage::Cache, 0)] {
|
||||
let r = cascade.resolve(Query::new("ping")).await;
|
||||
assert_eq!((r.stage, r.payload.as_str()), (stage, "pong"));
|
||||
assert_eq!(r.joule_cost.source, JouleSource::Estimated, "{}", r.joule_cost);
|
||||
let calibration = cascade.calibration(stage);
|
||||
eprintln!("eoc {stage} hit on this host: cost={} {}", r.joule_cost, hit_line(stage, meter, calibration));
|
||||
match calibration {
|
||||
Ok(c) => {
|
||||
assert!(!matches!(meter, "stub" | "powermetrics"), "{meter} calibrated a tier: {c:?}");
|
||||
assert!(c.batch >= 1 && c.span_microjoules >= 1 && c.empty_span_microjoules >= 1, "{c:?}");
|
||||
assert_eq!(r.joule_cost, JouleCost::estimated(own + c.per_hit_nanojoules / 1_000), "{c:?}");
|
||||
}
|
||||
Err(why) => {
|
||||
assert_eq!(r.joule_cost, JouleCost::estimated(own), "{why:?}");
|
||||
let expected: &[NoCalibration] = match meter {
|
||||
// The stub reads, and never moves.
|
||||
"stub" => &[NoCalibration::CounterHeld],
|
||||
// The stand-in's every read fails.
|
||||
"powermetrics" => &[NoCalibration::ReadFailed],
|
||||
// A live counter: only a read it declined to take.
|
||||
_ => &[NoCalibration::ReadFailed],
|
||||
};
|
||||
assert!(expected.contains(&why), "{meter}: {why:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The `hit` line says what a hit was charged and over what it was
|
||||
/// read, or why nothing was.
|
||||
#[test]
|
||||
fn the_hit_line_shows_the_calibration_or_why_there_is_none() {
|
||||
let c = HitCalibration {
|
||||
per_hit_nanojoules: 2_803,
|
||||
batch: 488,
|
||||
span_microjoules: 1_368,
|
||||
empty_span_microjoules: 16,
|
||||
took: Duration::from_micros(1_017),
|
||||
};
|
||||
assert_eq!(
|
||||
hit_line(Stage::Kv, "process-energy", Ok(c)),
|
||||
"a kv hit is charged 2803 nJ on this host, on top of the stage's own figure: process-energy read 1368 µJ \
|
||||
over 488 lookups in one span (16 µJ over none)"
|
||||
);
|
||||
assert_eq!(
|
||||
hit_line(Stage::Cache, "stub", Err(NoCalibration::CounterHeld)),
|
||||
"the cache stage's own figure; no calibration by stub (the counter did not advance)"
|
||||
);
|
||||
assert_eq!(
|
||||
hit_line(Stage::Graph, "powermetrics", Err(NoCalibration::ReadFailed)),
|
||||
"the graph stage's own figure; no calibration by powermetrics (a read of the meter failed)"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_counter_that_did_not_read_is_never_printed_as_a_figure() {
|
||||
let failed = idle_line("powermetrics", Err(Error::NoCounter), Err(Error::NoCounter));
|
||||
|
|
|
|||
Loading…
Reference in a new issue