osm-katana/benches/bz2_block_parallel.rs
//! **Block-parallel bz2: before and after, on the same bytes.**
//!
//! bzip2 compresses in independent, bit-aligned blocks. Reading a `.osm.bz2` one
//! block at a time on one core is the "before" every other tool on this box is
//! stuck with — `bench_history.json` (2026-06-18) records osmium taking 32.4 of
//! 162 GB in ~97 min on its single-stream bzip2 reader, ~8 h projected for the
//! planet. Reading the same blocks across gatling workers is the "after". This
//! bench measures both arms of that, over the same input, in one run — so the
//! ratio is a measurement and not a story.
//!
//! Two arms:
//! * **decompress** — [`osm_katana::xml_scan::bz2_decode_probe`]: decode every
//! block, keep nothing. The bz2 primitive with the payload removed, so its
//! number is attributable to the decompression alone.
//! * **admin-scan** (optional) — the whole three-pass boundary derivation
//! ([`osm_katana::stream_admin_region_polygon_detailed`]) at the same worker
//! counts, so the end-to-end job can be compared with the primitive.
//!
//! Timing and occupancy come from `osm_katana::phase_log` — the SAME channel
//! `convert --log` writes, so a bench row and a production run are directly
//! comparable (and `cpu_cores` / `busy_pct` are measured, not assumed).
//!
//! Run (defaults to the 5.2 MB in-repo fixture, seconds):
//! ```text
//! cargo bench -p osm-katana --bench bz2_block_parallel
//! ```
//! Planet-scale — state the input in any number you report:
//! ```text
//! OSM_KATANA_BENCH_INPUT=~/work/planet-260427.osm.bz2 \
//! OSM_KATANA_BENCH_WORKERS=1,31 \
//! OSM_KATANA_BENCH_REGION=sweden \
//! cargo bench -p osm-katana --bench bz2_block_parallel
//! ```
//! A 1-worker arm on the 173 GB planet takes HOURS by construction — that is the
//! point of the number, but pick the worker list deliberately.
use std::path::{Path, PathBuf};
use std::time::Instant;
/// Default corpus: a real, multi-block `.osm.bz2` that lives in the repo, so the
/// bench runs with no arguments and no download, ever.
const DEFAULT_INPUT: &str = concat!(
env!("CARGO_MANIFEST_DIR"),
"/../lbzip2/test_data/liechtenstein.osm.bz2"
);
fn env_path(key: &str, default: &str) -> PathBuf {
PathBuf::from(std::env::var(key).unwrap_or_else(|_| default.to_string()))
}
fn workers_list() -> Vec<usize> {
let all = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(2);
match std::env::var("OSM_KATANA_BENCH_WORKERS") {
Ok(s) => s.split(',').filter_map(|t| t.trim().parse().ok()).collect(),
Err(_) => vec![1, 2, 4, 8, all.saturating_sub(1).max(1)],
}
}
/// One measured arm.
struct Row {
name: String,
workers: usize,
wall_s: f64,
/// Compressed input bytes actually swept.
input_mb: f64,
/// Decompressed bytes produced (0 when not counted).
decoded_mb: f64,
}
impl Row {
fn mbs(&self) -> f64 {
self.input_mb / self.wall_s
}
}
fn main() {
let input = env_path("OSM_KATANA_BENCH_INPUT", DEFAULT_INPUT);
if !input.exists() {
eprintln!("bz2_block_parallel: input {} not found", input.display());
std::process::exit(1);
}
let input_mb =
std::fs::metadata(&input).map(|m| m.len()).unwrap_or(0) as f64 / (1024.0 * 1024.0);
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let workers = workers_list();
// Phase log → stderr as JSONL: wall_s, cpu_cores, busy_pct per arm, from the
// same mechanism convert uses. No second timing channel (LAW 5).
let sink = osm_katana::phase_log::PhaseSink::new(None).expect("phase sink");
eprintln!(
"bz2_block_parallel: input {} ({input_mb:.1} MiB compressed), {cores} cores, \
workers {workers:?}",
input.display(),
);
let mut rows: Vec<Row> = Vec::new();
// ── Arm 1: decompression only ─────────────────────────────────────────────
// `OSM_KATANA_BENCH_SKIP_CHANGESETS=1` starts at the first data block, the
// same span the admin scan's node pass covers — set it when the number is
// meant to be the FLOOR under that pass rather than the whole file (a planet
// dump carries a changeset prologue the scan never reads).
let skip_changesets =
std::env::var("OSM_KATANA_BENCH_SKIP_CHANGESETS").is_ok_and(|v| v != "0" && !v.is_empty());
for &w in &workers {
let phase = sink.phase(
"bz2_decompress",
&[
("workers", &w.to_string()),
("skip_changesets", &skip_changesets.to_string()),
],
);
let t0 = Instant::now();
let decoded = osm_katana::xml_scan::bz2_decode_probe(&input, w, skip_changesets)
.expect("bz2 decode probe");
let wall_s = t0.elapsed().as_secs_f64();
phase.done(&[("decoded_bytes", decoded.to_string())]);
rows.push(Row {
name: "bz2_decompress".into(),
workers: w,
wall_s,
input_mb,
decoded_mb: decoded as f64 / (1024.0 * 1024.0),
});
}
// ── Arm 2: the whole admin-boundary derivation (opt-in) ───────────────────
if let Ok(region) = std::env::var("OSM_KATANA_BENCH_REGION") {
let level = std::env::var("OSM_KATANA_BENCH_ADMIN_LEVEL").unwrap_or_else(|_| "2".into());
for &w in &workers {
let phase = sink.phase("admin_region_polygon", &[("workers", &w.to_string())]);
let t0 = Instant::now();
let opts = osm_katana::AdminPolyOptions {
admin_level: Some(&level),
n_workers: w,
log_path: None,
};
let got = osm_katana::stream_admin_region_polygon_detailed(
input.to_str().expect("utf8 input path"),
®ion,
&opts,
)
.expect("admin scan");
let wall_s = t0.elapsed().as_secs_f64();
let verts = got.as_ref().map(|d| d.polygon.vertex_count()).unwrap_or(0);
// A bench arm that measured NOTHING is worse than no bench arm.
assert!(
verts > 0,
"admin scan for {region:?} found no boundary — nothing was benched"
);
phase.done(&[("vertices", verts.to_string())]);
rows.push(Row {
name: format!("admin_poly_{region}"),
workers: w,
wall_s,
input_mb,
decoded_mb: 0.0,
});
}
}
// ── bench_history-shaped summary ──────────────────────────────────────────
println!(
"\n{:<24} {:>8} {:>10} {:>10} {:>12} {:>12}",
"name", "workers", "input_mb", "wall_s", "in_mbs", "decoded_mb",
);
for r in &rows {
println!(
"{:<24} {:>8} {:>10.1} {:>10.2} {:>12.1} {:>12.1}",
r.name,
r.workers,
r.input_mb,
r.wall_s,
r.mbs(),
r.decoded_mb,
);
}
// The before/after ratio, per arm: fewest workers vs most.
let mut names: Vec<&str> = rows.iter().map(|r| r.name.as_str()).collect();
names.dedup();
println!();
for name in names {
let arm: Vec<&Row> = rows.iter().filter(|r| r.name == name).collect();
let (Some(lo), Some(hi)) = (
arm.iter().min_by_key(|r| r.workers),
arm.iter().max_by_key(|r| r.workers),
) else {
continue;
};
if lo.workers == hi.workers {
continue;
}
println!(
"{name}: {} worker {:.1} MB/s → {} workers {:.1} MB/s = {:.2}x \
(input {}, {:.1} MiB compressed, {cores} cores)",
lo.workers,
lo.mbs(),
hi.workers,
hi.mbs(),
hi.mbs() / lo.mbs(),
input.file_name().and_then(|f| f.to_str()).unwrap_or("?"),
input_mb,
);
}
let _ = Path::new("");
}