From 0e93bc7bfa9885bedec4e19ab2625dcd232e3cdb Mon Sep 17 00:00:00 2001 From: Bryan O'Sullivan <23374+bos@users.noreply.github.com> Date: Sat, 24 Oct 2026 13:05:40 -0700 Subject: [PATCH] Subgraph lists shutdown candidates once each, rather than sorting and deduplicating `propagate_pointstamps` collected the node of every pushed change into `maybe_shutdown` and then sorted and deduplicated it. A per-child flag now lists each node once, as it is first seen. The candidates feed a binary heap, so their order does not matter. --- timely/src/progress/subgraph.rs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index d863132fb..a08cd8556 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -210,6 +210,7 @@ where activations, temp_active: BinaryHeap::new(), maybe_shutdown: Vec::new(), + maybe_shutdown_listed: vec![false; children.len()], children, input_messages: self.input_messages, output_capabilities: vec![MutableAntichain::new(); self.outputs], @@ -254,6 +255,8 @@ where activations: Rc>, temp_active: BinaryHeap>, maybe_shutdown: Vec, + // whether each child is in `maybe_shutdown`. + maybe_shutdown_listed: Vec, // shared state written to by the datapath, counting records entering this subgraph instance. input_messages: Vec>>>, @@ -476,7 +479,10 @@ where // Drain propagated information into shared progress structure. let (pushed, operators) = self.pointstamp_tracker.pushed(); for ((location, time), diff) in pushed.drain() { - self.maybe_shutdown.push(location.node); + if !self.maybe_shutdown_listed[location.node] { + self.maybe_shutdown_listed[location.node] = true; + self.maybe_shutdown.push(location.node); + } // Targets are actionable, sources are not. if let crate::progress::Port::Target(port) = location.port { // Activate based on expressed frontier interest for this input. @@ -497,9 +503,8 @@ where } // Consider scheduling each recipient of progress information to shut down. - self.maybe_shutdown.sort_unstable(); - self.maybe_shutdown.dedup(); for child_index in self.maybe_shutdown.drain(..) { + self.maybe_shutdown_listed[child_index] = false; let child_state = self.pointstamp_tracker.node_state(child_index); let frontiers_empty = child_state.targets.iter().all(|x| x.implications.is_empty()); let no_capabilities = child_state.cap_counts == 0;