ldeflate 0.1.1

//! **The git-object bake-off**: `ldeflate` against raw `zlib-rs` (through
//! `flate2`), on real git objects from the 19 `workspace_nordisk` repositories.
//!
//! The framing this harness exists to correct: **`ldeflate` IS `zlib-rs`.**
//! `ldeflate/Cargo.toml` pins `flate2 = { default-features = false,
//! features = ["zlib-rs"] }`, so the entropy coder on both sides of every
//! comparison below is the same code. What differs is the *scheduling*. A
//! naive "ldeflate vs zlib-rs" benchmark therefore compares a thing to itself
//! plus overhead, and single-stream it can only lose.
//!
//! So arm 1 is the **control**: how much does the scheduling cost when there is
//! nothing to schedule? Only against that number does arm 2's parallel speedup
//! mean anything.
//!
//! ```sh
//! CARGO_TARGET_DIR=/home/rickard/scratch/cargo/zoomies-deflate \
//!   taskset -c 0-31 cargo run --offline --release -p ldeflate \
//!   --example gitbench -- control crossover
//! ```
//!
//! Arms: `corpus` · `control` · `crossover` · `batch` · `all`.
//!
//! ## Metric choice
//!
//! Wall clock on this box is worthless on its own — load ran 8–135 all session.
//! Every arm therefore reports **CPU-seconds** (from `/proc/self/stat`,
//! utime+stime over every thread, 10 ms granularity) beside wall, and the
//! *control* arm is decided on CPU-seconds alone. A speedup is quoted only
//! together with the cores it actually held (`cpu / wall`).

use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{Duration, Instant};

use ldeflate::{Config, Encoder, compress_objects, compress_objects_serial};

/// The 19 members of `workspace_nordisk`, in descriptor order.
const REPOS: &[&str] = &[
    "nornir",
    "modgunn",
    "funnel-core",
    "holger",
    "facett",
    "edda",
    "skade",
    "znippy",
    "znippy-zoomies",
    "knut",
    "korp",
    "korp-collectors",
    "ordning",
    "Skidbladnir",
    "tunnr",
    "draupnir",
    "dwarves",
    "bifrost",
    "workspace_nordisk",
];

fn corpus_cache() -> PathBuf {
    // md1, never /tmp (that is RAM) and never the repo.
    PathBuf::from("/home/rickard/scratch/ldeflate-corpus/nordisk19.bin")
}

// ── Corpus ──────────────────────────────────────────────────────────────────

