From d9a6520a2470c411f6e18ae36512dde81bc83c77 Mon Sep 17 00:00:00 2001 From: smarm-dev Date: Fri, 29 May 2026 22:40:07 +0000 Subject: [PATCH] feat(cancel): cooperative cancellation via sentinel unwind request_stop(pid) sets a per-actor flag and wakes a parked target. The actor realizes the stop as a controlled unwind at its next observation point (check!()/alloc via maybe_preempt, or the wakeup side of any blocking park): a StopSentinel panic tears the stack down via the trampoline's existing catch_unwind, running Drop, and is reported as the new Outcome::Stopped (distinct from a user Panic). Death surface kept distinct from Exit: Signal::Stopped(pid) + DownReason::Stopped, so the supervisor await logic to come can tell a requested stop apart from a self-termination. Flag lives on Actor behind Arc; the scheduler resume path takes a raw pointer into it (no per-resume refcount traffic), keeping yield throughput at baseline. --- src/actor.rs | 31 ++++++++++- src/lib.rs | 4 +- src/monitor.rs | 2 + src/preempt.rs | 53 +++++++++++++++++++ src/runtime.rs | 43 +++++++++------ src/scheduler.rs | 50 +++++++++++++++++- src/supervisor.rs | 6 +++ tests/cancel.rs | 132 ++++++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 301 insertions(+), 20 deletions(-) create mode 100644 tests/cancel.rs diff --git a/src/actor.rs b/src/actor.rs index 58ebd04..145f93c 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -24,8 +24,22 @@ use std::panic; pub enum Outcome { Exit, Panic(Box), + /// The actor was cooperatively cancelled via `request_stop`: a sentinel + /// panic unwound its stack (running Drop) and the trampoline recognised + /// the sentinel. Distinct from a user `Panic` — there is no payload to + /// propagate. + Stopped, } +/// The payload of the sentinel panic raised at an observation point when an +/// actor has been flagged for cooperative cancellation. Zero-sized; its only +/// job is to be recognisable in the trampoline's `catch_unwind` so the +/// teardown is reported as `Outcome::Stopped` rather than `Outcome::Panic`. +/// +/// User code that wraps its own `catch_unwind` can swallow this (cf. Erlang's +/// `catch`); the stop flag stays set, so the next observation point re-raises. +pub(crate) struct StopSentinel; + // Thread-locals that the scheduler writes immediately before `switch_to_actor`. thread_local! { /// The closure for the actor we're about to resume *for the first time*. @@ -83,8 +97,14 @@ pub extern "C-unwind" fn trampoline() { .expect("trampoline entered without a closure set"); let outcome = match panic::catch_unwind(panic::AssertUnwindSafe(b)) { - Ok(()) => Outcome::Exit, - Err(payload) => Outcome::Panic(payload), + Ok(()) => Outcome::Exit, + Err(payload) => { + if payload.is::() { + Outcome::Stopped + } else { + Outcome::Panic(payload) + } + } }; LAST_OUTCOME.with(|r| *r.borrow_mut() = Some(outcome)); @@ -107,4 +127,11 @@ pub struct Actor { 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 + /// parked actor); the actor observes it lock-free at its next observation + /// point — see `preempt::check_cancelled`. Shared via `Arc` so the running + /// actor can poll it without taking the shared mutex. Lives on the `Actor` + /// (constructed fresh at spawn), so unlike a `Slot` field it needs no reset + /// in the slot-recycling paths. + pub stop: std::sync::Arc, } diff --git a/src/lib.rs b/src/lib.rs index 58dff9a..956b259 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -43,8 +43,8 @@ pub use mutex::{LockTimeout, Mutex, MutexGuard}; pub use pid::Pid; pub use runtime::{init, Config, Runtime}; pub use scheduler::{ - block_on_io, run, self_pid, sleep, spawn, spawn_under, wait_readable, wait_writable, - yield_now, JoinError, JoinHandle, + block_on_io, request_stop, run, self_pid, sleep, spawn, spawn_under, wait_readable, + wait_writable, yield_now, JoinError, JoinHandle, }; pub use supervisor::{ChildSpec, OneForOne, Restart, Signal}; diff --git a/src/monitor.rs b/src/monitor.rs index 3821633..44d6495 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -47,6 +47,8 @@ pub enum DownReason { /// The target panicked. The payload is delivered to the actor's joiner, /// not to monitors. Panic, + /// The target was cooperatively cancelled via `request_stop`. + Stopped, /// The target was already gone (finished and reclaimed, or never alive) /// at the moment `monitor()` was called. NoProc, diff --git a/src/preempt.rs b/src/preempt.rs index 40cf570..e8b88c1 100644 --- a/src/preempt.rs +++ b/src/preempt.rs @@ -27,6 +27,7 @@ use std::alloc::{GlobalAlloc, Layout, System}; use std::cell::Cell; +use std::sync::atomic::{AtomicBool, Ordering}; pub const DEFAULT_ALLOC_INTERVAL: u32 = 128; pub const DEFAULT_TIMESLICE_CYCLES: u64 = 300_000; // ≈ 100µs on a 3 GHz CPU @@ -49,6 +50,54 @@ thread_local! { /// Per-thread copy of the configured timeslice, written once at /// scheduler-thread startup. static CONFIGURED_TIMESLICE_CYCLES: Cell = const { Cell::new(DEFAULT_TIMESLICE_CYCLES) }; + + /// Raw pointer to the on-CPU actor's cancellation flag, set by the + /// scheduler on resume and reset to null when control returns to it. + /// Null while no actor is running, so observation points are inert on the + /// scheduler's own stack. A raw pointer (not a cloned `Arc`) keeps the + /// resume path free of atomic ref-count traffic; see `check_cancelled` for + /// the safety argument. + static CURRENT_STOP: Cell<*const AtomicBool> = const { Cell::new(std::ptr::null()) }; +} + +// --------------------------------------------------------------------------- +// Cooperative-cancellation observation +// --------------------------------------------------------------------------- + +/// Bind the on-CPU actor's stop flag. Called by the scheduler immediately +/// before `switch_to_actor`. `flag` must point into the resuming actor's +/// `Arc` heap box. +pub(crate) fn set_current_stop(flag: *const AtomicBool) { + CURRENT_STOP.with(|c| c.set(flag)); +} + +/// Unbind the stop flag. Called by the scheduler on the return path from an +/// actor, so the pointer never leaks into the scheduler loop or the next actor. +pub(crate) fn clear_current_stop() { + CURRENT_STOP.with(|c| c.set(std::ptr::null())); +} + +/// Observation point for cooperative cancellation. If the on-CPU actor has +/// been flagged for stop, raise the sentinel panic so the trampoline's +/// `catch_unwind` tears the stack down (running Drop) and reports +/// `Outcome::Stopped`, distinct from a user `Panic`. A no-op (one TLS load, +/// one relaxed atomic load) on the common path. +/// +/// Called from `maybe_preempt` (amortised, the `check!()`/alloc path) and from +/// the wakeup side of every blocking park (`park_current`/`yield_now`), which +/// is past the prep-to-park window — so it can never lose a wakeup. +#[inline] +pub fn check_cancelled() { + let p = CURRENT_STOP.with(|c| c.get()); + // SAFETY: `p` is either null (no actor on-CPU — the scheduler clears it on + // every return) or a pointer into the on-CPU actor's `Arc` + // heap box. That box outlives the run: an actor's slot is not finalized or + // reclaimed while the actor is on-CPU (finalize runs only after it yields + // back), and the Arc keeps the box alive at a fixed address even if the + // slot table Vec reallocates underneath it. + if !p.is_null() && unsafe { (*p).load(Ordering::Relaxed) } { + std::panic::panic_any(crate::actor::StopSentinel); + } } /// Called once per scheduler thread at startup (before any actor runs). @@ -122,6 +171,10 @@ pub fn maybe_preempt() { let n = c.get(); if n == 0 { c.set(CONFIGURED_ALLOC_INTERVAL.with(|i| i.get())); + // Cooperative cancellation shares the amortised cadence with the + // timeslice check. Observe a pending stop first: if we are being + // cancelled there is no point yielding, we unwind instead. + check_cancelled(); if PREEMPTION_ENABLED.with(|e| e.get()) { let start = TIMESLICE_START.with(|s| s.get()); if rdtsc().saturating_sub(start) > CONFIGURED_TIMESLICE_CYCLES.with(|t| t.get()) { diff --git a/src/runtime.rs b/src/runtime.rs index 20ab2a3..afb71c7 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -44,7 +44,7 @@ use crate::supervisor::Signal; use crate::timer::Timers; use std::collections::VecDeque; -use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::thread; @@ -542,6 +542,10 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { Signal::Panic(pid, Box::new(()) as Box), DownReason::Panic, ), + // Cooperative cancellation: kept distinct from a normal Exit so a + // supervisor's await logic (roadmap #2) can tell "I stopped it" apart + // from "it finished on its own". + Outcome::Stopped => (Outcome::Stopped, Signal::Stopped(pid), DownReason::Stopped), }; let (waiters, supervisor_pid, monitors, recycled_stack) = inner.with_shared(|s| { @@ -730,7 +734,7 @@ fn schedule_loop(inner: &Arc, slot: usize) { // None with all_clear=false — idle, wait // ---------------------------------------------------------------- enum PopResult { - Work(Pid, usize, Option), + Work(Pid, usize, *const AtomicBool, Option), Idle { next_deadline: Option, io_outstanding: u32, @@ -738,23 +742,30 @@ fn schedule_loop(inner: &Arc, slot: usize) { }, AllDone, } - // SAFETY: the raw pointer is the actor's stack pointer, valid for the - // lifetime of the actor's Slot. It is only dereferenced by - // switch_to_actor() on this same thread immediately after this block. - unsafe impl Send for PopResult {} // closure is Send; usize sp is plain data + // 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 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 and optional first closure. - let sp = s.slot(pid) - .and_then(|slot| slot.actor.as_ref().map(|a| a.sp)); - match sp { - Some(sp) => { + // 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(pid) + .and_then(|slot| slot.actor.as_ref() + .map(|a| (a.sp, std::sync::Arc::as_ptr(&a.stop)))); + match info { + Some((sp, stop)) => { let closure = s.pop_pending_closure(pid); - PopResult::Work(pid, sp, closure) + PopResult::Work(pid, sp, stop, closure) } None => { // Stale PID (actor already finalized). Treat as idle @@ -792,10 +803,10 @@ fn schedule_loop(inner: &Arc, slot: usize) { } }); - let (pid, sp, first_closure) = match pop { - PopResult::Work(pid, sp, closure) => { + let (pid, sp, stop_flag, first_closure) = match pop { + PopResult::Work(pid, sp, stop, closure) => { crate::te!(crate::trace::Event::Dequeue(pid)); - (pid, sp, closure) + (pid, sp, stop, closure) } PopResult::AllDone => return, PopResult::Idle { next_deadline, io_outstanding, wake_fd } => { @@ -841,6 +852,7 @@ fn schedule_loop(inner: &Arc, slot: usize) { set_actor_sp(sp); set_current_pid(pid); + crate::preempt::set_current_stop(stop_flag); reset_actor_done(); YIELD_INTENT.with(|c| c.set(YieldIntent::Yield)); crate::preempt::reset_timeslice(); @@ -852,6 +864,7 @@ fn schedule_loop(inner: &Arc, slot: usize) { PREEMPTION_ENABLED.with(|c| c.set(false)); stats.current_pid_index.store(u32::MAX, Ordering::Relaxed); clear_current_pid(); + crate::preempt::clear_current_stop(); let intent = YIELD_INTENT.with(|c| c.get()); let new_sp = get_actor_sp(); diff --git a/src/scheduler.rs b/src/scheduler.rs index 1df6a5e..b9d3639 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -79,6 +79,10 @@ impl JoinHandle { return match o { Outcome::Exit => Ok(()), Outcome::Panic(p) => Err(JoinError { payload: p }), + // A cooperative stop carries no panic payload to + // propagate; the *reason* is observable via monitors + // (DownReason::Stopped). join() therefore reports Ok. + Outcome::Stopped => Ok(()), }; } None => { @@ -147,7 +151,13 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHa 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 }); + slot.actor = Some(crate::actor::Actor { + pid, + stack, + sp, + supervisor, + stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + }); slot.state = crate::runtime::State::Runnable; slot.outstanding_handles = 1; slot.outcome = None; @@ -185,11 +195,18 @@ pub fn self_pid() -> Pid { pub fn yield_now() { runtime::set_yield_intent(YieldIntent::Yield); unsafe { crate::context::switch_to_scheduler() }; + // Observation point: we may have been resumed only to be cancelled. + crate::preempt::check_cancelled(); } pub fn park_current() { runtime::set_yield_intent(YieldIntent::Park); unsafe { crate::context::switch_to_scheduler() }; + // Observation point on the wakeup side of every blocking primitive + // (recv/sleep/mutex/io/join). Past the prep-to-park window, so this never + // races a wakeup: a stop unparks us, we resume here, and unwind out of + // whatever blocking call parked us — running Drop along the way. + crate::preempt::check_cancelled(); } pub fn unpark(pid: Pid) { @@ -220,6 +237,37 @@ pub fn unpark(pid: Pid) { let _ = result; } +/// 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`. +/// +/// This is best-effort and cooperative: an actor that never reaches an +/// observation point — a tight loop with no `check!()`, no allocation, and no +/// blocking op — cannot be stopped, exactly as it cannot be preempted. A no-op +/// 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)); + } + } + }) + }); +} + // --------------------------------------------------------------------------- // NoPreempt // --------------------------------------------------------------------------- diff --git a/src/supervisor.rs b/src/supervisor.rs index c0d69dc..e0ad0d1 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -16,6 +16,10 @@ pub enum Signal { Exit(Pid), /// The child panicked. Payload is whatever `panic!` was called with. Panic(Pid, Box), + /// The child was cooperatively cancelled via `request_stop`. Carries no + /// payload — there is nothing to propagate. Kept distinct from `Exit` so a + /// supervisor can distinguish a stop it requested from a self-termination. + Stopped(Pid), } impl std::fmt::Debug for Signal { @@ -23,6 +27,7 @@ impl std::fmt::Debug for Signal { match self { Signal::Exit(pid) => write!(f, "Signal::Exit({:?})", pid), Signal::Panic(pid, _) => write!(f, "Signal::Panic({:?}, ..)", pid), + Signal::Stopped(pid) => write!(f, "Signal::Stopped({:?})", pid), } } } @@ -32,6 +37,7 @@ impl Signal { match self { Signal::Exit(p) => *p, Signal::Panic(p, _) => *p, + Signal::Stopped(p) => *p, } } } diff --git a/tests/cancel.rs b/tests/cancel.rs new file mode 100644 index 0000000..f166e61 --- /dev/null +++ b/tests/cancel.rs @@ -0,0 +1,132 @@ +//! Cooperative cancellation tests (roadmap item #1, the keystone). +//! +//! `request_stop(pid)` flags an actor for cancellation. The actor realizes the +//! stop as a *controlled unwind*: at the next observation point (a `check!()` +//! / allocation via `maybe_preempt`, or the wakeup side of any blocking park) +//! a dedicated sentinel panic is raised, the trampoline's `catch_unwind` tears +//! the stack down — running Drop guards — and reports `Outcome::Stopped`, +//! distinct from a user `Panic`. Monitors see `DownReason::Stopped`. +//! +//! These cases are deterministic under the single-thread runtime: the parent +//! runs until it parks, so the relative order of `request_stop`, the child +//! reaching its observation point, and the monitor `Down` is fixed. + +use smarm::{channel, monitor, request_stop, run, spawn, yield_now, DownReason}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; + +/// Sets its flag when dropped — used to prove the cancellation unwind runs +/// Drop guards rather than leaking the stack. +struct DropFlag(Arc); +impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } +} + +#[test] +fn looping_actor_on_check_is_stopped() { + let dropped = Arc::new(AtomicBool::new(false)); + let saw_stopped = Arc::new(AtomicBool::new(false)); + let (d, s) = (dropped.clone(), saw_stopped.clone()); + run(move || { + let h = spawn(move || { + let _g = DropFlag(d); + // Tight loop whose only observation point is check!(). + loop { + smarm::check!(); + } + }); + let pid = h.pid(); + let down = monitor(pid); + // Flag the stop before the child is ever resumed; it will observe the + // flag once its check!() loop reaches the amortised preempt check. + request_stop(pid); + let dn = down.recv().expect("monitor channel closed before Down"); + assert_eq!(dn.pid, pid); + if matches!(dn.reason, DownReason::Stopped) { + s.store(true, Ordering::SeqCst); + } + let _ = h.join(); + }); + assert!(saw_stopped.load(Ordering::SeqCst), "expected DownReason::Stopped"); + assert!(dropped.load(Ordering::SeqCst), "Drop guard must run during the cancellation unwind"); +} + +#[test] +fn parked_on_recv_actor_is_stopped() { + let dropped = Arc::new(AtomicBool::new(false)); + let saw_stopped = Arc::new(AtomicBool::new(false)); + let (d, s) = (dropped.clone(), saw_stopped.clone()); + run(move || { + let h = spawn(move || { + let _g = DropFlag(d); + let (tx, rx) = channel::(); + // Keep a sender alive so the channel stays open and recv() parks + // indefinitely rather than returning Err. + let _keep = tx; + let _ = rx.recv(); // parks here until the stop unwinds us out + }); + let pid = h.pid(); + let down = monitor(pid); + // Let the child run and park in recv() before we request the stop. + yield_now(); + request_stop(pid); + let dn = down.recv().expect("monitor channel closed before Down"); + assert_eq!(dn.pid, pid); + if matches!(dn.reason, DownReason::Stopped) { + s.store(true, Ordering::SeqCst); + } + let _ = h.join(); + }); + assert!(saw_stopped.load(Ordering::SeqCst), "expected DownReason::Stopped"); + assert!(dropped.load(Ordering::SeqCst), "Drop guard must run on cancellation of a parked actor"); +} + +#[test] +fn no_check_no_alloc_loop_is_not_stopped() { + // Documents the inherent gap: an actor that never reaches an observation + // point (no check!(), no allocation, no blocking op) cannot be stopped — + // the same limitation as preemption. Here the child runs a bounded, + // allocation-free arithmetic loop, so the stop flagged before it runs is + // silently never honored and the actor exits normally. + let saw_exit = Arc::new(AtomicBool::new(false)); + let s = saw_exit.clone(); + run(move || { + let h = spawn(|| { + let mut x: u64 = 0; + for i in 0..2_000_000u64 { + x = x.wrapping_add(i ^ (x >> 1)); + } + std::hint::black_box(x); + }); + let pid = h.pid(); + let down = monitor(pid); + request_stop(pid); // no observation point => ignored + let dn = down.recv().expect("monitor channel closed before Down"); + if matches!(dn.reason, DownReason::Exit) { + s.store(true, Ordering::SeqCst); + } + let _ = h.join(); + }); + assert!( + saw_exit.load(Ordering::SeqCst), + "a loop with no observation points must exit normally, never Stopped" + ); +} + +#[test] +fn join_on_stopped_actor_returns_ok() { + // A cooperative stop carries no panic payload to propagate, so join() + // reports Ok(()); the *fact* of the stop is observable via monitors + // (DownReason::Stopped), which is the channel that carries termination + // reason. + run(|| { + let h = spawn(|| loop { + smarm::check!(); + }); + let pid = h.pid(); + request_stop(pid); + assert!(h.join().is_ok(), "join on a stopped actor returns Ok(())"); + }); +}