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();