plugins/native/znippy-plugin-git/examples/lazy_explode_cost.rs
//! **What a push costs on disk, measured — the lane Q harness.**
//!
//! Pushes one real packfile into a fresh store and reports, in order:
//!
//! 1. bytes under the store at **ack** (`put_pack` returned),
//! 2. bytes and wall time at **index-complete**,
//! 3. first-read against second-read latency for a sample of objects,
//! 4. the **plateau**: bytes under the store while every object in the
//! repository is read once, which is the full-history browse that would
//! otherwise rebuild the eager tier.
//!
//! Every figure is `du`-style applied output — `st_blocks * 512`, the bytes the
//! filesystem is really holding — with the logical size beside it. Both,
//! because they disagree while a file is being written: ext4's speculative
//! preallocation charged the 8.73 GB pack 12.36 GB of blocks at the instant
//! `put_pack` returned and gave it back later. Reporting only one of them makes
//! an ack look 1.42× the pack on one run and 1.05× on the next.
//!
//! ```text
//! ZNIPPY_GIT_EXPLODE=graph ZNIPPY_GIT_EXPLODE_BUDGET=8G \
//! cargo run --release -p znippy-plugin-git --example lazy_explode_cost -- \
//! <pack file> <store dir> [sample] [scan]
//! ```
//!
//! `sample` is how many objects the first/second-read table is built from
//! (default 100); `scan` is `all` to read every object (the plateau run) or a
//! count.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::time::Instant;
use anyhow::{Context, Result};
use git_storage_trait::GitOps as _;
use znippy_plugin_git::git_ops::GitStore;
use znippy_plugin_git::object::GitObjectKind;
fn main() -> Result<()> {
let mut args = std::env::args().skip(1);
let pack_path = PathBuf::from(args.next().context("usage: <pack> <dir> [sample] [scan]")?);
let dir = PathBuf::from(args.next().context("usage: <pack> <dir> [sample] [scan]")?);
let sample: usize = args.next().unwrap_or_else(|| "100".into()).parse().unwrap_or(100);
let scan = args.next().unwrap_or_else(|| "0".into());
// `seal` mode: open an existing store, seal the index tail into its
// reserved sections and report the index's own bytes per object. Separate
// from the push run because sealing is `repo gc`'s job, not a push's, and
// measuring it inside the push run would report a number no push produces.
if scan == "seal" {
let store = GitStore::open(&dir, "alice").context("opening the store")?;
let before = usage(&dir)?;
let objects = store.object_count().max(1);
let t = Instant::now();
let sections = store.seal().context("sealing")?;
let after = usage(&dir)?;
let idx_before = before.get("objects.tail").map(|(b, _)| *b).unwrap_or(0);
let idx_after: u64 = after
.iter()
.filter(|(k, _)| k.starts_with("objects.tail") || k.starts_with("objects.idx"))
.map(|(_, (b, _))| *b)
.sum();
println!(
"SEAL {} sections in {:.1} s; index {} B → {} B over {} objects = {:.1} → {:.1} B/object",
sections.len(),
t.elapsed().as_secs_f64(),
idx_before,
idx_after,
objects,
idx_before as f64 / objects as f64,
idx_after as f64 / objects as f64,
);
for (name, (bytes, _)) in &after {
if *bytes > 0 {
println!(" {name:<28} {bytes:>14} B");
}
}
return Ok(());
}
if dir.exists() {
std::fs::remove_dir_all(&dir).with_context(|| format!("clearing {}", dir.display()))?;
}
std::fs::create_dir_all(&dir)?;
// The pack, mapped rather than read: 8 GiB copied into a `Vec` is 8 GiB of
// anon this measurement did not want to be about.
let f = std::fs::File::open(&pack_path)
.with_context(|| format!("opening {}", pack_path.display()))?;
let map = unsafe { memmap2::Mmap::map(&f)? };
let pack_bytes = map.len() as u64;
println!("pack {:>14} B {}", pack_bytes, pack_path.display());
println!("store {}", dir.display());
println!(
"policy {} budget {}",
std::env::var("ZNIPPY_GIT_EXPLODE").unwrap_or_else(|_| "(default)".into()),
std::env::var("ZNIPPY_GIT_EXPLODE_BUDGET").unwrap_or_else(|_| "(default)".into()),
);
println!("loadavg {}", loadavg());
let store = GitStore::open(&dir, "alice").context("opening the store")?;
// ── (1) the ACK ─────────────────────────────────────────────────────────
let t0 = Instant::now();
let tx = store.put_pack(&map).context("the push")?;
let ack = t0.elapsed();
let at_ack = usage(&dir)?;
report("ACK", pack_bytes, &at_ack, ack.as_secs_f64(), rss_peak());
// ── (2) INDEX-COMPLETE ──────────────────────────────────────────────────
store.wait_indexed();
let indexed = t0.elapsed();
let at_index = usage(&dir)?;
report("INDEX", pack_bytes, &at_index, indexed.as_secs_f64(), rss_peak());
println!(
"objects {} explode {:?}",
store.object_count(),
store.explode_cache_use()
);
// ── (3) FIRST vs SECOND read ────────────────────────────────────────────
let extent = tx.extent.context("the push reported no extent")?;
let t = Instant::now();
let oids = store.oids_in_extent(extent).context("listing the pack's oids")?;
println!(
"oids {} listed in {:.1} s",
oids.len(),
t.elapsed().as_secs_f64()
);
let picks: Vec<&Vec<u8>> = stride(&oids, sample);
let mut cold = Vec::with_capacity(picks.len());
// Split by what the read actually PAID, which is the only split that means
// anything once the cache admits on rebuild cost: a shallow object is never
// cached, so its cold latency is its latency for ever.
let mut shallow = Vec::new();
let mut deep = Vec::new();
let mut kinds: BTreeMap<&'static str, usize> = BTreeMap::new();
let mut depths: BTreeMap<u32, usize> = BTreeMap::new();
for oid in &picks {
let t = Instant::now();
let (kind, _payload, cost) = store
.content_costed(oid)?
.context("an object of the pack is absent")?;
let us = t.elapsed().as_secs_f64() * 1e6;
cold.push(us);
*kinds.entry(kind_name(kind)).or_default() += 1;
if let Some(c) = cost {
*depths.entry(c.depth).or_default() += 1;
if c.depth <= znippy_plugin_git::exploded_cache::CHEAP_DEPTH {
shallow.push(us);
} else {
deep.push(us);
}
}
}
store.explode_cache_maintain()?;
let mut warm = Vec::with_capacity(picks.len());
for oid in &picks {
let t = Instant::now();
store.content(oid)?.context("an object went missing between reads")?;
warm.push(t.elapsed().as_secs_f64() * 1e6);
}
println!(
"read {:>4} FIRST p50 {:.1} µs p90 {:.1} µs p99 {:.1} µs SECOND p50 {:.1} µs \
p90 {:.1} µs kinds {:?}",
picks.len(),
pct(&mut cold.clone(), 50.0),
pct(&mut cold.clone(), 90.0),
pct(&mut cold.clone(), 99.0),
pct(&mut warm.clone(), 50.0),
pct(&mut warm.clone(), 90.0),
kinds
);
println!(
"cold-by-cost SHALLOW (depth <= {}) n={} p50 {:.1} µs p99 {:.1} µs DEEP n={} p50 \
{:.1} µs p99 {:.1} µs depths {:?}",
znippy_plugin_git::exploded_cache::CHEAP_DEPTH,
shallow.len(),
pct(&mut shallow.clone(), 50.0),
pct(&mut shallow.clone(), 99.0),
deep.len(),
pct(&mut deep.clone(), 50.0),
pct(&mut deep.clone(), 99.0),
depths
);
let band = store.explode_cache_use();
println!(
"admission {} written back, {} read directly and NOT cached ({:.1} % of {} \
rebuilt reads)",
band.written_back,
band.direct_reads,
100.0 * band.direct_reads as f64 / (band.direct_reads + band.written_back).max(1) as f64,
band.direct_reads + band.written_back,
);
// ── (4) THE PLATEAU ─────────────────────────────────────────────────────
let want: usize = if scan == "all" {
oids.len()
} else {
scan.parse().unwrap_or(0)
};
if want > 0 {
let t = Instant::now();
let mut peak = 0u64;
let mut peak_at = 0usize;
let mut done = 0usize;
let mut missing = 0usize;
for (i, oid) in oids.iter().enumerate() {
if i >= want {
break;
}
match store.content(oid) {
Ok(Some(_)) => done += 1,
Ok(None) => missing += 1,
Err(e) => {
eprintln!("read {}: {e:#}", hex(oid));
missing += 1;
}
}
if i % 20_000 == 19_999 {
store.explode_cache_maintain()?;
let now = usage(&dir)?;
let total: u64 = now.values().map(|(b, _)| b).sum();
if total > peak {
peak = total;
peak_at = i + 1;
}
println!(
" scan {:>9}/{:<9} {:>7.1} s store {:>8.2} GiB ({:.2}x pack) {:?}",
i + 1,
want,
t.elapsed().as_secs_f64(),
total as f64 / (1u64 << 30) as f64,
total as f64 / pack_bytes as f64,
store.explode_cache_use()
);
}
}
store.explode_cache_maintain()?;
let at_scan = usage(&dir)?;
let total: u64 = at_scan.values().map(|(b, _)| b).sum();
peak = peak.max(total);
report("SCAN", pack_bytes, &at_scan, t.elapsed().as_secs_f64(), rss_peak());
let band = store.explode_cache_use();
let eager = at_scan.get("objects.exploded").map(|(b, _)| *b).unwrap_or(0);
let cached = at_scan.get("objects.exploded.c*").map(|(b, _)| *b).unwrap_or(0);
let rest = total.saturating_sub(eager + cached);
println!(
"SPLIT eager (the graph) {:.3}x · cached-because-expensive {:.3}x · \
pack+index {:.3}x",
eager as f64 / pack_bytes as f64,
cached as f64 / pack_bytes as f64,
rest as f64 / pack_bytes as f64,
);
println!(
"ADMISSION {} written back, {} read directly and NOT cached ({:.1} % of {} \
rebuilt reads); index {:.1} B/object over {} objects",
band.written_back,
band.direct_reads,
100.0 * band.direct_reads as f64
/ (band.direct_reads + band.written_back).max(1) as f64,
band.direct_reads + band.written_back,
at_scan.get("objects.tail").map(|(b, _)| *b).unwrap_or(0) as f64
/ store.object_count().max(1) as f64,
store.object_count(),
);
println!(
"PLATEAU peak {:.2} GiB ({:.2}x pack) at object {}; end {:.2} GiB ({:.2}x); \
{} read, {} missing; {:?}",
peak as f64 / (1u64 << 30) as f64,
peak as f64 / pack_bytes as f64,
peak_at,
total as f64 / (1u64 << 30) as f64,
total as f64 / pack_bytes as f64,
done,
missing,
store.explode_cache_use()
);
}
Ok(())
}
/// Every file under `dir`, by name, with the bytes the filesystem is really
/// holding — `st_blocks * 512`, so a sparse file counts what it costs and not
/// what it claims.
fn usage(dir: &Path) -> Result<BTreeMap<String, (u64, u64)>> {
use std::os::unix::fs::MetadataExt;
let mut out: BTreeMap<String, (u64, u64)> = BTreeMap::new();
let mut stack = vec![dir.to_path_buf()];
while let Some(d) = stack.pop() {
for e in std::fs::read_dir(&d)?.flatten() {
let m = e.metadata()?;
if m.is_dir() {
stack.push(e.path());
} else {
let name = e.file_name().to_string_lossy().into_owned();
// Segment files roll up under one name so the table stays
// readable when there are sixteen of them.
let key = match name.rsplit_once(".c") {
Some((head, tail)) if tail.chars().all(|c| c.is_ascii_digit()) => {
format!("{head}.c*")
}
_ => name,
};
let e = out.entry(key).or_insert((0, 0));
e.0 += m.blocks() * 512;
e.1 += m.len();
}
}
}
Ok(out)
}
fn report(phase: &str, pack: u64, files: &BTreeMap<String, (u64, u64)>, secs: f64, rss: u64) {
let total: u64 = files.values().map(|(b, _)| b).sum();
let logical: u64 = files.values().map(|(_, l)| l).sum();
println!(
"{phase:<11} {:>14} B {:.3}x pack (logical {:>14} B {:.3}x) {:>7.1} s peak RSS {:.2} GiB",
total,
total as f64 / pack as f64,
logical,
logical as f64 / pack as f64,
secs,
rss as f64 / (1u64 << 30) as f64
);
for (name, (bytes, len)) in files {
if *bytes > 0 {
println!(
" {name:<28} {:>14} B {:.3}x (logical {:>14} B)",
bytes,
*bytes as f64 / pack as f64,
len
);
}
}
}
fn stride<'a>(oids: &'a [Vec<u8>], n: usize) -> Vec<&'a Vec<u8>> {
if oids.is_empty() || n == 0 {
return Vec::new();
}
let step = (oids.len() / n).max(1);
oids.iter().step_by(step).take(n).collect()
}
fn pct(v: &mut Vec<f64>, p: f64) -> f64 {
if v.is_empty() {
return 0.0;
}
v.sort_by(|a, b| a.partial_cmp(b).unwrap());
v[(((v.len() - 1) as f64) * p / 100.0).round() as usize]
}
fn kind_name(k: GitObjectKind) -> &'static str {
match k {
GitObjectKind::Commit => "commit",
GitObjectKind::Tree => "tree",
GitObjectKind::Blob => "blob",
GitObjectKind::Tag => "tag",
}
}
fn hex(b: &[u8]) -> String {
b.iter().map(|x| format!("{x:02x}")).collect()
}
/// `VmHWM` — the peak resident set this process has ever held.
fn rss_peak() -> u64 {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.starts_with("VmHWM:"))
.and_then(|l| l.split_whitespace().nth(1).and_then(|n| n.parse::<u64>().ok()))
})
.map(|kb| kb * 1024)
.unwrap_or(0)
}
fn loadavg() -> String {
std::fs::read_to_string("/proc/loadavg")
.unwrap_or_default()
.split_whitespace()
.take(3)
.collect::<Vec<_>>()
.join(" ")
}