examples/sat_probe.rs
//! Focused core-saturation probe for the Mimir bulk paths under optimization
//! (cycle 2). Runs ONE production path in a wall-clock loop under the real bench
//! `Sampler` (the same /proc 1 Hz telemetry the suite folds per row) and prints
//! `proc_cores_max / n_cores` + throughput — so a single path's saturation can be
//! proven in seconds instead of re-running the whole ~5 min suite (the 10M arms).
//!
//! `cores` is **this process's** occupancy (task-clock ÷ wall, `/proc/self/stat`).
//! `host-busy` is the whole box (`/proc/stat`) and is printed only as ambient
//! context: on a loaded machine it reads N/N no matter what the probe does, so
//! reading a saturation verdict off it is how a serial path came to look
//! 32/32-core parallel. Never compare probes on `host-busy`.
//!
//! cargo run --release --example sat_probe --features vector -- <probe> <size> <secs>
//!
//! probes: context | context_radius | append | embeddings | ragnar
//!
//! The `ragnar` probe needs no `--features vector`:
//! cargo run --release --example sat_probe -- ragnar 10000000 4
//!
//! Not shipped, not registered — a developer probe for the never-ending bench
//! loop. The authoritative numbers still come from `examples/nornir-bench`.
use anyhow::Result;
use nornir::bench::telemetry::Sampler;
use nornir::knowledge::git_heat::{GitHeatRow, GitHeatScan};
use nornir::knowledge::query::KnowledgeView;
use nornir::knowledge::symbols::{CallEdgeRow, SymbolRow};
use nornir::vector::chunk::{Chunk, ChunkOptions, hash_text};
use nornir::vector::store::{IndexParams, ModelProfile, chunker_hash, index_snapshot};
use nornir::warehouse::iceberg::IcebergWarehouse;
fn mk_vec(seed: u64, dim: usize) -> Vec<f32> {
(0..dim)
.map(|d| {
let x = (seed.wrapping_mul(2_654_435_761).wrapping_add(d as u64)) as f32;
(x * 0.000_1).sin() + 0.001 * (d as f32 + 1.0)
})
.collect()
}
fn profile(dim: usize) -> ModelProfile {
ModelProfile {
model_name: "bench-embed".into(),
weights_sha: "w0".into(),
tokenizer_sha: "t0".into(),
pooling: "mean".into(),
normalize: true,
dim,
dtype: "f32".into(),
}
}
fn mk_chunks(n: usize) -> Vec<Chunk> {
(0..n)
.map(|i| {
let text = format!("chunk_{i}_token_{}", i % 997);
Chunk {
file: format!("src/f{}.rs", i % 512),
start_line: (i % 4000) + 1,
end_line: (i % 4000) + 12,
content_hash: hash_text(&text),
text,
}
})
.collect()
}
fn mk_symbol(i: usize) -> SymbolRow {
SymbolRow {
crate_name: "app".into(),
module_path: format!("app::m{}", i % 512),
item_kind: if i % 4 == 0 { "struct" } else { "fn" }.into(),
item_name: format!("f{i}"),
visibility: if i % 3 == 0 { "pub" } else { "pub(crate)" }.into(),
file: format!("src/m{}.rs", i % 512),
line: (i % 4000) as u32 + 1,
doc_lines: (i % 5) as u32,
signature: Some(format!("fn f{i}(x: u32) -> u32")),
}
}
fn mk_call(i: usize, n: usize) -> Vec<CallEdgeRow> {
let mut out = Vec::with_capacity(2);
let caller = format!("app::m{}::f{}", i % 512, i);
if i + 1 < n {
out.push(CallEdgeRow {
crate_name: "app".into(),
caller_path: caller.clone(),
callee_ident: format!("app::m{}::f{}", (i + 1) % 512, i + 1),
call_kind: "call".into(),
file: format!("src/m{}.rs", i % 512),
line: (i % 4000) as u32 + 1,
});
}
out.push(CallEdgeRow {
crate_name: "app".into(),
caller_path: caller,
callee_ident: "util::helper".into(),
call_kind: "call".into(),
file: format!("src/m{}.rs", i % 512),
line: (i % 4000) as u32 + 2,
});
out
}
/// The in-memory knowledge view the context probe joins against.
fn mk_view(n_sym: usize, n_call: usize) -> KnowledgeView {
let symbols: Vec<SymbolRow> = (0..n_sym).map(mk_symbol).collect();
let mut calls = Vec::with_capacity(n_call * 2);
for i in 0..n_call {
calls.extend(mk_call(i, n_call));
}
KnowledgeView::new(symbols, calls)
}
/// A `git_heat_facts` scan of `rows` files — the append probe's commit payload.
fn mk_heat(repo: &str, rows: usize) -> GitHeatScan {
let ts = chrono::Utc::now();
GitHeatScan {
snapshot_id: uuid::Uuid::new_v4(),
ts,
repo: repo.into(),
files: (0..rows)
.map(|i| GitHeatRow {
file: format!("src/m{}/f{i}.rs", i % 512),
commits_total: (i % 97) as i64 + 1,
commits_30d: (i % 7) as i64,
commits_90d: (i % 23) as i64,
authors_total: (i % 5) as i64 + 1,
last_commit_ts: ts,
})
.collect(),
}
}
fn seed_snapshot(wh: &IcebergWarehouse, repo: &str, chunks: &[Chunk], dim: usize) -> Result<()> {
let m = profile(dim);
let ch = chunker_hash(&ChunkOptions::default());
let params = IndexParams {
workspace: "ws",
repo,
git_sha: "sha1",
branch: "main",
model: &m,
chunker_hash: &ch,
complete: true,
};
index_snapshot(wh, ¶ms, chunks, |missing| {
Ok(missing
.iter()
.enumerate()
.map(|(i, _)| mk_vec(i as u64, dim))
.collect())
})?;
Ok(())
}
fn main() -> Result<()> {
let args: Vec<String> = std::env::args().collect();
let probe = args.get(1).map(|s| s.as_str()).unwrap_or("context");
let size: usize = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(20_000);
let secs: f64 = args.get(3).and_then(|s| s.parse().ok()).unwrap_or(4.0);
let budget = std::time::Duration::from_secs_f64(secs);
match probe {
"context" | "context_radius" => {
use nornir::knowledge::context::{ChangeRadius, ContextQuery, context_slices};
let dim = 32usize;
let chunks = mk_chunks(size);
let dir = tempfile::tempdir()?;
let wh = IcebergWarehouse::open(dir.path())?;
seed_snapshot(&wh, "repo", &chunks, dim)?;
let n_sym = size.min(20_000);
let view = mk_view(n_sym, n_sym);
let mp = profile(dim).id();
let radius = if probe == "context_radius" {
Some(ChangeRadius {
seeds: vec!["f0".into(), "f1".into()],
hops: 3,
include_callers: true,
include_callees: true,
})
} else {
None
};
let k: usize = args.get(4).and_then(|s| s.parse().ok()).unwrap_or(10);
let mk_q = |_qi: u64| ContextQuery {
repo: "repo",
git_sha: Some("sha1"),
model_profile: &mp,
k,
radius: radius.clone(),
};
let _ = context_slices(&wh, &view, &mk_q(1), &mk_vec(1, dim))?; // warm
let sampler = Sampler::start();
let t0 = std::time::Instant::now();
let mut q = 0u64;
let mut slices = 0u64;
while t0.elapsed() < budget {
let qv = mk_vec(q.wrapping_mul(31).wrapping_add(7), dim);
slices += context_slices(&wh, &view, &mk_q(q), &qv)?.len() as u64;
q += 1;
}
let el = t0.elapsed();
let t = sampler.stop();
println!(
"probe={probe} size={size} queries={q} slices={slices} qps={:.0} cores={:.2}/{} host-busy={} cpu_avg={:.0}%",
q as f64 / el.as_secs_f64(),
t.proc_cores_max.unwrap_or(f64::NAN),
t.n_cores,
t.cores_busy_max,
t.cpu_pct_avg
);
}
"append" => {
let dir = tempfile::tempdir()?;
let wh = IcebergWarehouse::open(dir.path())?;
wh.append_git_heat_scan(&mk_heat("warm", 16))?;
let sampler = Sampler::start();
let t0 = std::time::Instant::now();
let mut commits = 0u64;
while t0.elapsed() < budget {
wh.append_git_heat_scan(&mk_heat("repo", size))?;
commits += 1;
}
let el = t0.elapsed();
let t = sampler.stop();
println!(
"probe=append rows_each={size} commits={commits} rows_per_sec={:.0} cores={:.2}/{} host-busy={} cpu_avg={:.0}%",
(commits as usize * size) as f64 / el.as_secs_f64(),
t.proc_cores_max.unwrap_or(f64::NAN),
t.n_cores,
t.cores_busy_max,
t.cpu_pct_avg
);
}
"embeddings" => {
let dim = 32usize;
let sampler = Sampler::start();
let t0 = std::time::Instant::now();
let mut runs = 0u64;
while t0.elapsed() < budget {
let chunks = mk_chunks(size);
let dir = tempfile::tempdir()?;
let wh = IcebergWarehouse::open(dir.path())?;
seed_snapshot(&wh, "repo", &chunks, dim)?;
runs += 1;
}
let el = t0.elapsed();
let t = sampler.stop();
println!(
"probe=embeddings chunks={size} runs={runs} chunks_per_sec={:.0} cores={:.2}/{} host-busy={} cpu_avg={:.0}%",
(runs as usize * size) as f64 / el.as_secs_f64(),
t.proc_cores_max.unwrap_or(f64::NAN),
t.n_cores,
t.cores_busy_max,
t.cpu_pct_avg
);
}
// CYCLE-3: RagnarLog::build index-build sort — serial `slice::sort_by`
// vs the repointed `gatling_sort_by`. Same `(key, orig, payload)` tuples
// the build sorts; runs each in a wall-clock loop under the Sampler and
// prints the A/B. Also asserts the two orders are byte-identical.
"ragnar" | "ragnar_build" | "sort" => {
use std::hint::black_box;
use std::time::Instant;
use znippy_zoomies::gatling_sort::gatling_sort_by;
// 4th arg: "sorted" (default, = the ragnar_build bench fixture) or
// "shuffled" (keys in random order — the adversarial case for an
// adaptive serial sort).
let shape = args.get(4).map(|s| s.as_str()).unwrap_or("sorted");
let rows: Vec<(i64, u64)> = if shape == "shuffled" {
(0..size as u64)
.map(|k| {
// xorshift-scrambled key so the input is NOT monotonic.
let mut x = k.wrapping_add(0x9E37_79B9_7F4A_7C15);
x ^= x >> 30;
x = x.wrapping_mul(0xBF58_476D_1CE4_E5B9);
x ^= x >> 27;
x = x.wrapping_mul(0x94D0_49BB_1331_11EB);
x ^= x >> 31;
((x % (size as u64 * 2)) as i64, k)
})
.collect()
} else {
(0..size as i64).map(|k| (k * 2, k as u64)).collect()
};
eprintln!("shape={shape}");
let cmp =
|a: &(i64, usize, u64), b: &(i64, usize, u64)| a.0.cmp(&b.0).then(b.1.cmp(&a.1));
let tag = |r: &[(i64, u64)]| -> Vec<(i64, usize, u64)> {
r.iter()
.copied()
.enumerate()
.map(|(i, (k, v))| (k, i, v))
.collect()
};
// BYTE-IDENTICAL proof: the two sorts produce the same order.
let mut a = tag(&rows);
let mut b = a.clone();
a.sort_by(cmp);
gatling_sort_by(&mut b, cmp);
assert_eq!(a, b, "gatling_sort_by diverged from slice::sort_by");
println!("probe=ragnar size={size} byte_identical=true");
let run = |parallel: bool| {
let s = Sampler::start();
let t0 = Instant::now();
let (mut builds, mut acc) = (0u64, 0i64);
while t0.elapsed() < budget {
let mut t = tag(&rows);
if parallel {
gatling_sort_by(&mut t, cmp);
} else {
t.sort_by(cmp);
}
t.dedup_by_key(|x| x.0);
acc = acc.wrapping_add(t.first().map(|x| x.0).unwrap_or(0));
black_box(&t);
builds += 1;
}
let el = t0.elapsed();
let telem = s.stop();
let mode = if parallel { "gatling" } else { "serial " };
println!(
"probe=ragnar mode={mode} keys={size} builds={builds} keys_per_sec={:.0} cores={:.2}/{} host-busy={} cpu_avg={:.0}% (chk={acc})",
(builds as usize * size) as f64 / el.as_secs_f64(),
telem.proc_cores_max.unwrap_or(f64::NAN),
telem.n_cores,
telem.cores_busy_max,
telem.cpu_pct_avg
);
};
run(false); // A: serial slice::sort_by (the old path)
run(true); // B: gatling_sort_by (the repointed path)
}
other => anyhow::bail!("unknown probe {other}"),
}
Ok(())
}