From 4ba0cf196e5a803d439c843023db5793087e3087 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-threads)" Date: Sat, 10 Oct 2026 21:00:10 -0700 Subject: [PATCH] feat(incremental): a scheduler runs sub-circuits on a worker pool; the aggregate and grid stages placed on two units Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/improvise-incremental.md | 48 ++++ context/code/zset.md | 2 + .../test-evidence/improvise-incremental.md | 28 ++ context/test-evidence/zset.md | 5 + crates/improvise-incremental/src/lib.rs | 2 + crates/improvise-incremental/src/sched.rs | 234 ++++++++++++++++ crates/improvise-incremental/src/staged.rs | 216 +++++++++++++++ .../tests/sched.proptest-regressions | 9 + crates/improvise-incremental/tests/sched.rs | 257 ++++++++++++++++++ crates/zset/src/edge.rs | 22 +- crates/zset/tests/edge.rs | 21 ++ 11 files changed, 843 insertions(+), 1 deletion(-) create mode 100644 crates/improvise-incremental/src/sched.rs create mode 100644 crates/improvise-incremental/src/staged.rs create mode 100644 crates/improvise-incremental/tests/sched.proptest-regressions create mode 100644 crates/improvise-incremental/tests/sched.rs diff --git a/context/code/improvise-incremental.md b/context/code/improvise-incremental.md index 986b9ee4..81dce313 100644 --- a/context/code/improvise-incremental.md +++ b/context/code/improvise-incremental.md @@ -504,3 +504,51 @@ 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`), and `Pool`, the only thing that spawns a thread, is +`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 (`Flow::Backlogged`) is also `more`: the unit retries, and coalesces + meanwhile. 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. **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). diff --git a/context/code/zset.md b/context/code/zset.md index 246be461..a050ac5d 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -363,3 +363,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..33aa3e1b 100644 --- a/context/test-evidence/improvise-incremental.md +++ b/context/test-evidence/improvise-incremental.md @@ -957,3 +957,31 @@ 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. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index 29166d5d..2fae93af 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -689,3 +689,8 @@ with "did not panic". Three hundred commands on a producer thread, a consumer on another, capacity 2, the producer 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. 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..0c9ead12 --- /dev/null +++ b/crates/improvise-incremental/src/sched.rs @@ -0,0 +1,234 @@ +//! 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::{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, or an edge was full: run it again. + 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, +} + +struct State { + slots: Vec, + queue: VecDeque, + running: usize, + shutdown: bool, +} + +/// 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) + } +} + +/// 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`. +pub struct Executor { + state: Mutex, + wake: Condvar, + budget: usize, +} + +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 { + Executor { + state: Mutex::new(State { + slots: units + .into_iter() + .map(|u| Slot { + unit: Some(u), + queued: false, + running: false, + again: false, + }) + .collect(), + queue: VecDeque::new(), + running: 0, + shutdown: false, + }), + wake: Condvar::new(), + budget: budget.max(1), + } + } + + pub fn budget(&self) -> usize { + self.budget + } + + fn lock(&self) -> MutexGuard<'_, State> { + self.state.lock().expect("a unit panicked in a turn") + } + + fn kick_in(st: &mut State, id: usize) { + let slot = &mut st.slots[id]; + if slot.running { + slot.again = true; + } else if !slot.queued { + slot.queued = true; + st.queue.push_back(id); + } + } + + /// Marks `id` runnable: it has input, or the host wants it to look. + pub fn kick(&self, id: usize) { + 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` after `shutdown`. + 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"); + } + } + + /// 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. + pub fn finish(&self, claimed: Claimed, turn: Turn) { + let Claimed { id, unit } = claimed; + let mut st = self.lock(); + let neighbours = unit.neighbours().to_vec(); + 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(); + } + + /// 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) { + 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"); + } + } + + /// Tells the workers to leave once the queue is empty. + 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>, +} + +#[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); + std::thread::spawn(move || { + while let Some(mut claimed) = exec.claim_blocking() { + let turn = claimed.run(exec.budget()); + exec.finish(claimed, turn); + } + }) + }) + .collect(); + Pool { exec, workers } + } + + pub fn executor(&self) -> &Executor { + &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(); + self.exec.shutdown(); + for w in self.workers { + w.join().expect("a worker panicked"); + } + self.exec + } +} diff --git a/crates/improvise-incremental/src/staged.rs b/crates/improvise-incremental/src/staged.rs new file mode 100644 index 00000000..1b18f2f2 --- /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, Flow, 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); + } + let flow = self.sink.pump(); + Turn { + progress: self.sink.traffic() != before, + more: flow == Flow::Backlogged, + } + } + + 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); + } + let flow = self.sink.pump(); + Turn { + progress: self.source.traffic() + self.sink.traffic() != before, + more: read == budget || flow == Flow::Backlogged, + } + } + + 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.proptest-regressions b/crates/improvise-incremental/tests/sched.proptest-regressions new file mode 100644 index 00000000..01cef711 --- /dev/null +++ b/crates/improvise-incremental/tests/sched.proptest-regressions @@ -0,0 +1,9 @@ +# Seeds for failure cases proptest has generated in the past. It is +# automatically read and these particular cases re-run before any +# novel cases are generated. +# +# It is recommended to check this file in to source control so that +# everyone who runs the test benefits from these saved cases. +cc 67b92f77a748cf7ce3b29898ed519a2926626971d2e38b295b3cd6616ea26b43 # shrinks to commands = [ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }], workers = 3, cap = 1, budget = 1, choices = [148, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], gaps = [0] +cc caa4c353b78a43303e8a95654896b57d676c5d95f3198ef4eb4b3d0c7531fec9 # shrinks to commands = [ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {(0, 0, 0): -1} }, ZSet { weights: {} }], workers = 2, cap = 1, budget = 1, choices = [92, 71, 34, 29, 122, 72, 14, 0, 2, 6, 73, 132, 236, 26, 2, 20, 10], gaps = [2, 2, 4, 1, 5, 3] +cc 4d5dba8a92cdebce288f317349a8b961e23dc0b627d27ddbeb78f728a6868b16 # shrinks to commands = [ZSet { weights: {} }, ZSet { weights: {} }], workers = 1, cap = 1, budget = 1, choices = [130, 0, 10, 0, 132, 62], gaps = [3, 3] diff --git a/crates/improvise-incremental/tests/sched.rs b/crates/improvise-incremental/tests/sched.rs new file mode 100644 index 00000000..d4dee3dd --- /dev/null +++ b/crates/improvise-incremental/tests/sched.rs @@ -0,0 +1,257 @@ +//! 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`. + +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 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() { + 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(); + 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() { + 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"); +} diff --git a/crates/zset/src/edge.rs b/crates/zset/src/edge.rs index 2b1415d8..c65a5ff1 100644 --- a/crates/zset/src/edge.rs +++ b/crates/zset/src/edge.rs @@ -144,6 +144,7 @@ pub struct EdgeSink { waiting: Delta, owes_correction: bool, closed: bool, + traffic: u64, } impl, P: Pipe> EdgeSink { @@ -169,6 +170,7 @@ impl, P: Pipe> EdgeSink { }, owes_correction: false, closed: false, + traffic: 0, } } @@ -179,6 +181,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.owes_correction @@ -217,6 +224,7 @@ impl, P: Pipe> EdgeSink { while !self.closed { match self.pipe.try_recv() { Ok(wire) => { + self.traffic += 1; if let Ok(Frame::Request(_)) = self.codec.decode(wire) { self.owes_correction = true; } @@ -241,6 +249,7 @@ impl, P: Pipe> EdgeSink { }; match self.pipe.try_send(self.codec.encode(&frame)) { Ok(()) => { + self.traffic += 1; self.owes_correction = false; self.waiting = Delta { epoch: self.epoch, @@ -275,6 +284,7 @@ pub struct EdgeSource { stale: bool, asked: bool, closed: bool, + traffic: u64, } impl, P: Pipe> EdgeSource { @@ -291,6 +301,7 @@ impl, P: Pipe> EdgeSou stale: true, asked: false, closed: false, + traffic: 0, } } @@ -302,6 +313,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 @@ -326,6 +342,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), @@ -396,7 +413,10 @@ impl, P: Pipe> EdgeSou .codec .encode(&Frame::Request(Request { have: self.stamp() })); match self.pipe.try_send(wire) { - Ok(()) => self.asked = true, + Ok(()) => { + self.asked = true; + self.traffic += 1; + } Err(Refused::Full(_)) => {} Err(Refused::Closed) => self.closed = true, } diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs index c177ef0c..5a3bafb9 100644 --- a/crates/zset/tests/edge.rs +++ b/crates/zset/tests/edge.rs @@ -702,3 +702,24 @@ 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); + for through in 0..3 { + let target = Z::new(); + let frame = Frame::Correction(Correction { + epoch: 1, + through, + 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 })); +}