/// Read every object of `repo` with `git cat-file --batch-all-objects --batch`.
/// The size distribution is therefore the REAL one; nothing here invents sizes.
fn git_objects(repo: &Path) -> Vec<Vec<u8>> {
    let Ok(out) = Command::new("git")
        .args([
            "-C",
            &repo.display().to_string(),
            "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() {
        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 Some(size) = h
            .split_whitespace()
            .nth(2)
            .and_then(|s| s.parse::<usize>().ok())
        else {
            break;
        };
        if i + size > buf.len() {
            break;
        }
        objs.push(buf[i..i + size].to_vec());
        i += size + 1;
    }
    objs
}

/// Build (or load) the corpus. **Cached to one file**: a corpus read live from
/// working repositories GROWS between runs — the design doc's own Finding 4b
/// records a 1.2% drift that made two tables incomparable. Freezing it is what
/// lets arm 1 and arm 2 be divided by each other.
fn corpus() -> Vec<Vec<u8>> {
    let cache = corpus_cache();
    if let Ok(bytes) = std::fs::read(&cache) {
        let mut objs = Vec::new();
        let mut i = 0usize;
        while i + 4 <= bytes.len() {
            let n = u32::from_le_bytes(bytes[i..i + 4].try_into().unwrap()) as usize;
            i += 4;
            if i + n > bytes.len() {
                break;
            }
            objs.push(bytes[i..i + n].to_vec());
            i += n;
        }
        eprintln!(
            "corpus: {} objects from cache {}",
            objs.len(),
            cache.display()
        );
        return objs;
    }

    let mut all: Vec<Vec<u8>> = Vec::new();
    for r in REPOS {
        let p = PathBuf::from("/home/rickard/git").join(r);
        if !p.join(".git").exists() {
            eprintln!("  corpus: {r} MISSING — skipped");
            continue;
        }
        let n0 = all.len();
        all.extend(git_objects(&p));
        eprintln!("  corpus: {r} -> {} objects", all.len() - n0);
    }
    let _ = std::fs::create_dir_all(cache.parent().unwrap());
    let mut flat = Vec::new();
    for o in &all {
        flat.extend_from_slice(&(o.len() as u32).to_le_bytes());
        flat.extend_from_slice(o);
    }
    let _ = std::fs::write(&cache, &flat);
    eprintln!(
        "corpus: {} objects, cached to {}",
        all.len(),
        cache.display()
    );
    all
}

fn describe(objs: &[Vec<u8>]) {
    let mut sizes: Vec<usize> = objs.iter().map(|o| o.len()).collect();
    sizes.sort_unstable();
    let total: usize = sizes.iter().sum();
    let pct = |p: f64| sizes[((sizes.len() as f64 * p) as usize).min(sizes.len() - 1)];
    let under = |n: usize| sizes.iter().filter(|&&s| s < n).count() as f64 / sizes.len() as f64;
    let bytes_under =
        |n: usize| sizes.iter().filter(|&&s| s < n).sum::<usize>() as f64 / total as f64;
    println!("\n## Corpus — 19 workspace_nordisk repositories\n");
    println!("| statistic | value |");
    println!("|---|---|");
    println!("| objects | {} |", sizes.len());
    println!("| total | {:.1} MiB |", total as f64 / (1 << 20) as f64);
    println!("| mean | {} B |", total / sizes.len());
    println!("| p10 / **p50** | {} / **{} B** |", pct(0.10), pct(0.50));
    println!("| p90 / p99 | {} / {} B |", pct(0.90), pct(0.99));
    println!(
        "| p99.9 / max | {} / {} B |",
        pct(0.999),
        sizes[sizes.len() - 1]
    );
    println!("| under 1 KiB | {:.1}% |", under(1024) * 100.0);
    println!("| under 64 KiB | **{:.1}%** |", under(64 << 10) * 100.0);
    println!(
        "| **BYTES** under 64 KiB | **{:.1}%** |",
        bytes_under(64 << 10) * 100.0
    );
}

// ── Instrumentation ─────────────────────────────────────────────────────────

/// CPU-seconds this process has burned across **all** its threads, from
/// `/proc/self/stat` (utime + stime). 10 ms granularity, so every timed region
/// below runs for seconds, not milliseconds.
fn cpu_seconds() -> f64 {
    let s = std::fs::read_to_string("/proc/self/stat").unwrap_or_default();
    // `comm` may contain spaces; split after the last ')'.
    let Some(rest) = s.rsplit_once(especially_close_paren()) else {
        return 0.0;
    };
    let f: Vec<&str> = rest.1.split_whitespace().collect();
    // After ") " the fields are state(0) ppid(1) ... utime is field 14 overall,
    // i.e. index 11 here; stime index 12.
    let u: f64 = f.get(11).and_then(|x| x.parse().ok()).unwrap_or(0.0);
    let k: f64 = f.get(12).and_then(|x| x.parse().ok()).unwrap_or(0.0);
    (u + k) / clock_ticks()
}

fn especially_close_paren() -> char {
    ')'
}

fn clock_ticks() -> f64 {
    // `getconf CLK_TCK` is 100 on every Linux this ships to; read it rather
    // than assume, because a wrong divisor silently scales every CPU number.
    static CACHE: std::sync::OnceLock<f64> = std::sync::OnceLock::new();
    *CACHE.get_or_init(|| {
        Command::new("getconf")
            .arg("CLK_TCK")
            .output()
            .ok()
            .and_then(|o| {
                String::from_utf8_lossy(&o.stdout)
                    .trim()
                    .parse::<f64>()
                    .ok()
            })
            .unwrap_or(100.0)
    })
}

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)
}

struct Sample {
    wall: Duration,
    cpu: f64,
    out_bytes: usize,
}

/// Run `f` `n` times and keep the **min wall** sample — the least-perturbed
/// one, the standard choice on a shared box where noise only ever slows a run.
/// The CPU figure reported is the one from that same sample.
fn min_of<F: FnMut() -> usize>(n: usize, mut f: F) -> Sample {
    let mut best: Option<Sample> = None;
    for _ in 0..n {
        let c0 = cpu_seconds();
        let t0 = Instant::now();
        let out_bytes = f();
        let wall = t0.elapsed();
        let cpu = cpu_seconds() - c0;
        if best.as_ref().is_none_or(|b| wall < b.wall) {
            best = Some(Sample {
                wall,
                cpu,
                out_bytes,
            });
        }
    }
    best.expect("n >= 1")
}

