open-standards/eoc/eoc-rs/crates/eoc-cascade/tests/hit_calibration_process_energy.rs

//! What a hit is charged, against the hardware: the cascade on this
//! process's own CPU energy counter, each tier's calibrated per-hit figure
//! held against an oracle that reads the same count its own way.
//!
//! A hit on a deterministic tier reads no meter. It is charged a figure the
//! cascade calibrated on this host: the counter's reading over one span
//! around a batch of the tier's lookup, divided by the batch. Here every
//! read the cascade takes of its counter has the oracle's own read on either
//! side of it, so the span a calibration kept lies between two windows of
//! the oracle's around the same batch, and the figure is held against both.
//!
//! One test binary with one test: the counter's domain is every thread of
//! the process, and the oracle's check is the process's context-switch
//! count, so nothing else may run here. No joule figure and no timing is
//! asserted: what a lookup costs follows the core that ran it, and a loaded
//! host stretches everything's wall time.

#![cfg(target_os = "macos")]

mod oracle;

use std::sync::Arc;

use eoc_cache::{LruCache, Stage as _};
use eoc_cascade::{Cascade, HitCalibration, NoCalibration};
use eoc_core::{JouleSource, Query, Response, Stage};
use eoc_graph::{GraphStage, Triple};
use eoc_kv::{KvBackend, KvStage, MemoryKvBackend};
use eoc_meter::{CounterStatus, JouleCounter, MacosProcessEnergyCounter};
use eoc_neural::{EchoBackend, NeuralStage};

use oracle::{Counted, Span, Watched};

const TIERS: [Stage; 3] = [Stage::Cache, Stage::Kv, Stage::Graph];

/// A cascade whose three deterministic tiers each answer one query, metered
/// by `meter`. It does not memoize, so a KV or graph hit stays one.
fn cascade(meter: Arc<dyn JouleCounter>) -> Cascade {
    let kv = Box::new(MemoryKvBackend::new());
    kv.put("ping", b"pong".to_vec());
    let graph = GraphStage::new();
    graph.extend([
        Triple::new("Paris", "capital of", "France"),
        Triple::new("Tokyo", "capital of", "Japan"),
        Triple::new("Mars", "fourth planet from", "the Sun"),
    ]);
    let cascade = Cascade::new(
        Arc::new(LruCache::new(1024)),
        Arc::new(KvStage::new(kv)),
        Arc::new(graph),
        Arc::new(NeuralStage::new(Box::new(EchoBackend::new()))),
    )
    .with_meter(meter)
    .without_memoization();
    cascade.cache().insert(&answered_by(Stage::Cache), "answer");
    cascade
}

/// The query `stage` answers in [`cascade`].
fn answered_by(stage: Stage) -> Query {
    Query::new(match stage {
        Stage::Cache => "seen before",
        Stage::Kv => "ping",
        Stage::Graph => "Mars is the fourth planet from what?",
        Stage::Neural => "nothing knows this",
    })
}

/// What `stage` reports for a hit of its own, in micro-joules.
fn own_cost(stage: Stage) -> u64 {
    match stage {
        Stage::Cache => 0,
        Stage::Kv => 50,
        Stage::Graph => 500,
        Stage::Neural => unreachable!("not a deterministic tier"),
    }
}

/// `stage`'s own lookup of `q`, and the response minted for it: what a
/// calibration repeats.
async fn lookup(cascade: &Cascade, stage: Stage, q: &Query) {
    let r = match stage {
        Stage::Cache => cascade.cache().try_resolve(q).await,
        Stage::Kv => cascade.kv().try_resolve(q).await,
        Stage::Graph => cascade.graph().try_resolve(q).await,
        Stage::Neural => unreachable!("not a deterministic tier"),
    }
    .expect("the tier answers its query");
    std::hint::black_box(Response::new(r.query_id, r.payload, r.stage, r.joule_cost));
}

fn in_words(c: &HitCalibration) -> String {
    format!(
        "{} nJ a hit ({} lookups in one span of {} µJ, {} times the {} µJ an empty span read, in {:?})",
        c.per_hit_nanojoules,
        c.batch,
        c.span_microjoules,
        c.span_microjoules / c.empty_span_microjoules.max(1),
        c.empty_span_microjoules,
        c.took
    )
}

