diff --git a/context/code/improvise-incremental.md b/context/code/improvise-incremental.md index 986b9ee4..566da8da 100644 --- a/context/code/improvise-incremental.md +++ b/context/code/improvise-incremental.md @@ -504,3 +504,77 @@ change log, no stems. These are the readings taken. - **The two-second limit decided the shape of the tests** more than the design did: the corpus differential over a scanning source is 2.5s untrimmed, so it follows `corpus_eager`'s trimming, and each random test is its own binary. + +## Scheduling: sub-circuits on a worker pool (user ruling 2026-10-11) + +Multi-threading here is **scheduling**. Each sub-circuit stays synchronous and single-threaded +(`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`), 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 + `neighbours()`). Placement is manual: which nodes sit in which unit is how the graph was built, + and `staged::PLACEMENT` names it for the example. Two node-building functions (`aggregate_nodes`, + `grid_nodes`) build the same nodes into one circuit (`single_circuit`, the reference) or into + two. +2. **Runnable means queued.** `Executor::kick(id)` queues a unit once. A unit is kicked by the + host (`Staged::submit`), and by a neighbour whose turn moved a frame (`Turn::progress`, which a + unit derives from its edges' `traffic()`). A kick reaches both directions, because a request + flows back along an edge. Progress is a count of frames, so it ends: two neighbours cannot kick + each other for ever. +3. **The ownership token is the unit itself.** A claimed unit is taken out of its slot, so a second + worker cannot reach it. A kick that arrives while it runs sets `again`, and the owner queues it + when it hands the unit back (`finish`). No lock is held during a turn; one mutex guards the + 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 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 + units held at once and the turn of a held unit already run, so a kick can land on a running + unit. Both drive the same code. +6. **Results do not depend on N or the order.** The deltas sum, and every operator in the graph is + a function of its input's integral. So any schedule settles where the single circuit does + (`placed_circuits_equal_one_circuit_for_any_workers_and_order`, 1 to 4 workers, random + commands, random order, commands arriving while it runs; and a real pool of 1 to 4 threads). + Per-tick outputs (`delay`, `differentiate`) would differ with coalescing, and the example has + none. +7. **What the example is.** `Staged`: the host's end of the first edge, an aggregate stage (sum of + a cell's value per row and column, `aggregate`) and a grid projection (the cell each group draws, + 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. **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_waits_for_its_workers`), discarding what is queued. A dead unit is + dropped inside `catch_unwind` (`bury`), because a drop that panics would otherwise kill the + worker after the bookkeeping and hang or shrink the pool; `Pool::join` does not panic over a + dead worker while the host is already unwinding. `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/code/zset.md b/context/code/zset.md index ec838232..3f0215d1 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -396,3 +396,5 @@ Nothing here spawns a thread; whoever owns each circuit calls `pump` and `poll`. `Moved` correction). A pipe that blocks or notifies, because `try_send` and `try_recv` are polling. Key-sharded data parallelism inside a circuit: its exchange step would be this same edge, and nothing here forbids it. + +11. **Progress is counted.** `EdgeSink::traffic` and `EdgeSource::traffic` count frames sent and read and requests sent and read, so a scheduler can tell that a turn moved something (`improvise-incremental.md`, "Scheduling"). diff --git a/context/test-evidence/improvise-incremental.md b/context/test-evidence/improvise-incremental.md index abb3b697..ab0cd8ed 100644 --- a/context/test-evidence/improvise-incremental.md +++ b/context/test-evidence/improvise-incremental.md @@ -957,3 +957,112 @@ differs). `Pushdown::new` took a generic `Arc` that could not take a `SharedSource`; the test builds one from the alias and did not compile against the generic signature. + +### placed_circuits_equal_one_circuit_for_any_workers_and_order + +Random commands (twelve at most), 1 to 4 virtual workers, edge capacity 1 to 3, a budget of 1 to +3 frames and a random order of claim and hand-back, with commands arriving between scheduler steps, +settle to the grid of one circuit running the whole graph. Rivals, each failing it: a kick of a +running unit queues it as well (S1, claiming a running unit panics), a kick while running is lost +(S2), a unit that wants more goes to the front of the queue (S3, it never lets the others run), and +progress that does not kick the neighbours (S4, the graph stalls). + +### real_worker_threads_equal_one_circuit + +The same graph on a real `Pool` of 1 to 4 threads, 400 commands, capacity 2, budget 3. Rivals S1 +and S4 fail it. + +### a_unit_that_wants_more_goes_to_the_back_of_the_queue + +Units 0 (wants two more turns), 1 and 2 run as 0, 1, 2, 0, 0. Rival S3 runs 0 again before 1. + +### a_kick_while_a_unit_runs_queues_it_again_and_never_claims_it_twice + +A claimed unit is not claimable; a kick during its turn leaves the executor not idle after it is +handed back. Rivals S1 and S2 fail it. + +### at_most_one_worker_runs_a_unit_however_often_it_is_kicked + +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_waits_for_its_workers + +A unit sleeps 300 ms in its turn; the pool is dropped while it runs. Drop must join, so the turn +has finished and no worker is left when it returns. The first version counted workers after the +drop, which a `Drop` that did not join could pass by timing. Rival D2 (shutdown without join) +fails it every time. + +### 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 frames reach the projection's edge while the projection is never run (it is claimed and +handed back unrun), then it is kicked once with a budget of 1. It must ask to run again, or the +second frame is never read. The first version let the units alternate, which hid the clause and +made the evidence wrong; rival M5 now fails this test as well as +`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. + +### a_unit_whose_drop_panics_does_not_hang_or_shrink_the_pool + +`bury` dropped the dead unit outside any catch_unwind, so a unit whose drop panics killed its +worker after the bookkeeping: on a one-worker pool the next queued unit never ran and +`Pool::finish` hung (a 5 s watchdog). The drop is now caught. The test checks the one worker is +still alive, the dead unit is reported and the other unit ran. Rival D1 fails it. + +### dropping_a_pool_while_unwinding_does_not_abort + +`Pool::drop` joined with `expect`, so a dead worker double-panicked while the host unwound and +aborted the process. Now a dead worker is a panic only when not already panicking. With the drop +caught no worker dies, so the test also passes without that guard (rival D3 does not fail it); +the guard is defence for a worker lost another way, and is not pinned. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index 6361d626..64cd492b 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -747,6 +747,11 @@ Three hundred commands on a producer thread, a consumer on another, capacity 2, yielding at uneven points and pumping on only two of three commands. The consumer's output equals the combined circuit's. Rival R1 fails it. +### a_poll_reads_at_most_the_frames_it_is_given + +Three corrections waiting, `poll(c, 2)` reads two and the next reads one, so a unit's turn is +bounded by its budget. Rival: the bound removed from the loop fails the first assertion. + Ids index an arena, so a foreign handle from a circuit of the same shape reads or wires the other circuit's slot. Ten cases (`current`, `take`, `push`, a unary operator, both operands of a binary one, `output`, `delay`, `integrate`, `distinct`) each need the panic text "handle from @@ -764,6 +769,12 @@ 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. + ### review_token_collision_after_new_source The old receiver's request 2 is answered by E1 and the reply is delayed; E1 dies, E2 takes over; diff --git a/crates/improvise-incremental/src/lib.rs b/crates/improvise-incremental/src/lib.rs index e809ed5a..b926303c 100644 --- a/crates/improvise-incremental/src/lib.rs +++ b/crates/improvise-incremental/src/lib.rs @@ -12,7 +12,9 @@ mod incremental; mod pushdown; mod reading; mod reads; +pub mod sched; mod shapes; +pub mod staged; pub use delta::{CellDelta, DeltaValue, cell_delta, moved_records}; pub use eager::Eager; diff --git a/crates/improvise-incremental/src/sched.rs b/crates/improvise-incremental/src/sched.rs new file mode 100644 index 00000000..0fd1e4fd --- /dev/null +++ b/crates/improvise-incremental/src/sched.rs @@ -0,0 +1,362 @@ +//! A scheduler for sub-circuits joined by edges: multi-threading as scheduling. +//! 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. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +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 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, +} + +/// A sub-circuit and the edge ends around it. One thread runs it at a time. +pub trait Unit: Send { + /// Runs at most `budget` frames' worth of work, then returns. + fn turn(&mut self, budget: usize) -> Turn; + /// The units on the other end of its edges, in either direction. + fn neighbours(&self) -> &[usize]; +} + +struct Slot { + /// `None` while a worker owns it: this is the ownership token. + unit: Option>, + queued: bool, + 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 { + slots: Vec, + queue: VecDeque, + running: usize, + shutdown: bool, + panics: Vec, +} + +/// A unit a worker owns until it hands it back with `finish`. +pub struct Claimed { + id: usize, + unit: Box, +} + +impl Claimed { + pub fn id(&self) -> usize { + self.id + } + + pub fn run(&mut self, budget: usize) -> Turn { + self.unit.turn(budget) + } +} + +/// 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`, 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 + .into_iter() + .map(|u| Slot { + unit: Some(u), + 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), + } + } + + pub fn budget(&self) -> usize { + 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("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 { + slot.queued = true; + st.queue.push_back(id); + } + } + + #[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(); + } + + pub fn kick_all(&self) { + let mut st = self.lock(); + for id in 0..st.slots.len() { + Self::kick_in(&mut st, id); + } + drop(st); + self.wake.notify_all(); + } + + fn take(st: &mut State, at: usize) -> Claimed { + let id = st.queue.remove(at).expect("a queued unit"); + let slot = &mut st.slots[id]; + slot.queued = false; + slot.running = true; + st.running += 1; + Claimed { + id, + unit: slot.unit.take().expect("a queued unit is at rest"), + } + } + + /// Takes the queued unit that `pick(queue length)` names, if any is queued. + pub fn claim(&self, pick: impl FnOnce(usize) -> usize) -> Option { + let mut st = self.lock(); + let n = st.queue.len(); + (n > 0).then(|| Self::take(&mut st, pick(n) % n)) + } + + /// 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.shutdown { + return None; + } + 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, 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 slot = &mut st.slots[id]; + slot.unit = Some(unit); + slot.running = false; + let again = std::mem::take(&mut slot.again); + st.running -= 1; + if again || turn.more { + Self::kick_in(&mut st, id); + } + if turn.progress { + for n in neighbours { + Self::kick_in(&mut st, n); + } + } + drop(st); + 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); + self.wake.notify_all(); + // A unit whose drop panics must not take the calling thread with it. + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || drop(unit))); + } + + /// 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`, 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("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 at once, whatever is queued. + pub fn shutdown(&self) { + self.lock().shutdown = true; + self.wake.notify_all(); + } +} + +/// N worker threads over one executor. Native only: a wasm32 build has no threads to spawn. +#[cfg(not(target_arch = "wasm32"))] +pub struct Pool { + exec: std::sync::Arc, + 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(|_| { + 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 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)), + } + } + }) + }) + .collect(); + Pool { exec, workers } + } + + pub fn executor(&self) -> &Executor { + &self.exec + } + + fn join(&mut self) { + self.exec.shutdown(); + for w in std::mem::take(&mut self.workers) { + let died = w.join().is_err(); + assert!( + !died || std::thread::panicking(), + "a worker died outside a turn" + ); + } + } + + /// 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 new file mode 100644 index 00000000..a45ef06b --- /dev/null +++ b/crates/improvise-incremental/src/staged.rs @@ -0,0 +1,216 @@ +//! One graph in two sub-circuits: the aggregate stage and the grid projection, placed on +//! separate units of the scheduler. Reasoning: `context/code/improvise-incremental.md`, "Scheduling". + +use crate::sched::{Executor, Turn, Unit}; +use std::sync::{Arc, Mutex}; +use zset::{ + ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Frame, Input, Output, Stream, ZSet, + channel_pipes, +}; + +/// A stored cell: row, column and a value; its weight counts it in or out. +pub type Cell = (u8, u8, i64); +/// A group of cells: its `(row, column)` and the sum of its values. +pub type Group = ((u8, u8), i64); +type P = ChannelPipe>; + +/// What the grid draws: each group's cell, and a total per row. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct Grid { + pub cells: ZSet<(u8, u8, i64)>, + pub rows: ZSet<(u8, i64)>, +} + +/// Which unit each node of the graph runs on (manual placement): the head is the host's end of +/// the first edge, the aggregate stage and the grid projection are circuits of their own. +pub const PLACEMENT: [(&str, usize); 3] = [("head", 0), ("aggregate", 1), ("grid", 2)]; + +/// The aggregate stage's nodes: the sum of the values of each `(row, column)`. +fn aggregate_nodes(c: &mut Circuit, cells: &Stream>) -> Stream> { + let keyed = c.map(cells, |(r, col, v)| ((*r, *col), *v)); + c.aggregate(&keyed, |_, vals| { + Some(vals.iter().map(|(v, w)| v * w).sum()) + }) +} + +type Drawn = (Output<(u8, u8, i64)>, Output<(u8, i64)>); + +/// The grid projection's nodes: the cell each group draws, and the sum of each row. +fn grid_nodes(c: &mut Circuit, groups: &Stream>) -> Drawn { + let drawn = c.map(groups, |((r, col), s)| (*r, *col, *s)); + let by_row = c.map(groups, |((r, _), s)| (*r, *s)); + let totals = c.aggregate(&by_row, |_, vals| { + Some(vals.iter().map(|(v, w)| v * w).sum()) + }); + let totals = c.map(&totals, |(r, t)| (*r, *t)); + (c.output(&drawn), c.output(&totals)) +} + +/// The whole graph in one circuit: what the placed one must equal. +pub fn single_circuit(commands: &[ZSet]) -> Grid { + let mut c = Circuit::new(); + let (input, cells) = c.input::(); + let groups = aggregate_nodes(&mut c, &cells); + let (drawn, totals) = grid_nodes(&mut c, &groups); + for d in commands { + c.push(&input, d.clone()); + c.step(); + } + Grid { + cells: c.take(&drawn), + rows: c.take(&totals), + } +} + +/// The host's end of the first edge: commands come in through the inbox, each one stamped. +struct Head { + sink: EdgeSink>, + inbox: Arc>>>, + next: [usize; 1], +} + +impl Unit for Head { + fn turn(&mut self, _budget: usize) -> Turn { + let before = self.sink.traffic(); + let commands = std::mem::take(&mut *self.inbox.lock().expect("the inbox")); + for d in commands { + self.sink.committed(d); + } + self.sink.pump(); + Turn { + progress: self.sink.traffic() != before, + more: false, + } + } + + fn neighbours(&self) -> &[usize] { + &self.next + } +} + +struct Aggregate { + circuit: Circuit, + out: Output, + source: EdgeSource>, + sink: EdgeSink>, + around: [usize; 2], +} + +impl Unit for Aggregate { + fn turn(&mut self, budget: usize) -> Turn { + let before = self.source.traffic() + self.sink.traffic(); + let read = self.source.poll(&mut self.circuit, budget); + if read > 0 { + self.circuit.begin(); + self.circuit.step(); + self.circuit.commit(); + self.sink.committed_from(&mut self.circuit, &self.out); + } + self.sink.pump(); + Turn { + progress: self.source.traffic() + self.sink.traffic() != before, + more: false, + } + } + + fn neighbours(&self) -> &[usize] { + &self.around + } +} + +struct Projection { + circuit: Circuit, + drawn: Output<(u8, u8, i64)>, + totals: Output<(u8, i64)>, + source: EdgeSource>, + grid: Arc>, + back: [usize; 1], +} + +impl Unit for Projection { + fn turn(&mut self, budget: usize) -> Turn { + let before = self.source.traffic(); + let read = self.source.poll(&mut self.circuit, budget); + if read > 0 { + self.circuit.step(); + let mut grid = self.grid.lock().expect("the grid"); + grid.cells += self.circuit.take(&self.drawn); + grid.rows += self.circuit.take(&self.totals); + } + Turn { + progress: self.source.traffic() != before, + more: read == budget, + } + } + + fn neighbours(&self) -> &[usize] { + &self.back + } +} + +/// The placed graph: three units on an executor, and the host's two handles on it. +pub struct Staged { + exec: Arc, + inbox: Arc>>>, + grid: Arc>, +} + +impl Staged { + /// `cap` frames per edge direction, `budget` frames per unit turn. + pub fn new(cap: usize, budget: usize) -> Self { + let inbox = Arc::new(Mutex::new(Vec::new())); + let grid = Arc::new(Mutex::new(Grid::default())); + let (head_end, agg_in) = channel_pipes::>(cap); + let (agg_out, grid_in) = channel_pipes::>(cap); + + let mut c = Circuit::new(); + let (input, cells): (Input, _) = c.input::(); + let groups = aggregate_nodes(&mut c, &cells); + let out = c.output(&groups); + let aggregate = Aggregate { + circuit: c, + out, + source: EdgeSource::new(Direct, agg_in, input), + sink: EdgeSink::new(Direct, agg_out, 2), + around: [0, 2], + }; + + let mut c = Circuit::new(); + let (input, groups) = c.input::(); + let (drawn, totals) = grid_nodes(&mut c, &groups); + let projection = Projection { + circuit: c, + drawn, + totals, + source: EdgeSource::new(Direct, grid_in, input), + grid: Arc::clone(&grid), + back: [1], + }; + + let head = Head { + sink: EdgeSink::new(Direct, head_end, 1), + inbox: Arc::clone(&inbox), + next: [1], + }; + let units: Vec> = + vec![Box::new(head), Box::new(aggregate), Box::new(projection)]; + let exec = Arc::new(Executor::new(units, budget)); + exec.kick_all(); + Staged { exec, inbox, grid } + } + + pub fn executor(&self) -> &Arc { + &self.exec + } + + /// One committed command from the host. + pub fn submit(&self, delta: ZSet) { + self.inbox.lock().expect("the inbox").push(delta); + self.exec.kick(0); + } + + /// What the grid has drawn so far. + pub fn grid(&self) -> Grid { + self.grid.lock().expect("the grid").clone() + } +} diff --git a/crates/improvise-incremental/tests/sched.rs b/crates/improvise-incremental/tests/sched.rs new file mode 100644 index 00000000..57047438 --- /dev/null +++ b/crates/improvise-incremental/tests/sched.rs @@ -0,0 +1,264 @@ +//! 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> { + prop::collection::vec(((0u8..3, 0u8..3, -4i64..=4), -2i64..=2), 0..4) + .prop_map(|v| v.into_iter().collect()) +} + +/// Runs the executor with `workers` virtual workers, choosing each step from `choices`: either +/// claim a queued unit (running its turn at once) or hand a held one back. Returns the steps. +fn run_deterministic( + exec: &Executor, + workers: usize, + choices: &mut impl Iterator, +) -> usize { + let mut held: Vec<(Claimed, Turn)> = Vec::new(); + let mut steps = 0; + loop { + let c = choices.next().unwrap_or(0) as usize; + let can_claim = held.len() < workers; + let mut started = false; + if can_claim + && (held.is_empty() || c.is_multiple_of(2)) + && let Some(mut claimed) = exec.claim(|n| c / 2 % n) + { + let turn = claimed.run(exec.budget()); + held.push((claimed, turn)); + started = true; + } + if !started { + if held.is_empty() { + return steps; + } + let (claimed, turn) = held.remove(c / 2 % held.len()); + exec.finish(claimed, turn); + } + steps += 1; + assert!(steps < 100_000, "[sched] the executor never went quiet"); + } +} + +proptest! { + /// The placed graph, on 1 to 4 workers in a random order with commands arriving while it + /// runs, ends where one circuit running the whole graph ends. + #[test] + fn placed_circuits_equal_one_circuit_for_any_workers_and_order( + commands in prop::collection::vec(command(), 0..12), + workers in 1usize..5, + cap in 1usize..4, + budget in 1usize..4, + choices in prop::collection::vec(any::(), 0..400), + gaps in prop::collection::vec(0usize..6, 0..12), + ) { + let staged = Staged::new(cap, budget); + let mut it = choices.into_iter(); + for (i, d) in commands.iter().enumerate() { + staged.submit(d.clone()); + // A few scheduler steps before the next command arrives. + let mut few = it.by_ref().take(gaps.get(i).copied().unwrap_or(0)); + run_some(staged.executor(), workers, &mut few); + } + run_deterministic(staged.executor(), workers, &mut std::iter::repeat(0)); + prop_assert_eq!(staged.grid(), single_circuit(&commands), "[sched] the placed graph"); + } +} + +/// A bounded number of claim-or-finish steps, leaving whatever is left in the queue. +fn run_some(exec: &Executor, workers: usize, choices: &mut impl Iterator) { + let mut held: Vec<(Claimed, Turn)> = Vec::new(); + for c in choices { + let c = c as usize; + if held.len() < workers + && (held.is_empty() || c.is_multiple_of(2)) + && let Some(mut claimed) = exec.claim(|n| c / 2 % n) + { + let turn = claimed.run(exec.budget()); + held.push((claimed, turn)); + continue; + } + if !held.is_empty() { + let (claimed, turn) = held.remove(c / 2 % held.len()); + exec.finish(claimed, turn); + } + } + for (claimed, turn) in held { + exec.finish(claimed, turn); + } +} + +#[test] +fn real_worker_threads_equal_one_circuit() { + 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" + ); + } + }); +} + +/// A unit that records which turns ran, and asks for `more` a fixed number of times. +struct Log { + id: usize, + more: usize, + ran: Arc>>, +} + +impl Unit for Log { + fn turn(&mut self, _: usize) -> Turn { + self.ran.lock().unwrap().push(self.id); + let more = self.more > 0; + self.more = self.more.saturating_sub(1); + Turn { + progress: false, + more, + } + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + +#[test] +fn a_unit_that_wants_more_goes_to_the_back_of_the_queue() { + let ran = Arc::new(std::sync::Mutex::new(Vec::new())); + let unit = |id, more| -> Box { + Box::new(Log { + id, + more, + ran: ran.clone(), + }) + }; + let exec = Executor::new(vec![unit(0, 2), unit(1, 0), unit(2, 0)], 1); + exec.kick_all(); + while let Some(mut c) = exec.claim(|_| 0) { + let t = c.run(exec.budget()); + exec.finish(c, t); + } + assert_eq!( + *ran.lock().unwrap(), + vec![0, 1, 2, 0, 0], + "[fair] the others run between its turns" + ); + assert!(exec.idle()); +} + +#[test] +fn a_kick_while_a_unit_runs_queues_it_again_and_never_claims_it_twice() { + let ran = Arc::new(std::sync::Mutex::new(Vec::new())); + let exec = Executor::new( + vec![Box::new(Log { + id: 0, + more: 0, + ran: ran.clone(), + })], + 1, + ); + exec.kick(0); + let mut claimed = exec.claim(|_| 0).unwrap(); + exec.kick(0); + assert!( + exec.claim(|_| 0).is_none(), + "[own] a running unit is not claimable" + ); + let t = claimed.run(1); + exec.finish(claimed, t); + assert!( + !exec.idle(), + "[again] the kick that arrived mid-turn is kept" + ); + let mut again = exec.claim(|_| 0).unwrap(); + let t = again.run(1); + exec.finish(again, t); + assert!(exec.idle()); +} + +/// A unit that fails if two turns of it ever overlap. +struct Exclusive { + inside: Arc, + turns: Arc, +} + +impl Unit for Exclusive { + fn turn(&mut self, _: usize) -> Turn { + assert!( + !self.inside.swap(true, Ordering::SeqCst), + "[own] two workers ran one unit" + ); + self.turns.fetch_add(1, Ordering::SeqCst); + std::thread::yield_now(); + self.inside.store(false, Ordering::SeqCst); + Turn { + progress: false, + more: false, + } + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + +#[test] +fn at_most_one_worker_runs_a_unit_however_often_it_is_kicked() { + 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().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..bf15ed9a --- /dev/null +++ b/crates/improvise-incremental/tests/sched_dag.rs @@ -0,0 +1,383 @@ +//! 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, 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()); + } + } + for s in &mut self.sinks { + s.pump(); + } + let after: u64 = self.sinks.iter().map(|s| s.traffic()).sum(); + Turn { + progress: after != before, + more: false, + } + } + 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); + // Only a leaf (nothing downstream to kick it back) asks to run again at its budget. + more |= self.sinks.is_empty() && 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 { + s.pump(); + } + 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_drop.rs b/crates/improvise-incremental/tests/sched_drop.rs new file mode 100644 index 00000000..6810903f --- /dev/null +++ b/crates/improvise-incremental/tests/sched_drop.rs @@ -0,0 +1,87 @@ +//! A unit that panics again when it is dropped must not cost the pool a worker. +//! Evidence: `context/test-evidence/improvise-incremental.md`. + +mod support; + +use improvise_incremental::sched::{Executor, Pool, Turn, Unit}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use support::watchdog::within; + +/// A unit that panics in its turn and again when it is dropped by the unwind's aftermath. +struct BoomTwice; + +impl Unit for BoomTwice { + fn turn(&mut self, _: usize) -> Turn { + panic!("boom in the turn"); + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + +impl Drop for BoomTwice { + fn drop(&mut self) { + if !std::thread::panicking() { + panic!("boom in the drop"); + } + } +} + +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_unit_whose_drop_panics_does_not_hang_or_shrink_the_pool() { + let turns = Arc::new(AtomicUsize::new(0)); + let t = turns.clone(); + let (err, workers) = within(5, "a pool with a unit that panics in its drop", move || { + let units: Vec> = vec![Box::new(BoomTwice), Box::new(Count(t))]; + let exec = Arc::new(Executor::new(units, 1)); + let pool = Pool::start(exec.clone(), 1); + exec.kick(0); + exec.kick(1); + let idle = exec.wait_idle(); + let workers = exec.workers(); + let _ = pool.finish(); + (idle.err(), workers) + }); + assert_eq!( + workers, 1, + "[drop] the only worker survived the dead unit's drop" + ); + assert_eq!(err.map(|e| e.unit), Some(0)); + assert_eq!( + turns.load(Ordering::SeqCst), + 1, + "[drop] the other unit still ran" + ); +} + +#[test] +fn dropping_a_pool_while_unwinding_does_not_abort() { + within(5, "unwinding past a pool", || { + let exec = Arc::new(Executor::new(vec![Box::new(BoomTwice)], 1)); + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _pool = Pool::start(exec.clone(), 1); + exec.kick(0); + let _ = exec.wait_idle(); + panic!("the host fails with a pool alive"); + })); + assert!(r.is_err()); + }); +} diff --git a/crates/improvise-incremental/tests/sched_threads.rs b/crates/improvise-incremental/tests/sched_threads.rs new file mode 100644 index 00000000..8fa5d76c --- /dev/null +++ b/crates/improvise-incremental/tests/sched_threads.rs @@ -0,0 +1,259 @@ +//! 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"); + }); +} + +/// A unit whose turn is slow, so a drop that does not join returns before it ends. +struct Slow { + started: Arc, + finished: Arc, +} + +impl Unit for Slow { + fn turn(&mut self, _: usize) -> Turn { + self.started.fetch_add(1, Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_millis(300)); + self.finished.fetch_add(1, Ordering::SeqCst); + Turn { + progress: false, + more: false, + } + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + +#[test] +fn a_pool_that_is_dropped_waits_for_its_workers() { + within(10, "dropping a pool", || { + let (started, finished) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0))); + let unit = Slow { + started: started.clone(), + finished: finished.clone(), + }; + let exec = Arc::new(Executor::new(vec![Box::new(unit)], 1)); + { + let _pool = Pool::start(exec.clone(), 3); + assert_eq!(exec.workers(), 3); + exec.kick(0); + while started.load(Ordering::SeqCst) == 0 { + std::thread::yield_now(); + } + } + assert_eq!( + finished.load(Ordering::SeqCst), + 1, + "[drop] the turn in flight was joined" + ); + assert_eq!(exec.workers(), 0, "[drop] no thread is left"); + }); +} + +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 frames reach the projection's edge while the projection is never run: its kicks + // coalesce into one queue entry, and a budget of 1 reads one frame per turn. + let skip = Turn { + progress: false, + more: false, + }; + for c in [cmd(0, 0, 1), cmd(1, 1, 2)] { + staged.submit(c); + while let Some(mut u) = exec.claim(|_| 0) { + if u.id() == 2 { + exec.finish(u, skip); + continue; + } + let t = u.run(1); + exec.finish(u, t); + } + } + exec.kick(2); + 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/src/edge.rs b/crates/zset/src/edge.rs index 8b23e41b..7461511f 100644 --- a/crates/zset/src/edge.rs +++ b/crates/zset/src/edge.rs @@ -149,6 +149,7 @@ pub struct EdgeSink { /// The token of the latest request read and not yet answered. owed: Option, closed: bool, + traffic: u64, } impl, P: Pipe> EdgeSink { @@ -174,6 +175,7 @@ impl, P: Pipe> EdgeSink { }, owed: None, closed: false, + traffic: 0, } } @@ -184,6 +186,11 @@ impl, P: Pipe> EdgeSink { } } + /// Frames sent and requests read so far: a host sees progress when it moves. + pub fn traffic(&self) -> u64 { + self.traffic + } + /// Whether any committed command has not been sent. pub fn backlog(&self) -> bool { self.waiting.to > self.waiting.from || self.owed.is_some() @@ -191,7 +198,11 @@ impl, P: Pipe> EdgeSink { /// The same pipe and codec, after a restart: a new incarnation at `seq` holding `integral`. pub fn restart(self, epoch: u64, seq: u64, integral: ZSet) -> Self { - Self::resume(self.codec, self.pipe, epoch, seq, integral) + let traffic = self.traffic; + EdgeSink { + traffic, + ..Self::resume(self.codec, self.pipe, epoch, seq, integral) + } } /// One committed command changed the output by `delta` (empty is still a command). @@ -224,6 +235,7 @@ impl, P: Pipe> EdgeSink { while !self.closed { match self.pipe.try_recv() { Ok(wire) => { + self.traffic += 1; if let Ok(Frame::Request(r)) = self.codec.decode(wire) { self.owed = Some(r.token); } @@ -249,6 +261,7 @@ impl, P: Pipe> EdgeSink { }; match self.pipe.try_send(self.codec.encode(&frame)) { Ok(()) => { + self.traffic += 1; self.owed = None; self.waiting = Delta { epoch: self.epoch, @@ -302,6 +315,7 @@ pub struct EdgeSource { /// The foreign incarnation a re-ask was already made for. noted: Option, closed: bool, + traffic: u64, } impl, P: Pipe> EdgeSource { @@ -321,6 +335,7 @@ impl, P: Pipe> EdgeSou ask_marks: 0, noted: None, closed: false, + traffic: 0, } } @@ -332,6 +347,11 @@ impl, P: Pipe> EdgeSou }) } + /// Frames read and requests sent so far: a host sees progress when it moves. + pub fn traffic(&self) -> u64 { + self.traffic + } + /// Whether a correction is wanted and has not arrived. pub fn stale(&self) -> bool { self.stale @@ -339,9 +359,10 @@ impl, P: Pipe> EdgeSou /// The same pipe and codec, after the consumer lost its state: it holds nothing again. pub fn restart(self, input: Input) -> Self { - let token = self.token; + let (token, traffic) = (self.token, self.traffic); EdgeSource { token, + traffic, ..Self::new(self.codec, self.pipe, input) } } @@ -369,6 +390,7 @@ impl, P: Pipe> EdgeSou match self.pipe.try_recv() { Ok(wire) => { read += 1; + self.traffic += 1; match self.codec.decode(wire) { Ok(Frame::Delta(d)) => self.on_delta(circuit, d), Ok(Frame::Correction(c)) => self.on_correction(circuit, c), @@ -453,6 +475,7 @@ impl, P: Pipe> EdgeSou match self.pipe.try_send(wire) { Ok(()) => { self.asked = true; + self.traffic += 1; self.token = token; self.ask_marks = self.marks; } diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs index 5062ccde..08854e36 100644 --- a/crates/zset/tests/edge.rs +++ b/crates/zset/tests/edge.rs @@ -617,6 +617,24 @@ fn an_answer_older_than_the_watermark_does_not_clear_stale() { ); } +#[test] +fn traffic_never_goes_down_across_a_restart_of_either_end() { + let (mut sink, mut src, mut c, _, _) = mirror(8); + sink.committed(z(&[(1, 1)])); + drive(&mut sink, &mut src, &mut c); + let (s, r) = (sink.traffic(), src.traffic()); + assert!(s > 0 && r > 0, "[setup] frames moved both ways"); + let sink = sink.restart(2, 0, Z::new()); + 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" + ); +} + #[test] fn a_new_epoch_frame_re_asks_once_after_a_lost_request() { let (a, mut b) = channel_pipes::>(64); @@ -873,3 +891,29 @@ fn a_circuit_on_each_of_two_threads_equals_one_circuit() { "[threads] the consumer after the producer thread finishes" ); } + +#[test] +fn a_poll_reads_at_most_the_frames_it_is_given() { + let (a, mut b) = channel_pipes::>(8); + let mut c = Circuit::new(); + let (i, _) = c.input::(); + let mut src = EdgeSource::new(Direct, a, i); + src.poll(&mut c, 0); + let Ok(Frame::Request(request)) = b.try_recv() else { + panic!("a request") + }; + for through in 0..3 { + let target = Z::new(); + let frame = Frame::Correction(Correction { + epoch: 1, + through, + token: request.token, + target, + }); + b.try_send(frame).unwrap(); + } + assert_eq!(src.poll(&mut c, 2), 2, "[budget] two of three"); + assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 1 })); + assert_eq!(src.poll(&mut c, 2), 1, "[budget] the rest"); + assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 2 })); +}