fiddlerwoaroof/improvise
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #364 from fiddlerwoaroof/dbsp/sched

Browse Source 
feat(incremental): scheduler runs sub-circuits on a worker pool
This commit is contained in:
Ed Langley
2026-10-10 22:20:34 -07:00
committed by GitHub
parent 3d1dcfd5e4 9c10c7cf6b
commit d63b2c1956
15 changed files with 1862 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
context/code/improvise-incremental.md
+74
View File
@@ -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.
context/code/zset.md
+2
View File
@@ -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").
context/test-evidence/improvise-incremental.md
+109
View File
@@ -957,3 +957,112 @@ differs).
`Pushdown::new` took a generic `Arc<Q>` 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.
context/test-evidence/zset.md
+11
View File
@@ -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;
crates/improvise-incremental/src/lib.rs
+2
View File
@@ -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;
crates/improvise-incremental/src/sched.rs
+362
View File
@@ -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<Box<dyn Unit>>,
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<Slot>,
queue: VecDeque<usize>,
running: usize,
shutdown: bool,
panics: Vec<UnitPanic>,
}
/// A unit a worker owns until it hands it back with `finish`.
pub struct Claimed {
id: usize,
unit: Box<dyn Unit>,
}
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::<String>()
.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<State>,
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<Box<dyn Unit>>, 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<UnitPanic> {
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<Claimed> {
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<Claimed> {
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<dyn Unit>, 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<Executor>,
workers: Vec<std::thread::JoinHandle<()>>,
}
/// Counts a worker as gone however its thread ends.
#[cfg(not(target_arch = "wasm32"))]
struct Alive(std::sync::Arc<Executor>);
#[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<Executor>, 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<std::sync::Arc<Executor>, 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();
}
}
crates/improvise-incremental/src/staged.rs
+216
View File
@@ -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<K> = ChannelPipe<Frame<K>>;
/// 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<ZSet<Cell>>) -> Stream<ZSet<Group>> {
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<ZSet<Group>>) -> 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<Cell>]) -> Grid {
let mut c = Circuit::new();
let (input, cells) = c.input::<Cell>();
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<Cell, Direct, P<Cell>>,
inbox: Arc<Mutex<Vec<ZSet<Cell>>>>,
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<Group>,
source: EdgeSource<Cell, Direct, P<Cell>>,
sink: EdgeSink<Group, Direct, P<Group>>,
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<Group, Direct, P<Group>>,
grid: Arc<Mutex<Grid>>,
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<Executor>,
inbox: Arc<Mutex<Vec<ZSet<Cell>>>>,
grid: Arc<Mutex<Grid>>,
}
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::<Frame<Cell>>(cap);
let (agg_out, grid_in) = channel_pipes::<Frame<Group>>(cap);
let mut c = Circuit::new();
let (input, cells): (Input<Cell>, _) = c.input::<Cell>();
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::<Group>();
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<Box<dyn Unit>> =
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<Executor> {
&self.exec
}
/// One committed command from the host.
pub fn submit(&self, delta: ZSet<Cell>) {
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()
}
}
crates/improvise-incremental/tests/sched.rs
+264
View File
@@ -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<Value = ZSet<Cell>> {
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<Item = u8>,
) -> 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::<u8>(), 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<Item = u8>) {
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<ZSet<Cell>> = (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<std::sync::Mutex<Vec<usize>>>,
}
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<dyn Unit> {
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<AtomicBool>,
turns: Arc<AtomicUsize>,
}
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");
});
}
crates/improvise-incremental/tests/sched_dag.rs
+383
View File
@@ -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<Frame<i64>>;
type Z = ZSet<i64>;
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<Z>) -> Stream<Z> {
c.map(s, |x| x * 2)
}
fn st_b(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
c.filter(s, |x| x % 3 != 0)
}
fn st_c(c: &mut Circuit, a: &Stream<Z>, b: &Stream<Z>) -> Stream<Z> {
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::<i64>()));
c.map(&g, |(k, s)| k * 1_000_000 + s)
}
fn st_d(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
let m = c.map(s, |x| x.rem_euclid(1000));
c.distinct(&m)
}
fn st_e(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
c.filter(s, |x| x % 2 == 0)
}
fn st_f(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
c.map(s, |x| x / 2)
}
fn reference(cmds: &[Z]) -> (Z, Z) {
let mut c = Circuit::new();
let (i, s) = c.input::<i64>();
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<EdgeSink<i64, Direct, P>>,
inbox: Arc<Mutex<Vec<Z>>>,
around: Vec<usize>,
}
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<EdgeSource<i64, Direct, P>>,
sinks: Vec<(EdgeSink<i64, Direct, P>, Output<i64>)>,
collect: Option<(Output<i64>, Arc<Mutex<Z>>)>,
around: Vec<usize>,
}
impl Node {
fn traffic(&self) -> u64 {
self.sources.iter().map(|s| s.traffic()).sum::<u64>()
+ self.sinks.iter().map(|(s, _)| s.traffic()).sum::<u64>()
}
}
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<Executor>,
inbox: Arc<Mutex<Vec<Z>>>,
d: Arc<Mutex<Z>>,
f: Arc<Mutex<Z>>,
}
fn dag(rng: &mut Rng, budget: usize) -> Dag {
let mut cap = || 1 + rng.below(3) as usize;
let (h_a, a_in) = channel_pipes::<Frame<i64>>(cap());
let (h_b, b_in) = channel_pipes::<Frame<i64>>(cap());
let (a_c, c_in_a) = channel_pipes::<Frame<i64>>(cap());
let (b_c, c_in_b) = channel_pipes::<Frame<i64>>(cap());
let (c_d, d_in) = channel_pipes::<Frame<i64>>(cap());
let (c_e, e_in) = channel_pipes::<Frame<i64>>(cap());
let (e_f, f_in) = channel_pipes::<Frame<i64>>(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<Z>) -> Stream<Z>,
out: Option<P>,
col: Option<Arc<Mutex<Z>>>,
around: Vec<usize>| {
let mut c = Circuit::new();
let (i, s) = c.input::<i64>();
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::<i64>();
let (ib, sb) = c.input::<i64>();
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<Box<dyn Unit>> = 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<Z> {
(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<Vec<Z>> = (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}");
}
}
crates/improvise-incremental/tests/sched_drop.rs
+87
View File
@@ -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<AtomicUsize>);
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<Box<dyn Unit>> = 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());
});
}
crates/improvise-incremental/tests/sched_threads.rs
+259
View File
@@ -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<AtomicUsize>);
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<Box<dyn Unit>> = 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<Box<dyn Unit>> =
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<AtomicUsize>,
finished: Arc<AtomicUsize>,
}
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<Cell> {
[((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"
);
}
crates/improvise-incremental/tests/support/mod.rs
+1
View File
@@ -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;
crates/improvise-incremental/tests/support/watchdog.rs
+23
View File
@@ -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<T: Send + 'static>(
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"),
}
}
crates/zset/src/edge.rs
+25 -2
View File
@@ -149,6 +149,7 @@ pub struct EdgeSink<K, C, P> {
/// The token of the latest request read and not yet answered.
owed: Option<u64>,
closed: bool,
traffic: u64,
}
impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
@@ -174,6 +175,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
},
owed: None,
closed: false,
traffic: 0,
}
}
@@ -184,6 +186,11 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
}
}
/// 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<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
/// 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<K>) -> 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<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
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<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
};
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<K, C, P> {
/// The foreign incarnation a re-ask was already made for.
noted: Option<u64>,
closed: bool,
traffic: u64,
}
impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSource<K, C, P> {
@@ -321,6 +335,7 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
ask_marks: 0,
noted: None,
closed: false,
traffic: 0,
}
}
@@ -332,6 +347,11 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> 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<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
/// The same pipe and codec, after the consumer lost its state: it holds nothing again.
pub fn restart(self, input: Input<K>) -> 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<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> 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<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
match self.pipe.try_send(wire) {
Ok(()) => {
self.asked = true;
self.traffic += 1;
self.token = token;
self.ask_marks = self.marks;
}
crates/zset/tests/edge.rs
+44
View File
@@ -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::<u8>();
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::<Frame<u8>>(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::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, _) = c.input::<u8>();
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 }));
}