feat(scheduler): RFC 005 wake slot — per-scheduler capacity-one wake cache
A thread-local Cell<Option<Pid>> per scheduler, checked before the shared
run queue. Runtime-selected via Config { wake_slot: bool }, default OFF
until the slot shootout accepts it (one binary benches both arms).
Push policy: slot-eligible iff the wake originates from actor context
(current_pid().is_some()) — the slot push replaces run_queue.push at the
tail of the unpark protocol's Parked → Queued CAS, so at-most-once-enqueued
holds verbatim as (slot ⊕ shared queue). Scheduler-context wakes (timer/IO
drain) and spawns always go shared (spawns never reach unpark_inner at all).
Displacement is Go semantics: newest wake takes the slot, occupant pushed
shared — moved, never copied.
Pop order: slot, then shared. Slot-popped actors skip reset_timeslice() and
inherit the waker's remaining slice; a handoff chain is bounded by one
slice, after which the preempt-yield re-enqueue goes shared (a yield is not
a wake) — the one-slice starvation bound, zero new counters. Idle and
AllDone are only reachable with an empty local slot by pop order; an
occupied slot elsewhere holds a Queued (live) actor, so the counter-first
termination argument is untouched.
Observability: per-thread slot_hits / slot_displacements (reset at run()
start so post-run stats() reads are per-run), SlotPush/SlotPop trace events.
Bench plan (roadmap v0.9 item 2): rq_runtime gains the slot on/off
dimension (SMARM_BENCH_SLOT, default "0 1") — ping-pong-pairs is the
target metric, yield-storm the regression guard, spawn-storm the
neutrality check. RQCSV grows a slot column; RQSLOT lines carry the
counters; bench_rq.sh aggregates both. Tests pin the push policy through
the counters (actor-context hits, spawn/join bypass, displacement,
default-off, per-run reset).
This commit is contained in:
Claude
+102
-38
@@ -1,25 +1,39 @@
|
||||
//! Runtime-level run-queue benches (ROADMAP_v0.5 phase 4).
|
||||
//! Runtime-level run-queue benches (ROADMAP_v0.5 phase 4; slot dimension
|
||||
//! added for the v0.9 slot shootout, RFC 005).
|
||||
//!
|
||||
//! 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:
|
||||
//! scripts/bench_rq.sh does. The RFC 005 wake slot is a *runtime* Config
|
||||
//! knob, so one binary benches both arms; the slot on/off sweep happens
|
||||
//! inside this binary. Workloads:
|
||||
//!
|
||||
//! yield-storm — N actors yield K times each. Pure queue churn:
|
||||
//! every yield is a push + pop with nothing in between.
|
||||
//! Slot role: REGRESSION GUARD — yields never touch the
|
||||
//! slot, any slot-on delta is pop-path overhead.
|
||||
//! ping-pong-pairs — P channel pairs, M roundtrips each. Park/unpark
|
||||
//! latency through the queue.
|
||||
//! Slot role: TARGET METRIC — every send-wake is an
|
||||
//! actor-context unpark, the slot's home pattern.
|
||||
//! spawn-storm — S spawn+join of trivial actors. Slab + queue + free
|
||||
//! list under churn.
|
||||
//! Slot role: NEUTRALITY CHECK — spawns bypass the slot
|
||||
//! by policy; join wakes fire from finalize (scheduler
|
||||
//! context), also shared.
|
||||
//!
|
||||
//! Knobs (env):
|
||||
//! SMARM_BENCH_THREADS scheduler-count sweep, default "1 2 4"
|
||||
//! SMARM_BENCH_SLOT wake-slot sweep, default "0 1" (off then on)
|
||||
//! 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>
|
||||
//! RQCSV,runtime,<variant>,<slot>,<bench>,<threads>,<work>,<median_us>,<ops_per_s>
|
||||
//! plus, for slot-on configs, the RFC 005 observability counters:
|
||||
//! RQSLOT,<variant>,<bench>,<threads>,<slot_hits>,<slot_displacements>
|
||||
//! (hits/displacements are taken from the same run as the median time).
|
||||
//!
|
||||
//! NOTE: a 1-core sandbox validates the harness, not the scaling story;
|
||||
//! real curves come from the many-core box.
|
||||
@@ -49,9 +63,30 @@ fn env_threads() -> Vec<usize> {
|
||||
.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));
|
||||
fn env_slots() -> Vec<bool> {
|
||||
std::env::var("SMARM_BENCH_SLOT")
|
||||
.map(|v| {
|
||||
v.split_whitespace()
|
||||
.filter_map(|t| match t {
|
||||
"0" | "off" | "false" => Some(false),
|
||||
"1" | "on" | "true" => Some(true),
|
||||
_ => None,
|
||||
})
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_else(|_| vec![false, true])
|
||||
}
|
||||
|
||||
/// One measured run: (total_ops, elapsed_µs, slot_hits, slot_displacements).
|
||||
struct Sample {
|
||||
ops: u64,
|
||||
us: u128,
|
||||
hits: u64,
|
||||
displacements: u64,
|
||||
}
|
||||
|
||||
fn yield_storm(threads: usize, slot: bool, actors: usize, yields: usize) -> Sample {
|
||||
let rt = init(Config::exact(threads).wake_slot(slot));
|
||||
let start = Instant::now();
|
||||
rt.run(move || {
|
||||
let handles: Vec<_> = (0..actors)
|
||||
@@ -67,11 +102,18 @@ fn yield_storm(threads: usize, actors: usize, yields: usize) -> (u64, u128) {
|
||||
let _ = h.join();
|
||||
}
|
||||
});
|
||||
((actors * yields) as u64, start.elapsed().as_micros())
|
||||
let us = start.elapsed().as_micros();
|
||||
let stats = rt.stats();
|
||||
Sample {
|
||||
ops: (actors * yields) as u64,
|
||||
us,
|
||||
hits: stats.slot_hits(),
|
||||
displacements: stats.slot_displacements(),
|
||||
}
|
||||
}
|
||||
|
||||
fn ping_pong_pairs(threads: usize, pairs: usize, roundtrips: usize) -> (u64, u128) {
|
||||
let rt = init(Config::exact(threads));
|
||||
fn ping_pong_pairs(threads: usize, slot: bool, pairs: usize, roundtrips: usize) -> Sample {
|
||||
let rt = init(Config::exact(threads).wake_slot(slot));
|
||||
let total = Arc::new(AtomicU64::new(0));
|
||||
let t2 = total.clone();
|
||||
let start = Instant::now();
|
||||
@@ -103,11 +145,18 @@ fn ping_pong_pairs(threads: usize, pairs: usize, roundtrips: usize) -> (u64, u12
|
||||
let _ = h.join();
|
||||
}
|
||||
});
|
||||
(total.load(Ordering::Relaxed), start.elapsed().as_micros())
|
||||
let us = start.elapsed().as_micros();
|
||||
let stats = rt.stats();
|
||||
Sample {
|
||||
ops: total.load(Ordering::Relaxed),
|
||||
us,
|
||||
hits: stats.slot_hits(),
|
||||
displacements: stats.slot_displacements(),
|
||||
}
|
||||
}
|
||||
|
||||
fn spawn_storm(threads: usize, spawns: usize) -> (u64, u128) {
|
||||
let rt = init(Config::exact(threads));
|
||||
fn spawn_storm(threads: usize, slot: bool, spawns: usize) -> Sample {
|
||||
let rt = init(Config::exact(threads).wake_slot(slot));
|
||||
let start = Instant::now();
|
||||
rt.run(move || {
|
||||
// Batches bound simultaneous liveness well below the slab cap.
|
||||
@@ -122,11 +171,19 @@ fn spawn_storm(threads: usize, spawns: usize) -> (u64, u128) {
|
||||
left -= n;
|
||||
}
|
||||
});
|
||||
(spawns as u64, start.elapsed().as_micros())
|
||||
let us = start.elapsed().as_micros();
|
||||
let stats = rt.stats();
|
||||
Sample {
|
||||
ops: spawns as u64,
|
||||
us,
|
||||
hits: stats.slot_hits(),
|
||||
displacements: stats.slot_displacements(),
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let threads_sweep = env_threads();
|
||||
let slot_sweep = env_slots();
|
||||
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);
|
||||
@@ -134,55 +191,62 @@ fn main() {
|
||||
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!("\n{}", "=".repeat(106));
|
||||
println!(
|
||||
"{:>16} | {:>7} | {:>16} | {:>10} | {:>14}",
|
||||
"bench", "threads", "work", "median µs", "ops/s"
|
||||
" runtime benches — variant={}, runs={runs} (median)",
|
||||
variant()
|
||||
);
|
||||
println!("{}", "-".repeat(86));
|
||||
println!("{}", "=".repeat(106));
|
||||
println!(
|
||||
"{:>16} | {:>7} | {:>4} | {:>16} | {:>10} | {:>14} | {:>10} | {:>9}",
|
||||
"bench", "threads", "slot", "work", "median µs", "ops/s", "slot hits", "displaced"
|
||||
);
|
||||
println!("{}", "-".repeat(106));
|
||||
|
||||
type Bench = (&'static str, String, Box<dyn Fn(usize) -> (u64, u128)>);
|
||||
type Bench = (&'static str, String, Box<dyn Fn(usize, bool) -> Sample>);
|
||||
let benches: Vec<Bench> = vec![
|
||||
(
|
||||
"yield-storm",
|
||||
format!("{ya}x{yy}"),
|
||||
Box::new(move |t| yield_storm(t, ya, yy)),
|
||||
Box::new(move |t, s| yield_storm(t, s, ya, yy)),
|
||||
),
|
||||
(
|
||||
"ping-pong-pairs",
|
||||
format!("{pp}x{pr}"),
|
||||
Box::new(move |t| ping_pong_pairs(t, pp, pr)),
|
||||
Box::new(move |t, s| ping_pong_pairs(t, s, pp, pr)),
|
||||
),
|
||||
(
|
||||
"spawn-storm",
|
||||
format!("{ss}"),
|
||||
Box::new(move |t| spawn_storm(t, ss)),
|
||||
Box::new(move |t, s| spawn_storm(t, s, 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;
|
||||
for &slot in &slot_sweep {
|
||||
let mut samples: Vec<Sample> = (0..runs).map(|_| f(t, slot)).collect();
|
||||
// Median by elapsed time; report the counters from that
|
||||
// same run so hits/time stay paired.
|
||||
samples.sort_unstable_by_key(|s| s.us);
|
||||
let mid = &samples[samples.len() / 2];
|
||||
let per_s = (mid.ops as f64 / (mid.us as f64 / 1e6)) as u64;
|
||||
let slot_str = if slot { "on" } else { "off" };
|
||||
println!(
|
||||
"{:>16} | {:>7} | {:>16} | {:>10} | {:>14}",
|
||||
name, t, work, median, per_s
|
||||
"{:>16} | {:>7} | {:>4} | {:>16} | {:>10} | {:>14} | {:>10} | {:>9}",
|
||||
name, t, slot_str, work, mid.us, per_s, mid.hits, mid.displacements
|
||||
);
|
||||
println!(
|
||||
"RQCSV,runtime,{},{},{},{},{},{}",
|
||||
variant(), name, t, work, median, per_s
|
||||
"RQCSV,runtime,{},{},{},{},{},{},{}",
|
||||
variant(), slot_str, name, t, work, mid.us, per_s
|
||||
);
|
||||
if slot {
|
||||
println!(
|
||||
"RQSLOT,{},{},{},{},{}",
|
||||
variant(), name, t, mid.hits, mid.displacements
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+22
-7
@@ -1,31 +1,46 @@
|
||||
#!/usr/bin/env bash
|
||||
# Run-queue shootout driver (ROADMAP_v0.5 phase 4).
|
||||
# Run-queue shootout driver (ROADMAP_v0.5 phase 4; RFC 005 slot dimension
|
||||
# added for the v0.9 slot shootout).
|
||||
#
|
||||
# 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.
|
||||
# microbench once (it covers all structures in a single binary). The RFC 005
|
||||
# wake slot is a runtime Config knob, NOT a feature — each rq_runtime binary
|
||||
# sweeps slot off/on internally (SMARM_BENCH_SLOT, default "0 1"). Results
|
||||
# land in bench_results/ as full logs; the RQCSV lines are aggregated into
|
||||
# bench_results/summary.csv and the RQSLOT counter lines (slot hits /
|
||||
# displacements, slot-on configs only) into bench_results/slot_counters.csv.
|
||||
#
|
||||
# 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
|
||||
# Slot-only re-run against the frozen rq-mutex substrate:
|
||||
# SMARM_BENCH_SLOT="0 1" cargo bench --bench rq_runtime
|
||||
set -euo pipefail
|
||||
cd "$(dirname "$0")/.."
|
||||
|
||||
OUT=bench_results
|
||||
mkdir -p "$OUT"
|
||||
: "${SMARM_BENCH_THREADS:=1 2 4}"
|
||||
export SMARM_BENCH_THREADS
|
||||
: "${SMARM_BENCH_SLOT:=0 1}"
|
||||
export SMARM_BENCH_THREADS SMARM_BENCH_SLOT
|
||||
|
||||
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 =="
|
||||
echo "== runtime benches: $v (slot sweep: $SMARM_BENCH_SLOT) =="
|
||||
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"
|
||||
# runtime rows: kind,variant,slot,bench,threads,work,median_us,ops_per_s
|
||||
# micro rows: kind,structure,threads,p:c,items,median_us,items_per_s (one
|
||||
# column narrower, as before — split on kind when plotting)
|
||||
echo "kind,a,b,c,d,e,median_us,ops_per_s" > "$OUT/summary.csv"
|
||||
grep -h '^RQCSV,' "$OUT"/*.txt | sed 's/^RQCSV,//' >> "$OUT/summary.csv"
|
||||
|
||||
echo "variant,bench,threads,slot_hits,slot_displacements" > "$OUT/slot_counters.csv"
|
||||
grep -h '^RQSLOT,' "$OUT"/*.txt | sed 's/^RQSLOT,//' >> "$OUT/slot_counters.csv" || true
|
||||
|
||||
echo
|
||||
echo "Summary: $OUT/summary.csv ($(($(wc -l < "$OUT/summary.csv") - 1)) rows)"
|
||||
echo "Slot counters: $OUT/slot_counters.csv ($(($(wc -l < "$OUT/slot_counters.csv") - 1)) rows)"
|
||||
|
||||
+6
-2
@@ -25,8 +25,12 @@
|
||||
//! - **Occupancy is bounded by `max_actors`.** A pid is in the queue at most
|
||||
//! once (pushes pair 1:1 with transitions into `Queued`; only the
|
||||
//! scheduler transitions `Queued → Running` — see the state-machine docs
|
||||
//! in `runtime.rs`), and at most `max_actors` actors exist. The bounded
|
||||
//! rings are sized ≥ `max_actors`, so **`push` is infallible**; a full
|
||||
//! in `runtime.rs`), and at most `max_actors` actors exist. With the RFC
|
||||
//! 005 wake slot enabled the invariant reads "in (slot ⊕ shared queue) at
|
||||
//! most once" — a slot push *replaces* the queue push at the same
|
||||
//! protocol point, and a displacement moves the occupant, never copies
|
||||
//! it — so the bound holds verbatim. The bounded rings are sized ≥
|
||||
//! `max_actors`, so **`push` is infallible**; a full
|
||||
//! ring is an invariant violation and panics loudly rather than spinning.
|
||||
//! - **Preemption must be disabled around every push/pop** (debug-asserted).
|
||||
//! For the mutex variant this is the usual no-switch/no-unwind-under-lock
|
||||
|
||||
+142
-3
@@ -154,6 +154,7 @@ pub struct Config {
|
||||
timeslice_cycles: u64,
|
||||
stack_pool_cap: usize,
|
||||
max_actors: usize,
|
||||
wake_slot: bool,
|
||||
}
|
||||
|
||||
impl Config {
|
||||
@@ -166,6 +167,7 @@ impl Config {
|
||||
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
|
||||
stack_pool_cap: n * 4,
|
||||
max_actors: DEFAULT_MAX_ACTORS,
|
||||
wake_slot: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -182,6 +184,7 @@ impl Config {
|
||||
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
|
||||
stack_pool_cap: max * 4,
|
||||
max_actors: DEFAULT_MAX_ACTORS,
|
||||
wake_slot: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -226,6 +229,18 @@ impl Config {
|
||||
self
|
||||
}
|
||||
|
||||
/// Enable the per-scheduler wake slot (RFC 005): a thread-local,
|
||||
/// capacity-one wake cache checked before the shared run queue. A wake
|
||||
/// performed from actor context parks the woken pid in the waking
|
||||
/// thread's slot; it is resumed next on that core and inherits the
|
||||
/// remainder of the waker's timeslice. Scheduler-context wakes
|
||||
/// (timer/IO drain) and spawns always go to the shared queue.
|
||||
/// Default: `false` (off until the slot shootout accepts it).
|
||||
pub fn wake_slot(mut self, on: bool) -> Self {
|
||||
self.wake_slot = on;
|
||||
self
|
||||
}
|
||||
|
||||
/// The number of scheduler threads this config resolves to.
|
||||
pub fn resolved_thread_count(&self) -> usize {
|
||||
if let Some(e) = self.exact {
|
||||
@@ -249,6 +264,7 @@ impl Default for Config {
|
||||
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
|
||||
stack_pool_cap: avail * 4,
|
||||
max_actors: DEFAULT_MAX_ACTORS,
|
||||
wake_slot: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -264,6 +280,10 @@ pub struct SchedulerStats {
|
||||
pub current_pid_index: AtomicU32,
|
||||
/// Snapshot of run queue length maintained on every push/pop.
|
||||
pub run_queue_len: AtomicU64,
|
||||
/// RFC 005: wakes resumed from this thread's wake slot.
|
||||
pub slot_hits: AtomicU64,
|
||||
/// RFC 005: slot occupants displaced to the shared queue by a newer wake.
|
||||
pub slot_displacements: AtomicU64,
|
||||
}
|
||||
|
||||
impl SchedulerStats {
|
||||
@@ -271,6 +291,8 @@ impl SchedulerStats {
|
||||
Self {
|
||||
current_pid_index: AtomicU32::new(u32::MAX),
|
||||
run_queue_len: AtomicU64::new(0),
|
||||
slot_hits: AtomicU64::new(0),
|
||||
slot_displacements: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -305,6 +327,23 @@ impl RuntimeStats {
|
||||
pub fn sleeping_count(&self) -> u32 {
|
||||
self.inner.sleeping.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// RFC 005: total wakes resumed from a wake slot, summed across
|
||||
/// scheduler threads. Counters are reset at the start of each `run()`,
|
||||
/// so after a run this reads that run's total.
|
||||
pub fn slot_hits(&self) -> u64 {
|
||||
self.inner.stats.iter()
|
||||
.map(|s| s.slot_hits.load(Ordering::Relaxed))
|
||||
.sum()
|
||||
}
|
||||
|
||||
/// RFC 005: total slot occupants displaced to the shared queue, summed
|
||||
/// across scheduler threads. Reset at the start of each `run()`.
|
||||
pub fn slot_displacements(&self) -> u64 {
|
||||
self.inner.stats.iter()
|
||||
.map(|s| s.slot_displacements.load(Ordering::Relaxed))
|
||||
.sum()
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -460,6 +499,9 @@ pub(crate) struct RuntimeInner {
|
||||
/// Preemption knobs, written into each scheduler thread's locals on startup.
|
||||
pub(crate) alloc_interval: u32,
|
||||
pub(crate) timeslice_cycles: u64,
|
||||
/// RFC 005: whether actor-context wakes route through the per-scheduler
|
||||
/// wake slot. Read-only after init; one predictable branch per wake.
|
||||
pub(crate) wake_slot: bool,
|
||||
/// The name <-> pid registry (bidirectional). RawMutex Leaf: never held
|
||||
/// with any other lock; liveness checks under it read only the atomic
|
||||
/// slot word.
|
||||
@@ -477,6 +519,7 @@ impl RuntimeInner {
|
||||
timeslice_cycles: u64,
|
||||
stack_pool_cap: usize,
|
||||
max_actors: usize,
|
||||
wake_slot: bool,
|
||||
) -> Arc<Self> {
|
||||
let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
|
||||
let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect();
|
||||
@@ -496,6 +539,7 @@ impl RuntimeInner {
|
||||
sleeping: AtomicU32::new(0),
|
||||
alloc_interval,
|
||||
timeslice_cycles,
|
||||
wake_slot,
|
||||
registry: RawMutex::new(crate::registry::Registry::new()),
|
||||
stack_pool: RawMutex::new(Vec::new()),
|
||||
stack_pool_cap,
|
||||
@@ -568,8 +612,21 @@ impl RuntimeInner {
|
||||
match slot.word.unpark(pid.generation(), want) {
|
||||
Unpark::Enqueue => {
|
||||
crate::te!(crate::trace::Event::UnparkDirect(pid));
|
||||
// RFC 005: a wake from ACTOR context is slot-eligible —
|
||||
// the woken actor's message bytes are hot in this core's
|
||||
// cache. Scheduler-context wakes (timer/IO drain, where
|
||||
// current_pid is None) have no locality to exploit,
|
||||
// arrive in bursts that would thrash the slot, and
|
||||
// concentrate on the drain winner by construction:
|
||||
// shared queue, always. Spawns never come through here
|
||||
// (install_actor enqueues directly), so they bypass the
|
||||
// slot by construction.
|
||||
if self.wake_slot && crate::actor::current_pid().is_some() {
|
||||
self.slot_push(pid);
|
||||
} else {
|
||||
self.enqueue(pid);
|
||||
}
|
||||
}
|
||||
Unpark::Notified => {
|
||||
crate::te!(crate::trace::Event::UnparkDeferred(pid));
|
||||
}
|
||||
@@ -578,6 +635,38 @@ impl RuntimeInner {
|
||||
}
|
||||
}
|
||||
|
||||
/// RFC 005: park `pid` in this thread's wake slot instead of the shared
|
||||
/// queue. Replaces `enqueue` at the tail of the wake protocol's
|
||||
/// `Parked → Queued` CAS, so the caller has just transitioned the pid
|
||||
/// into `Queued` — same precondition, same invariant, different home.
|
||||
/// Displacement: the NEW wake takes the slot (newest is hottest; the old
|
||||
/// occupant was about to lose its locality window anyway) and the old
|
||||
/// occupant is pushed to the shared queue.
|
||||
fn slot_push(&self, pid: Pid) {
|
||||
debug_assert!(
|
||||
!crate::preempt::PREEMPTION_ENABLED.with(|c| c.get()),
|
||||
"slot_push with preemption enabled — a switch mid-op could \
|
||||
migrate the actor and split the slot access across threads"
|
||||
);
|
||||
debug_assert!(
|
||||
self.slot_at(pid).map(|s| s.word.load()).is_some_and(|w| {
|
||||
crate::slot_state::word_gen(w) == pid.generation()
|
||||
&& crate::slot_state::word_state(w) == crate::slot_state::ST_QUEUED
|
||||
}),
|
||||
"slot_push of a pid not in (gen, Queued)"
|
||||
);
|
||||
let displaced = WAKE_SLOT.with(|s| s.replace(Some(pid)));
|
||||
crate::te!(crate::trace::Event::SlotPush(pid));
|
||||
if let Some(old) = displaced {
|
||||
SCHED_SLOT.with(|s| {
|
||||
self.stats[s.get()]
|
||||
.slot_displacements
|
||||
.fetch_add(1, Ordering::Relaxed)
|
||||
});
|
||||
self.enqueue(old);
|
||||
}
|
||||
}
|
||||
|
||||
/// Allocate the next process-unique `MonitorId`. Lock-free; monitors are a
|
||||
/// cold path but there is no reason to serialize id minting under any lock.
|
||||
pub(crate) fn alloc_monitor_id(&self) -> MonitorId {
|
||||
@@ -621,6 +710,7 @@ pub fn init(config: Config) -> Runtime {
|
||||
config.timeslice_cycles,
|
||||
config.stack_pool_cap,
|
||||
config.max_actors,
|
||||
config.wake_slot,
|
||||
),
|
||||
thread_count: n,
|
||||
}
|
||||
@@ -673,6 +763,13 @@ impl Runtime {
|
||||
);
|
||||
*self.inner.io.lock().unwrap() = Some(IoThread::start().expect("failed to start IO thread"));
|
||||
|
||||
// RFC 005: slot counters reset at the START of a run (not the end),
|
||||
// so `stats()` read after `run()` returns reports that run's totals.
|
||||
for stat in &self.inner.stats {
|
||||
stat.slot_hits.store(0, Ordering::Relaxed);
|
||||
stat.slot_displacements.store(0, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
// Spawn the initial actor through the public spawn path (which
|
||||
// requires a running runtime in the thread-local).
|
||||
RUNTIME.with(|r| *r.borrow_mut() = Some(self.inner.clone()));
|
||||
@@ -750,6 +847,17 @@ thread_local! {
|
||||
/// This scheduler thread's index into RuntimeInner::stats.
|
||||
static SCHED_SLOT: Cell<usize> = const { Cell::new(0) };
|
||||
|
||||
/// RFC 005: the per-scheduler wake slot — a capacity-one wake cache
|
||||
/// checked before the shared run queue. Holds a pid in state
|
||||
/// `(gen, Queued)` exactly as a shared-queue entry would; the
|
||||
/// at-most-once-enqueued invariant reads "in (slot ⊕ shared queue) at
|
||||
/// most once". All access is from the owning thread with preemption
|
||||
/// disabled (the existing queue-op contract), so plain Cell ops suffice:
|
||||
/// no atomics, nothing to steal, nothing to model. Empty whenever the
|
||||
/// thread reaches the idle or termination path (pop order drains it
|
||||
/// first), so it never holds a pid across the end of a run.
|
||||
static WAKE_SLOT: Cell<Option<Pid>> = const { Cell::new(None) };
|
||||
|
||||
/// What the actor wants when it yields back to the scheduler.
|
||||
static YIELD_INTENT: Cell<YieldIntent> = const { Cell::new(YieldIntent::Yield) };
|
||||
}
|
||||
@@ -1060,8 +1168,9 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||||
} // drain_guard drops here
|
||||
|
||||
// ----------------------------------------------------------------
|
||||
// 2. Pop a runnable pid. The queue mutex covers ONLY the pop; the
|
||||
// slot's own atomics carry everything needed to resume.
|
||||
// 2. Pop a runnable pid. Pop order (RFC 005): wake slot first, then
|
||||
// shared queue. The queue mutex covers ONLY the pop; the slot's
|
||||
// own atomics carry everything needed to resume.
|
||||
// ----------------------------------------------------------------
|
||||
enum Pop {
|
||||
Got(Pid),
|
||||
@@ -1069,6 +1178,24 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||||
AllDone,
|
||||
}
|
||||
|
||||
// 2a. RFC 005: drain this thread's wake slot before touching the
|
||||
// shared queue. Two consequences fall out of slot-first order:
|
||||
// the idle path below is only reachable with an empty slot, and so
|
||||
// is AllDone — an occupied slot on ANOTHER thread holds a Queued
|
||||
// (hence live) actor, so `live_actors > 0` and termination cannot
|
||||
// fire; the counter-first argument is untouched.
|
||||
let slot_pid = if inner.wake_slot {
|
||||
WAKE_SLOT.with(|s| s.take())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let from_slot = slot_pid.is_some();
|
||||
|
||||
let pid = if let Some(pid) = slot_pid {
|
||||
stats.slot_hits.fetch_add(1, Ordering::Relaxed);
|
||||
crate::te!(crate::trace::Event::SlotPop(pid));
|
||||
pid
|
||||
} else {
|
||||
// Read IO liveness BEFORE the queue lock (phase-1 ordering: a
|
||||
// completion resurrects an actor only via the drain path, whose
|
||||
// enqueue would be visible under the queue lock we take next).
|
||||
@@ -1103,7 +1230,7 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||||
}
|
||||
};
|
||||
|
||||
let pid = match pop {
|
||||
match pop {
|
||||
Pop::Got(pid) => pid,
|
||||
Pop::AllDone => {
|
||||
// Remaining timer entries are orphaned (no live actor can be
|
||||
@@ -1140,6 +1267,7 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// ----------------------------------------------------------------
|
||||
@@ -1172,7 +1300,18 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||||
crate::preempt::set_current_stop(stop_flag);
|
||||
reset_actor_done();
|
||||
YIELD_INTENT.with(|c| c.set(YieldIntent::Yield));
|
||||
// RFC 005 timeslice inheritance: a slot-popped actor does NOT get a
|
||||
// fresh slice — it inherits the waker's remaining one (this thread's
|
||||
// TIMESLICE_START/ALLOC_COUNT carry over from the waker's run, with
|
||||
// only scheduler bookkeeping in between). A chain of slot handoffs
|
||||
// is therefore collectively bounded by one slice, after which
|
||||
// preemption fires and the preempt-yield re-enqueue goes to the
|
||||
// SHARED queue (a yield is not a wake — never slot-eligible). The
|
||||
// shared queue is thus consulted at least once per slice per
|
||||
// scheduler: the one-slice starvation bound, zero new counters.
|
||||
if !from_slot {
|
||||
crate::preempt::reset_timeslice();
|
||||
}
|
||||
PREEMPTION_ENABLED.with(|c| c.set(true));
|
||||
|
||||
crate::te!(crate::trace::Event::Resume(pid));
|
||||
|
||||
@@ -58,6 +58,9 @@ mod inner {
|
||||
// Queue
|
||||
Enqueue(Pid),
|
||||
Dequeue(Pid),
|
||||
// RFC 005 wake slot
|
||||
SlotPush(Pid), // actor-context wake parked in the waking thread's slot
|
||||
SlotPop(Pid), // scheduler resumed a pid from its own slot
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
@@ -237,6 +240,8 @@ mod inner {
|
||||
Event::RecvWake(p) => ("recv_wake".into(), p.index()),
|
||||
Event::Enqueue(p) => ("enqueue".into(), p.index()),
|
||||
Event::Dequeue(p) => ("dequeue".into(), p.index()),
|
||||
Event::SlotPush(p) => ("slot_push".into(), p.index()),
|
||||
Event::SlotPop(p) => ("slot_pop".into(), p.index()),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
//! RFC 005 wake-slot tests: correctness with the slot on, push-policy
|
||||
//! discrimination (actor vs scheduler context, spawns), displacement, the
|
||||
//! default-off contract, and per-run counter reset.
|
||||
//!
|
||||
//! The slot is same-thread plain ops behind the existing queue-op contract,
|
||||
//! so there is nothing new to model-check (RFC 005 §Summary); these tests
|
||||
//! pin the *policy* — who lands in the slot, who never does — which is
|
||||
//! observable through the `slot_hits` / `slot_displacements` counters.
|
||||
|
||||
use smarm::channel::channel;
|
||||
use smarm::runtime::{init, Config};
|
||||
use smarm::spawn;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Ping-pong over channels: the slot's home pattern. Returns total roundtrips.
|
||||
fn ping_pong(pairs: usize, roundtrips: u64) -> impl FnOnce() + Send + 'static {
|
||||
move || {
|
||||
let handles: Vec<_> = (0..pairs)
|
||||
.map(|_| {
|
||||
spawn(move || {
|
||||
let (tx_ab, rx_ab) = channel::<u64>();
|
||||
let (tx_ba, rx_ba) = channel::<u64>();
|
||||
let echo = spawn(move || {
|
||||
for _ in 0..roundtrips {
|
||||
let v = rx_ab.recv().expect("echo recv");
|
||||
tx_ba.send(v + 1).expect("echo send");
|
||||
}
|
||||
});
|
||||
for i in 0..roundtrips {
|
||||
tx_ab.send(i).expect("ping send");
|
||||
assert_eq!(rx_ba.recv().expect("ping recv"), i + 1);
|
||||
}
|
||||
let _ = echo.join();
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
for h in handles {
|
||||
h.join().expect("pair");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Correctness: messaging workloads complete with the slot on, 1 and N threads
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn ping_pong_correct_with_slot_on_single_thread() {
|
||||
let rt = init(Config::exact(1).wake_slot(true));
|
||||
rt.run(ping_pong(4, 200));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ping_pong_correct_with_slot_on_multi_thread() {
|
||||
let rt = init(Config::exact(4).wake_slot(true));
|
||||
rt.run(ping_pong(8, 200));
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Push policy: actor-context wakes hit the slot; the default is off
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn messaging_workload_exercises_the_slot() {
|
||||
let rt = init(Config::exact(1).wake_slot(true));
|
||||
rt.run(ping_pong(2, 100));
|
||||
let stats = rt.stats();
|
||||
assert!(
|
||||
stats.slot_hits() > 0,
|
||||
"send-wakes are actor-context unparks — the slot must see hits, got 0"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn slot_off_means_zero_slot_traffic() {
|
||||
// Off explicitly…
|
||||
let rt = init(Config::exact(1).wake_slot(false));
|
||||
rt.run(ping_pong(2, 100));
|
||||
assert_eq!(rt.stats().slot_hits(), 0);
|
||||
assert_eq!(rt.stats().slot_displacements(), 0);
|
||||
|
||||
// …and off by default (RFC 005: default off until the shootout accepts).
|
||||
let rt = init(Config::exact(1));
|
||||
rt.run(ping_pong(2, 100));
|
||||
assert_eq!(rt.stats().slot_hits(), 0, "wake_slot must default to OFF");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Push policy: spawns and join-wakes (finalize = scheduler context) bypass
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn spawn_join_workload_bypasses_the_slot() {
|
||||
let rt = init(Config::exact(1).wake_slot(true));
|
||||
rt.run(|| {
|
||||
for _ in 0..20 {
|
||||
let handles: Vec<_> = (0..50).map(|_| spawn(|| {})).collect();
|
||||
for h in handles {
|
||||
h.join().expect("trivial actor");
|
||||
}
|
||||
}
|
||||
});
|
||||
let stats = rt.stats();
|
||||
assert_eq!(
|
||||
stats.slot_hits(),
|
||||
0,
|
||||
"spawns go shared by policy and joiner wakes fire from finalize \
|
||||
(scheduler context) — a pure spawn/join workload must never touch \
|
||||
the slot"
|
||||
);
|
||||
assert_eq!(stats.slot_displacements(), 0);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Displacement: newest wake takes the slot, occupant goes shared — and runs
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn displaced_occupant_reaches_the_shared_queue_and_runs() {
|
||||
// Single thread, strict FIFO (rq-mutex default): rx1 and rx2 park on
|
||||
// their recvs before the sender runs; the sender's back-to-back sends
|
||||
// then produce two actor-context wakes — the second displaces the first.
|
||||
let rt = init(Config::exact(1).wake_slot(true));
|
||||
let ran = Arc::new(AtomicU64::new(0));
|
||||
let (r1, r2) = (ran.clone(), ran.clone());
|
||||
rt.run(move || {
|
||||
let (tx1, rx1) = channel::<u32>();
|
||||
let (tx2, rx2) = channel::<u32>();
|
||||
let h1 = spawn(move || {
|
||||
assert_eq!(rx1.recv().expect("rx1"), 1);
|
||||
r1.fetch_add(1, Ordering::Relaxed);
|
||||
});
|
||||
let h2 = spawn(move || {
|
||||
assert_eq!(rx2.recv().expect("rx2"), 2);
|
||||
r2.fetch_add(1, Ordering::Relaxed);
|
||||
});
|
||||
let sender = spawn(move || {
|
||||
tx1.send(1).expect("send 1"); // rx1 → slot
|
||||
tx2.send(2).expect("send 2"); // rx2 → slot, rx1 displaced → shared
|
||||
});
|
||||
sender.join().expect("sender");
|
||||
h1.join().expect("h1");
|
||||
h2.join().expect("h2");
|
||||
});
|
||||
assert_eq!(ran.load(Ordering::Relaxed), 2, "both receivers must run");
|
||||
let stats = rt.stats();
|
||||
assert!(
|
||||
stats.slot_displacements() >= 1,
|
||||
"back-to-back wakes of parked receivers must displace at least once"
|
||||
);
|
||||
assert!(stats.slot_hits() >= 1, "the displacing wake is slot-popped");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Counters reset at the start of each run() on a reused Runtime
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn slot_counters_reset_per_run() {
|
||||
let rt = init(Config::exact(1).wake_slot(true));
|
||||
rt.run(ping_pong(2, 100));
|
||||
let first = rt.stats().slot_hits();
|
||||
assert!(first > 0);
|
||||
|
||||
// A slot-bypassing run on the same handle must read 0, not `first`.
|
||||
rt.run(|| {
|
||||
let h = spawn(|| {});
|
||||
h.join().expect("trivial");
|
||||
});
|
||||
assert_eq!(
|
||||
rt.stats().slot_hits(),
|
||||
0,
|
||||
"counters are reset at run() start; the second run had no slot traffic"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user