// ── Arm 1: the control — single stream, no scheduling to do ─────────────────

/// Raw `zlib-rs` through `flate2`, one reused `Compress`, one thread. This is
/// the thing `ldeflate` is built on, driven directly. **Nothing here can beat
/// it by compressing better** — it is the same coder.
fn raw_zlib_serial(objs: &[&[u8]], level: u32) -> usize {
    use flate2::{Compress, Compression, FlushCompress, Status};
    let mut z = Compress::new(Compression::new(level), true);
    let mut total = 0usize;
    for o in objs {
        z.reset();
        // **The same reserve-and-grow policy `Encoder::drive` uses**, and a
        // FRESH buffer per object, not a reused one whose capacity has already
        // grown to the largest object seen.
        //
        // This is not pedantry. Running out of output space mid-stream makes
        // `deflate()` return and resume, and the resulting bitstream is not
        // always identical to the one a single uninterrupted call produces: the
        // diff arm found **3 of 76 616 objects** (all 665 KB-1.7 MB) where the
        // two call sequences give the same LENGTH but different bytes near the
        // end. Both are valid DEFLATE and both inflate to the object; they are
        // simply not the same encoding. A baseline that reserved differently
        // would therefore fail the byte-equality assert for a reason that has
        // nothing to do with what is being measured.
        let mut out: Vec<u8> = Vec::with_capacity((o.len() / 2).max(64) + 64);
        let mut consumed = 0usize;
        loop {
            if out.len() == out.capacity() {
                out.reserve(out.capacity().max(1024));
            }
            let spare = out.capacity() - out.len();
            let at = out.len();
            let ob = z.total_out();
            let ib = z.total_in();
            let dst = unsafe { std::slice::from_raw_parts_mut(out.as_mut_ptr().add(at), spare) };
            let flush = if consumed == o.len() {
                FlushCompress::Finish
            } else {
                FlushCompress::None
            };
            let st = z.compress(&o[consumed..], dst, flush).unwrap();
            let produced = (z.total_out() - ob) as usize;
            consumed += (z.total_in() - ib) as usize;
            unsafe { out.set_len(at + produced) };
            if matches!(st, Status::StreamEnd) {
                break;
            }
            if produced == spare {
                out.reserve(spare.max(1024));
            }
        }
        total += out.len();
    }
    total
}

/// The same loop, but through `ldeflate::Encoder` — isolating the kernel
/// wrapper's cost from the *scheduler's* cost.
fn ldeflate_kernel_serial(objs: &[&[u8]], level: u32) -> usize {
    let mut enc = Encoder::new(level, true).unwrap();
    let mut out = Vec::with_capacity(1 << 20);
    let mut total = 0usize;
    for o in objs {
        out.clear();
        enc.compress_into(o, &mut out).unwrap();
        total += out.len();
    }
    total
}

