From 2add5331046f1ec63bfc0f3369401a464096b091 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Tue, 13 Oct 2026 01:26:37 -0400 Subject: [PATCH 1/4] eoc-cascade: a hit reads no meter and is charged a calibrated estimate; the span opens at the neural stage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Cascade::resolve` took one read of its meter before the walk and one after, whatever answered. Since 0f6a9a3b a read of this process's own energy counter first sees the reading thread leave its core, so it costs about 80 µs, and a span holds its closing read's own wake-up. A cache, KV or graph hit is well under a microsecond. So a resolve one of them answered paid two reads to bracket a lookup three hundred times shorter than they are, and what it then sealed as the hit's cost, tagged measured (f84f0b2f's last case: an estimated zero on a live span is the span's reading), was the meter's own. Measured on the build host (M3 Ultra, macOS 27.0.1, load average 11 to 19, `examples/hit_cost`, release), parent ac93d9e5 against this commit, 10,000 resolves answered by the cache: - before: 1,647 ms in all, 164 µs a resolve at the median; every one tagged measured, 13 to 28 µJ, 141,447 µJ in all. A neural stage that does nothing read the same: 164 µs and 15 to 20 µJ. - after: 4.4 ms in all, 0.42 µs a resolve at the median; none tagged measured, 16,650 µJ in all, which is the 1,665 nJ a hit the tier was calibrated at. The rule is unchanged: nothing that was not read from a counter is measured. What changes is where the meter is read. - A resolve the neural stage answers is read as before, but the span opens once the three deterministic tiers have missed. The misses are spent before it and are not in the figure: 0.95 µJ and 223 ns for the three together, against the 13 to 24 µJ the span holds of the meter itself. So the measured figure is still not under the neural stage's energy. `cascade_cost` and `live_span` are as they were. A failed opening read is no longer followed by a closing one, which could not make a span. - A resolve a deterministic tier answers reads no meter and is never measured. It is charged the cost its stage reported plus a figure calibrated on this host: the first time a tier answers, the cascade reads the counter over no work until it moves (the meter's own cost), then repeats that tier's lookup inside one span, growing the batch until the span reads fifty times that (and 200 µJ at least) or 40 ms have gone, and keeps the span's reading per lookup, in nanojoules. It is an estimate: read by this counter on this host, but over a batch, and a mean is not a reading of any one hit. - Per-hit figures are around a microjoule and `JouleCost` is whole microjoules, so each tier carries the nanojoules not yet charged. The microjoules charged over any run of hits are the whole microjoules in what those hits accrued: never each hit rounded up, never a string of zeros. - The stage's own figure is added, as `cascade_cost` adds an estimate to a span and for its reason: it may be energy the counter never saw. A cached response that carries a measured figure is charged as an estimate too: it replays what was read for an earlier resolve. - No live counter, no calibration: the cascade given no meter (nothing is read to find the stub does not move), a failed read, a counter that holds for 2 ms of back-to-back reads or over every batch. The hit then keeps its stage's cost. The attempt is made once a tier; later hits read nothing. - `Cascade::calibration` reads what a tier holds (per-hit nJ, batch, span µJ, empty-span µJ, time taken) or why it holds nothing; `Cascade::calibrate` calibrates the tier that answers a query now, again if asked, and keeps the carry; `without_hit_calibration` switches a tier off, for a KV backend whose `get` must not be repeated. The three stages here are safe to repeat: the cache's lookup only refreshes a recency the hit already refreshed, the KV and graph lookups only read. Calibrated on that host: 1,356 to 1,665 nJ a cache hit, 2,669 to 3,250 nJ a KV hit, 2,946 to 3,152 nJ a graph hit, each from 224 to 4,032 lookups in a span of 0.7 to 5.9 mJ, 54 to 347 times the 12 to 29 µJ an empty span read, in 0.85 to 2.4 ms. Tests, on scripted counters and on a model host whose meter, clock and KV backend are one count: a hit takes no read once its tier is calibrated (not one in a thousand hits, each tier) and a neural resolve exactly two, the first after the KV lookup; the microjoules charged over 10,000 hits equal the calibrated nanojoules at every hit, for 139 nJ and 2,359 nJ a hit, and across eight threads; a calibration's batch is the lookups between its last two reads and its span their delta; a 1 ms lookup is repeated 39 times and a 100 ms one once; with no live counter every resolve keeps its stage's cost; and no hit is measured, on any counter, the calibrating hit included. Co-Authored-By: Claude Opus 5.5 --- .../crates/eoc-cascade/examples/hit_cost.rs | 156 +++ eoc/eoc-rs/crates/eoc-cascade/src/lib.rs | 1228 ++++++++++++++++- 2 files changed, 1321 insertions(+), 63 deletions(-) create mode 100644 eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs diff --git a/eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs b/eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs new file mode 100644 index 000000000..be33f5ab8 --- /dev/null +++ b/eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs @@ -0,0 +1,156 @@ +//! 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()) + } +} + +fn cascade() -> 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"), + ]); + Cascade::new( + Arc::new(LruCache::new(1024)), + Arc::new(KvStage::new(kv)), + Arc::new(graph), + Arc::new(NeuralStage::new(Box::new(Nothing))), + ) + .with_meter(Arc::from(detect_in_process())) + .without_memoization() +} + +/// 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(); + 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. + timed(&c, "neural (a stage that does nothing)", &Query::new("nothing knows this"), 2_000).await; + misses(&c, 100_000).await; + std::thread::sleep(Duration::from_millis(1)); +} diff --git a/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs b/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs index a063a5ec5..6c8c0f57f 100644 --- a/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs +++ b/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs @@ -1,15 +1,53 @@ //! EOC cascade — the four-stage memoizing pipeline. //! //! `Cascade::resolve` walks `cache → kv → graph → neural` in order. The -//! first stage to return `Some(Response)` wins. Its joule cost combines the -//! cost the stage reported with what the cascade's meter read over the whole -//! call, when that meter is live; it is `Measured` only when every -//! micro-joule of it was read from a live counter. Any micro-joule that was -//! estimated makes the whole an estimate. +//! first stage to return `Some(Response)` wins. +//! +//! # What a resolve is charged +//! +//! The rule, as the owner set it: a live counter's figure is `Measured`, and +//! nothing that was not read from one may be. The cascade keeps it by using +//! its meter only where the meter can resolve the work. +//! +//! **A resolve the neural stage answers is read.** The meter is read once the +//! three deterministic tiers have missed, and again after the neural stage +//! and the memoizing write: two reads, one span. [`cascade_cost`] combines +//! that span with the cost the stage reported, and the total is `Measured` +//! only when every micro-joule of it was read from a live counter. The three +//! misses are spent before the opening read, so they are not in the figure. +//! They are lookups of under a microsecond each, well under what the span +//! holds of the meter itself, its closing read's own wake-up. So the measured +//! figure is the neural stage's energy, the memoizing write's and the meter's +//! own closing cost: over the neural stage's energy, never under it. +//! +//! **A resolve a deterministic tier answers (cache, kv, graph) reads no +//! meter**, and is never `Measured`. A read of a counter costs more than the +//! hit it would bracket, and what such a span reads is the read. On an M3 +//! Ultra (macOS 27.0.1), with this process's own energy counter, whose every +//! read first sees the reading thread leave its core: a resolve answered by +//! the cache took 164 µs at the median with a span around it, against well +//! under a microsecond for the lookup, and the span read 13 to 28 µJ, 14 µJ +//! on average over 10,000 hits, the same as a span around a neural stage that +//! does nothing (17 to 19 µJ). That figure is the meter's own cost, not the +//! hit's; tagged `Measured` and sealed in the hit's receipt it was a true +//! reading booked to the wrong work. So a hit is charged a figure calibrated +//! on this host instead ([`HitCalibration`]): the first time a tier answers, +//! the cascade repeats that lookup back to back inside one span, enough times +//! that the span is large against the meter's own cost, and keeps the span's +//! reading per hit. Every later hit on that tier is charged that figure, with +//! no read. It is `Estimated`: it was read by this counter, on this host, but +//! over a batch, not over this resolve. A per-operation figure for work +//! shorter than the meter can resolve is sound only as the mean of many +//! operations in one span, and a mean is not a reading of any one of them. +//! +//! With no live counter — the cascade was given none, a read failed, or the +//! counter did not advance — there is no calibration and no span, and every +//! resolve keeps the cost its stage reported, figure and tag. #![forbid(unsafe_code)] -use std::sync::Arc; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; use eoc_cache::{LruCache, Stage}; use eoc_core::{JouleCost, JouleSource, Query, Response}; @@ -18,6 +56,193 @@ use eoc_kv::KvStage; use eoc_meter::{JouleCounter, StubCounter}; use eoc_neural::NeuralStage; +/// How many times the meter's own cost a calibration span must read before +/// its reading is taken as the batch's: the meter's share of the per-hit +/// figure is then a fiftieth at most. +const SPAN_TO_EMPTY: u64 = 50; + +/// The least a calibration span must read, in micro-joules, whatever the +/// meter's own cost. The per-hit figure is the span's whole micro-joules over +/// the batch, so a span of 200 µJ keeps the counter's own rounding under +/// half a percent of it. +const SMALLEST_SPAN_MICROJOULES: u64 = 200; + +/// How much larger each calibration batch is than the one before it. +const BATCH_GROWTH: u64 = 8; + +/// The wall time the second batch aims at: the first batch is one lookup, +/// timed, and the second is as many as fill this, so a lookup of well under +/// a microsecond does not climb to a readable batch eight at a time. +const SECOND_BATCH: Duration = Duration::from_micros(100); + +/// The wall time after which a calibration starts no further batch. It keeps +/// the largest span it read by then, whatever that span's size against the +/// meter's own cost ([`HitCalibration`] reports both). +const CALIBRATION_CAP: Duration = Duration::from_millis(40); + +/// How long a counter may hold over back-to-back reads before a calibration +/// takes it for one that is not live. +const HELD_AFTER: Duration = Duration::from_millis(2); + +/// The largest batch a calibration runs, whatever the clock says. +const LARGEST_BATCH: u64 = 1 << 22; + +/// What one hit on a deterministic tier costs on this host: the cascade's +/// meter read over a batch of that tier's lookup, divided by the batch. +/// +/// Each lookup in the batch is what a resolve answered by that tier does once +/// the cheaper tiers have missed: the tier's own lookup of the query, and the +/// minting of the response the cascade returns for it. The misses of the +/// cheaper tiers before it and the memoizing write after it are not in the +/// figure. Nor is the meter's own cost taken out of it: the span holds its +/// closing read, as every span does, so the figure is over the batch's energy +/// by at most `empty_span_microjoules` in `span_microjoules`. +/// +/// The figure is that of the query the tier first answered (or the one +/// [`Cascade::calibrate`] was given), charged to every hit on the tier: a +/// lookup of another key, or a longer scan, costs what it costs, and is +/// charged this. +/// +/// The counter reads its whole domain over the batch. On this process's own +/// counter that is every thread of the process, so work on other threads +/// during the batch is in the figure; on a whole-machine counter, so is the +/// machine's. Calibrate on a quiet process, read the record, and call +/// [`Cascade::calibrate`] again if it was not. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct HitCalibration { + /// The span's reading per lookup, in nanojoules, to the nearest one. + pub per_hit_nanojoules: u64, + /// How many times the lookup ran inside the span. + pub batch: u64, + /// What the meter read over the batch, in micro-joules: one span, two + /// reads. + pub span_microjoules: u64, + /// What the meter read over no work at all, in micro-joules: the first + /// advance it showed across back-to-back reads. It is the meter's own + /// cost in a span, and the batch was grown until its span read fifty + /// times this, or until the time allowed ran out. + pub empty_span_microjoules: u64, + /// The wall time the calibration took, its reads and every batch. + pub took: Duration, +} + +/// Why a tier has no [`HitCalibration`]. Its hits keep the cost the stage +/// reported. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NoCalibration { + /// The neural stage is not calibrated: a resolve it answers is read. + NotADeterministicTier, + /// The tier has not answered yet. + NotYet, + /// A resolve is calibrating the tier now. + InProgress, + /// Switched off for this tier ([`Cascade::without_hit_calibration`]). + Disabled, + /// The cascade was given no meter: it holds the stub. + NoMeter, + /// This target has no monotonic clock to bound the calibration by. + NoClock, + /// A read of the meter failed. + ReadFailed, + /// The counter did not advance: over back-to-back reads, or over any + /// batch. + CounterHeld, + /// The lookup stopped answering while it was repeated. + LookupStoppedAnswering, +} + +impl std::fmt::Display for NoCalibration { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::NotADeterministicTier => "the neural stage is read, not calibrated", + Self::NotYet => "the tier has not answered yet", + Self::InProgress => "the tier is being calibrated", + Self::Disabled => "calibration is switched off for the tier", + Self::NoMeter => "the cascade was given no meter", + Self::NoClock => "no monotonic clock on this target", + Self::ReadFailed => "a read of the meter failed", + Self::CounterHeld => "the counter did not advance", + Self::LookupStoppedAnswering => "the lookup stopped answering while it was repeated", + }) + } +} + +/// A deterministic tier: one whose hits are calibrated, not read. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Tier { + Cache, + Kv, + Graph, +} + +impl Tier { + fn of(stage: eoc_core::Stage) -> Option { + match stage { + eoc_core::Stage::Cache => Some(Self::Cache), + eoc_core::Stage::Kv => Some(Self::Kv), + eoc_core::Stage::Graph => Some(Self::Graph), + eoc_core::Stage::Neural => None, + } + } + + fn stage(self) -> eoc_core::Stage { + match self { + Self::Cache => eoc_core::Stage::Cache, + Self::Kv => eoc_core::Stage::Kv, + Self::Graph => eoc_core::Stage::Graph, + } + } +} + +/// What the cascade holds about one deterministic tier's hits. +#[derive(Debug)] +struct TierHits { + calibration: Result, + /// The nanojoules calibrated hits have accrued that no charge has + /// carried yet: under 1,000 after every charge. + carry_nanojoules: u64, +} + +impl TierHits { + fn new() -> Self { + Self { calibration: Err(NoCalibration::NotYet), carry_nanojoules: 0 } + } + + /// The whole micro-joules one more hit is charged. A hit accrues the + /// tier's per-hit nanojoules; what does not make a micro-joule is carried + /// to the next hit. So the micro-joules charged over any run of hits sum + /// to the nanojoules those hits accrued, to within the one micro-joule + /// still carried — never a round-up of each hit, and never a string of + /// zeros for a figure under a micro-joule. Nothing where the tier has no + /// calibration. + fn charge(&mut self) -> u64 { + let Ok(calibration) = self.calibration else { + return 0; + }; + let accrued = self.carry_nanojoules.saturating_add(calibration.per_hit_nanojoules); + self.carry_nanojoules = accrued % 1_000; + accrued / 1_000 + } +} + +/// A monotonic clock: the time since some fixed instant, or `None` where the +/// target has none. +type Clock = Arc Option + Send + Sync>; + +/// The time since this process first asked. +#[cfg(not(target_arch = "wasm32"))] +fn monotonic() -> Option { + static FIRST: std::sync::OnceLock = std::sync::OnceLock::new(); + Some(FIRST.get_or_init(std::time::Instant::now).elapsed()) +} + +/// `wasm32-unknown-unknown` has no clock (`Instant::now` panics there), and +/// no energy counter either: nothing is calibrated. +#[cfg(target_arch = "wasm32")] +fn monotonic() -> Option { + None +} + /// The four-stage cascade. pub struct Cascade { cache: Arc, @@ -25,8 +250,14 @@ pub struct Cascade { graph: Arc, neural: Arc, meter: Arc, + /// Whether a meter was given ([`Self::with_meter`]): without one, nothing + /// is calibrated and no read is spent finding the stub does not move. + metered: bool, /// Whether to memoize neural/graph/kv answers back into the cache. memoize: bool, + /// What a hit costs on each deterministic tier: cache, kv, graph. + hits: [Mutex; 3], + clock: Clock, } impl Cascade { @@ -43,13 +274,22 @@ impl Cascade { graph, neural, meter: Arc::new(StubCounter), + metered: false, memoize: true, + hits: [Mutex::new(TierHits::new()), Mutex::new(TierHits::new()), Mutex::new(TierHits::new())], + clock: Arc::new(monotonic), } } /// Swap in a different joule counter. + /// + /// The counter is read twice around the neural stage of a resolve that + /// reaches it, and around the batches that calibrate each deterministic + /// tier the first time it answers ([`HitCalibration`]); never around a + /// hit. pub fn with_meter(mut self, meter: Arc) -> Self { self.meter = meter; + self.metered = true; self } @@ -59,6 +299,24 @@ impl Cascade { self } + /// Never calibrate `stage`'s hits: they keep the cost the stage reports. + /// + /// A calibration repeats the tier's lookup of one query, up to the time + /// allowed (tens of milliseconds). That is safe for the stages this + /// workspace ships: the cache's lookup only refreshes the entry's + /// recency, which the hit before it already did; the KV stage's reads its + /// backend ([`eoc_kv::KvBackend::get`], and `embeddings` for a similarity + /// match); the graph stage's scans its triples. A KV backend whose `get` + /// is not a plain read — one that bills, rate-limits or counts each call + /// — should not be repeated, and its tier is switched off here. No effect + /// for the neural stage, which is never calibrated. + pub fn without_hit_calibration(self, stage: eoc_core::Stage) -> Self { + if let Some(tier) = Tier::of(stage) { + self.hits(tier).calibration = Err(NoCalibration::Disabled); + } + self + } + /// Borrow the cache (for warmup / wiring). pub fn cache(&self) -> &LruCache { &self.cache @@ -80,38 +338,81 @@ impl Cascade { self.meter.as_ref() } + /// What a hit on `stage` is charged on this host, or why it is charged + /// nothing beyond the stage's own cost. + pub fn calibration(&self, stage: eoc_core::Stage) -> Result { + match Tier::of(stage) { + Some(tier) => self.hits(tier).calibration, + None => Err(NoCalibration::NotADeterministicTier), + } + } + + /// Calibrate, now, the deterministic tier that answers `q`, whether or + /// not it was calibrated before: the figure replaces the one held, and + /// the next hit on that tier is charged it. For a host whose state has + /// changed since the first hit, or to calibrate before the first resolve + /// rather than inside it. + /// + /// Returns the stage that answered and its calibration, or why there is + /// none; `None` when no deterministic tier answers `q`. Nothing is + /// memoized and no resolve is charged. A tier switched off by + /// [`Self::without_hit_calibration`] stays off. + pub async fn calibrate(&self, q: &Query) -> Option<(eoc_core::Stage, Result)> { + let mut answering = None; + for tier in [Tier::Cache, Tier::Kv, Tier::Graph] { + if self.lookup(tier, q).await.is_some() { + answering = Some(tier); + break; + } + } + let tier = answering?; + let held = self.hits(tier).calibration; + if held == Err(NoCalibration::Disabled) { + return Some((tier.stage(), held)); + } + let calibration = self.calibrate_tier(tier, q).await; + self.hits(tier).calibration = calibration; + Some((tier.stage(), calibration)) + } + /// Resolve a query through the cascade. /// /// On a miss at one stage, falls through to the next. The neural /// stage always answers, so this function always returns `Some` if /// the neural stage's backend is well-behaved. We still return /// `Option` for symmetry with the `Stage` trait. + /// + /// The meter is read only when the walk reaches the neural stage; a + /// resolve a deterministic tier answers reads none (see the crate + /// documentation), apart from the one that first calibrates its tier. pub async fn resolve(&self, q: Query) -> Response { - let start = self.meter.read_microjoules().ok(); + // Walk the deterministic tiers in order. None of them is read. + if let Some(r) = self.cache.try_resolve(&q).await { + return self.hit(Tier::Cache, &q, r).await; + } + if let Some(r) = self.kv.try_resolve(&q).await { + self.maybe_memoize(&q, &r); + return self.hit(Tier::Kv, &q, r).await; + } + if let Some(r) = self.graph.try_resolve(&q).await { + self.maybe_memoize(&q, &r); + return self.hit(Tier::Graph, &q, r).await; + } - // Walk stages in order. - let result = if let Some(r) = self.cache.try_resolve(&q).await { - r - } else if let Some(r) = self.kv.try_resolve(&q).await { - self.maybe_memoize(&q, &r); - r - } else if let Some(r) = self.graph.try_resolve(&q).await { - self.maybe_memoize(&q, &r); - r - } else { - // Neural always answers. - let r = self - .neural - .try_resolve(&q) - .await - .expect("neural stage must always return Some"); - self.maybe_memoize(&q, &r); - r - }; + // All three missed: only now is the meter read, so the span is the + // neural stage's, and the misses, spent before it, are not in it. + let start = self.meter.read_microjoules().ok(); + // Neural always answers. + let r = self + .neural + .try_resolve(&q) + .await + .expect("neural stage must always return Some"); + self.maybe_memoize(&q, &r); // Re-derive the response with the cascade-attributed joule cost: the // stage's own cost, combined with the meter's span when it is live. - attach_cascade_cost(result, start, self.meter.as_ref()) + attach_cascade_cost(r, start, self.meter.as_ref()) } fn maybe_memoize(&self, _q: &Query, r: &Response) { @@ -125,23 +426,147 @@ impl Cascade { self.cache.insert_response(r.query_id, cached); } } + + fn hits(&self, tier: Tier) -> MutexGuard<'_, TierHits> { + // The state is two plain figures, whole at every point a holder + // could have panicked. + self.hits[tier as usize].lock().unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// `tier`'s own lookup of `q`. + async fn lookup(&self, tier: Tier, q: &Query) -> Option { + match tier { + Tier::Cache => self.cache.try_resolve(q).await, + Tier::Kv => self.kv.try_resolve(q).await, + Tier::Graph => self.graph.try_resolve(q).await, + } + } + + /// The response for a resolve `tier` answered with `r`: charged the + /// tier's calibrated share on top of the stage's own cost, an estimate, + /// with no read of the meter. The first hit on a tier calibrates it. + async fn hit(&self, tier: Tier, q: &Query, r: Response) -> Response { + let first = { + let mut hits = self.hits(tier); + let first = hits.calibration == Err(NoCalibration::NotYet); + if first { + // Claimed before the lock is dropped, so a resolve on + // another thread does not calibrate the tier a second time: + // it is charged the stage's own cost until this one is done. + hits.calibration = Err(NoCalibration::InProgress); + } + first + }; + if first { + let calibration = self.calibrate_tier(tier, q).await; + self.hits(tier).calibration = calibration; + } + let share = self.hits(tier).charge(); + // Rebuild the response so the receipt covers the final joule cost. + Response::new(r.query_id, r.payload, r.stage, hit_cost(r.joule_cost, share)) + } + + /// What one hit on `tier` costs on this host, from one span of the meter + /// around a batch of `tier`'s lookup of `q`. + /// + /// First the meter's own cost: back-to-back reads until the counter + /// moves. A counter that holds for [`HELD_AFTER`] is not live. Then the + /// batch, grown until its span reads [`SPAN_TO_EMPTY`] times that (and + /// [`SMALLEST_SPAN_MICROJOULES`] at least), each batch inside one span + /// of its own; no further batch is started once [`CALIBRATION_CAP`] of + /// wall time has gone, or once another would not fit in what is left of + /// it, and the largest span read by then is kept. The stage's lookups + /// here do not yield, so the batch runs on the calling thread between + /// the span's two reads. + async fn calibrate_tier(&self, tier: Tier, q: &Query) -> Result { + if !self.metered { + return Err(NoCalibration::NoMeter); + } + let clock = self.clock.as_ref(); + let started = clock().ok_or(NoCalibration::NoClock)?; + let since = |then: Duration| clock().map_or(Duration::MAX, |now| now.saturating_sub(then)); + let meter = self.meter.as_ref(); + let read = || meter.read_microjoules().map_err(|_| NoCalibration::ReadFailed); + + let base = read()?; + let empty = loop { + let now = read()?; + if now > base { + break now - base; + } + if since(started) >= HELD_AFTER { + return Err(NoCalibration::CounterHeld); + } + }; + let enough = empty.saturating_mul(SPAN_TO_EMPTY).max(SMALLEST_SPAN_MICROJOULES); + + let mut batch = 1u64; + let mut largest = None; + loop { + let open = read()?; + let began = clock().ok_or(NoCalibration::NoClock)?; + for _ in 0..batch { + let Some(r) = self.lookup(tier, q).await else { + return Err(NoCalibration::LookupStoppedAnswering); + }; + // The response the cascade mints for a hit, as `hit` does. + std::hint::black_box(Response::new(r.query_id, r.payload, r.stage, r.joule_cost)); + } + let lookups = since(began); + let close = read()?; + if close > open { + largest = Some((batch, close - open)); + if close - open >= enough { + break; + } + } + let each = (lookups.as_nanos() / u128::from(batch)).max(1); + let fit = CALIBRATION_CAP.saturating_sub(since(started)).as_nanos() / each; + let second = u64::try_from(SECOND_BATCH.as_nanos() / each).unwrap_or(u64::MAX); + let wanted = batch.saturating_mul(BATCH_GROWTH).max(second); + let next = wanted.min(u64::try_from(fit).unwrap_or(u64::MAX)).min(LARGEST_BATCH); + if next <= batch { + break; + } + batch = next; + } + let (batch, span) = largest.ok_or(NoCalibration::CounterHeld)?; + let per_hit = (u128::from(span) * 1_000 + u128::from(batch) / 2) / u128::from(batch); + Ok(HitCalibration { + per_hit_nanojoules: u64::try_from(per_hit).unwrap_or(u64::MAX), + batch, + span_microjoules: span, + empty_span_microjoules: empty, + took: since(started), + }) + } } +/// The response for a resolve the neural stage answered: its cost combined +/// with the span the cascade's meter read around the stage, from `start`, +/// read before it, to a second read taken here. No second read is taken +/// where the first failed: no span could come of it. fn attach_cascade_cost( response: Response, start: Option, meter: &dyn JouleCounter, ) -> Response { - let end = meter.read_microjoules().ok(); + let end = start.and_then(|_| meter.read_microjoules().ok()); let joule_cost = cascade_cost(response.joule_cost, live_span(start, end)); // Rebuild the response so the receipt covers the final joule cost. Response::new(response.query_id, response.payload, response.stage, joule_cost) } -/// What the cascade's meter read over one resolve: the delta between its two -/// reads, only when both succeeded and the counter advanced. A counter that -/// read but did not move — the stub, a frozen rail — measured nothing, and a -/// failed read leaves no span; neither is a figure. +/// What the cascade's meter read around the neural stage of one resolve: the +/// delta between its two reads, only when both succeeded and the counter +/// advanced. A counter that read but did not move — the stub, a frozen rail +/// — measured nothing, and a failed read leaves no span; neither is a figure. +/// +/// The first read is taken after the cache, the KV stage and the graph have +/// missed, and the second after the neural stage and the memoizing write. So +/// the span holds the neural stage, that write, and the meter's own closing +/// cost (on this process's counter, the wake-up of the read that closes it); +/// it does not hold the three misses, which were spent before it opened. fn live_span(start: Option, end: Option) -> Option { match (start, end) { (Some(s), Some(e)) if e > s => Some(e - s), @@ -149,8 +574,9 @@ fn live_span(start: Option, end: Option) -> Option { } } -/// The joule cost of one resolve, from the answering stage's own cost and -/// the cascade meter's live span over the whole resolve, if there was one. +/// The joule cost of one resolve the neural stage answered, from the stage's +/// own cost and the cascade meter's live span around the stage, if there was +/// one. /// /// It is `Measured` only when every micro-joule of it was read from a live /// counter; any micro-joule that was estimated makes the whole an estimate. @@ -165,8 +591,22 @@ fn live_span(start: Option, end: Option) -> Option { /// - A live span and an estimated stage: the estimate is energy no counter /// read, possibly spent elsewhere (a remote backend), so it is added to the /// span and the total is an estimate. An estimated zero, which is what a -/// cache hit costs, adds nothing: the cost is the span's reading alone, and -/// measured. +/// stage reports when it has no figure of its own, adds nothing: the cost +/// is the span's reading alone, and measured. +/// +/// A measured figure is what the counter read from the opening read to the +/// closing one: the neural stage, the memoizing write and the meter's own +/// closing cost. It leaves out the three deterministic misses before the +/// opening read, which are smaller than that closing cost, so it is not +/// under the neural stage's energy. +/// +/// This rule is never applied to a resolve a deterministic tier answered. +/// It once was, and its last case then read: an estimated zero on a live +/// span, a cache hit, is the span's reading, measured. With a counter whose +/// read costs more than the hit, that reading was the meter's own cost (on +/// an M3 Ultra, 14 µJ on average over 10,000 cache hits, each bracketed by +/// two reads taking 164 µs between them), sealed as the hit's. A hit is +/// charged by [`hit_cost`] instead, and no meter is read for it. fn cascade_cost(stage: JouleCost, span: Option) -> JouleCost { let Some(span) = span else { return stage; @@ -178,6 +618,27 @@ fn cascade_cost(stage: JouleCost, span: Option) -> JouleCost { } } +/// The joule cost of one resolve a deterministic tier answered: the cost the +/// stage reported plus the tier's calibrated share for this hit, in whole +/// micro-joules (nothing, where the tier has no calibration). Always an +/// estimate. +/// +/// No counter was read for this resolve, so none of it is measured. Not the +/// calibrated share: it was read on this host, by the cascade's own counter, +/// but over a batch of the lookup, and a mean is not a reading of this hit. +/// And not a figure the stage's response carries tagged measured: a cached +/// response replays what was read for an earlier resolve, not this one. +/// +/// The stage's own figure is added, not replaced, for the reason +/// [`cascade_cost`] adds an estimate to a span: it may be energy the counter +/// never saw. Where it is the stage's guess at the very lookup the +/// calibration read (the reference KV and graph stages report a fixed figure +/// for theirs), the two overlap; the cascade cannot tell, and an estimate +/// that is over says so by its tag. +fn hit_cost(stage: JouleCost, share_microjoules: u64) -> JouleCost { + JouleCost::estimated(stage.microjoules.saturating_add(share_microjoules)) +} + #[cfg(test)] mod tests { use super::*; @@ -186,6 +647,15 @@ mod tests { use eoc_graph::{GraphStage, Triple}; use eoc_kv::{KvBackend, KvStage, MemoryKvBackend}; use eoc_neural::{EchoBackend, NeuralStage}; + use std::sync::atomic::{AtomicU64, Ordering}; + + impl Cascade { + /// Bound calibrations by a scripted clock instead of this host's. + fn with_clock(mut self, clock: Clock) -> Self { + self.clock = clock; + self + } + } fn fixture() -> Cascade { let cache = Arc::new(LruCache::new(64)); @@ -200,6 +670,33 @@ mod tests { Cascade::new(cache, kv, graph, neural) } + /// A query each deterministic tier of [`fixture`] answers, once the + /// cache has been handed its own. + fn answered_by(cascade: &Cascade, stage: StageKind) -> Query { + match stage { + StageKind::Cache => { + let q = Query::new("seen before"); + cascade.cache().insert(&q, "answer"); + q + } + StageKind::Kv => Query::new("known key"), + StageKind::Graph => Query::new("Mars is the fourth planet from what?"), + StageKind::Neural => Query::new("totally novel question"), + } + } + + /// What each deterministic stage of [`fixture`] reports for a hit. + fn own_cost(stage: StageKind) -> u64 { + match stage { + StageKind::Cache => 0, + StageKind::Kv => 50, + StageKind::Graph => 500, + StageKind::Neural => 50_000_000, + } + } + + const TIERS: [StageKind; 3] = [StageKind::Cache, StageKind::Kv, StageKind::Graph]; + #[tokio::test] async fn cascade_uses_kv_when_it_can() { let cascade = fixture(); @@ -240,21 +737,39 @@ mod tests { /// A counter that reads the scripted values in turn, then the last one /// again; `None` is a failed read. One instance can be shared by the - /// cascade and a stage, as one host counter would be. + /// cascade and a stage, as one host counter would be. It counts its + /// reads, and keeps a clock that only its reads move, a millisecond each, + /// so a calibration against it ends by the script and not by this host's + /// time. A lookup takes none of that clock's time: a script must end a + /// calibration by its reads — the span it scripts is enough, the counter + /// holds from the start, or a read fails. struct Scripted { reads: std::sync::Mutex>>, + taken: AtomicU64, } impl Scripted { fn new(reads: &[Option]) -> Arc { Arc::new(Self { reads: std::sync::Mutex::new(reads.iter().copied().collect()), + taken: AtomicU64::new(0), }) } + + /// How many reads have been taken. + fn taken(&self) -> u64 { + self.taken.load(Ordering::SeqCst) + } + + fn clock(self: &Arc) -> Clock { + let scripted = self.clone(); + Arc::new(move || Some(Duration::from_millis(scripted.taken()))) + } } impl JouleCounter for Scripted { fn read_microjoules(&self) -> eoc_core::Result { + self.taken.fetch_add(1, Ordering::SeqCst); let mut r = self.reads.lock().unwrap(); let next = if r.len() > 1 { r.pop_front().unwrap() @@ -268,6 +783,11 @@ mod tests { } } + /// A script that calibrates a tier in four reads: the counter moves 10 µJ + /// over no work, so a span of 500 µJ is enough, and the first batch, one + /// lookup, reads 700. Every hit is then charged 700 µJ. + const CALIBRATES_AT_700: [Option; 4] = [Some(1_000), Some(1_010), Some(2_000), Some(2_700)]; + /// A neural backend that meters itself the way the local backends do: it /// reads its counter before and after, and reports the delta as measured /// when the counter advanced. @@ -288,14 +808,21 @@ mod tests { } } - fn metered(meter: Arc, neural: Box) -> Cascade { + /// [`fixture`]'s stages around a given neural backend, metered by a + /// scripted counter and bounded by its clock. + fn metered(meter: &Arc, neural: Box) -> Cascade { + let kv_backend = Box::new(MemoryKvBackend::new()); + kv_backend.put("known key", b"known value".to_vec()); + let graph = GraphStage::new(); + graph.insert(Triple::new("Mars", "fourth planet from", "the Sun")); Cascade::new( Arc::new(LruCache::new(8)), - Arc::new(KvStage::new(Box::new(MemoryKvBackend::new()))), - Arc::new(GraphStage::new()), + Arc::new(KvStage::new(kv_backend)), + Arc::new(graph), Arc::new(NeuralStage::new(neural)), ) - .with_meter(meter) + .with_meter(meter.clone()) + .with_clock(meter.clock()) } /// No meter is read for a cache hit, so it claims no measurement: not a @@ -316,12 +843,12 @@ mod tests { assert_eq!(inserted.joule_cost, JouleCost::estimated(0)); } - /// A live counter advanced 500 µJ over the resolve, and the stage reported - /// a 50 J estimate. The total holds an estimate, so it is one. + /// A live counter advanced 500 µJ over the neural stage, and the stage + /// reported a 50 J estimate. The total holds an estimate, so it is one. #[tokio::test] async fn an_estimate_plus_a_live_span_is_an_estimate() { let meter = Scripted::new(&[Some(1_000), Some(1_500)]); - let cascade = metered(meter, Box::new(EchoBackend::new().with_cost(50_000_000))); + let cascade = metered(&meter, Box::new(EchoBackend::new().with_cost(50_000_000))); let r = cascade.resolve(Query::new("totally novel question")).await; assert_eq!(r.stage, StageKind::Neural); assert_eq!(r.joule_cost, JouleCost::estimated(50_000_500)); @@ -335,39 +862,28 @@ mod tests { #[tokio::test] async fn a_stage_metered_on_the_cascade_counter_is_counted_once() { let meter = Scripted::new(&[Some(1_000), Some(1_100), Some(1_400), Some(1_500)]); - let shared: Arc = meter; - let cascade = metered(shared.clone(), Box::new(SelfMetered { meter: shared })); + let cascade = metered(&meter, Box::new(SelfMetered { meter: meter.clone() })); let r = cascade.resolve(Query::new("totally novel question")).await; assert_eq!(r.stage, StageKind::Neural); assert_eq!(r.joule_cost, JouleCost::measured(500)); } - /// A cache hit adds no figure of its own, so on a live counter the cost is - /// the span's reading alone, every micro-joule of it read: measured. - #[tokio::test] - async fn a_cache_hit_on_a_live_counter_is_the_span_reading() { - let meter = Scripted::new(&[Some(2_000), Some(2_250)]); - let cascade = metered(meter, Box::new(EchoBackend::new())); - let q = Query::new("seen before"); - cascade.cache().insert(&q, "answer"); - let r = cascade.resolve(q).await; - assert_eq!(r.stage, StageKind::Cache); - assert_eq!(r.joule_cost, JouleCost::measured(250)); - } - /// A counter that reads but does not move, or fails a read, measured - /// nothing: the stage's own cost stands, figure and tag. + /// nothing: the stage's own cost stands, figure and tag. A failed first + /// read is not followed by a second. #[tokio::test] async fn a_span_that_is_not_live_adds_nothing() { - for reads in [vec![Some(7_000)], vec![None, Some(9_000)], vec![Some(9_000), None]] { + for (reads, taken) in [(vec![Some(7_000)], 2), (vec![None, Some(9_000)], 1), (vec![Some(9_000), None], 2)] { let meter = Scripted::new(&reads); - let cascade = metered(meter, Box::new(EchoBackend::new().with_cost(42))); + let cascade = metered(&meter, Box::new(EchoBackend::new().with_cost(42))); let r = cascade.resolve(Query::new("totally novel question")).await; assert_eq!(r.joule_cost, JouleCost::estimated(42), "reads {reads:?}"); + assert_eq!(meter.taken(), taken, "reads {reads:?}"); } } - /// Every branch of the rule, on the figures alone. + /// Every branch of the rule for a resolve the neural stage answered, on + /// the figures alone. #[test] fn the_cost_is_measured_only_when_every_microjoule_was_read() { let e = JouleCost::estimated; @@ -390,4 +906,590 @@ mod tests { assert_eq!(live_span(Some(5), None), None); assert_eq!(live_span(Some(5), Some(8)), Some(3)); } + + /// Every branch of the rule for a resolve a deterministic tier answered: + /// the stage's figure plus the calibrated share, and an estimate whatever + /// the stage's own tag was. + #[test] + fn a_hits_cost_is_an_estimate_whatever_it_is_made_of() { + let e = JouleCost::estimated; + let m = JouleCost::measured; + assert_eq!(hit_cost(e(0), 0), e(0), "no calibration, a cache hit"); + assert_eq!(hit_cost(e(50), 0), e(50), "no calibration, a KV hit"); + assert_eq!(hit_cost(e(0), 3), e(3), "the calibrated share alone"); + assert_eq!(hit_cost(e(50), 3), e(53), "the stage's own estimate is added, not replaced"); + assert_eq!(hit_cost(m(9), 0), e(9), "a replayed reading is not a reading of this resolve"); + assert_eq!(hit_cost(m(9), 3), e(12)); + assert_eq!(hit_cost(e(u64::MAX), 3), e(u64::MAX)); + } + + /// The carry, on the figures alone: after any number of hits the + /// micro-joules charged are the whole micro-joules in the nanojoules + /// accrued, the rest is carried, and the carry stays under a micro-joule. + #[test] + fn the_carry_conserves_every_nanojoule() { + for per_hit in [0u64, 1, 137, 999, 1_000, 1_001, 2_345, 123_456_789] { + let mut hits = TierHits::new(); + hits.calibration = Ok(HitCalibration { + per_hit_nanojoules: per_hit, + batch: 1, + span_microjoules: 1, + empty_span_microjoules: 1, + took: Duration::ZERO, + }); + let mut charged = 0u64; + for n in 1..=5_000u64 { + charged += hits.charge(); + assert_eq!(charged, n * per_hit / 1_000, "{per_hit} nJ a hit, after {n} hits"); + assert_eq!(hits.carry_nanojoules, n * per_hit % 1_000, "{per_hit} nJ a hit, after {n} hits"); + } + } + // No calibration, no charge, and the carry is left as it stands. + let mut hits = TierHits::new(); + hits.carry_nanojoules = 600; + for why in [NoCalibration::NotYet, NoCalibration::InProgress, NoCalibration::CounterHeld] { + hits.calibration = Err(why); + assert_eq!((hits.charge(), hits.carry_nanojoules), (0, 600), "{why:?}"); + } + } + + /// A host in miniature, for calibrations that must come out the same on + /// every run: one cumulative energy count and one clock, which a read of + /// the meter and a KV lookup move, each by a fixed amount, and nothing + /// else does. It is the cascade's meter, its clock and its KV backend, + /// and it keeps what it saw in order. Only the KV tier can be calibrated + /// against it: a cache or graph lookup moves neither count. + struct Host { + nanojoules: AtomicU64, + nanos: AtomicU64, + /// What a read of the meter costs: nanojoules, nanoseconds. + read: (u64, u64), + /// What a KV lookup costs: nanojoules, nanoseconds. + get: (AtomicU64, u64), + /// Once this many reads of the meter have been taken the count + /// holds, whatever is spent. + holds_after: Option, + store: MemoryKvBackend, + seen: Mutex>, + } + + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum Seen { + /// A read of the meter, and the micro-joules it returned. + Read(u64), + /// A KV lookup. + Get, + /// The neural stage. + Infer, + } + + impl Host { + /// A host whose meter costs 10 µJ and 80 µs a read, and whose KV + /// lookup costs `get_nanojoules` and `get_nanos`. + fn new(get_nanojoules: u64, get_nanos: u64) -> Arc { + Self::holding_after(get_nanojoules, get_nanos, None) + } + + fn holding_after(get_nanojoules: u64, get_nanos: u64, holds_after: Option) -> Arc { + let store = MemoryKvBackend::new(); + store.put("known key", b"known value".to_vec()); + Arc::new(Self { + nanojoules: AtomicU64::new(1_000_000), + nanos: AtomicU64::new(0), + read: (10_000, 80_000), + get: (AtomicU64::new(get_nanojoules), get_nanos), + holds_after, + store, + seen: Mutex::new(Vec::new()), + }) + } + + fn spend(&self, nanojoules: u64, nanos: u64) { + if self.holds_after.is_none_or(|n| self.reads() < n) { + self.nanojoules.fetch_add(nanojoules, Ordering::SeqCst); + } + self.nanos.fetch_add(nanos, Ordering::SeqCst); + } + + fn elapsed(&self) -> Duration { + Duration::from_nanos(self.nanos.load(Ordering::SeqCst)) + } + + fn clock(self: &Arc) -> Clock { + let host = self.clone(); + Arc::new(move || Some(host.elapsed())) + } + + fn seen(&self) -> Vec { + self.seen.lock().unwrap().clone() + } + + fn count(&self, which: impl Fn(&Seen) -> bool) -> u64 { + self.seen.lock().unwrap().iter().filter(|s| which(s)).count() as u64 + } + + fn reads(&self) -> u64 { + self.count(|s| matches!(s, Seen::Read(_))) + } + + fn gets(&self) -> u64 { + self.count(|s| *s == Seen::Get) + } + + /// A cascade on this host that does not memoize, so a KV hit stays a + /// KV hit and the cache never answers, around a neural stage that + /// reports `neural`. + fn cascade(self: &Arc, neural: JouleCost) -> Cascade { + Cascade::new( + Arc::new(LruCache::new(8)), + Arc::new(KvStage::new(Box::new(HostKv(self.clone())))), + Arc::new(GraphStage::new()), + Arc::new(NeuralStage::new(Box::new(HostNeural(self.clone(), neural)))), + ) + .with_meter(Arc::new(HostMeter(self.clone()))) + .with_clock(self.clock()) + .without_memoization() + } + } + + struct HostMeter(Arc); + + impl JouleCounter for HostMeter { + /// The read's own cost is spent before the count is taken, so a span + /// holds its closing read, as a span on a real counter does. + fn read_microjoules(&self) -> eoc_core::Result { + self.0.spend(self.0.read.0, self.0.read.1); + let microjoules = self.0.nanojoules.load(Ordering::SeqCst) / 1_000; + self.0.seen.lock().unwrap().push(Seen::Read(microjoules)); + Ok(microjoules) + } + fn name(&self) -> &'static str { + "host" + } + } + + struct HostKv(Arc); + + impl KvBackend for HostKv { + fn get(&self, key: &str) -> Option> { + self.0.seen.lock().unwrap().push(Seen::Get); + self.0.spend(self.0.get.0.load(Ordering::SeqCst), self.0.get.1); + self.0.store.get(key) + } + fn put(&self, key: &str, value: Vec) { + self.0.store.put(key, value) + } + fn embeddings(&self) -> Vec<(Vec, String)> { + self.0.store.embeddings() + } + fn put_embedding(&self, embedding: Vec, key: String) { + self.0.store.put_embedding(embedding, key) + } + } + + struct HostNeural(Arc, JouleCost); + + #[async_trait::async_trait] + impl eoc_neural::NeuralBackend for HostNeural { + async fn infer(&self, q: &Query) -> Response { + self.0.seen.lock().unwrap().push(Seen::Infer); + Response::new(q.id, "inferred".to_string(), StageKind::Neural, self.1) + } + } + + /// The micro-joules of a KV hit's cost that are the calibrated share: + /// the rest is the 50 µJ the KV stage reports for itself. + fn share(r: &Response) -> u64 { + assert_eq!((r.stage, r.joule_cost.source), (StageKind::Kv, JouleSource::Estimated), "{}", r.joule_cost); + r.joule_cost.microjoules - own_cost(StageKind::Kv) + } + + /// The first hit on a tier calibrates it, and after that no hit on the + /// tier reads the meter: not one read in a thousand hits, on any of the + /// three tiers. Every hit is an estimate. + #[tokio::test] + async fn a_hit_reads_no_meter_once_its_tier_is_calibrated() { + for stage in TIERS { + let meter = Scripted::new(&CALIBRATES_AT_700); + let cascade = metered(&meter, Box::new(EchoBackend::new())).without_memoization(); + let q = answered_by(&cascade, stage); + assert_eq!(cascade.calibration(stage), Err(NoCalibration::NotYet)); + let first = cascade.resolve(q.clone()).await; + assert_eq!(first.stage, stage); + assert_eq!(meter.taken(), 4, "{stage}: the calibration's reads"); + assert_eq!(cascade.calibration(stage).map(|c| c.per_hit_nanojoules), Ok(700_000), "{stage}"); + for _ in 0..1_000 { + let r = cascade.resolve(q.clone()).await; + assert_eq!(r.stage, stage); + assert_eq!(r.joule_cost, JouleCost::estimated(own_cost(stage) + 700), "{stage}"); + assert!(r.receipt.verify(&r.query_id, &r.stage, &r.joule_cost, &r.payload)); + } + assert_eq!(first.joule_cost, JouleCost::estimated(own_cost(stage) + 700), "{stage}: the first hit"); + assert_eq!(meter.taken(), 4, "{stage}: a hit read the meter"); + // The other tiers have not answered, and were not calibrated. + for other in TIERS.into_iter().filter(|other| *other != stage) { + assert_eq!(cascade.calibration(other), Err(NoCalibration::NotYet), "{stage}: {other}"); + } + } + + // The same on a host whose every read and lookup is counted. + let host = Host::new(137, 150); + let cascade = host.cascade(JouleCost::estimated(42)); + let q = Query::new("known key"); + cascade.resolve(q.clone()).await; + cascade.calibration(StageKind::Kv).expect("the KV tier is calibrated"); + let (reads, gets) = (host.reads(), host.gets()); + assert!(reads >= 4, "{reads} reads calibrated the tier"); + for _ in 0..1_000 { + cascade.resolve(q.clone()).await; + } + assert_eq!(host.reads(), reads, "a hit read the meter"); + assert_eq!(host.gets(), gets + 1_000, "a hit is one lookup"); + } + + /// A resolve the neural stage answers reads the meter exactly twice: once + /// after the deterministic tiers have missed, once after the stage. What + /// the KV lookup spent before the first read is not in the span. + #[tokio::test] + async fn a_neural_resolve_reads_the_meter_twice_and_only_after_the_walk() { + // A miss in the KV stage here spends 2 mJ, forty times the 50 µJ + // span the two reads then bracket. + let host = Host::new(2_000_000, 150); + let cascade = host.cascade(JouleCost::zero()); + let before = host.nanojoules.load(Ordering::SeqCst) / 1_000; + let r = cascade.resolve(Query::new("nothing knows this")).await; + let seen = host.seen(); + let &[Seen::Get, Seen::Read(open), Seen::Infer, Seen::Read(close)] = seen.as_slice() else { + panic!("the walk, a read, the neural stage, a read: {seen:?}"); + }; + assert_eq!(r.stage, StageKind::Neural); + assert_eq!(r.joule_cost, JouleCost::measured(close - open)); + assert_eq!(close - open, 10, "the span holds its closing read's 10 µJ and nothing of the walk"); + assert_eq!(open - before, 2_010, "the miss's 2 mJ and the opening read were spent before the span"); + + // And an estimating stage: its estimate plus the span, an estimate. + let host = Host::new(2_000_000, 150); + let r = host.cascade(JouleCost::estimated(42)).resolve(Query::new("nothing knows this")).await; + assert_eq!(r.joule_cost, JouleCost::estimated(52)); + assert_eq!(host.reads(), 2); + } + + /// What a calibration records is what it did: the batch is the number of + /// lookups between the two reads of its last span, the span is those two + /// reads' delta, the empty span is the counter's first advance over no + /// work, and the per-hit figure is the span over the batch. The batch + /// was grown until its span read fifty times the empty one. + #[tokio::test] + async fn a_calibration_records_the_batch_it_ran_and_the_span_it_read() { + for get_nanojoules in [137, 2_345, 40_000] { + let host = Host::new(get_nanojoules, 150); + let cascade = host.cascade(JouleCost::estimated(42)); + cascade.resolve(Query::new("known key")).await; + let c = cascade.calibration(StageKind::Kv).expect("calibrated"); + let seen = host.seen(); + let reads: Vec = (0..seen.len()).filter(|i| matches!(seen[*i], Seen::Read(_))).collect(); + let at = |i: usize| match seen[i] { + Seen::Read(microjoules) => microjoules, + other => panic!("{other:?} is not a read"), + }; + // The hit's own lookup comes first, then the two reads that + // found the meter's own cost, with nothing between them. + assert_eq!(seen[0], Seen::Get); + assert_eq!((reads[0], reads[1]), (1, 2), "{seen:?}"); + assert_eq!(c.empty_span_microjoules, at(reads[1]) - at(reads[0])); + assert_eq!(c.empty_span_microjoules, 10); + // The last span: its two reads, and the lookups between them. + let (open, close) = (reads[reads.len() - 2], reads[reads.len() - 1]); + assert_eq!(close, seen.len() - 1, "nothing follows the closing read"); + assert_eq!(c.batch, (close - open - 1) as u64, "{get_nanojoules} nJ: the batch is the lookups in the span"); + assert!(seen[open + 1..close].iter().all(|s| *s == Seen::Get)); + assert_eq!(c.span_microjoules, at(close) - at(open), "{get_nanojoules} nJ"); + assert_eq!(c.per_hit_nanojoules, (c.span_microjoules * 1_000 + c.batch / 2) / c.batch); + assert!(c.span_microjoules >= SPAN_TO_EMPTY * c.empty_span_microjoules, "{c:?}"); + // All the host's time but the hit's own lookup, which came first. + assert_eq!(c.took, host.elapsed() - Duration::from_nanos(150)); + // Every span is a pair of reads: the empty one, then each batch. + assert_eq!(reads.len() % 2, 0, "{seen:?}"); + // The figure is the lookup's own, over by the meter's share of + // the span and no more: 10 µJ in a span of 500 at least. + let over = c.per_hit_nanojoules as f64 / get_nanojoules as f64; + assert!((1.0..=1.03).contains(&over), "{get_nanojoules} nJ a lookup calibrated as {c:?}"); + } + } + + /// Integer conservation. Over any number of hits the micro-joules + /// charged sum to the hits' calibrated nanojoules, to within the one + /// micro-joule still carried: for a figure under a micro-joule, where + /// most hits are charged nothing and some one, and for a figure over it. + #[tokio::test] + async fn the_microjoules_charged_sum_to_the_calibrated_total() { + const HITS: u64 = 10_000; + for get_nanojoules in [137u64, 2_345] { + let host = Host::new(get_nanojoules, 150); + let cascade = host.cascade(JouleCost::estimated(42)); + let q = Query::new("known key"); + let mut charges = Vec::new(); + for _ in 0..HITS { + charges.push(share(&cascade.resolve(q.clone()).await)); + } + let per_hit = cascade.calibration(StageKind::Kv).expect("calibrated").per_hit_nanojoules; + assert_eq!(per_hit < 1_000, get_nanojoules < 1_000, "{per_hit} nJ calibrated for {get_nanojoules}"); + let mut charged = 0u64; + for (n, charge) in charges.iter().enumerate() { + charged += charge; + let accrued = (n as u64 + 1) * per_hit; + assert_eq!(charged, accrued / 1_000, "{per_hit} nJ a hit, after {} hits", n + 1); + } + let total = (HITS * per_hit + 500) / 1_000; + assert!(charged.abs_diff(total) <= 1, "{charged} µJ charged for {total} µJ calibrated"); + // Neither every hit rounded up nor a string of zeros. + let (least, most) = (per_hit / 1_000, per_hit.div_ceil(1_000)); + assert!(charges.iter().all(|c| *c == least || *c == most), "{per_hit} nJ a hit"); + assert!(charges.contains(&least) && charges.contains(&most), "{per_hit} nJ a hit"); + assert_ne!(charged, HITS * most, "every hit was rounded up"); + assert_ne!(charged, HITS * least, "every hit was rounded down"); + } + } + + /// The carry is one count for the tier, whichever thread a hit runs on: + /// eight threads' hits together are charged the calibrated total. + #[test] + fn hits_on_many_threads_are_charged_the_calibrated_total() { + const THREADS: u64 = 8; + const HITS: u64 = 500; + /// The micro-joules `n` hits are charged, on a runtime of this + /// thread's own. + fn block_on(cascade: &Cascade, n: u64) -> u64 { + let runtime = tokio::runtime::Builder::new_current_thread().build().unwrap(); + runtime.block_on(async { + let mut charged = 0; + for _ in 0..n { + charged += share(&cascade.resolve(Query::new("known key")).await); + } + charged + }) + } + let host = Host::new(2_345, 150); + let cascade = Arc::new(host.cascade(JouleCost::estimated(42))); + // One hit first, so the tier is calibrated before the threads start. + let first = block_on(&cascade, 1); + let per_hit = cascade.calibration(StageKind::Kv).expect("calibrated").per_hit_nanojoules; + let threads: Vec<_> = (0..THREADS) + .map(|_| { + let cascade = cascade.clone(); + std::thread::spawn(move || block_on(&cascade, HITS)) + }) + .collect(); + let charged: u64 = first + threads.into_iter().map(|t| t.join().unwrap()).sum::(); + assert_eq!(charged, (THREADS * HITS + 1) * per_hit / 1_000); + } + + /// A lookup that is slow is repeated only as far as the time allowed: no + /// batch is started that would not fit, and the largest span read by + /// then is kept, short of fifty empty spans as it is. A lookup slower + /// than the whole allowance is repeated once. + #[tokio::test] + async fn a_slow_lookup_is_repeated_only_as_far_as_the_time_allowed() { + // 1 ms a lookup: 40 fit. + let host = Host::new(2_000, 1_000_000); + let cascade = host.cascade(JouleCost::estimated(42)); + cascade.resolve(Query::new("known key")).await; + let c = cascade.calibration(StageKind::Kv).expect("a live span is a calibration, however short"); + assert!(c.took <= CALIBRATION_CAP, "{c:?}"); + assert!(host.gets() <= 1 + 40, "{} lookups of 1 ms", host.gets()); + assert!(c.batch > 8 && c.batch <= 40, "{c:?}"); + assert!(c.span_microjoules < SPAN_TO_EMPTY * c.empty_span_microjoules, "{c:?}"); + assert_eq!(c.per_hit_nanojoules, (c.span_microjoules * 1_000 + c.batch / 2) / c.batch); + + // 100 ms a lookup: one, in one span. + let host = Host::new(2_000, 100_000_000); + let cascade = host.cascade(JouleCost::estimated(42)); + let r = cascade.resolve(Query::new("known key")).await; + let c = cascade.calibration(StageKind::Kv).expect("calibrated"); + assert_eq!((c.batch, host.gets()), (1, 2), "the hit's lookup and one more"); + assert_eq!(c.span_microjoules, 12, "the lookup's 2 µJ and the closing read's 10"); + assert_eq!(r.joule_cost, JouleCost::estimated(50 + 12)); + } + + /// With no live counter a hit is what its stage reported, and so is a + /// neural resolve: the cascade given no meter at all, a counter whose + /// read fails, one that holds from the start, and one that holds once + /// the batches begin. The attempt is made once: later hits read nothing. + #[tokio::test] + async fn with_no_live_counter_every_resolve_keeps_its_stages_cost() { + // No meter given: nothing is read, and nothing is calibrated. + let cascade = fixture().without_memoization(); + for stage in TIERS { + let q = answered_by(&cascade, stage); + for _ in 0..3 { + let r = cascade.resolve(q.clone()).await; + assert_eq!((r.stage, r.joule_cost), (stage, JouleCost::estimated(own_cost(stage)))); + } + assert_eq!(cascade.calibration(stage), Err(NoCalibration::NoMeter)); + } + let r = cascade.resolve(Query::new("totally novel question")).await; + assert_eq!(r.joule_cost, JouleCost::estimated(50_000_000)); + + // A failed read (one read spent), and a counter that holds over + // back-to-back reads (given up after 2 ms of them). + for (reads, why, spent) in [ + (vec![None], NoCalibration::ReadFailed, 1), + (vec![Some(7_000), None], NoCalibration::ReadFailed, 2), + (vec![Some(7_000)], NoCalibration::CounterHeld, 2), + ] { + for stage in TIERS { + let meter = Scripted::new(&reads); + let cascade = metered(&meter, Box::new(EchoBackend::new().with_cost(42))).without_memoization(); + let q = answered_by(&cascade, stage); + for _ in 0..3 { + let r = cascade.resolve(q.clone()).await; + assert_eq!((r.stage, r.joule_cost), (stage, JouleCost::estimated(own_cost(stage))), "{reads:?}"); + } + assert_eq!(cascade.calibration(stage), Err(why), "{stage} on {reads:?}"); + assert_eq!(meter.taken(), spent, "{stage} on {reads:?}: the one attempt's reads"); + } + } + + // A counter that moved over no work and then held over every batch: + // the batches stop at the time allowed (the last of them fits in it, + // its two reads of 80 µs come on top, and the hit's own lookup of + // 20 µs came before), with no span to keep. + let host = Host::holding_after(2_000, 20_000, Some(2)); + let cascade = host.cascade(JouleCost::estimated(42)); + let r = cascade.resolve(Query::new("known key")).await; + assert_eq!(r.joule_cost, JouleCost::estimated(50)); + assert_eq!(cascade.calibration(StageKind::Kv), Err(NoCalibration::CounterHeld)); + assert!(host.elapsed() <= CALIBRATION_CAP + Duration::from_micros(160 + 20), "{:?}", host.elapsed()); + assert!(host.elapsed() > CALIBRATION_CAP / 2, "the batches stopped early: {:?}", host.elapsed()); + let reads = host.reads(); + for _ in 0..3 { + let r = cascade.resolve(Query::new("known key")).await; + assert_eq!(r.joule_cost, JouleCost::estimated(50)); + } + assert_eq!(host.reads(), reads, "a hit read the meter"); + let r = cascade.resolve(Query::new("nothing knows this")).await; + assert_eq!(r.joule_cost, JouleCost::estimated(42), "the neural stage's own cost"); + assert_eq!(host.reads(), reads + 2); + } + + /// The old behaviour is gone: a resolve a deterministic tier answers is + /// never measured. Not on a live counter, where a span around the hit + /// would have read (1_000 -> 1_010 here, once `measured(10)`); not the + /// hit that calibrates its tier; not where the calibration failed; and + /// not a cached response that carries a measured figure of its own. + #[tokio::test] + async fn a_hit_is_never_measured() { + let scripts: [&[Option]; 5] = [ + &CALIBRATES_AT_700, + &[Some(1_000), Some(1_010), Some(2_000), None], + &[Some(2_000), Some(2_250), None], + &[Some(7_000)], + &[None], + ]; + for script in scripts { + for stage in TIERS { + let meter = Scripted::new(script); + let cascade = metered(&meter, Box::new(EchoBackend::new())).without_memoization(); + let q = answered_by(&cascade, stage); + for hit in 0..5 { + let r = cascade.resolve(q.clone()).await; + assert_eq!(r.stage, stage); + let cost = r.joule_cost; + assert_eq!(cost.source, JouleSource::Estimated, "{stage} hit {hit} on {script:?}: {cost}"); + } + } + } + + // A memoized answer, on a live counter: the neural resolve is read, + // the hit that follows is not. + let meter = Scripted::new(&[Some(1_000), Some(1_500), Some(2_000), Some(2_010), Some(3_000), Some(3_700)]); + let cascade = metered(&meter, Box::new(EchoBackend::new().with_cost(0))); + let q = Query::new("totally novel question"); + assert_eq!(cascade.resolve(q.clone()).await.joule_cost, JouleCost::measured(500)); + let hit = cascade.resolve(q).await; + assert_eq!((hit.stage, hit.joule_cost), (StageKind::Cache, JouleCost::estimated(700))); + + // A cached response that carries a measured figure: what was read + // for an earlier resolve is not a reading of this one. + for cascade in [fixture(), metered(&Scripted::new(&CALIBRATES_AT_700), Box::new(EchoBackend::new()))] { + let q = Query::new("answered and read once"); + let read_once = Response::new(q.id, "answer".to_string(), StageKind::Cache, JouleCost::measured(9)); + cascade.cache().insert_response(q.id, read_once); + let r = cascade.resolve(q).await; + assert_eq!(r.joule_cost.source, JouleSource::Estimated, "{}", r.joule_cost); + assert!(r.joule_cost.microjoules >= 9); + assert!(r.receipt.verify(&r.query_id, &r.stage, &r.joule_cost, &r.payload)); + } + } + + /// While one resolve calibrates a tier, another that the tier answers is + /// charged the stage's own cost and reads nothing; and a tier switched + /// off is never calibrated, by a hit or by `calibrate`. + #[tokio::test] + async fn a_tier_being_calibrated_or_switched_off_charges_its_stages_cost() { + let host = Host::new(137, 150); + let cascade = host.cascade(JouleCost::estimated(42)); + cascade.hits(Tier::Kv).calibration = Err(NoCalibration::InProgress); + let r = cascade.resolve(Query::new("known key")).await; + assert_eq!(r.joule_cost, JouleCost::estimated(50)); + assert_eq!((host.reads(), cascade.calibration(StageKind::Kv)), (0, Err(NoCalibration::InProgress))); + + let host = Host::new(137, 150); + let cascade = host.cascade(JouleCost::estimated(42)).without_hit_calibration(StageKind::Kv); + let q = Query::new("known key"); + assert_eq!(cascade.resolve(q.clone()).await.joule_cost, JouleCost::estimated(50)); + assert_eq!(cascade.calibrate(&q).await, Some((StageKind::Kv, Err(NoCalibration::Disabled)))); + assert_eq!(cascade.resolve(q).await.joule_cost, JouleCost::estimated(50)); + assert_eq!(host.reads(), 0); + // The neural stage is read, never calibrated, and cannot be switched. + let cascade = cascade.without_hit_calibration(StageKind::Neural); + assert_eq!(cascade.calibration(StageKind::Neural), Err(NoCalibration::NotADeterministicTier)); + } + + /// `calibrate` calibrates the tier that answers a query without a + /// resolve, and again whenever it is asked: the new figure replaces the + /// old, and the nanojoules carried under the old one are not dropped. + #[tokio::test] + async fn a_tier_can_be_calibrated_again() { + let host = Host::new(137, 150); + let cascade = host.cascade(JouleCost::estimated(42)); + let q = Query::new("known key"); + assert_eq!(cascade.calibrate(&Query::new("nothing knows this")).await, None); + assert_eq!(host.reads(), 0, "no tier answered, so nothing was read"); + + // Before any resolve: the first hit then finds the tier calibrated. + let (stage, first) = cascade.calibrate(&q).await.expect("the KV stage answers"); + let first = first.expect("calibrated"); + assert_eq!(stage, StageKind::Kv); + assert_eq!(cascade.calibration(StageKind::Kv), Ok(first)); + let reads = host.reads(); + let mut charged = 0; + for _ in 0..7 { + charged += share(&cascade.resolve(q.clone()).await); + } + assert_eq!(host.reads(), reads, "a hit read the meter"); + assert_eq!(charged, 7 * first.per_hit_nanojoules / 1_000); + + // The lookup now costs seventeen times as much. + host.get.0.store(2_345, Ordering::SeqCst); + let (_, second) = cascade.calibrate(&q).await.expect("the KV stage answers"); + let second = second.expect("calibrated"); + assert!(second.per_hit_nanojoules > 10 * first.per_hit_nanojoules, "{first:?} then {second:?}"); + assert_eq!(cascade.calibration(StageKind::Kv), Ok(second)); + for _ in 0..7 { + charged += share(&cascade.resolve(q.clone()).await); + } + assert_eq!(charged, (7 * first.per_hit_nanojoules + 7 * second.per_hit_nanojoules) / 1_000); + } + + /// A resolve and a calibration can be spawned onto a runtime's other + /// threads: no lock is held across a stage's lookup. + #[test] + fn a_resolve_can_move_between_threads() { + fn sendable(_: &T) {} + let cascade = fixture(); + let q = Query::new("known key"); + sendable(&cascade.resolve(q.clone())); + sendable(&cascade.calibrate(&q)); + } } -- 2.54.0 (Apple Git-157) From bd865deb9a233d7c76e273955f25f567337877c3 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Tue, 13 Oct 2026 01:37:05 -0400 Subject: [PATCH 2/4] eoc-cascade: a hit's calibrated figure and a neural resolve's span, against an oracle on the real counter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 0d78becb's rules were tested on scripted counters. These two hold them against this process's own CPU energy counter, each in a test binary of its own with one test, as eoc-meter's are: the counter's domain is every thread of the process, and the oracle's check is the process's context-switch count. Gated the same way: macOS only, and the counter must be live. The oracle (`tests/oracle`) shares no call and no number with the counter: `proc_pid_rusage` declared there and read at the SDK's offset, taken once the test's thread has left its core by the process's context-switch count rising across a sleep. A `Watched` counter puts that read on either side of every read the cascade takes, so a span lies between two windows of the oracle's: the one inside it and the one around it and both reads. That a reading lies between them holds for any count that only grows, so it is asserted every time; what is allowed to miss a tenth of the time is how far apart the two windows are, which is the two readers' own reads and wake-ups. `hit_calibration_process_energy`: - each tier calibrated 20 times through the watched counter: the span the calibration kept is its last two reads, it lies between the oracle's two windows around the same batch, the per-hit figure times the batch is that span, and the empty span it reports is the counter's first advance over no work. The inner window was 952/1000 of the outer at least, quiet and under 40 busy loops, so the figure is pinned within a twentieth; a fifth is asked. - on the bare counter, after a tier's first hit, 10,000 hits take no read of the meter (counted), every one is an estimate, and the microjoules charged are the calibrated nanojoules to the one microjoule still carried. - what the change removes: a span around one hit, two reads about one lookup, reads more than the hit is charged, every time; in an optimized build 5.8 to 17 times as much at the median, and three times is asked. In an unoptimized build the lookup itself costs about what the meter does (22 µJ a cache hit against 1.4 optimized), so no multiple is asked there. `neural_span_process_energy`, with a KV stage that burns 8 ms of the thread's CPU on its way to missing: - a neural stage that burns 2 ms: exactly two reads, the first after the KV lookup, and the measured figure between the oracle's window inside the span and its window around it, which opens after the burn. The inner window was 785/1000 of the outer at least under load (992 quiet); two thirds is asked. - a neural stage that only sleeps 2 ms, on the bare counter, nothing taking the thread off its core between the burn and the cascade's first read: the figure is under a quarter of the oracle's window over the whole resolve (at most 32/1000 of it). A span opened before the walk reads the burn. No joule figure and no timing is asserted. A calibration that lands on an efficiency core reads a fifth of what one on a performance core does (253 nJ a cache hit against 1,400): the figure follows the core the batch ran on. Co-Authored-By: Claude Opus 5.5 --- .../tests/hit_calibration_process_energy.rs | 304 ++++++++++++++++++ .../tests/neural_span_process_energy.rs | 201 ++++++++++++ .../crates/eoc-cascade/tests/oracle/mod.rs | 213 ++++++++++++ 3 files changed, 718 insertions(+) create mode 100644 eoc/eoc-rs/crates/eoc-cascade/tests/hit_calibration_process_energy.rs create mode 100644 eoc/eoc-rs/crates/eoc-cascade/tests/neural_span_process_energy.rs create mode 100644 eoc/eoc-rs/crates/eoc-cascade/tests/oracle/mod.rs diff --git a/eoc/eoc-rs/crates/eoc-cascade/tests/hit_calibration_process_energy.rs b/eoc/eoc-rs/crates/eoc-cascade/tests/hit_calibration_process_energy.rs new file mode 100644 index 000000000..c2b641d19 --- /dev/null +++ b/eoc/eoc-rs/crates/eoc-cascade/tests/hit_calibration_process_energy.rs @@ -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) -> 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:?}" + ); + } +} diff --git a/eoc/eoc-rs/crates/eoc-cascade/tests/neural_span_process_energy.rs b/eoc/eoc-rs/crates/eoc-cascade/tests/neural_span_process_energy.rs new file mode 100644 index 000000000..301a541f0 --- /dev/null +++ b/eoc/eoc-rs/crates/eoc-cascade/tests/neural_span_process_energy.rs @@ -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); + +impl KvBackend for BurnsThenMisses { + fn get(&self, _key: &str) -> Option> { + burn(BEFORE_THE_SPAN); + self.0.fetch_add(1, Ordering::SeqCst); + None + } + fn put(&self, _key: &str, _value: Vec) {} + fn embeddings(&self) -> Vec<(Vec, String)> { + Vec::new() + } + fn put_embedding(&self, _embedding: Vec, _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, neural: Neural, lookups: &Arc) -> 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"); +} diff --git a/eoc/eoc-rs/crates/eoc-cascade/tests/oracle/mod.rs b/eoc/eoc-rs/crates/eoc-cascade/tests/oracle/mod.rs new file mode 100644 index 000000000..f28911806 --- /dev/null +++ b/eoc/eoc-rs/crates/eoc-cascade/tests/oracle/mod.rs @@ -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 ``. +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, + 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 { + counter: C, + reads: Mutex>, + count: Arc, +} + +impl Watched { + 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) -> Self { + Self { counter, reads: Mutex::new(Vec::new()), count } + } + + /// The reads taken since the last call, in order. + pub fn take(&self) -> Vec { + std::mem::take(&mut *self.reads.lock().unwrap()) + } +} + +impl JouleCounter for Watched { + fn read_microjoules(&self) -> eoc_core::Result { + 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, + /// 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 { + counter: C, + reads: Mutex, +} + +impl Counted { + pub fn new(counter: C) -> Self { + Self { counter, reads: Mutex::new(0) } + } + + pub fn reads(&self) -> u64 { + *self.reads.lock().unwrap() + } +} + +impl JouleCounter for Counted { + fn read_microjoules(&self) -> eoc_core::Result { + *self.reads.lock().unwrap() += 1; + self.counter.read_microjoules() + } + fn name(&self) -> &'static str { + self.counter.name() + } +} -- 2.54.0 (Apple Git-157) From 3d941db7059bd6e567e529a2fbab93391f2d3925 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Tue, 13 Oct 2026 01:39:37 -0400 Subject: [PATCH 3/4] eoc: a hit prints its estimate and the calibration it came from MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With 0d78becb a prompt the cache, the KV store or the graph answers is no longer bracketed by two reads of the meter: it is charged the stage's own figure plus what a hit on that tier costs on this host, calibrated over a batch the first time the tier answers. The CLI resolves one prompt a process, so its hit is always the one that calibrates. `eoc query` therefore prints, for a hit, an estimate (it printed `measured(the meter's own cost)` for a cache hit on a live counter, and the stage's estimate plus that cost for a KV or graph hit), and a `hit` line that says where the figure came from: the per-hit nanojoules, the batch, what the meter read over it and over no work; or why the tier has no calibration and the cost is the stage's own figure. On the build host (M3 Ultra, release): $ eoc query ping stage : kv payload: pong 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) A prompt that falls through to the neural stage prints as before: the echo backend's 50 J estimate plus what the counter read around the stage, an estimate; measured only where the stage estimated nothing, which the existing in-process test pins. `eoc meter`'s `query` line says how the meter is used. Tests: the CLI's own path for `ping`, a KV hit and then a cache hit, is an estimate on every host; where the tier was calibrated the cost is the stage's figure plus the whole microjoules of the calibrated one, and where it was not, the stage's figure and a reason this host's meter can give (the stub holds, the powermetrics stand-in fails its reads). It takes the lock the burning tests take, since the meter's domain is the whole process. The `hit` line is pinned on its own. Co-Authored-By: Claude Opus 5.5 --- eoc/eoc-rs/crates/eoc/src/main.rs | 121 +++++++++++++++++++++++++++--- 1 file changed, 111 insertions(+), 10 deletions(-) diff --git a/eoc/eoc-rs/crates/eoc/src/main.rs b/eoc/eoc-rs/crates/eoc/src/main.rs index 5c011344c..280524c6e 100644 --- a/eoc/eoc-rs/crates/eoc/src/main.rs +++ b/eoc/eoc-rs/crates/eoc/src/main.rs @@ -3,7 +3,14 @@ //! Subcommands: //! //! - `eoc query ""` — 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) -> 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) -> 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, 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)); -- 2.54.0 (Apple Git-157) From 23eacd79a5ae8109ac555a8c1ea2a2b0b411dfe4 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Tue, 13 Oct 2026 01:44:36 -0400 Subject: [PATCH 4/4] eoc docs: what a resolve is charged, with the measured figures; the example prints the unmetered baseline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The README said only that joule cost is the unit of accounting. It now has the rule and where the cascade reads its meter since 0d78becb: around the neural stage, never around a hit, which is charged a per-hit figure calibrated on the host. The example output shows the `hit` line, with and without a live counter. eoc-cascade's own docs carry the final figures from the build host (M3 Ultra, macOS 27.0.1, release, load average 5 to 14, three runs of `examples/hit_cost`): - 10,000 cache hits: 4.18 to 4.20 ms, 0.375 µs a resolve at the median (parent: 1,641 to 1,651 ms, 162 to 164 µs). A cascade given no meter at all: 3.55 to 3.60 ms, 0.333 µs. KV 0.667 µs, graph 0.75 to 0.79 µs. - calibrated: 1,413 to 1,716 nJ a cache hit, 2,672 to 2,961 nJ a KV hit, 3,007 to 3,163 nJ a graph hit, each from 416 to 3,328 lookups in a span 51 to 351 times the 12 to 19 µJ an empty span read; 0.85 to 2.3 ms for a first hit, calibration included, 0.96 to 1.35 ms for one asked again. - a neural stage that does nothing: 162.4 to 162.6 µs a resolve and 14.3 to 14.8 µJ measured, the cascade's two reads; 0.79 to 0.83 µs with no meter. The three misses before the opening read: 0.95 µJ and 220 ns together. `HitCalibration`'s docs say the figure is that of the core the batch ran on and of the build, and that the calibration's own energy is charged to no resolve. The module docs no longer link the private `cascade_cost`. eoc-meter's crate docs and `detect_in_process` say how eoc-cascade uses a counter now (it was the example of a caller that should bracket a batch, not each operation). eoc-core's `JouleCost::zero` and eoc-cache said a lookup's energy is below any counter's resolution; it is about a microjoule, readable as the mean of a batch, and less than a read costs, which is why no counter is read for a hit. `examples/hit_cost` also runs the same stages with no meter, the baseline a metered hit is now within 0.05 µs of. Co-Authored-By: Claude Opus 5.5 --- eoc/eoc-rs/README.md | 27 ++++++++- eoc/eoc-rs/crates/eoc-cache/src/lib.rs | 8 ++- .../crates/eoc-cascade/examples/hit_cost.rs | 23 ++++++-- eoc/eoc-rs/crates/eoc-cascade/src/lib.rs | 59 +++++++++++-------- eoc/eoc-rs/crates/eoc-core/src/joule_cost.rs | 7 ++- eoc/eoc-rs/crates/eoc-meter/src/lib.rs | 10 +++- 6 files changed, 97 insertions(+), 37 deletions(-) diff --git a/eoc/eoc-rs/README.md b/eoc/eoc-rs/README.md index e194e041d..1d75efcb0 100644 --- a/eoc/eoc-rs/README.md +++ b/eoc/eoc-rs/README.md @@ -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: diff --git a/eoc/eoc-rs/crates/eoc-cache/src/lib.rs b/eoc/eoc-rs/crates/eoc-cache/src/lib.rs index f165c7858..fbbafe6c0 100644 --- a/eoc/eoc-rs/crates/eoc-cache/src/lib.rs +++ b/eoc/eoc-rs/crates/eoc-cache/src/lib.rs @@ -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)] diff --git a/eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs b/eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs index be33f5ab8..a3c1fd358 100644 --- a/eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs +++ b/eoc/eoc-rs/crates/eoc-cascade/examples/hit_cost.rs @@ -30,7 +30,10 @@ impl NeuralBackend for Nothing { } } -fn cascade() -> Cascade { +/// 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(); @@ -39,14 +42,14 @@ fn cascade() -> Cascade { Triple::new("Tokyo", "capital of", "Japan"), Triple::new("Mars", "fourth planet from", "the Sun"), ]); - Cascade::new( + let cascade = Cascade::new( Arc::new(LruCache::new(1024)), Arc::new(KvStage::new(kv)), Arc::new(graph), Arc::new(NeuralStage::new(Box::new(Nothing))), ) - .with_meter(Arc::from(detect_in_process())) - .without_memoization() + .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 @@ -140,7 +143,7 @@ async fn misses(cascade: &Cascade, n: u64) { #[tokio::main(flavor = "current_thread")] async fn main() { - let c = cascade(); + let c = cascade(true); println!("meter: {}", c.meter().name()); let cached = Query::new("seen before"); @@ -149,8 +152,16 @@ async fn main() { 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. + // 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)); } diff --git a/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs b/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs index 6c8c0f57f..c6d29b4b4 100644 --- a/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs +++ b/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs @@ -11,34 +11,36 @@ //! //! **A resolve the neural stage answers is read.** The meter is read once the //! three deterministic tiers have missed, and again after the neural stage -//! and the memoizing write: two reads, one span. [`cascade_cost`] combines -//! that span with the cost the stage reported, and the total is `Measured` +//! and the memoizing write: two reads, one span. That span is combined with +//! the cost the stage reported (`cascade_cost`), and the total is `Measured` //! only when every micro-joule of it was read from a live counter. The three //! misses are spent before the opening read, so they are not in the figure. -//! They are lookups of under a microsecond each, well under what the span -//! holds of the meter itself, its closing read's own wake-up. So the measured +//! They are small against what the span holds of the meter itself, its +//! closing read's own wake-up: on an M3 Ultra the three together took 220 ns +//! and 0.95 µJ, and a span over no work read 12 to 29 µJ. So the measured //! figure is the neural stage's energy, the memoizing write's and the meter's //! own closing cost: over the neural stage's energy, never under it. //! //! **A resolve a deterministic tier answers (cache, kv, graph) reads no //! meter**, and is never `Measured`. A read of a counter costs more than the //! hit it would bracket, and what such a span reads is the read. On an M3 -//! Ultra (macOS 27.0.1), with this process's own energy counter, whose every -//! read first sees the reading thread leave its core: a resolve answered by -//! the cache took 164 µs at the median with a span around it, against well -//! under a microsecond for the lookup, and the span read 13 to 28 µJ, 14 µJ -//! on average over 10,000 hits, the same as a span around a neural stage that -//! does nothing (17 to 19 µJ). That figure is the meter's own cost, not the -//! hit's; tagged `Measured` and sealed in the hit's receipt it was a true -//! reading booked to the wrong work. So a hit is charged a figure calibrated -//! on this host instead ([`HitCalibration`]): the first time a tier answers, -//! the cascade repeats that lookup back to back inside one span, enough times -//! that the span is large against the meter's own cost, and keeps the span's -//! reading per hit. Every later hit on that tier is charged that figure, with -//! no read. It is `Estimated`: it was read by this counter, on this host, but -//! over a batch, not over this resolve. A per-operation figure for work -//! shorter than the meter can resolve is sound only as the mean of many -//! operations in one span, and a mean is not a reading of any one of them. +//! Ultra (macOS 27.0.1, an optimized build), with this process's own energy +//! counter, whose every read first sees the reading thread leave its core: a +//! resolve answered by the cache took 164 µs at the median with a span around +//! it, 0.38 µs without, and the span read 13 to 28 µJ, 14 µJ on average over +//! 10,000 hits, as a span around a neural stage that does nothing did (14 to +//! 19 µJ on average), where the hit itself is calibrated at 1.2 to 1.7 µJ. +//! That figure is the meter's own cost, not the hit's; tagged `Measured` and +//! sealed in the hit's receipt it was a true reading booked to the wrong +//! work. So a hit is charged a figure calibrated on this host instead +//! ([`HitCalibration`]): the first time a tier answers, the cascade repeats +//! that lookup back to back inside one span, enough times that the span is +//! large against the meter's own cost, and keeps the span's reading per hit. +//! Every later hit on that tier is charged that figure, with no read. It is +//! `Estimated`: it was read by this counter, on this host, but over a batch, +//! not over this resolve. A per-operation figure for work shorter than the +//! meter can resolve is sound only as the mean of many operations in one +//! span, and a mean is not a reading of any one of them. //! //! With no live counter — the cascade was given none, a read failed, or the //! counter did not advance — there is no calibration and no span, and every @@ -101,13 +103,23 @@ const LARGEST_BATCH: u64 = 1 << 22; /// The figure is that of the query the tier first answered (or the one /// [`Cascade::calibrate`] was given), charged to every hit on the tier: a /// lookup of another key, or a longer scan, costs what it costs, and is -/// charged this. +/// charged this. What the calibration itself spent, its spans and its +/// reads, is charged to no resolve. /// /// The counter reads its whole domain over the batch. On this process's own /// counter that is every thread of the process, so work on other threads /// during the batch is in the figure; on a whole-machine counter, so is the /// machine's. Calibrate on a quiet process, read the record, and call /// [`Cascade::calibrate`] again if it was not. +/// +/// The figure is also that of the core the batch ran on, and of this build. +/// On an M3 Ultra (macOS 27.0.1, optimized, this process's counter): 1.2 to +/// 1.7 µJ a cache hit, 2.6 to 3.4 µJ a KV hit and 2.9 to 3.4 µJ a graph hit +/// on a performance core, from 224 to 4,032 lookups in a span of 0.7 to +/// 5.9 mJ, 51 to 351 times the 12 to 29 µJ a span over no work read, +/// calibrated in 0.85 to 2.4 ms; a batch that landed on an efficiency core +/// read a fifth of that (253 nJ a cache hit). An unoptimized build's cache +/// hit read 22 µJ, most of it the receipt. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct HitCalibration { /// The span's reading per lookup, in nanojoules, to the nearest one. @@ -605,8 +617,9 @@ fn live_span(start: Option, end: Option) -> Option { /// span, a cache hit, is the span's reading, measured. With a counter whose /// read costs more than the hit, that reading was the meter's own cost (on /// an M3 Ultra, 14 µJ on average over 10,000 cache hits, each bracketed by -/// two reads taking 164 µs between them), sealed as the hit's. A hit is -/// charged by [`hit_cost`] instead, and no meter is read for it. +/// two reads taking 164 µs between them, where the hit is 0.38 µs and about +/// 1.5 µJ), sealed as the hit's. A hit is charged by [`hit_cost`] instead, +/// and no meter is read for it. fn cascade_cost(stage: JouleCost, span: Option) -> JouleCost { let Some(span) = span else { return stage; diff --git a/eoc/eoc-rs/crates/eoc-core/src/joule_cost.rs b/eoc/eoc-rs/crates/eoc-core/src/joule_cost.rs index aa458a6e1..2c1f06ecb 100644 --- a/eoc/eoc-rs/crates/eoc-core/src/joule_cost.rs +++ b/eoc/eoc-rs/crates/eoc-core/src/joule_cost.rs @@ -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 diff --git a/eoc/eoc-rs/crates/eoc-meter/src/lib.rs b/eoc/eoc-rs/crates/eoc-meter/src/lib.rs index e1e2a3d9b..fa24bebeb 100644 --- a/eoc/eoc-rs/crates/eoc-meter/src/lib.rs +++ b/eoc/eoc-rs/crates/eoc-meter/src/lib.rs @@ -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 { detect_in_process_with(process_energy_status, detect) } -- 2.54.0 (Apple Git-157)