#[tokio::test]
async fn a_hit_is_charged_what_an_oracle_reads_around_the_same_batch() {
    const TRIES: usize = 20;
    const HITS: u64 = 10_000;
    let status = eoc_meter::process_energy_status();
    eprintln!("hit calibration: status={status:?}");
    assert_eq!(status, CounterStatus::Live, "this process's CPU energy counter is not live here");

    // 1. Each tier calibrated again and again through a counter the oracle
    //    reads on either side of. The span a calibration kept is its last
    //    two reads, and it lies between the oracle's window inside it and
    //    the oracle's window around it and both reads: that much holds for
    //    any count that only grows, so a figure outside the two windows is
    //    not this process's count over that batch. What the windows leave
    //    open is their own distance apart, the two readers' reads and
    //    wake-ups: the figure is pinned within a fifth when the window
    //    inside is at least four fifths of the one around (it was 96% at
    //    least on a quiet M3 Ultra).
    let watched = Arc::new(Watched::new(MacosProcessEnergyCounter::try_new().expect("the process counter is live here")));
    let through_oracle = cascade(watched.clone());
    let (mut not_calibrated, mut loose, mut tight) = (Vec::new(), Vec::new(), Vec::new());
    for stage in TIERS {
        let q = answered_by(stage);
        let (mut per_hit, mut short) = (Vec::new(), 0usize);
        for attempt in 0..TRIES {
            watched.take();
            let (answered, calibration) = through_oracle.calibrate(&q).await.expect("a deterministic tier answers");
            assert_eq!(answered, stage);
            let reads = watched.take();
            let c = match calibration {
                Ok(c) => c,
                Err(why) => {
                    not_calibrated.push((stage, attempt, why));
                    continue;
                }
            };
            assert_eq!(through_oracle.calibration(stage), Ok(c));

            // The empty span the calibration reports is the counter's first
            // advance over no work.
            let base = reads[0].microjoules.expect("a calibration's first read");
            let advanced = reads[1..].iter().filter_map(|r| r.microjoules).find(|uj| *uj > base);
            assert_eq!(Some(c.empty_span_microjoules), advanced.map(|uj| uj - base), "{stage} #{attempt}: {c:?}");
            assert!(c.empty_span_microjoules > 0 && c.batch > 0, "{stage} #{attempt}: {c:?}");

            // The span it kept, and the oracle's two windows around the
            // same batch.
            let span = Span::between(&reads[reads.len() - 2], &reads[reads.len() - 1]);
            assert_eq!(span.read, Some(c.span_microjoules * 1_000), "{stage} #{attempt}: {c:?} is not its last span");
            assert!(
                span.holds(c.span_microjoules * 1_000),
                "{stage} #{attempt}: a span of {} µJ outside the oracle's windows around it, {} nJ inside and {} nJ \
                 around",
                c.span_microjoules,
                span.inside,
                span.around
            );
            // The per-hit figure times the batch is that span, to the half
            // nanojoule a lookup the figure is rounded by.
            let figure = c.per_hit_nanojoules * c.batch;
            assert!(figure.abs_diff(c.span_microjoules * 1_000) <= c.batch / 2 + 1, "{stage} #{attempt}: {c:?}");
            if span.tightness() < 800 {
                loose.push((stage, attempt, figure, span.inside, span.around));
            } else {
                tight.push(span.tightness());
            }
            short += usize::from(c.span_microjoules < 50 * c.empty_span_microjoules);
            per_hit.push(c.per_hit_nanojoules);
        }
        per_hit.sort_unstable();
        eprintln!(
            "hit calibration: {stage} through the oracle x {TRIES}: {} to {} nJ a hit, {} at the median; {short} \
             stopped by the time allowed short of fifty empty spans; last: {}",
            per_hit.first().copied().unwrap_or(0),
            per_hit.last().copied().unwrap_or(0),
            per_hit.get(per_hit.len() / 2).copied().unwrap_or(0),
            through_oracle.calibration(stage).map(|c| in_words(&c)).unwrap_or_else(|why| why.to_string()),
        );
    }
    eprintln!(
        "hit calibration: {} of {} figures within the oracle's two windows with the inner at least {}/1000 of the \
         outer; {} looser {loose:?} (tier, try, figure nJ, inside nJ, around nJ); {} not calibrated {not_calibrated:?}",
        tight.len(),
        3 * TRIES,
        tight.iter().min().copied().unwrap_or(0),
        loose.len(),
        not_calibrated.len()
    );

    // 2. The cascade as it is used, on the bare counter: after a tier's
    //    first hit no hit reads the meter, every hit is an estimate, and the
    //    micro-joules charged over ten thousand hits are the calibrated
    //    nanojoules, to the one micro-joule still carried.
    let counted = Arc::new(Counted::new(MacosProcessEnergyCounter::try_new().expect("the process counter is live here")));
    let bare = cascade(counted.clone());
    let (mut over, mut not_over, mut unread, mut medians) = (0usize, Vec::new(), 0usize, Vec::new());
    for stage in TIERS {
        let q = answered_by(stage);
        let first = bare.resolve(q.clone()).await;
        assert_eq!((first.stage, first.joule_cost.source), (stage, JouleSource::Estimated), "{}", first.joule_cost);
        // A read whose thread was not seen to leave its core is no reading
        // by design, and leaves the tier uncalibrated: ask again.
        let mut held = bare.calibration(stage);
        for _ in 0..3 {
            if held.is_err() {
                held = bare.calibrate(&q).await.expect("a deterministic tier answers").1;
            }
        }
        let c = held.unwrap_or_else(|why| panic!("{stage} was not calibrated on a live counter: {why}"));
        let reads = counted.reads();
        let mut charged = 0u64;
        for _ in 0..HITS {
            let r = bare.resolve(q.clone()).await;
            assert_eq!((r.stage, r.joule_cost.source), (stage, JouleSource::Estimated), "{}", r.joule_cost);
            charged += r.joule_cost.microjoules - own_cost(stage);
        }
        assert_eq!(counted.reads(), reads, "{stage}: a hit read the meter");
        let calibrated = (HITS * c.per_hit_nanojoules + 500) / 1_000;
        assert!(charged.abs_diff(calibrated) <= 1, "{stage}: {charged} µJ charged over {HITS} hits of {c:?}");
        eprintln!(
            "hit calibration: {stage} on the bare counter: {}; {HITS} hits charged {charged} µJ beside the stage's \
             own {} µJ each, with no read",
            in_words(&c),
            own_cost(stage)
        );

        // 3. What this change removes: a span of the meter around one hit,
        //    two reads about one lookup, as every resolve took before. Such
        //    a span reads the hit and the meter's own cost, so it reads more
        //    than the hit is charged now. In an optimized build, where the
        //    lookup is a fraction of a microsecond, it reads several times
        //    as much: at least three times, at the median of the tries. An
        //    unoptimized lookup is itself about as dear as the meter, and
        //    there only the first is asked. Each try calibrates the tier and
        //    then reads one such span, so both ran on the host as it was
        //    then; the median, because the meter's own cost in one span
        //    moves severalfold from span to span on a loaded host.
        let mut tenths = Vec::new();
        for attempt in 0..TRIES {
            let Ok(c) = bare.calibrate(&q).await.expect("a deterministic tier answers").1 else {
                unread += 1;
                continue;
            };
            let open = counted.read_microjoules();
            lookup(&bare, stage, &q).await;
            let close = counted.read_microjoules();
            let (Ok(open), Ok(close)) = (open, close) else {
                unread += 1;
                continue;
            };
            let around_one_hit = close.saturating_sub(open) * 1_000;
            tenths.push(around_one_hit * 10 / c.per_hit_nanojoules.max(1));
            if around_one_hit > c.per_hit_nanojoules {
                over += 1;
            } else {
                not_over.push((stage, attempt, c.per_hit_nanojoules, around_one_hit));
            }
        }
        tenths.sort_unstable();
        let times = |tenths: Option<&u64>| tenths.map_or(0.0, |t| *t as f64 / 10.0);
        eprintln!(
            "hit calibration: {stage}: a span around one hit read {:.1} to {:.1} times what the hit is charged, {:.1} \
             at the median ({TRIES} tries)",
            times(tenths.first()),
            times(tenths.last()),
            times(tenths.get(tenths.len() / 2))
        );
        medians.push((stage, times(tenths.get(tenths.len() / 2))));
    }
    eprintln!(
        "hit calibration: {over} of {} spans around one hit read more than the hit's calibrated figure; {} did not \
         {not_over:?} (tier, try, per-hit nJ, span nJ); {unread} not read; at the median {medians:?} times the figure{}",
        3 * TRIES,
        not_over.len(),
        if cfg!(debug_assertions) { " (an unoptimized build: no multiple is asked)" } else { "" }
    );

    // A tier that cannot be calibrated says why, and is not charged.
    assert_eq!(bare.calibration(Stage::Neural), Err(NoCalibration::NotADeterministicTier));

    // A tenth may miss: on a loaded host one reader's wake-up can land on a
    // costlier core than the batch did, and a read whose thread was not
    // seen to leave its core is no reading by design.
    assert!(
        (not_calibrated.len() + loose.len()) * 10 <= 3 * TRIES,
        "{} calibrations failed {not_calibrated:?} and {} were not pinned within a fifth by the oracle {loose:?}",
        not_calibrated.len(),
        loose.len()
    );
    assert!(
        (not_over.len() + unread) * 10 <= 3 * TRIES,
        "{} spans around one hit read no more than the hit's figure {not_over:?}, and {unread} were not read",
        not_over.len()
    );
    if !cfg!(debug_assertions) {
        assert!(
            medians.iter().all(|(_, times)| *times >= 3.0),
            "a span around one hit read under three times the hit's figure at the median: {medians:?}"
        );
    }
}