diff --git a/benches/rq_runtime.rs b/benches/rq_runtime.rs index 75d254d..fef71d6 100644 --- a/benches/rq_runtime.rs +++ b/benches/rq_runtime.rs @@ -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,,,,,, +//! RQCSV,runtime,,,,,,, +//! plus, for slot-on configs, the RFC 005 observability counters: +//! RQSLOT,,,,, +//! (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 { .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 { + 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 (u64, u128)>); + type Bench = (&'static str, String, Box Sample>); let benches: Vec = 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 = (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 - ); + for &slot in &slot_sweep { + let mut samples: Vec = (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} | {:>4} | {:>16} | {:>10} | {:>14} | {:>10} | {:>9}", + name, t, slot_str, work, mid.us, per_s, mid.hits, mid.displacements + ); + println!( + "RQCSV,runtime,{},{},{},{},{},{},{}", + variant(), slot_str, name, t, work, mid.us, per_s + ); + if slot { + println!( + "RQSLOT,{},{},{},{},{}", + variant(), name, t, mid.hits, mid.displacements + ); + } + } } } } diff --git a/scripts/bench_rq.sh b/scripts/bench_rq.sh index 2edf50a..1854db4 100755 --- a/scripts/bench_rq.sh +++ b/scripts/bench_rq.sh @@ -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)" diff --git a/src/run_queue.rs b/src/run_queue.rs index 91775f7..18a4e1e 100644 --- a/src/run_queue.rs +++ b/src/run_queue.rs @@ -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 diff --git a/src/runtime.rs b/src/runtime.rs index 2b428d2..b693274 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -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 { 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,7 +612,20 @@ impl RuntimeInner { match slot.word.unpark(pid.generation(), want) { Unpark::Enqueue => { crate::te!(crate::trace::Event::UnparkDirect(pid)); - self.enqueue(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 = 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> = const { Cell::new(None) }; + /// What the actor wants when it yields back to the scheduler. static YIELD_INTENT: Cell = const { Cell::new(YieldIntent::Yield) }; } @@ -1060,8 +1168,9 @@ fn schedule_loop(inner: &Arc, 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,76 +1178,95 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { AllDone, } - // 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). - let (io_out, io_fd) = match inner.io.lock().unwrap().as_ref() { - Some(io) => (io.outstanding + io.waiters.len() as u32, Some(io.wake_fd())), - None => (0, None), + // 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(); - stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed); - let pop = match inner.run_queue.pop() { - Some(pid) => Pop::Got(pid), - None => { - // Termination does not lean on pop-None being a fence (with - // the ring queues it is only a snapshot). The argument is - // counter-first: every queue entry's target stays `Queued` — - // hence un-finalized, hence counted live — until that very - // entry is popped. So `live == 0` (Acquire, pairing with - // finalize's Release decrement, which strictly follows all - // wakeup enqueues) by itself implies no entry is in, or can - // ever again enter, the queue: enqueues only target live - // actors, and a spawner is itself live. The pop-None above - // is then just the cheap fast-path filter; io_out was read - // before it per the phase-1 ordering. `live == 0` is also - // final — no spawn can resurrect the count — so every - // scheduler thread independently reaches this same verdict. - let live = inner.live_actors.load(Ordering::Acquire); - if live == 0 && io_out == 0 { - Pop::AllDone - } else { - Pop::Idle { io_outstanding: io_out, wake_fd: io_fd } + 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). + let (io_out, io_fd) = match inner.io.lock().unwrap().as_ref() { + Some(io) => (io.outstanding + io.waiters.len() as u32, Some(io.wake_fd())), + None => (0, None), + }; + + stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed); + let pop = match inner.run_queue.pop() { + Some(pid) => Pop::Got(pid), + None => { + // Termination does not lean on pop-None being a fence (with + // the ring queues it is only a snapshot). The argument is + // counter-first: every queue entry's target stays `Queued` — + // hence un-finalized, hence counted live — until that very + // entry is popped. So `live == 0` (Acquire, pairing with + // finalize's Release decrement, which strictly follows all + // wakeup enqueues) by itself implies no entry is in, or can + // ever again enter, the queue: enqueues only target live + // actors, and a spawner is itself live. The pop-None above + // is then just the cheap fast-path filter; io_out was read + // before it per the phase-1 ordering. `live == 0` is also + // final — no spawn can resurrect the count — so every + // scheduler thread independently reaches this same verdict. + let live = inner.live_actors.load(Ordering::Acquire); + if live == 0 && io_out == 0 { + Pop::AllDone + } else { + Pop::Idle { io_outstanding: io_out, wake_fd: io_fd } + } } - } - }; + }; - let pid = match pop { - Pop::Got(pid) => pid, - Pop::AllDone => { - // Remaining timer entries are orphaned (no live actor can be - // woken by them — e.g. a sleeper cancelled out of its sleep); - // they must not keep the runtime alive. Drop them on the way out. - inner.timers.lock().unwrap().clear(); - return; - } - Pop::Idle { io_outstanding, wake_fd } => { - // Something is still in flight. Sleep on the appropriate - // source to avoid hammering the queue mutex; retry on wake. - let next_deadline = inner.timers.lock().unwrap().peek_deadline(); - match (next_deadline, wake_fd) { - (Some(deadline), fd_opt) => { - let now = std::time::Instant::now(); - if deadline > now { - let timeout = deadline - now; - match fd_opt { - Some(fd) => { - crate::io::poll_wake(fd, Some(timeout)); - crate::io::drain_wake_pipe(fd); + match pop { + Pop::Got(pid) => pid, + Pop::AllDone => { + // Remaining timer entries are orphaned (no live actor can be + // woken by them — e.g. a sleeper cancelled out of its sleep); + // they must not keep the runtime alive. Drop them on the way out. + inner.timers.lock().unwrap().clear(); + return; + } + Pop::Idle { io_outstanding, wake_fd } => { + // Something is still in flight. Sleep on the appropriate + // source to avoid hammering the queue mutex; retry on wake. + let next_deadline = inner.timers.lock().unwrap().peek_deadline(); + match (next_deadline, wake_fd) { + (Some(deadline), fd_opt) => { + let now = std::time::Instant::now(); + if deadline > now { + let timeout = deadline - now; + match fd_opt { + Some(fd) => { + crate::io::poll_wake(fd, Some(timeout)); + crate::io::drain_wake_pipe(fd); + } + None => thread::sleep(timeout), } - None => thread::sleep(timeout), } } + (None, Some(fd)) if io_outstanding > 0 => { + crate::io::poll_wake(fd, None); + crate::io::drain_wake_pipe(fd); + } + _ => { + thread::sleep(std::time::Duration::from_micros(100)); + } } - (None, Some(fd)) if io_outstanding > 0 => { - crate::io::poll_wake(fd, None); - crate::io::drain_wake_pipe(fd); - } - _ => { - thread::sleep(std::time::Duration::from_micros(100)); - } + continue; } - continue; } }; @@ -1172,7 +1300,18 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { crate::preempt::set_current_stop(stop_flag); reset_actor_done(); YIELD_INTENT.with(|c| c.set(YieldIntent::Yield)); - crate::preempt::reset_timeslice(); + // 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)); diff --git a/src/trace.rs b/src/trace.rs index 592b040..8296456 100644 --- a/src/trace.rs +++ b/src/trace.rs @@ -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()), } } diff --git a/tests/wake_slot.rs b/tests/wake_slot.rs new file mode 100644 index 0000000..e51ffdd --- /dev/null +++ b/tests/wake_slot.rs @@ -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::(); + let (tx_ba, rx_ba) = channel::(); + 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::(); + let (tx2, rx2) = channel::(); + 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" + ); +}