From 9c10c7cf6b8c2f6d62658b3918eba7386704bbb8 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-threads)" Date: Sat, 10 Oct 2026 22:06:33 -0700 Subject: [PATCH] fix(incremental): a dead unit is dropped inside catch_unwind; pool drop joins deterministically; the leaf's more is pinned Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/improvise-incremental.md | 5 +- .../test-evidence/improvise-incremental.md | 29 +++++-- crates/improvise-incremental/src/sched.rs | 9 +- .../improvise-incremental/tests/sched_dag.rs | 13 ++- .../improvise-incremental/tests/sched_drop.rs | 87 +++++++++++++++++++ .../tests/sched_threads.rs | 70 ++++++++++++--- crates/zset/tests/edge.rs | 5 +- 7 files changed, 189 insertions(+), 29 deletions(-) create mode 100644 crates/improvise-incremental/tests/sched_drop.rs diff --git a/context/code/improvise-incremental.md b/context/code/improvise-incremental.md index 01e9baef..566da8da 100644 --- a/context/code/improvise-incremental.md +++ b/context/code/improvise-incremental.md @@ -569,7 +569,10 @@ the browser bundle: `Pool`, the only thing that spawns a thread, and the blockin 9. **Workers are counted and a dropped `Pool` stops them.** `Executor::workers()` counts live worker threads, so a worker that leaves on a spurious wake is visible (`every_worker_stays_through_wakes_with_nothing_queued`); `Drop for Pool` shuts down and joins - (`a_pool_that_is_dropped_stops_its_threads`), discarding what is queued. `claim_blocking` + (`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` diff --git a/context/test-evidence/improvise-incremental.md b/context/test-evidence/improvise-incremental.md index ed05866e..ab0cd8ed 100644 --- a/context/test-evidence/improvise-incremental.md +++ b/context/test-evidence/improvise-incremental.md @@ -1015,9 +1015,12 @@ Four workers, 200 kicks, the worker count stays four and is zero after `finish`. worker returns on a wake with an empty queue) fails it. Rivals W3 and W4 (a lost `notify_all` in `kick` or `finish`) fail it by the watchdog. -### a_pool_that_is_dropped_stops_its_threads +### a_pool_that_is_dropped_waits_for_its_workers -Dropping a pool leaves no worker. Rival W2 (an empty `Drop`) fails it. +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 @@ -1027,9 +1030,11 @@ turns. ### a_unit_whose_kicks_coalesced_still_drains_its_edge -Two commands cross before the projection's kicks are served, with a budget of 1. The projection -must ask to run again; the second frame is otherwise never read. Rival M5 fails it (and -`placed_circuits_equal_one_circuit_for_any_workers_and_order`). The same clause on the aggregate +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. @@ -1047,3 +1052,17 @@ it by the watchdog. ### a_dag_on_virtual_workers_in_any_order_equals_one_circuit The same graph under a random claim and hand-back order of 1 to 8 virtual workers. + +### a_unit_whose_drop_panics_does_not_hang_or_shrink_the_pool + +`bury` dropped the dead unit outside any catch_unwind, so a unit whose drop panics killed its +worker after the bookkeeping: on a one-worker pool the next queued unit never ran and +`Pool::finish` hung (a 5 s watchdog). The drop is now caught. The test checks the one worker is +still alive, the dead unit is reported and the other unit ran. Rival D1 fails it. + +### dropping_a_pool_while_unwinding_does_not_abort + +`Pool::drop` joined with `expect`, so a dead worker double-panicked while the host unwound and +aborted the process. Now a dead worker is a panic only when not already panicking. With the drop +caught no worker dies, so the test also passes without that guard (rival D3 does not fail it); +the guard is defence for a worker lost another way, and is not pinned. diff --git a/crates/improvise-incremental/src/sched.rs b/crates/improvise-incremental/src/sched.rs index bf12c27f..0fd1e4fd 100644 --- a/crates/improvise-incremental/src/sched.rs +++ b/crates/improvise-incremental/src/sched.rs @@ -256,8 +256,9 @@ impl Executor { st.running -= 1; st.panics.push(UnitPanic { unit: id, message }); drop(st); - drop(unit); 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. @@ -335,7 +336,11 @@ impl Pool { fn join(&mut self) { self.exec.shutdown(); for w in std::mem::take(&mut self.workers) { - w.join().expect("a worker panicked outside a turn"); + let died = w.join().is_err(); + assert!( + !died || std::thread::panicking(), + "a worker died outside a turn" + ); } } diff --git a/crates/improvise-incremental/tests/sched_dag.rs b/crates/improvise-incremental/tests/sched_dag.rs index 47648d45..bf15ed9a 100644 --- a/crates/improvise-incremental/tests/sched_dag.rs +++ b/crates/improvise-incremental/tests/sched_dag.rs @@ -7,8 +7,7 @@ use improvise_incremental::sched::{Claimed, Executor, Pool, Turn, Unit}; use std::sync::{Arc, Mutex}; use support::watchdog::within; use zset::{ - ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Flow, Frame, Output, Stream, ZSet, - channel_pipes, + ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Frame, Output, Stream, ZSet, channel_pipes, }; type P = ChannelPipe>; @@ -81,14 +80,13 @@ impl Unit for Head { s.committed(d.clone()); } } - let mut more = false; for s in &mut self.sinks { - more |= s.pump() == Flow::Backlogged; + s.pump(); } let after: u64 = self.sinks.iter().map(|s| s.traffic()).sum(); Turn { progress: after != before, - more, + more: false, } } fn neighbours(&self) -> &[usize] { @@ -116,7 +114,8 @@ impl Unit for Node { let mut more = false; for s in &mut self.sources { let r = s.poll(&mut self.c, budget); - more |= r == 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 { @@ -131,7 +130,7 @@ impl Unit for Node { } } for (s, _) in &mut self.sinks { - more |= s.pump() == Flow::Backlogged; + s.pump(); } Turn { progress: self.traffic() != before, diff --git a/crates/improvise-incremental/tests/sched_drop.rs b/crates/improvise-incremental/tests/sched_drop.rs new file mode 100644 index 00000000..6810903f --- /dev/null +++ b/crates/improvise-incremental/tests/sched_drop.rs @@ -0,0 +1,87 @@ +//! A unit that panics again when it is dropped must not cost the pool a worker. +//! Evidence: `context/test-evidence/improvise-incremental.md`. + +mod support; + +use improvise_incremental::sched::{Executor, Pool, Turn, Unit}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use support::watchdog::within; + +/// A unit that panics in its turn and again when it is dropped by the unwind's aftermath. +struct BoomTwice; + +impl Unit for BoomTwice { + fn turn(&mut self, _: usize) -> Turn { + panic!("boom in the turn"); + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + +impl Drop for BoomTwice { + fn drop(&mut self) { + if !std::thread::panicking() { + panic!("boom in the drop"); + } + } +} + +struct Count(Arc); + +impl Unit for Count { + fn turn(&mut self, _: usize) -> Turn { + self.0.fetch_add(1, Ordering::SeqCst); + Turn { + progress: false, + more: false, + } + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + +#[test] +fn a_unit_whose_drop_panics_does_not_hang_or_shrink_the_pool() { + let turns = Arc::new(AtomicUsize::new(0)); + let t = turns.clone(); + let (err, workers) = within(5, "a pool with a unit that panics in its drop", move || { + let units: Vec> = vec![Box::new(BoomTwice), Box::new(Count(t))]; + let exec = Arc::new(Executor::new(units, 1)); + let pool = Pool::start(exec.clone(), 1); + exec.kick(0); + exec.kick(1); + let idle = exec.wait_idle(); + let workers = exec.workers(); + let _ = pool.finish(); + (idle.err(), workers) + }); + assert_eq!( + workers, 1, + "[drop] the only worker survived the dead unit's drop" + ); + assert_eq!(err.map(|e| e.unit), Some(0)); + assert_eq!( + turns.load(Ordering::SeqCst), + 1, + "[drop] the other unit still ran" + ); +} + +#[test] +fn dropping_a_pool_while_unwinding_does_not_abort() { + within(5, "unwinding past a pool", || { + let exec = Arc::new(Executor::new(vec![Box::new(BoomTwice)], 1)); + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _pool = Pool::start(exec.clone(), 1); + exec.kick(0); + let _ = exec.wait_idle(); + panic!("the host fails with a pool alive"); + })); + assert!(r.is_err()); + }); +} diff --git a/crates/improvise-incremental/tests/sched_threads.rs b/crates/improvise-incremental/tests/sched_threads.rs index 2ca25165..8fa5d76c 100644 --- a/crates/improvise-incremental/tests/sched_threads.rs +++ b/crates/improvise-incremental/tests/sched_threads.rs @@ -135,15 +135,51 @@ fn every_worker_stays_through_wakes_with_nothing_queued() { }); } +/// A unit whose turn is slow, so a drop that does not join returns before it ends. +struct Slow { + started: Arc, + finished: Arc, +} + +impl Unit for Slow { + fn turn(&mut self, _: usize) -> Turn { + self.started.fetch_add(1, Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_millis(300)); + self.finished.fetch_add(1, Ordering::SeqCst); + Turn { + progress: false, + more: false, + } + } + + fn neighbours(&self) -> &[usize] { + &[] + } +} + #[test] -fn a_pool_that_is_dropped_stops_its_threads() { +fn a_pool_that_is_dropped_waits_for_its_workers() { within(10, "dropping a pool", || { - let exec = Arc::new(Executor::new(vec![Box::new(Count(Arc::default()))], 1)); + 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!(exec.workers(), 0, "[drop] no thread is left parked"); + assert_eq!( + finished.load(Ordering::SeqCst), + 1, + "[drop] the turn in flight was joined" + ); + assert_eq!(exec.workers(), 0, "[drop] no thread is left"); }); } @@ -192,16 +228,24 @@ fn a_unit_whose_kicks_coalesced_still_drains_its_edge() { let t = u.run(exec.budget()); exec.finish(u, t); } - // Two commands cross before the aggregate runs: its two kicks are one queue entry. - staged.submit(cmd(0, 0, 1)); - let mut head = exec.claim(|_| 0).unwrap(); - let t = head.run(1); - exec.finish(head, t); - staged.submit(cmd(1, 1, 2)); - let mut head = exec.claim(|n| n - 1).unwrap(); - assert_eq!(head.id(), 0); - let t = head.run(1); - exec.finish(head, t); + // 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); diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs index 608624e3..08854e36 100644 --- a/crates/zset/tests/edge.rs +++ b/crates/zset/tests/edge.rs @@ -899,12 +899,15 @@ fn a_poll_reads_at_most_the_frames_it_is_given() { let (i, _) = c.input::(); let mut src = EdgeSource::new(Direct, a, i); src.poll(&mut c, 0); + let Ok(Frame::Request(request)) = b.try_recv() else { + panic!("a request") + }; for through in 0..3 { let target = Z::new(); let frame = Frame::Correction(Correction { epoch: 1, through, - token: 1, + token: request.token, target, }); b.try_send(frame).unwrap();