ldeflate/examples/gitbench.rs
//! **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=~/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",
];
/// `$HOME` — the corpus repos are the sibling checkouts under `$HOME/git`.
fn home() -> PathBuf {
PathBuf::from(std::env::var_os("HOME").unwrap_or_default())
}
fn corpus_cache() -> PathBuf {
// md1, never /tmp (that is RAM) and never the repo.
home().join("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 = home().join("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 = home()
.join("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);
}
}