From 7746dca69b15438cebb93c589a32c35e8aab76c1 Mon Sep 17 00:00:00 2001 From: smarm Date: Tue, 26 May 2026 21:58:14 +0200 Subject: [PATCH] fix(runtime): handle timer firing while actor is still Runnable in sleep() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 3 + README.md | 6 +- benches/baseline.json | 294 +++++++++++++++++++++--------- src/runtime.rs | 21 ++- tests/many_timers_multi_thread.rs | 98 ++++++++++ tests/runtime.rs | 64 +++++++ 6 files changed, 395 insertions(+), 91 deletions(-) create mode 100644 tests/many_timers_multi_thread.rs diff --git a/Cargo.toml b/Cargo.toml index 7b458d8..34a0e1e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,3 +38,6 @@ harness = false name = "smarm_favored" harness = false +[[bench]] +name = "tokio_favored" +harness = false diff --git a/README.md b/README.md index 8d7dd45..92733dc 100644 --- a/README.md +++ b/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 -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. --- -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. + diff --git a/benches/baseline.json b/benches/baseline.json index 4a72d5b..ff05710 100644 --- a/benches/baseline.json +++ b/benches/baseline.json @@ -2,185 +2,313 @@ "chained_spawn": { "smarm 1-thread": { "result": 1000, - "median": 273, - "min": 261, - "max": 292 + "median": 266, + "min": 242, + "max": 351 }, "smarm 24-thread": { "result": 1000, - "median": 837, - "min": 788, - "max": 913 + "median": 742, + "min": 696, + "max": 860 }, "tokio current_thread": { "result": 1000, - "median": 68, + "median": 62, "min": 61, - "max": 113 + "max": 68 }, "tokio multi-thread": { "result": 1000, - "median": 200, - "min": 176, - "max": 230 + "median": 190, + "min": 169, + "max": 207 } }, "yield_many": { "smarm 1-thread": { "result": 200000, - "median": 19465, - "min": 19299, - "max": 20007 + "median": 19071, + "min": 18776, + "max": 19396 }, "smarm 24-thread": { "result": 200000, - "median": 163921, - "min": 155700, - "max": 167921 + "median": 172454, + "min": 166246, + "max": 174230 }, "tokio current_thread": { "result": 200000, - "median": 5022, - "min": 5001, - "max": 5530 + "median": 4737, + "min": 4644, + "max": 5065 }, "tokio multi-thread": { "result": 200000, - "median": 8341, - "min": 7412, - "max": 9199 + "median": 8738, + "min": 7852, + "max": 9770 } }, "fan_out_compute": { "smarm 1-thread": { "result": 33860, - "median": 14096, - "min": 13982, - "max": 17406 + "median": 13234, + "min": 13196, + "max": 13390 }, "smarm 24-thread": { "result": 33860, - "median": 2475, - "min": 2264, - "max": 2589 + "median": 2244, + "min": 2162, + "max": 2380 }, "tokio current_thread": { "result": 33860, - "median": 12640, - "min": 12548, - "max": 15881 + "median": 14049, + "min": 14035, + "max": 14300 }, "tokio multi-thread": { "result": 33860, - "median": 1483, - "min": 1395, - "max": 1710 + "median": 1474, + "min": 1285, + "max": 1823 } }, "ping_pong_oneshot": { "smarm 1-thread": { "result": 1000, - "median": 801, - "min": 720, - "max": 929 + "median": 751, + "min": 727, + "max": 913 }, "smarm 24-thread": { "result": 1000, - "median": 1449, - "min": 1292, - "max": 1530 + "median": 1308, + "min": 1227, + "max": 1396 }, "tokio current_thread": { "result": 1000, - "median": 398, - "min": 369, - "max": 407 + "median": 407, + "min": 400, + "max": 444 }, "tokio multi-thread": { "result": 1000, - "median": 10497, - "min": 9238, - "max": 11961 + "median": 10869, + "min": 8683, + "max": 11688 + } + }, + "spawn_storm_busy": { + "smarm 1-thread": { + "result": 10000, + "median": 112045, + "min": 99936, + "max": 117329 + }, + "smarm 24-thread": { + "result": 10000, + "median": 137105, + "min": 130852, + "max": 147707 + }, + "tokio current_thread": { + "result": 10000, + "median": 1128, + "min": 1123, + "max": 1435 + }, + "tokio multi-thread": { + "result": 10000, + "median": 19674, + "min": 16013, + "max": 27234 + } + }, + "mpsc_contention": { + "smarm 1-thread": { + "result": 320000, + "median": 3667, + "min": 3608, + "max": 4126 + }, + "smarm 24-thread": { + "result": 320000, + "median": 45681, + "min": 31908, + "max": 51287 + }, + "tokio current_thread": { + "result": 320000, + "median": 6228, + "min": 6210, + "max": 6514 + }, + "tokio multi-thread": { + "result": 320000, + "median": 66173, + "min": 42208, + "max": 83255 + } + }, + "many_timers": { + "smarm 1-thread": { + "result": 10000, + "median": 119988, + "min": 107308, + "max": 123557 + }, + "smarm 24-thread": { + "result": 10000, + "median": 218842, + "min": 182009, + "max": 256988 + }, + "tokio current_thread": { + "result": 10000, + "median": 12432, + "min": 12308, + "max": 13468 + }, + "tokio multi-thread": { + "result": 10000, + "median": 16311, + "min": 15026, + "max": 16897 + } + }, + "multi_thread_scaling": { + "smarm 1-thread": { + "result": 33860, + "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": 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": 136, - "min": 117, - "max": 163 + "median": 102, + "min": 96, + "max": 123 }, "smarm 24-thread": { "result": 1, - "median": 743, - "min": 654, - "max": 845 + "median": 597, + "min": 576, + "max": 682 }, "tokio current_thread": { "result": 1, - "median": 14, - "min": 14, - "max": 16 + "median": 13, + "min": 11, + "max": 35 }, "tokio multi-thread": { "result": 1, - "median": 61, - "min": 47, - "max": 94 + "median": 56, + "min": 46, + "max": 65 } }, "yield_in_hot_loop": { "smarm 1-thread": { "result": 1000000, - "median": 83281, - "min": 80264, - "max": 85697 + "median": 80680, + "min": 80308, + "max": 81845 }, "tokio current_thread": { "result": 1000000, - "median": 75538, - "min": 69842, - "max": 78823 + "median": 72606, + "min": 72154, + "max": 77206 } }, "uncontended_channel": { "smarm 1-thread": { "result": 1000000, - "median": 9440, - "min": 9393, - "max": 13168 + "median": 9257, + "min": 9223, + "max": 12049 }, "tokio current_thread": { "result": 1000000, - "median": 17817, - "min": 16926, - "max": 18572 + "median": 16925, + "min": 16848, + "max": 17019 } }, "catch_unwind_panics": { "smarm 1-thread": { "result": 10000, - "median": 133840, - "min": 130425, - "max": 139709 + "median": 116821, + "min": 111345, + "max": 128261 }, "smarm 24-thread": { "result": 10000, - "median": 128729, - "min": 108091, - "max": 132700 + "median": 117487, + "min": 107011, + "max": 129307 }, "tokio current_thread": { "result": 10000, - "median": 10681, - "min": 10158, - "max": 11644 + "median": 10425, + "min": 10141, + "max": 10604 }, "tokio multi-thread": { "result": 10000, - "median": 6700, - "min": 2778, - "max": 7453 + "median": 6418, + "min": 3715, + "max": 7144 } } } \ No newline at end of file diff --git a/src/runtime.rs b/src/runtime.rs index 68259aa..e1a2541 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -603,10 +603,23 @@ fn schedule_loop(inner: &Arc, 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 => {} } } }); diff --git a/tests/many_timers_multi_thread.rs b/tests/many_timers_multi_thread.rs new file mode 100644 index 0000000..10ff947 --- /dev/null +++ b/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); +} diff --git a/tests/runtime.rs b/tests/runtime.rs index 86ccb47..b41445e 100644 --- a/tests/runtime.rs +++ b/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, + ); +}