Compare commits

...
3 Commits
Author SHA1 Message Date
smarm 7746dca69b 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.
2026-05-26 21:58:14 +02:00
smarm 72f5d38e5d perf: reduce scheduler mutex contention + stack pool
Three changes, each independently measured, landed together.

Fuse pre-resume mutex acquisitions
-----------------------------------
The schedule loop previously took the shared lock three separate times
per actor resume: once to pop the run queue, once to read the stack
pointer, and once to pop the first-resume closure. These are now a
single acquisition that returns everything needed to resume the actor.

As a side effect, pending_closures was changed from a Vec<(Pid, Closure)>
with O(n) linear scan to a Vec<Option<Closure>> indexed by slot index,
making first-closure lookup O(1).

Move stack allocation outside the shared lock
----------------------------------------------
Stack::new() (mmap + mprotect) was previously called inside with_shared,
stalling every other scheduler thread for the duration of two syscalls on
every spawn. It now runs before the lock is acquired.

Stack pool
----------
Rather than munmap-ing a stack when an actor finishes and mmap-ing a fresh
one on the next spawn, stacks are now recycled through a per-Runtime pool
(Mutex<Vec<Stack>>). finalize_actor extracts the stack from the Actor
before clearing the slot and pushes it to the pool outside the shared lock.
spawn_under pops from the pool before falling back to Stack::new().

The pool is unbounded for now (shrink policy TBD) but capped at
stack_pool_cap stacks on return, defaulting to thread_count * 4.
The cap is configurable via Config::stack_pool_cap(n).

Results (24-thread, against stored baseline)
--------------------------------------------
chained_spawn      smarm 1-thread:   9763 → 261 µs   (-97%)
chained_spawn      smarm 24-thread: 23562 → 838 µs   (-96%)
ping_pong_oneshot  smarm 1-thread:  18409 → 742 µs   (-96%)
ping_pong_oneshot  smarm 24-thread: 44596 → 1425 µs  (-97%)
catch_unwind_panics smarm 24-thread: 267812 → 124094 µs (-54%)
fan_out_compute    smarm 24-thread:   2839 → 2226 µs  (-22%)

