436 lines
18 KiB
Rust
436 lines
18 KiB
Rust
//! 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,<variant>,<workload>,<threads>,<budget>,<spinners>,<gap_us>,
|
|
//! <n>,<p50_us>,<p90_us>,<p99_us>,<min_us>,<max_us>,<cores_busy>,<ops_s>
|
|
//! ("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<usize> {
|
|
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<u64> {
|
|
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<Option<u32>> {
|
|
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<u32>) -> 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<u64>,
|
|
/// 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<u64>,
|
|
cpu_delta: f64,
|
|
wall_s: f64,
|
|
ops: u64,
|
|
}
|
|
|
|
fn finish(scratch: Arc<Mutex<Scratch>>) -> 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<u32>, 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::<Instant>();
|
|
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<u32>, 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<u32>, 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<u32>, 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<Sample>) -> Aggregate {
|
|
let mut pooled: Vec<u64> = Vec::new();
|
|
let mut cores: Vec<f64> = Vec::new();
|
|
let mut ops_s: Vec<f64> = 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<dyn Fn(usize, u64, Option<u32>) -> Sample>);
|
|
let workloads: Vec<Workload> = 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<Sample> =
|
|
(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
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|