Activate operators that may want to shut down
@@ -200,6 +200,7 @@ where
incomplete_count,
activations,
temp_active: BinaryHeap::new(),
+ maybe_shutdown: Vec::new(),
children: self.children,
input_messages: self.input_messages,
output_capabilities: self.output_capabilities,
@@ -242,6 +243,7 @@ where
// shared activations (including children).
activations: Rc<RefCell<Activations>>,
temp_active: BinaryHeap<Reverse<usize>>,
+ maybe_shutdown: Vec<usize>,
// shared state written to by the datapath, counting records entering this subgraph instance.
input_messages: Vec<Rc<RefCell<ChangeBatch<TInner>>>>,
@@ -461,6 +463,7 @@ where
// Drain propagated information into shared progress structure.
for ((location, time), diff) in self.pointstamp_tracker.pushed().drain() {
+ self.maybe_shutdown.push(location.node);
// Targets are actionable, sources are not.
if let crate::progress::Port::Target(port) = location.port {
if self.children[location.node].notify {
@@ -477,6 +480,18 @@ where
}
}
+ // Consider scheduling each recipient of progress information to shut down.
+ self.maybe_shutdown.sort();
+ self.maybe_shutdown.dedup();
+ for child_index in self.maybe_shutdown.drain(..) {
+ 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.sources.iter().all(|x| x.pointstamps.is_empty());
+ if frontiers_empty && no_capabilities {
+ self.temp_active.push(Reverse(child_index));
+ }
+ }
+
// Extract child zero frontier changes and report as internal capability changes.
for (output, internal) in self.shared_progress.borrow_mut().internals.iter_mut().enumerate() {
self.pointstamp_tracker
