From 87416f903a818f8ce91095b344fa88ea73674c1e Mon Sep 17 00:00:00 2001 From: smarm-agent Date: Sun, 14 Jun 2026 07:05:29 +0000 Subject: [PATCH] =?UTF-8?q?RFC=20004=20chunk=203:=20spinning-worker=20regr?= =?UTF-8?q?ession=20tests.=20Watchdog-timeout=20harness=20turns=20a=20lost?= =?UTF-8?q?=20wakeup=20into=20a=20failed=20assertion.=20Key=20test=20shutd?= =?UTF-8?q?own=5Freleases=5Fpinned=5Fparked=5Fsiblings=20pins=20siblings?= =?UTF-8?q?=20in=20futex=5Fwait=20while=20live>0=20=E2=80=94=20verified=20?= =?UTF-8?q?to=20FAIL=20(15s=20timeout)=20when=20the=20AllDone=20broadcast?= =?UTF-8?q?=20is=20gutted,=20pass=20when=20restored.=20Also=20covers=20bou?= =?UTF-8?q?nce/fanout=20(park+hot-spinner),=20N=3D1=20cap=3D0=20inertness,?= =?UTF-8?q?=20budget-0=20opt-out,=20repeated=20lifecycles.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/spinning_workers.rs | 260 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 260 insertions(+) create mode 100644 tests/spinning_workers.rs diff --git a/tests/spinning_workers.rs b/tests/spinning_workers.rs new file mode 100644 index 0000000..b58bf2a --- /dev/null +++ b/tests/spinning_workers.rs @@ -0,0 +1,260 @@ +//! RFC 004 spinning-worker tests. +//! +//! The futex scheduler-park path is NOT loom-modelled (sync_shim.rs: the full +//! runtime — context switches, futexes, TLS — is never run under cfg(loom)), +//! so the Dekker submit/park ordering is guarded here instead, by driving the +//! one true deadlock state and catching it as a timeout rather than a hung +//! binary: every scheduler parked on the futex while runnable work exists. +//! +//! All tests use a NO-IO runtime (pure message passing) so the idle path is +//! the bare `_` arm RFC 004 replaces — the futex park — not the timer/io arms, +//! which keep their own wake sources. A lost wakeup therefore manifests as a +//! permanent stall: a `recv` that never returns, an `rt.run` that never +//! reaches AllDone. The watchdog turns that into a failed assertion. + +use smarm::channel::channel; +use smarm::runtime::{init, Config}; +use smarm::spawn; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +/// Run `body` on a side thread; fail (rather than hang the test binary) if it +/// does not finish within `secs`. A hang here means a scheduler parked on the +/// futex with work outstanding and nothing woke it — exactly the RFC 004 +/// failure mode. The leaked thread is acceptable: the process exits non-zero. +fn run_with_timeout(secs: u64, body: impl FnOnce() + Send + 'static) { + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + body(); + let _ = tx.send(()); + }); + if rx.recv_timeout(Duration::from_secs(secs)).is_err() { + panic!( + "RFC 004 deadlock: workload did not terminate within {secs}s \ + — a scheduler wakeup was lost (work stranded with all schedulers \ + parked on the futex)" + ); + } +} + +/// A single token ping-ponging between two actors for `rounds` rounds, on a +/// `threads`-wide no-io runtime with the given spin budget. With more threads +/// than active actors, the surplus schedulers go idle and park; every send to +/// the (parked) peer is a cross-thread wakeup. A lost wake stalls the bounce. +/// The post-run count confirms no message was silently dropped. +fn bounce(threads: usize, rounds: u64, budget: u64) { + run_with_timeout(30, move || { + let rt = init(Config::exact(threads).spin_budget_cycles(budget)); + let echoed = Arc::new(AtomicU64::new(0)); + let echoed_actor = echoed.clone(); + rt.run(move || { + let (tx_ab, rx_ab) = channel::(); + let (tx_ba, rx_ba) = channel::(); + let echo = spawn(move || { + for _ in 0..rounds { + let v = rx_ab.recv().expect("echo recv"); + echoed_actor.fetch_add(1, Ordering::Relaxed); + tx_ba.send(v + 1).expect("echo send"); + } + }); + for i in 0..rounds { + tx_ab.send(i).expect("ping send"); + let r = rx_ba.recv().expect("ping recv"); + assert_eq!(r, i + 1, "token corrupted in flight"); + } + echo.join().expect("echo join"); + }); + assert_eq!( + echoed.load(Ordering::Relaxed), + rounds, + "messages were lost, not just delayed" + ); + }); +} + +/// Tiny budget: schedulers park almost immediately, so essentially every +/// cross-thread handoff exercises the futex wake. Sharpest stress on the +/// submit-rule / lost-wakeup guards. +#[test] +fn bounce_forces_park_and_wake() { + bounce(4, 20_000, 2_000); +} + +/// Large budget: idle schedulers stay hot spinners, so the submit rule mostly +/// takes the `n_spinning > 0` skip and the last-spinner handoff (rule 4) does +/// the waking. Confirms that path also makes progress and stays correct. +#[test] +fn bounce_with_hot_spinners() { + bounce(4, 20_000, 2_000_000); +} + +/// Oversubscribed: more active actors than schedulers, so the run queue is +/// rarely empty and the park/spin/submit machinery is hammered from every +/// thread at once. +#[test] +fn bounce_oversubscribed() { + bounce(2, 30_000, 5_000); +} + +/// Bursty fork: a parent makes many children runnable at once (the motivating +/// fan-out), each sends one reply, the parent collects all of them. Exercises +/// the submit rule under a burst — the first job eats one wake, the woken +/// worker becomes a spinner and absorbs the rest — and confirms none are lost. +fn fanout(threads: usize, children: u64, budget: u64) { + run_with_timeout(30, move || { + let rt = init(Config::exact(threads).spin_budget_cycles(budget)); + rt.run(move || { + let (tx, rx) = channel::(); + for c in 0..children { + let tx = tx.clone(); + spawn(move || { + tx.send(c).expect("child send"); + }); + } + drop(tx); + let mut sum = 0u64; + for _ in 0..children { + sum += rx.recv().expect("collector recv"); + } + assert_eq!(sum, (0..children).sum(), "a child reply was lost"); + }); + }); +} + +#[test] +fn fanout_burst_parks() { + fanout(4, 256, 2_000); +} + +#[test] +fn fanout_burst_hot() { + fanout(4, 256, 2_000_000); +} + +/// Shutdown guard: a mostly-idle spinning runtime must still terminate. Most +/// schedulers park on the futex with no work ever coming; only the AllDone +/// broadcast (`unpark_all_schedulers`) can release them. If that broadcast is +/// missing, `rt.run` never returns and the watchdog fires. +#[test] +fn idle_runtime_shuts_down() { + run_with_timeout(15, || { + let rt = init(Config::exact(8).spin_budget_cycles(2_000)); + rt.run(|| { + // One trivial cross-thread handoff, then everyone goes idle and the + // runtime must drain to AllDone through parked schedulers. + let (tx, rx) = channel::(); + let h = spawn(move || { + tx.send(42).expect("send"); + }); + assert_eq!(rx.recv().expect("recv"), 42); + h.join().expect("join"); + }); + }); +} + +/// Repeated lifecycles: init→run→AllDone several times in one process, to +/// catch any park/broadcast state that fails to reset between runtimes. +#[test] +fn repeated_spinning_lifecycles() { + run_with_timeout(30, || { + for _ in 0..20 { + let rt = init(Config::exact(4).spin_budget_cycles(2_000)); + let echoed = Arc::new(AtomicU64::new(0)); + let e = echoed.clone(); + rt.run(move || { + let (tx, rx) = channel::(); + let worker = spawn(move || { + for _ in 0..500 { + let v = rx.recv().expect("recv"); + e.fetch_add(v, Ordering::Relaxed); + } + }); + for i in 0..500u64 { + tx.send(i).expect("send"); + } + worker.join().expect("worker join"); + }); + assert_eq!(echoed.load(Ordering::Relaxed), (0..500u64).sum()); + } + }); +} + +/// Shutdown with siblings *pinned* in the futex park — the test that actually +/// exercises the AllDone broadcast. A single long-running actor holds `live` +/// at 1 (CPU-bound, never sending or spawning) long enough for every other +/// scheduler to idle, spin out its budget, and commit to `futex_wait`. Because +/// nothing is ever enqueued during that window, the submit-rule wakes never +/// fire and those siblings stay parked; the final finalize enqueues no one, so +/// the live-check self-termination cannot save them either. Only +/// `unpark_all_schedulers` at AllDone can. Gut that broadcast and this hangs. +#[test] +fn shutdown_releases_pinned_parked_siblings() { + run_with_timeout(15, || { + let rt = init(Config::exact(8).spin_budget_cycles(2_000)); + rt.run(|| { + // ~50ms of pure CPU: far longer than the µs it takes the other + // seven schedulers to park, so they are provably in futex_wait + // before this returns and live drops to 0. + let start = std::time::Instant::now(); + let mut x = 0u64; + while start.elapsed() < Duration::from_millis(50) { + x = x.wrapping_mul(6364136223846793005).wrapping_add(1); + std::hint::black_box(x); + } + std::hint::black_box(x); + }); + }); +} + +/// cross-thread handoff — but the futex park path must still be inert and +/// correct (the lone thread is never parked while it has work). Pins the +/// degenerate cap=0 case. +#[test] +fn single_thread_spinning_is_inert() { + run_with_timeout(15, || { + let rt = init(Config::exact(1).spin_budget_cycles(2_000)); + let echoed = Arc::new(AtomicU64::new(0)); + let e = echoed.clone(); + rt.run(move || { + let (tx, rx) = channel::(); + let worker = spawn(move || { + for _ in 0..1_000 { + let v = rx.recv().expect("recv"); + e.fetch_add(v, Ordering::Relaxed); + } + }); + for i in 0..1_000u64 { + tx.send(i).expect("send"); + } + worker.join().expect("join"); + }); + assert_eq!(echoed.load(Ordering::Relaxed), (0..1_000u64).sum()); + }); +} + +/// `spin_budget_cycles == 0` is the opt-out: the feature is fully off and the +/// runtime uses the historical `thread::sleep` idle path. Must still run a +/// cross-thread workload to completion (it always did) — this just guards that +/// the gating left the off-path intact. +#[test] +fn budget_zero_keeps_old_behaviour() { + run_with_timeout(20, || { + let rt = init(Config::exact(4).spin_budget_cycles(0)); + let sum = Arc::new(AtomicU64::new(0)); + let s = sum.clone(); + rt.run(move || { + let (tx, rx) = channel::(); + let worker = spawn(move || { + for _ in 0..1_000 { + s.fetch_add(rx.recv().expect("recv"), Ordering::Relaxed); + } + }); + for i in 0..1_000u64 { + tx.send(i).expect("send"); + } + worker.join().expect("join"); + }); + assert_eq!(sum.load(Ordering::Relaxed), (0..1_000u64).sum()); + }); +}