fix(runtime): handle timer firing while actor is still Runnable in sleep()

While running benchmarks, a hang surfaced in timer-only workloads — actors
sleeping with no IO in flight. Tracing it down, the race lives in sleep():
between the call to `timers.insert_sleep` and the subsequent `park_current`
yield, the actor is still in State::Runnable. If the timer fires in that
window, the old code's `if matches!(slot.state, State::Parked)` guard silently
drops the wakeup. The actor then parks normally and never gets re-queued —
it sleeps forever.

The fix mirrors what scheduler::unpark() and the IO FdReady path already do:
when the timer fires and the slot is still Runnable, set `pending_unpark`
instead of re-queuing immediately. The upcoming Park yield sees the flag and
re-queues the actor rather than suspending it, closing the race without any
new synchronisation.

Adds a regression test: 100 actors doing pure timer sleeps across ≥2
scheduler threads. The test asserts both correctness (all actors complete)
and timeliness (wall time < 2×sleep duration), which is enough signal to
catch a stuck actor even on a single-core CI runner.

Cargo.toml

@@ -29,3 +29,15 @@ harness = false
 [[bench]]
 name = "multi_scheduler"
 harness = false
+
+[[bench]]
+name = "general"
+harness = false
+
+[[bench]]
+name = "smarm_favored"
+harness = false
+
+[[bench]]
+name = "tokio_favored"
+harness = false

README.md

@@ -91,10 +91,8 @@ mechanism we know how to add; none belongs in this iteration.
 
 ## Contributing
 
-This is a personal proof-of-concept. There's no PR workflow — if you fork it
+This is a personal proof-of-concept. There's no PR workflow. If you fork it and do something interesting, just send me an email. If it's nice, I'll upstream the changes. 
-and do something interesting, just send me an email. I'd genuinely like to
-hear about it.
 
 ---
 
-<sub>The name is a recursive acronym. The M is for Marks, as in the BEAM — Bogdan/Björn's Erlang Abstract Machine, the virtual machine that runs Erlang and Elixir. smarm is not the BEAM. It just admires it from a safe distance.</sub>
+

benches/baseline.json

