osm-katana/benches/pbf_pass2_plan.rs
//! **What the pass-2 blob plan is worth, on a corpus that fills all 32 cores.**
//!
//! A `.pbf` convert reads the file twice. Pass 1 inflates every blob to build the
//! node store; pass 2 then inflated every blob AGAIN — including the node-only
//! ones, which are the large majority of an OSM extract and cannot, by
//! construction, contain a way. That second inflate is pure waste, and
//! decompression is 85.5 % of a pass.
//!
//! The mmap viewer path (`reader::read_pass2_mmap_planned`) has skipped those
//! blobs for months. The converter could not: a `gatling::TypedCodec::transform`
//! is handed a **slot-relative** slice, so it had no way to name a blob in file
//! coordinates and no key to look a plan up by. `xml_to_pbf::SlotCursor` carries
//! the slot's absolute base on the segment descriptor, which is the key — and
//! the same key is what lets SEVERAL consumers ride one pass instead of each
//! paying for their own.
//!
//! ## What this measures, and why in this shape
//!
//! Both arms are the SHIPPED converter on the same bytes, the same machine and
//! the same minute — the plan arm as it runs, and `OSM_KATANA_PASS2_PLAN=0` for
//! the pre-change behaviour. A remembered number from another day is not a
//! control.
//!
//! Every row carries three things, because a wall-clock figure alone is half a
//! result:
//!
//! * **wall seconds** per phase, straight out of `phase_log` — the same channel
//! `convert --log` writes, so a bench figure and a production figure can never
//! drift.
//! * **core occupancy** (`cpu_cores` of 32, `busy_pct`). A change that got
//! faster by leaving 20 cores idle has not got faster, it has got smaller.
//! * **peak RSS**, sampled per arm plus the kernel's own process-wide `VmHWM`.
//! oden was killed by a 474.9 GiB RSS on 2026-08-03; a speed number without a
//! memory number is how that happens.
//!
//! ## Corpus
//!
//! `OSM_KATANA_BENCH_PBF` names a real `.osm.pbf`. Without it the bench builds
//! one by concatenating the in-repo Liechtenstein extract until it passes
//! `OSM_KATANA_BENCH_MB` (default 1024). PBF is a flat sequence of self-framed
//! blobs, so a concatenation is a well-formed PBF with a real OSM blob mix — the
//! node:way blob ratio, which is the whole subject of this bench, is
//! Liechtenstein's own and not a generator's guess. What it is NOT is a fresh id
//! space: the ids repeat per copy. That inflates the node store and does not
//! change which blobs pass 2 may skip, so it is fine here and would not be fine
//! for a correctness claim — which is why correctness is asserted by
//! `xml_reader::pass2_plan_tests` on a fixture with unique ids, not here.
//!
//! **Size, and why 1 GiB.** The engine reads `.pbf` in 64 MiB slots and cuts
//! each into `2 × n_workers` segments. At 1 GiB that is 16 slots × 62 = ~990
//! work units for 31 workers — ~32 apiece, enough that the tail is not the
//! measurement. Going much smaller is the documented trap: a 2 GiB slice that
//! is only ~10 of an engine's slots measured 16 workers at 73.2 MB/s and 31
//! workers at 63.7 — a REGRESSION produced by the corpus, not the code. Going
//! much bigger only buys wall time, since the node store grows with the corpus
//! and this box has already been killed once today.
//!
//! ```text
//! cargo bench -p osm-katana --bench pbf_pass2_plan
//! OSM_KATANA_BENCH_PBF=/path/to/x.osm.pbf cargo bench -p osm-katana --bench pbf_pass2_plan
//! OSM_KATANA_BENCH_SABOTAGE=1 cargo bench -p osm-katana --bench pbf_pass2_plan # gate self-test
//! ```
/// **A debug build measures nothing here.** `opt-level = 0` leaves the protobuf
/// varint walk, the zstd frame and the bounds checks unoptimised — and it slows
/// the two arms by DIFFERENT factors, because the arm that skips a blob skips a
/// different mix of work than the arm that inflates it. The numbers would still
/// print and would still look like results.
///
/// Refuse instead. A bench that cannot tell you it was run wrong belongs to the
/// same family as a guard that can never report red.
#[cfg(debug_assertions)]
fn main() {
eprintln!(
"pbf_pass2_plan: REFUSING to run under debug_assertions — every figure this \
bench prints would be an artefact of opt-level=0, and the two ARMS would not \
even slow down by the same factor. Run it as \
`cargo bench -p osm-katana --bench pbf_pass2_plan`."
);
std::process::exit(2);
}
#[cfg(not(debug_assertions))]
mod bench {
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
/// `$HOME/work/liechtenstein-latest.osm.pbf` — the default corpus.
fn liechtenstein_pbf() -> String {
format!("{}/work/liechtenstein-latest.osm.pbf", std::env::var("HOME").unwrap_or_default())
}
/// One arm's measurements, all from the run itself.
pub struct Arm {
pub name: &'static str,
pub plan_on: bool,
pub pass1_s: f64,
pub pass2_s: f64,
pub total_s: f64,
pub pass1_cores: f64,
pub pass2_cores: f64,
pub pass1_busy: f64,
pub pass2_busy: f64,
pub peak_rss: u64,
pub blobs: u64,
pub wanted: u64,
pub opened: u64,
pub skipped: u64,
pub nodes: u64,
pub ways: u64,
pub rels: u64,
pub way_bytes: u64,
}
pub fn gib(b: u64) -> f64 {
b as f64 / (1024.0 * 1024.0 * 1024.0)
}
fn env_usize(k: &str, d: usize) -> usize {
std::env::var(k)
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(d)
}
/// Peak-RSS sampler for ONE arm. `VmHWM` is the kernel's exact answer but is
/// monotone and process-wide, so it cannot say which arm peaked; this can.
struct RssWatch {
stop: Arc<AtomicBool>,
peak: Arc<AtomicU64>,
h: Option<std::thread::JoinHandle<()>>,
}
impl RssWatch {
fn start() -> Self {
let stop = Arc::new(AtomicBool::new(false));
let peak = Arc::new(AtomicU64::new(0));
let (s, p) = (Arc::clone(&stop), Arc::clone(&peak));
let h = std::thread::spawn(move || {
while !s.load(Ordering::Relaxed) {
let r = osm_katana::phase_log::rss_bytes();
p.fetch_max(r, Ordering::Relaxed);
std::thread::sleep(std::time::Duration::from_millis(50));
}
p.fetch_max(osm_katana::phase_log::rss_bytes(), Ordering::Relaxed);
});
Self {
stop,
peak,
h: Some(h),
}
}
fn finish(mut self) -> u64 {
self.stop.store(true, Ordering::Relaxed);
if let Some(h) = self.h.take() {
let _ = h.join();
}
self.peak.load(Ordering::Relaxed)
}
}
// ── phase-log parsing (the SAME channel `convert --log` writes) ───────────
fn done_line<'a>(log: &'a str, phase: &str) -> Option<&'a str> {
log.lines()
.filter(|l| {
l.contains(&format!("\"phase\":\"{phase}\"")) && l.contains("\"ev\":\"done\"")
})
.next_back()
}
fn field(line: &str, key: &str) -> Option<f64> {
let pat = format!("\"{key}\":");
let i = line.find(&pat)? + pat.len();
let rest = &line[i..];
let end = rest
.find(|c: char| !(c.is_ascii_digit() || c == '.' || c == '-' || c == '"'))
.unwrap_or(rest.len());
rest[..end].trim_matches('"').parse().ok()
}
fn f(log: &str, phase: &str, key: &str) -> f64 {
done_line(log, phase)
.and_then(|l| field(l, key))
.unwrap_or(0.0)
}
// ── corpus ────────────────────────────────────────────────────────────────
/// The corpus, and the one-line provenance string printed with the results.
pub fn corpus(dir: &Path) -> (PathBuf, String) {
if let Ok(p) = std::env::var("OSM_KATANA_BENCH_PBF") {
let p = PathBuf::from(p);
let mb = std::fs::metadata(&p).map(|m| m.len()).unwrap_or(0) / (1024 * 1024);
let s = format!("REAL {} ({mb} MiB)", p.display());
return (p, s);
}
let want_mb = env_usize("OSM_KATANA_BENCH_MB", 1024);
let default_pbf = liechtenstein_pbf();
let src = std::fs::read(&default_pbf).unwrap_or_else(|e| {
eprintln!(
"pbf_pass2_plan: no corpus. Set OSM_KATANA_BENCH_PBF=<file.osm.pbf>, \
or make {default_pbf} readable ({e})."
);
std::process::exit(2);
});
let copies = ((want_mb * 1024 * 1024) / src.len().max(1)).max(1);
let out = dir.join("bench.osm.pbf");
let mut buf: Vec<u8> = Vec::with_capacity(src.len() * copies);
for _ in 0..copies {
buf.extend_from_slice(&src);
}
std::fs::write(&out, &buf).expect("write bench corpus");
let s = format!(
"SYNTHETIC {copies}× liechtenstein-latest.osm.pbf = {} MiB (ids repeat per copy)",
buf.len() / (1024 * 1024)
);
(out, s)
}
// ── one arm ───────────────────────────────────────────────────────────────
pub fn run(
name: &'static str,
plan_on: bool,
input: &Path,
work: &Path,
workers: usize,
) -> Arm {
let out = work.join(name);
let _ = std::fs::remove_dir_all(&out);
std::fs::create_dir_all(&out).expect("out dir");
let log = out.join("phases.jsonl");
// SAFETY: single-threaded here — no arm is running, and the sampler
// thread below is started after this.
unsafe {
std::env::set_var("OSM_KATANA_PASS2_PLAN", if plan_on { "1" } else { "0" });
}
let opts = osm_katana::ConvertOptions {
output_dir: out.clone(),
include_nodes: true,
include_ways: true,
include_rels: true,
compression: String::from("zstd"),
vtd_workers: workers,
geometry: String::from("resolved"),
skip_changesets: false,
log_path: Some(log.clone()),
clip: None,
keep_node_coords: false,
cancel: None,
};
let watch = RssWatch::start();
let t0 = std::time::Instant::now();
osm_katana::convert(input, &opts).expect("convert");
let wall = t0.elapsed().as_secs_f64();
let peak_rss = watch.finish();
let text = std::fs::read_to_string(&log).expect("phase log");
Arm {
name,
plan_on,
pass1_s: f(&text, "pass1", "wall_s"),
pass2_s: f(&text, "pass2", "wall_s"),
total_s: wall,
pass1_cores: f(&text, "pass1", "cpu_cores"),
pass2_cores: f(&text, "pass2", "cpu_cores"),
pass1_busy: f(&text, "pass1", "busy_pct"),
pass2_busy: f(&text, "pass2", "busy_pct"),
peak_rss,
blobs: f(&text, "pass1", "blobs") as u64,
wanted: f(&text, "pass1", "blobs_with_way_or_rel") as u64,
opened: f(&text, "pass2", "blobs_opened") as u64,
skipped: f(&text, "pass2", "blobs_skipped") as u64,
nodes: f(&text, "pass1", "nodes") as u64,
ways: f(&text, "pass2", "ways") as u64,
rels: f(&text, "pass2", "rels") as u64,
way_bytes: f(&text, "pass2", "way_bytes") as u64,
}
}
}
#[cfg(not(debug_assertions))]
fn main() {
use bench::{Arm, corpus, gib, run};
let n_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let workers = n_cores.saturating_sub(1).max(1);
let dir = tempfile::tempdir().expect("tempdir");
let (input, provenance) = corpus(dir.path());
println!("\npbf_pass2_plan — the pass-2 blob plan on the streaming .pbf path");
println!("corpus: {provenance}");
println!("machine: {n_cores} logical cores, {workers} convert workers");
println!(
"note: both arms are the SHIPPED converter; the `no-plan` arm is \
OSM_KATANA_PASS2_PLAN=0\n"
);
// no-plan FIRST so the plan arm never runs on a colder page cache than its
// control — the direction that would flatter the result.
let no_plan = run("no-plan", false, &input, dir.path(), workers);
let sabotage = std::env::var("OSM_KATANA_BENCH_SABOTAGE").is_ok();
if sabotage {
println!(
"!! OSM_KATANA_BENCH_SABOTAGE set: the `plan` arm runs WITHOUT the plan. \
The gate below must FAIL. This is the gate's own red test.\n"
);
}
let plan = run("plan", !sabotage, &input, dir.path(), workers);
let arms: [&Arm; 2] = [&no_plan, &plan];
println!(
"{:<9} {:>9} {:>9} {:>9} {:>11} {:>11} {:>9} {:>9} {:>9}",
"arm",
"pass1_s",
"pass2_s",
"total_s",
"p1 cores",
"p2 cores",
"peakRSS",
"opened",
"skipped"
);
println!("{}", "-".repeat(104));
for a in arms {
println!(
"{:<9} {:>9.2} {:>9.2} {:>9.2} {:>5.1}/{:<5.0} {:>5.1}/{:<5.0} {:>7.2}GiB {:>9} {:>9}",
a.name,
a.pass1_s,
a.pass2_s,
a.total_s,
a.pass1_cores,
a.pass1_busy,
a.pass2_cores,
a.pass2_busy,
gib(a.peak_rss),
a.opened,
a.skipped,
);
}
println!(
"\nblobs: {} total, {} carry a way or relation ({:.1} %)",
plan.blobs,
plan.wanted,
100.0 * plan.wanted as f64 / plan.blobs.max(1) as f64,
);
println!(
"rows: no-plan {}n/{}w/{}r plan {}n/{}w/{}r ways.parquet {} vs {} bytes",
no_plan.nodes,
no_plan.ways,
no_plan.rels,
plan.nodes,
plan.ways,
plan.rels,
no_plan.way_bytes,
plan.way_bytes,
);
let pass2_speedup = no_plan.pass2_s / plan.pass2_s.max(1e-9);
let total_speedup = no_plan.total_s / plan.total_s.max(1e-9);
println!(
"\npass 2: {:.2} s → {:.2} s = {:.2}× whole convert: {:.2} s → {:.2} s = {:.2}×",
no_plan.pass2_s, plan.pass2_s, pass2_speedup, no_plan.total_s, plan.total_s, total_speedup,
);
println!(
"VmHWM (kernel, whole process across both arms): {:.2} GiB",
gib(osm_katana::phase_log::peak_rss_bytes())
);
// ── the gate ──────────────────────────────────────────────────────────────
//
// Four independent ways to fail, because the interesting failures are not
// "it got slower". Each is stated as what it would catch.
let mut fail: Vec<String> = Vec::new();
// 1. Identical output. The plan is only ever allowed to change WHICH blobs
// get opened. A row that disappeared is not a speedup.
if (no_plan.nodes, no_plan.ways, no_plan.rels) != (plan.nodes, plan.ways, plan.rels) {
fail.push(format!(
"row counts differ: no-plan {}n/{}w/{}r vs plan {}n/{}w/{}r — the plan dropped data",
no_plan.nodes, no_plan.ways, no_plan.rels, plan.nodes, plan.ways, plan.rels
));
}
if no_plan.way_bytes != plan.way_bytes {
fail.push(format!(
"ways.parquet differs in size: {} vs {} bytes",
no_plan.way_bytes, plan.way_bytes
));
}
// 2. The plan must actually skip. This is the arm that would have caught the
// change landing as a no-op — which is the state it came from.
if plan.skipped == 0 {
fail.push(format!(
"the plan arm skipped 0 blobs of {} — the plan is not in effect",
plan.blobs
));
}
if no_plan.skipped != 0 {
fail.push(format!(
"the CONTROL arm skipped {} blobs — it is not a control",
no_plan.skipped
));
}
if plan.opened + plan.skipped != plan.blobs {
fail.push(format!(
"opened {} + skipped {} != {} blobs seen — a blob was counted twice or lost",
plan.opened, plan.skipped, plan.blobs
));
}
// 3. Speed, with a threshold tied to what the corpus can possibly give:
// skipping k % of the blobs cannot save more than k % of pass 2, and must
// save something. Half of the theoretical maximum is the bar.
let skip_frac = plan.skipped as f64 / plan.blobs.max(1) as f64;
let floor = 1.0 / (1.0 - 0.5 * skip_frac).max(1e-9);
if pass2_speedup < floor {
fail.push(format!(
"pass 2 sped up {pass2_speedup:.2}×, below the {floor:.2}× floor implied by \
skipping {:.1} % of blobs",
100.0 * skip_frac
));
}
// 4. Occupancy and memory. A convert that got faster by going serial, or by
// holding the box, is a regression wearing a speedup's clothes.
if plan.pass2_busy < 0.80 * no_plan.pass2_busy {
fail.push(format!(
"pass-2 core occupancy fell from {:.1} % to {:.1} % — fewer cores, not less work",
no_plan.pass2_busy, plan.pass2_busy
));
}
if plan.peak_rss > no_plan.peak_rss + (1 << 30) {
fail.push(format!(
"peak RSS rose {:.2} → {:.2} GiB (> 1 GiB) — the plan is not free",
gib(no_plan.peak_rss),
gib(plan.peak_rss)
));
}
if fail.is_empty() {
if sabotage {
println!(
"\nGATE SELF-TEST FAILED: the gate PASSED with the plan deliberately \
disabled. It cannot report red and is decoration."
);
std::process::exit(1);
}
println!("\nGATE: PASS — same rows, plan in effect, faster, no core or memory regression.");
return;
}
println!("\nGATE: FAIL");
for f in &fail {
println!(" - {f}");
}
if sabotage {
println!("\nGATE SELF-TEST PASSED: the gate reported red on a deliberately disabled plan.");
return;
}
std::process::exit(1);
}