//! 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:?}" ); } }