Tokio regressions in the checker output are baseline measurement drift;
no tokio code was changed.
2026-05-25 23:28:11 +02:00
smarm 64be3f23ac Baseline benchmarks on my machine 2026-05-25 23:23:12 +02:00
8 changed files with 521 additions and 164 deletions
+12
View File
@@ -29,3 +29,15 @@ harness = false
[[bench]] [[bench]]
name = "multi_scheduler" name = "multi_scheduler"
harness = false harness = false
[[bench]]
name = "general"
harness = false
[[bench]]
name = "smarm_favored"
harness = false
[[bench]]
name = "tokio_favored"
harness = false
+2 -4
View File
@@ -91,10 +91,8 @@ mechanism we know how to add; none belongs in this iteration.
## Contributing ## 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>
+189 -99
View File
@@ -2,223 +2,313 @@
"chained_spawn": { "chained_spawn": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 1000, "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": { "tokio current_thread": {
"result": 1000, "result": 1000,
"median": 124, "median": 62,
"min": 124, "min": 61,
"max": 153 "max": 68
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 1000, "result": 1000,
"median": 188, "median": 190,
"min": 183, "min": 169,
"max": 229 "max": 207
} }
}, },
"yield_many": { "yield_many": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 200000, "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": { "tokio current_thread": {
"result": 200000, "result": 200000,
"median": 15085, "median": 4737,
"min": 15013, "min": 4644,
"max": 15274 "max": 5065
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 200000, "result": 200000,
"median": 15964, "median": 8738,
"min": 15880, "min": 7852,
"max": 17959 "max": 9770
} }
}, },
"fan_out_compute": { "fan_out_compute": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 33860, "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": { "tokio current_thread": {
"result": 33860, "result": 33860,
"median": 28503, "median": 14049,
"min": 28391, "min": 14035,
"max": 28866 "max": 14300
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 33860, "result": 33860,
"median": 34542, "median": 1474,
"min": 34396, "min": 1285,
"max": 36111 "max": 1823
} }
}, },
"ping_pong_oneshot": { "ping_pong_oneshot": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 1000, "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": { "tokio current_thread": {
"result": 1000, "result": 1000,
"median": 879, "median": 407,
"min": 868, "min": 400,
"max": 973 "max": 444
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 1000, "result": 1000,
"median": 4328, "median": 10869,
"min": 4223, "min": 8683,
"max": 4461 "max": 11688
} }
}, },
"spawn_storm_busy": { "spawn_storm_busy": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 10000, "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": { "tokio current_thread": {
"result": 10000, "result": 10000,
"median": 2772, "median": 1128,
"min": 2641, "min": 1123,
"max": 4367 "max": 1435
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 10000, "result": 10000,
"median": 7462, "median": 19674,
"min": 4469, "min": 16013,
"max": 12892 "max": 27234
} }
}, },
"mpsc_contention": { "mpsc_contention": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 320000, "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": { "tokio current_thread": {
"result": 320000, "result": 320000,
"median": 17570, "median": 6228,
"min": 17213, "min": 6210,
"max": 18276 "max": 6514
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 320000, "result": 320000,
"median": 17593, "median": 66173,
"min": 17452, "min": 42208,
"max": 19564 "max": 83255
} }
}, },
"many_timers": { "many_timers": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 10000, "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": { "tokio current_thread": {
"result": 10000, "result": 10000,
"median": 14462, "median": 12432,
"min": 13555, "min": 12308,
"max": 15457 "max": 13468
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 10000, "result": 10000,
"median": 15011, "median": 16311,
"min": 14655, "min": 15026,
"max": 15368 "max": 16897
} }
}, },
"multi_thread_scaling": { "multi_thread_scaling": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 33860, "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": { "tokio multi 1-thread": {
"result": 33860, "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": { "deep_recursion": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 1, "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": { "tokio current_thread": {
"result": 1, "result": 1,
"median": 25, "median": 13,
"min": 25, "min": 11,
"max": 33 "max": 35
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 1, "result": 1,
"median": 59, "median": 56,
"min": 47, "min": 46,
"max": 205 "max": 65
} }
}, },
"yield_in_hot_loop": { "yield_in_hot_loop": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 1000000, "result": 1000000,
"median": 188753, "median": 80680,
"min": 187007, "min": 80308,
"max": 194366 "max": 81845
}, },
"tokio current_thread": { "tokio current_thread": {
"result": 1000000, "result": 1000000,
"median": 153929, "median": 72606,
"min": 152712, "min": 72154,
"max": 158749 "max": 77206
} }
}, },
"uncontended_channel": { "uncontended_channel": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 1000000, "result": 1000000,
"median": 26811, "median": 9257,
"min": 26498, "min": 9223,
"max": 29069 "max": 12049
}, },
"tokio current_thread": { "tokio current_thread": {
"result": 1000000, "result": 1000000,
"median": 51888, "median": 16925,
"min": 51530, "min": 16848,
"max": 52708 "max": 17019
} }
}, },
"catch_unwind_panics": { "catch_unwind_panics": {
"smarm 1-thread": { "smarm 1-thread": {
"result": 10000, "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": { "tokio current_thread": {
"result": 10000, "result": 10000,
"median": 682295, "median": 10425,
"min": 670281, "min": 10141,
"max": 700774 "max": 10604
}, },
"tokio multi-thread": { "tokio multi-thread": {
"result": 10000, "result": 10000,
"median": 662688, "median": 6418,
"min": 641453, "min": 3715,
"max": 681868 "max": 7144
} }
} }
} }
+9
View File
@@ -272,6 +272,8 @@ fn bench_panic_smarm(threads: usize) -> (u64, u128) {
let err = Arc::new(AtomicU64::new(0)); let err = Arc::new(AtomicU64::new(0));
let ok2 = ok.clone(); let ok2 = ok.clone();
let err2 = err.clone(); let err2 = err.clone();
let prev_hook = std::panic::take_hook();
std::panic::set_hook(Box::new(|_| {}));
let start = Instant::now(); let start = Instant::now();
smarm::runtime::init(bench_cfg(threads)).run(move || { smarm::runtime::init(bench_cfg(threads)).run(move || {
let mut handles = Vec::new(); 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); let total = ok.load(Ordering::Relaxed) + err.load(Ordering::Relaxed);
(total, start.elapsed().as_micros()) (total, start.elapsed().as_micros())
} }
@@ -299,6 +302,8 @@ fn bench_panic_tokio_current() -> (u64, u128) {
let ok2 = ok.clone(); let ok2 = ok.clone();
let err2 = err.clone(); let err2 = err.clone();
let rt = tokio::runtime::Builder::new_current_thread().build().unwrap(); 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 start = Instant::now();
let local = tokio::task::LocalSet::new(); let local = tokio::task::LocalSet::new();
local.block_on(&rt, async move { 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); let total = ok.load(Ordering::Relaxed) + err.load(Ordering::Relaxed);
(total, start.elapsed().as_micros()) (total, start.elapsed().as_micros())
} }
@@ -330,6 +336,8 @@ fn bench_panic_tokio_multi() -> (u64, u128) {
.worker_threads(available_threads()) .worker_threads(available_threads())
.build() .build()
.unwrap(); .unwrap();
let prev_hook = std::panic::take_hook();
std::panic::set_hook(Box::new(|_| {}));
let start = Instant::now(); let start = Instant::now();
rt.block_on(async move { rt.block_on(async move {
let mut handles = Vec::new(); 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); let total = ok.load(Ordering::Relaxed) + err.load(Ordering::Relaxed);
(total, start.elapsed().as_micros()) (total, start.elapsed().as_micros())
} }
+117 -40
View File
@@ -72,6 +72,7 @@ pub struct Config {
exact: Option<usize>, exact: Option<usize>,
alloc_interval: u32, alloc_interval: u32,
timeslice_cycles: u64, timeslice_cycles: u64,
stack_pool_cap: usize,
} }
impl Config { impl Config {
@@ -82,6 +83,7 @@ impl Config {
min: n, max: n, exact: Some(n), min: n, max: n, exact: Some(n),
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL, alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES, timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
stack_pool_cap: n * 4,
} }
} }
@@ -96,6 +98,7 @@ impl Config {
min, max, exact, min, max, exact,
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL, alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES, timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
stack_pool_cap: max * 4,
} }
} }
@@ -116,6 +119,14 @@ impl Config {
self 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. /// The number of scheduler threads this config resolves to.
pub fn resolved_thread_count(&self) -> usize { pub fn resolved_thread_count(&self) -> usize {
if let Some(e) = self.exact { if let Some(e) = self.exact {
@@ -137,6 +148,7 @@ impl Default for Config {
min: 1, max: avail, exact: None, min: 1, max: avail, exact: None,
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL, alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES, 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) root_pid: Option<Pid>,
pub(crate) timers: Timers, pub(crate) timers: Timers,
pub(crate) io: Option<IoThread>, 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 { impl SharedState {
@@ -257,7 +271,7 @@ impl SharedState {
root_pid: None, root_pid: None,
timers: Timers::new(), timers: Timers::new(),
io: None, 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> { 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. /// Preemption knobs, written into each scheduler thread's locals on startup.
pub(crate) alloc_interval: u32, pub(crate) alloc_interval: u32,
pub(crate) timeslice_cycles: u64, 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 { 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(); let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
Arc::new(Self { Arc::new(Self {
shared: Mutex::new(SharedState::new()), shared: Mutex::new(SharedState::new()),
@@ -317,6 +337,8 @@ impl RuntimeInner {
sleeping: AtomicU32::new(0), sleeping: AtomicU32::new(0),
alloc_interval, alloc_interval,
timeslice_cycles, timeslice_cycles,
stack_pool: Mutex::new(Vec::new()),
stack_pool_cap,
}) })
} }
@@ -346,7 +368,7 @@ pub struct Runtime {
pub fn init(config: Config) -> Runtime { pub fn init(config: Config) -> Runtime {
let n = config.resolved_thread_count(); let n = config.resolved_thread_count();
Runtime { 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, 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 slot = s.slot_mut(pid).expect("finalize_actor: slot vanished");
let sup = slot.actor.as_ref().map(|a| a.supervisor); 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.outcome = Some(joiner_outcome);
slot.state = State::Done; 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. // Deliver to supervisor.
if let Some(sup) = supervisor_pid { if let Some(sup) = supervisor_pid {
let sender = inner.with_shared(|s| { let sender = inner.with_shared(|s| {
@@ -568,11 +603,24 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
crate::timer::Reason::Sleep => { crate::timer::Reason::Sleep => {
inner.with_shared(|s| { inner.with_shared(|s| {
if let Some(slot) = s.slot_mut(entry.pid) { if let Some(slot) = s.slot_mut(entry.pid) {
if matches!(slot.state, State::Parked) { match slot.state {
State::Parked => {
slot.state = State::Runnable; slot.state = State::Runnable;
s.run_queue.push_back(entry.pid); s.run_queue.push_back(entry.pid);
crate::te!(crate::trace::Event::Enqueue(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 } // 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; let len = s.run_queue.len() as u64;
stats.run_queue_len.store(len, Ordering::Relaxed); 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 => { 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 // 1. run queue is still empty
// 2. no live actors (nothing parked, nothing mid-finalize) // 2. no live actors (nothing parked, nothing mid-finalize)
// 3. no pending timers // 3. no pending timers
// 4. no outstanding IO // 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 next = s.timers.peek_deadline();
let (out, fd) = match s.io.as_ref() { let (out, fd) = match s.io.as_ref() {
Some(io) => ( Some(io) => (
@@ -672,15 +749,23 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
None => (0, None), None => (0, None),
}; };
let live = s.slots.iter().filter(|slot| slot.actor.is_some()).count(); 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 // Something is still in flight. Sleep on the appropriate source
// to avoid hammering the mutex; the loop will retry on wake. // to avoid hammering the mutex; the loop will retry on wake.
match (next_deadline, wake_fd) { match (next_deadline, wake_fd) {
@@ -712,17 +797,9 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
// ---------------------------------------------------------------- // ----------------------------------------------------------------
// 3. Resume the actor. // 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. // 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); set_current_actor_box(b);
} }
+16 -7
View File
@@ -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 { 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| { let pid = with_runtime(|inner| {
inner.with_shared(|s| { inner.with_shared(|s| {
let (idx, gen) = s.allocate_slot(); let (idx, gen) = s.allocate_slot();
let pid = Pid::new(idx, gen); 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]; 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 });
slot.state = crate::runtime::State::Runnable; 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_unpark = false;
slot.pending_io_result = None; slot.pending_io_result = None;
s.run_queue.push_back(pid); 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::Spawn { parent: supervisor, child: pid });
crate::te!(crate::trace::Event::Enqueue(pid)); crate::te!(crate::trace::Event::Enqueue(pid));
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) { pub fn run<F: FnOnce() + Send + 'static>(f: F) {
crate::runtime::init(crate::runtime::Config::exact(1)).run(f); crate::runtime::init(crate::runtime::Config::exact(1)).run(f);
} }
+98
View File
@@ -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);
}
+64
View File
@@ -421,3 +421,67 @@ fn ping_pong_completes() {
}); });
assert_eq!(final_val.load(Ordering::SeqCst), ROUNDS); 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,
);
}