plugins/native/znippy-plugin-git/examples/resolve_scaling.rs
//! **What one pack costs at N workers** — the measurement behind
//! [`znippy_plugin_git::forest`].
//!
//! It maps a real `.pack`, walks it once, and then resolves it once per worker
//! count, reporting for each arm:
//!
//! * wall time, objects/s and MB of pack per second,
//! * the process's `VmHWM` **delta** over the arm (the mapped pack is in
//! `VmHWM` too, so the absolute number is not the resolver's footprint —
//! `mem_available` on the same row is what the watchdog acts on),
//! * `ResolveReport` in full: re-inflations, evictions, peak held bases,
//! stalls at the lookahead ceiling,
//! * CPU utilisation over the arm, as `utime+stime / wall` — which on a
//! 4-vCPU box is the number that says whether three of them stopped idling,
//! * and a **fingerprint of the ordered output**: a SHA-256 over the exact
//! sequence of rows and sink calls. Two arms with the same fingerprint
//! produced byte-identical output; that is the determinism proof, on the
//! real pack rather than on a fixture.
//!
//! ```text
//! cargo run --release --example resolve_scaling -- <pack> [workers,...]
//! taskset -c 0-3 cargo run --release --example resolve_scaling -- \
//! ~/scratch-t14s/linux/linux.git/objects/pack/pack-….pack 1,2,4
//! ```
//!
//! The pack is resolved against [`NoBases`], so a **thin** pack is refused by
//! name — which is correct: this measures a repository's own pack, and a
//! repository's own pack is self-contained.
use std::io::Write;
use std::sync::Mutex;
use std::time::Instant;
use anyhow::{Result, anyhow, bail};
use sha2::{Digest, Sha256};
use znippy_plugin_git::exploded::PayloadSink;
use znippy_plugin_git::object::{GitHashKind, GitObjectKind};
use znippy_plugin_git::resolve::{NoBases, ResolveLimits, Resolved, resolve_streaming};
/// A sink that keeps nothing but the **shape** of what it was handed, and one
/// that mimics the shipped `ExplodePolicy::Commits` by wanting only commits and
/// tags — so the arm measures the ordered hand-over the appliance actually does.
struct Fingerprint {
h: Mutex<Sha256>,
commits_only: bool,
seen: std::sync::atomic::AtomicU64,
/// 🔴 **OFF for the timing arms.** Hashing every row costs a mutex and a
/// SHA-256 update per object on the ONE thread the ordered drain runs on,
/// which is a serial section the resolver does not have — MEASURED, the
/// kernel pack: it is the difference between 10 396 obj/s and the rate the
/// appliance shows for the same code. Timing and fingerprinting are
/// separate arms for that reason.
on: bool,
}
impl Fingerprint {
fn feed(&self, tag: u8, parts: &[&[u8]]) {
if !self.on {
return;
}
let mut h = self.h.lock().unwrap_or_else(|e| e.into_inner());
h.update([tag]);
for p in parts {
h.update((p.len() as u64).to_le_bytes());
h.update(p);
}
}
}
impl PayloadSink for Fingerprint {
fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
self.seen.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if self.on {
// The payload's own hash, not the payload: a kernel's commits are
// 400 MB and the fingerprint must fit in a register file.
let mut p = Sha256::new();
p.update(payload);
self.feed(b'E', &[oid, &[kind as u8], &p.finalize()]);
}
Ok(())
}
fn wants(&self, kind: GitObjectKind) -> bool {
!self.commits_only || matches!(kind, GitObjectKind::Commit | GitObjectKind::Tag)
}
}
fn cpu_ms() -> u64 {
// utime + stime of the whole process, in ms. Fields 14 and 15 of
// /proc/self/stat, after the parenthesised comm.
let Ok(s) = std::fs::read_to_string("/proc/self/stat") else {
return 0;
};
let Some(rest) = s.rsplit_once(close_paren()).map(|(_, r)| r) else {
return 0;
};
let f: Vec<&str> = rest.split_whitespace().collect();
let tick = 100u64; // CLK_TCK on every Linux this ships to
let get = |i: usize| f.get(i).and_then(|x| x.parse::<u64>().ok()).unwrap_or(0);
// rest[0] is `state`, so utime is index 11 and stime index 12.
(get(11) + get(12)) * (1000 / tick)
}
/// The `comm` field of `/proc/self/stat` is parenthesised and may itself
/// contain spaces and parentheses, so the fields after it are found from the
/// LAST close-paren rather than by counting from the left.
fn close_paren() -> char {
')'
}
fn hwm_bytes() -> u64 {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|s| {
s.lines().find_map(|l| {
l.strip_prefix("VmHWM:")
.and_then(|v| v.split_whitespace().next())
.and_then(|v| v.parse::<u64>().ok())
})
})
.unwrap_or(0)
* 1024
}
fn mib(n: u64) -> String {
format!("{:.1} MiB", n as f64 / 1048576.0)
}
fn main() -> Result<()> {
let mut args = std::env::args().skip(1);
let path = args
.next()
.ok_or_else(|| anyhow!("usage: resolve_scaling <pack> [workers,…] [--commits-only]"))?;
let arms: Vec<usize> = match args.next() {
Some(a) if !a.starts_with("--") => a
.split(',')
.map(|x| x.trim().parse::<usize>())
.collect::<Result<_, _>>()?,
_ => vec![1, 2, 4],
};
let commits_only = std::env::args().any(|a| a == "--commits-only");
let fingerprint = !std::env::args().any(|a| a == "--no-fingerprint");
// **The appliance's ceiling, on a box that is not the appliance.** The
// resolver's own default is 25 % of this machine's `MemTotal`, and oden has
// 499 GB — so a run here would be unbounded where the appliance's is 4 GiB.
// Named on the command line rather than inferred, so the number in the
// report is the number that was in force.
let base_budget: u64 = std::env::var("SCALING_BASE_BUDGET_BYTES")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(0);
let f = std::fs::File::open(&path)?;
let map = unsafe { memmap2::Mmap::map(&f)? };
let pack: &[u8] = ↦
println!(
"pack {path}\n {} bytes ({}), arms {arms:?}, sink wants {}, base budget {}",
pack.len(),
mib(pack.len() as u64),
if commits_only { "commits+tags" } else { "everything" },
if base_budget == 0 { "unbounded".to_string() } else { mib(base_budget) }
);
if !fingerprint {
println!(" --no-fingerprint: TIMING ARM, the ordered output is not hashed");
}
let t0 = Instant::now();
let w = znippy_plugin_git::pack_walk::walk(pack, GitHashKind::Sha1.oid_len())?;
let deltas = w
.entries
.iter()
.filter(|e| !matches!(e.delta_base, znippy_plugin_git::pack_walk::DeltaBase::None))
.count();
println!(
" walk {} ms: {} entries, {deltas} deltas",
t0.elapsed().as_millis(),
w.entries.len()
);
let forest = znippy_plugin_git::forest::DeltaForest::build(&w)?;
println!(
" forest: {} whole roots, {} ref roots — {} independent subtrees",
forest.whole_roots().len(),
forest.ref_roots().len(),
forest.whole_roots().len() + forest.ref_roots().len()
);
let mut prints: Vec<String> = Vec::new();
let mut first_fp: Option<String> = None;
for workers in arms {
let sink = Fingerprint {
h: Mutex::new(Sha256::new()),
commits_only,
seen: std::sync::atomic::AtomicU64::new(0),
on: fingerprint,
};
let hwm0 = hwm_bytes();
let cpu0 = cpu_ms();
let t = Instant::now();
let mut rows_seen = 0u64;
let report = resolve_streaming(
pack,
&w,
GitHashKind::Sha1,
0,
&NoBases,
&sink,
ResolveLimits {
workers,
base_budget,
..ResolveLimits::none()
},
&mut |rows: &mut Vec<Resolved>| {
for r in rows.iter() {
sink.feed(
b'R',
&[
&r.oid,
&r.offset.to_le_bytes(),
&r.len.to_le_bytes(),
&[r.stored_type as u8, r.kind as u8],
&r.uncompressed_size.to_le_bytes(),
&r.delta_base.to_le_bytes(),
],
);
}
rows_seen += rows.len() as u64;
rows.clear();
Ok(())
},
)?;
let wall = t.elapsed();
let cpu = cpu_ms().saturating_sub(cpu0);
let hwm = hwm_bytes();
let fp = hex::encode(
sink.h
.into_inner()
.unwrap_or_else(|e| e.into_inner())
.finalize(),
);
let fp = if fingerprint { fp } else { "off".repeat(11) };
let same = match &first_fp {
None => {
first_fp = Some(fp.clone());
"reference".to_string()
}
Some(f0) if *f0 == fp => "IDENTICAL".to_string(),
Some(_) => "DIFFERENT".to_string(),
};
let secs = wall.as_secs_f64().max(1e-9);
let line = format!(
"workers={workers:<3} wall={:>8.1}s {:>9.0} obj/s {:>7.1} MB/s cpu={:>6.2}x \
VmHWM {} (+{}) peak_bases={} reinfl={} evict={} stalls={} batches={} explodes={} fp={}… {}",
secs,
rows_seen as f64 / secs,
pack.len() as f64 / secs / 1e6,
cpu as f64 / 1000.0 / secs,
mib(hwm),
mib(hwm.saturating_sub(hwm0)),
mib(report.peak_base_bytes),
report.reinflations,
report.evictions,
report.stalls,
report.batches,
sink.seen.load(std::sync::atomic::Ordering::Relaxed),
&fp[..16],
same
);
println!("{line}");
std::io::stdout().flush().ok();
prints.push(line);
if rows_seen != w.entries.len() as u64 {
bail!("{rows_seen} rows for {} entries", w.entries.len());
}
}
println!("\n─── summary ───");
for l in prints {
println!("{l}");
}
Ok(())
}