From de7bb9a9edcc35cf5832d12c50c14d5299fb9818 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 30 Oct 2026 18:38:55 -0400 Subject: [PATCH 1/4] Subgraph: schedule inbound recipients in the pass that harvests their records `harvest_inputs` records +1 at each inner target for records entering the scope, but did not queue those targets, unlike `extract_progress`, which queues the recipients of records produced inside the scope. The first inner operator therefore ran a pass later, and the scope's progress batch for one record split in two: the receipt with +1 at the inner target, then the -1 at that target with whatever followed. Queue the targets beside the +1. This changes no counts and defers or splits no batch; it only makes the recipient eligible in the current pass. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o --- timely/src/progress/subgraph.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index d8bdf243d..98577cfa3 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -417,6 +417,9 @@ where for (time, delta) in borrowed.drain() { for target in &self.children[0].edges[input] { self.local_pointstamp.update((Location::from(*target), time.clone()), delta); + // Schedule the recipient in this pass, as `extract_progress` does for messages + // produced inside the scope, so that it consumes the records it was just sent. + self.temp_active.push(Reverse(target.node)); } self.local_pointstamp.update((source, time), -delta); } From ae8d48aa9c1bb7123a6abb6d4821832e19d8ade6 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 30 Oct 2026 19:52:03 -0400 Subject: [PATCH 2/4] Subgraph: let the parent schedule recipients of records leaving a scope in the same pass A scope reports records leaving it as produced only once its own progress batch has been exchanged, and only then does the parent queue the downstream operator, so the record waited for the parent's next pass and the frontier work done before it. Add `pushed: Vec` to `SharedProgress`: after its children run, a subgraph marks each output with a positive change at a scope-output target in its pending batch, and the parent's `extract_progress` queues the recipients of marked outputs. No counts change. A downstream operator may now consume a record before the scope's exchanged report adds the matching +1; until then the frontier is held by the +1 at the scope's input, which the same worker sends in the same or an earlier batch, as with records on an exchange channel. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o --- timely/src/progress/operate.rs | 6 ++++++ timely/src/progress/subgraph.rs | 18 ++++++++++++++++++ 2 files changed, 24 insertions(+) diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index dcd377d7e..28f214c25 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -191,6 +191,11 @@ pub struct SharedProgress { pub internals: Vec>, /// Produced message changes reported by the child operator. pub produceds: Vec>, + /// Outputs on which the child sent records this activation, before reporting them as produced. + /// + /// A scope reports records leaving it as produced only once its own progress updates have been + /// exchanged. This lets its parent schedule the recipients sooner, which changes no counts. + pub pushed: Vec, } impl SharedProgress { @@ -201,6 +206,7 @@ impl SharedProgress { consumeds: vec![ChangeBatch::new(); inputs], internals: vec![ChangeBatch::new(); outputs], produceds: vec![ChangeBatch::new(); outputs], + pushed: vec![false; outputs], } } } diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 98577cfa3..83ed642a2 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -326,6 +326,17 @@ where } } + // Note scope outputs that were sent records in this pass, so that the parent can schedule + // their recipients now rather than after these counts are exchanged and reported upward. + { + let mut shared = self.shared_progress.borrow_mut(); + for ((location, _time), diff) in self.local_pointstamp.iter() { + if let (0, Port::Target(port), true) = (location.node, location.port, *diff > 0) { + shared.pushed[port] = true; + } + } + } + // Transmit produced progress updates. self.send_progress(); @@ -784,6 +795,13 @@ impl PerOperatorState { } } } + for (output, pushed) in shared_progress.pushed.iter_mut().enumerate() { + if std::mem::take(pushed) { + for target in self.edges[output].iter() { + temp_active.push(Reverse(target.node)); + } + } + } } /// Test the validity of `self.shared_progress`. From 0241ce6bfafd3af629b07a3206eaa84dc5b7d27c Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 31 Oct 2026 14:45:35 -0400 Subject: [PATCH 3/4] Subgraph: pass output marks up through scopes that wrap scopes A scope marks the outputs it sent records to, so its parent can schedule their recipients in the same pass. When a scope's output is wired straight to its parent's output (node zero), the parent queued node zero, which its scheduling loop skips, and its own scan for marks did not see the record, whose count reaches it only after the inner scope's progress exchange. So a record leaving two or more snug scopes still arrived a root pass late. A marked output wired to node zero now marks the enclosing scope's own output, during the enclosing scope's schedule, so the mark reaches the parent in the same pass at any depth. No counts change. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o --- timely/src/progress/subgraph.rs | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 83ed642a2..6326f548d 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -391,11 +391,12 @@ where } // Extract progress statements into either pre- or post-exchange buffers. + let outer_pushed = &mut self.shared_progress.borrow_mut().pushed[..]; if child.local { - child.extract_progress(&mut self.local_pointstamp, &mut self.temp_active); + child.extract_progress(&mut self.local_pointstamp, &mut self.temp_active, outer_pushed); } else { - child.extract_progress(&mut self.final_pointstamp, &mut self.temp_active); + child.extract_progress(&mut self.final_pointstamp, &mut self.temp_active, outer_pushed); } incomplete @@ -604,7 +605,7 @@ where // We introduce these into the progress tracker to determine the scope's initial // internal capabilities. for child in self.children.iter_mut() { - child.extract_progress(&mut self.final_pointstamp, &mut self.temp_active); + child.extract_progress(&mut self.final_pointstamp, &mut self.temp_active, &mut []); } self.propagate_pointstamps(); // Propagate expressed capabilities to output frontiers. @@ -766,7 +767,11 @@ impl PerOperatorState { } /// Extracts shared progress information and converts to pointstamp changes. - fn extract_progress(&self, pointstamps: &mut ChangeBatch<(Location, T)>, temp_active: &mut BinaryHeap>) { + /// + /// `outer_pushed` holds the enclosing scope's own output marks: a marked output wired to the + /// scope's outputs (node zero) marks them in turn, so that records leaving nested scopes + /// reach their recipients in the same pass however many scope boundaries they cross. + fn extract_progress(&self, pointstamps: &mut ChangeBatch<(Location, T)>, temp_active: &mut BinaryHeap>, outer_pushed: &mut [bool]) { let shared_progress = &mut *self.shared_progress.borrow_mut(); @@ -798,7 +803,8 @@ impl PerOperatorState { for (output, pushed) in shared_progress.pushed.iter_mut().enumerate() { if std::mem::take(pushed) { for target in self.edges[output].iter() { - temp_active.push(Reverse(target.node)); + if target.node == 0 { outer_pushed[target.port] = true; } + else { temp_active.push(Reverse(target.node)); } } } } From c5819a8f6068b0386a9484da53eb4f692a5d9575 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 31 Oct 2026 14:51:05 -0400 Subject: [PATCH 4/4] Subgraph: queue inbound recipients once per input; say only scopes set `pushed` `harvest_inputs` queued each recipient once per (time, target) pair; the scheduling loop removes duplicates, so this was only heap churn. Queue each recipient once per input that received records. The `pushed` documentation now says that only scopes set it. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o --- timely/src/progress/operate.rs | 7 ++++--- timely/src/progress/subgraph.rs | 9 ++++++--- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 28f214c25..088bd8940 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -191,10 +191,11 @@ pub struct SharedProgress { pub internals: Vec>, /// Produced message changes reported by the child operator. pub produceds: Vec>, - /// Outputs on which the child sent records this activation, before reporting them as produced. + /// Outputs on which a scope sent records this activation, before reporting them as produced. /// - /// A scope reports records leaving it as produced only once its own progress updates have been - /// exchanged. This lets its parent schedule the recipients sooner, which changes no counts. + /// Only scopes set this: a scope reports records leaving it as produced only once its own + /// progress updates have been exchanged. This lets its parent schedule the recipients sooner, + /// which changes no counts. Ordinary operators report produced records directly. pub pushed: Vec, } diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 6326f548d..6ed89c9a9 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -426,15 +426,18 @@ where for input in 0 .. self.inputs { let source = Location::new_source(0, input); let mut borrowed = self.input_messages[input].borrow_mut(); + if borrowed.is_empty() { continue; } for (time, delta) in borrowed.drain() { for target in &self.children[0].edges[input] { self.local_pointstamp.update((Location::from(*target), time.clone()), delta); - // Schedule the recipient in this pass, as `extract_progress` does for messages - // produced inside the scope, so that it consumes the records it was just sent. - self.temp_active.push(Reverse(target.node)); } self.local_pointstamp.update((source, time), -delta); } + // Schedule the recipients in this pass, as `extract_progress` does for messages + // produced inside the scope, so that they consume the records they were just sent. + for target in &self.children[0].edges[input] { + self.temp_active.push(Reverse(target.node)); + } } }