feat(causal): pipeline demo modes for the reserve-shortfall experiment

SMARM_CAUSAL_MODE selects the reserve stage's guard placement:
work (default, unchanged) | wide (guard over recv+work+send, the whole
serialized per-item path) | occupancy (no experiments; per-segment
timing of reserve's loop at baseline, reporting the unguarded
remainder δ and the impact ceiling it implies).

Discriminates the two candidate explanations for the demo's +84-vs-+100
@50% shortfall: physical recv/send time outside the guard (occupancy
sees δ≈30µs, wide recovers ~2x) vs. injection-side credit loss
(occupancy sees δ≈0, wide caps at ~+84 too — guard cadence identical).
This commit is contained in:
Claude (sandbox)
2026-07-13 09:40:27 +00:00
parent efbc254634
commit d5b6a8f66f
+104 -1
View File
@@ -20,6 +20,16 @@
//! 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.
//!
//! 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
@@ -55,7 +65,32 @@ fn calibrate_iters_per_us() -> u64 {
(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);
@@ -86,7 +121,8 @@ fn main() {
});
// Reserve: the true bottleneck (~400µs of work per item).
let reserve = smarm::spawn(move || {
let reserve = smarm::spawn(move || match mode {
Mode::Work => {
while let Ok(item) = rx_ab.recv() {
{
let _g = smarm::causal_site!("reserve");
@@ -96,6 +132,61 @@ fn main() {
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.
@@ -121,6 +212,18 @@ fn main() {
// 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),