diff --git a/src/runtime.rs b/src/runtime.rs index 8aa26f2..974bfa7 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -797,6 +797,11 @@ fn schedule_loop(inner: &Arc, slot: usize) { // ---------------------------------------------------------------- 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, @@ -840,13 +845,8 @@ fn schedule_loop(inner: &Arc, slot: usize) { PopResult::Work(pid, sp, stop, closure) } None => { - // Stale PID (actor already finalized). Treat as idle - // so the outer loop retries immediately. - PopResult::Idle { - next_deadline: None, - io_outstanding: 0, - wake_fd: None, - } + // Stale PID (actor already finalized): retry at once. + PopResult::Retry } } } else { @@ -911,6 +911,7 @@ fn schedule_loop(inner: &Arc, slot: usize) { (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. diff --git a/tests/poison_stop.rs b/tests/poison_stop.rs index 316730c..3845ab5 100644 --- a/tests/poison_stop.rs +++ b/tests/poison_stop.rs @@ -10,7 +10,10 @@ use std::sync::Arc; #[test] fn stop_storm_does_not_poison_runtime() { - let rt = init(Config::exact(4)); + // alloc_interval(1): every allocation is an observation point, so pre-fix + // the sentinel fires inside with_shared's alloc sites (run_queue.push_back, + // waiters.push, ...) with near-certainty rather than 1-in-128. + let rt = init(Config::exact(4).alloc_interval(1)); let completed = Arc::new(AtomicUsize::new(0)); let c = completed.clone(); rt.run(move || { @@ -40,3 +43,47 @@ fn stop_storm_does_not_poison_runtime() { }); assert_eq!(completed.load(Ordering::SeqCst), 1, "root completed cleanly"); } + +/// The sharper repro: a stop-flagged actor whose *next allocation* is the +/// `Box::new(closure)` inside `spawn`'s `with_shared` critical section. +/// Pre-fix, the ungated `check_cancelled` in `maybe_preempt` raises the +/// sentinel right there, unwinding while the global shared mutex is held → +/// poisoned → every later `with_shared` panics on every scheduler thread and +/// the whole runtime collapses. Post-fix, observation is deferred to the next +/// lock-free point and each actor dies a clean `Outcome::Stopped`. +/// +/// `request_stop(self_pid())` sets the flag without parking (we're running), +/// and the capturing closure makes `Box::new(f)` a real allocation. With +/// `alloc_interval(1)` the check fires every other allocation, so each actor +/// has ~50% odds of the check landing on the under-lock alloc; 64 actors make +/// a pre-fix escape astronomically unlikely. +#[test] +fn self_stop_during_spawn_does_not_poison_shared_mutex() { + let rt = init(Config::exact(4).alloc_interval(1)); + let completed = Arc::new(AtomicUsize::new(0)); + let c = completed.clone(); + rt.run(move || { + let mut handles = Vec::new(); + for i in 0..64usize { + let h = smarm::spawn(move || { + // Vary pre-stop allocation count to cover both phases of the + // every-other-allocation check cadence. + for _ in 0..(i % 2) { + let _phase: Vec = Vec::with_capacity(8); + } + let payload = vec![0u8; 64]; // captured → Box::new(f) allocates + smarm::request_stop(smarm::self_pid()); + let grandchild = smarm::spawn(move || drop(payload)); + let _ = grandchild.join(); + }); + handles.push(h); + } + for h in handles { + // Stopped reports Ok from join; the point is that join itself + // (with_shared) doesn't panic on a poisoned mutex. + let _ = h.join(); + } + c.fetch_add(1, Ordering::SeqCst); + }); + assert_eq!(completed.load(Ordering::SeqCst), 1, "root completed cleanly"); +}