feat(bench): phase 4 — run-queue bench harness + shootout driver
- benches/rq_micro.rs: raw-structure microbench, threads x producer:consumer ratio sweep. Benches all three queue types in one binary (they compile in every build; only the runtime alias is feature-selected), so no rebuild dance. Queues sized to the op count so the occupancy contract is met. - benches/rq_runtime.rs: whole-runtime benches with the selected variant: yield-storm (pure queue churn), ping-pong-pairs (park/unpark latency), spawn-storm (slab + free list + queue under churn). Scheduler-count sweep. - scripts/bench_rq.sh: rebuilds rq_runtime per rq-* feature, runs rq_micro once, aggregates RQCSV lines into bench_results/summary.csv. - All knobs via SMARM_BENCH_* env vars; house table format + machine lines. - run_queue module is now #[doc(hidden)] pub (types + push/pop/len + MpmcRing::with_capacity) solely so the external bench binary can drive the raw structures. docs(roadmap): phase 4 ticked (harness done; numbers from the 20-core box). New fast-follow per review: assert the invariants we lean on — debug_assert! on hot paths, loud assert!/panic on cold ones, at the point of reliance; sweep existing code during the phase-5 audit, adopt as house style. Validated end-to-end at smoke scale on the 1-core sandbox: full driver run, 24-row summary.csv across micro (3 structures x sweeps) and runtime (3 variants x 3 benches x thread sweep).
This commit is contained in:
@@ -1,3 +1,4 @@
|
|||||||
target
|
target
|
||||||
Cargo.lock
|
Cargo.lock
|
||||||
smarm_trace.json
|
smarm_trace.json
|
||||||
|
/bench_results/
|
||||||
|
|||||||
@@ -47,3 +47,11 @@ harness = false
|
|||||||
[[bench]]
|
[[bench]]
|
||||||
name = "tokio_favored"
|
name = "tokio_favored"
|
||||||
harness = false
|
harness = false
|
||||||
|
|
||||||
|
[[bench]]
|
||||||
|
name = "rq_micro"
|
||||||
|
harness = false
|
||||||
|
|
||||||
|
[[bench]]
|
||||||
|
name = "rq_runtime"
|
||||||
|
harness = false
|
||||||
|
|||||||
+22
-8
@@ -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
|
(`live == 0` alone implies the queue holds nothing actionable; argument
|
||||||
rewritten at the `schedule_loop` site).
|
rewritten at the `schedule_loop` site).
|
||||||
|
|
||||||
## Phase 4 — Bench harness
|
## Phase 4 — Bench harness ✅ DONE (harness; real numbers from the 20-core box)
|
||||||
- [ ] Raw-structure microbench: N threads, push/pop throughput vs thread count,
|
- [x] `benches/rq_micro.rs`: raw structures, threads × p:c ratio sweep. One
|
||||||
sweeping producer:consumer ratios. Isolates the data structure.
|
binary covers all three structures (types compile in every build).
|
||||||
- [ ] Runtime-level: yield-storm, ping-pong-pairs, spawn-storm, all sweeping
|
- [x] `benches/rq_runtime.rs`: yield-storm, ping-pong-pairs, spawn-storm,
|
||||||
scheduler count. Reuses existing `benches/` harness style.
|
sweeping scheduler count; variant baked in by feature.
|
||||||
- [ ] Driver script rebuilding per `rq-*` feature to compare variants in one go.
|
- [x] `scripts/bench_rq.sh`: rebuilds per `rq-*` feature, aggregates RQCSV
|
||||||
- [ ] Sandbox validates correctness via oversubscribed `Config::exact(N)` on 1
|
lines into `bench_results/summary.csv`. Knobs via SMARM_BENCH_* env;
|
||||||
core; real contention numbers come from the 20-core box.
|
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
|
## Phase 5 — Safety hardening & model checking
|
||||||
- [ ] `loom` (dev-dependency only, x86 Linux) model tests for the slot state
|
- [ ] `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)
|
## 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<T>` is `Arc<Mutex<_>>` of the
|
- **Channel mutex migration.** `channel::Inner<T>` is `Arc<Mutex<_>>` of the
|
||||||
same poison class as the old shared lock; `recv_match` even runs a user
|
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
|
predicate under it. The Phase-1 `check_cancelled` gating already removes the
|
||||||
|
|||||||
@@ -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,<structure>,<threads>,<p:c>,<items>,<median_us>,<items_per_s>
|
||||||
|
//!
|
||||||
|
//! 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<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, 4])
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Generic driver: `producers` threads push `items` total, `consumers`
|
||||||
|
/// threads pop until everything is accounted for. Returns elapsed µs.
|
||||||
|
fn drive<Q: Send + Sync + 'static>(
|
||||||
|
q: Arc<Q>,
|
||||||
|
push: fn(&Q, Pid),
|
||||||
|
pop: fn(&Q) -> Option<Pid>,
|
||||||
|
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: &Q, push: fn(&Q, Pid), pop: fn(&Q) -> Option<Pid>, 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<u128> = (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
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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,<variant>,<bench>,<threads>,<work>,<median_us>,<ops_per_s>
|
||||||
|
//!
|
||||||
|
//! 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<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, 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::<u64>();
|
||||||
|
let (tx_ba, rx_ba) = smarm::channel::channel::<u64>();
|
||||||
|
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<dyn Fn(usize) -> (u64, u128)>);
|
||||||
|
let benches: Vec<Bench> = 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<u128> = (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
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Executable
+31
@@ -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)"
|
||||||
+2
-1
@@ -27,7 +27,8 @@ pub mod link;
|
|||||||
pub mod gen_server;
|
pub mod gen_server;
|
||||||
pub mod runtime;
|
pub mod runtime;
|
||||||
pub(crate) mod raw_mutex;
|
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;
|
pub mod trace;
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
+18
-16
@@ -94,13 +94,13 @@ fn assert_no_preempt() {
|
|||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
pub(crate) struct MutexQueue {
|
pub struct MutexQueue {
|
||||||
q: std::sync::Mutex<std::collections::VecDeque<Pid>>,
|
q: std::sync::Mutex<std::collections::VecDeque<Pid>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
impl MutexQueue {
|
impl MutexQueue {
|
||||||
pub(crate) fn new(_threads: usize, max_actors: usize) -> Self {
|
pub fn new(_threads: usize, max_actors: usize) -> Self {
|
||||||
Self {
|
Self {
|
||||||
// Pre-size: the queue can never outgrow the slab, and one
|
// Pre-size: the queue can never outgrow the slab, and one
|
||||||
// allocation at init beats reallocating under the lock later.
|
// 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();
|
assert_no_preempt();
|
||||||
self.q.lock().unwrap().push_back(pid);
|
self.q.lock().unwrap().push_back(pid);
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn pop(&self) -> Option<Pid> {
|
pub fn pop(&self) -> Option<Pid> {
|
||||||
assert_no_preempt();
|
assert_no_preempt();
|
||||||
self.q.lock().unwrap().pop_front()
|
self.q.lock().unwrap().pop_front()
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn len(&self) -> u64 {
|
pub fn len(&self) -> u64 {
|
||||||
self.q.lock().unwrap().len() as u64
|
self.q.lock().unwrap().len() as u64
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -145,7 +145,7 @@ struct Cell {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
pub(crate) struct MpmcRing {
|
pub struct MpmcRing {
|
||||||
buf: Box<[Cell]>,
|
buf: Box<[Cell]>,
|
||||||
mask: usize,
|
mask: usize,
|
||||||
enqueue_pos: CachePadded<AtomicUsize>,
|
enqueue_pos: CachePadded<AtomicUsize>,
|
||||||
@@ -159,11 +159,13 @@ unsafe impl Sync for MpmcRing {}
|
|||||||
|
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
impl MpmcRing {
|
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)
|
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
|
// Occupancy ≤ max_actors (queue contract), so capacity = the next
|
||||||
// power of two ≥ max_actors can never overflow. (≥ 2 so mask works.)
|
// power of two ≥ max_actors can never overflow. (≥ 2 so mask works.)
|
||||||
let cap = min_cap.next_power_of_two().max(2);
|
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_no_preempt();
|
||||||
assert!(
|
assert!(
|
||||||
self.try_push(pid),
|
self.try_push(pid),
|
||||||
@@ -220,7 +222,7 @@ impl MpmcRing {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn pop(&self) -> Option<Pid> {
|
pub fn pop(&self) -> Option<Pid> {
|
||||||
assert_no_preempt();
|
assert_no_preempt();
|
||||||
let mut pos = self.dequeue_pos.0.load(Ordering::Relaxed);
|
let mut pos = self.dequeue_pos.0.load(Ordering::Relaxed);
|
||||||
loop {
|
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 e = self.enqueue_pos.0.load(Ordering::Relaxed);
|
||||||
let d = self.dequeue_pos.0.load(Ordering::Relaxed);
|
let d = self.dequeue_pos.0.load(Ordering::Relaxed);
|
||||||
e.saturating_sub(d) as u64
|
e.saturating_sub(d) as u64
|
||||||
@@ -276,7 +278,7 @@ impl MpmcRing {
|
|||||||
// terminates (in practice on the first stripe).
|
// terminates (in practice on the first stripe).
|
||||||
|
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
pub(crate) struct StripedRing {
|
pub struct StripedRing {
|
||||||
stripes: Box<[MpmcRing]>,
|
stripes: Box<[MpmcRing]>,
|
||||||
/// Stripe count minus one (count is a power of two).
|
/// Stripe count minus one (count is a power of two).
|
||||||
stripe_mask: usize,
|
stripe_mask: usize,
|
||||||
@@ -286,7 +288,7 @@ pub(crate) struct StripedRing {
|
|||||||
|
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
impl StripedRing {
|
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 —
|
// One stripe per scheduler thread, rounded up to a power of two —
|
||||||
// more stripes than threads buys nothing (at most `threads` ops are
|
// more stripes than threads buys nothing (at most `threads` ops are
|
||||||
// in flight) and costs pop-probe latency when mostly empty.
|
// 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();
|
assert_no_preempt();
|
||||||
let home = self.push_ticket.0.fetch_add(1, Ordering::Relaxed);
|
let home = self.push_ticket.0.fetch_add(1, Ordering::Relaxed);
|
||||||
// Probe from the home stripe; capacity headroom (Σ ≥ 2×occupancy)
|
// Probe from the home stripe; capacity headroom (Σ ≥ 2×occupancy)
|
||||||
@@ -322,7 +324,7 @@ impl StripedRing {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn pop(&self) -> Option<Pid> {
|
pub fn pop(&self) -> Option<Pid> {
|
||||||
assert_no_preempt();
|
assert_no_preempt();
|
||||||
let home = self.pop_ticket.0.fetch_add(1, Ordering::Relaxed);
|
let home = self.pop_ticket.0.fetch_add(1, Ordering::Relaxed);
|
||||||
for i in 0..=self.stripe_mask {
|
for i in 0..=self.stripe_mask {
|
||||||
@@ -333,7 +335,7 @@ impl StripedRing {
|
|||||||
None // snapshot miss possible across stripes; idle-retry absorbs it
|
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()
|
self.stripes.iter().map(|s| s.len()).sum()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user