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.
This commit is contained in:
smarm
2026-05-26 21:58:14 +02:00
parent 72f5d38e5d
commit 7746dca69b
6 changed files with 395 additions and 91 deletions
+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);
}