Files
smarm/tests/wake_slot.rs
Claude 2708042990 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).
2026-06-11 20:20:07 +02:00

177 lines
6.4 KiB
Rust

//! 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"
);
}