crates/mimir-graph/examples/graph_bench.rs
//! Mimir v2 graph-engine bench: CSR import + lens queries on a random
//! power-law graph. Prints a markdown table (median of R runs) and peak RSS.
//!
//! cargo run --release -p nornir-mimir-graph --example graph_bench -- \
//! [--nodes 1000000] [--edges 10000000] [--sub-nodes 100000] [--sub-edges 1000000] [--runs 5]
//!
//! An example runner rather than a `register_bench!` in nornir's registry:
//! the registry compiles into the `nornir` lib itself, which would make every
//! nornir build (and the release gate) link libgraphblas.
//!
//! Generator: `from = π(⌊n·u^1.5⌋)`, `to = π(⌊n·u^2⌋)` with `u` uniform and
//! `π` a fixed bijection of `0..n` (so hubs are spread over the id space) —
//! skewed out-degree, heavier-skewed in-degree (popular callees), like a call
//! graph. Edges are then sorted by (from, to) and de-duplicated, which is the
//! shape a sorted `code_edges` scan delivers.
use std::sync::Arc;
use std::time::{Duration, Instant};
use arrow_array::{ArrayRef, RecordBatch, UInt64Array};
use mimir_graph::arrow::from_arrow_edge_batches;
use mimir_graph::{Matrix, Vector, ops, runtime};
struct Rng(u64);
impl Rng {
fn next(&mut self) -> u64 {
// splitmix64
self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
let mut z = self.0;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^ (z >> 31)
}
fn unit(&mut self) -> f64 {
(self.next() >> 11) as f64 / (1u64 << 53) as f64
}
}
/// Bijection of 0..n: multiply by a constant coprime to n, then add.
fn perm(x: u64, n: u64) -> u64 {
let mut a = 2_654_435_761 % n;
while gcd(a, n) != 1 {
a += 1;
}
((x as u128 * a as u128 + 12_345) % n as u128) as u64
}
fn gcd(a: u64, b: u64) -> u64 {
if b == 0 { a } else { gcd(b, a % b) }
}
fn power_law_edges(n: u64, m: usize, seed: u64) -> (Vec<u64>, Vec<u64>) {
let mut r = Rng(seed);
let mut e: Vec<(u64, u64)> = (0..m)
.map(|_| {
let f = ((n as f64 * r.unit().powf(1.5)) as u64).min(n - 1);
let t = ((n as f64 * r.unit().powf(2.0)) as u64).min(n - 1);
(perm(f, n), perm(t, n))
})
.collect();
e.sort_unstable();
e.dedup();
e.into_iter().unzip()
}
fn median(mut v: Vec<Duration>) -> Duration {
v.sort();
v[v.len() / 2]
}
/// (min, median) wall time over `runs` runs. On a shared, oversubscribed box
/// the min is the closer estimate of the op's own cost; both are printed.
fn time<T>(runs: usize, mut f: impl FnMut() -> T) -> ((Duration, Duration), T) {
let mut ds = Vec::new();
let mut last = None;
for _ in 0..runs {
let t = Instant::now();
let out = f();
ds.push(t.elapsed());
last = Some(out);
}
let min = *ds.iter().min().unwrap();
((min, median(ds)), last.unwrap())
}
fn ms((min, med): (Duration, Duration)) -> (String, String) {
(
format!("{:.1}", min.as_secs_f64() * 1e3),
format!("{:.1}", med.as_secs_f64() * 1e3),
)
}
fn peak_rss_mb() -> String {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.starts_with("VmHWM:"))
.map(|l| l.split_whitespace().nth(1).unwrap_or("0").to_string())
})
.and_then(|kb| kb.parse::<u64>().ok())
.map(|kb| format!("{:.0}", kb as f64 / 1024.0))
.unwrap_or_else(|| "?".into())
}
fn arg(name: &str, default: u64) -> u64 {
let a: Vec<String> = std::env::args().collect();
a.iter()
.position(|x| x == name)
.and_then(|i| a.get(i + 1))
.and_then(|v| v.parse().ok())
.unwrap_or(default)
}
fn seeds(n: u64, k: usize, seed: u64) -> Vector<bool> {
let mut r = Rng(seed);
let idx: Vec<u64> = (0..k).map(|_| r.next() % n).collect();
Vector::<bool>::from_indices(n, &idx).unwrap()
}
fn main() {
let n = arg("--nodes", 1_000_000);
let m = arg("--edges", 10_000_000) as usize;
let sub_n = arg("--sub-nodes", 100_000);
let sub_m = arg("--sub-edges", 1_000_000) as usize;
let runs = arg("--runs", 5) as usize;
let cores = std::thread::available_parallelism()
.map(|c| c.get())
.unwrap_or(1) as i32;
runtime::set_threads(cores).unwrap();
let load = std::fs::read_to_string("/proc/loadavg").unwrap_or_default();
let t = Instant::now();
let (from, to) = power_law_edges(n, m, 42);
let nnz = from.len();
eprintln!(
"generated {nnz} unique edges in {:.1}s",
t.elapsed().as_secs_f64()
);
// Prebuilt CSR (for from_csr / from_csr_import).
let mut off = vec![0u64; n as usize + 1];
for &f in &from {
off[f as usize + 1] += 1;
}
for i in 0..n as usize {
off[i + 1] += off[i];
}
let mut rows: Vec<(String, (String, String), String)> = Vec::new();
let mut row = |name: &str, d: (Duration, Duration), note: String| {
rows.push((name.to_string(), ms(d), note))
};
let ver = runtime::library_version().unwrap();
// ── Import ──────────────────────────────────────────────────────────────
let batch = RecordBatch::try_from_iter([
(
"from_ord",
Arc::new(UInt64Array::from(from.clone())) as ArrayRef,
),
(
"to_ord",
Arc::new(UInt64Array::from(to.clone())) as ArrayRef,
),
])
.unwrap();
let (d, (a, path)) = time(3, || {
from_arrow_edge_batches(std::slice::from_ref(&batch), "from_ord", "to_ord", n).unwrap()
});
a.wait().unwrap();
row(
"import Arrow, sorted (from,to) → pack",
d,
format!("{path:?}, {} edges", a.nvals().unwrap()),
);
// Same edges shuffled → GraphBLAS build path.
let mut r = Rng(7);
let mut order: Vec<usize> = (0..nnz).collect();
for i in (1..nnz).rev() {
order.swap(i, (r.next() % (i as u64 + 1)) as usize);
}
let sf: Vec<u64> = order.iter().map(|&i| from[i]).collect();
let st: Vec<u64> = order.iter().map(|&i| to[i]).collect();
drop(order);
let shuffled = RecordBatch::try_from_iter([
("from_ord", Arc::new(UInt64Array::from(sf)) as ArrayRef),
("to_ord", Arc::new(UInt64Array::from(st)) as ArrayRef),
])
.unwrap();
let (d, (b, path)) = time(3, || {
let (b, p) =
from_arrow_edge_batches(std::slice::from_ref(&shuffled), "from_ord", "to_ord", n)
.unwrap();
b.wait().unwrap();
(b, p)
});
assert_eq!(b.nvals().unwrap(), nnz as u64);
row("import Arrow, shuffled → GxB build", d, format!("{path:?}"));
drop((b, shuffled));
let (d, c) = time(3, || {
let c = Matrix::<bool>::from_csr(n, n, &off, &to, None).unwrap();
c.wait().unwrap();
c
});
row(
"from_csr (validate + 1 memcpy + GxB pack)",
d,
"zero-copy pack after one copy".into(),
);
drop(c);
let (d, c) = time(3, || {
let c = Matrix::<bool>::from_csr_import(n, n, &off, &to, None).unwrap();
c.wait().unwrap();
c
});
row(
"from_csr_import (validate + GrB_Matrix_import)",
d,
"GraphBLAS copies".into(),
);
drop(c);
drop(batch);
// ── Queries ─────────────────────────────────────────────────────────────
let one = {
let fo = ops::fan_out(&a).unwrap();
let (i, x) = fo.to_tuples().unwrap();
// a node of median out-degree
let mut by_deg: Vec<(u64, u64)> = x.into_iter().zip(i).collect();
by_deg.sort_unstable();
Vector::<bool>::from_indices(n, &[by_deg[by_deg.len() / 2].1]).unwrap()
};
let many = seeds(n, 10_000, 99);
let (d, v) = time(runs, || ops::reach(&one, &a, None).unwrap());
row(
"reach, 1 seed",
d,
format!("{} reached", v.nvals().unwrap()),
);
let (d, v) = time(runs, || ops::reach(&many, &a, None).unwrap());
row(
"reach, 10 000 seeds",
d,
format!("{} reached", v.nvals().unwrap()),
);
let (d, v) = time(runs, || ops::reach(&many, &a, Some(2)).unwrap());
row(
"reach, 10 000 seeds, ≤ 2 hops",
d,
format!("{} reached", v.nvals().unwrap()),
);
let (d, v) = time(runs, || ops::reverse_reach(&one, &a, None).unwrap());
row(
"reverse_reach (mxv on A), 1 seed",
d,
format!("{} reached", v.nvals().unwrap()),
);
let (d, v) = time(runs, || ops::reverse_reach(&many, &a, None).unwrap());
row(
"reverse_reach (mxv on A), 10 000 seeds",
d,
format!("{} reached", v.nvals().unwrap()),
);
let (d, at) = time(3, || {
let t = a.transpose().unwrap();
t.wait().unwrap();
t
});
row("transpose Aᵀ (once, cacheable)", d, String::new());
let (d, v) = time(runs, || ops::reach(&one, &at, None).unwrap());
row(
"reverse walk = reach on cached Aᵀ, 1 seed",
d,
format!("{} reached", v.nvals().unwrap()),
);
let (d, v) = time(runs, || ops::reach(&many, &at, None).unwrap());
row(
"reverse walk = reach on cached Aᵀ, 10 000 seeds",
d,
format!("{} reached", v.nvals().unwrap()),
);
drop(at);
let (d, v) = time(runs, || ops::fan_in(&a).unwrap());
row(
"fan_in (ones PLUS.PAIR A)",
d,
format!("{} nodes with in-edges", v.nvals().unwrap()),
);
let (d, v) = time(runs, || ops::fan_out(&a).unwrap());
row(
"fan_out (row reduce)",
d,
format!("{} nodes with out-edges", v.nvals().unwrap()),
);
let (d, v) = time(runs, || ops::min_hops(&many, &a, None).unwrap());
row(
"min_hops (MIN_PLUS), 10 000 seeds",
d,
format!("{} reached", v.nvals().unwrap()),
);
let sev = {
let mut r = Rng(5);
let idx: Vec<u64> = (0..1000).map(|_| r.next() % n).collect();
let val: Vec<u64> = (0..1000).map(|_| r.next() % 10).collect();
Vector::<u64>::from_tuples(n, &idx, &val).unwrap()
};
let (d, v) = time(3, || ops::max_on_path(&a, &sev).unwrap());
row(
"max_on_path (MAX_SECOND), 1 000 valued nodes",
d,
format!("{} nodes reach a value", v.nvals().unwrap()),
);
let (d, v) = time(3, || ops::components(&a).unwrap());
let ncomp = {
let (i, l) = v.to_tuples().unwrap();
i.iter().zip(&l).filter(|(i, l)| i == l).count()
};
row("components (WCC)", d, format!("{ncomp} components"));
let scope = seeds(n, sub_n as usize, 11);
let (d, s) = time(3, || {
let s = ops::mask_by(&a, &scope).unwrap();
s.wait().unwrap();
s
});
row(
"mask_by, 1e5-node scope",
d,
format!("{} edges kept", s.nvals().unwrap()),
);
drop(s);
// Second snapshot: every 100th edge removed, 1 % new random edges added.
let (from2, to2): (Vec<u64>, Vec<u64>) = {
let (nf, nt) = power_law_edges(n, nnz / 100, 43);
from.iter()
.zip(&to)
.enumerate()
.filter(|(i, _)| i % 100 != 0)
.map(|(_, (&f, &t))| (f, t))
.chain(nf.into_iter().zip(nt))
.unzip()
};
let b2 = Matrix::<bool>::from_edges(n, n, &from2, &to2).unwrap();
drop((from2, to2));
let (d, (ad, rm)) = time(3, || ops::diff(&a, &b2).unwrap());
row(
"diff vs a snapshot with 1 % removed / 1 % added",
d,
format!("+{} / −{}", ad.nvals().unwrap(), rm.nvals().unwrap()),
);
drop((ad, rm, b2));
// ── A·Aᵀ on a separate 1e5-node graph (same generator) ──────────────────
let (sf, st) = power_law_edges(sub_n, sub_m, 44);
let sub = Matrix::<bool>::from_edges(sub_n, sub_n, &sf, &st).unwrap();
let (d, c) = time(3, || {
let c = ops::similarity(&sub).unwrap();
c.wait().unwrap();
c
});
row(
"similarity A·Aᵀ (PLUS_PAIR)",
d,
format!(
"{sub_n} nodes / {} edges → {} entries",
sf.len(),
c.nvals().unwrap()
),
);
drop(c);
println!(
"GraphBLAS {}.{}.{}, {} threads, {} nodes, {} edges (unique), runs = {} for queries / 3 for imports and heavy ops",
ver.0,
ver.1,
ver.2,
runtime::threads().unwrap(),
n,
nnz,
runs
);
println!("loadavg at start: {}", load.trim());
println!();
println!("| op | min ms | median ms | note |");
println!("|---|---:|---:|---|");
for (a, (b0, b1), c) in rows {
println!("| {a} | {b0} | {b1} | {c} |");
}
println!();
println!("peak RSS (VmHWM): {} MB", peak_rss_mb());
}