diff --git a/Cargo.toml b/Cargo.toml index f24a59f..335bf38 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,10 @@ expect_used = "deny" [features] default = ["rq-mutex"] smarm-trace = [] +# RFC 007: native causal profiling. Zero cost when off (cf. smarm-trace): the +# hook in `maybe_preempt` and the resume-path fast-forward compile away; the +# two Slot ledger fields exist regardless and stay 0 (budget_cycles precedent). +smarm-causal = [] # RFC 016 Chunk 2: cycle-accurate per-actor time-budget accounting. Off by # default — it costs two extra RDTSC reads per actor resume on the hot path # (D6). The `ActorInfo.budget_cycles` field exists regardless; it just stays 0 @@ -89,3 +93,7 @@ harness = false [[example]] name = "observer" required-features = ["observer"] + +[[example]] +name = "causal_pipeline" +required-features = ["smarm-causal"] diff --git a/examples/causal_pipeline.rs b/examples/causal_pipeline.rs new file mode 100644 index 0000000..5109074 --- /dev/null +++ b/examples/causal_pipeline.rs @@ -0,0 +1,177 @@ +//! 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 +//! +//! 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) +} + +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 mut failures: Vec = Vec::new(); + + 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::(); + + // 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 || { + while let Ok(item) = rx_ab.recv() { + { + let _g = smarm::causal_site!("reserve"); + work_us(400); + } + if tx_bc.send(item).is_err() { + break; + } + } + }); + + // 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)); + + 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)); + 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); + } + }); +} diff --git a/examples/causal_probe.rs b/examples/causal_probe.rs new file mode 100644 index 0000000..441cdbd --- /dev/null +++ b/examples/causal_probe.rs @@ -0,0 +1,145 @@ +//! Diagnostic probe for RFC 007 on a target box. Measures, in order: +//! 1. TSC frequency against `Instant` (the crate assumes 3 GHz). +//! 2. TSC sanity under actor migration: distribution of wall time actually +//! spent in `burn_us(400)` across many runs — a bimodal/short tail means +//! cross-core TSC offsets are cutting burns short. +//! 3. Pipeline stage rates with no experiment running (who is the real +//! bottleneck?). +//! 4. The same rates during a 50% experiment on `background-compaction` +//! (a correct implementation must slow every stage; an off-critical-path +//! target must reduce end-to-end throughput proportionally). +//! +//! Run: cargo run --release --example causal_probe --features smarm-causal + +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +fn burn_us(us: u64) { + let cycles = us * 3_000; + let start = smarm::preempt::rdtsc(); + while smarm::preempt::rdtsc().saturating_sub(start) < cycles { + smarm::check!(); + } +} + +fn main() { + // 1. TSC calibration (plain OS thread, before the runtime starts). + let c0 = smarm::preempt::rdtsc(); + let t0 = Instant::now(); + std::thread::sleep(Duration::from_millis(200)); + let hz = (smarm::preempt::rdtsc() - c0) as f64 / t0.elapsed().as_secs_f64(); + println!("tsc_hz: {:.3e} (crate assumes 3.0e9)", hz); + + smarm::init(smarm::Config::default()).run(move || { + // 2. burn_us(400) wall-time distribution inside a migrating actor. + let h = smarm::spawn(|| { + let mut samples: Vec = (0..500) + .map(|_| { + let t = Instant::now(); + burn_us(400); + t.elapsed().as_micros() as u64 + }) + .collect(); + samples.sort_unstable(); + println!( + "burn_us(400) wall us: min {} p10 {} p50 {} p90 {} max {}", + samples[0], samples[50], samples[250], samples[450], samples[499] + ); + }); + h.join().unwrap(); + + // 3+4. Pipeline with per-stage counters. + let stop = Arc::new(AtomicBool::new(false)); + let produced = Arc::new(AtomicU64::new(0)); + let reserved = Arc::new(AtomicU64::new(0)); + let notified = Arc::new(AtomicU64::new(0)); + + let (tx_ab, rx_ab) = smarm::channel::(); + let (tx_bc, rx_bc) = smarm::channel::(); + + let stop_p = stop.clone(); + let produced2 = produced.clone(); + let producer = smarm::spawn(move || { + let mut i = 0u64; + while !stop_p.load(Ordering::Relaxed) { + { + let _g = smarm::causal_site!("serialize"); + burn_us(200); + } + if tx_ab.send(i).is_err() { + break; + } + produced2.fetch_add(1, Ordering::Relaxed); + i += 1; + } + }); + + let reserved2 = reserved.clone(); + let reserve = smarm::spawn(move || { + while let Ok(item) = rx_ab.recv() { + { + let _g = smarm::causal_site!("reserve"); + burn_us(400); + } + reserved2.fetch_add(1, Ordering::Relaxed); + if tx_bc.send(item).is_err() { + break; + } + } + }); + + let notified2 = notified.clone(); + let notify = smarm::spawn(move || { + while rx_bc.recv().is_ok() { + { + let _g = smarm::causal_site!("notify"); + burn_us(50); + } + notified2.fetch_add(1, Ordering::Relaxed); + 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"); + burn_us(500); + } + }); + + smarm::sleep(Duration::from_millis(300)); + + let window = |label: &str| { + let (p0, r0, n0) = ( + produced.load(Ordering::Relaxed), + reserved.load(Ordering::Relaxed), + notified.load(Ordering::Relaxed), + ); + let d0 = smarm::causal::global_delay_cycles(); + let t = Instant::now(); + smarm::sleep(Duration::from_millis(700)); + let secs = t.elapsed().as_secs_f64(); + println!( + "{label}: produced {:.0}/s reserved {:.0}/s notified {:.0}/s injected {:.0}ms(assumed-3GHz)", + (produced.load(Ordering::Relaxed) - p0) as f64 / secs, + (reserved.load(Ordering::Relaxed) - r0) as f64 / secs, + (notified.load(Ordering::Relaxed) - n0) as f64 / secs, + (smarm::causal::global_delay_cycles() - d0) as f64 / 3.0e9 * 1e3, + ); + }; + + window("no-experiment "); + smarm::causal::begin_experiment_for_test("background-compaction", 50); + window("bg-comp @ 50% "); + smarm::causal::end_experiment_for_test(); + window("post-experiment"); + + stop.store(true, Ordering::Relaxed); + producer.join().unwrap(); + reserve.join().unwrap(); + notify.join().unwrap(); + background.join().unwrap(); + }); +} diff --git a/src/causal.rs b/src/causal.rs new file mode 100644 index 0000000..1cc8cb7 --- /dev/null +++ b/src/causal.rs @@ -0,0 +1,607 @@ +//! Native causal profiling (RFC 007). Enabled by `--features smarm-causal`; +//! zero cost without it (same discipline as `smarm-trace`). +//! +//! The Coz algorithm, transposed onto actors: to estimate what speeding up +//! code site S by p% would do to throughput, we instead *slow everything +//! else down* by p% of the time spent in S, and watch the progress-point +//! rates respond. Where Coz must inject real `usleep`s into OS threads from +//! the outside, smarm owns every clock that matters: +//! +//! - Sampling and delay injection happen at `maybe_preempt`'s amortised +//! cadence — an existing, safe hook (never inside a prep-to-park region). +//! - Injected delay is subtracted from the actor's timeslice +//! (`preempt::extend_timeslice`), so experiments don't perturb scheduling. +//! - Delay bookkeeping is *actor*-granular: each `Slot` carries an absorbed- +//! delay ledger, compared against a global ledger. Parked actors absorb +//! accrued delay for free on resume (Coz's blocked-thread rule) — waiting +//! is never penalised. +//! +//! v1 scope (per RFC discussion): explicit scoped sites (`causal_site!`) +//! rather than PC sampling (jar Q1 stays open); throughput progress points +//! only; timer-heap deadlines are *not* shifted (documented gap — long +//! experiments can make real-time timeouts fire early in virtual terms); +//! multi-scheduler coherence is best-effort via global atomics. +//! +//! Usage: +//! ```ignore +//! let _g = smarm::causal_site!("inventory-reserve"); // in suspect code +//! smarm::progress!("orders-processed"); // per unit of work +//! let results = smarm::causal::run_experiments(&Default::default()); +//! print!("{}", smarm::causal::render_summary(&results)); +//! std::fs::write("profile.coz", smarm::causal::render_coz(&results))?; +//! ``` + +#[cfg(feature = "smarm-causal")] +mod inner { + use crate::preempt; + use std::cell::Cell; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::{Mutex, OnceLock}; + use std::time::{Duration, Instant}; + + // ----------------------------------------------------------------------- + // Global state + // ----------------------------------------------------------------------- + + /// Active experiment, packed `(site_id << 32) | speedup_pct`. 0 = idle. + /// A single word so the hot path reads one atomic; experiments are global + /// across scheduler threads (jar Q7, v1: plain Relaxed atomics). + static EXPERIMENT: AtomicU64 = AtomicU64::new(0); + + /// Global virtual-delay ledger, in TSC cycles: the total delay every + /// actor *should* have experienced since startup. Grows while a sample + /// lands in the experiment's target site; each actor's `Slot` ledger + /// chases it by spin-absorbing at preemption checks. + static GLOBAL_DELAY: AtomicU64 = AtomicU64::new(0); + + /// Registered site names; site id = index + 1 (0 = "no site"). + static SITES: OnceLock>> = OnceLock::new(); + + /// Registered progress points (leaked for `'static`, like trace's drain + /// state — the set is small and lives for the process). + static PROGRESS: OnceLock>> = OnceLock::new(); + + thread_local! { + /// TSC at this thread's previous causal check, the sample "period" + /// denominator. Re-armed on every actor resume so scheduler time and + /// a previous actor's tail never count toward a sample. 0 = unarmed. + static LAST_SAMPLE_TSC: Cell = const { Cell::new(0) }; + } + + /// Guard against TSC weirdness (migration between unsynced sockets, + /// virtualisation steps): a single sample interval larger than this is + /// discarded rather than believed. ~33ms at 3 GHz — far beyond any real + /// gap between preemption checks inside a slice. + const MAX_SAMPLE_CYCLES: u64 = 100_000_000; + + /// Cap on delay spun in one visit, so one check can never wedge an actor + /// for a human-visible pause; the remainder is absorbed on later visits. + /// ~3ms at 3 GHz. + const MAX_SPIN_PER_VISIT: u64 = 10_000_000; + + fn sites() -> &'static Mutex> { + SITES.get_or_init(|| Mutex::new(Vec::new())) + } + + fn progress_points() -> &'static Mutex> { + PROGRESS.get_or_init(|| Mutex::new(Vec::new())) + } + + /// Recover from lock poisoning: all these registries hold plain data that + /// is valid at every instruction boundary, so a panicked registrant can't + /// leave them torn. + fn lock_unpoisoned(m: &Mutex) -> std::sync::MutexGuard<'_, T> { + match m.lock() { + Ok(g) => g, + Err(poisoned) => poisoned.into_inner(), + } + } + + // ----------------------------------------------------------------------- + // Sites + // ----------------------------------------------------------------------- + + /// Register (or look up) a causal site by name; returns its nonzero id. + /// Called once per `causal_site!` expansion via a `OnceLock`, so the + /// mutex is off every hot path. + pub fn site_id(name: &'static str) -> u32 { + let mut v = lock_unpoisoned(sites()); + if let Some(pos) = v.iter().position(|n| *n == name) { + return (pos + 1) as u32; + } + v.push(name); + v.len() as u32 + } + + fn site_name(id: u32) -> Option { + if id == 0 { + return None; + } + let v = lock_unpoisoned(sites()); + v.get((id - 1) as usize).map(|s| (*s).to_string()) + } + + /// RAII marker: while alive, the current *actor* (not thread — the id + /// lives in its `Slot` and survives preemption/migration) is "inside" + /// the site. Nesting restores the outer site on drop. Inert outside an + /// actor (scheduler/OS-thread stacks). + pub struct SiteGuard { + /// Slot of the actor that entered, null if entered outside an actor. + /// Valid for the guard's whole life: the guard lives on the actor's + /// stack, and a slot is never reclaimed while its actor is alive — + /// the same argument as `preempt::check_cancelled`. + slot: *const crate::runtime::Slot, + prev: u32, + } + + impl SiteGuard { + /// Enter `site` for the on-CPU actor. + pub fn enter(site: u32) -> Self { + let slot = preempt::current_slot_ptr(); + if slot.is_null() { + return SiteGuard { slot, prev: 0 }; + } + // SAFETY: non-null ⇒ points at the on-CPU actor's slot; see the + // field docs for the lifetime argument. + let prev = unsafe { (*slot).causal_site() }; + unsafe { (*slot).set_causal_site(site) }; + SiteGuard { slot, prev } + } + } + + 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. + unsafe { (*self.slot).set_causal_site(self.prev) }; + } + } + } + + /// Name of the site the on-CPU actor is currently inside, if any. + /// (Introspection/testing; not a hot path.) + pub fn current_site_name() -> Option { + let slot = preempt::current_slot_ptr(); + if slot.is_null() { + return None; + } + // SAFETY: on-CPU actor's slot, valid for the whole resume. + site_name(unsafe { (*slot).causal_site() }) + } + + // ----------------------------------------------------------------------- + // Progress points + // ----------------------------------------------------------------------- + + /// A named throughput counter. One per distinct name; `progress!` call + /// sites sharing a name share the counter. + pub struct ProgressPoint { + name: &'static str, + count: AtomicU64, + } + + impl ProgressPoint { + /// The hot path: one Relaxed RMW. (Contended across actors by design + /// — a progress point is a global rate meter.) + #[inline] + pub fn bump(&self) { + self.count.fetch_add(1, Ordering::Relaxed); + } + } + + /// Register (or look up) a progress point. Called once per `progress!` + /// expansion via a `OnceLock`; the mutex is off the hot path. + pub fn register_progress(name: &'static str) -> &'static ProgressPoint { + let mut v = lock_unpoisoned(progress_points()); + if let Some(p) = v.iter().find(|p| p.name == name) { + return p; + } + let p: &'static ProgressPoint = Box::leak(Box::new(ProgressPoint { + name, + count: AtomicU64::new(0), + })); + v.push(p); + p + } + + /// Snapshot of all progress points as `(name, count)`. + pub fn progress_snapshot() -> Vec<(String, u64)> { + lock_unpoisoned(progress_points()) + .iter() + .map(|p| (p.name.to_string(), p.count.load(Ordering::Relaxed))) + .collect() + } + + // ----------------------------------------------------------------------- + // The hot hook: sample + absorb + // ----------------------------------------------------------------------- + + /// Called from `maybe_preempt` at the amortised timeslice-check cadence, + /// under the `PREEMPTION_ENABLED` gate (so never in a prep-to-park or + /// no-preempt region — spinning here is as safe as yielding is). + /// + /// One Relaxed load and out when no experiment is running. + #[inline] + pub(crate) fn check() { + let exp = EXPERIMENT.load(Ordering::Relaxed); + if exp == 0 { + return; + } + cold_check(exp); + } + + /// The experiment-active path, kept out of the inlined fast path. + #[cold] + fn cold_check(exp: u64) { + let slot = preempt::current_slot_ptr(); + if slot.is_null() { + return; + } + let now = preempt::rdtsc(); + let last = LAST_SAMPLE_TSC.with(|c| c.replace(now)); + let target_site = (exp >> 32) as u32; + let pct = exp & 0xffff_ffff; + + // SAFETY (both derefs below): non-null ⇒ the on-CPU actor's slot, + // never reclaimed while the actor runs — see `check_cancelled`. + let my_site = unsafe { (*slot).causal_site() }; + + if my_site == target_site && pct > 0 { + // A sample landed in the target site: everyone else must fall + // behind by pct% of the sampled interval. Grow the global ledger + // and credit ourselves the same amount — the credited gap *is* + // the virtual speedup. + 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)) }; + } else { + // Not the winner: chase the global ledger by spinning off the + // difference, then push the slice start forward so injected + // delay never counts as compute (the clock correction that Coz + // cannot do from outside). + let global = GLOBAL_DELAY.load(Ordering::Relaxed); + let mine = unsafe { (*slot).causal_delay() }; + if mine >= global { + return; + } + let spin = (global - mine).min(MAX_SPIN_PER_VISIT); + let start = preempt::rdtsc(); + while preempt::rdtsc().saturating_sub(start) < spin { + core::hint::spin_loop(); + } + unsafe { (*slot).set_causal_delay(mine.wrapping_add(spin)) }; + preempt::extend_timeslice(spin); + // The spin is not part of the next sample interval either. + 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 + /// delay accrued meanwhile for free — Coz's blocked-thread rule, which + /// keeps experiments from punishing actors for waiting. An actor that + /// merely yielded (slice expiry) was runnable the whole time and keeps + /// its debt: it must pay by spinning at its next check. Forgiving on + /// every resume would make any yield-cadence actor delay-immune and + /// experiments inert (found live on a 24-core run: nothing slowed). + /// - Arm this thread's sample clock so the first interval of the resume + /// excludes scheduler time. + #[inline] + pub(crate) fn on_resume(slot: &crate::runtime::Slot) { + if slot.take_causal_parked() { + let global = GLOBAL_DELAY.load(Ordering::Relaxed); + if slot.causal_delay() < global { + slot.set_causal_delay(global); + } + } + LAST_SAMPLE_TSC.with(|c| c.set(preempt::rdtsc())); + } + + // ----------------------------------------------------------------------- + // Experiments + // ----------------------------------------------------------------------- + + fn begin(site: u32, pct: u32) { + EXPERIMENT.store(((site as u64) << 32) | pct as u64, Ordering::Relaxed); + } + + fn end() { + EXPERIMENT.store(0, Ordering::Relaxed); + } + + /// Total virtual delay injected so far, in TSC cycles. + pub fn global_delay_cycles() -> u64 { + GLOBAL_DELAY.load(Ordering::Relaxed) + } + + /// Absorbed-delay ledger of the on-CPU actor (testing/introspection). + pub fn my_absorbed_delay_cycles() -> u64 { + let slot = preempt::current_slot_ptr(); + if slot.is_null() { + return 0; + } + // SAFETY: on-CPU actor's slot; see `check_cancelled`. + unsafe { (*slot).causal_delay() } + } + + /// Test support: start an experiment targeting `site_name` at `pct`% + /// virtual speedup. Registers the site if needed. + pub fn begin_experiment_for_test(name: &'static str, pct: u32) { + begin(site_id(name), pct); + } + + /// Test support: stop the running experiment. + pub fn end_experiment_for_test() { + end(); + } + + /// Controller parameters: which speedups to try per site, and the + /// experiment/cooldown windows. + pub struct ExperimentPlan { + pub speedups_pct: Vec, + pub experiment: Duration, + pub cooldown: Duration, + } + + impl Default for ExperimentPlan { + fn default() -> Self { + ExperimentPlan { + speedups_pct: vec![0, 25, 50], + experiment: Duration::from_millis(500), + cooldown: Duration::from_millis(100), + } + } + } + + /// One completed experiment cell. + pub struct ExperimentResult { + pub site: String, + pub speedup_pct: u32, + pub duration: Duration, + /// Progress-point deltas over the window, `(name, count)`. + pub deltas: Vec<(String, u64)>, + /// Virtual delay injected during the window (cycles). + pub injected_cycles: u64, + } + + /// Run the plan synchronously on the calling (OS) thread: for every + /// registered site × speedup, run one experiment window and record + /// progress-point deltas, with a cooldown between cells. Sites and + /// progress points must already be registered (the workload has to be + /// running); the caller owns workload start/stop. + /// + /// v1 controller: exhaustive sweep, fixed windows, no adaptive site + /// selection or confidence stopping (jar Q5). + /// + /// Callable from a plain OS thread *or* from inside an actor: sleeping + /// parks the green thread when we're on one (so no scheduler thread is + /// blocked), and falls back to `thread::sleep` otherwise. + pub fn run_experiments(plan: &ExperimentPlan) -> Vec { + fn controller_sleep(d: Duration) { + if preempt::current_slot_ptr().is_null() { + std::thread::sleep(d); + } else { + crate::scheduler::sleep(d); + } + } + // Calibrate before any window so report rendering never has to sleep. + let _ = tsc_hz(); + let site_list: Vec<(u32, String)> = { + let v = lock_unpoisoned(sites()); + v.iter() + .enumerate() + .map(|(i, n)| ((i + 1) as u32, (*n).to_string())) + .collect() + }; + let mut out = Vec::new(); + for (sid, sname) in &site_list { + for &pct in &plan.speedups_pct { + let before = progress_snapshot(); + let injected_before = global_delay_cycles(); + let t0 = Instant::now(); + begin(*sid, pct); + controller_sleep(plan.experiment); + end(); + let elapsed = t0.elapsed(); + let after = progress_snapshot(); + let deltas = after + .iter() + .map(|(n, c)| { + let b = before + .iter() + .find(|(bn, _)| bn == n) + .map(|(_, bc)| *bc) + .unwrap_or(0); + (n.clone(), c.saturating_sub(b)) + }) + .collect(); + out.push(ExperimentResult { + site: sname.clone(), + speedup_pct: pct, + duration: elapsed, + deltas, + injected_cycles: global_delay_cycles() - injected_before, + }); + controller_sleep(plan.cooldown); + } + } + out + } + + // ----------------------------------------------------------------------- + // Reports + // ----------------------------------------------------------------------- + + /// Measured TSC frequency (Hz), calibrated once. The crate-wide 3 GHz + /// constant is fine for the *relative* timeslice check, but report + /// normalisation divides wall time by injected time, so a 20% Hz error + /// skews every impact number — measured live: a 3.7 GHz box inflated all + /// baselines uniformly. Calibrated against `Instant` over ~50ms on first + /// use; `run_experiments` triggers it before its first window (using the + /// park-aware sleep, so no scheduler thread is blocked when called from + /// an actor). + static TSC_HZ_MEASURED: OnceLock = OnceLock::new(); + + /// Measured TSC frequency in Hz. Calibrates on first call (~50ms). + pub fn tsc_hz() -> f64 { + *TSC_HZ_MEASURED.get_or_init(|| { + let c0 = preempt::rdtsc(); + let t0 = Instant::now(); + let d = Duration::from_millis(50); + if preempt::current_slot_ptr().is_null() { + std::thread::sleep(d); + } else { + crate::scheduler::sleep(d); + } + (preempt::rdtsc().wrapping_sub(c0)) as f64 / t0.elapsed().as_secs_f64() + }) + } + + /// Normalized rate for one cell: count over the *virtual* window + /// (wall − injected) — Coz's normalization: injected delay does not + /// exist in the virtual timeline. A bottleneck site keeps its raw count + /// while shrinking the divisor → positive impact; a fully overlapped + /// site loses count proportionally → ~zero. + fn normalized_rate(r: &ExperimentResult, point: &str) -> Option { + let count = r.deltas.iter().find(|(n, _)| n == point).map(|(_, c)| *c)?; + let injected_secs = r.injected_cycles as f64 / tsc_hz(); + let virtual_secs = (r.duration.as_secs_f64() - injected_secs).max(1e-9); + Some(count as f64 / virtual_secs) + } + + /// Impact of virtually speeding up `site` by `speedup_pct` on progress + /// point `point`, in percent relative to that site's own 0% baseline + /// cell. `None` if either cell or the point is missing, or the baseline + /// rate is zero. This is the machine-readable form of the summary's + /// "vs baseline" column, for programmatic checks (CI, examples). + pub fn impact_pct( + results: &[ExperimentResult], + site: &str, + speedup_pct: u32, + point: &str, + ) -> Option { + let cell = results + .iter() + .find(|r| r.site == site && r.speedup_pct == speedup_pct)?; + let base = results.iter().find(|r| r.site == site && r.speedup_pct == 0)?; + let rate = normalized_rate(cell, point)?; + let b = normalized_rate(base, point)?; + if b <= 0.0 { + return None; + } + Some((rate / b - 1.0) * 100.0) + } + + /// Human-readable summary: per (site, progress point), the throughput at + /// each virtual speedup and the change relative to that site's own 0% + /// baseline. A near-zero column across speedups means: optimising this + /// site buys you nothing — the RFC's headline answer. + pub fn render_summary(results: &[ExperimentResult]) -> String { + use std::fmt::Write; + let mut s = String::new(); + let _ = writeln!(s, "== smarm causal profile =="); + let mut sites_seen: Vec<&str> = Vec::new(); + for r in results { + if !sites_seen.contains(&r.site.as_str()) { + sites_seen.push(&r.site); + } + } + for site in sites_seen { + let _ = writeln!(s, "site {site}"); + for r in results.iter().filter(|r| r.site == site) { + for (name, _) in &r.deltas { + let rate = match normalized_rate(r, name) { + Some(x) => x, + None => continue, + }; + let rel = impact_pct(results, site, r.speedup_pct, name) + .map(|p| format!("{p:+.1}%")) + .unwrap_or_else(|| "n/a".to_string()); + let _ = writeln!( + s, + " speedup {:>3}% {name:<24} {rate:>12.1}/s vs baseline {rel} (injected {:.1}ms)", + r.speedup_pct, + r.injected_cycles as f64 / tsc_hz() * 1e3 + ); + } + } + } + s + } + + /// Coz-compatible profile text (`profile.coz`), so Coz's existing plot + /// tooling renders our experiments — the RFC's "don't build a UI" call. + pub fn render_coz(results: &[ExperimentResult]) -> String { + use std::fmt::Write; + let mut s = String::new(); + let _ = writeln!(s, "startup\ttime=0"); + for r in results { + let _ = writeln!( + s, + "experiment\tselected={}\tspeedup={:.2}\tduration={}\tselected-samples=1", + r.site, + r.speedup_pct as f64 / 100.0, + r.duration.as_nanos() + ); + for (name, count) in &r.deltas { + let _ = writeln!(s, "throughput-point\tname={name}\tdelta={count}"); + } + } + s + } +} + +#[cfg(feature = "smarm-causal")] +pub use inner::*; + +/// Mark one unit of useful work complete at a named throughput progress +/// point (RFC 007). One Relaxed increment when `smarm-causal` is on; nothing +/// at all when it's off. +#[cfg(feature = "smarm-causal")] +#[macro_export] +macro_rules! progress { + ($name:literal) => {{ + static __SMARM_PP: ::std::sync::OnceLock<&'static $crate::causal::ProgressPoint> = + ::std::sync::OnceLock::new(); + __SMARM_PP + .get_or_init(|| $crate::causal::register_progress($name)) + .bump(); + }}; +} + +#[cfg(not(feature = "smarm-causal"))] +#[macro_export] +macro_rules! progress { + ($name:literal) => {{}}; +} + +/// Enter a named causal-profiling site for the current actor; the returned +/// guard exits it (restoring any enclosing site) on drop. Site identity is +/// stored in the actor's slot, so it survives preemption and migration. +/// Expands to a unit no-op without `smarm-causal`. +#[cfg(feature = "smarm-causal")] +#[macro_export] +macro_rules! causal_site { + ($name:literal) => {{ + static __SMARM_SITE: ::std::sync::OnceLock = ::std::sync::OnceLock::new(); + $crate::causal::SiteGuard::enter(*__SMARM_SITE.get_or_init(|| $crate::causal::site_id($name))) + }}; +} + +#[cfg(not(feature = "smarm-causal"))] +#[macro_export] +macro_rules! causal_site { + ($name:literal) => { + () + }; +} diff --git a/src/lib.rs b/src/lib.rs index e3016be..48e9ab9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -38,6 +38,7 @@ pub(crate) mod sync_shim; #[doc(hidden)] // pub only so benches/rq_micro.rs can drive the raw structures pub mod run_queue; pub mod trace; +pub mod causal; // --------------------------------------------------------------------------- // Global allocator diff --git a/src/preempt.rs b/src/preempt.rs index 5b61128..221119a 100644 --- a/src/preempt.rs +++ b/src/preempt.rs @@ -98,6 +98,25 @@ pub(crate) fn clear_current_slot() { CURRENT_SLOT.with(|c| c.set(std::ptr::null())); } +/// RFC 007 (`smarm-causal`) — raw pointer to the on-CPU actor's slot, null on +/// the scheduler's own stack. Same lifetime argument as `note_overrun`: the +/// slot is never reclaimed while its actor is on-CPU. +#[cfg(feature = "smarm-causal")] +#[inline] +pub(crate) fn current_slot_ptr() -> *const crate::runtime::Slot { + CURRENT_SLOT.with(|c| c.get()) +} + +/// RFC 007 (`smarm-causal`) — push the slice start forward by `cycles`, so +/// virtually-injected delay spun inside `maybe_preempt` does not count against +/// the actor's timeslice (the clock-correction half of the RFC: the runtime +/// owns this clock, so it can subtract its own perturbation). +#[cfg(feature = "smarm-causal")] +#[inline] +pub(crate) fn extend_timeslice(cycles: u64) { + TIMESLICE_START.with(|c| c.set(c.get().wrapping_add(cycles))); +} + /// Tally a timeslice overrun against the on-CPU actor (RFC 016 Chunk 2). A /// no-op if no actor is bound (the scheduler's own stack). Reached only from /// the slice-expiry branch, which is already the yield path, so its cost is @@ -247,6 +266,11 @@ pub fn maybe_preempt() { // Observe a pending stop first: if we are being cancelled // there is no point yielding, we unwind instead. check_cancelled(); + // RFC 007: causal-profiling sample/absorb point. Shares the + // amortised cadence, and the PREEMPTION_ENABLED gate — so it + // can never spin inside a prep-to-park or no-preempt region. + #[cfg(feature = "smarm-causal")] + crate::causal::check(); let start = TIMESLICE_START.with(|s| s.get()); if rdtsc().saturating_sub(start) > CONFIGURED_TIMESLICE_CYCLES.with(|t| t.get()) { // Tally the overrun (RFC 016 Chunk 2) before handing back — diff --git a/src/runtime.rs b/src/runtime.rs index a980420..db1162c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -445,6 +445,28 @@ pub(crate) struct Slot { /// and only when the `budget-accounting` feature is on (it costs two RDTSC /// per resume); stays 0 otherwise. Same single-writer Relaxed discipline. budget_cycles: AtomicU64, + /// RFC 007 (`smarm-causal`) — id of the causal-profiling site this actor + /// is currently inside (0 = none). Lives in the slot, not a thread-local, + /// so it survives preemption and cross-scheduler migration. Written only + /// by the actor itself (guard enter/exit on its own thread), read by that + /// same thread in `maybe_preempt` — single-writer Relaxed, like `overruns`. + /// Exists regardless of the feature; stays 0 without it. + causal_site: AtomicU32, + /// RFC 007 (`smarm-causal`) — virtual-speedup delay cycles this actor has + /// absorbed (or been credited). Compared against the global ledger in + /// `causal::check`; fast-forwarded on resume-from-park so blocked time + /// absorbs delay for free (Coz's blocked-thread rule). Same single-writer + /// discipline. + causal_delay: AtomicU64, + /// RFC 007 (`smarm-causal`) — true iff this actor's last deschedule was a + /// real park (a successful `park_return`, not a slice yield and not the + /// instant-wake re-queue). Consumed by `causal::on_resume`: only a wake + /// from genuine blocking forgives outstanding virtual delay; a merely + /// preempted (runnable) actor stays in debt and must pay by spinning. + /// Starts true so a fresh spawn doesn't inherit the process's whole delay + /// history. Written by the owning scheduler thread at the deschedule / + /// resume boundary only — same single-writer discipline as `causal_delay`. + causal_parked: AtomicBool, /// Cold lifecycle data. See [`SlotCold`]. pub(crate) cold: RawMutex, } @@ -459,6 +481,9 @@ impl Slot { overruns: AtomicU64::new(0), messages_received: AtomicU64::new(0), budget_cycles: AtomicU64::new(0), + causal_site: AtomicU32::new(0), + causal_delay: AtomicU64::new(0), + causal_parked: AtomicBool::new(true), cold: RawMutex::new(SlotCold { actor: None, waiters: Vec::new(), @@ -547,6 +572,61 @@ impl Slot { self.overruns.store(0, Ordering::Relaxed); self.messages_received.store(0, Ordering::Relaxed); self.budget_cycles.store(0, Ordering::Relaxed); + self.causal_site.store(0, Ordering::Relaxed); + self.causal_delay.store(0, Ordering::Relaxed); + // True, not false: the first resume of a fresh actor is a "wake" — + // it must not owe the process's accumulated delay history. + self.causal_parked.store(true, Ordering::Relaxed); + } + + /// RFC 007 — mark that this actor's deschedule was a genuine park. + /// Called from the scheduler's `YieldIntent::Park` branch on a successful + /// `park_return` only. + #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))] + #[inline] + pub(crate) fn set_causal_parked(&self) { + self.causal_parked.store(true, Ordering::Relaxed); + } + + /// RFC 007 — consume the parked marker at resume: returns whether the + /// last deschedule was a real park, and clears it so the next resume + /// defaults to "was runnable" unless the park branch says otherwise. + #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))] + #[inline] + pub(crate) fn take_causal_parked(&self) -> bool { + self.causal_parked.swap(false, Ordering::Relaxed) + } + + /// RFC 007 — current causal site id (0 = none). Single-writer: only the + /// on-CPU actor's thread writes, via the site-guard enter/exit. + #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))] + #[inline] + pub(crate) fn causal_site(&self) -> u32 { + self.causal_site.load(Ordering::Relaxed) + } + + /// RFC 007 — write the current causal site id (guard enter/exit). + #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))] + #[inline] + pub(crate) fn set_causal_site(&self, id: u32) { + self.causal_site.store(id, Ordering::Relaxed); + } + + /// RFC 007 — absorbed/credited virtual-delay cycles. + #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))] + #[inline] + pub(crate) fn causal_delay(&self) -> u64 { + self.causal_delay.load(Ordering::Relaxed) + } + + /// RFC 007 — set the absorbed-delay ledger (spin-absorb, credit, or the + /// resume-path fast-forward). Single-writer per the resume protocol: the + /// actor's own thread while on-CPU, the resuming scheduler thread at the + /// resume boundary — never both at once. + #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))] + #[inline] + pub(crate) fn set_causal_delay(&self, v: u64) { + self.causal_delay.store(v, Ordering::Relaxed); } /// A pid's-eye snapshot of the slot. Cold paths re-read this under the @@ -1646,6 +1726,11 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { crate::preempt::reset_timeslice(); } PREEMPTION_ENABLED.with(|c| c.set(true)); + // RFC 007: delay accrued while this actor was off-CPU is absorbed for + // free (Coz's blocked-thread rule) — fast-forward its ledger and arm + // the per-thread sample clock before it runs. + #[cfg(feature = "smarm-causal")] + crate::causal::on_resume(slot); crate::te!(crate::trace::Event::Resume(pid)); unsafe { switch_to_actor() }; @@ -1680,6 +1765,11 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { } YieldIntent::Park => { if slot.word.park_return(gen) { + // RFC 007: a real park — the eventual wake forgives + // delay accrued while blocked (the instant-wake + // re-queue below does NOT: that actor never blocked). + #[cfg(feature = "smarm-causal")] + slot.set_causal_parked(); crate::te!(crate::trace::Event::Park(pid)); } else { // An unpark landed in the prep-to-park window; the diff --git a/tests/causal.rs b/tests/causal.rs new file mode 100644 index 0000000..d8ad4d2 --- /dev/null +++ b/tests/causal.rs @@ -0,0 +1,334 @@ +//! Integration tests for `smarm-causal` — native causal profiling (RFC 007). +//! +//! Gated on the feature; run with: +//! cargo test --test causal --features smarm-causal +#![cfg(feature = "smarm-causal")] + +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +/// Progress points count, and snapshots expose deltas by name. +#[test] +fn progress_point_counts() { + smarm::init(smarm::Config::exact(1)).run(|| { + let before = smarm::causal::progress_snapshot(); + let h = smarm::spawn(|| { + for _ in 0..100 { + smarm::progress!("units-a"); + } + for _ in 0..7 { + smarm::progress!("units-b"); + } + }); + h.join().unwrap(); + let after = smarm::causal::progress_snapshot(); + let delta = |name: &str| { + after.iter().find(|(n, _)| n == name).map(|(_, c)| *c).unwrap() + - before + .iter() + .find(|(n, _)| n == name) + .map(|(_, c)| *c) + .unwrap_or(0) + }; + assert_eq!(delta("units-a"), 100); + assert_eq!(delta("units-b"), 7); + }); +} + +/// Site guards nest and restore, and site identity is per-actor: observable +/// from inside the actor and gone once the guard drops. +#[test] +fn site_guard_nesting_restores() { + smarm::init(smarm::Config::exact(1)).run(|| { + let h = smarm::spawn(|| { + assert_eq!(smarm::causal::current_site_name(), None); + { + let _outer = smarm::causal_site!("outer"); + assert_eq!( + smarm::causal::current_site_name().as_deref(), + Some("outer") + ); + { + let _inner = smarm::causal_site!("inner"); + assert_eq!( + smarm::causal::current_site_name().as_deref(), + Some("inner") + ); + } + assert_eq!( + smarm::causal::current_site_name().as_deref(), + Some("outer") + ); + } + assert_eq!(smarm::causal::current_site_name(), None); + }); + h.join().unwrap(); + }); +} + +/// The core Coz transposition: while an experiment targets site S with a +/// nonzero speedup, work inside S grows the global delay ledger and credits +/// itself; an actor *outside* S absorbs the difference (its per-actor ledger +/// catches up to the global one) instead of running delay-free. +#[test] +fn virtual_speedup_ledger() { + let global_out = Arc::new(AtomicU64::new(0)); + let absorbed_out = Arc::new(AtomicU64::new(0)); + let g_out = global_out.clone(); + let a_out = absorbed_out.clone(); + + smarm::init(smarm::Config::exact(2)).run(move || { + let stop = Arc::new(AtomicBool::new(false)); + + // Bystander: never in the target site; must absorb injected delay. + let stop2 = stop.clone(); + let out2 = a_out.clone(); + let bystander = smarm::spawn(move || { + while !stop2.load(Ordering::Relaxed) { + smarm::check!(); + out2.store( + smarm::causal::my_absorbed_delay_cycles(), + Ordering::Relaxed, + ); + } + }); + + // Target: spins inside the experiment's site. + let stop3 = stop.clone(); + let target = smarm::spawn(move || { + let _g = smarm::causal_site!("hot-site"); + while !stop3.load(Ordering::Relaxed) { + smarm::check!(); + } + }); + + let base = smarm::causal::global_delay_cycles(); + smarm::causal::begin_experiment_for_test("hot-site", 50); + smarm::sleep(Duration::from_millis(200)); + g_out.store( + smarm::causal::global_delay_cycles() - base, + Ordering::Relaxed, + ); + smarm::causal::end_experiment_for_test(); + + stop.store(true, Ordering::Relaxed); + bystander.join().unwrap(); + target.join().unwrap(); + }); + + let global = global_out.load(Ordering::Relaxed); + let absorbed = absorbed_out.load(Ordering::Relaxed); + + // The target's site work must have grown the ledger… + assert!(global > 0, "global delay ledger never grew"); + // …and the bystander must have absorbed a meaningful share of it (it can + // lag by a few check intervals; require half to be robust). + assert!( + absorbed >= global / 2, + "bystander absorbed {absorbed} of {global} global delay cycles" + ); +} + +/// The forgiveness rule must not leak: delay is forgiven on resume from a +/// *park* (Coz's blocked-thread rule), but an actor that merely yields on +/// timeslice expiry stays runnable and must PAY — its work rate has to drop +/// while an experiment targets someone else. (Regression: forgiving on every +/// resume made all bystanders delay-immune, so experiments moved nothing.) +#[test] +fn runnable_bystander_pays_delay() { + let baseline = Arc::new(AtomicU64::new(0)); + let during = Arc::new(AtomicU64::new(0)); + let b_out = baseline.clone(); + let d_out = during.clone(); + + smarm::init(smarm::Config::exact(1)).run(move || { + let stop = Arc::new(AtomicBool::new(false)); + + // Target: in-site spinner; grows the ledger while the experiment runs. + let stop_t = stop.clone(); + let target = smarm::spawn(move || { + let _g = smarm::causal_site!("pay-site"); + while !stop_t.load(Ordering::Relaxed) { + smarm::check!(); + } + }); + + // Bystander: pure check!-loop — yields on slice expiry, never parks. + let stop_b = stop.clone(); + let iters = Arc::new(AtomicU64::new(0)); + let iters2 = iters.clone(); + let absorbed = Arc::new(AtomicU64::new(0)); + let absorbed2 = absorbed.clone(); + let bystander = smarm::spawn(move || { + while !stop_b.load(Ordering::Relaxed) { + iters2.fetch_add(1, Ordering::Relaxed); + smarm::check!(); + absorbed2.store( + smarm::causal::my_absorbed_delay_cycles(), + Ordering::Relaxed, + ); + } + }); + + let window = |out: &AtomicU64| { + let i0 = iters.load(Ordering::Relaxed); + let t = std::time::Instant::now(); + smarm::sleep(Duration::from_millis(150)); + let rate = + (iters.load(Ordering::Relaxed) - i0) as f64 / t.elapsed().as_secs_f64(); + out.store(rate as u64, Ordering::Relaxed); + }; + + window(&b_out); + let d0 = smarm::causal::global_delay_cycles(); + smarm::causal::begin_experiment_for_test("pay-site", 50); + window(&d_out); + smarm::causal::end_experiment_for_test(); + let injected = smarm::causal::global_delay_cycles() - d0; + + stop.store(true, Ordering::Relaxed); + target.join().unwrap(); + bystander.join().unwrap(); + + // The never-parking bystander's ledger can only advance by actually + // spinning (forgiveness requires a real park), so this is no longer + // vacuous: most of the injected delay must have been paid for real. + let a = absorbed.load(Ordering::Relaxed); + assert!(injected > 0, "experiment injected nothing"); + assert!( + a >= injected / 2, + "bystander spun off {a} of {injected} injected cycles" + ); + }); + + let b = baseline.load(Ordering::Relaxed) as f64; + let d = during.load(Ordering::Relaxed) as f64; + // Time-shared single scheduler: target is on-CPU ~50% of wall, so a 50% + // virtual speedup injects ~25% of wall — expected slowdown ×0.8 (measured + // exactly that live); on real parallelism it's ×0.67. Either way <0.9. + assert!(b > 0.0, "bystander never ran"); + assert!( + d < b * 0.9, + "runnable bystander did not pay: baseline {b:.0}/s, during experiment {d:.0}/s" + ); +} + +/// A zero-percent experiment must inject nothing: the null experiment is the +/// baseline Coz relies on, and it doubles as the overhead sanity check. +#[test] +fn zero_speedup_injects_nothing() { + smarm::init(smarm::Config::exact(2)).run(|| { + let stop = Arc::new(AtomicBool::new(false)); + let stop2 = stop.clone(); + let worker = smarm::spawn(move || { + let _g = smarm::causal_site!("zero-site"); + while !stop2.load(Ordering::Relaxed) { + smarm::check!(); + } + }); + + let before = smarm::causal::global_delay_cycles(); + smarm::causal::begin_experiment_for_test("zero-site", 0); + smarm::sleep(Duration::from_millis(50)); + smarm::causal::end_experiment_for_test(); + let after = smarm::causal::global_delay_cycles(); + + stop.store(true, Ordering::Relaxed); + worker.join().unwrap(); + assert_eq!(after, before, "0% speedup must not grow the delay ledger"); + }); +} + +/// `impact_pct` computes the normalized throughput impact of a (site, +/// speedup) cell against that site's own 0% baseline: injected virtual delay +/// is removed from the divisor (Coz's normalization), so a bottleneck site +/// shows positive impact and a fully-overlapped one shows ~zero. +#[test] +fn impact_from_synthetic_results() { + use smarm::causal::{impact_pct, tsc_hz, ExperimentResult}; + let hz = tsc_hz(); + let secs = |s: f64| Duration::from_secs_f64(s); + let cycles = |s: f64| (s * hz) as u64; + let results = vec![ + // Baseline: 1000 units in 1s, nothing injected. + ExperimentResult { + site: "bottleneck".into(), + speedup_pct: 0, + duration: secs(1.0), + deltas: vec![("units".into(), 1000)], + injected_cycles: 0, + }, + // Bottleneck at 50%: raw count unchanged, 0.5s injected + // -> rate 1000/0.5 = 2x baseline -> +100%. + ExperimentResult { + site: "bottleneck".into(), + speedup_pct: 50, + duration: secs(1.0), + deltas: vec![("units".into(), 1000)], + injected_cycles: cycles(0.5), + }, + ExperimentResult { + site: "overlapped".into(), + speedup_pct: 0, + duration: secs(1.0), + deltas: vec![("units".into(), 1000)], + injected_cycles: 0, + }, + // Overlapped at 50%: everyone else (incl. the real bottleneck) + // slowed, count drops with the divisor -> ~0% impact. + ExperimentResult { + site: "overlapped".into(), + speedup_pct: 50, + duration: secs(1.0), + deltas: vec![("units".into(), 500)], + injected_cycles: cycles(0.5), + }, + ]; + let bn = impact_pct(&results, "bottleneck", 50, "units").unwrap(); + assert!((bn - 100.0).abs() < 5.0, "bottleneck impact: {bn}"); + let ov = impact_pct(&results, "overlapped", 50, "units").unwrap(); + assert!(ov.abs() < 5.0, "overlapped impact: {ov}"); + // Missing cells yield None, not garbage. + assert!(impact_pct(&results, "nosuch", 50, "units").is_none()); + assert!(impact_pct(&results, "bottleneck", 50, "nosuch").is_none()); +} + +/// End-to-end controller pass over a toy workload: the report must contain a +/// human summary naming every (site × speedup) cell and a Coz-compatible +/// section with experiment / throughput-point lines. +#[test] +fn controller_produces_report() { + smarm::init(smarm::Config::exact(2)).run(|| { + let stop = Arc::new(AtomicBool::new(false)); + let stop2 = stop.clone(); + let worker = smarm::spawn(move || { + while !stop2.load(Ordering::Relaxed) { + { + let _g = smarm::causal_site!("stage-x"); + for _ in 0..50 { + smarm::check!(); + } + } + smarm::progress!("items"); + } + }); + + let results = smarm::causal::run_experiments(&smarm::causal::ExperimentPlan { + speedups_pct: vec![0, 50], + experiment: Duration::from_millis(60), + cooldown: Duration::from_millis(20), + }); + stop.store(true, Ordering::Relaxed); + worker.join().unwrap(); + + assert!(!results.is_empty()); + let summary = smarm::causal::render_summary(&results); + assert!(summary.contains("stage-x"), "summary: {summary}"); + assert!(summary.contains("items"), "summary: {summary}"); + + let coz = smarm::causal::render_coz(&results); + assert!(coz.contains("experiment\tselected=stage-x"), "coz: {coz}"); + assert!(coz.contains("throughput-point\tname=items"), "coz: {coz}"); + }); +}