examples/lens_fanout_bench.rs
//! The fan-out, measured: the SAME room snapshot, 1 worker against N, each
//! `--runs` times, for the lenses whose hot loop fans out on gatling —
//! `group_twins` (candidate judging + in-function scan) and
//! `structural_clones` (the structural pair pass).
//!
//! ```text
//! cargo run --release --no-default-features --features code-room,vector \
//! --example lens_fanout_bench -- <lake-dir> <repo> [--runs 3] [--workers 1,6]
//! ```
//! `<lake-dir>` is a COPY of a server's lake (the lake is opened by this
//! process only). Per (lens, workers): every run's wall ms and cores used
//! (process CPU time / wall), then the median, and whether every run gave
//! the same answer as the 1-worker run.
#[cfg(feature = "code-room")]
fn cpu_ms() -> u64 {
// utime + stime of this process, in clock ticks (100 Hz on Linux)
let s = std::fs::read_to_string("/proc/self/stat").unwrap_or_default();
let rest = s.rsplit_once(')').map(|x| x.1).unwrap_or("");
let f: Vec<&str> = rest.split_whitespace().collect();
let t = |i: usize| f.get(i).and_then(|v| v.parse::<u64>().ok()).unwrap_or(0);
(t(11) + t(12)) * 10
}
#[cfg(feature = "code-room")]
fn rss_gb() -> f64 {
let s = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
s.lines().find_map(|l| l.strip_prefix("VmRSS:")).and_then(|v| v.split_whitespace().next()).and_then(|v| v.parse::<f64>().ok()).unwrap_or(0.0) / 1048576.0
}
#[cfg(feature = "code-room")]
fn measure<T>(runs: usize, mut f: impl FnMut() -> T) -> (Vec<(u64, f64)>, Vec<T>) {
let mut out = Vec::new();
let mut answers = Vec::new();
for _ in 0..runs {
let (c0, t0) = (cpu_ms(), std::time::Instant::now());
answers.push(f());
let wall = t0.elapsed().as_millis().max(1) as u64;
out.push((wall, (cpu_ms() - c0) as f64 / wall as f64));
}
(out, answers)
}
#[cfg(feature = "code-room")]
fn main() -> anyhow::Result<()> {
use mimir_room::{MirSide, Room, Scope};
use nornir::clone_signals::structure;
use nornir::mimir_groupclone_graph::{GraphOptions, gather, pair};
let mut args = std::env::args().skip(1);
let lake = std::path::PathBuf::from(args.next().expect("usage: lens_fanout_bench <lake-dir> <repo> [--runs N] [--workers 1,6]"));
let repo = args.next().expect("<repo>");
let (mut runs, mut workers) = (3usize, vec![1usize, 6]);
while let Some(a) = args.next() {
match a.as_str() {
"--runs" => runs = args.next().expect("--runs N").parse()?,
"--workers" => workers = args.next().expect("--workers 1,6").split(',').map(|w| w.parse()).collect::<Result<_, _>>()?,
o => anyhow::bail!("unknown argument {o}"),
}
}
if std::env::var("BENCH_DROP_ZONE").is_ok_and(|v| v == "1") {
// a lake copied without its READ-FAST zone: drop the interval tables
// from the catalog, so every read folds the event tables (the same
// rows: the zone is derived, readfast's oracle)
use skade::iceberg::Catalog as _;
let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build()?;
rt.block_on(async {
let wh = skade::open(&lake).await?;
for t in ["code_nodes_v", "code_edges_v", "code_spans_v"] {
let ident = wh.table_ident(&format!("mimir_v2.{t}"))?;
if wh.catalog().table_exists(&ident).await? {
wh.catalog().drop_table(&ident).await?;
println!("dropped {t} from the copy's catalog");
}
}
anyhow::Ok(())
})?;
}
let log = mimir_ingest::EventLog::open_lake_blocking(&lake)?;
if repo == "--rooms" {
// what each room costs: RSS after the workspace room, after the ring
// view beside it, and after the cache let go of their kept folds
println!("rss at start {:.2} GB", rss_gb());
let t = std::time::Instant::now();
let ws = Room::open_blocking(&log, &Scope::current())?.room;
let st = ws.stats();
println!("workspace room: {} nodes, {} edges, node_bytes {:.2} GB, {} ms; rss {:.2} GB", ws.len(), st.edges.values().sum::<u64>(), st.build.node_bytes as f64 / 1e9, t.elapsed().as_millis(), rss_gb());
let t = std::time::Instant::now();
let ring = Room::open_blocking(&log, &Scope::current().with_universe(true))?.room;
let st = ring.stats();
println!("ring view: {} nodes, {} edges, node_bytes {:.2} GB, {} ms; rss {:.2} GB", ring.len(), st.edges.values().sum::<u64>(), st.build.node_bytes as f64 / 1e9, t.elapsed().as_millis(), rss_gb());
let mut kinds: Vec<(String, u64)> = st.edges.iter().map(|(k, n)| (k.to_string(), *n)).collect();
kinds.sort_by(|a, b| b.1.cmp(&a.1));
println!("ring edges by kind: {:?}", &kinds[..kinds.len().min(8)]);
let mut nk: Vec<(String, u64)> = st.node_kinds.iter().map(|(k, n)| (k.clone(), *n)).collect();
nk.sort_by(|a, b| b.1.cmp(&a.1));
println!("ring nodes by kind: {:?}; universe rows {}", &nk[..nk.len().min(10)], st.universe_rows);
Room::clear_cache();
println!("cache cleared (folds dropped, both snapshots still held): rss {:.2} GB", rss_gb());
drop(ring);
println!("ring view dropped: rss {:.2} GB", rss_gb());
return Ok(());
}
let t = std::time::Instant::now();
let room = Room::open_blocking(&log, &Scope::current())?.room;
println!("room: {} nodes, {} ms", room.len(), t.elapsed().as_millis());
let upto = room.scope().repos.get(&repo).copied();
let side = log.block_on(MirSide::load(&log, &repo, upto, 2048 << 20))?;
let g = gather(&room, &repo, &side)?;
println!("group_twins input: {} fns of {repo}, {} MIR blocks beside the room", g.fns.len(), side.len());
let median = |v: &[(u64, f64)]| {
let mut w: Vec<(u64, f64)> = v.to_vec();
w.sort_by_key(|x| x.0);
w[w.len() / 2]
};
let report = |lens: &str, w: usize, m: &[(u64, f64)], same: bool| {
let (mw, mc) = median(m);
let all: Vec<String> = m.iter().map(|(w, c)| format!("{w} ms/{c:.1}c")).collect();
println!("{lens:<18} workers {w:>2}: median {mw} ms, {mc:.1} cores [{}] same answer as 1 worker: {same}", all.join(", "));
};
// group_twins: the budget out of the way, so every run judges every candidate
let mut base: Option<String> = None;
for &w in &workers {
let opts = GraphOptions { threads: w, budget_ms: 3_600_000, ..GraphOptions::from_env() };
let (m, answers) = measure(runs, || serde_json::to_string(&pair(&g.fns, &g.rarity, &opts).twins).unwrap_or_default());
let b = base.get_or_insert_with(|| answers[0].clone()).clone();
report("group_twins", w, &m, answers.iter().all(|a| *a == b));
}
// structural_clones: the structural pair pass over the room's functions of the repo
let mask = room.mask(&mimir_room::MaskSpec { repos: vec![repo.clone()], node_kinds: vec!["fn".into(), "method".into()], defined_only: true, is_test: Some(false), ..Default::default() })?;
let ords: Vec<u64> = mask.value.ordinals().to_vec();
let f = structure::gather(&room, &ords)?;
println!("structural input: {} fns", ords.len());
let mut base: Option<usize> = None;
for &w in &workers {
let (m, answers) = measure(runs, || {
let (p, _) = structure::structural_pairs_on(&f, structure::STRUCT_FLOOR, None, w);
p.iter().map(|x| (x.a as u64).wrapping_mul(1_000_003) ^ x.b as u64 ^ x.score.to_bits() as u64).fold(p.len(), |h, x| h.wrapping_mul(31).wrapping_add(x as usize))
});
let b = *base.get_or_insert(answers[0]);
report("structural_clones", w, &m, answers.iter().all(|a| *a == b));
}
Ok(())
}
#[cfg(not(feature = "code-room"))]
fn main() {
eprintln!("lens_fanout_bench needs --features code-room,vector");
}