@@ -2,223 +2,313 @@
   "chained_spawn": {
     "smarm 1-thread": {
       "result": 1000,
-      "median": 8637,
+      "median": 266,
-      "min": 8553,
+      "min": 242,
-      "max": 8933
+      "max": 351
+    },
+    "smarm 24-thread": {
+      "result": 1000,
+      "median": 742,
+      "min": 696,
+      "max": 860
     },
     "tokio current_thread": {
       "result": 1000,
-      "median": 124,
+      "median": 62,
-      "min": 124,
+      "min": 61,
-      "max": 153
+      "max": 68
     },
     "tokio multi-thread": {
       "result": 1000,
-      "median": 188,
+      "median": 190,
-      "min": 183,
+      "min": 169,
-      "max": 229
+      "max": 207
     }
   },
   "yield_many": {
     "smarm 1-thread": {
       "result": 200000,
-      "median": 41622,
+      "median": 19071,
-      "min": 41063,
+      "min": 18776,
-      "max": 44973
+      "max": 19396
+    },
+    "smarm 24-thread": {
+      "result": 200000,
+      "median": 172454,
+      "min": 166246,
+      "max": 174230
     },
     "tokio current_thread": {
       "result": 200000,
-      "median": 15085,
+      "median": 4737,
-      "min": 15013,
+      "min": 4644,
-      "max": 15274
+      "max": 5065
     },
     "tokio multi-thread": {
       "result": 200000,
-      "median": 15964,
+      "median": 8738,
-      "min": 15880,
+      "min": 7852,
-      "max": 17959
+      "max": 9770
     }
   },
   "fan_out_compute": {
     "smarm 1-thread": {
       "result": 33860,
-      "median": 29727,
+      "median": 13234,
-      "min": 29491,
+      "min": 13196,
-      "max": 31634
+      "max": 13390
+    },
+    "smarm 24-thread": {
+      "result": 33860,
+      "median": 2244,
+      "min": 2162,
+      "max": 2380
     },
     "tokio current_thread": {
       "result": 33860,
-      "median": 28503,
+      "median": 14049,
-      "min": 28391,
+      "min": 14035,
-      "max": 28866
+      "max": 14300
     },
     "tokio multi-thread": {
       "result": 33860,
-      "median": 34542,
+      "median": 1474,
-      "min": 34396,
+      "min": 1285,
-      "max": 36111
+      "max": 1823
     }
   },
   "ping_pong_oneshot": {
     "smarm 1-thread": {
       "result": 1000,
-      "median": 16848,
+      "median": 751,
-      "min": 16633,
+      "min": 727,
-      "max": 17301
+      "max": 913
+    },
+    "smarm 24-thread": {
+      "result": 1000,
+      "median": 1308,
+      "min": 1227,
+      "max": 1396
     },
     "tokio current_thread": {
       "result": 1000,
-      "median": 879,
+      "median": 407,
-      "min": 868,
+      "min": 400,
-      "max": 973
+      "max": 444
     },
     "tokio multi-thread": {
       "result": 1000,
-      "median": 4328,
+      "median": 10869,
-      "min": 4223,
+      "min": 8683,
-      "max": 4461
+      "max": 11688
     }
   },
   "spawn_storm_busy": {
     "smarm 1-thread": {
       "result": 10000,
-      "median": 130058,
+      "median": 112045,
-      "min": 126790,
+      "min": 99936,
-      "max": 134475
+      "max": 117329
+    },
+    "smarm 24-thread": {
+      "result": 10000,
+      "median": 137105,
+      "min": 130852,
+      "max": 147707
     },
     "tokio current_thread": {
       "result": 10000,
-      "median": 2772,
+      "median": 1128,
-      "min": 2641,
+      "min": 1123,
-      "max": 4367
+      "max": 1435
     },
     "tokio multi-thread": {
       "result": 10000,
-      "median": 7462,
+      "median": 19674,
-      "min": 4469,
+      "min": 16013,
-      "max": 12892
+      "max": 27234
     }
   },
   "mpsc_contention": {
     "smarm 1-thread": {
       "result": 320000,
-      "median": 9260,
+      "median": 3667,
-      "min": 9095,
+      "min": 3608,
-      "max": 10081
+      "max": 4126
+    },
+    "smarm 24-thread": {
+      "result": 320000,
+      "median": 45681,
+      "min": 31908,
+      "max": 51287
     },
     "tokio current_thread": {
       "result": 320000,
-      "median": 17570,
+      "median": 6228,
-      "min": 17213,
+      "min": 6210,
-      "max": 18276
+      "max": 6514
     },
     "tokio multi-thread": {
       "result": 320000,
-      "median": 17593,
+      "median": 66173,
-      "min": 17452,
+      "min": 42208,
-      "max": 19564
+      "max": 83255
     }
   },
   "many_timers": {
     "smarm 1-thread": {
       "result": 10000,
-      "median": 135806,
+      "median": 119988,
-      "min": 132573,
+      "min": 107308,
-      "max": 141651
+      "max": 123557
+    },
+    "smarm 24-thread": {
+      "result": 10000,
+      "median": 218842,
+      "min": 182009,
+      "max": 256988
     },
     "tokio current_thread": {
       "result": 10000,
-      "median": 14462,
+      "median": 12432,
-      "min": 13555,
+      "min": 12308,
-      "max": 15457
+      "max": 13468
     },
     "tokio multi-thread": {
       "result": 10000,
-      "median": 15011,
+      "median": 16311,
-      "min": 14655,
+      "min": 15026,
-      "max": 15368
+      "max": 16897
     }
   },
   "multi_thread_scaling": {
     "smarm 1-thread": {
       "result": 33860,
-      "median": 30029,
+      "median": 14908,
-      "min": 29720,
+      "min": 14857,
-      "max": 31351
+      "max": 15218
+    },
+    "smarm 2-thread": {
+      "result": 33860,
+      "median": 7834,
+      "min": 7717,
+      "max": 8033
+    },
+    "smarm 4-thread": {
+      "result": 33860,
+      "median": 4393,
+      "min": 4326,
+      "max": 4435
+    },
+    "smarm 24-thread": {
+      "result": 33860,
+      "median": 2173,
+      "min": 2068,
+      "max": 2405
     },
     "tokio multi 1-thread": {
       "result": 33860,
-      "median": 28983,
+      "median": 14432,
-      "min": 28908,
+      "min": 14219,
-      "max": 29323
+      "max": 14763
+    },
+    "tokio multi 2-thread": {
+      "result": 33860,
+      "median": 7333,
+      "min": 7222,
+      "max": 7477
+    },
+    "tokio multi 4-thread": {
+      "result": 33860,
+      "median": 3741,
+      "min": 3681,
+      "max": 3876
+    },
+    "tokio multi 24-thread": {
+      "result": 33860,
+      "median": 1513,
+      "min": 1375,
+      "max": 1979
     }
   },
   "deep_recursion": {
     "smarm 1-thread": {
       "result": 1,
-      "median": 83,
+      "median": 102,
-      "min": 78,
+      "min": 96,
-      "max": 587
+      "max": 123
+    },
+    "smarm 24-thread": {
+      "result": 1,
+      "median": 597,
+      "min": 576,
+      "max": 682
     },
     "tokio current_thread": {
       "result": 1,
-      "median": 25,
+      "median": 13,
-      "min": 25,
+      "min": 11,
-      "max": 33
+      "max": 35
     },
     "tokio multi-thread": {
       "result": 1,
-      "median": 59,
+      "median": 56,
-      "min": 47,
+      "min": 46,
-      "max": 205
+      "max": 65
     }
   },
   "yield_in_hot_loop": {
     "smarm 1-thread": {
       "result": 1000000,
-      "median": 188753,
+      "median": 80680,
-      "min": 187007,
+      "min": 80308,
-      "max": 194366
+      "max": 81845
     },
     "tokio current_thread": {
       "result": 1000000,
-      "median": 153929,
+      "median": 72606,
-      "min": 152712,
+      "min": 72154,
-      "max": 158749
+      "max": 77206
     }
   },
   "uncontended_channel": {
     "smarm 1-thread": {
       "result": 1000000,
-      "median": 26811,
+      "median": 9257,
-      "min": 26498,
+      "min": 9223,
-      "max": 29069
+      "max": 12049
     },
     "tokio current_thread": {
       "result": 1000000,
-      "median": 51888,
+      "median": 16925,
-      "min": 51530,
+      "min": 16848,
-      "max": 52708
+      "max": 17019
     }
   },
   "catch_unwind_panics": {
     "smarm 1-thread": {
       "result": 10000,
-      "median": 142215,
+      "median": 116821,
-      "min": 140189,
+      "min": 111345,
-      "max": 143570
+      "max": 128261
+    },
+    "smarm 24-thread": {
+      "result": 10000,
+      "median": 117487,
+      "min": 107011,
+      "max": 129307
     },
     "tokio current_thread": {
       "result": 10000,
-      "median": 682295,
+      "median": 10425,
-      "min": 670281,
+      "min": 10141,
-      "max": 700774
+      "max": 10604
     },
     "tokio multi-thread": {
       "result": 10000,
-      "median": 662688,
+      "median": 6418,
-      "min": 641453,
+      "min": 3715,
-      "max": 681868
+      "max": 7144
     }
   }
 }

