ldeflate 0.1.3

//! 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();
        }
    }
}