Markk116/smarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(runtime): phase 2 — fixed slab, per-slot packed-atomic state machine, raw cold locks

Browse Source 
The slot table split (ROADMAP_v0.5 phase 2): slot lookup is lock-free, the
run path (yield/park/unpark/pop/resume) takes zero locks beyond the queue
mutex itself, and SharedState shrinks to { run_queue }.

Core pieces:

- src/raw_mutex.rs: hand-rolled 3-state futex mutex (Drepper mutex3),
  non-poisoning; the guard enters NoPreempt, which both bounds hold times and
  structurally closes the unwind-under-lock hole (the stop sentinel shares
  the gate). Used for per-slot cold data, the free list, and the stack pool.

- Fixed slab Box<[Slot]> (default max_actors = 16_384, ~4 MiB, align(128)
  against false sharing). Slots never move -> stable addresses, lock-free
  index. Exhaustion panics loudly, naming Config::max_actors(n) as the fix.
  Unbounded/segmented slab stays deferred (see ROADMAP).

- Per-slot AtomicU64 packing (generation << 32 | state), states
  Vacant/Queued/Running/RunningNotified/Parked/Done. Every transition CASes
  the packed word, so the generation check is atomic with the transition: no
  ABA, no spurious unparks on recycled slots. RunningNotified replaces the
  pending_unpark bool (the lost-wakeup window is a state, not a flag) and
  uniformly fixes a LATENT LOST WAKEUP in the old Blocking-IO completion
  path, which set the result for a still-Running actor without flagging it.

- Invariant: a pid is in the run queue at most once (pushes pair 1:1 with
  transitions into Queued; only the scheduler does Queued->Running). This is
  what makes phase 3's bounded rings sound.

- sp -> relaxed AtomicUsize; stop flag + first-resume closure as AtomicPtrs
  (closure double-boxed for a thin pointer, swap-to-take): the resume path is
  fully atomic.

- finalize_actor: Done published under the dying slot's cold lock (join's
  check-or-register is linearized by it); link cascade locks peers ONE AT A
  TIME (cold locks are leaves) with the acyclicity argument written at the
  site; link() registers on the target first, then self (stale self-entries
  are benign, every walk re-verifies the peer's word).

- Termination by counters: live_actors incremented in spawn pre-enqueue,
  decremented at the very END of finalize after all wakeup enqueues; exit on
  io_out == 0 (read before the queue lock, phase-1 ordering) && queue empty
  && live == 0. Soundness note at the site: any enqueue targets a live actor.

- spawn boxes the closure and acquires the stack BEFORE any runtime lock: no
  allocation ever happens under a global lock anymore.

- with_runtime/try_with_runtime now enter NoPreempt for their full span.
  This fixes a bug the rework exposed: install_actor allocated with
  preemption enabled while the RUNTIME RefCell borrow was live; a timeslice
  preemption there migrates the actor across OS threads and the borrow guard
  increments one thread's RefCell count and decrements another's — underflow
  to 'permanently mutably borrowed', cascading panics (caught by stress
  suite: deterministic non-unwinding-panic abort in lost_wakeup_many_pairs).
  The old code was safe only by accident; now it's structural.

- Behavior note: request_stop on a RUNNING target now marks it
  RunningNotified, so its next park returns immediately to an observation
  point — faster stop observation; parked/queued/done semantics unchanged.

Validated: 22 suites green in release (1/2/8-thread oversubscribed on the
1-core sandbox) and in debug with all debug_asserts live; stress suite x5.
This commit is contained in:
Claude
2026-06-09 20:08:12 +00:00
parent 453f6f491f
commit 5e0c9d45da
7 changed files with 1070 additions and 578 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/actor.rs
+2 -2
View File
@@ -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
src/lib.rs
+1
View File
@@ -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;
// ---------------------------------------------------------------------------
src/link.rs
+66 -44
View File
@@ -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<ExitSignal> {
let (tx, rx) = channel::<ExitSignal>();
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<Option<Sender<ExitSignal>>> = 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);
}
})
}
});
}
src/monitor.rs
+24 -22
View File
@@ -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::<Down>();
// 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<MonitorId> {
// 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<Down>)> = 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)
src/raw_mutex.rs
+228
View File
@@ -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<T> {
state: AtomicU32,
data: UnsafeCell<T>,
}
// 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<T: Send> Send for RawMutex<T> {}
unsafe impl<T: Send> Sync for RawMutex<T> {}
impl<T> RawMutex<T> {
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<T>,
prev_preempt: bool,
}
impl<T> 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<T> 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<T> 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::<libc::timespec>(),
);
}
}
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);
}
}
src/runtime.rs
+644 -396
View File
File diff suppressed because it is too large Load Diff
src/scheduler.rs
+105 -114
View File
@@ -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<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> 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<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> R) -> Option<R> {
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::<T>().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<usize> {
// ---------------------------------------------------------------------------
pub fn register_supervisor_channel(pid: Pid, sender: Sender<Signal>) {
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);
});
}
// ---------------------------------------------------------------------------