benches/smarm_favored.rs

@@ -272,6 +272,8 @@ fn bench_panic_smarm(threads: usize) -> (u64, u128) {
     let err = Arc::new(AtomicU64::new(0));
     let ok2 = ok.clone();
     let err2 = err.clone();
+    let prev_hook = std::panic::take_hook();
+    std::panic::set_hook(Box::new(|_| {}));
     let start = Instant::now();
     smarm::runtime::init(bench_cfg(threads)).run(move || {
         let mut handles = Vec::new();
@@ -289,6 +291,7 @@ fn bench_panic_smarm(threads: usize) -> (u64, u128) {
             }
         }
     });
+    std::panic::set_hook(prev_hook);
     let total = ok.load(Ordering::Relaxed) + err.load(Ordering::Relaxed);
     (total, start.elapsed().as_micros())
 }
@@ -299,6 +302,8 @@ fn bench_panic_tokio_current() -> (u64, u128) {
     let ok2 = ok.clone();
     let err2 = err.clone();
     let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
+    let prev_hook = std::panic::take_hook();
+    std::panic::set_hook(Box::new(|_| {}));
     let start = Instant::now();
     let local = tokio::task::LocalSet::new();
     local.block_on(&rt, async move {
@@ -317,6 +322,7 @@ fn bench_panic_tokio_current() -> (u64, u128) {
             }
         }
     });
