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.
perf: reduce scheduler mutex contention + stack pool
2026-05-26 21:58:14 +02:00 · 2026-05-25 23:28:11 +02:00 · 2026-05-25 23:23:12 +02:00
8 changed files with 521 additions and 164 deletions
@@ -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
@@ -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
-and do something interesting, just send me an email. I'd genuinely like to
-hear about 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. 

 ---

-<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>
+
@@ -2,223 +2,313 @@
  "chained_spawn": {
    "smarm 1-thread": {
      "result": 1000,
-      "median": 8637,
-      "min": 8553,
-      "max": 8933
+      "median": 266,
+      "min": 242,
+      "max": 351
+    },
+    "smarm 24-thread": {
+      "result": 1000,
+      "median": 742,
+      "min": 696,
+      "max": 860
    },
    "tokio current_thread": {
      "result": 1000,
-      "median": 124,
-      "min": 124,
-      "max": 153
+      "median": 62,
+      "min": 61,
+      "max": 68
    },
    "tokio multi-thread": {
      "result": 1000,
-      "median": 188,
-      "min": 183,
-      "max": 229
+      "median": 190,
+      "min": 169,
+      "max": 207
    }
  },
  "yield_many": {
    "smarm 1-thread": {
      "result": 200000,
-      "median": 41622,
-      "min": 41063,
-      "max": 44973
+      "median": 19071,
+      "min": 18776,
+      "max": 19396
+    },
+    "smarm 24-thread": {
+      "result": 200000,
+      "median": 172454,
+      "min": 166246,
+      "max": 174230
    },
    "tokio current_thread": {
      "result": 200000,
-      "median": 15085,
-      "min": 15013,
-      "max": 15274
+      "median": 4737,
+      "min": 4644,
+      "max": 5065
    },
    "tokio multi-thread": {
      "result": 200000,
-      "median": 15964,
-      "min": 15880,
-      "max": 17959
+      "median": 8738,
+      "min": 7852,
+      "max": 9770
    }
  },
  "fan_out_compute": {
    "smarm 1-thread": {
      "result": 33860,
-      "median": 29727,
-      "min": 29491,
-      "max": 31634
+      "median": 13234,
+      "min": 13196,
+      "max": 13390
+    },
+    "smarm 24-thread": {
+      "result": 33860,
+      "median": 2244,
+      "min": 2162,
+      "max": 2380
    },
    "tokio current_thread": {
      "result": 33860,
-      "median": 28503,
-      "min": 28391,
-      "max": 28866
+      "median": 14049,
+      "min": 14035,
+      "max": 14300
    },
    "tokio multi-thread": {
      "result": 33860,
-      "median": 34542,
-      "min": 34396,
-      "max": 36111
+      "median": 1474,
+      "min": 1285,
+      "max": 1823
    }
  },
  "ping_pong_oneshot": {
    "smarm 1-thread": {
      "result": 1000,
-      "median": 16848,
-      "min": 16633,
-      "max": 17301
+      "median": 751,
+      "min": 727,
+      "max": 913
+    },
+    "smarm 24-thread": {
+      "result": 1000,
+      "median": 1308,
+      "min": 1227,
+      "max": 1396
    },
    "tokio current_thread": {
      "result": 1000,
-      "median": 879,
-      "min": 868,
-      "max": 973
+      "median": 407,
+      "min": 400,
+      "max": 444
    },
    "tokio multi-thread": {
      "result": 1000,
-      "median": 4328,
-      "min": 4223,
-      "max": 4461
+      "median": 10869,
+      "min": 8683,
+      "max": 11688
    }
  },
  "spawn_storm_busy": {
    "smarm 1-thread": {
      "result": 10000,
-      "median": 130058,
-      "min": 126790,
-      "max": 134475
+      "median": 112045,
+      "min": 99936,
+      "max": 117329
+    },
+    "smarm 24-thread": {
+      "result": 10000,
+      "median": 137105,
+      "min": 130852,
+      "max": 147707
    },
    "tokio current_thread": {
      "result": 10000,
-      "median": 2772,
-      "min": 2641,
-      "max": 4367
+      "median": 1128,
+      "min": 1123,
+      "max": 1435
    },
    "tokio multi-thread": {
      "result": 10000,
-      "median": 7462,
-      "min": 4469,
-      "max": 12892
+      "median": 19674,
+      "min": 16013,
+      "max": 27234
    }
  },
  "mpsc_contention": {
    "smarm 1-thread": {
      "result": 320000,
-      "median": 9260,
-      "min": 9095,
-      "max": 10081
+      "median": 3667,
+      "min": 3608,
+      "max": 4126
+    },
+    "smarm 24-thread": {
+      "result": 320000,
+      "median": 45681,
+      "min": 31908,
+      "max": 51287
    },
    "tokio current_thread": {
      "result": 320000,
-      "median": 17570,
-      "min": 17213,
-      "max": 18276
+      "median": 6228,
+      "min": 6210,
+      "max": 6514
    },
    "tokio multi-thread": {
      "result": 320000,
-      "median": 17593,
-      "min": 17452,
-      "max": 19564
+      "median": 66173,
+      "min": 42208,
+      "max": 83255
    }
  },
  "many_timers": {
    "smarm 1-thread": {
      "result": 10000,
-      "median": 135806,
-      "min": 132573,
-      "max": 141651
+      "median": 119988,
+      "min": 107308,
+      "max": 123557
+    },
+    "smarm 24-thread": {
+      "result": 10000,
+      "median": 218842,
+      "min": 182009,
+      "max": 256988
    },
    "tokio current_thread": {
      "result": 10000,
-      "median": 14462,
-      "min": 13555,
-      "max": 15457
+      "median": 12432,
+      "min": 12308,
+      "max": 13468
    },
    "tokio multi-thread": {
      "result": 10000,
-      "median": 15011,
-      "min": 14655,
-      "max": 15368
+      "median": 16311,
+      "min": 15026,
+      "max": 16897
    }
  },
  "multi_thread_scaling": {
    "smarm 1-thread": {
      "result": 33860,
-      "median": 30029,
-      "min": 29720,
-      "max": 31351
+      "median": 14908,
+      "min": 14857,
+      "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,
-      "min": 28908,
-      "max": 29323
+      "median": 14432,
+      "min": 14219,
+      "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,
-      "min": 78,
-      "max": 587
+      "median": 102,
+      "min": 96,
+      "max": 123
+    },
+    "smarm 24-thread": {
+      "result": 1,
+      "median": 597,
+      "min": 576,
+      "max": 682
    },
    "tokio current_thread": {
      "result": 1,
-      "median": 25,
-      "min": 25,
-      "max": 33
+      "median": 13,
+      "min": 11,
+      "max": 35
    },
    "tokio multi-thread": {
      "result": 1,
-      "median": 59,
-      "min": 47,
-      "max": 205
+      "median": 56,
+      "min": 46,
+      "max": 65
    }
  },
  "yield_in_hot_loop": {
    "smarm 1-thread": {
      "result": 1000000,
-      "median": 188753,
-      "min": 187007,
-      "max": 194366
+      "median": 80680,
+      "min": 80308,
+      "max": 81845
    },
    "tokio current_thread": {
      "result": 1000000,
-      "median": 153929,
-      "min": 152712,
-      "max": 158749
+      "median": 72606,
+      "min": 72154,
+      "max": 77206
    }
  },
  "uncontended_channel": {
    "smarm 1-thread": {
      "result": 1000000,
-      "median": 26811,
-      "min": 26498,
-      "max": 29069
+      "median": 9257,
+      "min": 9223,
+      "max": 12049
    },
    "tokio current_thread": {
      "result": 1000000,
-      "median": 51888,
-      "min": 51530,
-      "max": 52708
+      "median": 16925,
+      "min": 16848,
+      "max": 17019
    }
  },
  "catch_unwind_panics": {
    "smarm 1-thread": {
      "result": 10000,
-      "median": 142215,
-      "min": 140189,
-      "max": 143570
+      "median": 116821,
+      "min": 111345,
+      "max": 128261
+    },
+    "smarm 24-thread": {
+      "result": 10000,
+      "median": 117487,
+      "min": 107011,
+      "max": 129307
    },
    "tokio current_thread": {
      "result": 10000,
-      "median": 682295,
-      "min": 670281,
-      "max": 700774
+      "median": 10425,
+      "min": 10141,
+      "max": 10604
    },
    "tokio multi-thread": {
      "result": 10000,
-      "median": 662688,
-      "min": 641453,
-      "max": 681868
+      "median": 6418,
+      "min": 3715,
+      "max": 7144
    }
  }
 }
