diff --git a/.gitignore b/.gitignore index 01eb4b6..116a693 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ target Cargo.lock smarm_trace.json +/bench_results/ diff --git a/Cargo.toml b/Cargo.toml index 9e61d68..25da28e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -47,3 +47,11 @@ harness = false [[bench]] name = "tokio_favored" harness = false + +[[bench]] +name = "rq_micro" +harness = false + +[[bench]] +name = "rq_runtime" +harness = false diff --git a/ROADMAP_v0.5.md b/ROADMAP_v0.5.md index 576d60a..02e6f20 100644 --- a/ROADMAP_v0.5.md +++ b/ROADMAP_v0.5.md @@ -86,14 +86,17 @@ Make slot lookup lock-free and per-slot state independently mutable. (`live == 0` alone implies the queue holds nothing actionable; argument rewritten at the `schedule_loop` site). -## Phase 4 — Bench harness -- [ ] Raw-structure microbench: N threads, push/pop throughput vs thread count, - sweeping producer:consumer ratios. Isolates the data structure. -- [ ] Runtime-level: yield-storm, ping-pong-pairs, spawn-storm, all sweeping - scheduler count. Reuses existing `benches/` harness style. -- [ ] Driver script rebuilding per `rq-*` feature to compare variants in one go. -- [ ] Sandbox validates correctness via oversubscribed `Config::exact(N)` on 1 - core; real contention numbers come from the 20-core box. +## Phase 4 — Bench harness ✅ DONE (harness; real numbers from the 20-core box) +- [x] `benches/rq_micro.rs`: raw structures, threads × p:c ratio sweep. One + binary covers all three structures (types compile in every build). +- [x] `benches/rq_runtime.rs`: yield-storm, ping-pong-pairs, spawn-storm, + sweeping scheduler count; variant baked in by feature. +- [x] `scripts/bench_rq.sh`: rebuilds per `rq-*` feature, aggregates RQCSV + lines into `bench_results/summary.csv`. Knobs via SMARM_BENCH_* env; + e.g. `SMARM_BENCH_THREADS="1 2 4 8 16 20" ./scripts/bench_rq.sh`. +- [x] Harness validated end-to-end at smoke scale on the 1-core sandbox; + contention curves and the actual variant decision come from the + 20-core box. ## Phase 5 — Safety hardening & model checking - [ ] `loom` (dev-dependency only, x86 Linux) model tests for the slot state @@ -104,6 +107,17 @@ Make slot lookup lock-free and per-slot state independently mutable. --- ## Fast follow (post-v0.5, written down so it isn't lost) +- **Assert the invariants we lean on.** This cycle accumulated load-bearing + invariants: at-most-once-enqueued, queue-ops-under-NoPreempt, never two + cold locks, cold-path generation re-verify under the lock, finalize's + decrement-last, pushes-pair-with-Queued-transitions, thread-local guards + never crossing a switch point. Whenever code RELIES on one and a cheap + check exists, assert it at the point of reliance — `debug_assert!` on hot + paths, full `assert!`/loud panic on cold ones — so a violation fails at + the breakage site, not three modules downstream (the slab-overflow panic + and the queue-op preemption debug_assert are the pattern). Sweep the + existing code for missed spots; new code adopts it as house style. The + phase-5 audit is the natural vehicle for the sweep. - **Channel mutex migration.** `channel::Inner` is `Arc>` of the same poison class as the old shared lock; `recv_match` even runs a user predicate under it. The Phase-1 `check_cancelled` gating already removes the diff --git a/benches/rq_micro.rs b/benches/rq_micro.rs new file mode 100644 index 0000000..4b76d35 --- /dev/null +++ b/benches/rq_micro.rs @@ -0,0 +1,186 @@ +//! Raw run-queue microbench (ROADMAP_v0.5 phase 4). +//! +//! Benches the three queue STRUCTURES directly — no runtime, no actors — to +//! isolate the data structure under contention. All three types compile in +//! every build, so this binary covers the whole matrix in one run; it does +//! NOT need the rq-* feature rebuild dance (that's `rq_runtime`). +//! +//! Sweeps thread count × producer:consumer ratio. Queues are sized to the +//! item count, so the occupancy contract holds trivially and producers never +//! block on capacity. +//! +//! Knobs (env): +//! SMARM_BENCH_THREADS space-separated sweep, default "1 2 4" +//! SMARM_BENCH_ITEMS items per measurement, default 200_000 +//! SMARM_BENCH_RUNS repetitions per config (median reported), default 5 +//! +//! Output: the house table, plus one machine-readable line per config: +//! RQCSV,micro,,,,,, +//! +//! NOTE: numbers from a 1-core sandbox only validate the harness; real +//! contention curves come from the many-core box (scripts/bench_rq.sh). + +use smarm::pid::Pid; +use smarm::run_queue::{MpmcRing, MutexQueue, StripedRing}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Instant; + +fn env_usize(key: &str, default: usize) -> usize { + 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, 4]) +} + +/// Generic driver: `producers` threads push `items` total, `consumers` +/// threads pop until everything is accounted for. Returns elapsed µs. +fn drive( + q: Arc, + push: fn(&Q, Pid), + pop: fn(&Q) -> Option, + producers: usize, + consumers: usize, + items: usize, +) -> u128 { + let remaining = Arc::new(AtomicUsize::new(items)); + let start = Instant::now(); + let mut hs = Vec::new(); + let per = items / producers; + for p in 0..producers { + let q = q.clone(); + // Give the last producer the remainder. + let n = if p == producers - 1 { items - per * (producers - 1) } else { per }; + hs.push(std::thread::spawn(move || { + let pid = Pid::new(p as u32, 0); + for _ in 0..n { + push(&q, pid); + } + })); + } + for _ in 0..consumers { + let q = q.clone(); + let remaining = remaining.clone(); + hs.push(std::thread::spawn(move || loop { + // Claim-then-pop so consumers exit promptly when the budget hits + // zero; the claim is backed out on a miss. + let r = remaining.load(Ordering::Relaxed); + if r == 0 { + return; + } + if pop(&q).is_some() { + remaining.fetch_sub(1, Ordering::Relaxed); + } else { + std::hint::spin_loop(); + } + })); + } + for h in hs { + h.join().unwrap(); + } + start.elapsed().as_micros() +} + +/// Single-thread alternating push/pop (the T = 1 case). +fn drive_single(q: &Q, push: fn(&Q, Pid), pop: fn(&Q) -> Option, items: usize) -> u128 { + let pid = Pid::new(0, 0); + let start = Instant::now(); + for _ in 0..items { + push(q, pid); + assert!(pop(q).is_some()); + } + start.elapsed().as_micros() +} + +struct Case { + structure: &'static str, + threads: usize, + producers: usize, + consumers: usize, +} + +fn ratios_for(threads: usize) -> Vec<(usize, usize)> { + if threads < 2 { + return vec![(1, 1)]; // label only; T=1 runs the alternating driver + } + let mut v = vec![(threads / 2, threads - threads / 2)]; // balanced + if threads >= 4 { + v.push((3 * threads / 4, threads - 3 * threads / 4)); // producer-heavy + v.push((threads / 4, threads - threads / 4)); // consumer-heavy + } + v +} + +fn main() { + let threads_sweep = env_threads(); + let items = env_usize("SMARM_BENCH_ITEMS", 200_000); + let runs = env_usize("SMARM_BENCH_RUNS", 5); + + println!("\n{}", "=".repeat(86)); + println!(" run-queue raw structures — items={items}, runs={runs} (median)"); + println!("{}", "=".repeat(86)); + println!( + "{:>10} | {:>7} | {:>7} | {:>10} | {:>14}", + "structure", "threads", "p:c", "median µs", "items/s" + ); + println!("{}", "-".repeat(86)); + + let mut cases = Vec::new(); + for &t in &threads_sweep { + for (p, c) in ratios_for(t) { + for s in ["mutex", "mpmc", "striped"] { + cases.push(Case { structure: s, threads: t, producers: p, consumers: c }); + } + } + } + + for case in cases { + let mut times: Vec = (0..runs) + .map(|_| { + // Fresh queue per run; capacity = items so pushes never stall. + match case.structure { + "mutex" => { + let q = Arc::new(MutexQueue::new(case.threads, items)); + if case.threads < 2 { + drive_single(&*q, MutexQueue::push, MutexQueue::pop, items) + } else { + drive(q, MutexQueue::push, MutexQueue::pop, case.producers, case.consumers, items) + } + } + "mpmc" => { + let q = Arc::new(MpmcRing::with_capacity(items)); + if case.threads < 2 { + drive_single(&*q, MpmcRing::push, MpmcRing::pop, items) + } else { + drive(q, MpmcRing::push, MpmcRing::pop, case.producers, case.consumers, items) + } + } + "striped" => { + let q = Arc::new(StripedRing::new(case.threads.max(1), items)); + if case.threads < 2 { + drive_single(&*q, StripedRing::push, StripedRing::pop, items) + } else { + drive(q, StripedRing::push, StripedRing::pop, case.producers, case.consumers, items) + } + } + _ => unreachable!(), + } + }) + .collect(); + times.sort_unstable(); + let median = times[times.len() / 2]; + let per_s = (items as f64 / (median as f64 / 1e6)) as u64; + let ratio = format!("{}:{}", case.producers, case.consumers); + println!( + "{:>10} | {:>7} | {:>7} | {:>10} | {:>14}", + case.structure, case.threads, ratio, median, per_s + ); + println!( + "RQCSV,micro,{},{},{},{},{},{}", + case.structure, case.threads, ratio, items, median, per_s + ); + } +} diff --git a/benches/rq_runtime.rs b/benches/rq_runtime.rs new file mode 100644 index 0000000..75d254d --- /dev/null +++ b/benches/rq_runtime.rs @@ -0,0 +1,188 @@ +//! Runtime-level run-queue benches (ROADMAP_v0.5 phase 4). +//! +//! These exercise the WHOLE scheduler with the compile-time-selected queue, +//! so comparing variants means rebuilding per rq-* feature — that's what +//! scripts/bench_rq.sh does. Workloads: +//! +//! yield-storm — N actors yield K times each. Pure queue churn: +//! every yield is a push + pop with nothing in between. +//! ping-pong-pairs — P channel pairs, M roundtrips each. Park/unpark +//! latency through the queue. +//! spawn-storm — S spawn+join of trivial actors. Slab + queue + free +//! list under churn. +//! +//! Knobs (env): +//! SMARM_BENCH_THREADS scheduler-count sweep, default "1 2 4" +//! SMARM_BENCH_RUNS repetitions per config (median), default 5 +//! SMARM_BENCH_YIELD_ACTORS / _YIELDS default 200 / 500 +//! SMARM_BENCH_PAIRS / _ROUNDTRIPS default 32 / 1000 +//! SMARM_BENCH_SPAWNS default 5000 +//! +//! Output: house table + one line per config: +//! RQCSV,runtime,,,,,, +//! +//! NOTE: a 1-core sandbox validates the harness, not the scaling story; +//! real curves come from the many-core box. + +use smarm::runtime::{init, Config}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Instant; + +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_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, 4]) +} + +/// (total_ops, elapsed_µs) for one measured run. +fn yield_storm(threads: usize, actors: usize, yields: usize) -> (u64, u128) { + let rt = init(Config::exact(threads)); + let start = Instant::now(); + rt.run(move || { + let handles: Vec<_> = (0..actors) + .map(|_| { + smarm::spawn(move || { + for _ in 0..yields { + smarm::yield_now(); + } + }) + }) + .collect(); + for h in handles { + let _ = h.join(); + } + }); + ((actors * yields) as u64, start.elapsed().as_micros()) +} + +fn ping_pong_pairs(threads: usize, pairs: usize, roundtrips: usize) -> (u64, u128) { + let rt = init(Config::exact(threads)); + let total = Arc::new(AtomicU64::new(0)); + let t2 = total.clone(); + let start = Instant::now(); + rt.run(move || { + let handles: Vec<_> = (0..pairs) + .map(|_| { + let total = t2.clone(); + smarm::spawn(move || { + let (tx_ab, rx_ab) = smarm::channel::channel::(); + let (tx_ba, rx_ba) = smarm::channel::channel::(); + let n = roundtrips as u64; + let echo = smarm::spawn(move || { + for _ in 0..n { + let v = rx_ab.recv().expect("echo recv"); + tx_ba.send(v + 1).expect("echo send"); + } + }); + for i in 0..n { + tx_ab.send(i).expect("ping send"); + let v = rx_ba.recv().expect("ping recv"); + assert_eq!(v, i + 1); + } + let _ = echo.join(); + total.fetch_add(n, Ordering::Relaxed); + }) + }) + .collect(); + for h in handles { + let _ = h.join(); + } + }); + (total.load(Ordering::Relaxed), start.elapsed().as_micros()) +} + +fn spawn_storm(threads: usize, spawns: usize) -> (u64, u128) { + let rt = init(Config::exact(threads)); + let start = Instant::now(); + rt.run(move || { + // Batches bound simultaneous liveness well below the slab cap. + const BATCH: usize = 1024; + let mut left = spawns; + while left > 0 { + let n = left.min(BATCH); + let handles: Vec<_> = (0..n).map(|_| smarm::spawn(|| {})).collect(); + for h in handles { + let _ = h.join(); + } + left -= n; + } + }); + (spawns as u64, start.elapsed().as_micros()) +} + +fn main() { + let threads_sweep = env_threads(); + let runs = env_usize("SMARM_BENCH_RUNS", 5); + let ya = env_usize("SMARM_BENCH_YIELD_ACTORS", 200); + let yy = env_usize("SMARM_BENCH_YIELDS", 500); + let pp = env_usize("SMARM_BENCH_PAIRS", 32); + let pr = env_usize("SMARM_BENCH_ROUNDTRIPS", 1000); + let ss = env_usize("SMARM_BENCH_SPAWNS", 5000); + + println!("\n{}", "=".repeat(86)); + println!(" runtime benches — variant={}, runs={runs} (median)", variant()); + println!("{}", "=".repeat(86)); + println!( + "{:>16} | {:>7} | {:>16} | {:>10} | {:>14}", + "bench", "threads", "work", "median µs", "ops/s" + ); + println!("{}", "-".repeat(86)); + + type Bench = (&'static str, String, Box (u64, u128)>); + let benches: Vec = vec![ + ( + "yield-storm", + format!("{ya}x{yy}"), + Box::new(move |t| yield_storm(t, ya, yy)), + ), + ( + "ping-pong-pairs", + format!("{pp}x{pr}"), + Box::new(move |t| ping_pong_pairs(t, pp, pr)), + ), + ( + "spawn-storm", + format!("{ss}"), + Box::new(move |t| spawn_storm(t, ss)), + ), + ]; + + for (name, work, f) in &benches { + for &t in &threads_sweep { + let mut ops = 0u64; + let mut times: Vec = (0..runs) + .map(|_| { + let (o, us) = f(t); + ops = o; + us + }) + .collect(); + times.sort_unstable(); + let median = times[times.len() / 2]; + let per_s = (ops as f64 / (median as f64 / 1e6)) as u64; + println!( + "{:>16} | {:>7} | {:>16} | {:>10} | {:>14}", + name, t, work, median, per_s + ); + println!( + "RQCSV,runtime,{},{},{},{},{},{}", + variant(), name, t, work, median, per_s + ); + } + } +} diff --git a/scripts/bench_rq.sh b/scripts/bench_rq.sh new file mode 100755 index 0000000..2edf50a --- /dev/null +++ b/scripts/bench_rq.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Run-queue shootout driver (ROADMAP_v0.5 phase 4). +# +# Rebuilds the runtime bench once per rq-* feature and runs the raw-structure +# microbench once (it covers all structures in a single binary). Results land +# in bench_results/ as full logs; the RQCSV lines are aggregated into +# bench_results/summary.csv for plotting. +# +# Tune the sweep for the box, e.g. on the 20-core machine: +# SMARM_BENCH_THREADS="1 2 4 8 16 20" ./scripts/bench_rq.sh +set -euo pipefail +cd "$(dirname "$0")/.." + +OUT=bench_results +mkdir -p "$OUT" +: "${SMARM_BENCH_THREADS:=1 2 4}" +export SMARM_BENCH_THREADS + +echo "== raw structures (one binary, all variants) ==" +cargo bench --bench rq_micro 2>&1 | tee "$OUT/micro.txt" + +for v in rq-mutex rq-mpmc rq-striped; do + echo "== runtime benches: $v ==" + cargo bench --bench rq_runtime --no-default-features --features "$v" \ + 2>&1 | tee "$OUT/runtime-$v.txt" +done + +echo "bench,kind,a,b,c,d,median_us,ops_per_s" > "$OUT/summary.csv" +grep -h '^RQCSV,' "$OUT"/*.txt | sed 's/^RQCSV,//' >> "$OUT/summary.csv" +echo +echo "Summary: $OUT/summary.csv ($(($(wc -l < "$OUT/summary.csv") - 1)) rows)" diff --git a/src/lib.rs b/src/lib.rs index ebb2b6d..b31e4c3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,7 +27,8 @@ pub mod link; pub mod gen_server; pub mod runtime; pub(crate) mod raw_mutex; -pub(crate) mod run_queue; +#[doc(hidden)] // pub only so benches/rq_micro.rs can drive the raw structures +pub mod run_queue; pub mod trace; // --------------------------------------------------------------------------- diff --git a/src/run_queue.rs b/src/run_queue.rs index ef8f906..df46cee 100644 --- a/src/run_queue.rs +++ b/src/run_queue.rs @@ -94,13 +94,13 @@ fn assert_no_preempt() { // --------------------------------------------------------------------------- #[allow(dead_code)] -pub(crate) struct MutexQueue { +pub struct MutexQueue { q: std::sync::Mutex>, } #[allow(dead_code)] impl MutexQueue { - pub(crate) fn new(_threads: usize, max_actors: usize) -> Self { + pub fn new(_threads: usize, max_actors: usize) -> Self { Self { // Pre-size: the queue can never outgrow the slab, and one // allocation at init beats reallocating under the lock later. @@ -108,17 +108,17 @@ impl MutexQueue { } } - pub(crate) fn push(&self, pid: Pid) { + pub fn push(&self, pid: Pid) { assert_no_preempt(); self.q.lock().unwrap().push_back(pid); } - pub(crate) fn pop(&self) -> Option { + pub fn pop(&self) -> Option { assert_no_preempt(); self.q.lock().unwrap().pop_front() } - pub(crate) fn len(&self) -> u64 { + pub fn len(&self) -> u64 { self.q.lock().unwrap().len() as u64 } } @@ -145,7 +145,7 @@ struct Cell { } #[allow(dead_code)] -pub(crate) struct MpmcRing { +pub struct MpmcRing { buf: Box<[Cell]>, mask: usize, enqueue_pos: CachePadded, @@ -159,11 +159,13 @@ unsafe impl Sync for MpmcRing {} #[allow(dead_code)] impl MpmcRing { - pub(crate) fn new(_threads: usize, max_actors: usize) -> Self { + pub fn new(_threads: usize, max_actors: usize) -> Self { Self::with_capacity(max_actors) } - fn with_capacity(min_cap: usize) -> Self { + /// Pub for the raw-structure microbench (sized to the op count so the + /// occupancy contract is trivially met there). Runtime code uses `new`. + pub fn with_capacity(min_cap: usize) -> Self { // Occupancy ≤ max_actors (queue contract), so capacity = the next // power of two ≥ max_actors can never overflow. (≥ 2 so mask works.) let cap = min_cap.next_power_of_two().max(2); @@ -181,7 +183,7 @@ impl MpmcRing { } } - pub(crate) fn push(&self, pid: Pid) { + pub fn push(&self, pid: Pid) { assert_no_preempt(); assert!( self.try_push(pid), @@ -220,7 +222,7 @@ impl MpmcRing { } } - pub(crate) fn pop(&self) -> Option { + pub fn pop(&self) -> Option { assert_no_preempt(); let mut pos = self.dequeue_pos.0.load(Ordering::Relaxed); loop { @@ -252,7 +254,7 @@ impl MpmcRing { } } - pub(crate) fn len(&self) -> u64 { + pub fn len(&self) -> u64 { let e = self.enqueue_pos.0.load(Ordering::Relaxed); let d = self.dequeue_pos.0.load(Ordering::Relaxed); e.saturating_sub(d) as u64 @@ -276,7 +278,7 @@ impl MpmcRing { // terminates (in practice on the first stripe). #[allow(dead_code)] -pub(crate) struct StripedRing { +pub struct StripedRing { stripes: Box<[MpmcRing]>, /// Stripe count minus one (count is a power of two). stripe_mask: usize, @@ -286,7 +288,7 @@ pub(crate) struct StripedRing { #[allow(dead_code)] impl StripedRing { - pub(crate) fn new(threads: usize, max_actors: usize) -> Self { + pub fn new(threads: usize, max_actors: usize) -> Self { // One stripe per scheduler thread, rounded up to a power of two — // more stripes than threads buys nothing (at most `threads` ops are // in flight) and costs pop-probe latency when mostly empty. @@ -304,7 +306,7 @@ impl StripedRing { } } - pub(crate) fn push(&self, pid: Pid) { + pub fn push(&self, pid: Pid) { assert_no_preempt(); let home = self.push_ticket.0.fetch_add(1, Ordering::Relaxed); // Probe from the home stripe; capacity headroom (Σ ≥ 2×occupancy) @@ -322,7 +324,7 @@ impl StripedRing { } } - pub(crate) fn pop(&self) -> Option { + pub fn pop(&self) -> Option { assert_no_preempt(); let home = self.pop_ticket.0.fetch_add(1, Ordering::Relaxed); for i in 0..=self.stripe_mask { @@ -333,7 +335,7 @@ impl StripedRing { None // snapshot miss possible across stripes; idle-retry absorbs it } - pub(crate) fn len(&self) -> u64 { + pub fn len(&self) -> u64 { self.stripes.iter().map(|s| s.len()).sum() } }