smarm
7746dca69b
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.
488 lines
17 KiB
Rust
488 lines
17 KiB
Rust
//! Tests for the multi-scheduler runtime: Config, Runtime::run, and
|
||
//! correctness under genuine parallelism.
|
||
//!
|
||
//! The single-threaded correctness properties (channel ordering, mutex
|
||
//! fairness, timer accuracy, etc.) are already covered by the per-module
|
||
//! tests. This file focuses on what changes when N > 1 scheduler threads
|
||
//! are involved:
|
||
//!
|
||
//! - Config construction and validation
|
||
//! - Runtime::run blocks until all actors finish
|
||
//! - All existing cooperative behaviours hold under multi-threading
|
||
//! - Actors genuinely run on different OS threads
|
||
//! - No lost wakeups under concurrent park/unpark
|
||
//! - No slot leaks under high spawn/join churn
|
||
//! - Panic on one scheduler thread doesn't kill others
|
||
|
||
use smarm::{channel, runtime::{Config, Runtime}, spawn, yield_now, JoinHandle};
|
||
use std::sync::{atomic::{AtomicBool, AtomicU64, Ordering}, Arc};
|
||
use std::time::Duration;
|
||
use std::collections::HashSet;
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Helpers
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Build a runtime with exactly `n` scheduler threads.
|
||
fn rt(n: usize) -> Runtime {
|
||
smarm::runtime::init(Config::exact(n))
|
||
}
|
||
|
||
/// Convenient single-threaded runtime (regression guard).
|
||
fn rt1() -> Runtime { rt(1) }
|
||
|
||
/// Multi-threaded runtime using all available parallelism.
|
||
fn rt_par() -> Runtime {
|
||
smarm::runtime::init(Config::default())
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Config
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn config_exact_overrides_bounds() {
|
||
let c = Config::exact(3);
|
||
assert_eq!(c.resolved_thread_count(), 3);
|
||
}
|
||
|
||
#[test]
|
||
fn config_default_clamps_to_available_parallelism() {
|
||
let c = Config::default();
|
||
let n = c.resolved_thread_count();
|
||
let avail = std::thread::available_parallelism()
|
||
.map(|n| n.get())
|
||
.unwrap_or(1);
|
||
// Default min is 1, default max is available_parallelism.
|
||
assert!(n >= 1 && n <= avail);
|
||
}
|
||
|
||
#[test]
|
||
fn config_min_max_clamps() {
|
||
// Force a range that excludes exact: min=2, max=4, available might be >4.
|
||
let c = Config::new(2, 4, None);
|
||
let n = c.resolved_thread_count();
|
||
assert!(n >= 2 && n <= 4, "expected 2..=4, got {n}");
|
||
}
|
||
|
||
#[test]
|
||
fn config_min_1_max_1_is_single_threaded() {
|
||
let c = Config::new(1, 1, None);
|
||
assert_eq!(c.resolved_thread_count(), 1);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Runtime::run — basic lifecycle
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn runtime_run_executes_closure() {
|
||
let flag = Arc::new(AtomicBool::new(false));
|
||
let f = flag.clone();
|
||
rt(1).run(move || { f.store(true, Ordering::SeqCst); });
|
||
assert!(flag.load(Ordering::SeqCst));
|
||
}
|
||
|
||
#[test]
|
||
fn runtime_run_blocks_until_all_actors_done() {
|
||
// Spawn a chain of actors; the counter should be exactly N when run returns.
|
||
let counter = Arc::new(AtomicU64::new(0));
|
||
let c = counter.clone();
|
||
rt(2).run(move || {
|
||
let mut handles = Vec::new();
|
||
for _ in 0..20 {
|
||
let cc = c.clone();
|
||
handles.push(spawn(move || {
|
||
cc.fetch_add(1, Ordering::SeqCst);
|
||
}));
|
||
}
|
||
for h in handles {
|
||
h.join().unwrap();
|
||
}
|
||
});
|
||
assert_eq!(counter.load(Ordering::SeqCst), 20);
|
||
}
|
||
|
||
#[test]
|
||
fn runtime_can_be_used_multiple_times_sequentially() {
|
||
// Each call to run() is independent.
|
||
let r = rt(2);
|
||
let a = Arc::new(AtomicU64::new(0));
|
||
let b = Arc::new(AtomicU64::new(0));
|
||
let ac = a.clone();
|
||
let bc = b.clone();
|
||
r.run(move || { ac.fetch_add(1, Ordering::SeqCst); });
|
||
r.run(move || { bc.fetch_add(1, Ordering::SeqCst); });
|
||
assert_eq!(a.load(Ordering::SeqCst), 1);
|
||
assert_eq!(b.load(Ordering::SeqCst), 1);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Single-threaded regression: exact(1) must behave identically to old run()
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn exact_1_spawn_join_works() {
|
||
let v = Arc::new(AtomicU64::new(0));
|
||
let vc = v.clone();
|
||
rt1().run(move || {
|
||
let h = spawn(move || { vc.store(42, Ordering::SeqCst); });
|
||
h.join().unwrap();
|
||
});
|
||
assert_eq!(v.load(Ordering::SeqCst), 42);
|
||
}
|
||
|
||
#[test]
|
||
fn exact_1_channel_recv_parks_and_wakes() {
|
||
let v = Arc::new(AtomicU64::new(0));
|
||
let vc = v.clone();
|
||
rt1().run(move || {
|
||
let (tx, rx) = channel::<u64>();
|
||
let h = spawn(move || {
|
||
let val = rx.recv().unwrap();
|
||
vc.store(val, Ordering::SeqCst);
|
||
});
|
||
yield_now();
|
||
tx.send(99).unwrap();
|
||
h.join().unwrap();
|
||
});
|
||
assert_eq!(v.load(Ordering::SeqCst), 99);
|
||
}
|
||
|
||
#[test]
|
||
fn exact_1_panic_captured() {
|
||
let saw_err = Arc::new(AtomicBool::new(false));
|
||
let s = saw_err.clone();
|
||
rt1().run(move || {
|
||
let h = spawn(|| panic!("oops"));
|
||
if h.join().is_err() { s.store(true, Ordering::SeqCst); }
|
||
});
|
||
assert!(saw_err.load(Ordering::SeqCst));
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Multi-threaded correctness
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn multi_thread_all_actors_complete() {
|
||
let counter = Arc::new(AtomicU64::new(0));
|
||
let c = counter.clone();
|
||
rt_par().run(move || {
|
||
let mut handles = Vec::new();
|
||
for _ in 0..100 {
|
||
let cc = c.clone();
|
||
handles.push(spawn(move || {
|
||
cc.fetch_add(1, Ordering::SeqCst);
|
||
}));
|
||
}
|
||
for h in handles { h.join().unwrap(); }
|
||
});
|
||
assert_eq!(counter.load(Ordering::SeqCst), 100);
|
||
}
|
||
|
||
#[test]
|
||
fn multi_thread_channel_wakeup_across_threads() {
|
||
// Receiver parks; sender runs (potentially on a different OS thread).
|
||
// Verifies no lost wakeup.
|
||
let received = Arc::new(AtomicU64::new(0));
|
||
let rc = received.clone();
|
||
rt_par().run(move || {
|
||
let (tx, rx) = channel::<u64>();
|
||
let h = spawn(move || {
|
||
let v = rx.recv().unwrap();
|
||
rc.store(v, Ordering::SeqCst);
|
||
});
|
||
// Let receiver park.
|
||
yield_now();
|
||
tx.send(7).unwrap();
|
||
h.join().unwrap();
|
||
});
|
||
assert_eq!(received.load(Ordering::SeqCst), 7);
|
||
}
|
||
|
||
#[test]
|
||
fn multi_thread_many_channels_no_lost_wakeups() {
|
||
// N pairs of (sender actor, receiver actor). Each pair exchanges one
|
||
// message. All must complete — any lost wakeup causes a deadlock/timeout.
|
||
const PAIRS: usize = 50;
|
||
let count = Arc::new(AtomicU64::new(0));
|
||
let c = count.clone();
|
||
rt_par().run(move || {
|
||
let mut handles: Vec<JoinHandle> = Vec::new();
|
||
for _ in 0..PAIRS {
|
||
let (tx, rx) = channel::<u64>();
|
||
let cc = c.clone();
|
||
handles.push(spawn(move || {
|
||
let v = rx.recv().unwrap();
|
||
cc.fetch_add(v, Ordering::SeqCst);
|
||
}));
|
||
handles.push(spawn(move || {
|
||
tx.send(1).unwrap();
|
||
}));
|
||
}
|
||
for h in handles { h.join().unwrap(); }
|
||
});
|
||
assert_eq!(count.load(Ordering::SeqCst), PAIRS as u64);
|
||
}
|
||
|
||
#[test]
|
||
fn multi_thread_mutex_contention_no_deadlock() {
|
||
use smarm::Mutex;
|
||
const ACTORS: usize = 20;
|
||
const PER: u64 = 100;
|
||
let total = Arc::new(AtomicU64::new(0));
|
||
let t = total.clone();
|
||
rt_par().run(move || {
|
||
let m: Mutex<u64> = Mutex::new(0);
|
||
let mut handles = Vec::new();
|
||
for _ in 0..ACTORS {
|
||
let mc = m.clone();
|
||
let tc = t.clone();
|
||
handles.push(spawn(move || {
|
||
for _ in 0..PER {
|
||
let mut g = mc.lock_timeout(Duration::from_secs(5)).unwrap();
|
||
*g += 1;
|
||
tc.fetch_add(0, Ordering::SeqCst); // just a memory barrier
|
||
}
|
||
}));
|
||
}
|
||
for h in handles { h.join().unwrap(); }
|
||
let g = m.lock_timeout(Duration::from_secs(1)).unwrap();
|
||
t.store(*g, Ordering::SeqCst);
|
||
});
|
||
assert_eq!(total.load(Ordering::SeqCst), ACTORS as u64 * PER);
|
||
}
|
||
|
||
#[test]
|
||
fn multi_thread_join_across_threads() {
|
||
// Parent joins a child that may run on a different scheduler thread.
|
||
let v = Arc::new(AtomicU64::new(0));
|
||
let vc = v.clone();
|
||
rt_par().run(move || {
|
||
let h = spawn(move || {
|
||
// Do some work to make scheduling interesting.
|
||
for _ in 0..10 { yield_now(); }
|
||
vc.store(1, Ordering::SeqCst);
|
||
});
|
||
h.join().unwrap();
|
||
});
|
||
assert_eq!(v.load(Ordering::SeqCst), 1);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Actors run on distinct OS threads
|
||
//
|
||
// We collect the OS thread IDs that actors execute on. With N schedulers
|
||
// and enough actors, we expect to see more than one thread ID.
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn actors_run_on_multiple_os_threads() {
|
||
let thread_ids: Arc<smarm::Mutex<HashSet<u64>>> =
|
||
Arc::new(smarm::Mutex::new(HashSet::new()));
|
||
|
||
rt_par().run({
|
||
let ids = thread_ids.clone();
|
||
move || {
|
||
let mut handles = Vec::new();
|
||
for _ in 0..64 {
|
||
let idc = ids.clone();
|
||
handles.push(spawn(move || {
|
||
let tid = unsafe { libc::syscall(libc::SYS_gettid) as u64 };
|
||
let mut g = idc.lock_timeout(Duration::from_secs(1)).unwrap();
|
||
g.insert(tid);
|
||
}));
|
||
}
|
||
for h in handles { h.join().unwrap(); }
|
||
}
|
||
});
|
||
|
||
let n = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
|
||
|
||
let ids = thread_ids.lock_timeout(Duration::from_secs(1)).unwrap();
|
||
// If we have >1 scheduler threads, we expect >1 OS thread IDs.
|
||
// On a single-CPU machine this may be 1; we just assert ≥ 1.
|
||
assert!(!ids.is_empty());
|
||
if n > 1 {
|
||
// Strongly expect parallelism — not a hard assert since scheduling
|
||
// is non-deterministic, but 64 actors should spread.
|
||
// We log rather than assert to avoid flakiness on loaded CI.
|
||
if ids.len() == 1 {
|
||
eprintln!("WARNING: 64 actors all ran on the same OS thread (flaky on loaded system)");
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Scheduler stats (RFC 000 Layer 1 primitives)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn scheduler_stats_run_queue_len_is_observable() {
|
||
// After spawning actors but before they run, the queue should be non-empty.
|
||
// We can't observe this from inside run() without a snapshot API, but we
|
||
// can verify the stats struct is accessible and returns sane values after
|
||
// run() completes (queue len == 0 at quiescence).
|
||
let r = rt_par();
|
||
r.run(|| {
|
||
for _ in 0..10 { spawn(|| {}); }
|
||
// Don't join — let them drain naturally.
|
||
});
|
||
let stats = r.stats();
|
||
assert_eq!(stats.total_run_queue_len(), 0, "queue should be empty after run()");
|
||
}
|
||
|
||
#[test]
|
||
fn scheduler_stats_thread_count_matches_config() {
|
||
let r = rt(3);
|
||
r.run(|| {});
|
||
assert_eq!(r.stats().scheduler_count(), 3);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Panic isolation: a panicking actor doesn't kill the scheduler thread
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn panic_in_actor_does_not_kill_runtime() {
|
||
let completed = Arc::new(AtomicU64::new(0));
|
||
let c = completed.clone();
|
||
rt_par().run(move || {
|
||
// Spawn a panicker alongside well-behaved actors.
|
||
let bad = spawn(|| panic!("deliberate"));
|
||
let mut good_handles = Vec::new();
|
||
for _ in 0..10 {
|
||
let cc = c.clone();
|
||
good_handles.push(spawn(move || {
|
||
cc.fetch_add(1, Ordering::SeqCst);
|
||
}));
|
||
}
|
||
let _ = bad.join(); // expect Err
|
||
for h in good_handles { h.join().unwrap(); }
|
||
});
|
||
assert_eq!(completed.load(Ordering::SeqCst), 10);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// No slot leaks: rapid spawn/join churn
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn no_slot_leak_under_churn() {
|
||
// Spawn and join many short actors in a loop. If slots leak, the slot
|
||
// table grows unboundedly. We can't directly measure it without an
|
||
// introspection API, but the test at least checks correctness under
|
||
// churn and will OOM if there's a severe leak.
|
||
let counter = Arc::new(AtomicU64::new(0));
|
||
let c = counter.clone();
|
||
rt_par().run(move || {
|
||
for _ in 0..500 {
|
||
let cc = c.clone();
|
||
spawn(move || { cc.fetch_add(1, Ordering::SeqCst); })
|
||
.join()
|
||
.unwrap();
|
||
}
|
||
});
|
||
assert_eq!(counter.load(Ordering::SeqCst), 500);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Ping-pong: channel round-trips between two actors
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[test]
|
||
fn ping_pong_completes() {
|
||
const ROUNDS: u64 = 1_000;
|
||
let final_val = Arc::new(AtomicU64::new(0));
|
||
let fv = final_val.clone();
|
||
rt_par().run(move || {
|
||
let (tx_a, rx_a) = channel::<u64>();
|
||
let (tx_b, rx_b) = channel::<u64>();
|
||
let h_a = spawn(move || {
|
||
tx_a.send(0).unwrap();
|
||
for _ in 0..ROUNDS {
|
||
let v = rx_b.recv().unwrap();
|
||
tx_a.send(v + 1).unwrap();
|
||
}
|
||
});
|
||
let h_b = spawn(move || {
|
||
for _ in 0..=ROUNDS {
|
||
let v = rx_a.recv().unwrap();
|
||
if v < ROUNDS {
|
||
tx_b.send(v).unwrap();
|
||
} else {
|
||
fv.store(v, Ordering::SeqCst);
|
||
}
|
||
}
|
||
});
|
||
h_a.join().unwrap();
|
||
h_b.join().unwrap();
|
||
});
|
||
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,
|
||
);
|
||
}
|