Markk116/smarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(causal): native causal profiling behind smarm-causal (RFC 007 v1)

Browse Source 
causal_site! scoped site guards per actor slot, progress! throughput
points, and a Coz-style virtual-speedup engine hooked into
maybe_preempt's amortized cold block: target-site samples grow a global
delay ledger; bystanders spin-absorb their debt at the next causal
check, with timeslice extension so injected delay is not charged
against the slice.

Resume credit (Coz's blocked-thread rule) is gated on a causal_parked
slot bit set only by a real park: crediting on every resume made any
yield-cadence actor delay-immune and every experiment inert (found
live on a 24-core run — dead-flat deltas across all sites).

Report normalization uses a measured TSC frequency (~50ms calibration
on first use) instead of the crate-wide 3 GHz assumption, which
uniformly inflated impact numbers on a 3.7 GHz box. impact_pct() is
the machine-readable form of the summary for programmatic checks.

examples/causal_pipeline.rs burns fixed *work* (calibrated LCG loop),
not fixed wall time — a timed busy-wait absorbs injected delay into
its own budget and reads as a no-op. Self-checking: exits nonzero if
causal separation fails; skips the verdict below 4 cores. Validated
on a 24-core box: reserve (true bottleneck) +29.3%@25/+83.5%@50;
serialize and background-compaction ~0%.

Known v1 gaps (jar): timer-heap deadlines unshifted, no-check!/no-alloc
actors undelayable, multi-scheduler coherence best-effort Relaxed,
off-CPU blame punted, Instant::now() uncorrected.

Zero-cost with the feature off; clippy -D warnings clean both ways;
full suite green with and without smarm-causal.
This commit is contained in:
Claude (sandbox)
2026-07-12 19:10:04 +00:00
parent 1c90a4ef5e
commit 2668f4018f
8 changed files with 1386 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/causal.rs
+607
View File
@@ -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<Mutex<Vec<&'static str>>> = 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<Mutex<Vec<&'static ProgressPoint>>> = 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<u64> = 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<Vec<&'static str>> {
SITES.get_or_init(|| Mutex::new(Vec::new()))
}
fn progress_points() -> &'static Mutex<Vec<&'static ProgressPoint>> {
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<T>(m: &Mutex<T>) -> 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<String> {
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<String> {
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<u32>,
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<ExperimentResult> {
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<f64> = 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<f64> {
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<f64> {
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<u32> = ::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) => {
()
};
}