+    std::panic::set_hook(prev_hook);
     let total = ok.load(Ordering::Relaxed) + err.load(Ordering::Relaxed);
     (total, start.elapsed().as_micros())
 }
@@ -330,6 +336,8 @@ fn bench_panic_tokio_multi() -> (u64, u128) {
         .worker_threads(available_threads())
         .build()
         .unwrap();
+    let prev_hook = std::panic::take_hook();
+    std::panic::set_hook(Box::new(|_| {}));
     let start = Instant::now();
     rt.block_on(async move {
         let mut handles = Vec::new();
@@ -347,6 +355,7 @@ fn bench_panic_tokio_multi() -> (u64, u128) {
             }
         }
     });
+    std::panic::set_hook(prev_hook);
     let total = ok.load(Ordering::Relaxed) + err.load(Ordering::Relaxed);
     (total, start.elapsed().as_micros())
 }

src/runtime.rs

@@ -72,6 +72,7 @@ pub struct Config {
     exact: Option<usize>,
     alloc_interval: u32,
     timeslice_cycles: u64,
+    stack_pool_cap: usize,
 }
 
 impl Config {
@@ -82,6 +83,7 @@ impl Config {
             min: n, max: n, exact: Some(n),
             alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
             timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
+            stack_pool_cap: n * 4,
         }
     }
 
@@ -96,6 +98,7 @@ impl Config {
             min, max, exact,
             alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
             timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
+            stack_pool_cap: max * 4,
         }
     }
 
@@ -116,6 +119,14 @@ impl Config {
         self
     }
 
+    /// Maximum number of stacks kept in the pool for reuse across spawns.
+    /// A larger cap reduces `mmap`/`munmap` syscalls at the cost of idle memory.
+    /// Default: `thread_count * 4`.
+    pub fn stack_pool_cap(mut self, n: usize) -> Self {
+        self.stack_pool_cap = n;
+        self
+    }
+
     /// The number of scheduler threads this config resolves to.
     pub fn resolved_thread_count(&self) -> usize {
         if let Some(e) = self.exact {
@@ -137,6 +148,7 @@ impl Default for Config {
             min: 1, max: avail, exact: None,
             alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
             timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
+            stack_pool_cap: avail * 4,
         }
     }
 }
@@ -244,8 +256,10 @@ pub(crate) struct SharedState {
     pub(crate) root_pid: Option<Pid>,
     pub(crate) timers: Timers,
     pub(crate) io: Option<IoThread>,
-    /// Closures awaiting their first resume, keyed by Pid.
+    /// Closures awaiting their first resume, indexed by slot index.
-    pub(crate) pending_closures: Vec<(Pid, Closure)>,
+    /// `pending_closures[idx]` is `Some` iff the actor at slot `idx` has not
+    /// yet been resumed for the first time.  Grows on demand; never shrinks.
+    pub(crate) pending_closures: Vec<Option<Closure>>,
 }
 
 impl SharedState {
@@ -257,7 +271,7 @@ impl SharedState {
             root_pid: None,
             timers: Timers::new(),
             io: None,
-            pending_closures: Vec::new(),
+            pending_closures: Vec::new(), // indexed by slot index
         }
     }
 
