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<AtomicBool>; the scheduler resume path takes a raw pointer into it (no per-resume refcount traffic), keeping yield throughput at baseline.
This commit is contained in:
+29
-2
@@ -24,8 +24,22 @@ use std::panic;
|
||||
pub enum Outcome {
|
||||
Exit,
|
||||
Panic(Box<dyn Any + Send>),
|
||||
/// 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::<StopSentinel>() {
|
||||
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<std::sync::atomic::AtomicBool>,
|
||||
}
|
||||
|
||||
+2
-2
@@ -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};
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<u64> = 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<AtomicBool>` 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<AtomicBool>`
|
||||
// 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()) {
|
||||
|
||||
+28
-15
@@ -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<RuntimeInner>, pid: Pid, outcome: Outcome) {
|
||||
Signal::Panic(pid, Box::new(()) as Box<dyn std::any::Any + Send>),
|
||||
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<RuntimeInner>, slot: usize) {
|
||||
// None with all_clear=false — idle, wait
|
||||
// ----------------------------------------------------------------
|
||||
enum PopResult {
|
||||
Work(Pid, usize, Option<Closure>),
|
||||
Work(Pid, usize, *const AtomicBool, Option<Closure>),
|
||||
Idle {
|
||||
next_deadline: Option<std::time::Instant>,
|
||||
io_outstanding: u32,
|
||||
@@ -738,23 +742,30 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, 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<AtomicBool>` 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<AtomicBool>` 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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();
|
||||
|
||||
+49
-1
@@ -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
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -16,6 +16,10 @@ pub enum Signal {
|
||||
Exit(Pid),
|
||||
/// The child panicked. Payload is whatever `panic!` was called with.
|
||||
Panic(Pid, Box<dyn Any + Send>),
|
||||
/// 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,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user