bench/examples/stage-probe.rs
//! The stage decomposition, printed. A diagnostic, not a persisted row.
//!
//! ```text
//! # a clone and a negotiated fetch out of one bare repository
//! stage-probe fetch <repo.git> <want-oid> [have-oid ...]
//!
//! # the same fetch at N clients at once
//! stage-probe concurrent <repo.git> <width> <want-oid> [have-oid ...]
//!
//! # one push of <tip> from <source.git> into a fresh <target.git>
//! stage-probe push <source.git> <target.git> <tip-oid> refs/heads/<name>
//!
//! # the wire bytes this build produces, hashed — the fidelity check
//! stage-probe wirehash <repo.git> <want-oid> [have-oid ...]
//!
//! # the same over MANY requests out of ONE open store, which is the only
//! # shape that can see a selection that is wrong for a commit this fetch
//! # did not ask for. One line per request: `<want> [have ...]`.
//! stage-probe wirehash-set <repo.git> <pairs-file>
//! ```
//!
//! `wirehash` is what proves the instrumentation does not change what it
//! measures, and it is the half `gunnar.stage_overhead` cannot do from inside
//! one process: that arm compares *recording* against *not recording*, which is
//! blind to a probe that perturbed the code generation around it. This compares
//! **two builds**:
//!
//! ```text
//! cargo build --release --example stage-probe --features count-alloc # probes IN
//! cargo build --release --example stage-probe --no-default-features # probes OUT
//! ```
//!
//! and every corpus repository must print the same digest out of both. MEASURED
//! the desktop 2026-08-04: 19 of 19 identical. A benchmark whose instrument moves the
//! bytes is measuring itself.
//!
//! Output is one JSON object per line so `jq` can slice it; the human table is
//! on stderr.
use anyhow::{ensure, Context, Result};
use gix_hash::ObjectId;
use gunnar_bench::stages::{self, StageRun};
/// Always on here. See `gunnar_bench::stages::CountingAlloc` for why it is
/// *not* always on in `nornir-bench`.
#[global_allocator]
static ALLOC: stages::CountingAlloc = stages::CountingAlloc;
fn oid(text: &str) -> Result<ObjectId> {
ObjectId::from_hex(text.trim().as_bytes()).with_context(|| format!("{text:?} is not an oid"))
}
fn table(label: &str, run: &StageRun) {
eprintln!(
"\n── {label} ── {} objects, {} wire bytes, {:.2} ms wall, {:.2} ms cpu \
(u {:.2} / s {:.2}) [cpu={} rss={} alloc={}]",
run.objects,
run.wire_bytes,
run.wall_ns as f64 / 1e6,
run.cpu_ns as f64 / 1e6,
run.user_ns as f64 / 1e6,
run.sys_ns as f64 / 1e6,
run.instruments.cpu,
run.instruments.rss,
run.instruments.alloc,
);
eprintln!(
"{:<24} {:>7} {:>7} {:>7} {:>10} {:>10} {:>10} {:>9} {:>12}",
"stage", "wall%", "cpu%", "peak%", "wall_ms", "cpu_ms", "alloc_MiB", "reads", "read_MiB"
);
for row in &run.rows {
eprintln!(
"{:<24} {:>6.2}% {:>6.2}% {:>6.2}% {:>10.3} {:>10.3} {:>10.2} {:>9} {:>12.2}",
row.name,
row.wall_frac * 100.0,
row.cpu_frac * 100.0,
row.peak_frac * 100.0,
row.tally.excl_wall_ns as f64 / 1e6,
(row.tally.excl_user_ns + row.tally.excl_sys_ns) as f64 / 1e6,
row.tally.excl_alloc_bytes as f64 / 1048576.0,
row.store_reads,
row.store_read_bytes as f64 / 1048576.0,
);
}
eprintln!(
"owner: CPU = {}, memory = {}; the table accounts for {:.1}% of the process's \
{:.2} ms of CPU over the same window",
run.cpu_owner(),
run.memory_owner(),
run.attributed_frac() * 100.0,
run.process_cpu_ns as f64 / 1e6,
);
}
fn json(label: &str, run: &StageRun) {
let rows: Vec<serde_json::Value> = run
.rows
.iter()
.map(|r| {
serde_json::json!({
"stage": r.name,
"wall_frac": r.wall_frac,
"cpu_frac": r.cpu_frac,
"peak_frac": r.peak_frac,
"calls": r.tally.calls,
"excl_wall_ns": r.tally.excl_wall_ns,
"incl_wall_ns": r.tally.incl_wall_ns,
"excl_user_ns": r.tally.excl_user_ns,
"excl_sys_ns": r.tally.excl_sys_ns,
"excl_alloc_count": r.tally.excl_alloc_count,
"excl_alloc_bytes": r.tally.excl_alloc_bytes,
"peak_growth_bytes": r.tally.peak_growth_bytes,
"retained_bytes": r.tally.retained_bytes,
"rss_growth_bytes": r.tally.rss_growth_bytes,
"store_reads": r.store_reads,
"store_read_bytes": r.store_read_bytes,
})
})
.collect();
println!(
"{}",
serde_json::json!({
"label": label,
"objects": run.objects,
"wire_bytes": run.wire_bytes,
"wall_ns": run.wall_ns,
"cpu_ns": run.cpu_ns,
"user_ns": run.user_ns,
"sys_ns": run.sys_ns,
"process_cpu_ns": run.process_cpu_ns,
"attributed_frac": run.attributed_frac(),
"cpu_owner": run.cpu_owner(),
"memory_owner": run.memory_owner(),
"instruments": {
"cpu": run.instruments.cpu,
"rss": run.instruments.rss,
"alloc": run.instruments.alloc,
},
"stages": rows,
})
);
}
fn main() -> Result<()> {
let args: Vec<String> = std::env::args().skip(1).collect();
ensure!(!args.is_empty(), "usage: see the module docs");
match args[0].as_str() {
"fetch" => {
ensure!(args.len() >= 3, "fetch <repo.git> <want> [have ...]");
let repo = std::path::PathBuf::from(&args[1]);
let want = oid(&args[2])?;
let haves: Vec<ObjectId> = args[3..].iter().map(|s| oid(s)).collect::<Result<_>>()?;
let run = stages::fetch(&repo, &want, &haves)?;
let label = if haves.is_empty() {
"clone"
} else {
"negotiated fetch"
};
table(label, &run);
json(label, &run);
}
"concurrent" => {
ensure!(
args.len() >= 4,
"concurrent <repo.git> <width> <want> [have ...]"
);
let repo = std::path::PathBuf::from(&args[1]);
let width: usize = args[2].parse()?;
let want = oid(&args[3])?;
let haves: Vec<ObjectId> = args[4..].iter().map(|s| oid(s)).collect::<Result<_>>()?;
let run = stages::concurrent_fetch(&repo, &want, &haves, width)?;
let label = format!("fetch x{width}, per request");
table(&label, &run);
json(&label, &run);
}
"push" => {
ensure!(
args.len() >= 5,
"push <source.git> <target.git> <tip> <ref>"
);
let run = stages::push(
std::path::Path::new(&args[1]),
std::path::Path::new(&args[2]),
&oid(&args[3])?,
&args[4],
)?;
table("push", &run);
json("push", &run);
}
"wirehash" => {
ensure!(args.len() >= 3, "wirehash <repo.git> <want> [have ...]");
let repo = std::path::PathBuf::from(&args[1]);
let want = oid(&args[2])?;
let haves: Vec<ObjectId> = args[3..].iter().map(|s| oid(s)).collect::<Result<_>>()?;
// Served twice in the SAME binary: once with the probes recording
// and once with them off. Identical bytes is necessary but not
// sufficient — the compile-time-off build is the other half, and it
// is compared by running this same subcommand out of a `stages`-off
// build and diffing the digest.
for enabled in [true, false] {
if enabled {
gunnar_stage::enable();
} else {
gunnar_stage::disable();
}
let digest = stages::wire_digest(&repo, &want, &haves)?;
println!(
"{}",
serde_json::json!({
"recording": enabled,
"repo": repo.to_string_lossy(),
"digest": digest.0,
"wire_bytes": digest.1,
"objects": digest.2,
})
);
}
}
"wirehash-set" => {
ensure!(args.len() == 3, "wirehash-set <repo.git> <pairs-file>");
let repo = std::path::PathBuf::from(&args[1]);
let text = std::fs::read_to_string(&args[2])
.with_context(|| format!("reading the request set {}", &args[2]))?;
let mut requests: Vec<(ObjectId, Vec<ObjectId>)> = Vec::new();
for line in text.lines() {
let mut ids = line.split_whitespace();
let Some(want) = ids.next() else { continue };
requests.push((oid(want)?, ids.map(oid).collect::<Result<_>>()?));
}
ensure!(
requests.len() > 1,
"a one-request set is exactly the blindness `wirehash-set` exists to remove: a \
fetch exercises only its own tip, so one digest cannot see an index that is \
wrong for every other commit. Give it distinct wants."
);
// The combined digest is over the per-request digests **and their
// order**, so a corruption that swapped two answers is a different
// number rather than the same one.
let mut combined: u64 = 0xcbf2_9ce4_8422_2325;
let mut total_bytes = 0u64;
let mut total_objects = 0u64;
for ((want, haves), (digest, bytes, objects)) in requests
.iter()
.zip(stages::wire_digest_set(&repo, &requests)?)
{
for byte in digest.as_bytes() {
combined ^= u64::from(*byte);
combined = combined.wrapping_mul(0x0000_0100_0000_01b3);
}
total_bytes += bytes;
total_objects += objects;
println!(
"{}",
serde_json::json!({
"want": want.to_string(),
"haves": haves.len(),
"digest": digest,
"wire_bytes": bytes,
"objects": objects,
})
);
}
println!(
"{}",
serde_json::json!({
"requests": requests.len(),
"combined": format!("{combined:016x}"),
"wire_bytes": total_bytes,
"objects": total_objects,
})
);
}
other => anyhow::bail!("unknown subcommand {other:?}"),
}
Ok(())
}