fn control(objs: &[Vec<u8>], level: u32, iters: usize) {
    let r: Vec<&[u8]> = objs.iter().map(|v| v.as_slice()).collect();
    let src: usize = objs.iter().map(|v| v.len()).sum();
    let n = objs.len();

    println!("\n## Arm 1 — the CONTROL: single stream, one core\n");
    println!(
        "Load at start: {:.1}. {} objects, {:.1} MiB, level {level}, min-of-{iters}.\n",
        loadavg(),
        n,
        src as f64 / (1 << 20) as f64
    );

    // `Config::default()` has `SkipPolicy::resolve()` ON: an object that probes
    // as already-compressed is STORED, not deflated. That is real behaviour and
    // a real win — but it means the planner is doing LESS WORK than raw
    // zlib-rs, and comparing the two would be comparing two different
    // computations, which is exactly what LAW 2 forbids. So the control arm is
    // `compress_everything()` (byte-identical output to raw zlib-rs), and the
    // skip is measured separately as its own row.
    // ...and `without_splitting()` too. `compress_everything` alone still cuts
    // every object >= `split_above` (4 MiB) at full-flush boundaries, and a
    // full flush costs ratio: measured **+293 794 B on this corpus (+0.035%)**
    // against raw zlib-rs. The byte-equality assert in arm 2 is what caught
    // that; without it these rows would have been two different computations
    // divided by each other.
    let apples = Config::default()
        .level(level)
        .compress_everything()
        .without_splitting();

    let base = min_of(iters, || raw_zlib_serial(&r, level));
    let kern = min_of(iters, || ldeflate_kernel_serial(&r, level));
    let ser = min_of(iters, || {
        compress_objects_serial(&r, &apples)
            .unwrap()
            .iter()
            .map(|v| v.len())
            .sum()
    });
    let w1 = min_of(iters, || {
        compress_objects(&r, &apples.workers(1))
            .unwrap()
            .iter()
            .map(|v| v.len())
            .sum()
    });
    let skipd = min_of(iters, || {
        compress_objects(&r, &Config::default().level(level).workers(1))
            .unwrap()
            .iter()
            .map(|v| v.len())
            .sum()
    });

    println!("| path | wall s | cpu s | **cpu vs raw zlib-rs** | ns/object | out MiB |");
    println!("|---|---|---|---|---|---|");
    for (name, s) in [
        ("raw zlib-rs (flate2, reused `Compress`)", &base),
        ("`ldeflate::Encoder` (same, wrapped)", &kern),
        ("`compress_objects_serial`, compress_everything", &ser),
        ("`compress_objects` workers=1, compress_everything", &w1),
        (
            "`compress_objects` workers=1, **default (skip ON)**",
            &skipd,
        ),
    ] {
        println!(
            "| {name} | {:.3} | {:.2} | **{:.4}×** | {:.0} | {:.1} |",
            s.wall.as_secs_f64(),
            s.cpu,
            s.cpu / base.cpu,
            s.wall.as_nanos() as f64 / n as f64,
            s.out_bytes as f64 / (1 << 20) as f64,
        );
    }
    println!(
        "\nApples-to-apples check: raw {} B vs planner {} B — {}.",
        base.out_bytes,
        w1.out_bytes,
        if base.out_bytes == w1.out_bytes {
            "**IDENTICAL**, so the rows above differ only in scheduling"
        } else {
            "**DIFFERENT — the two arms are NOT compressing the same way, the ratio above is meaningless**"
        }
    );
    println!(
        "The skip row emits {} B ({:+.2}% vs deflating everything) for {:.3}× the CPU.",
        skipd.out_bytes,
        (skipd.out_bytes as f64 / w1.out_bytes as f64 - 1.0) * 100.0,
        skipd.cpu / w1.cpu,
    );
    println!("Load at end: {:.1}.", loadavg());
}

// ── Arm 2: parallel ldeflate vs serial raw zlib-rs ──────────────────────────

fn crossover(objs: &[Vec<u8>], level: u32, iters: usize, workers: &[usize]) {
    let r: Vec<&[u8]> = objs.iter().map(|v| v.as_slice()).collect();
    let src: usize = objs.iter().map(|v| v.len()).sum();

    println!("\n## Arm 2 — parallel `ldeflate` vs SERIAL raw `zlib-rs`\n");
    println!(
        "Load at start: {:.1}. {} objects, {:.1} MiB, level {level}, min-of-{iters}. \
         Available parallelism: {}.\n",
        loadavg(),
        objs.len(),
        src as f64 / (1 << 20) as f64,
        std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(0),
    );

    // Same apples-to-apples rule as arm 1: `compress_everything`, so every row
    // below is the SAME computation as the serial baseline and the speedup is a
    // scheduling number, not a "we did less work" number.
    let apples = Config::default()
        .level(level)
        .compress_everything()
        .without_splitting();
    let base = min_of(iters, || raw_zlib_serial(&r, level));
    println!(
        "| workers | wall s | speedup vs serial zlib-rs | cpu s | **cpu vs serial** | cores held | MB/s |"
    );
    println!("|---|---|---|---|---|---|---|");
    println!(
        "| serial raw zlib-rs | {:.3} | 1.00× | {:.2} | 1.000× | 1.00 | {:.1} |",
        base.wall.as_secs_f64(),
        base.cpu,
        src as f64 / 1e6 / base.wall.as_secs_f64()
    );
    for &w in workers {
        let s = min_of(iters, || {
            compress_objects(&r, &apples.workers(w))
                .unwrap()
                .iter()
                .map(|v| v.len())
                .sum()
        });
        // Reported, not asserted. The two binaries in the A/B differ in their
        // output-buffer reserve policy, and a mid-stream buffer-full changes
        // zlib's encoding for a handful of large objects (same length, different
        // bytes) — see `raw_zlib_serial`. A hard assert would abort the older
        // binary's run for a reason unrelated to what is being measured, so the
        // delta is PRINTED instead. Anything but 0 must be read before the row.
        if s.out_bytes != base.out_bytes {
            println!(
                "| ^^ **BYTE DELTA {:+} B ({:+.4}%) at workers={w} — rows above are not \
                 the same computation** | | | | | | |",
                s.out_bytes as i64 - base.out_bytes as i64,
                (s.out_bytes as f64 / base.out_bytes as f64 - 1.0) * 100.0,
            );
        }
        println!(
            "| {w} | {:.3} | **{:.2}×** | {:.2} | {:.3}× | {:.2} | {:.1} |",
            s.wall.as_secs_f64(),
            base.wall.as_secs_f64() / s.wall.as_secs_f64(),
            s.cpu,
            s.cpu / base.cpu,
            s.cpu / s.wall.as_secs_f64(),
            src as f64 / 1e6 / s.wall.as_secs_f64(),
        );
    }
    println!("Load at end: {:.1}.", loadavg());
}

