diff --git a/examples/causal_pipeline.rs b/examples/causal_pipeline.rs index 5109074..b06db8b 100644 --- a/examples/causal_pipeline.rs +++ b/examples/causal_pipeline.rs @@ -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,15 +121,71 @@ fn main() { }); // Reserve: the true bottleneck (~400µs of work per item). - let reserve = smarm::spawn(move || { - while let Ok(item) = rx_ab.recv() { - { - let _g = smarm::causal_site!("reserve"); - work_us(400); + 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) + ); + } } }); @@ -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),