@@ -283,8 +297,9 @@ impl SharedState {
     }
 
     pub(crate) fn pop_pending_closure(&mut self, pid: Pid) -> Option<Closure> {
-        let pos = self.pending_closures.iter().position(|(p, _)| *p == pid)?;
+        self.pending_closures
-        Some(self.pending_closures.swap_remove(pos).1)
+            .get_mut(pid.index() as usize)
+            .and_then(|cell| cell.take())
     }
 }
 
@@ -304,10 +319,15 @@ pub(crate) struct RuntimeInner {
     /// Preemption knobs, written into each scheduler thread's locals on startup.
     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<Vec<crate::stack::Stack>>,
+    /// 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) -> Arc<Self> {
+    fn new(thread_count: usize, alloc_interval: u32, timeslice_cycles: u64, stack_pool_cap: usize) -> Arc<Self> {
         let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
         Arc::new(Self {
             shared: Mutex::new(SharedState::new()),
@@ -317,6 +337,8 @@ impl RuntimeInner {
             sleeping: AtomicU32::new(0),
             alloc_interval,
             timeslice_cycles,
+            stack_pool: Mutex::new(Vec::new()),
+            stack_pool_cap,
         })
     }
 
@@ -346,7 +368,7 @@ 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),
+        inner: RuntimeInner::new(n, config.alloc_interval, config.timeslice_cycles, config.stack_pool_cap),
         thread_count: n,
     }
 }
@@ -514,15 +536,28 @@ fn finalize_actor(inner: &Arc<RuntimeInner>, pid: Pid, outcome: Outcome) {
         ),
     };
 
-    let (waiters, supervisor_pid) = inner.with_shared(|s| {
+    let (waiters, supervisor_pid, recycled_stack) = 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;
-        slot.actor = None;
+        (std::mem::take(&mut slot.waiters), sup, stack)
-        (std::mem::take(&mut slot.waiters), sup)
     });
 
+    // 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();
+        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| {
@@ -568,11 +603,24 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
                     crate::timer::Reason::Sleep => {
                         inner.with_shared(|s| {
                             if let Some(slot) = s.slot_mut(entry.pid) {
-                                if matches!(slot.state, State::Parked) {
+                                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 => {}
+                                }
                             }
                         });
                     }
@@ -640,29 +688,58 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
         } // drain_guard drops here
 
         // ----------------------------------------------------------------
-        // 2. Pop a runnable actor from the shared queue.
+        // 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<first-resume closure>))  — got work
+        //          None with all_clear=true                       — done, exit
+        //          None with all_clear=false                      — idle, wait
         // ----------------------------------------------------------------
