osm-katana/benches/tile_pyramid_parallel.rs
//! **Tiling a region: how many cores, and how much RAM.**
//!
//! [`osm_katana::tiles::tile_region`] used to scan `ways.parquet` and
//! `nodes.parquet` in a `for batch in reader` loop **on the calling thread** and
//! build the whole pyramid in HashMaps before writing a single tile. Measured on
//! this box over a full Sweden tiling, 2026-08-03:
//!
//! * **99.5–99.7 % CPU — exactly one core of 32 — from t=4 s to t=62 s**, then
//! ~6.7 cores for the closing write. 115.5 core-seconds of work spread over
//! 70 s of wall.
//! * **12.21 GB peak**, the curve monotone 0.5 → 11.4 GB with *nothing on disk
//! yet*. z15 alone wanted 36.36 GB.
//!
//! So this bench reports **core occupancy** (`cpu_cores`, from the same
//! `/proc/self/stat` counter `phase_log` publishes) and **memory** per arm, and
//! refuses to treat either alone as a result. A tiling that got faster by holding
//! 390 GB is a regression, and the gate below says so.
//!
//! ## RSS is a property of the PROCESS, and an arm is not a process
//!
//! The first version of this bench sampled RSS per arm and called the sample the
//! arm's cost. It is not. glibc opens an arena per contending thread, grows each to
//! its own high-water mark, and hands nothing back to the kernel on `free` — so a
//! fork-join arm leaves its peak behind as free list and the NEXT arm starts inside
//! it. Measured on oden over `sweden` at z6/9/12 on 2026-08-03: run third, the
//! bounded arm sampled **15.20 GiB**; run as the only arm in the process, the same
//! arm on the same corpus producing the same 28 974 tiles costs **5.92 GiB**. The
//! 9-point-something GiB difference was two earlier arms' allocator arenas, charged
//! to whichever arm ran last, and it held a 6x speed-up off `main` for a day.
//!
//! Each arm therefore `malloc_trim`s, records the RSS it STARTS from, and reports
//! `armGiB = peakRSS - baseline` beside the raw peak. The gate reads `armGiB` — and
//! reads the baseline first, because a large baseline makes the subtraction
//! meaningless and that is a verdict of its own (`INHERITED`).
//!
//! ## What the engine can name
//!
//! `tile_region` reports, at its scan high-water mark, RSS against the allocator's
//! own `mallinfo2` books against its own accounting — see
//! `TileStats::memory_attribution`. That is what turns "11.2 GiB is missing" into
//! three named terms: live heap, allocator free list, and the engine's unspillable
//! bookkeeping floor (`peak_bookkeeping_bytes`), which is the one the payload budget
//! structurally cannot bound.
//!
//! ## The arms
//!
//! | arm | workers | budget | what it is |
//! |---|---|---|---|
//! | `serial-unbounded` | 1 | none | the old engine's shape: one scan thread, whole pyramid resident |
//! | `parallel-unbounded` | all | none | fan-out alone — how much of the wall was the serial scan |
//! | `parallel-bounded` | all | `--budget` | fan-out **and** the spill: what ships |
//! | `dup-accumulator` | all | `--budget`, ×k concurrently | the trap: k private accumulators. **Must trip the memory ceiling.** |
//!
//! The `dup-accumulator` arm exists for LAW 2. A memory gate nobody has seen fire
//! is not a gate, so the bench runs `k` tilers at once — precisely the residency
//! shape of "give every worker its own `zoom_tiles` and merge at the end" — and
//! **asserts the ceiling rejects it**. If that arm ever passes the ceiling, the
//! ceiling is decoration and the bench fails on that, not on the tiling.
//!
//! ```text
//! OSM_KATANA_TILE_EXTRACT=~/.korp/work/sweden \
//! OSM_KATANA_TILE_ZOOMS=6,9,12 \
//! cargo bench -p osm-katana --bench tile_pyramid_parallel
//! ```
//! Env: `OSM_KATANA_TILE_OUT` (scratch, must be on md1 — never `/tmp`, that is
//! RAM and this writes gigabytes), `OSM_KATANA_TILE_BUDGET_GIB`,
//! `OSM_KATANA_TILE_CEILING_GIB`, `OSM_KATANA_TILE_DUP` (k, `0` skips the trap
//! arm), `OSM_KATANA_TILE_REGION` (`lon0,lat0,lon1,lat1`).
use std::path::PathBuf;
use std::time::Instant;
use osm_katana::Bounds;
use osm_katana::digest::PyramidFingerprint;
use osm_katana::phase_log::RssWatch;
use osm_katana::tiles::{TileOptions, TileSource, TileStats, tile_region};
/// One measured arm.
struct Row {
name: String,
workers: usize,
budget_gib: f64,
wall_s: f64,
scan_s: f64,
scan_cores: f64,
write_s: f64,
write_cores: f64,
peak_rss: u64,
/// RSS the process ALREADY held when this arm started, after a `malloc_trim`.
/// Subtract it and you have what the arm itself cost; ignore it and you charge
/// one arm for every arm before it.
rss_baseline: u64,
payload: u64,
spilled: u64,
tiles: usize,
/// The engine's own report for this arm, so the table and the attribution can
/// never describe two different runs.
st: TileStats,
}
impl Row {
/// **What this arm cost, as opposed to what the process was holding while it
/// ran.** The distinction is the whole of stinky `katana-15`.
fn arm_rss(&self) -> u64 {
self.peak_rss.saturating_sub(self.rss_baseline)
}
}
fn gib(b: u64) -> f64 {
b as f64 / (1u64 << 30) as f64
}
fn env_f64(k: &str, d: f64) -> f64 {
std::env::var(k)
.ok()
.and_then(|v| v.trim().parse().ok())
.unwrap_or(d)
}
/// 1-minute load average, from `/proc/loadavg`.
fn load_avg_1() -> f64 {
std::fs::read_to_string("/proc/loadavg")
.ok()
.and_then(|s| s.split_whitespace().next().and_then(|v| v.parse().ok()))
.unwrap_or(0.0)
}
/// **A busy box measures somebody else's job.** This bench's whole claim is core
/// occupancy: "the scan held 2.5 of 32 cores" means the fan-out failed *or* means
/// 29 of those cores belonged to another process, and the two are indistinguishable
/// from inside. Measured here on 2026-08-03 at load 59–192 — a concurrent
/// `osm-katana` at 1206 % and three `rustc` — and every arm read low by a factor
/// that could not be attributed.
///
/// So: refuse above `LOAD_CEILING × cores`, and say what the load was.
/// `OSM_KATANA_TILE_ALLOW_LOAD=1` overrides for a deliberate contended run — and
/// then the printout says so, in the table, so no figure escapes without it.
const LOAD_CEILING: f64 = 0.25;
/// **A `harness = false` bench must have an entry point in BOTH profiles.**
///
/// The real `main` below is `#[cfg(not(debug_assertions))]`, for the same reason
/// `clip_containment` and `pbf_pass2_plan` gate theirs: every figure this bench
/// prints under `opt-level=0` would be an artefact of the profile, and the arms
/// would not even slow down by the same factor. But a bench target with NO `main`
/// in the dev profile does not merely refuse to run — it fails to *compile*, and
/// `cargo check --all-targets` (the flag every "is the tree clean" gate reaches
/// for) can then never be green:
///
/// ```text
/// error[E0601]: `main` function not found in crate `tile_pyramid_parallel`
/// ```
///
/// One target that always fails trains everyone to read past the error, which is
/// how a real red goes unnoticed. So the dev profile gets a `main` that refuses
/// out loud, exactly as the two sibling benches already do.
/// `tests/bench_target_law.rs` holds every `harness = false` bench in the
/// workspace to this shape.
#[cfg(debug_assertions)]
fn main() {
eprintln!(
"tile_pyramid_parallel: REFUSING to run under debug_assertions — this bench \
reports core occupancy and peak RSS, and both figures would be artefacts of \
opt-level=0 (an unoptimised scan holds a core it would not otherwise need, \
and the arms do not slow down evenly). Run it as \
`cargo bench -p osm-katana --bench tile_pyramid_parallel`."
);
std::process::exit(2);
}
#[cfg(not(debug_assertions))]
fn main() {
let extract = PathBuf::from(
std::env::var("OSM_KATANA_TILE_EXTRACT")
.unwrap_or_else(|_| format!("{}/.korp/work/sweden", std::env::var("HOME").unwrap_or_default())),
);
// md1, never /tmp — /tmp is RAM on this box and an arm writes gigabytes of
// tiles plus its spill scratch.
let out_root = PathBuf::from(
std::env::var("OSM_KATANA_TILE_OUT")
.unwrap_or_else(|_| format!("{}/scratch/osm-katana-tilebench", std::env::var("HOME").unwrap_or_default())),
);
if !extract.join("ways.parquet").exists() && !extract.join("nodes.parquet").exists() {
eprintln!(
"tile_pyramid_parallel: no ways.parquet / nodes.parquet under {} — nothing to tile",
extract.display()
);
std::process::exit(1);
}
let zooms: Vec<u8> = std::env::var("OSM_KATANA_TILE_ZOOMS")
.unwrap_or_else(|_| String::from("6,9,12"))
.split(',')
.filter_map(|s| s.trim().parse().ok())
.collect();
assert!(!zooms.is_empty(), "OSM_KATANA_TILE_ZOOMS parsed to nothing");
let region: Option<Bounds> = std::env::var("OSM_KATANA_TILE_REGION").ok().and_then(|s| {
let v: Vec<f64> = s.split(',').filter_map(|t| t.trim().parse().ok()).collect();
(v.len() == 4).then(|| Bounds::new(v[0], v[1], v[2], v[3]))
});
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let budget_gib = env_f64("OSM_KATANA_TILE_BUDGET_GIB", 4.0);
let budget = (budget_gib * (1u64 << 30) as f64) as u64;
// The ceiling the bounded arm must live under. It is the resident-payload
// budget PLUS what the rest of the process legitimately needs at that worker
// count: the parquet decoder's per-row-group column buffers (one live row
// group per worker), arrow's staging, and the scan workers' placement
// staging. Stated as a headroom multiple + a constant so both terms are
// visible rather than a single magic gigabyte count.
let ceiling_gib = env_f64("OSM_KATANA_TILE_CEILING_GIB", budget_gib * 1.5 + 4.0);
let ceiling = (ceiling_gib * (1u64 << 30) as f64) as u64;
// Which arms to run. Full table by default; a subset when a measurement only
// needs one arm (the bounded arm alone is ~1/8 of the full run's wall).
let arms_filter: Option<Vec<String>> = std::env::var("OSM_KATANA_TILE_ARMS").ok().map(|s| {
s.split(',')
.map(|a| a.trim().to_string())
.filter(|a| !a.is_empty())
.collect()
});
let dup_k: usize = std::env::var("OSM_KATANA_TILE_DUP")
.ok()
.and_then(|v| v.trim().parse().ok())
.unwrap_or(3);
eprintln!(
"tile_pyramid_parallel: extract {} · zooms {zooms:?} · region {} · {cores} cores\n\
\tscratch {} · budget {budget_gib:.2} GiB · ceiling {ceiling_gib:.2} GiB · dup k={dup_k}",
extract.display(),
region
.map(|b| format!(
"{:.2},{:.2}..{:.2},{:.2}",
b.min_lon, b.min_lat, b.max_lon, b.max_lat
))
.unwrap_or_else(|| String::from("unbounded")),
out_root.display(),
);
let load0 = load_avg_1();
let contended = load0 > LOAD_CEILING * cores as f64;
let allow_load = std::env::var("OSM_KATANA_TILE_ALLOW_LOAD").is_ok_and(|v| v != "0");
if contended && !allow_load {
eprintln!(
"tile_pyramid_parallel: REFUSING to measure — 1-min load average is {load0:.1} on \
{cores} cores, over the {:.1} ceiling. Every occupancy figure this bench prints \
would be another process's contention wearing this one's name, and it would still \
look like a result. Wait for the box, or set OSM_KATANA_TILE_ALLOW_LOAD=1 to run \
anyway (the table then says the numbers are contended).",
LOAD_CEILING * cores as f64,
);
std::process::exit(2);
}
let sink = osm_katana::phase_log::PhaseSink::new(None).expect("phase sink");
let mut rows: Vec<Row> = Vec::new();
// The exclusion axis, so "what does the pyramid weigh with and without the road and
// rail networks in it" is one env flip on the same corpus rather than two builds.
let exclude_networks = std::env::var("OSM_KATANA_TILE_EXCLUDE_NETWORKS")
.map(|v| v != "0")
.unwrap_or(true);
eprintln!("tile_pyramid_parallel: exclude_networks={exclude_networks}");
let base = TileOptions {
zooms: zooms.clone(),
region,
exclude_networks,
..Default::default()
};
// **What an arm hands to the equivalence gate.** The tile COUNT is kept for the
// table; the FINGERPRINT is what the gate compares — see
// `osm_katana::digest::PyramidFingerprint` for why a count cannot do the job.
struct Arm {
tiles: usize,
fp: PyramidFingerprint,
}
// **Every arm name this run OFFERED**, whether or not the filter let it through.
// The valid-name list is therefore derived from the arms themselves and cannot rot
// into a second, stale copy of them beside the dispatch (LAW 5).
let offered: std::cell::RefCell<Vec<String>> = std::cell::RefCell::new(Vec::new());
let run = |name: &str, workers: usize, mem: u64, rows: &mut Vec<Row>| -> Option<Arm> {
offered.borrow_mut().push(name.to_string());
if let Some(f) = &arms_filter {
if !f.iter().any(|a| a == name) {
return None;
}
}
let out = out_root.join(name);
let _ = std::fs::remove_dir_all(&out);
std::fs::create_dir_all(&out).expect("create arm scratch");
let opts = TileOptions {
workers,
mem_budget_bytes: mem,
..base.clone()
};
let phase = sink.phase(
"tile_region",
&[("arm", name), ("workers", &workers.to_string())],
);
// **An arm inherits the arms before it.** RSS is a property of the PROCESS:
// glibc opens an arena per contending thread, grows each to its own high-water
// mark and returns nothing on `free`, so arm 3 starts inside whatever arms 1
// and 2 left behind and a sampler charges the whole of it to arm 3. That is
// how this bench read 15.20 GiB for a bounded arm that costs 5.92 GiB when it
// is the only arm in the process (oden, `sweden` z6/9/12, 2026-08-03).
// So: hand back what can be handed back, then record where this arm STARTS.
let returned = osm_katana::phase_log::trim_malloc();
let rss_baseline = osm_katana::phase_log::rss_bytes();
eprintln!(
"tile_pyramid_parallel: {name} starts at {:.2} GiB RSS ({:.2} GiB returned to the \
kernel first) — every RSS figure for this arm is measured FROM here",
gib(rss_baseline),
gib(returned),
);
let watch = RssWatch::start();
let t0 = Instant::now();
let st = tile_region(TileSource::Extract(&extract), &out, &opts).expect("tile_region");
let wall_s = t0.elapsed().as_secs_f64();
let peak_rss = watch.finish();
phase.done(&[("peak_rss_gib", format!("{:.3}", gib(peak_rss)))]);
let tiles: usize = st.zooms.iter().map(|z| z.tiles).sum();
// An arm that produced nothing measured nothing.
assert!(tiles > 0, "{name}: produced no tiles — nothing was benched");
// **The fingerprint comes off the pyramid BEFORE it is deleted**, and it is the
// only thing that survives an arm. Not gated on `OSM_KATANA_TILE_KEEP`: an
// equivalence gate that only runs when an operator happened to ask for the files
// to be left behind is a gate that never runs.
let fp = PyramidFingerprint::of(&st.tiles_dir)
.unwrap_or_else(|e| panic!("{name}: fingerprint the pyramid: {e:#}"));
if !std::env::var("OSM_KATANA_TILE_KEEP").is_ok_and(|v| v != "0") {
let _ = std::fs::remove_dir_all(&out);
} else {
eprintln!(
"tile_pyramid_parallel: {name} kept at {} ({} tiles)",
out.display(),
tiles
);
}
rows.push(Row {
name: name.into(),
workers,
budget_gib: gib(mem),
wall_s,
scan_s: st.scan_wall_s,
scan_cores: st.scan_cpu_cores,
write_s: st.write_wall_s,
write_cores: st.write_cpu_cores,
peak_rss,
rss_baseline,
payload: st.peak_resident_bytes,
spilled: st.spilled_bytes,
tiles,
st,
});
Some(Arm {
tiles: rows.last().expect("just pushed").tiles,
fp,
})
};
let n_serial = run("serial-unbounded", 1, 0, &mut rows);
let n_punb = run("parallel-unbounded", cores, 0, &mut rows);
let n_bnd = run("parallel-bounded", cores, budget, &mut rows);
// Optional scaling sweep: where does the fan-out stop paying, and is the knee
// the shard locks or the disk? A ratio between two points cannot tell you.
let mut sweeps: Vec<(String, Option<Arm>)> = Vec::new();
if let Ok(spec) = std::env::var("OSM_KATANA_TILE_SWEEP") {
for w in spec
.split(',')
.filter_map(|s| s.trim().parse::<usize>().ok())
{
let name = format!("sweep-w{w}");
let n = run(&name, w, budget, &mut rows);
sweeps.push((name, n));
}
}
// **Same input, same options bar the fan-out and the budget ⇒ the SAME PYRAMID.**
//
// Collected here and reported in the gate below beside every other verdict, rather
// than asserted inline: a bench that panics halfway prints no table, and the table
// is what tells you WHICH arm to look at.
//
// The comparison is `PyramidFingerprint`, not a tile count. The count was blind to a
// tile at a different (x, y) — the fan-out is exactly what decides placement — to
// rows lost INSIDE a tile, and to feature substitution. It was the arm meant to
// notice a 32-thread scan silently dropping geometry, reporting on a number that
// cannot see it.
let mut equivalence: Vec<String> = Vec::new();
let mut compare =
|what: &str, a: &Option<Arm>, b: &Option<Arm>| {
match (a, b) {
(Some(a), Some(b)) => {
let diffs = a.fp.differences(&b.fp);
if !diffs.is_empty() {
equivalence.push(format!(
"EQUIVALENCE: {what} — {} tiles/{} rows vs {} tiles/{} rows:\n · {}",
a.fp.tiles.len(), a.fp.rows(), b.fp.tiles.len(), b.fp.rows(),
diffs.join("\n · "),
));
}
}
// **The absence is RED, not silent.** `if let (Some(a), Some(b))` turned the
// whole equivalence check off in exactly the "I am iterating on one arm"
// mode where a regression is most likely, and said nothing at all.
_ => equivalence.push(format!(
"NOT CHECKED: {what} — OSM_KATANA_TILE_ARMS filtered out an arm this check \
needs, so nothing was compared. Unfiltered runs prove equivalence; this \
one does not.",
)),
}
};
compare("fanning the scan out", &n_serial, &n_punb);
compare("spilling to the budget", &n_serial, &n_bnd);
for (name, arm) in &sweeps {
compare(&format!("{name} against serial"), &n_serial, arm);
}
drop(compare);
// ── The trap arm: k private accumulators, run concurrently ────────────────
let mut dup_peak = 0u64;
let mut dup_baseline = 0u64;
if dup_k > 1 {
eprintln!(
"\ntile_pyramid_parallel: dup-accumulator arm — {dup_k} tilers at once, each \
holding its own pyramid. THIS IS THE SHAPE THAT MUST NOT PASS."
);
// The dup arm needs its baseline for the SAME reason every other arm does —
// and here the contamination would push the check the WRONG way, letting an
// inherited free list stand in for a ceiling that fired. A decoration check
// that is itself decorated is worse than none.
let _ = osm_katana::phase_log::trim_malloc();
dup_baseline = osm_katana::phase_log::rss_bytes();
let watch = RssWatch::start();
let t0 = Instant::now();
std::thread::scope(|s| {
let hs: Vec<_> = (0..dup_k)
.map(|k| {
let out = out_root.join(format!("dup{k}"));
let extract = &extract;
let base = &base;
s.spawn(move || {
let _ = std::fs::remove_dir_all(&out);
std::fs::create_dir_all(&out).expect("create dup scratch");
let opts = TileOptions {
workers: (cores / dup_k).max(1),
mem_budget_bytes: budget,
..base.clone()
};
let st = tile_region(TileSource::Extract(extract), &out, &opts)
.expect("dup tile_region");
let _ = std::fs::remove_dir_all(&out);
st.peak_resident_bytes
})
})
.collect();
let mut sum = 0u64;
for h in hs {
sum += h.join().expect("dup arm");
}
eprintln!(
"tile_pyramid_parallel: dup arm summed resident payload {:.2} GiB in {:.1} s",
gib(sum),
t0.elapsed().as_secs_f64()
);
});
dup_peak = watch.finish().saturating_sub(dup_baseline);
}
// ── table ─────────────────────────────────────────────────────────────────
println!(
"\n{:<20} {:>7} {:>9} {:>9} {:>8} {:>7} {:>8} {:>7} {:>8} {:>10} {:>8} {:>10} {:>9} {:>8}",
"arm",
"workers",
"budgetGiB",
"wall_s",
"scan_s",
"scan_c",
"write_s",
"wr_c",
"baseGiB",
"peakRSSGiB",
"armGiB",
"payloadGiB",
"spillGiB",
"tiles",
);
for r in &rows {
println!(
"{:<20} {:>7} {:>9.2} {:>9.2} {:>8.2} {:>7.1} {:>8.2} {:>7.1} {:>8.2} {:>10.2} \
{:>8.2} {:>10.2} {:>9.2} {:>8}",
r.name,
r.workers,
r.budget_gib,
r.wall_s,
r.scan_s,
r.scan_cores,
r.write_s,
r.write_cores,
gib(r.rss_baseline),
gib(r.peak_rss),
gib(r.arm_rss()),
gib(r.payload),
gib(r.spilled),
r.tiles,
);
}
println!(
" armGiB = peakRSSGiB - baseGiB: what the arm ITSELF cost. peakRSSGiB is the whole \n \u{20}\u{20}process and includes every earlier arm's allocator free list."
);
// ── where each arm's memory WENT ──────────────────────────────────────────
// The table above says how much; this says what of. Printed from the engine's
// own renderer so the bench cannot describe a different run than the engine did.
println!("\nmemory attribution, per arm (at each arm's scan high-water mark):");
for r in &rows {
println!(" {:<20} {}", r.name, r.st.memory_attribution());
}
let find = |n: &str| rows.iter().find(|r| r.name == n);
let serial = find("serial-unbounded");
let punb = find("parallel-unbounded");
let bounded = find("parallel-bounded");
if let (Some(serial), Some(bounded)) = (serial, bounded) {
println!(
"\nscan {:.2} s at {:.1} of {cores} cores → {:.2} s at {:.1} cores = {:.2}x\n\
whole {:.2} s → {:.2} s = {:.2}x (the write phase already fanned out, so the \
whole-tiling ratio is well below the scan ratio — quote THIS one)\n\
peak {:.2} GiB → {:.2} GiB = {:.2}x, of which {:.2} GiB of payload went to spill",
serial.scan_s,
serial.scan_cores,
bounded.scan_s,
bounded.scan_cores,
serial.scan_s / bounded.scan_s.max(1e-9),
serial.wall_s,
bounded.wall_s,
serial.wall_s / bounded.wall_s.max(1e-9),
gib(serial.peak_rss),
gib(bounded.peak_rss),
gib(serial.peak_rss) / gib(bounded.peak_rss).max(1e-9),
gib(bounded.spilled),
);
}
println!(
"VmHWM (kernel, whole process across every arm): {:.2} GiB",
gib(osm_katana::phase_log::peak_rss_bytes())
);
println!(
"box: {cores} cores, 1-min load {load0:.1} at start / {:.1} at end{}",
load_avg_1(),
if contended {
" ← CONTENDED: every occupancy figure above is an UNDER-count of what this \
tiling can hold, by an amount this bench cannot attribute"
} else {
""
},
);
// ── ZEROTH GATE: did this run measure ANYTHING? ───────────────────────────
//
// `OSM_KATANA_TILE_ARMS=bounded` matches no arm — they are `parallel-bounded`,
// `serial-unbounded`, `parallel-unbounded`. The filter silently selected nothing,
// every arm returned `None`, and the bench printed **`GATE: PASS`**. The per-arm
// `assert!(tiles > 0, "produced no tiles — nothing was benched")` could not catch
// it, because that assertion lives INSIDE an arm and no arm ran. A green from a
// bench that measured nothing is worse than a red: it is a measurement-shaped hole
// that a table can be written from.
//
// Both halves are needed. Naming a bad arm must fail even if a good arm was also
// named (or the typo just quietly narrows the run), and an empty result set must
// fail even when the filter was legal (a future dispatch change could produce one).
{
let offered = offered.borrow();
if let Some(f) = &arms_filter {
let unknown: Vec<&String> = f.iter().filter(|a| !offered.contains(a)).collect();
assert!(
unknown.is_empty(),
"OSM_KATANA_TILE_ARMS names {unknown:?}, which no arm of this bench is \
called. The arms offered by this run are {offered:?}. A filter that \
matches nothing runs nothing and still prints a verdict, so it is a \
hard error and not a warning."
);
}
assert!(
!rows.is_empty(),
"no arm ran: this bench measured NOTHING and must not report a verdict. \
Arms offered: {offered:?}; filter: {arms_filter:?}."
);
}
// ── the gate ──────────────────────────────────────────────────────────────
// Three independent verdicts, and the third is the one this bench got wrong on
// 2026-08-03: it bounded RSS and it bounded the payload, and it never asked
// whether the two were talking about the same memory. A budget that governs a
// quantity it cannot connect to RSS governs nothing.
let mut fail: Vec<String> = Vec::new();
fail.extend(equivalence);
if let Some(bounded) = bounded {
// 1. INHERITED — checked FIRST, because it decides whether verdict 2 means
// anything. If the process carried a large baseline into this arm, the
// arm's own figure is a difference of two big numbers and no memory verdict
// below it is worth reading. This is the 2026-08-03 defect made into a gate:
// the bounded arm was charged 15.20 GiB of a process it had joined, not
// filled, and the branch was held back on it.
if bounded.rss_baseline > ceiling / 2 {
fail.push(format!(
"INHERITED: parallel-bounded started at {:.2} GiB RSS — over half the \
{ceiling_gib:.2} GiB ceiling was already resident before the arm ran, and \
`malloc_trim` could not give it back. Every per-arm RSS figure in this table \
is then a subtraction of two large numbers. Run the arm alone \
(OSM_KATANA_TILE_ARMS=parallel-bounded).",
gib(bounded.rss_baseline)
));
}
// 2. MEMORY. What the arm ITSELF cost lives under the ceiling.
if bounded.arm_rss() > ceiling {
fail.push(format!(
"MEMORY: parallel-bounded cost {:.2} GiB RSS of its own ({:.2} peak - {:.2} \
baseline), over the {ceiling_gib:.2} GiB ceiling for a {budget_gib:.2} GiB \
payload budget",
gib(bounded.arm_rss()),
gib(bounded.peak_rss),
gib(bounded.rss_baseline),
));
}
// 3. BUDGET. The thing the budget bounds must be bounded. A 5 % slack because
// the check happens per flush, not per placement.
if budget > 0 && bounded.payload > budget + budget / 20 {
fail.push(format!(
"BUDGET: accounted payload peaked at {:.2} GiB against a {budget_gib:.2} GiB \
budget — the spill is not holding the line it exists to hold",
gib(bounded.payload)
));
}
// 4. FLOOR. The part of residency a spill CANNOT free — the per-tile dedup
// sets, the spilled-run indexes, the shard tables — must fit inside the
// budget, or the budget is not reachable at any spill rate and the caller
// is being told a number the engine cannot honour. This is the gate that
// turns "the accounting was extended" into something with teeth: raising
// the zoom set or lowering the budget makes it red, and it should.
let floor = bounded.st.peak_bookkeeping_bytes;
if budget > 0 && floor > budget {
fail.push(format!(
"FLOOR: the unspillable bookkeeping alone is {:.2} GiB — MORE than the \
{budget_gib:.2} GiB budget. Spilling every byte of payload would not reach \
that budget on this corpus; it is not a budget this engine can honour.",
gib(floor)
));
}
// 5. The budget must be doing the work, not the input being small: if nothing
// spilled AND the unbounded arm fitted anyway, the memory arms are hollow.
if let Some(punb) = punb {
if bounded.spilled == 0 && punb.arm_rss() <= ceiling {
fail.push(format!(
"HOLLOW: nothing spilled and the UNBOUNDED arm already fitted under the \
ceiling ({:.2} GiB) — this corpus cannot demonstrate the budget. Use a \
bigger extract, more zooms, or a smaller OSM_KATANA_TILE_BUDGET_GIB",
gib(punb.arm_rss())
));
}
}
// 6. TIME. The scan was one core; fanning it out must buy most of it back.
if let Some(serial) = serial {
const MIN_SCAN_SPEEDUP: f64 = 4.0;
let scan_x = serial.scan_s / bounded.scan_s.max(1e-9);
if scan_x < MIN_SCAN_SPEEDUP {
fail.push(format!(
"TIME: the scan went {:.2} s → {:.2} s = {scan_x:.2}x on {cores} cores, under \
the {MIN_SCAN_SPEEDUP:.1}x floor",
serial.scan_s, bounded.scan_s
));
}
}
// 7. TIME, occupancy: a wall-clock win with one core busy is a measurement
// bug — UNLESS the box was contended, in which case this bench has already
// said it cannot attribute the shortfall and must not convict on it.
if bounded.scan_cores < 4.0 && !contended {
fail.push(format!(
"OCCUPANCY: the bounded scan held {:.1} cores of {cores} — the fan-out is not \
fanning out",
bounded.scan_cores
));
}
}
// 2. MEMORY, the other direction — the ceiling must be capable of firing.
// The dup arm is the residency shape a per-worker accumulator would have.
if dup_k > 1 && dup_peak <= ceiling {
fail.push(format!(
"GATE IS DECORATION: {dup_k} concurrent private accumulators peaked at only \
{:.2} GiB, under the {ceiling_gib:.2} GiB ceiling — this ceiling could not \
report red, so nothing above it means anything",
gib(dup_peak)
));
}
if dup_k > 1 {
println!(
"\ndup-accumulator ({dup_k} private pyramids): {:.2} GiB of its OWN RSS \
(from a {:.2} GiB baseline) vs the {ceiling_gib:.2} GiB ceiling — the ceiling \
{} it, as it must.",
gib(dup_peak),
gib(dup_baseline),
if dup_peak > ceiling {
"REJECTS"
} else {
"ADMITS"
},
);
}
if fail.is_empty() {
println!(
"\nGATE: PASS — memory under ceiling, ceiling proven able to fire, scan fanned out."
);
return;
}
println!("\nGATE: FAIL");
for f in &fail {
println!(" · {f}");
}
std::process::exit(1);
}