// ── Arm 3: the BATCH-SIZE crossover — how many objects must a call carry? ───
//
// gunnar's batch tier does not hand a compressor 82 000 objects; it hands it
// whatever one operation touches. Below some batch size the fan-out cannot pay
// for itself, and that threshold — not the 82k-object headline — is what
// decides whether a caller should reach for this crate.

fn batch(objs: &[Vec<u8>], level: u32, iters: usize, workers: usize) {
    println!("\n## Arm 3 — how big must a batch be before fanning out pays?\n");
    println!(
        "Load at start: {:.1}. level {level}, min-of-{iters}, workers={workers}. Batches are \
         taken from the head of the corpus, so each row is a real prefix of real objects.\n",
        loadavg()
    );
    println!(
        "| objects in call | src KiB | serial zlib-rs ms | ldeflate ms | **speedup** | verdict |"
    );
    println!("|---|---|---|---|---|---|");
    for &n in &[1usize, 4, 16, 64, 256, 1024, 4096, 16384, 65536] {
        if n > objs.len() {
            continue;
        }
        let slice = &objs[..n];
        let r: Vec<&[u8]> = slice.iter().map(|v| v.as_slice()).collect();
        let src: usize = slice.iter().map(|v| v.len()).sum();
        // Repeat the call enough times that each timed region is >= ~50 ms even
        // for n = 1; otherwise the 10 ms CPU tick and the timer floor dominate.
        let reps = (200_000usize / n.max(1)).clamp(1, 20_000);
        let b = min_of(iters, || {
            let mut t = 0;
            for _ in 0..reps {
                t += raw_zlib_serial(&r, level);
            }
            t
        });
        let apples = Config::default()
            .level(level)
            .compress_everything()
            .without_splitting()
            .workers(workers);
        let p = min_of(iters, || {
            let mut t = 0;
            for _ in 0..reps {
                t += compress_objects(&r, &apples)
                    .unwrap()
                    .iter()
                    .map(|v| v.len())
                    .sum::<usize>();
            }
            t
        });
        if b.out_bytes != p.out_bytes {
            println!(
                "| ^^ byte delta {:+} B at n={n} |||||",
                p.out_bytes as i64 - b.out_bytes as i64
            );
        }
        let sp = b.wall.as_secs_f64() / p.wall.as_secs_f64();
        println!(
            "| {n} | {:.1} | {:.2} | {:.2} | **{sp:.2}×** | {} |",
            src as f64 / 1024.0,
            b.wall.as_secs_f64() * 1000.0 / reps as f64,
            p.wall.as_secs_f64() * 1000.0 / reps as f64,
            if sp > 1.0 {
                "fan-out wins"
            } else {
                "SERIAL WINS"
            },
        );
    }
    println!("Load at end: {:.1}.", loadavg());
}

// ── Arm 4: the 31+1 STREAMING path ──────────────────────────────────────────
//
// `.nornir/ldeflate-design.md` Finding 5 names `compress_objects_to_writer` as
// what "a real `repack()` should use" and records it as **correctness-tested
// but not perf-measured**. It is also the path that built a whole
// `Compress` per OBJECT, so it is where the encoder pool has the most to give.

