bench/examples/nornir-bench.rs
//! nornir Bencher harness for **dwarves** — the CI forge's incremental / action-
//! cache hot paths, historized in the warehouse + gated on regression by
//! `nornir bench run dwarves`.
//!
//! What's benched (the previously-unguarded perf surface — the forge scales with
//! the number of pipeline steps it re-plans on every push):
//! - `dwarves.incremental_cold` — `run_pipeline_cached` over a deep step DAG with
//! a COLD cache: every step is planned, hashed and run (the full-build walk).
//! - `dwarves.incremental_warm` — the same DAG against a WARM cache: every step's
//! input content-id is recomputed and hits the action cache, nothing executes
//! (the incremental re-plan the forge does most of the time).
//! - `dwarves.step_input_id` — the raw sha256 content-addressing primitive
//! (`step_input_id`) that keys the cache: ids hashed per second.
//!
//! Run: `nornir bench run dwarves` (or `cargo run --release --example nornir-bench`).
use std::collections::BTreeMap;
use anyhow::{ensure, Result};
use nornir::bench::api::Bencher;
use nornir::bench::BenchResult;
use dwarves::brokkr::dag::Dag;
use dwarves::brokkr::engine::RunRequest;
use dwarves::brokkr::model::{Pipeline, RunKind};
use dwarves::brokkr::PipelineDescriptor;
use dwarves::{
run_pipeline_cached, ActionCache, DemoRunner, IncrementalOptions, step_input_id,
};
/// Deep pipeline scale: `STEPS_HEAVY` steps, each its own stage, chained
/// `stepN needs stepN-1` — the forge's re-plan walk grows with this.
const STEPS_HEAVY: usize = 150;
/// PREVIEW mode (`NORNIR_BENCH_PREVIEW=1`) — the fast verification loop.
///
/// Every registered arm still runs, through the SAME code paths, emitting the SAME
/// metric keys; only the scale (pipeline depth + iteration counts) shrinks, so a
/// broken arm surfaces in seconds instead of minutes. `nornir_bench::api` stamps the
/// emitted run onto `<host>-preview`, so a preview number can never land in — or
/// overwrite — the real host's README row.
fn preview() -> bool {
nornir::bench::api::preview_run()
}
/// Scale knob: the heavy value, or a small one under `NORNIR_BENCH_PREVIEW`.
fn scaled(heavy: u64, light: u64) -> u64 {
if preview() { light } else { heavy }
}
/// Steps in the generated pipeline (heavy 150 / preview 10).
fn steps() -> usize {
scaled(STEPS_HEAVY as u64, 10) as usize
}
/// Generate a `brokkr.toml` with `steps()` variant steps in a dependency chain and
/// parse it into a validated pipeline + its topo-ordered DAG (built once, outside
/// the timed loop).
fn deep_pipeline() -> (Pipeline, Dag) {
let mut toml = String::from("[pipeline]\nname = \"bench\"\n");
for i in 0..steps() {
toml.push_str("[[pipeline.stage]]\n");
toml.push_str(&format!("name = \"s{i}\"\n"));
toml.push_str("[[pipeline.stage.step]]\n");
toml.push_str(&format!("name = \"step{i}\"\n"));
if i > 0 {
toml.push_str(&format!("needs = [\"step{}\"]\n", i - 1));
}
toml.push_str("run = { variant = \"server\" }\n");
}
let d = PipelineDescriptor::from_toml_str(&toml).expect("valid generated pipeline");
d.validate().expect("generated pipeline validates");
let dag = Dag::build(&d.pipeline).expect("dag builds");
(d.pipeline, dag)
}
fn req() -> RunRequest {
RunRequest { project: "bench".into(), trigger: "manual".into(), ..Default::default() }
}
fn metrics(pairs: &[(&str, serde_json::Value)]) -> serde_json::Map<String, serde_json::Value> {
pairs.iter().map(|(k, v)| (k.to_string(), v.clone())).collect()
}
/// Full cold-cache walk: every step planned + hashed + run (the demo runner greens
/// each step deterministically, so this isolates the forge's per-step overhead, not
/// a toolchain).
struct IncrementalCold;
impl Bencher for IncrementalCold {
fn id(&self) -> &'static str {
"dwarves.incremental_cold"
}
fn run(&self) -> Result<BenchResult> {
let (p, dag) = deep_pipeline();
let runner = DemoRunner;
let opts = IncrementalOptions::cached();
let iters = scaled(200, 3);
let t = std::time::Instant::now();
let mut ran = 0usize;
for k in 0..iters {
// A fresh cache each iter ⇒ genuinely cold (every step executes).
let mut cache = ActionCache::in_memory();
let out = run_pipeline_cached(&p, &dag, &runner, &req(), &format!("cold-{k}"), &mut cache, &opts);
ran = out.ran();
}
let s = t.elapsed().as_secs_f64();
ensure!(ran == steps(), "cold run should execute every step");
let steps_per_sec = (steps() as u64 * iters) as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("steps", steps().into()),
("steps_per_sec", steps_per_sec.round().into()),
("run_ms", (s * 1000.0 / iters as f64).into()),
("iters", iters.into()),
("elapsed_ms", (s * 1000.0).round().into()),
]),
))
}
}
/// Warm-cache re-plan: recompute every step's content-id and serve it from the
/// action cache (no execution) — the incremental path the forge takes on most
/// pushes.
struct IncrementalWarm;
impl Bencher for IncrementalWarm {
fn id(&self) -> &'static str {
"dwarves.incremental_warm"
}
fn run(&self) -> Result<BenchResult> {
let (p, dag) = deep_pipeline();
let runner = DemoRunner;
let opts = IncrementalOptions::cached();
// Warm the cache once with a full run.
let mut cache = ActionCache::in_memory();
run_pipeline_cached(&p, &dag, &runner, &req(), "warm", &mut cache, &opts);
let iters = scaled(500, 5);
let t = std::time::Instant::now();
let mut hits = 0usize;
for k in 0..iters {
let out = run_pipeline_cached(&p, &dag, &runner, &req(), &format!("warm-{k}"), &mut cache, &opts);
hits = out.cache_hits();
}
let s = t.elapsed().as_secs_f64();
ensure!(hits == steps(), "warm run should hit the cache on every step");
let steps_per_sec = (steps() as u64 * iters) as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("steps", steps().into()),
("cache_hits_per_sec", steps_per_sec.round().into()),
("replan_ms", (s * 1000.0 / iters as f64).into()),
("iters", iters.into()),
("elapsed_ms", (s * 1000.0).round().into()),
]),
))
}
}
/// The raw content-addressing primitive: how many step input-ids (sha256 over the
/// canonical {argv, matrix, upstream pins, source} framing) can be hashed per sec.
struct StepInputId;
impl Bencher for StepInputId {
fn id(&self) -> &'static str {
"dwarves.step_input_id"
}
fn run(&self) -> Result<BenchResult> {
let matrix: BTreeMap<String, String> =
[("rust".to_string(), "1.85".to_string()), ("os".to_string(), "linux".to_string())]
.into_iter()
.collect();
let pins = vec!["sha256:aaaa".to_string(), "sha256:bbbb".to_string()];
let iters = scaled(500_000, 5_000);
let t = std::time::Instant::now();
let mut last = String::new();
for i in 0..iters {
// Vary the argv each iter so the hash can't be hoisted/cached.
let run = RunKind::Variant(format!("server-{}", i & 0xff));
last = step_input_id(&run, &matrix, &pins, "rev-abc123");
}
let s = t.elapsed().as_secs_f64();
ensure!(last.starts_with("sha256:"), "input-id is a sha256");
let ids_per_sec = iters as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("ids_per_sec", ids_per_sec.round().into()),
("iters", iters.into()),
("elapsed_ms", (s * 1000.0).round().into()),
]),
))
}
}
/// DAG assembly: the petgraph topo-sort + cycle-detection walk that turns a
/// validated pipeline into an ordered run plan. Sindri rebuilds this on EVERY
/// dispatch (`Dag::build(®.pipeline)`), so its cost scales the scheduler — yet
/// every other arm builds the DAG once, OUTSIDE its timed loop, leaving this
/// primitive unguarded. This arm times `Dag::build` in isolation (the pipeline is
/// parsed once, outside the loop) so a regression in the topo walk is caught.
struct DagBuild;
impl Bencher for DagBuild {
fn id(&self) -> &'static str {
"dwarves.dag_build"
}
fn run(&self) -> Result<BenchResult> {
let (p, _dag) = deep_pipeline();
let iters = scaled(20_000, 200);
let t = std::time::Instant::now();
let mut last_len = 0usize;
for _ in 0..iters {
let dag = Dag::build(&p).expect("dag builds");
last_len = dag.order.len();
}
let s = t.elapsed().as_secs_f64();
ensure!(last_len == steps(), "the topo order carries every step");
let builds_per_sec = iters as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("steps", steps().into()),
("builds_per_sec", builds_per_sec.round().into()),
("build_ms", (s * 1000.0 / iters as f64).into()),
("iters", iters.into()),
("elapsed_ms", (s * 1000.0).round().into()),
]),
))
}
}
// ── holger artifact-store push/resolve (feature `holger`) ───────────────────────
// The forge's byte half: a produced artifact is pushed to holger (sealed via znippy)
// and resolved back by content id — the write + read legs a real build step drives so
// a cache hit on ANY machine materializes the real output. Behind the `holger` feature
// (which pulls holger-server-lib + znippy in-process); the Loki bencher runs it with
// `--features holger`. A build artifact is a binary, so blobs are a realistic size.
#[cfg(feature = "holger")]
mod holger_arms {
use super::*;
use dwarves::artifact_store::{ArtifactStore, HolgerArtifactStore};
/// A throwaway holger data dir, unique per bench arm + pid so concurrent runs
/// never collide. Cleaned up when the guard drops.
struct Scratch {
root: std::path::PathBuf,
}
impl Scratch {
fn new(tag: &str) -> Self {
let root = std::env::temp_dir().join(format!(
"dwarves-bench-holger-{}-{}-{}",
tag,
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0)
));
Self { root }
}
}
impl Drop for Scratch {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.root);
}
}
const BLOB_BYTES: usize = 256 * 1024;
fn distinct_blobs(n: usize) -> Vec<Vec<u8>> {
(0..n)
.map(|i| {
let mut b = vec![0u8; BLOB_BYTES];
b[..8].copy_from_slice(&(i as u64).to_le_bytes());
b
})
.collect()
}
/// holger push: how fast the forge seals produced artifact bytes into holger's
/// immutable znippy store (content-id + znippy seal + append), in objects/sec + MB/s.
pub struct HolgerPut;
impl Bencher for HolgerPut {
fn id(&self) -> &'static str {
"dwarves.holger_put"
}
fn run(&self) -> Result<BenchResult> {
let scratch = Scratch::new("put");
let store = HolgerArtifactStore::open(&scratch.root, "artifacts")?;
let n = scaled(500, 20) as usize;
let blobs = distinct_blobs(n);
let t = std::time::Instant::now();
let mut last = String::new();
for b in &blobs {
last = store.put_bytes(b)?;
}
let s = t.elapsed().as_secs_f64();
ensure!(last.starts_with("sha256:"), "put returns a content id");
let total_mb = (n * BLOB_BYTES) as f64 / (1024.0 * 1024.0);
Ok(BenchResult::measured(
self.id(),
metrics(&[
("objects", n.into()),
("blob_kib", (BLOB_BYTES / 1024).into()),
("puts_per_sec", (n as f64 / s).round().into()),
("put_mb_per_sec", (total_mb / s).round().into()),
("elapsed_ms", (s * 1000.0).round().into()),
]),
))
}
}
/// holger resolve: how fast a cache HIT reads an artifact back out of holger by
/// content id (the read leg a reused step drives), in objects/sec + MB/s.
pub struct HolgerGet;
impl Bencher for HolgerGet {
fn id(&self) -> &'static str {
"dwarves.holger_get"
}
fn run(&self) -> Result<BenchResult> {
let scratch = Scratch::new("get");
let store = HolgerArtifactStore::open(&scratch.root, "artifacts")?;
let n = scaled(500, 20) as usize;
let ids: Vec<String> =
distinct_blobs(n).iter().map(|b| store.put_bytes(b).unwrap()).collect();
let t = std::time::Instant::now();
let mut bytes_read = 0u64;
for id in &ids {
let got = store.get(id)?.expect("stored object present");
bytes_read += got.len() as u64;
}
let s = t.elapsed().as_secs_f64();
ensure!(bytes_read == (n * BLOB_BYTES) as u64, "all bytes resolved back");
let total_mb = bytes_read as f64 / (1024.0 * 1024.0);
Ok(BenchResult::measured(
self.id(),
metrics(&[
("objects", n.into()),
("blob_kib", (BLOB_BYTES / 1024).into()),
("gets_per_sec", (n as f64 / s).round().into()),
("get_mb_per_sec", (total_mb / s).round().into()),
("elapsed_ms", (s * 1000.0).round().into()),
]),
))
}
}
}
nornir::register_bench!(IncrementalCold);
nornir::register_bench!(IncrementalWarm);
nornir::register_bench!(StepInputId);
nornir::register_bench!(DagBuild);
#[cfg(feature = "holger")]
nornir::register_bench!(holger_arms::HolgerPut);
#[cfg(feature = "holger")]
nornir::register_bench!(holger_arms::HolgerGet);
fn main() -> Result<()> {
nornir::bench::api::run_main_json()
}