diff --git a/Cargo.toml b/Cargo.toml index 335bf38..a8b9cd5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -97,3 +97,11 @@ required-features = ["observer"] [[example]] name = "causal_pipeline" required-features = ["smarm-causal"] + +[[example]] +name = "causal_attrib_probe" +required-features = ["smarm-causal"] + +[[example]] +name = "causal_probe" +required-features = ["smarm-causal"] diff --git a/examples/causal_attrib_probe.rs b/examples/causal_attrib_probe.rs new file mode 100644 index 0000000..7f777f4 --- /dev/null +++ b/examples/causal_attrib_probe.rs @@ -0,0 +1,169 @@ +//! Attribution-efficiency probe (RFC 007 follow-up). +//! +//! Hypothesis: the ~4pt impact shortfall on the 24-core validation +//! (+29.3/+83.5 vs theoretical +33/+100) is a constant attribution +//! efficiency eff ≈ 0.91 — only ~91% of true in-target-site time is ever +//! sampled, because samples are taken only when `maybe_preempt`'s cold +//! block fires while in-site: the tail between the last in-site check and +//! `SiteGuard` drop (and any in-site time truncated by a deschedule) is +//! discarded. +//! +//! Measurement: same pipeline as `causal_pipeline`, but the `reserve` +//! actor also measures its raw in-site time directly (rdtsc at guard +//! enter/exit) and counts site entries. For each experiment window at +//! pct%: +//! +//! eff = (Δglobal_delay / (pct/100)) / Δin_site_cycles +//! +//! and the missing time per site entry localizes the leak: +//! +//! tail_us/entry = (Δin_site − Δglobal_delay/(pct/100)) / Δentries +//! +//! A constant eff across 25/50% with tail/entry in the tens of µs (about +//! half the causal-check cadence) confirms site-exit tail truncation. +//! +//! Run: cargo run --release --example causal_attrib_probe --features smarm-causal + +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +fn rdtsc() -> u64 { + // x86_64 only — same clock the ledger uses. + unsafe { core::arch::x86_64::_rdtsc() } +} + +/// Same fixed-work loop as causal_pipeline (dependent LCG, preemptible). +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!(); + } +} + +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) +} + +static IN_SITE_CYCLES: AtomicU64 = AtomicU64::new(0); +static SITE_ENTRIES: AtomicU64 = AtomicU64::new(0); + +fn main() { + 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 cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); + println!("cores: {cores}"); + if cores < 4 { + println!("probe: SKIPPED (needs the stages in parallel)"); + return; + } + + smarm::init(smarm::Config::default()).run(move || { + let stop = Arc::new(AtomicBool::new(false)); + let (tx_ab, rx_ab) = smarm::channel::(); + let (tx_bc, rx_bc) = smarm::channel::(); + + 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; + } + }); + + // Reserve: the target — instrumented with ground-truth in-site time. + let reserve = smarm::spawn(move || { + while let Ok(item) = rx_ab.recv() { + { + let t0 = rdtsc(); + let _g = smarm::causal_site!("reserve"); + work_us(400); + // Measured before guard drop: exactly the span the + // ledger should be attributing. + IN_SITE_CYCLES.fetch_add(rdtsc().saturating_sub(t0), Ordering::Relaxed); + SITE_ENTRIES.fetch_add(1, Ordering::Relaxed); + } + if tx_bc.send(item).is_err() { + break; + } + } + }); + + let notify = smarm::spawn(move || { + while rx_bc.recv().is_ok() { + { + let _g = smarm::causal_site!("notify"); + work_us(50); + } + smarm::progress!("orders-processed"); + } + }); + + 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); + } + }); + + smarm::sleep(Duration::from_millis(300)); + + let hz = smarm::causal::tsc_hz(); + println!("tsc_hz: {:.3} GHz", hz / 1e9); + + // Manual windows so ledger/ground-truth snapshots align exactly. + for &pct in &[25u32, 50, 50, 25] { + let g0 = smarm::causal::global_delay_cycles(); + let s0 = IN_SITE_CYCLES.load(Ordering::Relaxed); + let e0 = SITE_ENTRIES.load(Ordering::Relaxed); + smarm::causal::begin_experiment_for_test("reserve", pct); + smarm::sleep(Duration::from_millis(1000)); + smarm::causal::end_experiment_for_test(); + let injected = smarm::causal::global_delay_cycles() - g0; + let in_site = IN_SITE_CYCLES.load(Ordering::Relaxed) - s0; + let entries = SITE_ENTRIES.load(Ordering::Relaxed) - e0; + + let attributed = injected as f64 / (pct as f64 / 100.0); + let eff = attributed / in_site as f64; + let missing = in_site as f64 - attributed; + let tail_us = if entries > 0 { + missing / entries as f64 / hz * 1e6 + } else { + f64::NAN + }; + println!( + "pct {pct:>2}% in_site {:>8.1}ms attributed {:>8.1}ms eff {eff:.3} entries {entries} missing/entry {tail_us:.1}µs", + in_site as f64 / hz * 1e3, + attributed / hz * 1e3, + ); + smarm::sleep(Duration::from_millis(150)); + } + + stop.store(true, Ordering::Relaxed); + producer.join().unwrap(); + reserve.join().unwrap(); + notify.join().unwrap(); + background.join().unwrap(); + println!("probe: DONE"); + }); +} diff --git a/src/causal.rs b/src/causal.rs index 1cc8cb7..96a23cf 100644 --- a/src/causal.rs +++ b/src/causal.rs @@ -145,6 +145,7 @@ mod inner { // field docs for the lifetime argument. let prev = unsafe { (*slot).causal_site() }; unsafe { (*slot).set_causal_site(site) }; + site_transition(slot, prev, site); SiteGuard { slot, prev } } } @@ -152,9 +153,11 @@ mod inner { impl Drop for SiteGuard { fn drop(&mut self) { if !self.slot.is_null() { - // SAFETY: as in `enter` — the actor (and thus its slot) is - // alive for as long as this guard is on its stack. + // SAFETY (both): as in `enter` — the actor (and thus its + // slot) is alive for as long as this guard is on its stack. + let site = unsafe { (*self.slot).causal_site() }; unsafe { (*self.slot).set_causal_site(self.prev) }; + site_transition(self.slot, site, self.prev); } } } @@ -255,14 +258,9 @@ mod inner { if last == 0 { return; // unarmed clock: no interval to attribute } - let interval = now.saturating_sub(last); - if interval == 0 || interval > MAX_SAMPLE_CYCLES { - return; // clock hiccup: discard the sample, not the run - } - let delta = interval.saturating_mul(pct) / 100; - GLOBAL_DELAY.fetch_add(delta, Ordering::Relaxed); - let mine = unsafe { (*slot).causal_delay() }; - unsafe { (*slot).set_causal_delay(mine.wrapping_add(delta)) }; + // SAFETY: `slot` is the on-CPU actor's slot (checked non-null + // above); see `check_cancelled` for the lifetime argument. + unsafe { attribute(slot, now.saturating_sub(last), pct) }; } else { // Not the winner: chase the global ledger by spinning off the // difference, then push the slice start forward so injected @@ -285,6 +283,60 @@ mod inner { } } + /// Attribute one target-site sample of `interval` cycles at `pct`%: + /// grow the global ledger and credit the sampling actor's own ledger by + /// the same amount — the credited gap *is* the virtual speedup. Shared + /// by the cold check and the guard-boundary flush. Applies the same + /// clamps as sampling always has: zero intervals and clock hiccups are + /// discarded, not the run. + /// + /// SAFETY: `slot` must point at the on-CPU actor's slot (the + /// `check_cancelled` lifetime argument). + unsafe fn attribute(slot: *const crate::runtime::Slot, interval: u64, pct: u64) { + if interval == 0 || interval > MAX_SAMPLE_CYCLES { + return; + } + let delta = interval.saturating_mul(pct) / 100; + GLOBAL_DELAY.fetch_add(delta, Ordering::Relaxed); + let mine = (*slot).causal_delay(); + (*slot).set_causal_delay(mine.wrapping_add(delta)); + } + + /// Site-boundary hook, called by `SiteGuard` enter/drop when the + /// actor's current site changes from `old` to `new`. Sample-only — + /// never spins — so it is safe anywhere, including no-preempt regions + /// where `check()` cannot run. + /// + /// - Leaving the experiment's target site: flush the pending interval. + /// Cold checks only sample when they happen to fire in-site, so the + /// tail between the last check and the guard drop was otherwise + /// discarded on every site entry — measured live at ~22-29µs/entry, + /// ~6-7% of all target time (eff 0.93), which under-reported every + /// impact (+83.5% where theory says +100%). + /// - Entering the target site: re-arm the sample clock, so time spent + /// *before* the site can never be attributed to it by the first + /// in-site check (the symmetric over-attribution). + #[inline] + fn site_transition(slot: *const crate::runtime::Slot, old: u32, new: u32) { + let exp = EXPERIMENT.load(Ordering::Relaxed); + if exp == 0 || old == new { + return; + } + let target = (exp >> 32) as u32; + let pct = exp & 0xffff_ffff; + if old == target && new != target { + let now = preempt::rdtsc(); + let last = LAST_SAMPLE_TSC.with(|c| c.replace(now)); + if last != 0 && pct > 0 { + // SAFETY: forwarded from the guard, which holds the on-CPU + // actor's slot for its whole life (see `SiteGuard::slot`). + unsafe { attribute(slot, now.saturating_sub(last), pct) }; + } + } else if new == target && old != target { + LAST_SAMPLE_TSC.with(|c| c.set(preempt::rdtsc())); + } + } + /// Resume-path hook (scheduler thread, actor off-CPU). Two duties: /// /// - If the last deschedule was a *real park*, time blocked absorbs any diff --git a/tests/causal.rs b/tests/causal.rs index d8ad4d2..85d07c7 100644 --- a/tests/causal.rs +++ b/tests/causal.rs @@ -332,3 +332,59 @@ fn controller_produces_report() { assert!(coz.contains("throughput-point\tname=items"), "coz: {coz}"); }); } + +/// A target-site entry that never hits a cold causal check must still be +/// attributed: the guard boundaries flush the pending interval on exit and +/// re-arm the clock on entry. Without the exit flush, in-site time between +/// the last check and guard drop is silently discarded — measured live at +/// ~6-7% of all target time (eff 0.93), under-reporting every impact +/// (+83.5% where theory says +100%). +#[test] +fn site_boundaries_flush_tail() { + fn tsc() -> u64 { + // SAFETY: rdtsc has no preconditions on x86_64. + unsafe { core::arch::x86_64::_rdtsc() } + } + + let injected = Arc::new(AtomicU64::new(0)); + let spent = Arc::new(AtomicU64::new(0)); + let inj_out = injected.clone(); + let spent_out = spent.clone(); + + smarm::init(smarm::Config::exact(1)).run(move || { + // Experiment is live before the actor ever enters the site. + smarm::causal::begin_experiment_for_test("tail-site", 50); + let base = smarm::causal::global_delay_cycles(); + + let a = smarm::spawn(move || { + let t0 = tsc(); + { + let _g = smarm::causal_site!("tail-site"); + // Burn ~20M cycles with no allocations and no `check!()`: + // the amortised cold check can never fire, so only the + // guard-boundary flush can attribute this time. + let start = tsc(); + while tsc().wrapping_sub(start) < 20_000_000 { + core::hint::spin_loop(); + } + } + spent_out.store(tsc().wrapping_sub(t0), Ordering::Relaxed); + }); + a.join().unwrap(); + + smarm::causal::end_experiment_for_test(); + inj_out.store( + smarm::causal::global_delay_cycles() - base, + Ordering::Relaxed, + ); + }); + + let injected = injected.load(Ordering::Relaxed); + let spent = spent.load(Ordering::Relaxed); + // At 50% the flush should attribute ≈ half the in-guard time; allow a + // generous band for guard overhead and clock skew. + assert!( + injected >= spent * 40 / 100 && injected <= spent * 60 / 100, + "boundary flush attributed {injected} of {spent} spent cycles (want ~50%)" + ); +}