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:
Claude
@@ -1,3 +1,4 @@
|
||||
target
|
||||
Cargo.lock
|
||||
smarm_trace.json
|
||||
/bench_results/
|
||||
|
||||
@@ -47,3 +47,11 @@ harness = false
|
||||
[[bench]]
|
||||
name = "tokio_favored"
|
||||
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
|
||||
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<T>` is `Arc<Mutex<_>>` 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
|
||||
|
||||
@@ -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 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;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
+18
-16
@@ -94,13 +94,13 @@ fn assert_no_preempt() {
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub(crate) struct MutexQueue {
|
||||
pub struct MutexQueue {
|
||||
q: std::sync::Mutex<std::collections::VecDeque<Pid>>,
|
||||
}
|
||||
|
||||
#[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<Pid> {
|
||||
pub fn pop(&self) -> Option<Pid> {
|
||||
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<AtomicUsize>,
|
||||
@@ -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<Pid> {
|
||||
pub fn pop(&self) -> Option<Pid> {
|
||||
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<Pid> {
|
||||
pub fn pop(&self) -> Option<Pid> {
|
||||
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()
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user