diff --git a/context/code/improvise-incremental.md b/context/code/improvise-incremental.md index 81dce313..01e9baef 100644 --- a/context/code/improvise-incremental.md +++ b/context/code/improvise-incremental.md @@ -511,7 +511,9 @@ Multi-threading here is **scheduling**. Each sub-circuit stays synchronous and s (`zset.md`, "Threads"), the only thing that crosses threads is an edge (`zset.md`, "Edges"), and a small executor decides which thread runs which sub-circuit next. This is a ruling recorded here, not an ADR amendment. Native threads are assumed. `sched.rs` and `staged.rs` compile for wasm32 -(a `Mutex` and a `Condvar`), and `Pool`, the only thing that spawns a thread, is +(a `Mutex` and a `Condvar`), but `improvise-command` depends on this crate, so the module is in +the browser bundle: `Pool`, the only thing that spawns a thread, and the blocking calls +(`claim_blocking`, `wait_idle`, which `Condvar::wait` would make panic there) are `cfg(not(target_arch = "wasm32"))`. Nothing here runs on wasm. 1. **A unit is a sub-circuit with the edge ends around it** (`Unit`: `turn(budget)` and @@ -530,8 +532,15 @@ not an ADR amendment. Native threads are assumed. `sched.rs` and `staged.rs` com queue and the slots, and a turn takes microseconds beside it. 4. **Fairness is the budget and the back of the queue.** A turn handles at most `budget` frames (`EdgeSource::poll(max)`), then says `more`, and `finish` queues it behind the units already - waiting. A full edge (`Flow::Backlogged`) is also `more`: the unit retries, and coalesces - meanwhile. The resumable `step(deadline)` predicate of `Rebuild` is for a sliced snapshot load, + waiting. A full edge is **not** `more`: the sender would spin through the queue while its + consumer is not running (`a_full_edge_does_not_make_its_sender_spin`). The consumer that frees + room reports progress and kicks its producer back, so a backlogged sender runs when it can + send. For the same reason a middle unit (the aggregate) needs no `more` at its budget: its + consumer's progress kicks it. The tail (the projection) has no consumer to kick it back, so it + says `more` when it read a full budget (`a_unit_whose_kicks_coalesced_still_drains_its_edge`, + because kicks coalesce into one queue entry). Both clauses were first written untested; a + review injected their removal and everything passed, which is how the first was found to be a + spin and the second redundant. The resumable `step(deadline)` predicate of `Rebuild` is for a sliced snapshot load, and this example has none. 5. **The executor spawns nothing.** `claim` and `finish` are what a worker calls. `Pool` is N threads looping on them; the tests' scheduler calls them in an order a seed chooses, with up to N @@ -548,7 +557,21 @@ not an ADR amendment. Native threads are assumed. `sched.rs` and `staged.rs` com a total per row). The incremental backend (`Incremental`) keeps its groups in its own shapes and does not run on a `Circuit`, so the stage is the same computation built as zset nodes rather than a split of `Incremental` itself. Running `Incremental` on circuits is a later step. -8. **Not done.** A work-stealing or priority scheme (one FIFO queue); a unit with a `Rebuild` job - sliced by the deadline predicate; a timeout that calls `connection_lost` on the edges (the - edges here never lose a frame); a panic in a turn (the executor's mutex poisons and the pool - reports it). +8. **A panic in a turn kills that unit and nothing else.** A turn runs outside the lock, so + nothing is poisoned and, in the first version, the pool waited for ever on a unit that never + came back. Now the worker catches the unwind and calls `Executor::abandon`: ownership is + released, the unit is dead (never run or kicked again), the panic is recorded + (`Executor::panics`, `UnitPanic { unit, message }`), and neighbours go on. `wait_idle` and + `Pool::finish` return `Err(UnitPanic)` for the first one once the rest is quiet. A unit that + cannot name its neighbours is buried the same way. A kick of a unit that does not exist panics + before it takes the lock (`no such unit`), so a bad index cannot poison the executor; the + `lock()` message now names the only thing that can poison it, a panic in the executor itself. +9. **Workers are counted and a dropped `Pool` stops them.** `Executor::workers()` counts live + worker threads, so a worker that leaves on a spurious wake is visible + (`every_worker_stays_through_wakes_with_nothing_queued`); `Drop for Pool` shuts down and joins + (`a_pool_that_is_dropped_stops_its_threads`), discarding what is queued. `claim_blocking` + returns `None` as soon as `shutdown` is set. The real-thread tests run behind a watchdog + (`tests/support/watchdog.rs`), so a lost `notify_all` fails by name and not by hanging. +10. **Not done.** A work-stealing or priority scheme (one FIFO queue); a unit with a `Rebuild` + job sliced by the deadline predicate; a timeout that calls `connection_lost` on the edges (the + edges here never lose a frame); restarting a dead unit. diff --git a/context/test-evidence/improvise-incremental.md b/context/test-evidence/improvise-incremental.md index 33aa3e1b..ed05866e 100644 --- a/context/test-evidence/improvise-incremental.md +++ b/context/test-evidence/improvise-incremental.md @@ -985,3 +985,65 @@ handed back. Rivals S1 and S2 fail it. Four kicking threads and four workers over one unit that asserts no two turns overlap. Rival S1 fails it. + +### a_panicking_turn_does_not_hang_finish + +A unit whose turn panics, on a two-thread pool: `Pool::finish` hung for ever (a 3-second watchdog +reproduced it before the fix). The unwind is now caught and the unit buried. Rival P1 (the dead +unit still counted as running) fails it by the watchdog. + +### a_panicking_unit_is_reported_and_its_neighbours_still_run + +`finish` returns `UnitPanic { unit: 0, message }` containing the payload, and a live unit ran. +Rival P1 fails it. + +### a_dead_unit_is_never_run_again + +After `abandon`, a kick does not queue the unit. Rival P2 (the `dead` check removed) fails it. + +### a_kick_of_a_unit_that_does_not_exist_panics_before_the_lock + +`kick(5)` names the missing unit. Rival P3 (no index check) fails it with an index error. + +### an_executor_survives_a_bad_kick + +After the bad `kick`, `idle()` works: the lock was not poisoned. Rival P3 fails it. + +### every_worker_stays_through_wakes_with_nothing_queued + +Four workers, 200 kicks, the worker count stays four and is zero after `finish`. Rival W1 (a +worker returns on a wake with an empty queue) fails it. Rivals W3 and W4 (a lost `notify_all` in +`kick` or `finish`) fail it by the watchdog. + +### a_pool_that_is_dropped_stops_its_threads + +Dropping a pool leaves no worker. Rival W2 (an empty `Drop`) fails it. + +### a_full_edge_does_not_make_its_sender_spin + +The aggregate and projection are held, the head's edge (capacity 1) fills, and the head must not +be queued again. Rival M4 (`more` while backlogged) makes it ask for ever and fails it at 50 +turns. + +### a_unit_whose_kicks_coalesced_still_drains_its_edge + +Two commands cross before the projection's kicks are served, with a budget of 1. The projection +must ask to run again; the second frame is otherwise never read. Rival M5 fails it (and +`placed_circuits_equal_one_circuit_for_any_workers_and_order`). The same clause on the aggregate +is redundant, because the projection's progress kicks it back; removing it passes everything, so +it was deleted. + +### a_dag_on_real_threads_equals_one_circuit_and_idle_means_done + +The reviewer's seven-unit graph (fan-out, fan-in, fan-out) with random edge capacities and +budgets, 1 to 8 workers, `wait_idle` checkpoints mid-stream that must equal the reference for the +prefix submitted. Rivals W3 and W4 fail it by the watchdog. + +### submitter_threads_race_the_pool + +Four threads submit commands to the same pool; the result is the reference. Rivals W3 and W4 fail +it by the watchdog. + +### a_dag_on_virtual_workers_in_any_order_equals_one_circuit + +The same graph under a random claim and hand-back order of 1 to 8 virtual workers. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index 82bdfeda..59d4ceda 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -772,3 +772,9 @@ Rival: `unsafe impl Sync for Circuit {}` fails the build of `tests/threads.rs`. A sink over `Counted` keys clones each key once (the insert into its sum). Rival: a sink that clones its input piece first, as the reviewer injected, makes `N` more. + +### traffic_never_goes_down_across_a_restart_of_either_end + +`EdgeSink::restart` and `EdgeSource::restart` kept the frame counters at 0 (they rebuilt the +struct), which could make a unit read "no progress" and lose a kick. Rival: either restart +resetting its counter fails it. diff --git a/crates/improvise-incremental/src/sched.rs b/crates/improvise-incremental/src/sched.rs index 0c9ead12..bf12c27f 100644 --- a/crates/improvise-incremental/src/sched.rs +++ b/crates/improvise-incremental/src/sched.rs @@ -2,6 +2,7 @@ //! Reasoning: `context/code/improvise-incremental.md`, "Scheduling". use std::collections::VecDeque; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Condvar, Mutex, MutexGuard}; /// What one turn of a unit did. @@ -9,7 +10,8 @@ use std::sync::{Condvar, Mutex, MutexGuard}; pub struct Turn { /// A frame moved on one of its edges, so its neighbours may have work. pub progress: bool, - /// It stopped at its budget, or an edge was full: run it again. + /// It stopped at its budget with frames left: run it again. A full edge is not `more`; the + /// reader that frees room reports progress and kicks it. pub more: bool, } @@ -28,6 +30,8 @@ struct Slot { running: bool, /// Kicked while running: queue it again when the owner hands it back. again: bool, + /// Its turn panicked: it is never run or kicked again. + dead: bool, } struct State { @@ -35,6 +39,7 @@ struct State { queue: VecDeque, running: usize, shutdown: bool, + panics: Vec, } /// A unit a worker owns until it hands it back with `finish`. @@ -53,17 +58,37 @@ impl Claimed { } } +/// A unit whose turn panicked. It is dead: it is never run again, and its neighbours go on. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct UnitPanic { + pub unit: usize, + pub message: String, +} + +/// The text of a panic payload. +pub fn panic_text(payload: &(dyn std::any::Any + Send)) -> String { + payload + .downcast_ref::() + .cloned() + .or_else(|| payload.downcast_ref::<&str>().map(|s| s.to_string())) + .unwrap_or_else(|| "a panic with a non-string payload".to_string()) +} + /// The run queue and the ownership of every unit. It spawns nothing: workers (threads, or a test -/// scheduler choosing an order) call `claim` and `finish`. +/// scheduler choosing an order) call `claim` and `finish`, or `abandon` after a panic. pub struct Executor { state: Mutex, wake: Condvar, budget: usize, + units: usize, + /// Worker threads a `Pool` has started and not yet seen leave. + workers: AtomicUsize, } impl Executor { /// `budget` is the frames a unit may handle per turn; zero is read as one. pub fn new(units: Vec>, budget: usize) -> Self { + let count = units.len(); Executor { state: Mutex::new(State { slots: units @@ -73,14 +98,18 @@ impl Executor { queued: false, running: false, again: false, + dead: false, }) .collect(), queue: VecDeque::new(), running: 0, shutdown: false, + panics: Vec::new(), }), wake: Condvar::new(), budget: budget.max(1), + units: count, + workers: AtomicUsize::new(0), } } @@ -88,12 +117,29 @@ impl Executor { self.budget } + /// How many worker threads are alive. + pub fn workers(&self) -> usize { + self.workers.load(Ordering::SeqCst) + } + + /// The units that have panicked, in order. + pub fn panics(&self) -> Vec { + self.lock().panics.clone() + } + + /// Only a panic inside the executor's own code, under the lock, poisons it. A unit's turn + /// runs outside the lock, so a panic there is `abandon`ed instead. fn lock(&self) -> MutexGuard<'_, State> { - self.state.lock().expect("a unit panicked in a turn") + self.state + .lock() + .expect("the executor's lock was poisoned by a panic inside the executor") } fn kick_in(st: &mut State, id: usize) { let slot = &mut st.slots[id]; + if slot.dead { + return; + } if slot.running { slot.again = true; } else if !slot.queued { @@ -102,8 +148,14 @@ impl Executor { } } + #[track_caller] + fn exists(&self, id: usize) { + assert!(id < self.units, "zset sched: no such unit {id}"); + } + /// Marks `id` runnable: it has input, or the host wants it to look. pub fn kick(&self, id: usize) { + self.exists(id); Self::kick_in(&mut self.lock(), id); self.wake.notify_all(); } @@ -136,26 +188,41 @@ impl Executor { (n > 0).then(|| Self::take(&mut st, pick(n) % n)) } - /// Like `claim` for a worker thread: waits for work, and is `None` after `shutdown`. + /// Like `claim` for a worker thread: waits for work, and is `None` once `shutdown` is called. + #[cfg(not(target_arch = "wasm32"))] pub fn claim_blocking(&self) -> Option { let mut st = self.lock(); loop { - if !st.queue.is_empty() { - return Some(Self::take(&mut st, 0)); - } if st.shutdown { return None; } - st = self.wake.wait(st).expect("a unit panicked in a turn"); + if !st.queue.is_empty() { + return Some(Self::take(&mut st, 0)); + } + st = self + .wake + .wait(st) + .expect("the executor's lock was poisoned by a panic inside the executor"); } } - /// Hands a unit back; queues it again if it asked for more or was kicked meanwhile, and - /// kicks its neighbours if the turn moved a frame. + /// Hands a unit back, queued again if it wants more or was kicked; progress kicks its + /// neighbours. A unit that cannot name them is buried. pub fn finish(&self, claimed: Claimed, turn: Turn) { let Claimed { id, unit } = claimed; + let named = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let n = unit.neighbours().to_vec(); + assert!( + n.iter().all(|&u| u < self.units), + "zset sched: unit {id} names a neighbour that does not exist" + ); + n + })); + let neighbours = match named { + Ok(n) => n, + Err(p) => return self.bury(id, unit, panic_text(&*p)), + }; let mut st = self.lock(); - let neighbours = unit.neighbours().to_vec(); let slot = &mut st.slots[id]; slot.unit = Some(unit); slot.running = false; @@ -173,21 +240,46 @@ impl Executor { self.wake.notify_all(); } + /// A unit's turn panicked: it is dead, ownership is released and the panic is recorded. + /// `wait_idle` and `Pool::finish` report it, and nothing waits for the dead unit. + pub fn abandon(&self, claimed: Claimed, message: String) { + let Claimed { id, unit } = claimed; + self.bury(id, unit, message); + } + + fn bury(&self, id: usize, unit: Box, message: String) { + let mut st = self.lock(); + let slot = &mut st.slots[id]; + slot.running = false; + slot.dead = true; + slot.again = false; + st.running -= 1; + st.panics.push(UnitPanic { unit: id, message }); + drop(st); + drop(unit); + self.wake.notify_all(); + } + /// Whether nothing is queued and nothing is running. pub fn idle(&self) -> bool { let st = self.lock(); st.queue.is_empty() && st.running == 0 } - /// Blocks until `idle`. - pub fn wait_idle(&self) { + /// Blocks until `idle`, then reports the first unit that panicked, if any. + #[cfg(not(target_arch = "wasm32"))] + pub fn wait_idle(&self) -> Result<(), UnitPanic> { let mut st = self.lock(); while !(st.queue.is_empty() && st.running == 0) { - st = self.wake.wait(st).expect("a unit panicked in a turn"); + st = self + .wake + .wait(st) + .expect("the executor's lock was poisoned by a panic inside the executor"); } + st.panics.first().cloned().map_or(Ok(()), Err) } - /// Tells the workers to leave once the queue is empty. + /// Tells the workers to leave at once, whatever is queued. pub fn shutdown(&self) { self.lock().shutdown = true; self.wake.notify_all(); @@ -201,16 +293,34 @@ pub struct Pool { workers: Vec>, } +/// Counts a worker as gone however its thread ends. +#[cfg(not(target_arch = "wasm32"))] +struct Alive(std::sync::Arc); + +#[cfg(not(target_arch = "wasm32"))] +impl Drop for Alive { + fn drop(&mut self) { + self.0.workers.fetch_sub(1, Ordering::SeqCst); + } +} + #[cfg(not(target_arch = "wasm32"))] impl Pool { pub fn start(exec: std::sync::Arc, workers: usize) -> Self { let workers = (0..workers.max(1)) .map(|_| { - let exec = std::sync::Arc::clone(&exec); + exec.workers.fetch_add(1, Ordering::SeqCst); + let alive = Alive(std::sync::Arc::clone(&exec)); std::thread::spawn(move || { + let exec = std::sync::Arc::clone(&alive.0); while let Some(mut claimed) = exec.claim_blocking() { - let turn = claimed.run(exec.budget()); - exec.finish(claimed, turn); + let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + claimed.run(exec.budget()) + })); + match ran { + Ok(turn) => exec.finish(claimed, turn), + Err(p) => exec.abandon(claimed, panic_text(&*p)), + } } }) }) @@ -222,13 +332,26 @@ impl Pool { &self.exec } - /// Waits for every unit to go quiet, stops the workers and returns the executor. - pub fn finish(self) -> std::sync::Arc { - self.exec.wait_idle(); + fn join(&mut self) { self.exec.shutdown(); - for w in self.workers { - w.join().expect("a worker panicked"); + for w in std::mem::take(&mut self.workers) { + w.join().expect("a worker panicked outside a turn"); } - self.exec + } + + /// Waits for every unit to go quiet and stops the workers. A unit that panicked is the + /// error; the workers are stopped either way. + pub fn finish(mut self) -> Result, UnitPanic> { + let idle = self.exec.wait_idle(); + self.join(); + idle.map(|()| std::sync::Arc::clone(&self.exec)) + } +} + +/// A pool that is dropped stops its threads, discarding what is queued. +#[cfg(not(target_arch = "wasm32"))] +impl Drop for Pool { + fn drop(&mut self) { + self.join(); } } diff --git a/crates/improvise-incremental/src/staged.rs b/crates/improvise-incremental/src/staged.rs index 1b18f2f2..a45ef06b 100644 --- a/crates/improvise-incremental/src/staged.rs +++ b/crates/improvise-incremental/src/staged.rs @@ -4,7 +4,7 @@ use crate::sched::{Executor, Turn, Unit}; use std::sync::{Arc, Mutex}; use zset::{ - ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Flow, Frame, Input, Output, Stream, ZSet, + ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Frame, Input, Output, Stream, ZSet, channel_pipes, }; @@ -76,10 +76,10 @@ impl Unit for Head { for d in commands { self.sink.committed(d); } - let flow = self.sink.pump(); + self.sink.pump(); Turn { progress: self.sink.traffic() != before, - more: flow == Flow::Backlogged, + more: false, } } @@ -106,10 +106,10 @@ impl Unit for Aggregate { self.circuit.commit(); self.sink.committed_from(&mut self.circuit, &self.out); } - let flow = self.sink.pump(); + self.sink.pump(); Turn { progress: self.source.traffic() + self.sink.traffic() != before, - more: read == budget || flow == Flow::Backlogged, + more: false, } } diff --git a/crates/improvise-incremental/tests/sched.rs b/crates/improvise-incremental/tests/sched.rs index d4dee3dd..57047438 100644 --- a/crates/improvise-incremental/tests/sched.rs +++ b/crates/improvise-incremental/tests/sched.rs @@ -1,11 +1,14 @@ //! The scheduler: sub-circuits on an executor equal one circuit, whatever the number of workers //! and the order they run in. Evidence: `context/test-evidence/improvise-incremental.md`. +mod support; + use improvise_incremental::sched::{Claimed, Executor, Pool, Turn, Unit}; use improvise_incremental::staged::{Cell, Staged, single_circuit}; use proptest::prelude::*; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use support::watchdog::within; use zset::ZSet; fn command() -> impl Strategy> { @@ -96,29 +99,31 @@ fn run_some(exec: &Executor, workers: usize, choices: &mut impl Iterator> = (0..400u32) - .map(|n| { - [( - ((n % 3) as u8, (n / 3 % 3) as u8, (n % 7) as i64 - 3), - 1 - (n % 3) as i64, - )] - .into_iter() - .collect() - }) - .collect(); - for workers in 1..=4 { - let staged = Staged::new(2, 3); - let pool = Pool::start(staged.executor().clone(), workers); - for d in &commands { - staged.submit(d.clone()); + within(10, "real worker threads", || { + let commands: Vec> = (0..400u32) + .map(|n| { + [( + ((n % 3) as u8, (n / 3 % 3) as u8, (n % 7) as i64 - 3), + 1 - (n % 3) as i64, + )] + .into_iter() + .collect() + }) + .collect(); + for workers in 1..=4 { + let staged = Staged::new(2, 3); + let pool = Pool::start(staged.executor().clone(), workers); + for d in &commands { + staged.submit(d.clone()); + } + pool.finish().unwrap(); + assert_eq!( + staged.grid(), + single_circuit(&commands), + "[threads] {workers} workers" + ); } - pool.finish(); - assert_eq!( - staged.grid(), - single_circuit(&commands), - "[threads] {workers} workers" - ); - } + }); } /// A unit that records which turns ran, and asks for `more` a fixed number of times. @@ -226,32 +231,34 @@ impl Unit for Exclusive { #[test] fn at_most_one_worker_runs_a_unit_however_often_it_is_kicked() { - let (inside, turns) = ( - Arc::new(AtomicBool::new(false)), - Arc::new(AtomicUsize::new(0)), - ); - let exec = Arc::new(Executor::new( - vec![Box::new(Exclusive { - inside: inside.clone(), - turns: turns.clone(), - })], - 1, - )); - let pool = Pool::start(exec.clone(), 4); - let kickers: Vec<_> = (0..4) - .map(|_| { - let exec = exec.clone(); - std::thread::spawn(move || { - for _ in 0..500 { - exec.kick(0); - std::thread::yield_now(); - } + within(10, "kicking threads", || { + let (inside, turns) = ( + Arc::new(AtomicBool::new(false)), + Arc::new(AtomicUsize::new(0)), + ); + let exec = Arc::new(Executor::new( + vec![Box::new(Exclusive { + inside: inside.clone(), + turns: turns.clone(), + })], + 1, + )); + let pool = Pool::start(exec.clone(), 4); + let kickers: Vec<_> = (0..4) + .map(|_| { + let exec = exec.clone(); + std::thread::spawn(move || { + for _ in 0..500 { + exec.kick(0); + std::thread::yield_now(); + } + }) }) - }) - .collect(); - for k in kickers { - k.join().unwrap(); - } - pool.finish(); - assert!(turns.load(Ordering::SeqCst) > 0, "[own] it ran"); + .collect(); + for k in kickers { + k.join().unwrap(); + } + pool.finish().unwrap(); + assert!(turns.load(Ordering::SeqCst) > 0, "[own] it ran"); + }); } diff --git a/crates/improvise-incremental/tests/sched_dag.rs b/crates/improvise-incremental/tests/sched_dag.rs new file mode 100644 index 00000000..47648d45 --- /dev/null +++ b/crates/improvise-incremental/tests/sched_dag.rs @@ -0,0 +1,384 @@ +//! A 7-unit DAG with fan-out and fan-in on real threads and on virtual workers, against the same +//! graph in one circuit (from the review of #364). Evidence: `context/test-evidence/improvise-incremental.md`. + +mod support; + +use improvise_incremental::sched::{Claimed, Executor, Pool, Turn, Unit}; +use std::sync::{Arc, Mutex}; +use support::watchdog::within; +use zset::{ + ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Flow, Frame, Output, Stream, ZSet, + channel_pipes, +}; + +type P = ChannelPipe>; +type Z = ZSet; + +struct Rng(u64); +impl Rng { + fn next(&mut self) -> u64 { + self.0 ^= self.0 << 13; + self.0 ^= self.0 >> 7; + self.0 ^= self.0 << 17; + self.0 + } + fn below(&mut self, n: u64) -> u64 { + self.next() % n + } +} + +fn st_a(c: &mut Circuit, s: &Stream) -> Stream { + c.map(s, |x| x * 2) +} +fn st_b(c: &mut Circuit, s: &Stream) -> Stream { + c.filter(s, |x| x % 3 != 0) +} +fn st_c(c: &mut Circuit, a: &Stream, b: &Stream) -> Stream { + let u = c.plus(a, b); + let k = c.map(&u, |x| (x.rem_euclid(7), *x)); + let g = c.aggregate(&k, |_, vs| Some(vs.iter().map(|(v, w)| v * w).sum::())); + c.map(&g, |(k, s)| k * 1_000_000 + s) +} +fn st_d(c: &mut Circuit, s: &Stream) -> Stream { + let m = c.map(s, |x| x.rem_euclid(1000)); + c.distinct(&m) +} +fn st_e(c: &mut Circuit, s: &Stream) -> Stream { + c.filter(s, |x| x % 2 == 0) +} +fn st_f(c: &mut Circuit, s: &Stream) -> Stream { + c.map(s, |x| x / 2) +} + +fn reference(cmds: &[Z]) -> (Z, Z) { + let mut c = Circuit::new(); + let (i, s) = c.input::(); + let a = st_a(&mut c, &s); + let b = st_b(&mut c, &s); + let cc = st_c(&mut c, &a, &b); + let d = st_d(&mut c, &cc); + let e = st_e(&mut c, &cc); + let f = st_f(&mut c, &e); + let (od, of) = (c.output(&d), c.output(&f)); + for d in cmds { + c.push(&i, d.clone()); + c.step(); + } + (c.take(&od), c.take(&of)) +} + +struct Head { + sinks: Vec>, + inbox: Arc>>, + around: Vec, +} +impl Unit for Head { + fn turn(&mut self, _: usize) -> Turn { + let before: u64 = self.sinks.iter().map(|s| s.traffic()).sum(); + let cmds = std::mem::take(&mut *self.inbox.lock().unwrap()); + for d in cmds { + for s in &mut self.sinks { + s.committed(d.clone()); + } + } + let mut more = false; + for s in &mut self.sinks { + more |= s.pump() == Flow::Backlogged; + } + let after: u64 = self.sinks.iter().map(|s| s.traffic()).sum(); + Turn { + progress: after != before, + more, + } + } + fn neighbours(&self) -> &[usize] { + &self.around + } +} + +struct Node { + c: Circuit, + sources: Vec>, + sinks: Vec<(EdgeSink, Output)>, + collect: Option<(Output, Arc>)>, + around: Vec, +} +impl Node { + fn traffic(&self) -> u64 { + self.sources.iter().map(|s| s.traffic()).sum::() + + self.sinks.iter().map(|(s, _)| s.traffic()).sum::() + } +} +impl Unit for Node { + fn turn(&mut self, budget: usize) -> Turn { + let before = self.traffic(); + let mut read = 0; + let mut more = false; + for s in &mut self.sources { + let r = s.poll(&mut self.c, budget); + more |= r == budget; + read += r; + } + if read > 0 { + self.c.begin(); + self.c.step(); + self.c.commit(); + for (s, o) in &mut self.sinks { + s.committed_from(&mut self.c, o); + } + if let Some((o, z)) = &self.collect { + *z.lock().unwrap() += self.c.take(o); + } + } + for (s, _) in &mut self.sinks { + more |= s.pump() == Flow::Backlogged; + } + Turn { + progress: self.traffic() != before, + more, + } + } + fn neighbours(&self) -> &[usize] { + &self.around + } +} + +struct Dag { + exec: Arc, + inbox: Arc>>, + d: Arc>, + f: Arc>, +} + +fn dag(rng: &mut Rng, budget: usize) -> Dag { + let mut cap = || 1 + rng.below(3) as usize; + let (h_a, a_in) = channel_pipes::>(cap()); + let (h_b, b_in) = channel_pipes::>(cap()); + let (a_c, c_in_a) = channel_pipes::>(cap()); + let (b_c, c_in_b) = channel_pipes::>(cap()); + let (c_d, d_in) = channel_pipes::>(cap()); + let (c_e, e_in) = channel_pipes::>(cap()); + let (e_f, f_in) = channel_pipes::>(cap()); + let inbox = Arc::new(Mutex::new(Vec::new())); + let (dz, fz) = ( + Arc::new(Mutex::new(Z::new())), + Arc::new(Mutex::new(Z::new())), + ); + let head = Head { + sinks: vec![EdgeSink::new(Direct, h_a, 1), EdgeSink::new(Direct, h_b, 1)], + inbox: inbox.clone(), + around: vec![1, 2], + }; + let unary = |pipe_in: P, + f: fn(&mut Circuit, &Stream) -> Stream, + out: Option

, + col: Option>>, + around: Vec| { + let mut c = Circuit::new(); + let (i, s) = c.input::(); + let o = f(&mut c, &s); + let o = c.output(&o); + let source = EdgeSource::new(Direct, pipe_in, i); + let (sinks, collect) = match (out, col) { + (Some(p), None) => (vec![(EdgeSink::new(Direct, p, 1), o)], None), + (None, Some(z)) => (vec![], Some((o, z))), + _ => unreachable!(), + }; + Node { + c, + sources: vec![source], + sinks, + collect, + around, + } + }; + let a = unary(a_in, st_a, Some(a_c), None, vec![0, 3]); + let b = unary(b_in, st_b, Some(b_c), None, vec![0, 3]); + let cnode = { + let mut c = Circuit::new(); + let (ia, sa) = c.input::(); + let (ib, sb) = c.input::(); + let o = st_c(&mut c, &sa, &sb); + let o1 = c.output(&o); + let o2 = c.output(&o); + Node { + c, + sources: vec![ + EdgeSource::new(Direct, c_in_a, ia), + EdgeSource::new(Direct, c_in_b, ib), + ], + sinks: vec![ + (EdgeSink::new(Direct, c_d, 1), o1), + (EdgeSink::new(Direct, c_e, 1), o2), + ], + collect: None, + around: vec![1, 2, 4, 5], + } + }; + let d = unary(d_in, st_d, None, Some(dz.clone()), vec![3]); + let e = unary(e_in, st_e, Some(e_f), None, vec![3, 6]); + let f = unary(f_in, st_f, None, Some(fz.clone()), vec![5]); + let units: Vec> = vec![ + Box::new(head), + Box::new(a), + Box::new(b), + Box::new(cnode), + Box::new(d), + Box::new(e), + Box::new(f), + ]; + let exec = Arc::new(Executor::new(units, budget)); + exec.kick_all(); + Dag { + exec, + inbox, + d: dz, + f: fz, + } +} + +fn commands(rng: &mut Rng, n: usize) -> Vec { + (0..n) + .map(|_| { + let k = rng.below(4); + (0..k) + .map(|_| (rng.below(40) as i64 - 20, rng.below(5) as i64 - 2)) + .collect() + }) + .collect() +} + +fn submit(g: &Dag, d: Z) { + g.inbox.lock().unwrap().push(d); + g.exec.kick(0); +} + +fn out(g: &Dag) -> (Z, Z) { + (g.d.lock().unwrap().clone(), g.f.lock().unwrap().clone()) +} + +#[test] +fn a_dag_on_real_threads_equals_one_circuit_and_idle_means_done() { + within(30, "the DAG on a pool", || { + let mut rng = Rng(0x9e3779b97f4a7c15); + for round in 0..8 { + let workers = 1 + round % 8; + let budget = 1 + rng.below(4) as usize; + let g = dag(&mut rng, budget); + let cmds = commands(&mut rng, 300); + let pool = Pool::start(g.exec.clone(), workers); + // Concurrent submitter thread, with idle checkpoints from the host. + let mut at = 0; + while at < cmds.len() { + let chunk = 1 + rng.below(80) as usize; + let end = (at + chunk).min(cmds.len()); + for d in &cmds[at..end] { + submit(&g, d.clone()); + if rng.below(7) == 0 { + std::thread::yield_now(); + } + } + at = end; + if rng.below(3) == 0 { + pool.executor().wait_idle().unwrap(); + assert_eq!( + out(&g), + reference(&cmds[..at]), + "[idle] round {round} w{workers} at {at}" + ); + } + } + pool.finish().unwrap(); + assert_eq!( + out(&g), + reference(&cmds), + "[final] round {round} w{workers} b{budget}" + ); + } + }); +} + +#[test] +fn submitter_threads_race_the_pool() { + within(30, "racing submitters", || { + let mut rng = Rng(0x1234567); + for workers in 1..=4 { + let g = Arc::new(dag(&mut rng, 2)); + let cmds = commands(&mut rng, 600); + let pool = Pool::start(g.exec.clone(), workers); + // Four submitters; order between them is free, the result is a sum so order-free. + let parts: Vec> = (0..4) + .map(|k| cmds.iter().skip(k).step_by(4).cloned().collect()) + .collect(); + let hs: Vec<_> = parts + .into_iter() + .map(|p| { + let g = g.clone(); + std::thread::spawn(move || { + for d in p { + submit(&g, d); + } + }) + }) + .collect(); + for h in hs { + h.join().unwrap(); + } + pool.finish().unwrap(); + // `distinct` and aggregates are functions of the integral, so order-free. + assert_eq!(out(&g), reference(&cmds), "[race] w{workers}"); + } + }); +} + +fn run_virtual(exec: &Executor, workers: usize, rng: &mut Rng, steps: usize, drain: bool) { + let mut held: Vec<(Claimed, Turn)> = Vec::new(); + let mut n = 0; + loop { + if !drain && n >= steps { + break; + } + n += 1; + assert!(n < 2_000_000, "never quiet"); + let c = rng.next() as usize; + if held.len() < workers + && (held.is_empty() || !c.is_multiple_of(3)) + && let Some(mut cl) = exec.claim(|k| c / 3 % k) + { + let t = cl.run(exec.budget()); + held.push((cl, t)); + continue; + } + if held.is_empty() { + if drain { + break; + } + continue; + } + let (cl, t) = held.remove(c / 3 % held.len()); + exec.finish(cl, t); + } + for (cl, t) in held { + exec.finish(cl, t); + } +} + +#[test] +fn a_dag_on_virtual_workers_in_any_order_equals_one_circuit() { + let mut rng = Rng(0xdeadbeef); + for round in 0..60 { + let workers = 1 + round % 8; + let budget = 1 + rng.below(3) as usize; + let g = dag(&mut rng, budget); + let n = 1 + rng.below(40) as usize; + let cmds = commands(&mut rng, n); + for d in &cmds { + submit(&g, d.clone()); + let s = rng.below(10) as usize; + run_virtual(&g.exec, workers, &mut rng, s, false); + } + run_virtual(&g.exec, workers, &mut rng, 0, true); + assert!(g.exec.idle()); + assert_eq!(out(&g), reference(&cmds), "[virt] round {round}"); + } +} diff --git a/crates/improvise-incremental/tests/sched_threads.rs b/crates/improvise-incremental/tests/sched_threads.rs new file mode 100644 index 00000000..2ca25165 --- /dev/null +++ b/crates/improvise-incremental/tests/sched_threads.rs @@ -0,0 +1,215 @@ +//! The scheduler under panics, idle workers and a full edge, with real threads behind a +//! watchdog. Evidence: `context/test-evidence/improvise-incremental.md`. + +mod support; + +use improvise_incremental::sched::{Executor, Pool, Turn, Unit}; +use improvise_incremental::staged::{Cell, Staged, single_circuit}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use support::watchdog::within; +use zset::ZSet; + +struct Boom; + +impl Unit for Boom { + fn turn(&mut self, _: usize) -> Turn { + panic!("boom"); + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + +/// A unit that counts its turns. +struct Count(Arc); + +impl Unit for Count { + fn turn(&mut self, _: usize) -> Turn { + self.0.fetch_add(1, Ordering::SeqCst); + Turn { + progress: false, + more: false, + } + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + +#[test] +fn a_panicking_turn_does_not_hang_finish() { + within(5, "Pool::finish after a unit panicked", || { + let exec = Arc::new(Executor::new(vec![Box::new(Boom)], 1)); + let pool = Pool::start(exec.clone(), 2); + exec.kick(0); + let _ = pool.finish(); + }); +} + +#[test] +fn a_panicking_unit_is_reported_and_its_neighbours_still_run() { + let turns = Arc::new(AtomicUsize::new(0)); + let t = turns.clone(); + let err = within(5, "finish with one dead unit", move || { + let units: Vec> = vec![Box::new(Boom), Box::new(Count(t))]; + let exec = Arc::new(Executor::new(units, 1)); + let pool = Pool::start(exec.clone(), 2); + exec.kick_all(); + let err = pool.finish().err(); + exec.kick(0); + exec.kick(1); + err + }); + let err = err.expect("[panic] finish reports the panic"); + assert_eq!(err.unit, 0); + assert!( + err.message.contains("boom"), + "[panic] the payload: {}", + err.message + ); + assert!( + turns.load(Ordering::SeqCst) >= 1, + "[panic] the live unit ran" + ); +} + +#[test] +fn a_dead_unit_is_never_run_again() { + let exec = Executor::new(vec![Box::new(Boom)], 1); + exec.kick(0); + let mut c = exec.claim(|_| 0).unwrap(); + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| c.run(1))); + assert!(r.is_err()); + exec.abandon(c, "boom".to_string()); + exec.kick(0); + assert!( + exec.claim(|_| 0).is_none(), + "[dead] a kick does not revive it" + ); + assert!(exec.idle()); +} + +#[test] +#[should_panic(expected = "no such unit")] +fn a_kick_of_a_unit_that_does_not_exist_panics_before_the_lock() { + Executor::new(vec![Box::new(Boom)], 1).kick(5); +} + +#[test] +fn an_executor_survives_a_bad_kick() { + let exec = Executor::new(vec![Box::new(Boom)], 1); + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| exec.kick(5))); + assert!(r.is_err()); + assert!( + exec.idle(), + "[lock] the lock was not poisoned by the bad index" + ); +} + +#[test] +fn every_worker_stays_through_wakes_with_nothing_queued() { + within(10, "the pool with idle wakes", || { + let turns = Arc::new(AtomicUsize::new(0)); + let units: Vec> = + vec![Box::new(Count(turns.clone())), Box::new(Count(turns))]; + let exec = Arc::new(Executor::new(units, 1)); + let pool = Pool::start(exec.clone(), 4); + assert_eq!(exec.workers(), 4); + for round in 0..200 { + exec.kick(round % 2); + if round % 20 == 0 { + let _ = exec.wait_idle(); + } + } + let _ = exec.wait_idle(); + assert_eq!( + exec.workers(), + 4, + "[workers] a wake with nothing queued sent a worker home" + ); + let _ = pool.finish(); + assert_eq!(exec.workers(), 0, "[workers] finish stops them all"); + }); +} + +#[test] +fn a_pool_that_is_dropped_stops_its_threads() { + within(10, "dropping a pool", || { + let exec = Arc::new(Executor::new(vec![Box::new(Count(Arc::default()))], 1)); + { + let _pool = Pool::start(exec.clone(), 3); + assert_eq!(exec.workers(), 3); + } + assert_eq!(exec.workers(), 0, "[drop] no thread is left parked"); + }); +} + +fn cmd(r: u8, c: u8, v: i64) -> ZSet { + [((r, c, v), 1)].into_iter().collect() +} + +#[test] +fn a_full_edge_does_not_make_its_sender_spin() { + let staged = Staged::new(1, 1); + let exec = staged.executor(); + // Hold the aggregate and the projection, so nothing drains the head's edge. + let agg = exec.claim(|_| 1).unwrap(); + let proj = exec.claim(|_| 1).unwrap(); + assert_eq!((agg.id(), proj.id()), (1, 2)); + for n in 0..4 { + staged.submit(cmd(0, 0, n)); + let mut head = exec.claim(|_| 0).unwrap(); + let t = head.run(exec.budget()); + exec.finish(head, t); + } + let mut turns = 0; + while let Some(mut head) = exec.claim(|_| 0) { + turns += 1; + assert!( + turns < 50, + "[spin] the head keeps asking to run with its edge full" + ); + let t = head.run(exec.budget()); + exec.finish(head, t); + } + let t = Turn { + progress: false, + more: false, + }; + exec.finish(agg, t); + exec.finish(proj, t); +} + +#[test] +fn a_unit_whose_kicks_coalesced_still_drains_its_edge() { + let staged = Staged::new(4, 1); + let exec = staged.executor(); + // Settle the start-up requests. + while let Some(mut u) = exec.claim(|_| 0) { + let t = u.run(exec.budget()); + exec.finish(u, t); + } + // Two commands cross before the aggregate runs: its two kicks are one queue entry. + staged.submit(cmd(0, 0, 1)); + let mut head = exec.claim(|_| 0).unwrap(); + let t = head.run(1); + exec.finish(head, t); + staged.submit(cmd(1, 1, 2)); + let mut head = exec.claim(|n| n - 1).unwrap(); + assert_eq!(head.id(), 0); + let t = head.run(1); + exec.finish(head, t); + while let Some(mut u) = exec.claim(|_| 0) { + let t = u.run(1); + exec.finish(u, t); + } + let want = single_circuit(&[cmd(0, 0, 1), cmd(1, 1, 2)]); + assert_eq!( + staged.grid(), + want, + "[coalesced] the second frame was left unread" + ); +} diff --git a/crates/improvise-incremental/tests/support/mod.rs b/crates/improvise-incremental/tests/support/mod.rs index 7a396c42..3ccde543 100644 --- a/crates/improvise-incremental/tests/support/mod.rs +++ b/crates/improvise-incremental/tests/support/mod.rs @@ -3,6 +3,7 @@ pub mod live; pub mod remote; +pub mod watchdog; use improvise_core::formula::ast::AggFunc; use improvise_core::formula::parse_formula; diff --git a/crates/improvise-incremental/tests/support/watchdog.rs b/crates/improvise-incremental/tests/support/watchdog.rs new file mode 100644 index 00000000..ab09ab12 --- /dev/null +++ b/crates/improvise-incremental/tests/support/watchdog.rs @@ -0,0 +1,23 @@ +//! A hung real-thread test must fail by name, not hang the binary. + +use std::sync::mpsc; +use std::thread; +use std::time::Duration; + +/// Runs `f` on its own thread and fails with `what` if it has not returned in `secs`. A panic in +/// `f` is re-raised here. A timed-out thread is left behind; the process exits without it. +pub fn within( + secs: u64, + what: &str, + f: impl FnOnce() -> T + Send + 'static, +) -> T { + let (tx, rx) = mpsc::channel(); + thread::spawn(move || { + let _ = tx.send(std::panic::catch_unwind(std::panic::AssertUnwindSafe(f))); + }); + match rx.recv_timeout(Duration::from_secs(secs)) { + Ok(Ok(v)) => v, + Ok(Err(p)) => std::panic::resume_unwind(p), + Err(_) => panic!("[watchdog] {what} did not finish in {secs}s"), + } +} diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs index f6de5a09..1a4e0058 100644 --- a/crates/zset/tests/edge.rs +++ b/crates/zset/tests/edge.rs @@ -628,7 +628,11 @@ fn traffic_never_goes_down_across_a_restart_of_either_end() { let mut c2 = Circuit::new(); let (i2, _) = c2.input::(); let src = src.restart(i2); - assert_eq!((sink.traffic(), src.traffic()), (s, r), "[restart] counters kept"); + assert_eq!( + (sink.traffic(), src.traffic()), + (s, r), + "[restart] counters kept" + ); } #[test]