ldeflate/examples/measure.rs
//! The measurement harness behind every number in `.nornir/ldeflate-design.md`.
//!
//! Run it directly for the tuning loop; the publishable rows are re-run from the
//! repo-root `examples/nornir-bench.rs` (nornir bencher, not criterion — see
//! `znippy/.nornir/benching.md`).
//!
//! ```sh
//! CARGO_TARGET_DIR=~/scratch/cargo/zoomies-deflate \
//! cargo run --offline --release -p ldeflate --example measure -- all
//! ```
//!
//! Arms: `corpus` · `dispatch` · `window` · `threshold` · `scaling` ·
//! `sustained` · `uniform` · `large` · `precompressed` · `all`.
//!
//! ## Which workload each arm models — read this before quoting a number
//!
//! Every arm here models a **BATCH write**: `repack()`, a loose-object write
//! burst, a client-side pack write, or a cold-tier re-compress. `P-004` rejects
//! per-*request* fan-out on a loaded serving tier (1000 concurrent clones ×
//! 32 workers = 32000 threads); it does not reject fan-out inside a batch
//! operation, which is what all of this is.
//!
//! No arm here models **serving a clone**. On a clone the right answer is
//! pack-copy (`P-001`): the bytes are already deflated in the pack the client
//! pushed, and shipping them is a `memcpy` — measured 25.30 ms against 28609 ms
//! for per-object zlib over 74523 objects. No compressor beats not compressing.
use std::path::PathBuf;
use std::process::Command;
use std::time::Instant;
use ldeflate::{Config, compress_objects, compress_objects_serial};
// ── Corpora ─────────────────────────────────────────────────────────────────
/// Every object in `repo`. Uses `git cat-file --batch-all-objects --batch`, so
/// the size distribution is the REAL one — commits, trees and small blobs
/// dominate, with a thin tail of large blobs. A uniform-size synthetic corpus
/// would invalidate this whole design, so nothing here invents sizes.
fn git_objects(repo: &str, cap_objects: usize) -> Vec<Vec<u8>> {
let Ok(out) = Command::new("git")
.args([
"-C",
repo,
"cat-file",
"--batch-all-objects",
"--batch",
"--buffer",
])
.output()
else {
return Vec::new();
};
if !out.status.success() {
return Vec::new();
}
let buf = out.stdout;
let mut objs = Vec::new();
let mut i = 0usize;
while i < buf.len() && objs.len() < cap_objects {
let Some(nl) = buf[i..].iter().position(|&b| b == b'\n') else {
break;
};
let Ok(h) = std::str::from_utf8(&buf[i..i + nl]) else {
break;
};
i += nl + 1;
let size = h
.split_whitespace()
.nth(2)
.and_then(|s| s.parse::<usize>().ok());
let Some(size) = size else { break };
if i + size > buf.len() {
break;
}
objs.push(buf[i..i + size].to_vec());
i += size + 1;
}
objs
}
/// The local repositories every corpus in this file is drawn from. One list, so
/// the object corpus and the packfile corpus cannot drift onto different repos
/// and make two arms incomparable.
const REPOS: &[&str] = &[
"nornir",
"dwarves",
"korp",
"znippy",
"skade",
"edda",
"holger",
"znippy-zoomies",
];
/// `$HOME/git/<name>` — where the sibling checkouts live.
fn repo_dir(name: &str) -> String {
format!("{}/git/{name}", std::env::var("HOME").unwrap_or_default())
}
fn git_corpus() -> Vec<Vec<u8>> {
let mut all = Vec::new();
for name in REPOS.iter().copied() {
let r = repo_dir(name);
let r = r.as_str();
if !PathBuf::from(r).join(".git").exists() {
continue;
}
let n0 = all.len();
all.extend(git_objects(r, 200_000));
eprintln!(" corpus: {r} -> {} objects", all.len() - n0);
}
all
}
/// LFS-shaped corpus: a handful of LARGE objects, in two flavours, because the
/// distinction decides what the split path is even for.
/// * `compressible` — text-like; deflate has real work to do.
/// * incompressible — high-entropy, standing in for the video/image/archive
/// content that is *why* a file is large enough to end up in LFS.
fn large_corpus(total_mb: usize, compressible: bool) -> Vec<Vec<u8>> {
let each = 64usize << 20;
let n = ((total_mb << 20) / each).max(1);
(0..n)
.map(|k| {
let mut v = Vec::with_capacity(each);
let mut s: u64 = 0x243F6A8885A308D3 ^ (k as u64).wrapping_mul(0x9E3779B97F4A7C15);
let words: [&[u8]; 4] = [
b"the quick brown fox jumps over the lazy dog\n",
b"pub fn run() -> usize { let x = 1; x + 2 }\n",
b"2026-08-03T10:15:00Z INFO request id=af39c0 status=200 dur_ms=42\n",
b"use std::collections::HashMap;\n",
];
while v.len() < each {
s ^= s << 13;
s ^= s >> 7;
s ^= s << 17;
if compressible {
v.extend_from_slice(words[(s >> 33) as usize % words.len()]);
} else {
v.extend_from_slice(&s.to_le_bytes());
}
}
v.truncate(each);
v
})
.collect()
}
/// A synthetic corpus of `n` objects of `size` bytes each — deliberately
/// UNIFORM. It exists to be named as the wrong corpus: numbers taken on it
/// flatter any per-item scheduler and must never be quoted as a git result.
fn uniform_corpus(n: usize, size: usize) -> Vec<Vec<u8>> {
(0..n)
.map(|i| {
let mut v = Vec::with_capacity(size);
let tag = format!("uniform-{i}-");
while v.len() < size {
v.extend_from_slice(tag.as_bytes());
v.extend_from_slice(b"pub fn run() -> usize { 1 }\n");
}
v.truncate(size);
v
})
.collect()
}
/// A small compressible buffer of exactly `sz` bytes, for the micro arms.
fn tiny(sz: usize) -> Vec<u8> {
let mut v = Vec::with_capacity(sz);
let words: [&[u8]; 2] = [b"pub fn run() -> usize { 1 }\n", b"use std::io::Write;\n"];
let mut k = 0usize;
while v.len() < sz {
v.extend_from_slice(words[k % words.len()]);
k += 1;
}
v.truncate(sz);
v
}
// ── Instrumentation ─────────────────────────────────────────────────────────
fn cores() -> usize {
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
}
/// 1-minute load average. Every measurement on this box is only as good as this
/// number: oden is a SHARED developer machine and other agents' builds and
/// services run beside the bench. A run taken at load 84 on 32 cores is not a
/// measurement, and saying so is part of the result (LAW 7).
fn loadavg() -> f64 {
std::fs::read_to_string("/proc/loadavg")
.ok()
.and_then(|s| s.split_whitespace().next().and_then(|x| x.parse().ok()))
.unwrap_or(f64::NAN)
}
/// Process CPU seconds (user+sys, all threads) from `/proc/self/stat`.
fn proc_cpu_seconds() -> f64 {
let txt = std::fs::read_to_string("/proc/self/stat").unwrap_or_default();
let Some(r) = txt.rfind(')') else { return 0.0 };
let f: Vec<&str> = txt[r + 1..].split_whitespace().collect();
let u: f64 = f.get(11).and_then(|s| s.parse().ok()).unwrap_or(0.0);
let s: f64 = f.get(12).and_then(|s| s.parse().ok()).unwrap_or(0.0);
(u + s) / 100.0
}
/// One `/proc/stat` snapshot: (busy_jiffies, total_jiffies) per logical CPU.
fn cpu_snapshot() -> Vec<(u64, u64)> {
let txt = std::fs::read_to_string("/proc/stat").unwrap_or_default();
txt.lines()
.filter(|l| l.starts_with("cpu") && l.as_bytes().get(3).is_some_and(|b| b.is_ascii_digit()))
.map(|l| {
let v: Vec<u64> = l
.split_whitespace()
.skip(1)
.filter_map(|x| x.parse().ok())
.collect();
let total: u64 = v.iter().sum();
let idle = v.get(3).copied().unwrap_or(0) + v.get(4).copied().unwrap_or(0);
(total.saturating_sub(idle), total)
})
.collect()
}
fn cpu_delta(a: &[(u64, u64)], b: &[(u64, u64)]) -> Vec<f64> {
a.iter()
.zip(b)
.map(|(x, y)| {
let dt = y.1.saturating_sub(x.1);
if dt == 0 {
0.0
} else {
y.0.saturating_sub(x.0) as f64 / dt as f64
}
})
.collect()
}
struct Timed {
wall: f64,
cpu: f64,
out_bytes: usize,
}
impl Timed {
fn new() -> Self {
Timed {
wall: f64::INFINITY,
cpu: f64::INFINITY,
out_bytes: 0,
}
}
/// One timed rep, keeping the MIN wall AND — independently — the MIN CPU
/// seconds. Noise only ever ADDS time, so a min is the least-perturbed
/// sample. The two minima are tracked separately on purpose: on a loaded box
/// wall time is dominated by whoever else is running, while the CPU OUR
/// process burns is a far more stable read on scheduling overhead. Every
/// sweep here is decided on CPU seconds for exactly that reason; wall MB/s
/// is printed beside it and is indicative only.
fn sample(&mut self, f: &mut impl FnMut() -> usize) {
let c0 = proc_cpu_seconds();
let t0 = Instant::now();
let b = f();
let w = t0.elapsed().as_secs_f64();
let c = proc_cpu_seconds() - c0;
if w < self.wall {
self.wall = w;
}
if c < self.cpu {
self.cpu = c;
}
self.out_bytes = b;
}
fn cpu_s_per_gib(&self, in_bytes: usize) -> f64 {
self.cpu / (in_bytes as f64 / (1024.0 * 1024.0 * 1024.0))
}
}
fn time_run(iters: usize, mut f: impl FnMut() -> usize) -> Timed {
let mut t = Timed::new();
for _ in 0..iters {
t.sample(&mut f);
}
t
}
fn mbs(bytes: usize, secs: f64) -> f64 {
(bytes as f64 / (1024.0 * 1024.0)) / secs
}
fn run_cfg<'a>(refs: &'a [&'a [u8]], cfg: &'a Config) -> impl FnMut() -> usize + 'a {
move || {
compress_objects(refs, cfg)
.unwrap()
.iter()
.map(|v| v.len())
.sum()
}
}
// ── Arms ────────────────────────────────────────────────────────────────────
fn arm_corpus(objs: &[Vec<u8>], label: &str) {
let mut sizes: Vec<usize> = objs.iter().map(|o| o.len()).collect();
sizes.sort_unstable();
let n = sizes.len();
if n == 0 {
println!("{label}: EMPTY");
return;
}
let total: usize = sizes.iter().sum();
let pct = |p: f64| sizes[((n as f64 - 1.0) * p) as usize];
println!(
"\n-- corpus: {label} --\n\
objects {n}\n\
total bytes {total} ({:.1} MiB)\n\
mean {} B\n\
p10 / p50 {} / {} B\n\
p90 / p99 {} / {} B\n\
p99.9 / max {} / {} B\n\
objects under 1 KiB {:.1}% | 4 KiB {:.1}% | 64 KiB {:.1}%\n\
BYTES in objects < 64 KiB: {:.1}%",
total as f64 / (1024.0 * 1024.0),
total / n,
pct(0.10),
pct(0.50),
pct(0.90),
pct(0.99),
pct(0.999),
sizes[n - 1],
100.0 * sizes.iter().filter(|&&s| s < 1024).count() as f64 / n as f64,
100.0 * sizes.iter().filter(|&&s| s < 4096).count() as f64 / n as f64,
100.0 * sizes.iter().filter(|&&s| s < 65536).count() as f64 / n as f64,
100.0
* sizes
.iter()
.filter(|&&s| s < 65536)
.map(|&s| s as f64)
.sum::<f64>()
/ total as f64,
);
}
/// **The crossover the threshold is derived FROM**, rather than a constant
/// copied out of ljar. Measures, independently: the marginal cost of one gatling
/// unit carrying no work, and the marginal cost of one whole-object deflate at
/// each size.
fn arm_dispatch() {
println!("\n-- dispatch cost vs work cost (models: repack / loose-object write) --");
let w = cores();
for &n in &[10_000usize, 100_000, 1_000_000] {
let units: Vec<usize> = (0..n).collect();
for &claim in &[1usize, 4, 16] {
let t = time_run(5, || {
let out = gatling::gatling_forkjoin::gatling_map_balanced(
&units,
w,
claim,
|_| 1u64,
|_, &i| i,
);
std::hint::black_box(&out);
n
});
println!(
" {n:>9} EMPTY units, claim {claim:>2} -> {:>7.1} ns/unit",
t.wall * 1e9 / n as f64
);
}
}
println!("\n serial cost of ONE deflate at level 6, window 15 (one core, min-of-9):");
println!(" {:>10} {:>12} {:>14}", "size", "ns/object", "MB/s");
for &sz in &[
64usize, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 65536, 262144,
] {
let src = tiny(sz);
let mut enc = ldeflate::Encoder::new(6, true).unwrap();
let mut out = Vec::new();
let t = time_run(9, || {
for _ in 0..1000 {
out.clear();
enc.compress_into(&src, &mut out).unwrap();
}
sz * 1000
});
println!(
" {:>10} {:>12.0} {:>14.1}",
sz,
t.wall * 1e9 / 1000.0,
mbs(sz * 1000, t.wall)
);
}
}
/// **Why a 64-byte object costs microseconds.**
///
/// The dispatch arm shows a hard per-object floor at level 6 that is independent
/// of payload size — a 64 B object costs the same as a 2 KiB one. That is not
/// the deflate; it is zlib's per-stream **state reset**, which clears a hash
/// table sized by the window. Git's median object is ~1 KB, so a 32 KiB window
/// is dead weight: no match can reach beyond the object's own end.
///
/// A smaller window still produces a stream stock `git` reads — RFC 1950 puts
/// the window size in the header's CINFO field and every decoder allocates
/// 32 KiB regardless. `tests/stock_git.rs` proves it against the real binary.
fn arm_window(objs: &[Vec<u8>]) {
println!("\n-- window-bits sweep: where the per-object floor comes from --");
println!(" (models: loose-object writes, repack of small objects)\n");
println!(" cost of ONE deflate at level 6, by window (one core, min-of-9), ns/object:");
print!(" {:>10}", "size");
for wb in 9..=15u8 {
print!(" {:>8}", format!("wb{wb}"));
}
println!();
for &sz in &[64usize, 256, 1024, 4096, 16384] {
let src = tiny(sz);
print!(" {:>10}", sz);
for wb in 9..=15u8 {
let mut enc = ldeflate::Encoder::new_with_window(6, true, wb).unwrap();
let mut out = Vec::new();
let t = time_run(9, || {
for _ in 0..1000 {
out.clear();
enc.compress_into(&src, &mut out).unwrap();
}
sz * 1000
});
print!(" {:>8.0}", t.wall * 1e9 / 1000.0);
}
println!();
}
if objs.is_empty() {
return;
}
let refs: Vec<&[u8]> = objs.iter().map(|o| o.as_slice()).collect();
let total: usize = objs.iter().map(|o| o.len()).sum();
println!(
"\n end-to-end on {} REAL git objects ({:.1} MiB), {} workers, min-of-7 round-robin:",
objs.len(),
total as f64 / (1 << 20) as f64,
cores()
);
println!(
" {:>6} {:>13} {:>9} {:>10} {:>7} {:>12}",
"window", "cpu-s per GiB", "vs wb15", "wall MB/s", "ratio", "size vs wb15"
);
let wbs: Vec<u8> = (9..=15).collect();
let cfgs: Vec<Config> = wbs
.iter()
.map(|&wb| Config {
window_bits: wb,
..Config::default()
})
.collect();
let mut acc: Vec<Timed> = wbs.iter().map(|_| Timed::new()).collect();
for _ in 0..7 {
for (i, cfg) in cfgs.iter().enumerate() {
acc[i].sample(&mut run_cfg(&refs, cfg));
}
}
let base_cpu = acc[wbs.len() - 1].cpu;
let base_sz = acc[wbs.len() - 1].out_bytes as f64;
for (i, &wb) in wbs.iter().enumerate() {
println!(
" {:>6} {:>13.2} {:>8.2}x {:>10.1} {:>7.3} {:>12.4}",
wb,
acc[i].cpu_s_per_gib(total),
base_cpu / acc[i].cpu,
mbs(total, acc[i].wall),
total as f64 / acc[i].out_bytes as f64,
acc[i].out_bytes as f64 / base_sz
);
}
}
/// Sweep the coalescing threshold and the gatling claim size.
///
/// Configurations are sampled ROUND-ROBIN so a load spike from another tenant of
/// this box hits every configuration equally rather than ruining one row.
fn arm_threshold(objs: &[Vec<u8>], label: &str) {
const REPS: usize = 7;
if objs.is_empty() {
return;
}
let refs: Vec<&[u8]> = objs.iter().map(|o| o.as_slice()).collect();
let total: usize = objs.iter().map(|o| o.len()).sum();
let sizes: Vec<usize> = objs.iter().map(|o| o.len()).collect();
let mut names: Vec<String> = Vec::new();
let mut cfgs: Vec<Config> = Vec::new();
for &cb in &[0usize, 256, 1024, 4096, 16 << 10, 64 << 10, 256 << 10] {
for &claim in &[1usize, 4, 16] {
names.push(format!(
"{}/{}",
if cb == 0 {
"off".into()
} else {
format!("{cb}")
},
claim
));
cfgs.push(Config {
coalesce_below: cb,
claim,
..Config::default()
});
}
}
let mut acc: Vec<Timed> = cfgs.iter().map(|_| Timed::new()).collect();
let l0 = loadavg();
for rep in 0..REPS {
for (i, cfg) in cfgs.iter().enumerate() {
acc[i].sample(&mut run_cfg(&refs, cfg));
}
eprint!(
" threshold rep {}/{REPS} (load {:.0}) \r",
rep + 1,
loadavg()
);
}
let l1 = loadavg();
println!(
"\n-- threshold sweep: {label} --\n\
{} objects, {:.1} MiB, {} workers, level 6, min-of-{REPS} ROUND-ROBIN\n\
box load during the sweep: {l0:.0} -> {l1:.0} on {} cores (SHARED BOX)\n",
objs.len(),
total as f64 / (1 << 20) as f64,
cores(),
cores()
);
println!(
"{:>16} {:>8} {:>13} {:>10} {:>10} {:>7}",
"coalesce/claim", "units", "cpu-s per GiB", "vs off/1", "wall MB/s", "ratio"
);
let base = acc[0].cpu;
for (i, name) in names.iter().enumerate() {
println!(
"{:>16} {:>8} {:>13.2} {:>9.2}x {:>10.1} {:>7.3}",
name,
ldeflate::plan(&sizes, &cfgs[i]).unwrap().len(),
acc[i].cpu_s_per_gib(total),
base / acc[i].cpu,
mbs(total, acc[i].wall),
total as f64 / acc[i].out_bytes as f64,
);
}
}
/// Scaling 1 -> 32, with efficiency. Round-robin across worker counts, min-of-N.
fn arm_scaling(objs: &[Vec<u8>], cfg: &Config, label: &str) {
const REPS: usize = 5;
if objs.is_empty() {
return;
}
let refs: Vec<&[u8]> = objs.iter().map(|o| o.as_slice()).collect();
let total: usize = objs.iter().map(|o| o.len()).sum();
let l0 = loadavg();
let widths: Vec<usize> = vec![1, 2, 4, 8, 16, 24, 32];
let cfgs: Vec<Config> = widths
.iter()
.map(|&w| Config { workers: w, ..*cfg })
.collect();
let mut acc: Vec<Timed> = cfgs.iter().map(|_| Timed::new()).collect();
let mut ser = Timed::new();
for _ in 0..REPS {
ser.sample(&mut || {
compress_objects_serial(&refs, cfg)
.unwrap()
.iter()
.map(|v| v.len())
.sum()
});
for (i, c) in cfgs.iter().enumerate() {
acc[i].sample(&mut run_cfg(&refs, c));
}
}
let l1 = loadavg();
let ser_mbs = mbs(total, ser.wall);
println!(
"\n-- scaling: {label} --\n\
{:.1} MiB in, level {}, window {}, min-of-{REPS} round-robin, load {l0:.0} -> {l1:.0}\n",
total as f64 / (1 << 20) as f64,
cfg.level,
cfg.window_bits
);
println!(
"{:>8} {:>10} {:>9} {:>6} {:>11} {:>7}",
"threads", "MB/s", "speedup", "eff", "cores busy", "ratio"
);
println!(
"{:>8} {:>10.1} {:>9} {:>6} {:>11.2} {:>7.3}",
"serial",
ser_mbs,
"1.00x",
"100%",
ser.cpu / ser.wall,
total as f64 / ser.out_bytes as f64
);
for (i, &w) in widths.iter().enumerate() {
let m = mbs(total, acc[i].wall);
println!(
"{:>8} {:>10.1} {:>8.2}x {:>5.0}% {:>11.2} {:>7.3}",
w,
m,
m / ser_mbs,
100.0 * (m / ser_mbs) / w as f64,
acc[i].cpu / acc[i].wall,
total as f64 / acc[i].out_bytes as f64
);
}
}
/// A long run with a 100 ms per-core sampler behind it. Answers what the scaling
/// table cannot: do all 32 cores stay hot, or does load spike and collapse?
fn arm_sustained(objs: &[Vec<u8>], cfg: &Config, seconds: f64, label: &str) {
if objs.is_empty() {
return;
}
let refs: Vec<&[u8]> = objs.iter().map(|o| o.as_slice()).collect();
let total: usize = objs.iter().map(|o| o.len()).sum();
let ncpu = cores();
println!(
"\n-- sustained: {label} (target {seconds:.0}s, {ncpu} cores, load {:.0}) --",
loadavg()
);
let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let stop2 = stop.clone();
// The ONE sanctioned background thread (gatling::background) — LAW 3.
let sampler = gatling::background::Job::spawn(move || {
let mut samples: Vec<Vec<f64>> = Vec::new();
let mut prev = cpu_snapshot();
while !stop2.load(std::sync::atomic::Ordering::Relaxed) {
std::thread::sleep(std::time::Duration::from_millis(100));
let now = cpu_snapshot();
samples.push(cpu_delta(&prev, &now));
prev = now;
}
samples
});
let c0 = proc_cpu_seconds();
let t0 = Instant::now();
let mut passes = 0usize;
let mut bytes = 0usize;
while t0.elapsed().as_secs_f64() < seconds {
let out = compress_objects(&refs, cfg).unwrap();
bytes += total;
std::hint::black_box(&out);
passes += 1;
}
let wall = t0.elapsed().as_secs_f64();
let our_cores = (proc_cpu_seconds() - c0) / wall;
stop.store(true, std::sync::atomic::Ordering::Relaxed);
let samples = sampler.join().unwrap();
let s: Vec<&Vec<f64>> = samples
.iter()
.skip(1)
.take(samples.len().saturating_sub(2))
.collect();
if s.is_empty() {
println!(" (no samples)");
return;
}
let busy: Vec<f64> = s
.iter()
.map(|w| w.iter().filter(|&&u| u > 0.5).count() as f64)
.collect();
let summed: Vec<f64> = s.iter().map(|w| w.iter().sum()).collect();
let mean = |v: &[f64]| v.iter().sum::<f64>() / v.len() as f64;
let mut sorted = summed.clone();
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());
let p5 = sorted[(sorted.len() as f64 * 0.05) as usize];
let p50 = sorted[sorted.len() / 2];
let ncols = s[0].len();
let per_core: Vec<f64> = (0..ncols)
.map(|c| mean(&s.iter().map(|w| w[c]).collect::<Vec<_>>()))
.collect();
let min_core = per_core.iter().cloned().fold(f64::INFINITY, f64::min);
let max_core = per_core.iter().cloned().fold(0.0, f64::max);
println!(
" {passes} passes over {wall:.1}s -> {:.1} MB/s sustained\n\
\x20 cores >50% busy: mean {:.1} / {ncpu}, worst window {:.0}, best window {:.0}\n\
\x20 summed utilisation: p50 {:.1} / {ncpu} cores, p5 {:.1} (the dip floor)\n\
\x20 per-core mean over the run: min {:.0}%, max {:.0}% (spread {:.0} pp)\n\
\x20 {} sample windows of 100 ms\n\
\x20 OUR PROCESS held {:.2} of {ncpu} cores (cpu-seconds / wall) --- the /proc/stat\n\
\x20 figures above are WHOLE-BOX and include other tenants of this shared machine.\n\
\x20 Read the per-core spread as \"no core sat out\"; read OUR cores for our share.",
mbs(bytes, wall),
mean(&busy),
busy.iter().cloned().fold(f64::INFINITY, f64::min),
busy.iter().cloned().fold(0.0, f64::max),
p50,
p5,
min_core * 100.0,
max_core * 100.0,
(max_core - min_core) * 100.0,
s.len(),
our_cores,
);
}
/// The large-object / LFS arm: segment sweep, split-path scaling, and the
/// compressible-vs-incompressible split that decides what LFS actually needs.
fn arm_large() {
println!("\n=== LARGE / LFS-shaped arm (models: cold tier, LFS ingest) ===");
for &(compressible, label) in &[
(true, "compressible (text-like)"),
(false, "INCOMPRESSIBLE (video/archive-shaped)"),
] {
let objs = large_corpus(256, compressible);
arm_corpus(&objs, label);
let total: usize = objs.iter().map(|o| o.len()).sum();
let refs: Vec<&[u8]> = objs.iter().map(|o| o.as_slice()).collect();
println!(
"\n segment sweep ({label}, {} workers, min-of-3):",
cores()
);
println!(
" {:>10} {:>10} {:>7} {:>12}",
"segment", "MB/s", "ratio", "size vs none"
);
let unsplit_cfg = Config::default().without_splitting();
let t0 = time_run(3, run_cfg(&refs, &unsplit_cfg));
println!(
" {:>10} {:>10.1} {:>7.3} {:>12}",
"none",
mbs(total, t0.wall),
total as f64 / t0.out_bytes as f64,
"1.0000"
);
let unsplit = t0.out_bytes;
for &seg in &[64usize << 10, 256 << 10, 1 << 20, 4 << 20, 16 << 20] {
let cfg = Config {
split_above: 1 << 20,
segment: seg,
..Config::default()
};
let t = time_run(3, run_cfg(&refs, &cfg));
println!(
" {:>10} {:>10.1} {:>7.3} {:>12.4}",
seg,
mbs(total, t.wall),
total as f64 / t.out_bytes as f64,
t.out_bytes as f64 / unsplit as f64
);
}
arm_scaling(
&objs,
&Config {
split_above: 1 << 20,
segment: 1 << 20,
..Config::default()
},
&format!("SPLIT path, {label}"),
);
// Where does the time go: deflate, or hash? The question that decides
// whether LFS needs this compressor at all.
let one = &objs[0];
let t_def = time_run(3, || ldeflate::zlib_compress(one, 6).unwrap().len());
let t_sha = time_run(3, || {
use sha1::{Digest, Sha1};
let mut h = Sha1::new();
h.update(one);
std::hint::black_box(h.finalize());
one.len()
});
println!(
"\n ONE 64 MiB object on ONE core: deflate L6 {:.1} MB/s | SHA-1 {:.1} MB/s\n\
\x20 -> the hash runs {:.1}x the deflate rate, so on ALREADY-COMPRESSED content\n\
\x20 (most LFS content) the deflate is pure waste and the hash is the job.",
mbs(one.len(), t_def.wall),
mbs(one.len(), t_sha.wall),
(one.len() as f64 / t_sha.wall) / (one.len() as f64 / t_def.wall)
);
}
}
/// The negative control on the CORPUS itself: show that a uniform-size synthetic
/// corpus flatters the design, so nobody quotes one as a git result.
fn arm_uniform_control() {
println!("\n-- CORPUS CONTROL: uniform 1 MiB objects (the WRONG corpus) --");
println!(" Shown so it is on the record that this shape is not git-like.\n");
let objs = uniform_corpus(256, 1 << 20);
arm_corpus(&objs, "uniform 1 MiB x 256 (SYNTHETIC, not git-shaped)");
arm_scaling(
&objs,
&Config::default(),
"uniform 1 MiB objects (SYNTHETIC)",
);
}
// ── precompressed ────────────────────────────────────────────────────────────
/// Every real packfile and pack index under the local repositories. These are
/// the genuine article — bytes git itself already deflated — so this arm is not
/// a synthetic claim about a magic byte, it is the actual cost of re-compressing
/// content that cannot compress.
fn pack_corpus(cap_bytes: usize) -> Vec<Vec<u8>> {
let mut out = Vec::new();
let mut total = 0usize;
for name in REPOS.iter().copied() {
let dir = PathBuf::from(repo_dir(name)).join(".git/objects/pack");
let Ok(rd) = std::fs::read_dir(&dir) else {
continue;
};
let mut names: Vec<_> = rd.filter_map(|e| e.ok()).map(|e| e.path()).collect();
names.sort();
for path in names {
let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
if !matches!(ext, "pack" | "idx") {
continue;
}
let Ok(bytes) = std::fs::read(&path) else {
continue;
};
if bytes.is_empty() {
continue;
}
total += bytes.len();
out.push(bytes);
if total >= cap_bytes {
return out;
}
}
}
out
}
/// The cost of NOT probing: level-6 DEFLATE over content that is already
/// compressed, against storing it.
///
/// Decided on **CPU seconds**, not wall time — see [`Timed::sample`]. Both arms
/// run the identical scheduler over the identical objects; the only difference
/// is whether the bytes reach the entropy coder.
fn arm_precompressed(objs: &[Vec<u8>], label: &str) {
if objs.is_empty() {
println!("\n-- precompressed: {label} -- (no corpus, skipped)");
return;
}
let refs: Vec<&[u8]> = objs.iter().map(|o| o.as_slice()).collect();
let total: usize = refs.iter().map(|o| o.len()).sum();
println!(
"\n-- precompressed: {label} --\n{:.1} MiB in, {} objects, level 6, load {:.0}",
total as f64 / 1048576.0,
refs.len(),
loadavg()
);
let on = Config::default();
let off = Config::default().compress_everything();
let t_off = time_run(3, || {
compress_objects(&refs, &off)
.unwrap()
.iter()
.map(|o| o.len())
.sum()
});
let t_on = time_run(3, || {
compress_objects(&refs, &on)
.unwrap()
.iter()
.map(|o| o.len())
.sum()
});
println!(
" {:>14} {:>14} {:>12} {:>8}",
"probe", "cpu-s per GiB", "wall MB/s", "ratio"
);
for (name, t) in [("off (deflate)", &t_off), ("on (store)", &t_on)] {
println!(
" {name:>14} {:>14.2} {:>12.1} {:>8.4}",
t.cpu_s_per_gib(total),
total as f64 / t.wall / 1048576.0,
total as f64 / t.out_bytes as f64
);
}
println!(
" -> {:.1}x less CPU ({:.1}% saved), size {:+.2}%",
t_off.cpu / t_on.cpu.max(f64::MIN_POSITIVE),
100.0 * (1.0 - t_on.cpu / t_off.cpu),
100.0 * (t_on.out_bytes as f64 / t_off.out_bytes as f64 - 1.0)
);
}
fn main() {
let arm = std::env::args().nth(1).unwrap_or_else(|| "all".into());
println!(
"ldeflate measure -- {} cores, {}, load {:.0}",
cores(),
std::fs::read_to_string("/proc/cpuinfo")
.ok()
.and_then(|s| s
.lines()
.find(|l| l.starts_with("model name"))
.and_then(|l| l.split(':').nth(1))
.map(|m| m.trim().to_string()))
.unwrap_or_else(|| "unknown".into()),
loadavg()
);
let need_git = matches!(
arm.as_str(),
"all" | "corpus" | "threshold" | "scaling" | "sustained" | "window"
);
let git = if need_git {
eprintln!("building the git-object corpus ...");
let c = git_corpus();
eprintln!("git corpus: {} objects", c.len());
c
} else {
Vec::new()
};
let small: Vec<Vec<u8>> = git.iter().filter(|o| o.len() < 65536).cloned().collect();
match arm.as_str() {
"corpus" => {
arm_corpus(&git, "REAL git objects");
arm_corpus(&small, "REAL git objects < 64 KiB");
}
"dispatch" => arm_dispatch(),
"window" => arm_window(&small),
"threshold" => {
arm_corpus(&git, "REAL git objects");
arm_threshold(&git, "REAL git objects, whole corpus");
arm_corpus(
&small,
"REAL git objects < 64 KiB (the dispatch-bound majority)",
);
arm_threshold(&small, "REAL git objects < 64 KiB");
}
"scaling" => arm_scaling(&git, &Config::default(), "git objects, tuned config"),
"sustained" => arm_sustained(&git, &Config::default(), 30.0, "git objects, tuned config"),
"large" => arm_large(),
"precompressed" => {
arm_precompressed(&pack_corpus(512 << 20), "REAL git packfiles + indexes")
}
"uniform" => arm_uniform_control(),
_ => {
arm_corpus(&git, "REAL git objects");
arm_dispatch();
arm_window(&small);
arm_threshold(&git, "REAL git objects, whole corpus");
arm_corpus(
&small,
"REAL git objects < 64 KiB (the dispatch-bound majority)",
);
arm_threshold(&small, "REAL git objects < 64 KiB");
arm_scaling(
&git,
&Config::default().without_coalescing(),
"git objects, NO coalescing (one unit per object)",
);
arm_scaling(&git, &Config::default(), "git objects, tuned config");
arm_sustained(&git, &Config::default(), 30.0, "git objects, tuned config");
arm_uniform_control();
arm_large();
}
}
}