diff --git a/src/actor.rs b/src/actor.rs index c1e61a2..38bb7f7 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -122,9 +122,9 @@ pub struct Actor { /// The PID this actor was assigned at spawn time. pub pid: Pid, /// The stack the actor runs on. Dropped (munmap'd) when the actor dies. + /// (The saved stack pointer lives on the `Slot` as an atomic, not here: + /// it is hot scheduling state, read/written without the cold lock.) pub stack: Stack, - /// The saved stack pointer. Updated on every yield. - pub sp: usize, /// The PID of this actor's supervisor. Used to deliver `Signal` on death. pub supervisor: Pid, /// Cooperative-cancellation flag. `request_stop` sets it (and unparks a diff --git a/src/lib.rs b/src/lib.rs index 3faa229..d92d4ad 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,6 +26,7 @@ pub mod monitor; pub mod link; pub mod gen_server; pub mod runtime; +pub(crate) mod raw_mutex; pub mod trace; // --------------------------------------------------------------------------- diff --git a/src/link.rs b/src/link.rs index 65f6925..c56cc16 100644 --- a/src/link.rs +++ b/src/link.rs @@ -49,10 +49,9 @@ //! [`Down`]: crate::monitor::Down //! [`request_stop`]: crate::scheduler::request_stop -use crate::channel::{channel, Receiver, Sender}; +use crate::channel::{channel, Receiver}; use crate::monitor::DownReason; use crate::pid::Pid; -use crate::runtime::State; use crate::scheduler::{request_stop, self_pid, with_runtime}; /// A linked peer's death, delivered to a trapping actor's inbox. @@ -78,11 +77,13 @@ pub fn trap_exit() -> Receiver { let (tx, rx) = channel::(); let me = self_pid(); with_runtime(|inner| { - inner.with_shared(|s| { - if let Some(actor) = s.slot_mut(me).and_then(|slot| slot.actor.as_mut()) { + if let Some(slot) = inner.slot_at(me) { + let mut cold = slot.cold.lock(); + // Own slot: generation is necessarily current (we're running). + if let Some(actor) = cold.actor.as_mut() { actor.trap = Some(tx); } - }) + } }); rx } @@ -100,45 +101,61 @@ pub fn link(target: Pid) { return; } - // Under the lock: if the target is live, record the link both ways and - // return `None`. If it is gone, return the caller's trap sender (if any) - // so we can deliver the NoProc signal after releasing the lock. - let dead_action: Option>> = with_runtime(|inner| { - inner.with_shared(|s| { - let target_live = matches!( - s.slot(target), - Some(slot) if slot.actor.is_some() && !matches!(slot.state, State::Done) - ); - if target_live { - if let Some(slot) = s.slot_mut(me) { - if !slot.links.contains(&target) { - slot.links.push(target); - } + // Cold locks are leaves: never hold two at once. The link is recorded one + // side at a time, TARGET FIRST — that ordering is what makes the race + // window sound: + // + // - Once `me` is in `target.links`, the target's death always reaches us + // (its finalize cascade walks that list). So after step 1 succeeds, the + // link semantics are already live. + // - If the target dies between step 1 and step 2, its cascade removes + // `target` from OUR links (a no-op, we haven't added it yet) and + // delivers the exit signal — correct, the link was established. Our + // subsequent step-2 insert leaves a stale `target` entry in `me.links`; + // stale entries are benign by construction (every cascade walk + // re-verifies the peer's word; `unlink` removes them like any other). + // + // The reverse order would be unsound: target dying in the window would + // walk its links WITHOUT us — a silently dead link that we believe is live. + let registered_on_target = with_runtime(|inner| match inner.slot_at(target) { + Some(slot) => { + let mut cold = slot.cold.lock(); + if slot.is_live_for(target) && cold.actor.is_some() { + if !cold.links.contains(&me) { + cold.links.push(me); } - if let Some(slot) = s.slot_mut(target) { - if !slot.links.contains(&me) { - slot.links.push(me); - } - } - None + true } else { - // Grab our own trap sender so the NoProc delivery (below) - // doesn't need a second lock acquisition. - Some( - s.slot(me) - .and_then(|slot| slot.actor.as_ref()) - .and_then(|a| a.trap.clone()), - ) + false } - }) + } + None => false, }); - match dead_action { - None => {} // linked successfully - Some(Some(tx)) => { + if registered_on_target { + with_runtime(|inner| { + let slot = inner.slot_at(me).expect("link: own slot vanished"); + let mut cold = slot.cold.lock(); + if !cold.links.contains(&target) { + cold.links.push(target); + } + }); + return; + } + + // Target already gone: deliver NoProc to ourselves — as a message if + // trapping, otherwise as a cooperative stop. + let my_trap = with_runtime(|inner| { + inner.slot_at(me).and_then(|slot| { + let cold = slot.cold.lock(); + cold.actor.as_ref().and_then(|a| a.trap.clone()) + }) + }); + match my_trap { + Some(tx) => { let _ = tx.send(ExitSignal { from: target, reason: DownReason::NoProc }); } - Some(None) => request_stop(me), + None => request_stop(me), } } @@ -152,13 +169,18 @@ pub fn unlink(target: Pid) { return; } with_runtime(|inner| { - inner.with_shared(|s| { - if let Some(slot) = s.slot_mut(me) { - slot.links.retain(|p| *p != target); + // One cold lock at a time (leaf rule). Order is immaterial here: + // a half-removed link is just a stale entry on one side, and stale + // entries are benign (re-verified on every cascade walk). + if let Some(slot) = inner.slot_at(me) { + let mut cold = slot.cold.lock(); + cold.links.retain(|p| *p != target); + } + if let Some(slot) = inner.slot_at(target) { + let mut cold = slot.cold.lock(); + if slot.generation() == target.generation() { + cold.links.retain(|p| *p != me); } - if let Some(slot) = s.slot_mut(target) { - slot.links.retain(|p| *p != me); - } - }) + } }); } diff --git a/src/monitor.rs b/src/monitor.rs index ce473fd..d6eb158 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -47,7 +47,6 @@ use crate::channel::{channel, Receiver, Sender}; use crate::pid::Pid; -use crate::runtime::State; use crate::scheduler::with_runtime; /// Why a monitored actor went down. @@ -110,23 +109,24 @@ pub struct Monitor { pub fn monitor(target: Pid) -> Monitor { let (tx, rx) = channel::(); - // Register under the shared lock. We allocate the id and (if the target is - // live) clone the sender into its monitor list, keeping the original `tx` - // for the NoProc fallback. `tx.clone()` only touches the channel's own - // mutex, never the shared runtime mutex, so it is safe under the lock — but - // we must not *send* here, as `Sender::send` can call back in to unpark a - // parked receiver and the shared mutex is not reentrant. + // Register under the target's cold lock. `tx.clone()` only touches the + // channel's own mutex, never any runtime lock, so it is safe here — but + // we must not *send* under the lock, as `Sender::send` can unpark a + // parked receiver, and there's no reason to nest that. let (id, registered) = with_runtime(|inner| { let id = inner.alloc_monitor_id(); - let registered = inner.with_shared(|s| { - match s.slot_mut(target) { - Some(slot) if !matches!(slot.state, State::Done) => { - slot.monitors.push((id, tx.clone())); + let registered = match inner.slot_at(target) { + Some(slot) => { + let mut cold = slot.cold.lock(); + if slot.is_live_for(target) { + cold.monitors.push((id, tx.clone())); true + } else { + false } - _ => false, } - }); + None => false, + }; (id, registered) }); @@ -147,16 +147,18 @@ pub fn monitor(target: Pid) -> Monitor { /// dropping the [`Monitor`] closes its receiver and the queued notice goes with /// it — the analogue of Erlang's `demonitor(Ref, [flush])`. pub fn demonitor(m: &Monitor) -> Option { - // Remove the registration under the lock, but move the `Sender` *out* and - // let it drop only after the lock is released: dropping the last sender - // runs `Sender::drop`, which may unpark a parked receiver → `with_shared`, - // and the shared mutex is not reentrant. + // Remove the registration under the target's cold lock, but move the + // `Sender` *out* and let it drop only after the lock is released: + // dropping the last sender runs `Sender::drop`, which may unpark a parked + // receiver — legal under a cold lock, but pointless to nest. let removed: Option<(MonitorId, Sender)> = with_runtime(|inner| { - inner.with_shared(|s| { - let slot = s.slot_mut(m.target)?; - let pos = slot.monitors.iter().position(|(mid, _)| *mid == m.id)?; - Some(slot.monitors.remove(pos)) - }) + let slot = inner.slot_at(m.target)?; + let mut cold = slot.cold.lock(); + if slot.generation() != m.target.generation() { + return None; // slot reused; the Down already fired + } + let pos = cold.monitors.iter().position(|(mid, _)| *mid == m.id)?; + Some(cold.monitors.remove(pos)) }); // `removed`'s sender drops here, outside the lock. removed.map(|(id, _sender)| id) diff --git a/src/raw_mutex.rs b/src/raw_mutex.rs new file mode 100644 index 0000000..d6861ad --- /dev/null +++ b/src/raw_mutex.rs @@ -0,0 +1,228 @@ +//! A minimal futex-based mutex that cannot poison. +//! +//! `std::sync::Mutex` poisons on unwind, turning one panic into a cascade of +//! `lock().unwrap()` panics in every later user. The runtime's internal +//! critical sections must never unwind anyway (the stop sentinel is gated +//! behind `PREEMPTION_ENABLED`, and the guard below disables preemption), so +//! poisoning buys nothing and costs a failure mode. This mutex has no poison +//! state by construction. +//! +//! Two further properties the runtime wants: +//! +//! - **The guard enters `NoPreempt`.** A timeslice switch while holding an OS +//! mutex would suspend the actor with the lock held, stalling every other +//! OS thread that touches it until the actor is resumed. Disabling +//! preemption for the (short) critical section keeps lock hold times +//! bounded. It also closes the unwind hole structurally: with +//! `PREEMPTION_ENABLED` false, `maybe_preempt` neither yields nor raises +//! the stop sentinel, so no allocation inside the critical section can +//! unwind it. +//! - **No std machinery.** One `AtomicU32` and two futex syscalls; friendlier +//! to an eventual embedded port than `std::sync::Mutex` (swap the futex for +//! a spin or WFE backend). +//! +//! Algorithm: the classic three-state futex mutex (Drepper, "Futexes Are +//! Tricky", mutex3). 0 = unlocked, 1 = locked, 2 = locked with (possible) +//! waiters. Uncontended lock/unlock is one CAS / one swap, no syscall. +//! +//! Lock-order position: slot cold locks, the free list, and the stack pool +//! are all `RawMutex`es and all *leaves among themselves* — never hold two at +//! once. Holding a `RawMutex` while taking the run-queue mutex (`with_shared`) +//! is permitted (e.g. unpark from inside a cold section); the reverse — +//! taking any `RawMutex` from inside `with_shared` — is forbidden. + +use std::cell::UnsafeCell; +use std::ops::{Deref, DerefMut}; +use std::sync::atomic::{AtomicU32, Ordering}; + +const UNLOCKED: u32 = 0; +const LOCKED: u32 = 1; +const CONTENDED: u32 = 2; + +/// How many `pause` spins to burn before falling back to the futex. Critical +/// sections under this lock are tens of nanoseconds (push to a Vec, clone a +/// sender), so a short spin almost always avoids the syscall. +const SPIN_LIMIT: u32 = 64; + +pub(crate) struct RawMutex { + state: AtomicU32, + data: UnsafeCell, +} + +// SAFETY: standard mutex argument — exclusive access to `data` is mediated by +// `state`; `T: Send` suffices for both because `&RawMutex` only ever hands out +// access to one thread at a time. +unsafe impl Send for RawMutex {} +unsafe impl Sync for RawMutex {} + +impl RawMutex { + pub(crate) const fn new(data: T) -> Self { + Self { + state: AtomicU32::new(UNLOCKED), + data: UnsafeCell::new(data), + } + } + + #[inline] + pub(crate) fn lock(&self) -> RawMutexGuard<'_, T> { + // Enter NoPreempt *before* acquiring, so a preemption can't fire + // between acquisition and guard construction. + let prev_preempt = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); + if self + .state + .compare_exchange(UNLOCKED, LOCKED, Ordering::Acquire, Ordering::Relaxed) + .is_err() + { + self.lock_slow(); + } + RawMutexGuard { m: self, prev_preempt } + } + + #[cold] + fn lock_slow(&self) { + // Bounded spin first: the expected hold time is far below the cost of + // a futex round trip. + let mut spins = 0; + loop { + let s = self.state.load(Ordering::Relaxed); + if s == UNLOCKED + && self + .state + .compare_exchange_weak(UNLOCKED, LOCKED, Ordering::Acquire, Ordering::Relaxed) + .is_ok() + { + return; + } + spins += 1; + if spins >= SPIN_LIMIT { + break; + } + std::hint::spin_loop(); + } + // Futex path. Mark contended and sleep until woken; on wake, retake + // by swapping to CONTENDED (we cannot know whether other waiters + // remain, so we must conservatively keep the contended marker). + while self.state.swap(CONTENDED, Ordering::Acquire) != UNLOCKED { + futex_wait(&self.state, CONTENDED); + } + } + + #[inline] + fn unlock(&self) { + if self.state.swap(UNLOCKED, Ordering::Release) == CONTENDED { + futex_wake(&self.state, 1); + } + } +} + +pub(crate) struct RawMutexGuard<'a, T> { + m: &'a RawMutex, + prev_preempt: bool, +} + +impl Deref for RawMutexGuard<'_, T> { + type Target = T; + #[inline] + fn deref(&self) -> &T { + // SAFETY: guard existence implies exclusive ownership of the lock. + unsafe { &*self.m.data.get() } + } +} + +impl DerefMut for RawMutexGuard<'_, T> { + #[inline] + fn deref_mut(&mut self) -> &mut T { + // SAFETY: as above, plus &mut self. + unsafe { &mut *self.m.data.get() } + } +} + +impl Drop for RawMutexGuard<'_, T> { + #[inline] + fn drop(&mut self) { + self.m.unlock(); + // Restore preemption only after the lock is released. + crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(self.prev_preempt)); + } +} + +// --------------------------------------------------------------------------- +// futex (x86-64 Linux; master is x86-only, see arm-port branch) +// --------------------------------------------------------------------------- + +fn futex_wait(state: &AtomicU32, expected: u32) { + // SAFETY: `state` is a valid, aligned u32 for the duration of the call. + // Spurious wakeups and EAGAIN (value already changed) are both handled by + // the caller's retry loop. + unsafe { + libc::syscall( + libc::SYS_futex, + state.as_ptr(), + libc::FUTEX_WAIT | libc::FUTEX_PRIVATE_FLAG, + expected, + std::ptr::null::(), + ); + } +} + +fn futex_wake(state: &AtomicU32, n: i32) { + // SAFETY: as above. + unsafe { + libc::syscall( + libc::SYS_futex, + state.as_ptr(), + libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG, + n, + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + + #[test] + fn uncontended_lock_unlock() { + let m = RawMutex::new(0u64); + for _ in 0..1000 { + *m.lock() += 1; + } + assert_eq!(*m.lock(), 1000); + } + + #[test] + fn contended_counter_is_exact() { + const THREADS: usize = 8; + const PER: u64 = 50_000; + let m = Arc::new(RawMutex::new(0u64)); + let hs: Vec<_> = (0..THREADS) + .map(|_| { + let m = m.clone(); + std::thread::spawn(move || { + for _ in 0..PER { + *m.lock() += 1; + } + }) + }) + .collect(); + for h in hs { + h.join().unwrap(); + } + assert_eq!(*m.lock(), THREADS as u64 * PER); + } + + #[test] + fn no_poison_on_unwind() { + let m = Arc::new(RawMutex::new(0u64)); + let m2 = m.clone(); + let _ = std::thread::spawn(move || { + let _g = m2.lock(); + panic!("unwind while holding"); + }) + .join(); + // A std Mutex would now be poisoned; this one just works. + *m.lock() += 1; + assert_eq!(*m.lock(), 1); + } +} diff --git a/src/runtime.rs b/src/runtime.rs index 974bfa7..fe26801 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,50 +1,115 @@ //! Multi-scheduler runtime: configuration, initialisation, and the shared //! state that all scheduler OS threads operate against. //! -//! # Architecture +//! # Architecture (post slot-table split, ROADMAP_v0.5 phase 2) //! //! ```text //! init(Config) → Runtime (Arc) //! //! RuntimeInner { -//! shared: Mutex ← slot table, run queue, timers, IO +//! slots: Box<[Slot]> ← FIXED slab, max_actors entries, lock-free lookup +//! free: RawMutex> ← vacant slot indices +//! shared: Mutex ← the run queue (and nothing else) +//! timers: Mutex +//! io: Mutex> +//! live_actors: AtomicU32 ← spawned-but-not-finalized count (termination) //! stats: Vec ← one per thread, lockless atomics (RFC 000) -//! io_parked: AtomicU32 ← actors parked on IO -//! sleeping: AtomicU32 ← actors parked on timer +//! } +//! +//! Slot { +//! word: AtomicU64 ← (generation << 32) | state — THE state machine +//! sp: AtomicUsize ← saved stack pointer +//! stop_ptr:AtomicPtr<...> ← into the actor's Arc +//! closure: AtomicPtr<...> ← first-resume closure, swap-to-take +//! cold: RawMutex ← lifecycle collections (waiters/monitors/links/…) //! } //! ``` //! -//! `Runtime::run(f)` spawns N OS threads (one per `Config::resolved_thread_count()`), -//! each running `schedule_loop`. It blocks until all scheduler threads exit, -//! i.e. until the run queue is empty and nothing is pending. +//! # The per-slot state machine //! -//! Each scheduler thread holds an `Arc` clone. Per-thread -//! identity is a small integer index, stored in a thread-local, used to index -//! into `stats`. +//! Scheduling state lives in one atomic word per slot packing +//! `(generation, state)`, where state is one of: +//! +//! ```text +//! Vacant ─spawn→ Queued ─pop→ Running ─yield→ Queued +//! ↑ │ │ +//! │ park │ unpark while running +//! │ ↓ ↓ +//! unpark ←──── Parked RunningNotified ─park→ Queued +//! (re-queued immediately) +//! Running|RunningNotified ─actor returns→ Done ─reclaim→ Vacant(gen+1) +//! ``` +//! +//! Every transition is a CAS on the packed word, so: +//! +//! - The generation check is **atomic with the transition** — a stale `Pid` +//! can never act on a recycled slot (no ABA, no spurious unparks). +//! - `RunningNotified` replaces the old `pending_unpark` bool: an unpark that +//! races the prep-to-park window is a *state*, resolved by the scheduler's +//! park-return CAS, not a flag read under a lock. This also closes a latent +//! lost-wakeup in the old Blocking-IO completion path, which set the result +//! for a still-Running actor without flagging it. +//! - **A pid is in the run queue at most once**: the only pushes are paired +//! 1:1 with successful transitions *into* `Queued`, and only the scheduler +//! transitions `Queued → Running` (paired 1:1 with pops). +//! +//! Memory ordering: all word CASes are `AcqRel` (failure `Acquire`), plain +//! word stores are `Release`, loads are `Acquire`. The chain that matters: +//! the park path stores `sp` (Relaxed) *before* its Release transition; any +//! later Acquire transition/load of the word therefore observes that `sp`. +//! The run-queue mutex independently provides the same edges today; the +//! word's own ordering is what phase 3's lock-free queue will rely on. +//! +//! # Locks and ordering +//! +//! - `shared` now guards **only the run queue**. It is the innermost lock: +//! nothing else is ever acquired while holding it. +//! - Per-slot `cold` locks ([`RawMutex`], non-poisoning, guard enters +//! `NoPreempt`) guard the lifecycle collections. **Leaf rule: never hold +//! two cold locks at once** — `finalize_actor`'s link cascade and `link()` +//! lock peers one at a time (correctness arguments at the call sites). +//! Holding a cold lock while pushing to the run queue is permitted. +//! - Lock order overall: `io` → (slot `cold` | `free` | `stack_pool`, +//! mutually leaf) → `shared`/run-queue. `timers` is independent (never +//! nested with any of the above on either side). +//! +//! # Termination (counter-based) +//! +//! The old all-clear scanned the slot table under the big lock. Now: +//! exit when `io_out == 0` (read *before* the queue lock, phase-1 ordering) +//! and, under the queue lock, the queue is empty and `live_actors == 0`. +//! `live_actors` is incremented in `spawn` before the enqueue and decremented +//! at the very END of `finalize_actor`, strictly after every wakeup that +//! finalize produces has been enqueued. The soundness crux: any enqueue +//! targets a live (not-yet-finalized) actor, so `live == 0` implies no wakeup +//! can still be in flight; combined with "spawner is itself live", observing +//! `(queue empty, live == 0)` under the queue lock means no work can ever +//! appear again. //! //! # Timer / IO drain (try-lock, one-winner) //! -//! On each loop iteration every scheduler thread tries `try_lock()` on a -//! separate `drain_lock: Mutex<()>`. The winner drains due timers and IO -//! completions; losers skip and move straight to popping an actor from the -//! run queue. This is the simplest correct approach; revisit if the drain -//! becomes a measured bottleneck. +//! Unchanged from phase 1: one winner per round drains due timers and IO +//! completions from their own mutexes; wakeups go through the unpark +//! protocol like everyone else's. use crate::actor::{ - clear_current_pid, is_actor_done, reset_actor_done, set_current_actor_box, + clear_current_pid, is_actor_done, reset_actor_done, set_current_actor_box, set_current_pid, take_last_outcome, Actor, Outcome, }; use crate::channel::Sender; -use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor}; use crate::io::IoThread; use crate::monitor::{Down, DownReason, MonitorId}; use crate::pid::Pid; use crate::preempt::PREEMPTION_ENABLED; +use crate::raw_mutex::RawMutex; use crate::supervisor::Signal; use crate::timer::Timers; +use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor}; use std::collections::VecDeque; -use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; +use std::sync::atomic::{ + AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering, +}; use std::sync::{Arc, Mutex}; use std::thread; @@ -52,6 +117,10 @@ use std::thread; // Config // --------------------------------------------------------------------------- +/// Default capacity of the actor slot table. Slots are ~256 bytes, so the +/// default costs ~4 MiB, allocated once at `init`. See [`Config::max_actors`]. +pub const DEFAULT_MAX_ACTORS: usize = 16_384; + /// Runtime configuration. /// /// ``` @@ -74,6 +143,7 @@ pub struct Config { alloc_interval: u32, timeslice_cycles: u64, stack_pool_cap: usize, + max_actors: usize, } impl Config { @@ -85,6 +155,7 @@ impl Config { alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL, timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES, stack_pool_cap: n * 4, + max_actors: DEFAULT_MAX_ACTORS, } } @@ -100,6 +171,7 @@ impl Config { alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL, timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES, stack_pool_cap: max * 4, + max_actors: DEFAULT_MAX_ACTORS, } } @@ -128,6 +200,22 @@ impl Config { self } + /// Capacity of the actor slot table — the maximum number of + /// **simultaneously live** actors (total spawned over a run is unbounded; + /// slots are recycled). The table is a fixed slab allocated once at + /// `init`: slots never move, which is what makes lock-free slot lookup + /// sound. Exhausting it is a loud panic naming this knob. + /// Default: [`DEFAULT_MAX_ACTORS`] (16_384, ~4 MiB). + pub fn max_actors(mut self, n: usize) -> Self { + assert!(n >= 1, "max_actors must be ≥ 1"); + assert!( + n < u32::MAX as usize, + "max_actors must fit a u32 slot index (ROOT_PID reserves u32::MAX)" + ); + self.max_actors = n; + self + } + /// The number of scheduler threads this config resolves to. pub fn resolved_thread_count(&self) -> usize { if let Some(e) = self.exact { @@ -150,6 +238,7 @@ impl Default for Config { alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL, timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES, stack_pool_cap: avail * 4, + max_actors: DEFAULT_MAX_ACTORS, } } } @@ -209,18 +298,39 @@ impl RuntimeStats { } // --------------------------------------------------------------------------- -// Shared state (behind Mutex<>) +// Slot — packed state word + hot atomics + cold lifecycle data // --------------------------------------------------------------------------- pub(crate) const ACTOR_STACK_SIZE: usize = 64 * 1024; -#[derive(Debug)] -pub(crate) enum State { Runnable, Parked, Done } +// State encodings for the low 8 bits of `Slot::word`. +pub(crate) const ST_VACANT: u64 = 0; +pub(crate) const ST_QUEUED: u64 = 1; +pub(crate) const ST_RUNNING: u64 = 2; +pub(crate) const ST_RUNNING_NOTIFIED: u64 = 3; +pub(crate) const ST_PARKED: u64 = 4; +pub(crate) const ST_DONE: u64 = 5; -pub(crate) struct Slot { - pub(crate) generation: u32, +#[inline] +pub(crate) const fn pack(gen: u32, st: u64) -> u64 { + ((gen as u64) << 32) | st +} +#[inline] +pub(crate) const fn word_gen(w: u64) -> u32 { + (w >> 32) as u32 +} +#[inline] +pub(crate) const fn word_state(w: u64) -> u64 { + w & 0xFF +} + +pub(crate) type Closure = Box; + +/// Lifecycle data, mutated only under the slot's cold [`RawMutex`]. Everything +/// here is touched O(1) times per actor lifetime (spawn / join / monitor / +/// link / finalize), never on the yield/park/unpark hot path. +pub(crate) struct SlotCold { pub(crate) actor: Option, - pub(crate) state: State, pub(crate) waiters: Vec, pub(crate) outcome: Option, pub(crate) supervisor_channel: Option>, @@ -230,81 +340,164 @@ pub(crate) struct Slot { /// Distinct from `supervisor_channel`, which is the parent's single funnel. pub(crate) monitors: Vec<(MonitorId, Sender)>, /// Bidirectional links (roadmap #3). Each entry is a peer whose abnormal - /// death propagates to this actor (and vice versa — the relationship is - /// recorded on both slots). Walked + cleared in `finalize_actor`. Reset in - /// all three slot-lifecycle sites, like `monitors`. + /// death propagates to this actor (and vice versa). Entries may be + /// momentarily or persistently stale (peer already dead) — every walk + /// re-verifies the peer's word, so stale entries are benign no-ops. pub(crate) links: Vec, pub(crate) outstanding_handles: u32, pub(crate) pending_io_result: Option, - /// Set by `unpark()` when the actor is still running (not yet Parked). - /// The scheduler checks this after a Park yield and re-queues instead - /// of sleeping, closing the lost-wakeup window. - pub(crate) pending_unpark: bool, - /// This actor's closure awaiting its first resume. `Some` until the - /// scheduler pops it on first dequeue, `None` thereafter. Previously a - /// parallel `pending_closures: Vec>` indexed by slot - /// index; folded into the slot since it is per-actor data with the same - /// lifetime as the slot and no independent access pattern. - pub(crate) pending_closure: Option, +} + +/// One actor slot. Hot scheduling state is atomic; cold lifecycle state is +/// behind `cold`. Slots live in a fixed slab and never move. +/// +/// `align(128)` keeps two adjacent slots' hot words off each other's +/// cache-line pair (x86 prefetches lines in pairs), avoiding false sharing +/// between unrelated actors. +#[repr(align(128))] +pub(crate) struct Slot { + /// `(generation << 32) | state`. See the module docs for the machine. + word: AtomicU64, + /// Saved stack pointer. Written by the owning scheduler thread before the + /// Release transition out of Running; read after the Acquire transition + /// Queued→Running. Relaxed is sufficient — ordering rides on `word`. + sp: AtomicUsize, + /// Pointer into the actor's `Arc` stop flag. Set at spawn, + /// nulled at finalize. The box outlives every read: it is only ever read + /// on the resume path while the actor cannot be finalized (it is on-CPU). + stop_ptr: AtomicPtr, + /// First-resume closure, double-boxed so it fits an `AtomicPtr` + /// (`Box` is a thin pointer). Swap-to-take; null when absent. + closure: AtomicPtr, + /// Cold lifecycle data. See [`SlotCold`]. + pub(crate) cold: RawMutex, } impl Slot { fn vacant() -> Self { Self { - generation: 0, - actor: None, - state: State::Done, - waiters: Vec::new(), - outcome: None, - supervisor_channel: None, - monitors: Vec::new(), - links: Vec::new(), - outstanding_handles: 0, - pending_io_result: None, - pending_unpark: false, - pending_closure: None, + word: AtomicU64::new(pack(0, ST_VACANT)), + sp: AtomicUsize::new(0), + stop_ptr: AtomicPtr::new(std::ptr::null_mut()), + closure: AtomicPtr::new(std::ptr::null_mut()), + cold: RawMutex::new(SlotCold { + actor: None, + waiters: Vec::new(), + outcome: None, + supervisor_channel: None, + monitors: Vec::new(), + links: Vec::new(), + outstanding_handles: 0, + pending_io_result: None, + }), + } + } + + #[inline] + pub(crate) fn word(&self) -> u64 { + self.word.load(Ordering::Acquire) + } + + /// Current generation (of whatever occupies the slot — pair with a state + /// check or a CAS before acting on it). + #[inline] + pub(crate) fn generation(&self) -> u32 { + word_gen(self.word()) + } + + /// Does the slot currently hold the actor `pid` names, in a non-terminal + /// state? (Snapshot — callers that mutate must re-verify under `cold` or + /// CAS on the word.) + #[inline] + pub(crate) fn is_live_for(&self, pid: Pid) -> bool { + let w = self.word(); + word_gen(w) == pid.generation() + && !matches!(word_state(w), ST_VACANT | ST_DONE) + } + + #[inline] + fn store_word(&self, gen: u32, st: u64) { + self.word.store(pack(gen, st), Ordering::Release); + } + + #[inline] + fn cas_word(&self, gen: u32, from: u64, to: u64) -> Result { + self.word.compare_exchange( + pack(gen, from), + pack(gen, to), + Ordering::AcqRel, + Ordering::Acquire, + ) + } + + fn store_closure(&self, c: Closure) { + let raw = Box::into_raw(Box::new(c)); + let prev = self.closure.swap(raw, Ordering::Release); + debug_assert!(prev.is_null(), "slot already had a pending closure"); + } + + fn take_closure(&self) -> Option { + let raw = self.closure.swap(std::ptr::null_mut(), Ordering::Acquire); + if raw.is_null() { + None + } else { + // SAFETY: non-null values in `closure` are exclusively + // `Box::into_raw(Box)` from `store_closure`, and the + // swap above made us the unique owner. + Some(*unsafe { Box::from_raw(raw) }) + } + } + + /// The unpark protocol — the one way anything outside the scheduler makes + /// an actor runnable. Returns `true` iff the caller must enqueue `pid`. + /// + /// Loops on the packed word: + /// - generation mismatch → stale pid → no-op (atomic with the CAS: a + /// recycled slot can never be transitioned by an old pid). + /// - `Parked` → `Queued`: caller enqueues. + /// - `Running` → `RunningNotified`: the actor is inside a prep-to-park + /// window (registered somewhere, not yet parked). The scheduler's + /// park-return CAS will fail against this and re-queue immediately — + /// the lost-wakeup window is a state, not a flag. + /// - `Queued` / `RunningNotified`: someone already notified — coalesce. + /// - `Done` / `Vacant`: nothing to wake. + #[must_use] + pub(crate) fn unpark_action(&self, pid: Pid) -> bool { + let gen = pid.generation(); + loop { + let w = self.word(); + if word_gen(w) != gen { + return false; + } + match word_state(w) { + ST_PARKED => { + if self.cas_word(gen, ST_PARKED, ST_QUEUED).is_ok() { + return true; + } + } + ST_RUNNING => { + if self.cas_word(gen, ST_RUNNING, ST_RUNNING_NOTIFIED).is_ok() { + crate::te!(crate::trace::Event::UnparkDeferred(pid)); + return false; + } + } + _ => return false, // Queued | RunningNotified | Done | Vacant + } } } } -pub(crate) type Closure = Box; +// --------------------------------------------------------------------------- +// Shared state (behind Mutex<>) — now: the run queue, nothing else +// --------------------------------------------------------------------------- pub(crate) struct SharedState { - pub(crate) slots: Vec, - pub(crate) free_list: Vec, pub(crate) run_queue: VecDeque, - pub(crate) root_pid: Option, } impl SharedState { fn new() -> Self { - Self { - slots: Vec::new(), - free_list: Vec::new(), - run_queue: VecDeque::new(), - root_pid: None, - } - } - - pub(crate) fn allocate_slot(&mut self) -> (u32, u32) { - if let Some(idx) = self.free_list.pop() { - let gen = self.slots[idx as usize].generation; - (idx, gen) - } else { - let idx = self.slots.len() as u32; - self.slots.push(Slot::vacant()); - (idx, 0) - } - } - - pub(crate) fn slot(&self, pid: Pid) -> Option<&Slot> { - let s = self.slots.get(pid.index() as usize)?; - if s.generation == pid.generation() { Some(s) } else { None } - } - - pub(crate) fn slot_mut(&mut self, pid: Pid) -> Option<&mut Slot> { - let s = self.slots.get_mut(pid.index() as usize)?; - if s.generation == pid.generation() { Some(s) } else { None } + Self { run_queue: VecDeque::new() } } } @@ -313,18 +506,22 @@ impl SharedState { // --------------------------------------------------------------------------- pub(crate) struct RuntimeInner { + /// The run queue. Innermost lock: nothing else is acquired under it. + /// Phase 3 swaps this for the compile-time-selected `RunQueue`. pub(crate) shared: Mutex, - /// Timer heap, peeled out of `shared` (RFC: state decomposition). Its own - /// mutex: only the drain-winner and the blocking primitives (sleep, wait) - /// touch it, never on the pure-compute hot path. + /// The fixed actor slot table. Allocated once; slots never move. + pub(crate) slots: Box<[Slot]>, + /// Vacant slot indices. RawMutex leaf; never held with a cold lock. + pub(crate) free: RawMutex>, + /// Spawned-but-not-finalized actor count; the termination criterion. + /// Incremented in `spawn` before the enqueue; decremented at the very end + /// of `finalize_actor`, after every wakeup finalize produces. + pub(crate) live_actors: AtomicU32, + /// Timer heap. Independent lock: never nested with any other. pub(crate) timers: Mutex, - /// IO subsystem, peeled out of `shared`. `None` between runs; `Some` for - /// the duration of a `run()`. Holds its own completion mutex internally; - /// this outer mutex guards only the `Option`/`IoThread` bookkeeping - /// (submit, epoll register/deregister, completion drain). + /// IO subsystem. `None` between runs. Lock order: io before everything. pub(crate) io: Mutex>, - /// Monotonic `MonitorId` source, peeled out of `shared`. Never reused, so - /// `demonitor` can name one of several monitors on a target unambiguously. + /// Monotonic `MonitorId` source. Never reused. pub(crate) next_monitor_id: AtomicU64, /// Try-lock: exactly one scheduler thread drains timers/IO per iteration. drain_lock: Mutex<()>, @@ -337,17 +534,28 @@ pub(crate) struct RuntimeInner { pub(crate) alloc_interval: u32, pub(crate) timeslice_cycles: u64, /// Recycled stacks waiting to be reused by the next spawn. - /// Grows like a Vec (doubles capacity); shrink policy is TBD. - pub(crate) stack_pool: Mutex>, + pub(crate) stack_pool: RawMutex>, /// Maximum number of stacks to retain in the pool. pub(crate) stack_pool_cap: usize, } impl RuntimeInner { - fn new(thread_count: usize, alloc_interval: u32, timeslice_cycles: u64, stack_pool_cap: usize) -> Arc { + fn new( + thread_count: usize, + alloc_interval: u32, + timeslice_cycles: u64, + stack_pool_cap: usize, + max_actors: usize, + ) -> Arc { let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect(); + let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect(); + // Low indices on top of the stack so early spawns get low pids. + let free: Vec = (0..max_actors as u32).rev().collect(); Arc::new(Self { shared: Mutex::new(SharedState::new()), + slots, + free: RawMutex::new(free), + live_actors: AtomicU32::new(0), timers: Mutex::new(Timers::new()), io: Mutex::new(None), next_monitor_id: AtomicU64::new(0), @@ -357,28 +565,72 @@ impl RuntimeInner { sleeping: AtomicU32::new(0), alloc_interval, timeslice_cycles, - stack_pool: Mutex::new(Vec::new()), + stack_pool: RawMutex::new(Vec::new()), stack_pool_cap, }) } pub(crate) fn with_shared(&self, f: impl FnOnce(&mut SharedState) -> R) -> R { - // Preemption must be off while we hold the shared mutex. If an actor - // called with_shared (e.g. from spawn, join, sleep) and the allocator - // fired maybe_preempt() while the lock was held, switch_to_scheduler() - // would context-switch to the scheduler loop, which would immediately - // deadlock trying to acquire the same mutex. + // Preemption must be off while we hold the queue mutex: a + // preemption-driven context switch under it would carry the lock off + // to a suspended actor and stall every scheduler thread. (The stop + // sentinel shares the gate, so nothing can unwind in here either.) let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); let result = f(&mut self.shared.lock().unwrap()); crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev)); result } + /// Slot lookup by index only — bounds-checked, NOT generation-checked. + /// `ROOT_PID` (index `u32::MAX`) is out of bounds by construction and + /// resolves to `None`. Callers verify the generation atomically: either + /// inside a CAS on the word, or by re-reading the word under the cold lock. + #[inline] + pub(crate) fn slot_at(&self, pid: Pid) -> Option<&Slot> { + self.slots.get(pid.index() as usize) + } + + /// Push to the run queue. Callers must have just transitioned the pid + /// into `Queued` (spawn's publish, the unpark protocol, or the + /// scheduler's yield/notified-park return paths). + pub(crate) fn enqueue(&self, pid: Pid) { + self.with_shared(|s| s.run_queue.push_back(pid)); + crate::te!(crate::trace::Event::Enqueue(pid)); + } + + /// Make `pid` runnable if it is parked; coalesce or defer otherwise. + /// The runtime-internal core of `scheduler::unpark`. + pub(crate) fn unpark(&self, pid: Pid) { + if let Some(slot) = self.slot_at(pid) { + if slot.unpark_action(pid) { + crate::te!(crate::trace::Event::UnparkDirect(pid)); + self.enqueue(pid); + } + } + } + /// 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 { MonitorId(self.next_monitor_id.fetch_add(1, Ordering::Relaxed) + 1) } + + /// Pop a vacant slot index, or die loudly. The fixed slab is a deliberate + /// v0.5 simplification (ROADMAP: "Deferred"); the panic names the fix. + pub(crate) fn allocate_slot(&self) -> u32 { + match self.free.lock().pop() { + Some(idx) => idx, + None => panic!( + "smarm: actor slot table exhausted — {} actors are live \ + simultaneously, which is the configured maximum. \ + Fix: raise the cap at runtime init, e.g. \ + `smarm::init(Config::default().max_actors({}))`. \ + (Slots are ~256 bytes each; the table is allocated up-front.)", + self.slots.len(), + self.slots.len() * 2 + ), + } + } } // --------------------------------------------------------------------------- @@ -394,7 +646,13 @@ pub struct Runtime { pub fn init(config: Config) -> Runtime { let n = config.resolved_thread_count(); Runtime { - inner: RuntimeInner::new(n, config.alloc_interval, config.timeslice_cycles, config.stack_pool_cap), + inner: RuntimeInner::new( + n, + config.alloc_interval, + config.timeslice_cycles, + config.stack_pool_cap, + config.max_actors, + ), thread_count: n, } } @@ -437,9 +695,12 @@ impl Runtime { // Re-initialise shared state for this run. { - let mut s = self.inner.shared.lock().unwrap(); + let s = self.inner.shared.lock().unwrap(); assert!(s.run_queue.is_empty(), "run() called while previous run still active"); - s.root_pid = Some(ROOT_PID); + debug_assert_eq!( + self.inner.live_actors.load(Ordering::Acquire), 0, + "run() called while previous run still active" + ); } *self.inner.io.lock().unwrap() = Some(IoThread::start().expect("failed to start IO thread")); @@ -473,10 +734,17 @@ impl Runtime { // Drop initial handle (decrements outstanding_handles count). drop(initial_handle); - // Tear down IO and clean up shared state for the next run() call. + // Tear down IO and clean up for the next run() call. drop(self.inner.io.lock().unwrap().take()); // joins IO threads self.inner.timers.lock().unwrap().clear(); self.inner.next_monitor_id.store(0, Ordering::Relaxed); + // Every slot must have come back: any leak here is a runtime bug + // (a JoinHandle held across run() is decremented just above). + debug_assert_eq!( + self.inner.free.lock().len(), + self.inner.slots.len(), + "slot leak across run()" + ); // Reset per-thread stats. for stat in &self.inner.stats { stat.current_pid_index.store(u32::MAX, Ordering::Relaxed); @@ -528,28 +796,96 @@ pub(crate) fn set_yield_intent(i: YieldIntent) { // Sentinel root PID // --------------------------------------------------------------------------- +/// Index `u32::MAX` is out of bounds for any slab (Config asserts +/// `max_actors < u32::MAX`), so every slot lookup on ROOT_PID resolves to +/// `None` — the root "actor" silently absorbs supervisor signals. pub const ROOT_PID: Pid = Pid::new(u32::MAX, u32::MAX); +// --------------------------------------------------------------------------- +// Spawn-side slot installation +// --------------------------------------------------------------------------- + +/// Install a freshly spawned actor into the slot `idx` (which must have come +/// from `allocate_slot`) and publish it as Queued. Returns the new `Pid`. +/// Called by `scheduler::spawn_under`; lives here next to its inverse +/// (`reclaim_slot`) so the lifecycle is in one file. +pub(crate) fn install_actor( + inner: &RuntimeInner, + idx: u32, + sp: usize, + stack: crate::stack::Stack, + supervisor: Pid, + closure: Closure, +) -> Pid { + let slot = &inner.slots[idx as usize]; + let gen = slot.generation(); // stable: we own the vacant slot via the free list + let pid = Pid::new(idx, gen); + debug_assert_eq!(word_state(slot.word()), ST_VACANT); + + let stop = Arc::new(AtomicBool::new(false)); + slot.stop_ptr.store(Arc::as_ptr(&stop) as *mut _, Ordering::Release); + { + let mut cold = slot.cold.lock(); + debug_assert!(cold.actor.is_none(), "install over live actor"); + debug_assert!(cold.waiters.is_empty() && cold.monitors.is_empty() && cold.links.is_empty()); + cold.actor = Some(Actor { pid, stack, supervisor, stop, trap: None }); + cold.outstanding_handles = 1; + cold.outcome = None; + cold.pending_io_result = None; + } + slot.sp.store(sp, Ordering::Relaxed); + slot.store_closure(closure); + inner.live_actors.fetch_add(1, Ordering::Relaxed); + + // Publish: only now can pops, unparks, or stops find the actor. The + // Release store orders everything above before any Acquire reader. + slot.store_word(gen, ST_QUEUED); + inner.enqueue(pid); + crate::te!(crate::trace::Event::Spawn { parent: supervisor, child: pid }); + pid +} + // --------------------------------------------------------------------------- // Slot reclamation // --------------------------------------------------------------------------- -pub(crate) fn reclaim_slot(s: &mut SharedState, pid: Pid) { - let idx = pid.index(); - let slot = &mut s.slots[idx as usize]; - slot.generation = slot.generation.wrapping_add(1); - slot.actor = None; - slot.outcome = None; - slot.waiters.clear(); - slot.supervisor_channel = None; - slot.monitors.clear(); - slot.links.clear(); - slot.state = State::Done; - slot.outstanding_handles = 0; - slot.pending_unpark = false; - slot.pending_io_result = None; - slot.pending_closure = None; - s.free_list.push(idx); +/// Reclaim `pid`'s slot if (still) eligible: generation matches, state is +/// Done, and no handles are outstanding. Safe to call from racing sites +/// (finalize tail vs. JoinHandle drop): the first caller bumps the +/// generation under the cold lock, the loser sees the mismatch and no-ops. +/// +/// Channel senders extracted from the slot are dropped *after* the cold lock +/// is released — a last-sender drop can unpark a receiver, which takes the +/// run-queue mutex; legal under a cold lock, but pointless to nest. +pub(crate) fn reclaim_slot(inner: &RuntimeInner, pid: Pid) { + let Some(slot) = inner.slot_at(pid) else { return }; + let dropped_outside; + { + let mut cold = slot.cold.lock(); + let w = slot.word(); + if word_gen(w) != pid.generation() + || word_state(w) != ST_DONE + || cold.outstanding_handles != 0 + { + return; // already reclaimed, or not yet eligible + } + debug_assert!(cold.actor.is_none(), "reclaiming a slot that still owns an actor"); + dropped_outside = ( + cold.outcome.take(), + cold.supervisor_channel.take(), + cold.pending_io_result.take(), + slot.take_closure(), // an actor stopped before first resume + ); + cold.waiters.clear(); + cold.monitors.clear(); + cold.links.clear(); + slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release); + // The generation bump IS the reclaim: every stale pid is dead from + // this store onwards (unpark protocol, pops, cold-path re-verifies). + slot.store_word(pid.generation().wrapping_add(1), ST_VACANT); + } + drop(dropped_outside); + inner.free.lock().push(pid.index()); } // --------------------------------------------------------------------------- @@ -570,101 +906,118 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { Outcome::Stopped => (Outcome::Stopped, Signal::Stopped(pid), DownReason::Stopped), }; - let (waiters, supervisor_pid, monitors, recycled_stack, links) = inner.with_shared(|s| { - let slot = s.slot_mut(pid).expect("finalize_actor: slot vanished"); - let sup = slot.actor.as_ref().map(|a| a.supervisor); - // Extract the stack before clearing the actor so we can recycle it - // into the pool *outside* the shared lock. - let stack = slot.actor.take().map(|a| a.stack); - slot.outcome = Some(joiner_outcome); - slot.state = State::Done; - let monitors = std::mem::take(&mut slot.monitors); - let waiters = std::mem::take(&mut slot.waiters); - let links = std::mem::take(&mut slot.links); - // Drop this (dying) pid from each linked peer's list now, under the - // same lock. Done regardless of how we died, so a peer that later - // finalizes won't try to propagate back to this about-to-be-reclaimed - // slot — which also makes the abnormal-death cascade acyclic. - for &peer in &links { - if let Some(ps) = s.slot_mut(peer) { - ps.links.retain(|p| *p != pid); - } - } - (waiters, sup, monitors, stack, links) - }); + let slot = inner.slot_at(pid).expect("finalize_actor: pid out of range"); + let (waiters, monitors, links, actor) = { + let mut cold = slot.cold.lock(); + debug_assert_eq!(slot.generation(), pid.generation(), "finalize: slot vanished"); + let actor = cold.actor.take().expect("finalize_actor: actor vanished"); + cold.outcome = Some(joiner_outcome); + slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release); + // Done is published under the cold lock, so join's + // check-Done-or-register-waiter (also under it) can never miss: it + // either sees Done and takes the outcome, or its waiter registration + // happens before our take() below and is woken further down. + slot.store_word(pid.generation(), ST_DONE); + ( + std::mem::take(&mut cold.waiters), + std::mem::take(&mut cold.monitors), + std::mem::take(&mut cold.links), + actor, + ) + }; - // Return the stack to the pool outside the shared lock. The pool grows - // like a Vec (doubles capacity); if we are already at the cap we just - // drop the stack (munmap) instead of retaining it. - if let Some(stack) = recycled_stack { - let mut pool = inner.stack_pool.lock().unwrap(); + // Recycle the stack outside the cold lock; drop the rest of the Actor + // (the trap sender can unpark its receiver — keep that outside too). + let supervisor_pid = actor.supervisor; + let Actor { stack, .. } = actor; + { + let mut pool = inner.stack_pool.lock(); if pool.len() < inner.stack_pool_cap { pool.push(stack); } // else: drop here → munmap, same as before } - // Deliver to supervisor. - if let Some(sup) = supervisor_pid { - let sender = inner.with_shared(|s| { - s.slot(sup).and_then(|slot| slot.supervisor_channel.clone()) - }); - if let Some(sender) = sender { - let _ = sender.send(sup_signal); + // Deliver to supervisor. ROOT_PID resolves to no slot → silently absorbed. + let sender = inner.slot_at(supervisor_pid).and_then(|sup| { + let cold = sup.cold.lock(); + if sup.generation() == supervisor_pid.generation() { + cold.supervisor_channel.clone() + } else { + None } + }); + if let Some(sender) = sender { + let _ = sender.send(sup_signal); } - // Notify monitors. Sent outside the shared lock for the same reason as the - // supervisor signal: `send` may unpark a parked receiver, which re-takes - // the shared mutex. + // Notify monitors. Sent outside any slot lock: `send` may unpark a parked + // receiver, which takes the run-queue mutex. for (_, m) in monitors { let _ = m.send(Down { pid, reason: down_reason }); } - // Propagate to linked peers. A normal exit never propagates; an abnormal - // death (panic or cooperative stop) reaches each linked peer — a trapping - // peer gets an `ExitSignal` message and survives, a non-trapping peer is - // cooperatively stopped (which cascades along *its* links in turn). The - // reverse links were cleared above, so this never ping-pongs back. Decide - // trap-vs-stop under the lock, then act after releasing it: both - // `Sender::send` and `request_stop` re-enter the shared mutex. - if matches!(down_reason, DownReason::Panic | DownReason::Stopped) { - for peer in links { - let trap = inner.with_shared(|s| match s.slot(peer) { - Some(slot) if !matches!(slot.state, State::Done) => { - Some(slot.actor.as_ref().and_then(|a| a.trap.clone())) + // Walk linked peers ONE AT A TIME (cold locks are leaves). For every + // peer: remove the back-link to this (dying) actor; on abnormal death, + // also fetch its trap sender and deliver after unlocking. + // + // Acyclicity: the back-link removal happens under the peer's cold lock + // *before* any stop is delivered to it, so when the peer later dies its + // own cascade no longer contains us. Two peers finalizing concurrently + // each find the other already Done (set above, before any cascade) and + // skip — no ping-pong, no deadlock (never two cold locks held). + let abnormal = matches!(down_reason, DownReason::Panic | DownReason::Stopped); + for peer in links { + let trap = match inner.slot_at(peer) { + Some(ps) => { + let mut cold = ps.cold.lock(); + let w = ps.word(); + if word_gen(w) == peer.generation() + && !matches!(word_state(w), ST_DONE | ST_VACANT) + { + cold.links.retain(|p| *p != pid); + if abnormal { + Some(cold.actor.as_ref().and_then(|a| a.trap.clone())) + } else { + None // normal exit never propagates + } + } else { + None // peer already gone; nothing to do } - _ => None, // peer already gone; nothing to do - }); - match trap { - Some(Some(tx)) => { - let _ = tx.send(crate::link::ExitSignal { from: pid, reason: down_reason }); - } - Some(None) => crate::scheduler::request_stop(peer), - None => {} } + None => None, + }; + match trap { + Some(Some(tx)) => { + let _ = tx.send(crate::link::ExitSignal { from: pid, reason: down_reason }); + } + Some(None) => crate::scheduler::request_stop(peer), + None => {} } } // Unpark joiners. for joiner in waiters { - crate::scheduler::unpark(joiner); + inner.unpark(joiner); } - // Reclaim if no outstanding handles. - inner.with_shared(|s| { - let reclaim = s.slot(pid).map(|slot| slot.outstanding_handles == 0).unwrap_or(false); - if reclaim { reclaim_slot(s, pid); } - }); + // Reclaim if no outstanding handles (re-verified inside). + reclaim_slot(inner, pid); + + // The decrement is LAST: every wakeup this finalize produced (joiners, + // monitor/trap sends, stop cascades) is enqueued before `live_actors` + // can be observed at its decremented value. See the termination note in + // the module docs. + inner.live_actors.fetch_sub(1, Ordering::Release); } // --------------------------------------------------------------------------- // schedule_loop — runs on each scheduler OS thread // --------------------------------------------------------------------------- -fn schedule_loop(inner: &Arc, slot: usize) { +fn schedule_loop(inner: &Arc, slot_idx: usize) { crate::preempt::configure_preempt(inner.alloc_interval, inner.timeslice_cycles); - let stats = &inner.stats[slot]; + let stats = &inner.stats[slot_idx]; loop { // ---------------------------------------------------------------- @@ -672,17 +1025,10 @@ fn schedule_loop(inner: &Arc, slot: usize) { // losers skip immediately and proceed to step 2. // ---------------------------------------------------------------- if let Ok(_drain_guard) = inner.drain_lock.try_lock() { - // Drain due timers and IO completions. These now live behind their - // own mutexes (peeled out of `shared`), so the pure-yield / - // pure-compute hot path no longer contends the global lock just to - // discover there is nothing to drain. We still read the clock only - // when the timer heap is non-empty. - // - // IO completions are still drained unconditionally while the IO - // subsystem is present: the IO/pool threads push completions onto - // their own mutex, so the scheduler can't cheaply know in advance - // whether any arrived — it must look. That look is a single - // empty-VecDeque check on the empty path. + // Timers and IO live behind their own mutexes (phase 1), so the + // pure-yield / pure-compute hot path never contends a global lock + // just to discover there is nothing to drain. The clock is read + // only when the timer heap is non-empty. let due = { let mut t = inner.timers.lock().unwrap(); if t.is_empty() { @@ -697,32 +1043,14 @@ fn schedule_loop(inner: &Arc, slot: usize) { .unwrap_or_default(); for entry in due { match entry.reason { - crate::timer::Reason::Sleep => { - inner.with_shared(|s| { - if let Some(slot) = s.slot_mut(entry.pid) { - match slot.state { - State::Parked => { - slot.state = State::Runnable; - s.run_queue.push_back(entry.pid); - crate::te!(crate::trace::Event::Enqueue(entry.pid)); - } - // Actor is between `timers.insert_sleep` - // and `park_current` in `sleep()`. Set - // the flag so the upcoming Park yield - // re-queues instead of suspending. - // Mirrors scheduler::unpark() and the - // IO FdReady path below. - State::Runnable => { - slot.pending_unpark = true; - crate::te!(crate::trace::Event::UnparkDeferred(entry.pid)); - } - State::Done => {} - } - } - }); - } + // A sleep expiry is just an unpark: the protocol handles + // every interleaving — Parked (re-queue), Running (the + // actor is between `timers.insert_sleep` and + // `park_current`; RunningNotified makes the upcoming park + // re-queue), or gone (no-op). + crate::timer::Reason::Sleep => inner.unpark(entry.pid), crate::timer::Reason::WaitTimeout { target, wait_seq } => { - // Runs outside with_shared — the callback may call unpark. + // The callback may call unpark itself. target.on_timeout(entry.pid, wait_seq); } } @@ -734,90 +1062,52 @@ fn schedule_loop(inner: &Arc, slot: usize) { if let Some(io) = inner.io.lock().unwrap().as_mut() { io.outstanding = io.outstanding.saturating_sub(1); } - inner.with_shared(|s| { - if let Some(slot) = s.slot_mut(pid) { - slot.pending_io_result = Some(result); - if matches!(slot.state, State::Parked) { - slot.state = State::Runnable; - s.run_queue.push_back(pid); - crate::te!(crate::trace::Event::Enqueue(pid)); + // Stash the result under the cold lock, then unpark. + // The protocol also covers the submit→park window + // (RunningNotified), which the old code missed for + // Blocking completions — a latent lost wakeup. + if let Some(slot) = inner.slot_at(pid) { + { + let mut cold = slot.cold.lock(); + if slot.generation() == pid.generation() { + cold.pending_io_result = Some(result); + } else { + // Actor died (stopped) with the op in + // flight; discard the result. } } - }); + inner.unpark(pid); + } } crate::io::Completion::FdReady { fd, events: _ } => { - // Resolve the parked pid under the io lock, then do the - // state transition under the shared lock. Lock order is - // always io-before-shared; never the reverse. The two - // are independent here (the waiters map is io-private, - // the slot table is shared), so taking them in sequence - // rather than nested is both correct and cheaper. + // Resolve the parked pid under the io lock, then wake + // through the protocol. Lock order: io before all. let parked_pid = inner.io.lock().unwrap().as_mut().and_then(|io| { let pid = io.waiters.remove(&fd); io.epoll_deregister(fd); pid }); - inner.with_shared(|s| { - if let Some(pid) = parked_pid { - if let Some(slot) = s.slot_mut(pid) { - match slot.state { - State::Parked => { - slot.state = State::Runnable; - s.run_queue.push_back(pid); - crate::te!(crate::trace::Event::UnparkDirect(pid)); - crate::te!(crate::trace::Event::Enqueue(pid)); - } - // Actor is between epoll_register - // and park_current. Set the flag so - // the upcoming Park yield re-queues - // instead of suspending. Mirrors - // scheduler::unpark(). - State::Runnable => { - slot.pending_unpark = true; - crate::te!(crate::trace::Event::UnparkDeferred(pid)); - } - State::Done => {} - } - } - } - }); + if let Some(pid) = parked_pid { + inner.unpark(pid); + } } } } } // drain_guard drops here // ---------------------------------------------------------------- - // 2 + 3. Pop a runnable actor and grab everything we need to resume - // it — all in a single mutex acquisition. - // - // Returns: - // Some((pid, sp, Option)) — got work - // None with all_clear=true — done, exit - // None with all_clear=false — idle, wait + // 2. Pop a runnable pid. The queue mutex covers ONLY the pop; the + // slot's own atomics carry everything needed to resume. // ---------------------------------------------------------------- - enum PopResult { - Work(Pid, usize, *const AtomicBool, Option), - /// Popped a stale PID (actor already finalized). Loop again - /// immediately — the queue may hold runnable work right behind it, - /// so this must NOT take the idle path (which can sleep on a timer - /// deadline or the wake fd). - Retry, - Idle { - next_deadline: Option, - io_outstanding: u32, - wake_fd: Option, - }, + enum Pop { + Got(Pid), + Idle { io_outstanding: u32, wake_fd: Option }, AllDone, } - // SAFETY: the raw pointers are (a) the actor's stack pointer and (b) a - // pointer into the actor's `Arc` stop flag, both valid for - // the lifetime of the actor's Slot. They are only used by this same - // thread immediately after this block (switch_to_actor / set_current_stop). - unsafe impl Send for PopResult {} // closure is Send; usize sp + stop ptr are plain data - // Read IO liveness BEFORE taking the shared lock. Lock order is - // io-before-shared everywhere, and reading it first is also what makes - // the split termination check correct: see the all_clear note below. + // 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), @@ -826,95 +1116,38 @@ fn schedule_loop(inner: &Arc, slot: usize) { let pop = inner.with_shared(|s| { let len = s.run_queue.len() as u64; stats.run_queue_len.store(len, Ordering::Relaxed); - - if let Some(pid) = s.run_queue.pop_front() { - // Happy path: got a PID, grab SP, the stop-flag pointer, and - // optional first closure. We hand out a raw pointer into the - // actor's `Arc` rather than cloning the Arc: the - // box stays alive and at a fixed address for as long as the - // actor is on-CPU (it can't be finalized until it yields back), - // so the scheduler's resume path pays no atomic ref-count cost. - let info = s.slot_mut(pid) - .and_then(|slot| { - let closure = slot.pending_closure.take(); - slot.actor.as_ref() - .map(|a| (a.sp, std::sync::Arc::as_ptr(&a.stop), closure)) - }); - match info { - Some((sp, stop, closure)) => { - PopResult::Work(pid, sp, stop, closure) - } - None => { - // Stale PID (actor already finalized): retry at once. - PopResult::Retry - } - } - } else { - // Queue was empty. Re-examine to decide exit vs wait. - // We exit only when nothing can ever produce more work: - // 1. run queue is still empty - // 2. no live actors (nothing parked, nothing mid-finalize) - // 3. no outstanding IO - // - // `io_out` was read from the io mutex BEFORE we took `shared`. - // That ordering is what keeps the now-split check correct: an - // IO completion can only resurrect an actor by going through - // the drain path, which marks the actor Runnable under THIS - // shared lock. So if we read io first and saw `io_out == 0`, - // and then under `shared` we see the queue empty and no live - // actors, no completion that we missed could have made an actor - // live without also having taken `shared` after us — and it - // didn't, because we hold it now and the queue is empty. A - // completion landing after this point finds no live actor to - // wake. Reading io AFTER shared would be the buggy order: an op - // could complete in the gap and be lost. - // - // Pending timers are deliberately NOT a condition: a timer can - // only ever do something useful by waking a live actor, so once - // `live == 0` every remaining timer entry is orphaned (e.g. a - // sleeping actor that was cooperatively cancelled out of its - // sleep, leaving its deadline behind). Such entries must not - // keep the runtime alive until they fire — that could be an - // arbitrarily long hang. They are dropped on the way out. - let live = s.slots.iter().filter(|slot| slot.actor.is_some()).count(); - let all_clear = s.run_queue.is_empty() && live == 0 && io_out == 0; - if all_clear { - PopResult::AllDone - } else { - PopResult::Idle { - next_deadline: None, // filled in below from the timers lock - io_outstanding: io_out, - wake_fd: io_fd, + match s.run_queue.pop_front() { + Some(pid) => Pop::Got(pid), + None => { + // live_actors is read under the queue lock. If it is 0, + // every finalize fully completed — including its wakeup + // enqueues, which happen-before the decrement — so with + // the queue also empty, nothing can produce work again + // (any enqueue targets a live actor; a spawner is itself + // live). io_out was read before the lock per phase 1. + 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 } } } } }); - // If we're going idle, consult the timer heap for the soonest deadline - // (separate lock, taken after shared is released). If we decided - // AllDone, drop any orphaned timers on the way out. - let pop = match pop { - PopResult::Idle { next_deadline: _, io_outstanding, wake_fd } => { - let next = inner.timers.lock().unwrap().peek_deadline(); - PopResult::Idle { next_deadline: next, io_outstanding, wake_fd } - } - PopResult::AllDone => { + 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(); - PopResult::AllDone + return; } - other => other, - }; - - let (pid, sp, stop_flag, first_closure) = match pop { - PopResult::Work(pid, sp, stop, closure) => { - crate::te!(crate::trace::Event::Dequeue(pid)); - (pid, sp, stop, closure) - } - PopResult::AllDone => return, - PopResult::Retry => continue, - PopResult::Idle { next_deadline, io_outstanding, wake_fd } => { - // Something is still in flight. Sleep on the appropriate source - // to avoid hammering the mutex; the loop will retry on wake. + 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(); @@ -942,11 +1175,28 @@ fn schedule_loop(inner: &Arc, slot: usize) { }; // ---------------------------------------------------------------- - // 3. Resume the actor. + // 3. Claim and resume the actor: CAS Queued → Running. A failure + // means the pid is stale (slot recycled — generation mismatch); + // by the at-most-once-enqueued invariant nothing else can have + // changed the state of a queued actor. // ---------------------------------------------------------------- + let slot = match inner.slot_at(pid) { + Some(s) => s, + None => continue, // can't happen for real pids; defensive + }; + if let Err(actual) = slot.cas_word(pid.generation(), ST_QUEUED, ST_RUNNING) { + debug_assert_ne!( + word_gen(actual), pid.generation(), + "queued pid in unexpected state {}", word_state(actual) + ); + continue; // stale pid: retry immediately (never the idle path) + } + crate::te!(crate::trace::Event::Dequeue(pid)); + let sp = slot.sp.load(Ordering::Relaxed); + let stop_flag = slot.stop_ptr.load(Ordering::Relaxed); // First resume: move the closure into the trampoline's thread-local. - if let Some(b) = first_closure { + if let Some(b) = slot.take_closure() { set_current_actor_box(b); } @@ -970,44 +1220,42 @@ fn schedule_loop(inner: &Arc, slot: usize) { crate::preempt::clear_current_stop(); let intent = YIELD_INTENT.with(|c| c.get()); - let new_sp = get_actor_sp(); + slot.sp.store(get_actor_sp(), Ordering::Relaxed); if is_actor_done() { crate::te!(crate::trace::Event::Done(pid)); let outcome = take_last_outcome().unwrap_or(Outcome::Exit); finalize_actor(inner, pid, outcome); } else { - inner.with_shared(|s| { - if let Some(slot) = s.slot_mut(pid) { - if let Some(actor) = slot.actor.as_mut() { - actor.sp = new_sp; - } - match intent { - YieldIntent::Yield => { - crate::te!(crate::trace::Event::Yield(pid)); - slot.state = State::Runnable; - s.run_queue.push_back(pid); - crate::te!(crate::trace::Event::Enqueue(pid)); + let gen = pid.generation(); + match intent { + YieldIntent::Yield => { + // Running OR RunningNotified → Queued; a notification + // arriving mid-run coalesces into the re-queue. + crate::te!(crate::trace::Event::Yield(pid)); + let prev = slot.word.swap(pack(gen, ST_QUEUED), Ordering::AcqRel); + debug_assert!(matches!( + word_state(prev), ST_RUNNING | ST_RUNNING_NOTIFIED + )); + inner.enqueue(pid); + } + YieldIntent::Park => { + match slot.cas_word(gen, ST_RUNNING, ST_PARKED) { + Ok(_) => { + crate::te!(crate::trace::Event::Park(pid)); } - YieldIntent::Park => { - // Check if unpark() fired while the actor was - // still running (between registering in the - // channel and calling park_current). If so, - // re-queue immediately instead of parking. - if slot.pending_unpark { - slot.pending_unpark = false; - slot.state = State::Runnable; - s.run_queue.push_back(pid); - crate::te!(crate::trace::Event::UnparkFlagConsumed(pid)); - crate::te!(crate::trace::Event::Enqueue(pid)); - } else { - crate::te!(crate::trace::Event::Park(pid)); - slot.state = State::Parked; - } + Err(actual) => { + // An unpark landed in the prep-to-park window: + // we are RunningNotified. Re-queue instead of + // parking — the lost-wakeup window, closed. + debug_assert_eq!(word_state(actual), ST_RUNNING_NOTIFIED); + slot.store_word(gen, ST_QUEUED); + crate::te!(crate::trace::Event::UnparkFlagConsumed(pid)); + inner.enqueue(pid); } } } - }); + } } } } diff --git a/src/scheduler.rs b/src/scheduler.rs index 56e6147..9c59785 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -21,18 +21,39 @@ use std::sync::Arc; // --------------------------------------------------------------------------- /// Borrow the current runtime. Panics if called outside `Runtime::run()`. +/// +/// The whole span runs with preemption disabled. Two reasons, both load- +/// bearing: +/// +/// - The `RUNTIME` thread-local borrow is live across `f`. A preemption- +/// driven context switch inside `f` can resume the actor on a DIFFERENT OS +/// thread; the borrow guard would then increment this thread's RefCell but +/// decrement the other's — a count underflow that leaves that thread's +/// RefCell permanently "mutably borrowed". Holding a thread-local guard +/// across a potential switch point is the one unforgivable sin of green +/// threads; disabling preemption makes "no switch inside `f`" structural +/// instead of an accident of which bodies happen to allocate. +/// - `f` is runtime bookkeeping. Suspending an actor halfway through it (or +/// unwinding via the stop sentinel, which shares the gate) is never wanted. pub(crate) fn with_runtime(f: impl FnOnce(&Arc) -> R) -> R { - RUNTIME.with(|r| { + let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); + let result = RUNTIME.with(|r| { let b = r.borrow(); let inner = b.as_ref().expect("smarm: not inside Runtime::run()"); f(inner) - }) + }); + crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev)); + result } /// Borrow the runtime if present; returns `None` otherwise. /// Used on cleanup paths (channel Drop during teardown). +/// Same preemption gate as [`with_runtime`]; same reasons. pub(crate) fn try_with_runtime(f: impl FnOnce(&Arc) -> R) -> Option { - RUNTIME.with(|r| r.borrow().as_ref().map(|inner| f(inner))) + let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); + let result = RUNTIME.with(|r| r.borrow().as_ref().map(|inner| f(inner))); + crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev)); + result } // --------------------------------------------------------------------------- @@ -54,22 +75,30 @@ impl JoinHandle { pub fn join(mut self) -> Result<(), JoinError> { use crate::actor::Outcome; - use crate::runtime::State; // need State visibility let me = current_pid().expect("join() called outside an actor"); loop { + // Check-Done-or-register-waiter is atomic under the target's cold + // lock; finalize publishes Done and takes the waiter list under + // the same lock, so we either see the outcome or are woken. let outcome = with_runtime(|inner| { - inner.with_shared(|s| { - let slot = s.slot_mut(self.pid) - .expect("join: target slot has been reused"); - if matches!(slot.state, State::Done) { - Some(slot.outcome.take().expect("Done slot must have outcome")) - } else { - slot.waiters.push(me); - None - } - }) + let slot = inner.slot_at(self.pid) + .expect("join: pid index out of range"); + let mut cold = slot.cold.lock(); + let w = slot.word(); + // Our outstanding handle pins the slot: it cannot be + // reclaimed (generation cannot change) while we hold it. + assert_eq!( + crate::runtime::word_gen(w), self.pid.generation(), + "join: target slot has been reused" + ); + if crate::runtime::word_state(w) == crate::runtime::ST_DONE { + Some(cold.outcome.take().expect("Done slot must have outcome")) + } else { + cold.waiters.push(me); + None + } }); match outcome { @@ -95,20 +124,24 @@ impl JoinHandle { fn decrement_handle_count(&mut self) { with_runtime(|inner| { - inner.with_shared(|s| { - let should_reclaim = match s.slot_mut(self.pid) { - Some(slot) => { - slot.outstanding_handles = - slot.outstanding_handles.saturating_sub(1); - matches!(slot.state, crate::runtime::State::Done) - && slot.outstanding_handles == 0 + let should_reclaim = match inner.slot_at(self.pid) { + Some(slot) => { + let mut cold = slot.cold.lock(); + if slot.generation() == self.pid.generation() { + cold.outstanding_handles = cold.outstanding_handles.saturating_sub(1); + cold.outstanding_handles == 0 + && crate::runtime::word_state(slot.word()) + == crate::runtime::ST_DONE + } else { + false } - None => false, - }; - if should_reclaim { - crate::runtime::reclaim_slot(s, self.pid); } - }) + None => false, + }; + if should_reclaim { + // Re-verified inside; benign if finalize's reclaim won a race. + crate::runtime::reclaim_slot(inner, self.pid); + } }); } } @@ -129,51 +162,28 @@ impl Drop for JoinHandle { // --------------------------------------------------------------------------- pub fn spawn(f: impl FnOnce() + Send + 'static) -> JoinHandle { - let parent = current_pid() - .or_else(|| with_runtime(|inner| inner.with_shared(|s| s.root_pid))) - .expect("spawn() before run()"); + let parent = current_pid().unwrap_or_else(|| { + // Outside an actor but inside run(): the initial spawn. with_runtime + // panics with "not inside Runtime::run()" if there's no runtime at all. + with_runtime(|_| crate::runtime::ROOT_PID) + }); spawn_under(parent, f) } pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHandle { - // Try to reuse a stack from the pool; fall back to a fresh mmap if empty. - // Allocation happens before taking the shared lock so any syscall doesn't - // stall other scheduler threads. - let stack = with_runtime(|inner| inner.stack_pool.lock().unwrap().pop()) + // Stack + closure boxing happen before ANY runtime lock is taken: no + // syscall and no allocation ever stalls another scheduler thread. + let stack = with_runtime(|inner| inner.stack_pool.lock().pop()) .unwrap_or_else(|| { crate::stack::Stack::new(crate::runtime::ACTOR_STACK_SIZE) .expect("stack allocation failed") }); let sp = init_actor_stack(stack.top(), crate::actor::trampoline); + let closure: crate::runtime::Closure = Box::new(f); let pid = with_runtime(|inner| { - inner.with_shared(|s| { - let (idx, gen) = s.allocate_slot(); - let pid = Pid::new(idx, gen); - let slot = &mut s.slots[idx as usize]; - slot.actor = Some(crate::actor::Actor { - pid, - stack, - sp, - supervisor, - stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), - trap: None, - }); - slot.state = crate::runtime::State::Runnable; - slot.outstanding_handles = 1; - slot.outcome = None; - slot.waiters.clear(); - slot.supervisor_channel = None; - slot.monitors.clear(); - slot.links.clear(); - slot.pending_unpark = false; - slot.pending_io_result = None; - slot.pending_closure = Some(Box::new(f) as crate::runtime::Closure); - s.run_queue.push_back(pid); - crate::te!(crate::trace::Event::Spawn { parent: supervisor, child: pid }); - crate::te!(crate::trace::Event::Enqueue(pid)); - pid - }) + let idx = inner.allocate_slot(); // panics loudly on slab exhaustion + crate::runtime::install_actor(inner, idx, sp, stack, supervisor, closure) }); JoinHandle { pid, consumed: false } @@ -207,39 +217,19 @@ pub fn park_current() { } pub fn unpark(pid: Pid) { - let result = try_with_runtime(|inner| { - inner.with_shared(|s| { - if let Some(slot) = s.slot_mut(pid) { - match slot.state { - crate::runtime::State::Parked => { - // Actor is suspended — safe to re-queue immediately. - slot.state = crate::runtime::State::Runnable; - s.run_queue.push_back(pid); - crate::te!(crate::trace::Event::UnparkDirect(pid)); - crate::te!(crate::trace::Event::Enqueue(pid)); - } - crate::runtime::State::Runnable => { - // Actor is still running (between registering its - // parked_receiver and calling park_current). Set the - // flag; the scheduler will re-queue after the Park - // yield instead of sleeping. - slot.pending_unpark = true; - crate::te!(crate::trace::Event::UnparkDeferred(pid)); - } - crate::runtime::State::Done => {} - } - } - }) - }); - let _ = result; + // The whole protocol lives on the slot's packed word (gen + state in one + // CAS) — see Slot::unpark_action. No runtime lock unless we enqueue. + let _ = try_with_runtime(|inner| inner.unpark(pid)); } /// Request cooperative cancellation of `pid`. /// -/// Sets the actor's stop flag and, if it is parked, wakes it so it observes -/// the flag promptly. The actor realises the stop as a controlled unwind at -/// its next observation point (`check!()`/allocation, or the wakeup side of a -/// blocking park), terminating with `Outcome::Stopped`. +/// Sets the actor's stop flag and wakes it so it observes the flag promptly: +/// a parked target is re-queued; a running target is marked notified, so its +/// next park returns immediately to an observation point. The actor realises +/// the stop as a controlled unwind at its next observation point +/// (`check!()`/allocation, or the wakeup side of a blocking park), +/// terminating with `Outcome::Stopped`. /// /// This is best-effort and cooperative: an actor that never reaches an /// observation point — a tight loop with no `check!()`, no allocation, and no @@ -247,21 +237,19 @@ pub fn unpark(pid: Pid) { /// if `pid` is already gone. pub fn request_stop(pid: Pid) { let _ = try_with_runtime(|inner| { - inner.with_shared(|s| { - if let Some(slot) = s.slot_mut(pid) { - if let Some(actor) = slot.actor.as_ref() { - actor.stop.store(true, std::sync::atomic::Ordering::Relaxed); - } - // Wake a parked target so it reaches its post-park observation - // point now. A Runnable target will observe on its next yield; - // a Done target has nothing to stop. - if matches!(slot.state, crate::runtime::State::Parked) { - slot.state = crate::runtime::State::Runnable; - s.run_queue.push_back(pid); - crate::te!(crate::trace::Event::Enqueue(pid)); + if let Some(slot) = inner.slot_at(pid) { + { + let cold = slot.cold.lock(); + // Verify under the cold lock: generation can't change while + // we hold it (reclaim takes the same lock). + if slot.generation() == pid.generation() { + if let Some(actor) = cold.actor.as_ref() { + actor.stop.store(true, std::sync::atomic::Ordering::Relaxed); + } } } - }) + inner.unpark(pid); + } }); } @@ -333,13 +321,13 @@ where }); park_current(); } - let result = with_runtime(|inner| inner.with_shared(|s| { - s.slot_mut(me) - .expect("block_on_io: own slot vanished") - .pending_io_result + let result = with_runtime(|inner| { + let slot = inner.slot_at(me).expect("block_on_io: own slot vanished"); + let mut cold = slot.cold.lock(); + cold.pending_io_result .take() .expect("block_on_io: resumed without a result") - })); + }); match result { Ok(any) => *any.downcast::().expect("block_on_io: type mismatch"), Err(payload) => std::panic::resume_unwind(payload), @@ -382,13 +370,16 @@ pub fn write(fd: std::os::fd::RawFd, buf: &[u8]) -> std::io::Result { // --------------------------------------------------------------------------- pub fn register_supervisor_channel(pid: Pid, sender: Sender) { - with_runtime(|inner| inner.with_shared(|s| { - if let Some(slot) = s.slot_mut(pid) { - slot.supervisor_channel = Some(sender); - } else { - panic!("register_supervisor_channel: pid {:?} not found", pid); - } - })); + with_runtime(|inner| { + let slot = inner.slot_at(pid) + .unwrap_or_else(|| panic!("register_supervisor_channel: pid {:?} not found", pid)); + let mut cold = slot.cold.lock(); + assert_eq!( + slot.generation(), pid.generation(), + "register_supervisor_channel: pid {:?} not found", pid + ); + cold.supervisor_channel = Some(sender); + }); } // ---------------------------------------------------------------------------