From 7ec453920b93aaa1c4308e2ecedb84f423c85888 Mon Sep 17 00:00:00 2001 From: smarm-agent Date: Sun, 14 Jun 2026 07:52:21 +0000 Subject: [PATCH] RFC 004 bench: spin_sweep measurement tool. Latency (p50/p90/p99/min/max, pooled across runs) vs idle-CPU (cores-busy via getrusage RUSAGE_SELF) over spin_budget_cycles x max_spinners x threads. Four workloads: remote-wake (latency target), fork-join, half-load (cost target), pure-idle (floor check). Env-knobbed SMARM_SPIN_*, greppable SPINCSV lines. Smoke-runs clean on 1 core (no hang); 1-core correctly shows the spinner-starves-producer pathology, real curve needs the many-core box. --- Cargo.toml | 4 + benches/spin_sweep.rs | 435 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 439 insertions(+) create mode 100644 benches/spin_sweep.rs diff --git a/Cargo.toml b/Cargo.toml index acca772..2961105 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,3 +61,7 @@ harness = false [[bench]] name = "rq_runtime" harness = false + +[[bench]] +name = "spin_sweep" +harness = false diff --git a/benches/spin_sweep.rs b/benches/spin_sweep.rs new file mode 100644 index 0000000..0c6ef0e --- /dev/null +++ b/benches/spin_sweep.rs @@ -0,0 +1,435 @@ +//! RFC 004 spinning-worker measurement tool — the latency-vs-idle-CPU curve +//! over `spin_budget_cycles`. +//! +//! Spinning workers trade idle CPU for wake latency: an idle worker burns +//! cycles polling `queue_len` (up to `spin_budget_cycles`, capped at +//! `max_spinners` concurrent spinners) so that incoming work is claimed with +//! no futex syscall. `spin_budget_cycles = 0` is the feature fully off — the +//! historical `thread::sleep` park, exactly preserved. This binary measures +//! both sides of that trade so the sweep can pick the budget default (RFC 004 +//! Q3) and the on/off posture (Q4). +//! +//! Workloads — each one is a fresh runtime per sample (no cross-contamination): +//! +//! remote-wake — a consumer blocks on recv (its worker idles); a producer +//! on another scheduler timestamps just before send. Latency +//! = send -> receipt, the quantity spinning is meant to cut. +//! A fixed gap (SMARM_SPIN_GAP_US) between rounds lets the +//! consumer re-idle each round. THE LATENCY TARGET. +//! fork-join — root forks FANOUT trivial actors and joins them (one +//! burst), gap, repeat. Latency = fork -> all-joined; stresses +//! the fan-out wake. Secondary latency target. +//! half-load — work dispatched every PERIOD_US for ITEMS items, so +//! workers idle a fraction of each period. THE COST TARGET: +//! cores-busy climbs with budget as spin fills the idle gaps. +//! pure-idle — one actor sleeps WINDOW_MS, nothing else runs. The cleanest +//! single picture of spin burn: cores-busy is ~0 at budget 0 +//! and climbs toward max_spinners as the budget grows. +//! +//! Two axes, reported together per config: +//! latency — p50 / p90 / p99 / min / max over the POOLED rounds of every run +//! (pooling gives a stabler tail than a median-of-percentiles). +//! cost — cores-busy = process CPU-seconds (getrusage RUSAGE_SELF, all +//! runtime threads) / wall-seconds over the measured window. 1.0 = +//! one core pegged; approaches max_spinners when every spinner is +//! hot. Reported as the median over runs. +//! +//! The crossover the sweep reveals: spinning only wins when work arrives while +//! a worker is still inside its budget. With a fixed inter-arrival gap, that +//! means budgets whose spin-time (budget_cycles / TSC_freq) exceeds the gap +//! start catching work before the park; smaller budgets park and pay the full +//! wake. So GAP_US sets the knee — tune it against your box's TSC frequency. +//! +//! Knobs (env): +//! SMARM_BENCH_THREADS scheduler-count sweep default "1 2" +//! SMARM_BENCH_RUNS runs per config (pooled/median) default 3 +//! SMARM_SPIN_BUDGET spin_budget_cycles sweep default "0 1000 10000 100000 1000000" +//! SMARM_MAX_SPINNERS max_spinners sweep; "-" = runtime default (N/2) default "-" +//! SMARM_SPIN_GAP_US idle gap between rounds/bursts default 150 +//! SMARM_SPIN_ROUNDS remote-wake rounds default 100 +//! SMARM_SPIN_FANOUT fork-join actors per burst default 64 +//! SMARM_SPIN_BURSTS fork-join bursts default 30 +//! SMARM_SPIN_ITEMS half-load dispatch count default 150 +//! SMARM_SPIN_PERIOD_US half-load inter-dispatch period default 300 +//! SMARM_SPIN_WINDOW_MS pure-idle window default 25 +//! +//! Output: house table + one greppable line per config: +//! SPINCSV,,,,,,, +//! ,,,,,,, +//! ("spinners" is the literal "default" when SMARM_MAX_SPINNERS is "-".) +//! +//! NOTE: on a 1-core box (CI, sandboxes) spinning is *pathological* — a hot +//! spinner starves the very thread that would enqueue work, so the optimal +//! budget there is 0 and you only ever see the cost, never the benefit. This +//! binary on one core validates the harness and catches gross pathologies; the +//! real latency-vs-idle-CPU curve comes from the many-core box. + +use smarm::runtime::{init, Config}; +use smarm::{channel, sleep, spawn}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +// -------------------------------------------------------------------------- +// env helpers (house style, matching rq_runtime.rs) +// -------------------------------------------------------------------------- + +fn variant() -> &'static str { + if cfg!(feature = "rq-mpmc") { + "rq-mpmc" + } else if cfg!(feature = "rq-striped") { + "rq-striped" + } else { + "rq-mutex" + } +} + +fn env_usize(key: &str, default: usize) -> usize { + std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default) +} + +fn env_u64(key: &str, default: u64) -> u64 { + std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default) +} + +fn env_threads() -> Vec { + std::env::var("SMARM_BENCH_THREADS") + .map(|v| v.split_whitespace().filter_map(|t| t.parse().ok()).collect()) + .unwrap_or_else(|_| vec![1, 2]) +} + +fn env_budgets() -> Vec { + std::env::var("SMARM_SPIN_BUDGET") + .map(|v| v.split_whitespace().filter_map(|t| t.parse().ok()).collect()) + .unwrap_or_else(|_| vec![0, 1_000, 10_000, 100_000, 1_000_000]) +} + +/// `None` means "leave the runtime default (N/2)"; encoded in env as "-". +fn env_spinners() -> Vec> { + std::env::var("SMARM_MAX_SPINNERS") + .map(|v| { + v.split_whitespace() + .map(|t| if t == "-" { None } else { t.parse().ok() }) + .collect() + }) + .unwrap_or_else(|_| vec![None]) +} + +fn make_config(threads: usize, budget: u64, spinners: Option) -> Config { + let cfg = Config::exact(threads).spin_budget_cycles(budget); + match spinners { + Some(n) => cfg.max_spinners(n), + None => cfg, + } +} + +// -------------------------------------------------------------------------- +// CPU accounting — process-wide (all runtime threads) via getrusage +// -------------------------------------------------------------------------- + +/// Total CPU seconds (user + system) consumed by the whole process so far. +/// RUSAGE_SELF aggregates every thread, which is exactly what "how much did +/// the runtime burn" needs — the spinners are separate threads from the one +/// reading this. +fn cpu_seconds() -> f64 { + // SAFETY: getrusage with RUSAGE_SELF and a valid out-pointer; we check the + // return code and only read the (now-initialised) timeval fields. + let mut ru: libc::rusage = unsafe { std::mem::zeroed() }; + let rc = unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut ru) }; + assert_eq!(rc, 0, "getrusage(RUSAGE_SELF) failed"); + let tv = |t: libc::timeval| t.tv_sec as f64 + t.tv_usec as f64 * 1e-6; + tv(ru.ru_utime) + tv(ru.ru_stime) +} + +// -------------------------------------------------------------------------- +// One sample = one fresh runtime run +// -------------------------------------------------------------------------- + +struct Sample { + /// Per-event latencies in nanoseconds (empty for pure-idle). + lat_ns: Vec, + /// CPU-seconds / wall-seconds over the measured window. + cores_busy: f64, + /// Completed work units (for ops/s); 0 where not meaningful. + ops: u64, + /// Wall-seconds of the measured window (for ops/s). + wall_s: f64, +} + +/// Shared scratch the root closure writes and the caller reads after `run()` +/// returns. (`run` takes a `FnOnce` and yields no value, so we hand results +/// out through an Arc rather than a return.) +#[derive(Default)] +struct Scratch { + lat_ns: Vec, + cpu_delta: f64, + wall_s: f64, + ops: u64, +} + +fn finish(scratch: Arc>) -> Sample { + let mut s = scratch.lock().unwrap(); + let cores_busy = if s.wall_s > 0.0 { s.cpu_delta / s.wall_s } else { 0.0 }; + Sample { + lat_ns: std::mem::take(&mut s.lat_ns), + cores_busy, + ops: s.ops, + wall_s: s.wall_s, + } +} + +fn remote_wake(threads: usize, budget: u64, spinners: Option, rounds: usize, gap: Duration) -> Sample { + let rt = init(make_config(threads, budget, spinners)); + let scratch = Arc::new(Mutex::new(Scratch::default())); + let out = scratch.clone(); + rt.run(move || { + let (tx, rx) = channel::channel::(); + let lat_sink = out.clone(); + // Consumer parks on recv; its worker idles (spins or parks per budget) + // until each send wakes it. It records the wake latency itself. + let consumer = spawn(move || { + let mut lats = Vec::with_capacity(rounds); + for _ in 0..rounds { + let sent = rx.recv().expect("remote-wake recv"); + lats.push(sent.elapsed().as_nanos() as u64); + } + lat_sink.lock().unwrap().lat_ns = lats; + }); + // Producer: sleep the gap (so the consumer's worker re-idles), then + // timestamp and send. The send is the wake event under measurement. + let cpu0 = cpu_seconds(); + let w0 = Instant::now(); + for _ in 0..rounds { + sleep(gap); + tx.send(Instant::now()).expect("remote-wake send"); + } + let wall = w0.elapsed().as_secs_f64(); + let cpu = cpu_seconds() - cpu0; + let _ = consumer.join(); + let mut s = out.lock().unwrap(); + s.cpu_delta = cpu; + s.wall_s = wall; + s.ops = rounds as u64; + }); + finish(scratch) +} + +fn fork_join(threads: usize, budget: u64, spinners: Option, fanout: usize, bursts: usize, gap: Duration) -> Sample { + let rt = init(make_config(threads, budget, spinners)); + let scratch = Arc::new(Mutex::new(Scratch::default())); + let out = scratch.clone(); + rt.run(move || { + let mut lats = Vec::with_capacity(bursts); + let cpu0 = cpu_seconds(); + let w0 = Instant::now(); + for _ in 0..bursts { + // Gap lets the workers park between bursts, so the fan-out has real + // wakes to do rather than landing on already-hot workers. + sleep(gap); + let t = Instant::now(); + let handles: Vec<_> = (0..fanout).map(|_| spawn(|| {})).collect(); + for h in handles { + let _ = h.join(); + } + lats.push(t.elapsed().as_nanos() as u64); + } + let wall = w0.elapsed().as_secs_f64(); + let cpu = cpu_seconds() - cpu0; + let mut s = out.lock().unwrap(); + s.lat_ns = lats; + s.cpu_delta = cpu; + s.wall_s = wall; + s.ops = (bursts * fanout) as u64; + }); + finish(scratch) +} + +fn half_load(threads: usize, budget: u64, spinners: Option, items: usize, period: Duration) -> Sample { + let rt = init(make_config(threads, budget, spinners)); + let scratch = Arc::new(Mutex::new(Scratch::default())); + let out = scratch.clone(); + rt.run(move || { + let cpu0 = cpu_seconds(); + let w0 = Instant::now(); + // Dispatch one trivial actor every `period`. Between dispatches the + // workers have nothing to do, so they idle for a fraction of each + // period — that idle fraction is what spinning fills (or doesn't). + let mut handles = Vec::with_capacity(items); + for i in 0..items { + let target = period * i as u32; + let elapsed = w0.elapsed(); + if target > elapsed { + sleep(target - elapsed); + } + handles.push(spawn(|| {})); + } + for h in handles { + let _ = h.join(); + } + let wall = w0.elapsed().as_secs_f64(); + let cpu = cpu_seconds() - cpu0; + let mut s = out.lock().unwrap(); + s.cpu_delta = cpu; + s.wall_s = wall; + s.ops = items as u64; + }); + finish(scratch) +} + +fn pure_idle(threads: usize, budget: u64, spinners: Option, window: Duration) -> Sample { + let rt = init(make_config(threads, budget, spinners)); + let scratch = Arc::new(Mutex::new(Scratch::default())); + let out = scratch.clone(); + rt.run(move || { + // One actor sleeps; every other worker thread has nothing to do and + // spins-or-parks per budget for the whole window. + let cpu0 = cpu_seconds(); + let w0 = Instant::now(); + sleep(window); + let wall = w0.elapsed().as_secs_f64(); + let cpu = cpu_seconds() - cpu0; + let mut s = out.lock().unwrap(); + s.cpu_delta = cpu; + s.wall_s = wall; + s.ops = 0; + }); + finish(scratch) +} + +// -------------------------------------------------------------------------- +// stats +// -------------------------------------------------------------------------- + +/// Nearest-rank percentile over an already-sorted slice. `p` in [0, 100]. +fn pct(sorted: &[u64], p: f64) -> u64 { + if sorted.is_empty() { + return 0; + } + let idx = ((p / 100.0) * (sorted.len() - 1) as f64).round() as usize; + sorted[idx.min(sorted.len() - 1)] +} + +fn median_f64(xs: &mut [f64]) -> f64 { + if xs.is_empty() { + return 0.0; + } + xs.sort_by(|a, b| a.partial_cmp(b).unwrap()); + xs[xs.len() / 2] +} + +struct Aggregate { + n: usize, + p50_us: f64, + p90_us: f64, + p99_us: f64, + min_us: f64, + max_us: f64, + cores_busy: f64, + ops_s: f64, +} + +/// Pool latencies across all runs (stabler tail); take the median cores-busy +/// and median ops/s across runs. +fn aggregate(samples: Vec) -> Aggregate { + let mut pooled: Vec = Vec::new(); + let mut cores: Vec = Vec::new(); + let mut ops_s: Vec = Vec::new(); + for s in &samples { + pooled.extend_from_slice(&s.lat_ns); + cores.push(s.cores_busy); + if s.wall_s > 0.0 && s.ops > 0 { + ops_s.push(s.ops as f64 / s.wall_s); + } + } + pooled.sort_unstable(); + let ns_to_us = |ns: u64| ns as f64 / 1000.0; + Aggregate { + n: pooled.len(), + p50_us: ns_to_us(pct(&pooled, 50.0)), + p90_us: ns_to_us(pct(&pooled, 90.0)), + p99_us: ns_to_us(pct(&pooled, 99.0)), + min_us: pooled.first().map(|&v| ns_to_us(v)).unwrap_or(0.0), + max_us: pooled.last().map(|&v| ns_to_us(v)).unwrap_or(0.0), + cores_busy: median_f64(&mut cores), + ops_s: median_f64(&mut ops_s), + } +} + +// -------------------------------------------------------------------------- +// driver +// -------------------------------------------------------------------------- + +fn main() { + let threads_sweep = env_threads(); + let budgets = env_budgets(); + let spinners_sweep = env_spinners(); + let runs = env_usize("SMARM_BENCH_RUNS", 3); + let gap = Duration::from_micros(env_u64("SMARM_SPIN_GAP_US", 150)); + let rounds = env_usize("SMARM_SPIN_ROUNDS", 100); + let fanout = env_usize("SMARM_SPIN_FANOUT", 64); + let bursts = env_usize("SMARM_SPIN_BURSTS", 30); + let items = env_usize("SMARM_SPIN_ITEMS", 150); + let period = Duration::from_micros(env_u64("SMARM_SPIN_PERIOD_US", 300)); + let window = Duration::from_millis(env_u64("SMARM_SPIN_WINDOW_MS", 25)); + + // (name, closure producing one Sample for a (threads,budget,spinners) point) + type Workload = (&'static str, Box) -> Sample>); + let workloads: Vec = vec![ + ("remote-wake", Box::new(move |t, b, s| remote_wake(t, b, s, rounds, gap))), + ("fork-join", Box::new(move |t, b, s| fork_join(t, b, s, fanout, bursts, gap))), + ("half-load", Box::new(move |t, b, s| half_load(t, b, s, items, period))), + ("pure-idle", Box::new(move |t, b, s| pure_idle(t, b, s, window))), + ]; + + let width = 122; + println!("\n{}", "=".repeat(width)); + println!( + " spin sweep — variant={}, runs={runs} (latency pooled, cores median), gap={}µs", + variant(), + gap.as_micros() + ); + println!("{}", "=".repeat(width)); + println!( + "{:>11} | {:>3} | {:>9} | {:>8} | {:>6} | {:>9} | {:>9} | {:>9} | {:>9} | {:>9} | {:>7} | {:>10}", + "workload", "thr", "budget", "spinners", "n", "p50 µs", "p90 µs", "p99 µs", "min µs", "max µs", "cores", "ops/s" + ); + println!("{}", "-".repeat(width)); + + for (name, run_one) in &workloads { + for &threads in &threads_sweep { + for &budget in &budgets { + for &spinners in &spinners_sweep { + let samples: Vec = + (0..runs).map(|_| run_one(threads, budget, spinners)).collect(); + let a = aggregate(samples); + let sp_str = match spinners { + Some(n) => n.to_string(), + None => "default".to_string(), + }; + // Latency columns are blank for the no-latency probe. + let has_lat = a.n > 0; + let lat = |v: f64| if has_lat { format!("{v:.2}") } else { "-".to_string() }; + println!( + "{:>11} | {:>3} | {:>9} | {:>8} | {:>6} | {:>9} | {:>9} | {:>9} | {:>9} | {:>9} | {:>7.3} | {:>10}", + name, + threads, + budget, + sp_str, + a.n, + lat(a.p50_us), + lat(a.p90_us), + lat(a.p99_us), + lat(a.min_us), + lat(a.max_us), + a.cores_busy, + if a.ops_s > 0.0 { format!("{:.0}", a.ops_s) } else { "-".to_string() }, + ); + println!( + "SPINCSV,{},{},{},{},{},{},{},{:.3},{:.3},{:.3},{:.3},{:.3},{:.4},{:.0}", + variant(), name, threads, budget, sp_str, gap.as_micros(), + a.n, a.p50_us, a.p90_us, a.p99_us, a.min_us, a.max_us, a.cores_busy, a.ops_s + ); + } + } + } + } +}