fn stream_arm(objs: &[Vec<u8>], level: u32, iters: usize) {
    use std::io::Write;
    /// A sink that counts and discards. The point of this arm is the compressor,
    /// not the device; writing 800 MiB to disk would measure the NVMe.
    struct Null(u64);
    impl Write for Null {
        fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
            self.0 += b.len() as u64;
            Ok(b.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    let r: Vec<&[u8]> = objs.iter().map(|v| v.as_slice()).collect();
    let src: usize = objs.iter().map(|v| v.len()).sum();
    println!("\n## Arm 4 — the 31+1 streaming path (`compress_objects_to_writer`)\n");
    println!(
        "Load at start: {:.1}. {} objects, {:.1} MiB, level {level}, min-of-{iters}.\n",
        loadavg(),
        objs.len(),
        src as f64 / (1 << 20) as f64
    );
    let apples = Config::default()
        .level(level)
        .compress_everything()
        .without_splitting();
    println!("| workers | wall s | cpu s | ns/object | MB/s | cores held |");
    println!("|---|---|---|---|---|---|");
    for w in [1usize, 8, 31] {
        let s = min_of(iters, || {
            let mut sink = Null(0);
            ldeflate::compress_objects_to_writer(&r, &apples.workers(w), &mut sink).unwrap();
            sink.0 as usize
        });
        println!(
            "| {w} | {:.3} | {:.2} | {:.0} | {:.1} | {:.2} |",
            s.wall.as_secs_f64(),
            s.cpu,
            s.wall.as_nanos() as f64 / objs.len() as f64,
            src as f64 / 1e6 / s.wall.as_secs_f64(),
            s.cpu / s.wall.as_secs_f64(),
        );
    }
    println!("Load at end: {:.1}.", loadavg());
}

// ── Arm 5: REAL LFS objects ─────────────────────────────────────────────────
//
// The within-stream split path (`Unit::Segment`) is the axis this crate offers
// for LFS-shaped content, and Finding 6 measured 10.3x wall throughput on a
// SYNTHETIC compressible 64 MiB corpus. The open question is whether that
// transfers, because real LFS payloads are usually already-compressed media —
// in which case `SkipPolicy::resolve()` stores them and the split fires on
// chunking, not on compression.
//
// This arm answers it on the **98 real LFS objects on this box**, not on
// synthetic buffers.

fn lfs_corpus() -> Vec<(String, Vec<u8>)> {
    let mut out = Vec::new();
    for r in ["nornir", "korp", "edda", "facett", "holger", "skade"] {
        let dir = PathBuf::from("/home/rickard/git")
            .join(r)
            .join(".git/lfs/objects");
        let mut stack = vec![dir];
        while let Some(d) = stack.pop() {
            let Ok(rd) = std::fs::read_dir(&d) else {
                continue;
            };
            for e in rd.flatten() {
                let p = e.path();
                if p.is_dir() {
                    stack.push(p);
                } else if let Ok(b) = std::fs::read(&p) {
                    if !b.is_empty() {
                        out.push((
                            format!("{r}/{}", p.file_name().unwrap().to_string_lossy()),
                            b,
                        ));
                    }
                }
            }
        }
    }
    out.sort_by(|a, b| a.0.cmp(&b.0));
    out
}

fn lfs_arm(level: u32, iters: usize) {
    use ldeflate::{SkipPolicy, looks_compressed};
    let corpus = lfs_corpus();
    assert!(!corpus.is_empty(), "no LFS objects found");
    let bufs: Vec<&[u8]> = corpus.iter().map(|(_, b)| b.as_slice()).collect();
    let total: usize = bufs.iter().map(|b| b.len()).sum();

    println!("\n## Arm 5 — the REAL LFS corpus, and whether the split win transfers\n");
    println!(
        "Load at start: {:.1}. **{} real LFS objects, {:.1} MiB** from six repositories \
         on this box, level {level}, min-of-{iters}.\n",
        loadavg(),
        corpus.len(),
        total as f64 / (1 << 20) as f64
    );

    // ── (a) What does the probe SAY about real LFS content? ─────────────────
    let probed: usize = bufs.iter().filter(|b| looks_compressed(b)).count();
    let probed_bytes: usize = bufs
        .iter()
        .filter(|b| looks_compressed(b))
        .map(|b| b.len())
        .sum();
    let big = bufs.iter().filter(|b| b.len() >= 4 << 20).count();
    println!("| question | answer |");
    println!("|---|---|");
    println!(
        "| objects the magic probe calls already-compressed | **{probed} of {}** ({:.1}%) |",
        corpus.len(),
        probed as f64 / corpus.len() as f64 * 100.0
    );
    println!(
        "| BYTES the probe would store rather than deflate | **{:.1} of {:.1} MiB** ({:.1}%) |",
        probed_bytes as f64 / (1 << 20) as f64,
        total as f64 / (1 << 20) as f64,
        probed_bytes as f64 / total as f64 * 100.0
    );
    println!("| objects at or above `split_above` (4 MiB), i.e. eligible to split | **{big}** |");

    // ── (b) Does the split path actually pay on THIS content? ───────────────
    let mut rows: Vec<(&str, Config)> = Vec::new();
    rows.push((
        "default (skip ON, split ON), 1 worker",
        Config::default().level(level).workers(1),
    ));
    rows.push((
        "default (skip ON, split ON), 32 workers",
        Config::default().level(level).workers(32),
    ));
    rows.push((
        "compress_everything, split OFF, 1 worker",
        Config::default()
            .level(level)
            .compress_everything()
            .without_splitting()
            .workers(1),
    ));
    rows.push((
        "compress_everything, split ON, 32 workers",
        Config::default()
            .level(level)
            .compress_everything()
            .workers(32),
    ));
    rows.push((
        "compress_everything, split OFF, 32 workers",
        Config::default()
            .level(level)
            .compress_everything()
            .without_splitting()
            .workers(32),
    ));

    println!("\n| configuration | wall s | cpu s | MB/s | **ratio** | cores held |");
    println!("|---|---|---|---|---|---|");
    for (name, cfg) in rows {
        let s = min_of(iters, || {
            compress_objects(&bufs, &cfg)
                .unwrap()
                .iter()
                .map(|v| v.len())
                .sum()
        });
        println!(
            "| {name} | {:.3} | {:.2} | {:.1} | **{:.3}** | {:.2} |",
            s.wall.as_secs_f64(),
            s.cpu,
            total as f64 / 1e6 / s.wall.as_secs_f64(),
            total as f64 / s.out_bytes as f64,
            s.cpu / s.wall.as_secs_f64(),
        );
    }
    let _ = SkipPolicy::resolve();
    println!("Load at end: {:.1}.", loadavg());
}

// ── The diff arm ────────────────────────────────────────────────────────────
//
// Arm 2's byte-equality assert fired on a 293 794 B difference over the whole
// corpus, and reading the two code paths did not find it. This arm exists
// because a diff you can point at beats a theory about one.

fn diffarm(objs: &[Vec<u8>], level: u32) {
    use flate2::{Compress, Compression, FlushCompress, Status};
    let r: Vec<&[u8]> = objs.iter().map(|v| v.as_slice()).collect();
    let apples = Config::default()
        .level(level)
        .compress_everything()
        .without_splitting();
    let mine = compress_objects_serial(&r, &apples).unwrap();

    let mut z = Compress::new(Compression::new(level), true);
    let (mut shown, mut ndiff, mut bytes_diff) = (0usize, 0usize, 0i64);
    for (i, o) in r.iter().enumerate() {
        z.reset();
        let mut out: Vec<u8> = Vec::with_capacity((o.len() / 2).max(64) + 64);
        let mut consumed = 0usize;
        loop {
            if out.len() == out.capacity() {
                out.reserve(out.capacity().max(1024));
            }
            let spare = out.capacity() - out.len();
            let at = out.len();
            let (ob, ib) = (z.total_out(), z.total_in());
            let dst = unsafe { std::slice::from_raw_parts_mut(out.as_mut_ptr().add(at), spare) };
            let flush = if consumed == o.len() {
                FlushCompress::Finish
            } else {
                FlushCompress::None
            };
            let st = z.compress(&o[consumed..], dst, flush).unwrap();
            let produced = (z.total_out() - ob) as usize;
            consumed += (z.total_in() - ib) as usize;
            unsafe { out.set_len(at + produced) };
            if matches!(st, Status::StreamEnd) {
                break;
            }
        }
        if out != mine[i] {
            ndiff += 1;
            bytes_diff += mine[i].len() as i64 - out.len() as i64;
            if shown < 10 {
                shown += 1;
                println!(
                    "object {i}: src {} B, raw {} B, planner {} B, first differing byte at {:?}",
                    o.len(),
                    out.len(),
                    mine[i].len(),
                    out.iter().zip(&mine[i]).position(|(a, b)| a != b),
                );
            }
        }
    }
    println!(
        "\n{ndiff} of {} objects differ; planner is {bytes_diff:+} B overall",
        r.len()
    );
}

// ── Correctness, run before any timing ──────────────────────────────────────

/// Every arm above is meaningless if the two paths do not produce the same
/// bytes. Assert it on a real slice of the real corpus, and assert the *check*
/// can fail.
fn verify(objs: &[Vec<u8>], level: u32) {
    let take = objs.len().min(5000);
    let slice = &objs[..take];
    let r: Vec<&[u8]> = slice.iter().map(|v| v.as_slice()).collect();

    let par = compress_objects(&r, &Config::default().level(level)).unwrap();
    let ser = compress_objects_serial(&r, &Config::default().level(level)).unwrap();
    assert_eq!(par, ser, "parallel and serial disagree on the real corpus");

    for (i, (z, src)) in par.iter().zip(slice).enumerate() {
        let mut got = Vec::new();
        flate2::read::ZlibDecoder::new(&z[..])
            .read_to_end(&mut got)
            .unwrap_or_else(|e| panic!("object {i} ({} B) failed to inflate: {e}", src.len()));
        assert_eq!(&got, src, "object {i} round-trip differs");
    }

    // RED control: the round-trip must be able to FAIL. Splice two adjacent
    // streams together — the exact corruption a broken coalescing queue would
    // produce — and prove the decoder-plus-compare catches it.
    let mut caught = 0usize;
    for i in 0..par.len().saturating_sub(1) {
        let mut spliced = par[i].clone();
        spliced.truncate(spliced.len().saturating_sub(4).max(1));
        spliced.extend_from_slice(&par[i + 1]);
        let mut got = Vec::new();
        if flate2::read::ZlibDecoder::new(&spliced[..])
            .read_to_end(&mut got)
            .is_err()
            || got != slice[i]
        {
            caught += 1;
        }
        if i > 200 {
            break;
        }
    }
    assert!(
        caught > 0,
        "overlapping two objects' streams was NOT detected — the round-trip check \
         above is blind and proves nothing"
    );
    eprintln!("verify: {take} objects round-trip; splice control caught {caught}/201");
}

fn main() {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let arms: Vec<&str> = if args.is_empty() {
        vec!["all"]
    } else {
        args.iter().map(|s| s.as_str()).collect()
    };
    let all = arms.contains(&"all");
    let level: u32 = std::env::var("LDEFLATE_LEVEL")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(6);
    let iters: usize = std::env::var("LDEFLATE_ITERS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(5);

    let mut objs = corpus();
    // `LDEFLATE_MAX_OBJ` keeps only objects strictly smaller than N bytes.
    //
    // This exists because the whole-repository corpus is BYTES-dominated while
    // the per-object fixed costs this crate attacks are COUNT-dominated: 94.2%
    // of the objects hold 17.1% of the bytes, so a 4 us per-object saving is
    // 0.9% of a pass whose mean object costs 430 us. Finding 1 said exactly
    // this and the first A/B of the encoder pool ignored it. At
    // `LDEFLATE_MAX_OBJ=65536` the corpus is the 94.2% majority the coalescing
    // queue is built for, and a per-object cost is visible or it is not real.
    if let Some(cap) = std::env::var("LDEFLATE_MAX_OBJ")
        .ok()
        .and_then(|s| s.parse::<usize>().ok())
    {
        let before = objs.len();
        objs.retain(|o| o.len() < cap);
        eprintln!(
            "corpus: kept {} of {before} objects under {cap} B",
            objs.len()
        );
    }
    assert!(
        !objs.is_empty(),
        "empty corpus — no repositories were readable"
    );

    if all || arms.contains(&"corpus") {
        describe(&objs);
    }
    verify(&objs, level);
    if all || arms.contains(&"control") {
        control(&objs, level, iters);
    }
    if all || arms.contains(&"crossover") {
        let avail = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(4);
        let workers: Vec<usize> = [1usize, 2, 4, 8, 16, 32]
            .into_iter()
            .filter(|&w| w <= avail.max(1))
            .collect();
        crossover(&objs, level, iters, &workers);
    }
    if arms.contains(&"lfs") {
        lfs_arm(level, iters);
        return;
    }
    if arms.contains(&"diff") {
        diffarm(&objs, level);
    }
    if all || arms.contains(&"stream") {
        stream_arm(&objs, level, iters);
    }
    if all || arms.contains(&"batch") {
        let avail = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(4);
        batch(&objs, level, iters.min(3), avail);
    }
}