@@ -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())
 }
@@ -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.
-    pub(crate) pending_closures: Vec<(Pid, Closure)>,
+    /// Closures awaiting their first resume, indexed by slot index.
+    /// `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)?;
-        Some(self.pending_closures.swap_remove(pos).1)
+        self.pending_closures
+            .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)
+        (std::mem::take(&mut slot.waiters), sup, stack)
    });

+    // 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,10 +603,23 @@ 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) {
-                                    slot.state = State::Runnable;
-                                    s.run_queue.push_back(entry.pid);
-                                    crate::te!(crate::trace::Event::Enqueue(entry.pid));
+                                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,47 +688,84 @@ 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()
-        }) {
-            Some(p) => {
-                crate::te!(crate::trace::Event::Dequeue(p));
-                p
-            }
-            None => {
-                // Queue was empty when we popped. Re-examine under the lock to
-                // decide whether to exit or wait. All four conditions must hold
-                // simultaneously before we exit:
+
+            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) => {
+                        let closure = s.pop_pending_closure(pid);
+                        PopResult::Work(pid, sp, 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,
+                        }
+                    }
+                }
+            } 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) => (
-                                io.outstanding + io.waiters.len() as u32,
-                                Some(io.wake_fd()),
-                            ),
-                            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 = queue_empty && live == 0 && next.is_none() && out == 0;
-                        (next, out, fd, all_clear)
-                    });
-
+                let next = s.timers.peek_deadline();
+                let (out, fd) = match s.io.as_ref() {
+                    Some(io) => (
+                        io.outstanding + io.waiters.len() as u32,
+                        Some(io.wake_fd()),
+                    ),
+                    None => (0, None),
+                };
+                let live = s.slots.iter().filter(|slot| slot.actor.is_some()).count();
+                let all_clear = s.run_queue.is_empty() && live == 0
+                    && next.is_none() && out == 0;
                if all_clear {
-                    return;
+                    PopResult::AllDone
+                } else {
+                    PopResult::Idle { next_deadline: next, io_outstanding: out, wake_fd: fd }
                }
+            }
+        });

+        let (pid, sp, first_closure) = match pop {
+            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);
        }

@@ -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);
 }
-
-
-
@@ -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);
+}
@@ -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,
+    );
+}