diff --git a/crates/aura-bench/src/main.rs b/crates/aura-bench/src/main.rs index 58d7c24..cc2d672 100644 --- a/crates/aura-bench/src/main.rs +++ b/crates/aura-bench/src/main.rs @@ -4,6 +4,7 @@ //! bench instead of pinning a bogus baseline. mod baseline; +mod surfaces; mod synth; fn main() { diff --git a/crates/aura-bench/src/surfaces/engine.rs b/crates/aura-bench/src/surfaces/engine.rs new file mode 100644 index 0000000..cabea8b --- /dev/null +++ b/crates/aura-bench/src/surfaces/engine.rs @@ -0,0 +1,70 @@ +//! Engine throughput: bars/s through a bootstrapped harness — an SMA-cross +//! bias graph fed by the seeded synthetic walk via `VecSource`. Fingerprint: +//! the `SeriesReducer` finalize row `[last, max_drawdown, sign_flips]`. + +use super::{RepOutcome, Sizing}; +use crate::synth::synthetic_walk; +use aura_core::{Firing, Scalar, ScalarKind, Timestamp}; +use aura_engine::{GraphBuilder, Harness, Source, VecSource}; +use aura_std::{SeriesReducer, Sma, Sub}; +use std::collections::BTreeMap; +use std::sync::mpsc; +use std::time::Instant; + +pub const FULL_BARS: usize = 1_000_000; +pub const QUICK_BARS: usize = 2_000; + +pub fn rep(sizing: Sizing) -> Result { + let n_bars = if sizing.quick { QUICK_BARS } else { FULL_BARS }; + let (tx, rx) = mpsc::channel::<(Timestamp, Vec)>(); + let mut g = GraphBuilder::new("bench_sma_cross"); + let fast = g.add(Sma::builder().bind("length", Scalar::i64(12))); + let slow = g.add(Sma::builder().bind("length", Scalar::i64(48))); + let bias = g.add(Sub::builder()); + let sink = g.add(SeriesReducer::builder(Firing::Any, tx)); + let price = g.source_role("price", ScalarKind::F64); + g.feed(price, [fast.input("series"), slow.input("series")]); + g.connect(fast.output("value"), bias.input("lhs")); + g.connect(slow.output("value"), bias.input("rhs")); + g.connect(bias.output("value"), sink.input("col[0]")); + let flat = g + .build() + .map_err(|e| format!("bench graph must wire: {e:?}"))? + .compile_with_params(&[]) + .map_err(|e| format!("bench graph must compile: {e:?}"))?; + let mut h = Harness::bootstrap(flat).map_err(|e| format!("bench graph must bootstrap: {e:?}"))?; + let prices = synthetic_walk(n_bars); + let src: Vec> = vec![Box::new(VecSource::new(prices))]; + let t = Instant::now(); + h.run(src); + let wall = t.elapsed().as_secs_f64(); + let (_, row) = rx.try_iter().last().ok_or("SeriesReducer must emit its finalize row")?; + let fingerprint = format!( + "last={:.6} max_dd={:.6} flips={} n={}", + row[0].as_f64(), + row[1].as_f64(), + row[2].as_i64(), + n_bars + ); + let mut metrics = BTreeMap::new(); + metrics.insert("wall_s".to_string(), wall); + metrics.insert("bars_per_s".to_string(), n_bars as f64 / wall); + Ok(RepOutcome { metrics, fingerprint }) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The determinism smoke for the engine surface: two quick reps in-process + /// must produce byte-identical fingerprints (C1 — this is the harness's + /// entire premise, pinned as a currently-green test). + #[test] + fn engine_surface_fingerprint_is_deterministic_in_quick_mode() { + let a = rep(Sizing { quick: true }).expect("quick rep runs"); + let b = rep(Sizing { quick: true }).expect("quick rep runs"); + assert_eq!(a.fingerprint, b.fingerprint); + assert!(a.fingerprint.starts_with("last="), "shape: {}", a.fingerprint); + assert!(a.metrics.contains_key("bars_per_s")); + } +} diff --git a/crates/aura-bench/src/surfaces/ingest.rs b/crates/aura-bench/src/surfaces/ingest.rs new file mode 100644 index 0000000..836acf2 --- /dev/null +++ b/crates/aura-bench/src/surfaces/ingest.rs @@ -0,0 +1,67 @@ +//! Ingest throughput: bars/s draining one field column out of a synthetic +//! 48-byte M1 zip archive through the public ingest window API. A FRESH +//! `DataServer` per repetition, so the measured drain pays zip inflate + +//! record parse instead of the server's in-memory cache; the discarded warmup +//! rep leaves the OS page cache warm and consistent. + +use super::{RepOutcome, Sizing}; +use crate::synth::{write_symbol_archive, ScratchDir}; +use aura_ingest::{open_columns, DataServer, M1Field}; +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Instant; + +pub const SYMBOL: &str = "BENCHI"; +pub const FULL_MONTHS: ((i32, u32), (i32, u32)) = ((2024, 1), (2025, 12)); +pub const QUICK_MONTHS: ((i32, u32), (i32, u32)) = ((2024, 1), (2024, 2)); + +/// The archive is written once per surface run (setup, unmeasured) and shared +/// by every repetition. +pub struct IngestFixture { + pub dir: ScratchDir, +} + +pub fn setup(sizing: Sizing) -> Result { + let dir = ScratchDir::new("ingest").map_err(|e| format!("scratch dir: {e}"))?; + let (from, to) = if sizing.quick { QUICK_MONTHS } else { FULL_MONTHS }; + write_symbol_archive(&dir.0, SYMBOL, from, to); + Ok(IngestFixture { dir }) +} + +pub fn rep(fixture: &IngestFixture) -> Result { + let server = Arc::new(DataServer::new(&fixture.dir.0)); + let mut srcs = open_columns(&server, SYMBOL, None, None, &[M1Field::Close]) + .ok_or("bench archive must be present for the close column")?; + let mut src = srcs.pop().ok_or("one source for one field")?; + let t = Instant::now(); + let mut n: u64 = 0; + let mut xor: u64 = 0; + while let Some((_, v)) = src.next() { + n += 1; + xor ^= v.as_f64().to_bits(); + } + let wall = t.elapsed().as_secs_f64(); + if n == 0 { + return Err("the synthetic archive must yield bars".to_string()); + } + let mut metrics = BTreeMap::new(); + metrics.insert("wall_s".to_string(), wall); + metrics.insert("bars_per_s".to_string(), n as f64 / wall); + Ok(RepOutcome { metrics, fingerprint: format!("n={n} xor={xor:016x}") }) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Determinism smoke: two drains of the same archive fold to the same + /// fingerprint, and a fresh server per rep really re-reads the bytes. + #[test] + fn ingest_surface_fingerprint_is_deterministic_in_quick_mode() { + let fixture = setup(Sizing { quick: true }).expect("fixture writes"); + let a = rep(&fixture).expect("first drain"); + let b = rep(&fixture).expect("second drain"); + assert_eq!(a.fingerprint, b.fingerprint); + assert!(a.fingerprint.starts_with("n="), "shape: {}", a.fingerprint); + } +} diff --git a/crates/aura-bench/src/surfaces/mod.rs b/crates/aura-bench/src/surfaces/mod.rs new file mode 100644 index 0000000..7319919 --- /dev/null +++ b/crates/aura-bench/src/surfaces/mod.rs @@ -0,0 +1,141 @@ +//! The surface framework: a surface is a closure the harness calls once per +//! repetition; `run_reps` applies the measurement discipline (1 discarded +//! warmup + N measured reps, median per metric) and enforces cross-rep +//! fingerprint identity — a surface disagreeing with itself is nondeterminism, +//! reported as an infra error, never averaged away. +//! +//! The CLI glue that wires this framework's public surface into a running +//! binary lands in Task 7; until then nothing outside `#[cfg(test)]` calls +//! into it, so the module is exempted from the dead-code lint here rather +//! than left to trip the workspace's `-D warnings` clippy gate. +#![allow(dead_code)] + +pub mod engine; +pub mod ingest; + +use crate::baseline::Measured; +use std::collections::BTreeMap; + +/// Workload sizing: `quick` is the test/dev mode (tiny inputs, mechanics +/// only); the committed baselines are always full-size. +#[derive(Clone, Copy, Debug)] +pub struct Sizing { + pub quick: bool, +} + +/// One repetition's outcome. +pub struct RepOutcome { + pub metrics: BTreeMap, + pub fingerprint: String, +} + +pub fn median(mut xs: Vec) -> f64 { + xs.sort_by(|a, b| a.partial_cmp(b).expect("no NaN metrics")); + xs[xs.len() / 2] +} + +/// 1 discarded warmup + `reps` measured runs; median per metric key; +/// cross-rep fingerprint identity enforced. +pub fn run_reps( + surface: &str, + reps: u32, + mut rep: impl FnMut() -> Result, +) -> Result { + let reps = reps.max(1); + let _warmup = rep()?; // page/alloc warmup, discarded + let mut runs = Vec::new(); + for _ in 0..reps { + runs.push(rep()?); + } + let fingerprint = runs[0].fingerprint.clone(); + for r in &runs[1..] { + if r.fingerprint != fingerprint { + return Err(format!( + "surface {surface} is nondeterministic across repetitions: {fingerprint:?} vs {:?}", + r.fingerprint + )); + } + } + let mut metrics = BTreeMap::new(); + for key in runs[0].metrics.keys() { + let vals: Vec = runs.iter().filter_map(|r| r.metrics.get(key).copied()).collect(); + metrics.insert(key.clone(), median(vals)); + } + Ok(Measured { surface: surface.to_string(), metrics, fingerprint }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn median_picks_the_middle() { + assert_eq!(median(vec![3.0, 1.0, 2.0]), 2.0); + assert_eq!(median(vec![1.0, 2.0]), 2.0); + } + + /// Task 3's actual deliverable: `run_reps` composed with the real engine + /// surface (a re-bootstrapped SMA-cross harness per call), not a mock + /// closure. Protects the wiring end to end — a warmup plus three fresh + /// bootstraps must agree on the determinism fingerprint and the returned + /// metrics map must carry both keys a `BaselineDoc` for this surface + /// needs (`wall_s`, `bars_per_s`), positive and finite. + #[test] + fn run_reps_drives_the_real_engine_surface_deterministically() { + let measured = run_reps("engine", 3, || engine::rep(Sizing { quick: true })) + .expect("the real engine surface must agree with itself across repeated bootstraps"); + assert_eq!(measured.surface, "engine"); + let bars_per_s = *measured.metrics.get("bars_per_s").expect("bars_per_s metric present"); + assert!(bars_per_s.is_finite() && bars_per_s > 0.0, "bars_per_s: {bars_per_s}"); + assert!(measured.metrics.contains_key("wall_s")); + } + + /// Task 4's actual deliverable: `run_reps` composed with the real ingest + /// surface (a fresh `DataServer` re-opened and re-drained per call). + /// Protects that a warmup plus three fresh drains of the same synthetic + /// archive agree on the fingerprint, and that the metrics map carries + /// the schema the ingest `BaselineDoc` needs. + #[test] + fn run_reps_drives_the_real_ingest_surface_deterministically() { + let fixture = ingest::setup(Sizing { quick: true }).expect("ingest fixture must write"); + let measured = run_reps("ingest", 3, || ingest::rep(&fixture)) + .expect("the real ingest surface must agree with itself across repeated drains"); + assert_eq!(measured.surface, "ingest"); + let bars_per_s = *measured.metrics.get("bars_per_s").expect("bars_per_s metric present"); + assert!(bars_per_s.is_finite() && bars_per_s > 0.0, "bars_per_s: {bars_per_s}"); + assert!(measured.metrics.contains_key("wall_s")); + } + + #[test] + fn run_reps_rejects_cross_rep_fingerprint_drift() { + let mut n = 0; + let r = run_reps("t", 2, move || { + n += 1; + Ok(RepOutcome { metrics: BTreeMap::new(), fingerprint: format!("fp{n}") }) + }); + assert!(r.is_err(), "differing fingerprints must be an error"); + } + + /// The measurement discipline itself: the first (warmup) call is never + /// mixed into the reported metric — its outlier value would otherwise + /// skew the median — and the median is computed independently per + /// metric key across only the measured reps. + #[test] + fn run_reps_discards_warmup_and_medians_only_measured_reps() { + let mut n = 0; + let measured = run_reps("t", 3, move || { + n += 1; + // Call 1 is the discarded warmup: an outlier that must never + // enter the median. Calls 2..=4 (the measured reps) report + // 1.0, 2.0, 3.0, whose median is 2.0. + let v = if n == 1 { 1_000_000.0 } else { (n - 1) as f64 }; + let mut metrics = BTreeMap::new(); + metrics.insert("m".to_string(), v); + Ok(RepOutcome { metrics, fingerprint: "fp".to_string() }) + }) + .expect("agreeing fingerprints must succeed"); + assert_eq!(measured.surface, "t"); + assert_eq!(measured.metrics.get("m"), Some(&2.0), "warmup outlier must not pollute the median"); + assert_eq!(measured.fingerprint, "fp"); + } +}