Claude (sandbox)
9bfeb2c6a2
- causal_pipeline: SMARM_CAUSAL_AUDIT=1 appends render_ledger_audit() after the summary; pinned summary format untouched. - causal_attrib_probe: per-pct audit line (absorbed/forgiven/drops/ discards) under the existing eff line, so in-site-vs-attributed and the loss buckets land in one place for the sweep. 1-core smoke (work + wide): books balance — absorbed = 2x injected and forgiven = 2x injected, i.e. owed = injected x (N-1) with N=5 actors, zero outstanding. drop park = 0 even in wide mode (the bottleneck's queue is never empty, so its in-guard recv never parks); drop yield is ~1500 events but ~0.1ms per window, confirming slice-expiry yields sample at their own checkpoint. Deficit decomposition needs the parallel box.
297 lines
12 KiB
Rust
297 lines
12 KiB
Rust
//! Causal-profiling demo (RFC 007): a pipeline where conventional profiling
|
||
//! lies and causal profiling doesn't.
|
||
//!
|
||
//! producer --(serialize ~200µs/item)--> reserve --(~400µs/item)--> notify
|
||
//! background: an actor burning CPU constantly, fully off the critical path
|
||
//!
|
||
//! `reserve` is the true bottleneck. `serialize` is hot but overlapped with
|
||
//! `reserve`'s backlog, and `background` is the hottest code in the process
|
||
//! while contributing nothing to throughput. A cycle profiler ranks them
|
||
//! background > reserve ≈ 2×serialize; the causal report instead shows
|
||
//! throughput responding to virtual speedups of `reserve` and (near-)ignoring
|
||
//! `serialize` and `background`.
|
||
//!
|
||
//! Stage cost is fixed *work* (a calibrated arithmetic loop), not fixed wall
|
||
//! time. This matters: a timed busy-wait absorbs injected causal delay into
|
||
//! its own budget and finishes on schedule regardless, making every
|
||
//! experiment read as a no-op (found live on a 24-core run: dead-flat
|
||
//! deltas). Real workloads are work-shaped, so the demo must be too.
|
||
//!
|
||
//! Run:
|
||
//! cargo run --release --example causal_pipeline --features smarm-causal
|
||
//!
|
||
//! Modes (`SMARM_CAUSAL_MODE`), for probing what the guard placement leaves
|
||
//! out of the measurement (a site speeds up only what it wraps; `recv`/`send`
|
||
//! on the serialized stage sit outside the canonical guard):
|
||
//! work (default) — guard wraps only the 400µs of work.
|
||
//! wide — guard widened over recv + work + send, the whole
|
||
//! serialized per-item path.
|
||
//! occupancy — no experiments; times each segment of reserve's loop
|
||
//! at baseline and reports the unguarded per-item
|
||
//! overhead δ plus the impact ceiling it implies.
|
||
//!
|
||
//! Result (24-core run, 2026-07-13, job f9305cbb): δ measured 0.3µs/item —
|
||
//! 0.1% of the serialized path — and `wide` does not move the @50% cell
|
||
//! (+83.5/+86.3 vs work's +81.5/+86.7). This demo's +84-vs-+100 @50%
|
||
//! shortfall is therefore NOT unguarded stage time; it is controller-side:
|
||
//! injected delay reaches ~327ms of the ideal 350ms over the 700ms window,
|
||
//! plus a ~3% real-throughput dip while experiments run. Contrast urus's
|
||
//! causal_bench, where the same arithmetic identified a real ~70µs/request
|
||
//! unguarded remainder (recv/reply outside the store guard). Sites measure
|
||
//! what they wrap — and the occupancy probe tells you which case you're in.
|
||
//!
|
||
//! Prints a summary, writes `profile.coz` (Coz plot-compatible), and — given
|
||
//! enough cores for the pipeline to actually run in parallel — checks the
|
||
//! expected separation and exits nonzero if it doesn't hold, so a CI box can
|
||
//! run this as a smoke test.
|
||
|
||
use std::sync::atomic::{AtomicBool, Ordering};
|
||
use std::sync::Arc;
|
||
use std::time::{Duration, Instant};
|
||
|
||
/// LCG-mix `iters` times in dependent sequence (unvectorizable, un-elidable),
|
||
/// staying preemptible — and causal-sampleable/delayable — via `check!()`.
|
||
fn work_iters(iters: u64) {
|
||
let mut acc = 0x2545_f491_4f6c_dd1du64;
|
||
let mut i = 0u64;
|
||
while i < iters {
|
||
let chunk_end = (i + 256).min(iters);
|
||
while i < chunk_end {
|
||
acc = acc.wrapping_mul(6364136223846793005).wrapping_add(i);
|
||
i += 1;
|
||
}
|
||
std::hint::black_box(acc);
|
||
smarm::check!();
|
||
}
|
||
}
|
||
|
||
/// Measure how many `work_iters` iterations fit in a microsecond on this
|
||
/// machine, so stage costs below are meaningful in time while staying
|
||
/// work-shaped.
|
||
fn calibrate_iters_per_us() -> u64 {
|
||
let n = 8_000_000u64;
|
||
let t = Instant::now();
|
||
work_iters(n);
|
||
(n / (t.elapsed().as_micros().max(1) as u64)).max(1)
|
||
}
|
||
|
||
/// Guard placement for the `reserve` stage — see module doc.
|
||
#[derive(Clone, Copy, PartialEq)]
|
||
enum Mode {
|
||
Work,
|
||
Wide,
|
||
Occupancy,
|
||
}
|
||
|
||
fn main() {
|
||
let mode = match std::env::var("SMARM_CAUSAL_MODE").as_deref() {
|
||
Err(_) | Ok("") | Ok("work") => Mode::Work,
|
||
Ok("wide") => Mode::Wide,
|
||
Ok("occupancy") => Mode::Occupancy,
|
||
Ok(other) => {
|
||
eprintln!("unknown SMARM_CAUSAL_MODE {other:?} (work|wide|occupancy)");
|
||
std::process::exit(2);
|
||
}
|
||
};
|
||
println!(
|
||
"mode: {}",
|
||
match mode {
|
||
Mode::Work => "work",
|
||
Mode::Wide => "wide",
|
||
Mode::Occupancy => "occupancy",
|
||
}
|
||
);
|
||
let per_us = calibrate_iters_per_us();
|
||
println!("calibration: {per_us} work iters/µs");
|
||
let work_us = move |us: u64| work_iters(us * per_us);
|
||
|
||
let mut failures: Vec<String> = Vec::new();
|
||
|
||
smarm::init(smarm::Config::default()).run(move || {
|
||
let stop = Arc::new(AtomicBool::new(false));
|
||
|
||
let (tx_ab, rx_ab) = smarm::channel::<u64>();
|
||
let (tx_bc, rx_bc) = smarm::channel::<u64>();
|
||
|
||
// Producer: hot serialization, but upstream of the bottleneck.
|
||
let stop_p = stop.clone();
|
||
let producer = smarm::spawn(move || {
|
||
let mut i = 0u64;
|
||
while !stop_p.load(Ordering::Relaxed) {
|
||
{
|
||
let _g = smarm::causal_site!("serialize");
|
||
work_us(200);
|
||
}
|
||
if tx_ab.send(i).is_err() {
|
||
break;
|
||
}
|
||
i += 1;
|
||
}
|
||
// tx_ab drops here; downstream drains and exits.
|
||
});
|
||
|
||
// Reserve: the true bottleneck (~400µs of work per item).
|
||
let reserve = smarm::spawn(move || match mode {
|
||
Mode::Work => {
|
||
while let Ok(item) = rx_ab.recv() {
|
||
{
|
||
let _g = smarm::causal_site!("reserve");
|
||
work_us(400);
|
||
}
|
||
if tx_bc.send(item).is_err() {
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
// Whole serialized per-item path under the guard: a virtual
|
||
// speedup now also compresses recv/send, so the @50% cell should
|
||
// recover the theoretical 2× that `work` mode's placement caps.
|
||
Mode::Wide => loop {
|
||
let _g = smarm::causal_site!("reserve");
|
||
let Ok(item) = rx_ab.recv() else { break };
|
||
work_us(400);
|
||
if tx_bc.send(item).is_err() {
|
||
break;
|
||
}
|
||
},
|
||
// Time each segment at baseline; the recv+send remainder δ is
|
||
// the serialized time a `work`-placed guard cannot speed up.
|
||
Mode::Occupancy => {
|
||
let (mut recv_ns, mut work_ns, mut send_ns, mut n) = (0u64, 0u64, 0u64, 0u64);
|
||
loop {
|
||
let t0 = Instant::now();
|
||
let Ok(item) = rx_ab.recv() else { break };
|
||
let t1 = Instant::now();
|
||
{
|
||
let _g = smarm::causal_site!("reserve");
|
||
work_us(400);
|
||
}
|
||
let t2 = Instant::now();
|
||
if tx_bc.send(item).is_err() {
|
||
break;
|
||
}
|
||
recv_ns += (t1 - t0).as_nanos() as u64;
|
||
work_ns += (t2 - t1).as_nanos() as u64;
|
||
send_ns += t2.elapsed().as_nanos() as u64;
|
||
n += 1;
|
||
}
|
||
let items = n.max(1) as f64;
|
||
let (r, w, s) = (
|
||
recv_ns as f64 / items / 1e3,
|
||
work_ns as f64 / items / 1e3,
|
||
send_ns as f64 / items / 1e3,
|
||
);
|
||
let delta = r + s;
|
||
let total = w + delta;
|
||
println!("occupancy: {n} items; per item recv {r:.1}µs + work(guarded) {w:.1}µs + send {s:.1}µs");
|
||
println!(
|
||
"occupancy: unguarded δ = {delta:.1}µs/item = {:.1}% of the serialized path",
|
||
100.0 * delta / total
|
||
);
|
||
for pct in [25u32, 50] {
|
||
let f = 1.0 - f64::from(pct) / 100.0;
|
||
println!(
|
||
"occupancy: predicted reserve impact @{pct}% -> {:+.1}% (ceiling if δ were guarded: {:+.1}%)",
|
||
100.0 * (total / (f * w + delta) - 1.0),
|
||
100.0 * (1.0 / f - 1.0)
|
||
);
|
||
}
|
||
}
|
||
});
|
||
|
||
// Notify: light tail stage; marks the unit of useful work.
|
||
let notify = smarm::spawn(move || {
|
||
while rx_bc.recv().is_ok() {
|
||
{
|
||
let _g = smarm::causal_site!("notify");
|
||
work_us(50);
|
||
}
|
||
smarm::progress!("orders-processed");
|
||
}
|
||
});
|
||
|
||
// Background: hottest code in the process, zero throughput relevance.
|
||
let stop_bg = stop.clone();
|
||
let background = smarm::spawn(move || {
|
||
while !stop_bg.load(Ordering::Relaxed) {
|
||
let _g = smarm::causal_site!("background-compaction");
|
||
work_us(500);
|
||
}
|
||
});
|
||
|
||
// Warm up so queues reach steady state before measuring.
|
||
smarm::sleep(Duration::from_millis(300));
|
||
|
||
if mode == Mode::Occupancy {
|
||
// No experiments: hold steady state for a window, then drain and
|
||
// let the reserve actor print its segment report.
|
||
smarm::sleep(Duration::from_millis(1500));
|
||
stop.store(true, Ordering::Relaxed);
|
||
producer.join().unwrap();
|
||
reserve.join().unwrap();
|
||
notify.join().unwrap();
|
||
background.join().unwrap();
|
||
return;
|
||
}
|
||
|
||
let results = smarm::causal::run_experiments(&smarm::causal::ExperimentPlan {
|
||
speedups_pct: vec![0, 25, 50],
|
||
experiment: Duration::from_millis(700),
|
||
cooldown: Duration::from_millis(150),
|
||
});
|
||
|
||
stop.store(true, Ordering::Relaxed);
|
||
producer.join().unwrap();
|
||
reserve.join().unwrap();
|
||
notify.join().unwrap();
|
||
background.join().unwrap();
|
||
|
||
print!("{}", smarm::causal::render_summary(&results));
|
||
// RFC 007 deficit hunt: SMARM_CAUSAL_AUDIT=1 appends the per-cell
|
||
// ledger audit (injected/absorbed/forgiven + drop and discard
|
||
// buckets) without touching the pinned summary format.
|
||
if std::env::var_os("SMARM_CAUSAL_AUDIT").is_some() {
|
||
print!("{}", smarm::causal::render_ledger_audit(&results));
|
||
}
|
||
let coz = smarm::causal::render_coz(&results);
|
||
match std::fs::write("profile.coz", coz) {
|
||
Ok(()) => println!("\nwrote profile.coz"),
|
||
Err(e) => eprintln!("\nfailed to write profile.coz: {e}"),
|
||
}
|
||
|
||
// Verdict. The separation only exists when the four pipeline actors
|
||
// actually run in parallel; on a small box, report and skip.
|
||
let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
|
||
if cores < 4 {
|
||
println!("verdict: SKIPPED ({cores} cores; separation needs the stages in parallel)");
|
||
return;
|
||
}
|
||
let impact = |site: &str| {
|
||
smarm::causal::impact_pct(&results, site, 25, "orders-processed")
|
||
};
|
||
let mut expect = |site: &str, ok: &dyn Fn(f64) -> bool, want: &str| match impact(site) {
|
||
Some(p) => {
|
||
let verdict = if ok(p) { "ok" } else { "FAIL" };
|
||
println!("verdict: {site} @25% -> {p:+.1}% (want {want}) {verdict}");
|
||
if !ok(p) {
|
||
failures.push(format!("{site}: {p:+.1}% (want {want})"));
|
||
}
|
||
}
|
||
None => {
|
||
println!("verdict: {site} @25% -> missing cell FAIL");
|
||
failures.push(format!("{site}: missing cell"));
|
||
}
|
||
};
|
||
expect("reserve", &|p| p > 15.0, "> +15%");
|
||
expect("serialize", &|p| p < 10.0, "< +10%");
|
||
expect("background-compaction", &|p| p < 10.0, "< +10%");
|
||
|
||
if failures.is_empty() {
|
||
println!("verdict: PASS — causal separation holds");
|
||
} else {
|
||
println!("verdict: FAIL — {}", failures.join("; "));
|
||
std::process::exit(1);
|
||
}
|
||
});
|
||
}
|