-        let pid = match inner.with_shared(|s| {
+        enum PopResult {
+            Work(Pid, usize, Option<Closure>),
+            Idle {
+                next_deadline: Option<std::time::Instant>,
+                io_outstanding: u32,
+                wake_fd: Option<std::os::fd::RawFd>,
+            },
+            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
+
+        let pop = inner.with_shared(|s| {
             let len = s.run_queue.len() as u64;
             stats.run_queue_len.store(len, Ordering::Relaxed);
-            s.run_queue.pop_front()
+
-        }) {
+            if let Some(pid) = s.run_queue.pop_front() {
-            Some(p) => {
+                // Happy path: got a PID, grab SP and optional first closure.
-                crate::te!(crate::trace::Event::Dequeue(p));
+                let sp = s.slot(pid)
-                p
+                    .and_then(|slot| slot.actor.as_ref().map(|a| a.sp));
+                match sp {
+                    Some(sp) => {
+                        let closure = s.pop_pending_closure(pid);
+                        PopResult::Work(pid, sp, closure)
                     }
                     None => {
-                // Queue was empty when we popped. Re-examine under the lock to
+                        // Stale PID (actor already finalized). Treat as idle
-                // decide whether to exit or wait. All four conditions must hold
+                        // so the outer loop retries immediately.
-                // simultaneously before we exit:
+                        PopResult::Idle {
+                            next_deadline: None,
+                            io_outstanding: 0,
+                            wake_fd: None,
+                        }
+                    }
+                }
+            } else {
+                // Queue was empty. Re-examine to decide exit vs wait.
+                // All four conditions must hold simultaneously before we exit:
                 //   1. run queue is still empty
                 //   2. no live actors (nothing parked, nothing mid-finalize)
                 //   3. no pending timers
                 //   4. no outstanding IO
-                // If any is non-zero we keep spinning — "check the fridge is
-                // empty before you leave for the airport".
-                let (next_deadline, io_outstanding, wake_fd, all_clear) =
-                    inner.with_shared(|s| {
                 let next = s.timers.peek_deadline();
                 let (out, fd) = match s.io.as_ref() {
                     Some(io) => (
@@ -672,15 +749,23 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
                     None => (0, None),
                 };
                 let live = s.slots.iter().filter(|slot| slot.actor.is_some()).count();
-                        let queue_empty = s.run_queue.is_empty();
+                let all_clear = s.run_queue.is_empty() && live == 0
-                        let all_clear = queue_empty && live == 0 && next.is_none() && out == 0;
+                    && next.is_none() && out == 0;
-                        (next, out, fd, all_clear)
+                if all_clear {
+                    PopResult::AllDone
+                } else {
+                    PopResult::Idle { next_deadline: next, io_outstanding: out, wake_fd: fd }
+                }
+            }
         });
 
-                if all_clear {
+        let (pid, sp, first_closure) = match pop {
-                    return;
+            PopResult::Work(pid, sp, closure) => {
+                crate::te!(crate::trace::Event::Dequeue(pid));
+                (pid, sp, closure)
             }
-
+            PopResult::AllDone => return,
+            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.
                 match (next_deadline, wake_fd) {
@@ -712,17 +797,9 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
         // ----------------------------------------------------------------
         // 3. Resume the actor.
         // ----------------------------------------------------------------
-        let sp = match inner.with_shared(|s| {
-            s.slot(pid).and_then(|slot| slot.actor.as_ref().map(|a| a.sp))
-        }) {
-            Some(sp) => sp,
-            None => {
-                continue; // stale pid
-            }
-        };
 
         // First resume: move the closure into the trampoline's thread-local.
-        if let Some(b) = inner.with_shared(|s| s.pop_pending_closure(pid)) {
+        if let Some(b) = first_closure {
             set_current_actor_box(b);
         }
 

src/scheduler.rs

@@ -132,13 +132,20 @@ pub fn spawn(f: impl FnOnce() + Send + 'static) -> JoinHandle {
 }
 
 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())
+        .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 pid = with_runtime(|inner| {
         inner.with_shared(|s| {
             let (idx, gen) = s.allocate_slot();
             let pid = Pid::new(idx, gen);
-            let stack = crate::stack::Stack::new(crate::runtime::ACTOR_STACK_SIZE)
-                .expect("stack allocation failed");
-            let sp = init_actor_stack(stack.top(), crate::actor::trampoline);
             let slot = &mut s.slots[idx as usize];
             slot.actor = Some(crate::actor::Actor { pid, stack, sp, supervisor });
             slot.state = crate::runtime::State::Runnable;
@@ -149,7 +156,12 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHa
             slot.pending_unpark = false;
             slot.pending_io_result = None;
             s.run_queue.push_back(pid);
-            s.pending_closures.push((pid, Box::new(f) as crate::runtime::Closure));
+            // Grow the closures vec to cover this slot index, then store.
+            let idx = idx as usize;
+            if s.pending_closures.len() <= idx {
+                s.pending_closures.resize_with(idx + 1, || None);
+            }
+            s.pending_closures[idx] = Some(Box::new(f) as crate::runtime::Closure);
             crate::te!(crate::trace::Event::Spawn { parent: supervisor, child: pid });
             crate::te!(crate::trace::Event::Enqueue(pid));
             pid
@@ -344,6 +356,3 @@ pub fn register_supervisor_channel(pid: Pid, sender: Sender<Signal>) {
 pub fn run<F: FnOnce() + Send + 'static>(f: F) {
     crate::runtime::init(crate::runtime::Config::exact(1)).run(f);
 }
-
-
-

tests/many_timers_multi_thread.rs

@@ -0,0 +1,98 @@
+//! Regression test: multi-thread sleep timer lost-wakeup.
+//!
+//! Lifted from the `many_timers` workload in `benches/tokio_favored.rs`:
+//! N actors each sleep for a short randomised duration and join. On a
+//! multi-threaded runtime this hangs because the timer drain path drops
+//! wakeups for actors that are still in `State::Runnable` at drain time
+//! (i.e. between `timers.insert_sleep` and `park_current` in `sleep()`).
+//!
+//! The test runs the workload on a background OS thread with a wall-clock
+//! watchdog so a deadlock fails the test instead of hanging the suite.
+
+use smarm::runtime::Config;
+use std::sync::mpsc;
+use std::thread;
+use std::time::{Duration, Instant};
+
+// Same shape as benches/tokio_favored.rs::timer_delay_ms — deterministic
+// per-actor delay so the test is comparable across runs.
+const TIMER_ACTORS: u64 = 10_000;
+const TIMER_MIN_MS: u64 = 1;
+const TIMER_MAX_MS: u64 = 10;
+
+fn timer_delay_ms(i: u64) -> u64 {
+    TIMER_MIN_MS + (i.wrapping_mul(2654435761u64) >> 32) % (TIMER_MAX_MS - TIMER_MIN_MS + 1)
+}
+
+fn many_timers_workload(threads: usize) {
+    smarm::runtime::init(Config::exact(threads)).run(|| {
+        let mut handles = Vec::with_capacity(TIMER_ACTORS as usize);
+        for i in 0..TIMER_ACTORS {
+            let ms = timer_delay_ms(i);
+            handles.push(smarm::spawn(move || {
+                smarm::sleep(Duration::from_millis(ms));
+            }));
+        }
+        for h in handles {
+            h.join().unwrap();
+        }
+    });
+}
+
+/// Wall-clock budget for the workload. The 1-thread version completes in
+/// ~130ms on the reporter's machine; the workload's intrinsic minimum is
+/// `TIMER_MAX_MS` (10ms). Anything past several seconds is a hang.
+const WATCHDOG: Duration = Duration::from_secs(15);
+
+fn run_with_watchdog(threads: usize) {
+    let (done_tx, done_rx) = mpsc::channel::<()>();
+    let worker = thread::Builder::new()
+        .name(format!("many_timers-{}-thread", threads))
+        .spawn(move || {
+            many_timers_workload(threads);
+            let _ = done_tx.send(());
+        })
+        .expect("spawn worker thread");
+
+    let start = Instant::now();
+    match done_rx.recv_timeout(WATCHDOG) {
+        Ok(()) => {
+            let _ = worker.join();
+            eprintln!(
+                "many_timers ({} threads): completed in {:?}",
+                threads,
+                start.elapsed()
+            );
+        }
+        Err(_) => {
+            // The worker is wedged on a smarm scheduler that won't unwind.
+            // We can't join it; report and abort so the test fails loudly
+            // instead of the harness hanging on drop.
+            eprintln!(
+                "many_timers ({} threads): DEADLOCK — no progress in {:?}",
+                threads, WATCHDOG
+            );
+            std::process::exit(101);
+        }
+    }
+}
+
+#[test]
+fn many_timers_single_thread_completes() {
+    run_with_watchdog(1);
+}
+
+#[test]
+fn many_timers_multi_thread_completes() {
+    // Repro: this is the case that hangs in benches/tokio_favored.rs.
+    //
+    // The bug needs ≥2 scheduler threads (one draining timers while
+    // another is running sleep() between insert and park_current). The
+    // reporter hit it at 24; we use max(8, available_parallelism()) so
+    // single-CPU CI sandboxes still trigger it.
+    let n = thread::available_parallelism()
+        .map(|n| n.get())
+        .unwrap_or(1)
+        .max(8);
+    run_with_watchdog(n);
+}

tests/runtime.rs

@@ -421,3 +421,67 @@ fn ping_pong_completes() {
     });
     assert_eq!(final_val.load(Ordering::SeqCst), ROUNDS);
 }
+
+/// Regression test for the multi-scheduler timer-only hang.
+///
+/// Bug: when the run queue is empty and only timers are pending (no IO
+/// outstanding), all N scheduler threads called `poll_wake(wake_fd,
+/// Some(timeout))` on the *same* pipe fd. When the timer fired, the one
+/// thread that won `drain_lock` consumed the single wake byte and re-queued
+/// the actors; the other N-1 threads stayed blocked in `poll()` for the full
+/// timeout duration. After actors completed and `all_clear` became true, the
+/// stuck threads had to wait out the remainder of their poll timeout before
+/// noticing — adding up to one full sleep duration of extra latency.
+///
+/// Fix: use `thread::sleep(timeout)` when `io_outstanding == 0`, so every
+/// scheduler thread independently wakes at the deadline without contending
+/// on the wake pipe.
+///
+/// Signal: with SLEEP_MS=300 and ≥2 scheduler threads, the broken impl
+/// takes ≥2×SLEEP_MS (actors sleep + stuck threads drain their poll timeout
+/// before seeing all_clear). The fixed impl takes ≈SLEEP_MS + epsilon.
+#[test]
+fn multi_thread_timer_only_no_pipe_contention() {
+    const SLEEP_MS: u64 = 300;
+    const ACTORS: usize = 100;
+
+    // Need at least 2 scheduler threads: one wins drain_lock and does work,
+    // the rest pile into poll_wake and get stuck.
+    let n = std::thread::available_parallelism()
+        .map(|n| n.get())
+        .unwrap_or(1)
+        .max(2);
+    let r = smarm::runtime::init(Config::exact(n.min(4)));
+
+    let count = Arc::new(AtomicU64::new(0));
+    let c = count.clone();
+    let start = std::time::Instant::now();
+
+    r.run(move || {
+        let mut handles = Vec::new();
+        for _ in 0..ACTORS {
+            let cc = c.clone();
+            handles.push(spawn(move || {
+                // Pure timer sleep: no channels, no IO.
+                // All scheduler threads will be idle simultaneously while
+                // these timers are pending — the condition that triggers the bug.
+                smarm::sleep(Duration::from_millis(SLEEP_MS));
+                cc.fetch_add(1, Ordering::SeqCst);
+            }));
+        }
+        for h in handles {
+            h.join().unwrap();
+        }
+    });
+
+    assert_eq!(count.load(Ordering::SeqCst), ACTORS as u64, "not all actors completed");
+
+    let elapsed = start.elapsed();
+    assert!(
+        elapsed < Duration::from_millis(SLEEP_MS * 2),
+        "run() took {elapsed:?} — expected <{}ms; likely stuck threads draining \
+         their poll_wake timeout after all_clear (bug: scheduler threads poll \
+         the shared wake pipe instead of sleeping independently for timer-only workloads)",
+        SLEEP_MS * 2,
+    );
+}