From 0605d7825c6c54b6151f0b23d72a566eec97f88b Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 25 Sep 2026 17:59:49 +0200 Subject: [PATCH 1/5] communication: add broadcast_peers and Pull::quiet `Allocate::broadcast_peers` allocates a broadcast channel that delivers to every worker except the sender, while the sender's puller still receives what other workers push. The default implementation drops the sender's own pusher from `allocate`, and the zero-copy allocators share their existing broadcast paths, so remote processes and local peers still receive one serialization. `Pull::quiet` stops a puller from recording an event for its channel once it has been drained. By default, a drained channel that yielded messages schedules its recipient once more, which a recipient that finishes its work in the draining call does not need. The default implementation does nothing, and the counting and zero-copy pullers forward it to the pullers they wrap. Co-Authored-By: Claude Opus 5.5 --- communication/src/allocator/counters.rs | 15 +++- communication/src/allocator/generic.rs | 11 +++ communication/src/allocator/mod.rs | 18 ++++ .../src/allocator/zero_copy/allocator.rs | 87 +++++++++++-------- .../allocator/zero_copy/allocator_process.rs | 64 ++++++++------ .../src/allocator/zero_copy/push_pull.rs | 1 + communication/src/lib.rs | 7 ++ 7 files changed, 136 insertions(+), 67 deletions(-) diff --git a/communication/src/allocator/counters.rs b/communication/src/allocator/counters.rs index 809829807..afb105695 100644 --- a/communication/src/allocator/counters.rs +++ b/communication/src/allocator/counters.rs @@ -118,6 +118,8 @@ pub struct Puller> { events: Rc>>, puller: P, phantom: ::std::marker::PhantomData, + /// Whether to record an event once the channel has been drained. + echo: bool, } impl> Puller { @@ -129,6 +131,7 @@ impl> Puller { events, puller, phantom: ::std::marker::PhantomData, + echo: true, } } } @@ -138,9 +141,11 @@ impl> Pull for Puller { let result = self.puller.pull(); if result.is_none() { if self.count != 0 { - self.events - .borrow_mut() - .push(self.index); + if self.echo { + self.events + .borrow_mut() + .push(self.index); + } self.count = 0; } } @@ -150,4 +155,8 @@ impl> Pull for Puller { result } + fn quiet(&mut self) { + self.echo = false; + self.puller.quiet(); + } } diff --git a/communication/src/allocator/generic.rs b/communication/src/allocator/generic.rs index a3f68b12e..03506499e 100644 --- a/communication/src/allocator/generic.rs +++ b/communication/src/allocator/generic.rs @@ -56,6 +56,14 @@ impl Allocator { Allocator::Tcp(z) => z.broadcast(identifier), } } + /// Constructs a broadcast channel that does not deliver to the sender. + pub fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + match self { + Allocator::Thread(t) => t.broadcast_peers(identifier), + Allocator::Process(p) => p.broadcast_peers(identifier), + Allocator::Tcp(z) => z.broadcast_peers(identifier), + } + } /// Perform work before scheduling operators. pub fn receive(&mut self) { match self { @@ -111,6 +119,9 @@ impl Allocate for Allocator { fn broadcast(&mut self, identifier: usize) -> (Box>, Box>) { self.broadcast(identifier) } + fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_peers(identifier) + } fn receive(&mut self) { self.receive(); } fn release(&mut self) { self.release(); } fn events(&self) -> &Rc>> { self.events() } diff --git a/communication/src/allocator/mod.rs b/communication/src/allocator/mod.rs index 8c7cf6893..5c47b0ee9 100644 --- a/communication/src/allocator/mod.rs +++ b/communication/src/allocator/mod.rs @@ -92,6 +92,16 @@ pub(crate) trait Allocate { let (pushers, pull) = self.allocate(identifier); (Box::new(Broadcaster { spare: None, pushers }), pull) } + + /// Allocates a broadcast channel, where each pushed message is received by all workers except the sender. + /// + /// The puller still receives the messages that other workers push. + fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + let index = self.index(); + let (mut pushers, pull) = self.allocate(identifier); + pushers.remove(index); + (Box::new(Broadcaster { spare: None, pushers }), pull) + } } /// An adapter to broadcast any pushed element. @@ -208,6 +218,14 @@ impl Process { Process::Bytes(pb) => pb.broadcast(identifier), } } + pub(crate) fn broadcast_peers(&mut self, identifier: usize) + -> (Box>, Box>) + { + match self { + Process::Typed(p) => p.broadcast_peers(identifier), + Process::Bytes(pb) => pb.broadcast_peers(identifier), + } + } pub(crate) fn receive(&mut self) { match self { Process::Typed(p) => p.receive(), diff --git a/communication/src/allocator/zero_copy/allocator.rs b/communication/src/allocator/zero_copy/allocator.rs index 59ce805a1..e3076fea7 100644 --- a/communication/src/allocator/zero_copy/allocator.rs +++ b/communication/src/allocator/zero_copy/allocator.rs @@ -149,6 +149,51 @@ pub struct TcpAllocator { to_local: HashMap>>>, // to worker-local typed pullers. } +impl TcpAllocator { + /// Allocates a broadcast channel that delivers to this worker only if `include_self` is set. + fn broadcast_to(&mut self, identifier: usize, include_self: bool) -> (Box>, Box>) { + + // Assume and enforce in-order identifier allocation. + if let Some(bound) = self.channel_id_bound { + assert!(bound < identifier); + } + self.channel_id_bound = Some(identifier); + + // Result list of boxed pushers. + // One entry for each process. + let mut pushes = Vec::>>::with_capacity(self.sends.len() + 1); + + // Inner exchange allocations. + let inner_peers = self.inner.peers(); + let (inner_send, inner_recv) = if include_self { self.inner.broadcast(identifier) } else { self.inner.broadcast_peers(identifier) }; + + pushes.push(inner_send); + for (mut index, send) in self.sends.iter().enumerate() { + // The span of worker indexes jumps by `inner_peers` as we skip our own process. + // We bump `index` by one as we pass `self.index/inner_peers` to effect this. + if index >= self.index/inner_peers { index += 1; } + let header = MessageHeader { + channel: identifier, + source: self.index, + target_lower: index * inner_peers, + target_upper: index * inner_peers + inner_peers, + length: 0, + seqno: 0, + }; + pushes.push(Box::new(Pusher::new(header, Rc::clone(send)))) + } + + let channel = Rc::clone(self.to_local.entry(identifier).or_default()); + + use crate::allocator::counters::Puller as CountPuller; + let canary = Canary::new(identifier, Rc::clone(&self.canaries)); + let puller = Box::new(CountPuller::new(PullerInner::new(inner_recv, channel, canary), identifier, Rc::clone(self.events()))); + + let pushes = Box::new(crate::allocator::Broadcaster { spare: None, pushers: pushes }); + (pushes, puller, ) + } +} + impl Allocate for TcpAllocator { fn index(&self) -> usize { self.index } fn peers(&self) -> usize { self.peers } @@ -202,45 +247,11 @@ impl Allocate for TcpAllocator { } fn broadcast(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_to(identifier, true) + } - // Assume and enforce in-order identifier allocation. - if let Some(bound) = self.channel_id_bound { - assert!(bound < identifier); - } - self.channel_id_bound = Some(identifier); - - // Result list of boxed pushers. - // One entry for each process. - let mut pushes = Vec::>>::with_capacity(self.sends.len() + 1); - - // Inner exchange allocations. - let inner_peers = self.inner.peers(); - let (inner_send, inner_recv) = self.inner.broadcast(identifier); - - pushes.push(inner_send); - for (mut index, send) in self.sends.iter().enumerate() { - // The span of worker indexes jumps by `inner_peers` as we skip our own process. - // We bump `index` by one as we pass `self.index/inner_peers` to effect this. - if index >= self.index/inner_peers { index += 1; } - let header = MessageHeader { - channel: identifier, - source: self.index, - target_lower: index * inner_peers, - target_upper: index * inner_peers + inner_peers, - length: 0, - seqno: 0, - }; - pushes.push(Box::new(Pusher::new(header, Rc::clone(send)))) - } - - let channel = Rc::clone(self.to_local.entry(identifier).or_default()); - - use crate::allocator::counters::Puller as CountPuller; - let canary = Canary::new(identifier, Rc::clone(&self.canaries)); - let puller = Box::new(CountPuller::new(PullerInner::new(inner_recv, channel, canary), identifier, Rc::clone(self.events()))); - - let pushes = Box::new(crate::allocator::Broadcaster { spare: None, pushers: pushes }); - (pushes, puller, ) + fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_to(identifier, false) } // Perform preparatory work, most likely reading binary buffers from self.recv. diff --git a/communication/src/allocator/zero_copy/allocator_process.rs b/communication/src/allocator/zero_copy/allocator_process.rs index 7377b00bc..132263476 100644 --- a/communication/src/allocator/zero_copy/allocator_process.rs +++ b/communication/src/allocator/zero_copy/allocator_process.rs @@ -155,8 +155,10 @@ impl BytesPush for Fanout { } /// A pusher that serializes once for all other workers, and hands the element itself to this worker. +/// +/// Without a local pusher, the element is left with the caller. struct BroadcastPusher { - local: ThreadPusher, + local: Option>, remote: Option>, } @@ -164,7 +166,7 @@ impl Push for BroadcastPusher { fn push(&mut self, element: &mut Option) { // The serializing pusher reads the element and leaves it in place. if let Some(remote) = self.remote.as_mut() { remote.push(element); } - self.local.push(element); + if let Some(local) = self.local.as_mut() { local.push(element); } } } @@ -184,6 +186,36 @@ impl ProcessAllocator { let puller = Box::new(CountPuller::new(puller, identifier, Rc::clone(&self.events))); (local_send, puller) } + + /// Allocates a broadcast channel that delivers to this worker only if `include_self` is set. + fn broadcast_to(&mut self, identifier: usize, include_self: bool) -> (Box>, Box>) { + + // Assume and enforce in-order identifier allocation. + if let Some(bound) = self.channel_id_bound { + assert!(bound < identifier); + } + self.channel_id_bound = Some(identifier); + + let (local, puller) = self.local_channel::(identifier); + + // Serialize once, and hand every other worker a reference to the bytes. + let targets: Vec<_> = (0 .. self.peers).filter(|&target| target != self.index).map(|target| Rc::clone(&self.sends[target])).collect(); + let remote = if targets.is_empty() { None } else { + let header = MessageHeader { + channel: identifier, + source: self.index, + target_lower: 0, + target_upper: self.peers, + length: 0, + seqno: 0, + }; + let endpoint = SendEndpoint::new(Fanout { targets }, self.refill.clone()); + Some(Pusher::new(header, Rc::new(RefCell::new(endpoint)))) + }; + + let local = if include_self { Some(local) } else { None }; + (Box::new(BroadcastPusher { local, remote }), puller) + } } impl Allocate for ProcessAllocator { @@ -225,31 +257,11 @@ impl Allocate for ProcessAllocator { } fn broadcast(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_to(identifier, true) + } - // Assume and enforce in-order identifier allocation. - if let Some(bound) = self.channel_id_bound { - assert!(bound < identifier); - } - self.channel_id_bound = Some(identifier); - - let (local, puller) = self.local_channel::(identifier); - - // Serialize once, and hand every other worker a reference to the bytes. - let targets: Vec<_> = (0 .. self.peers).filter(|&target| target != self.index).map(|target| Rc::clone(&self.sends[target])).collect(); - let remote = if targets.is_empty() { None } else { - let header = MessageHeader { - channel: identifier, - source: self.index, - target_lower: 0, - target_upper: self.peers, - length: 0, - seqno: 0, - }; - let endpoint = SendEndpoint::new(Fanout { targets }, self.refill.clone()); - Some(Pusher::new(header, Rc::new(RefCell::new(endpoint)))) - }; - - (Box::new(BroadcastPusher { local, remote }), puller) + fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_to(identifier, false) } // Perform preparatory work, most likely reading binary buffers from self.recv. diff --git a/communication/src/allocator/zero_copy/push_pull.rs b/communication/src/allocator/zero_copy/push_pull.rs index 0e7ada2d3..adbf4bf36 100644 --- a/communication/src/allocator/zero_copy/push_pull.rs +++ b/communication/src/allocator/zero_copy/push_pull.rs @@ -137,4 +137,5 @@ impl Pull for PullerInner { &mut self.current } } + fn quiet(&mut self) { self.inner.quiet() } } diff --git a/communication/src/lib.rs b/communication/src/lib.rs index 3daac7b43..d6699d1aa 100644 --- a/communication/src/lib.rs +++ b/communication/src/lib.rs @@ -160,11 +160,18 @@ pub trait Pull { /// Takes an `Option` and leaves `None` behind. #[inline] fn recv(&mut self) -> Option { self.pull().take() } + /// Stops the puller from announcing its channel again once it has been drained. + /// + /// By default, draining a channel that yielded messages records an event for the channel, + /// which schedules its recipient once more. A recipient that finishes its work in the call + /// that drains the channel can opt out. + fn quiet(&mut self) { } } impl> Pull for Box

{ #[inline] fn pull(&mut self) -> &mut Option { (**self).pull() } + fn quiet(&mut self) { (**self).quiet() } } From ce40ba5f1a2b84f20f70f4eb3432230a45d64c54 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 25 Sep 2026 18:00:20 +0200 Subject: [PATCH 2/5] Apply child scope progress before a subgraph returns A subgraph collects the progress its child scopes report while scheduling them, and previously applied it only on its next activation, which it requested for itself. A change deep in nested scopes therefore climbed one level per worker step, and each step scheduled every enclosing scope again. The subgraph now propagates these updates before it returns, so that the effect on its outputs reaches the parent in the same call. Children whose frontiers change as a result still run on the next activation, which the subgraph requests when any remain. Propagating early can drain the tracker while such children have yet to observe their new frontiers. A subgraph with pending children now reports itself incomplete, so that the worker does not drop the dataflow before a probe sees its final frontier. Co-Authored-By: Claude Opus 5.5 --- timely/src/progress/subgraph.rs | 13 +++- timely/tests/nested_progress.rs | 126 ++++++++++++++++++++++++++++++++ 2 files changed, 136 insertions(+), 3 deletions(-) create mode 100644 timely/tests/nested_progress.rs diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index d863132fb..9d2e2e6a3 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -329,16 +329,23 @@ where // Transmit produced progress updates. self.send_progress(); - // If child scopes surface more final pointstamp updates we must re-execute. + // Apply the updates of child scopes now, rather than on a later activation, so + // that their effect on our outputs reaches the parent in this call. Children + // whose frontiers change as a result run on the next activation. if !self.final_pointstamp.is_empty() { + self.propagate_pointstamps(); + } + if !self.temp_active.is_empty() { self.activations.borrow_mut().activate(&self.path[..]); } - // A subgraph is incomplete if any child is incomplete, or there are outstanding messages. + // A subgraph is incomplete if any child is incomplete, there are outstanding messages, + // or children have yet to observe changes to their frontiers. let incomplete = self.incomplete_count > 0; let tracking = self.pointstamp_tracker.tracking_anything(); + let pending = !self.temp_active.is_empty(); - incomplete || tracking + incomplete || tracking || pending } } diff --git a/timely/tests/nested_progress.rs b/timely/tests/nested_progress.rs new file mode 100644 index 000000000..5233a385b --- /dev/null +++ b/timely/tests/nested_progress.rs @@ -0,0 +1,126 @@ +//! Progress tracking through nested regions and an iterative scope. +//! +//! Records pass through several levels of regions, each holding them until the input frontier +//! passes their time, and then circulate through a loop for a fixed number of rounds. The test +//! checks that every round completes, that all records arrive exactly once, and that the +//! dataflow shuts down once its input closes. + +use std::cell::RefCell; +use std::collections::BTreeMap; +use std::rc::Rc; + +use timely::dataflow::channels::pact::Exchange; +use timely::dataflow::operators::generic::Operator; +use timely::dataflow::operators::vec::{Filter, Map}; +use timely::dataflow::operators::{CapabilitySet, Concat, ConnectLoop, Enter, Feedback, Input, Inspect, Leave, Probe}; +use timely::dataflow::{InputHandle, ProbeHandle, Stream}; +use timely::order::Product; +use timely::progress::Timestamp; + +/// Holds records until the input frontier passes every time in their stamp. +fn hold<'s, T: Timestamp>(stream: Stream<'s, T, Vec<(u64, u64)>>) -> Stream<'s, T, Vec<(u64, u64)>> { + stream.unary_frontier(Exchange::new(|x: &(u64, u64)| x.0), "Hold", |_cap, _info| { + let mut stash: BTreeMap, (CapabilitySet, Vec<(u64, u64)>)> = BTreeMap::new(); + move |(input, frontier), output| { + input.for_each_stamp(|cap, data| { + let key = cap.stamp().elements().to_vec(); + let entry = stash.entry(key).or_insert_with(|| (cap.retain_stamp(0), Vec::new())); + for d in data { entry.1.extend(d.drain(..)); } + }); + let ready: Vec> = stash.keys().filter(|s| !s.iter().any(|t| frontier.less_equal(t))).cloned().collect(); + for s in ready { + let (caps, data) = stash.remove(&s).unwrap(); + output.session(&caps).give_iterator(data.into_iter()); + } + } + }) +} + +fn nest<'s, T: Timestamp>(stream: Stream<'s, T, Vec<(u64, u64)>>, depth: usize) -> Stream<'s, T, Vec<(u64, u64)>> { + let outer = stream.scope(); + outer.region(|inner| { + let s = hold(stream.enter(inner)); + let s = if depth > 1 { + nest(s, depth - 1) + } + else { + inner.iterative::(|it| { + let (handle, cycle) = it.feedback(Product::new(Default::default(), 1)); + // The low two bits count down the remaining rounds. + let body = hold(s.enter(it).map(|(k, v)| (k, v * 4 + 3)).concat(cycle)); + body.clone().filter(|x| x.1 % 4 != 0).map(|(k, v)| (k, v - 1)).connect_loop(handle); + body.filter(|x| x.1 % 4 == 0).map(|(k, v)| (k, v / 4)).leave(inner) + }) + }; + hold(s).leave(outer) + }) +} + +fn run(config: timely::Config) { + let results = timely::execute(config, |worker| { + let index = worker.index(); + let mut input = InputHandle::new(); + let probe = ProbeHandle::new(); + let seen = Rc::new(RefCell::new(Vec::new())); + let seen2 = Rc::clone(&seen); + worker.dataflow::(|scope| { + nest(scope.input_from(&mut input).container::>(), 4) + .inspect(move |x| seen2.borrow_mut().push(*x)) + .probe_with(&probe); + }); + for round in 0 .. 20u64 { + if index == 0 { + for i in 0 .. 3 { input.send((round * 3 + i, round * 3 + i)); } + } + input.advance_to(round + 1); + let mut steps = 0; + while probe.less_than(input.time()) { + worker.step(); + steps += 1; + assert!(steps < 100_000, "round {round} did not complete"); + } + } + drop(input); + let mut steps = 0; + while worker.step() { + steps += 1; + assert!(steps < 100_000, "dataflow did not shut down"); + } + // The probe must observe the final frontier before the dataflow shuts down. + assert!(probe.done()); + seen.take() + }).unwrap().join(); + let mut all: Vec<(u64, u64)> = results.into_iter().flat_map(|r| r.unwrap()).collect(); + all.sort(); + let expected: Vec<(u64, u64)> = (0 .. 60).map(|x| (x, x)).collect(); + assert_eq!(all, expected); +} + +#[test] +fn nested_progress_thread() { run(timely::Config::thread()); } + +#[test] +fn nested_progress_process() { run(timely::Config::process(4)); } + +/// A worker's own final update must reach operators before the dataflow completes. +fn probe_sees_final_frontier(config: timely::Config) { + timely::execute(config, |worker| { + let (mut input, probe) = worker.dataflow::(|scope| { + let (input, stream) = scope.new_input::>(); + (input, stream.probe().0) + }); + for round in 0 .. 5 { + input.advance_to(round + 1); + worker.step(); + } + drop(input); + while worker.step() { } + assert!(probe.done()); + }).unwrap(); +} + +#[test] +fn probe_sees_final_frontier_thread() { probe_sees_final_frontier(timely::Config::thread()); } + +#[test] +fn probe_sees_final_frontier_process() { probe_sees_final_frontier(timely::Config::process(2)); } From d91d149bdcd98438d4ea5c12607371e556ad26ad Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 25 Sep 2026 18:00:30 +0200 Subject: [PATCH 3/5] Apply a worker's own progress updates without a channel `Progcaster` broadcast each batch of progress updates to all workers, including the sender. The sender's copy went through its own channel, and on arrival activated the scope and every scope around it on a later step. The progress channel now excludes the sender, and `Progcaster::send` adds the sender's copy to the caller's batch of received updates. Correctness needs each worker's updates to reach every worker in the order sent, and batches applied locally keep that order. `Subgraph::schedule` propagates the copy before it returns, together with the progress of its child scopes. Progress logging records the local delivery as a receive, as before. Co-Authored-By: Claude Opus 5.5 --- timely/src/progress/broadcast.rs | 20 ++++++++++++++++++-- timely/src/progress/subgraph.rs | 8 ++++---- timely/src/worker.rs | 9 +++++++++ 3 files changed, 31 insertions(+), 6 deletions(-) diff --git a/timely/src/progress/broadcast.rs b/timely/src/progress/broadcast.rs index 2c77f223f..2ea719901 100644 --- a/timely/src/progress/broadcast.rs +++ b/timely/src/progress/broadcast.rs @@ -35,7 +35,8 @@ impl Progcaster { pub fn new(worker: &crate::worker::Worker, addr: Rc<[usize]>, identifier: usize, mut logging: Option, progress_logging: Option>) -> Progcaster { let channel_identifier = worker.new_identifier(); - let (pusher, puller) = worker.broadcast(channel_identifier, addr); + // The channel excludes this worker, and `send` applies our own updates directly. + let (pusher, puller) = worker.broadcast_peers(channel_identifier, addr); logging.as_mut().map(|l| l.log(crate::logging::CommChannelsEvent { identifier: channel_identifier, kind: crate::logging::CommChannelKind::Progress, @@ -53,7 +54,10 @@ impl Progcaster { } /// Sends pointstamp changes to all workers. - pub fn send(&mut self, changes: &mut ChangeBatch<(Location, T)>) { + /// + /// Other workers receive the changes through the channel, and this worker's copy is added to `local`. + /// Each worker's changes reach every worker in the order sent, which is all that `recv` relies on. + pub fn send(&mut self, changes: &mut ChangeBatch<(Location, T)>, local: &mut ChangeBatch<(Location, T)>) { changes.compact(); if !changes.is_empty() { @@ -83,11 +87,23 @@ impl Progcaster { channel: self.channel_identifier, seq_no: self.counter, identifier: self.identifier, + messages: messages.clone(), + internal: internal.clone(), + }); + // Log the local delivery as a receive, as it would appear had it used the channel. + l.log(crate::logging::TimelyProgressEvent { + is_send: false, + source: self.source, + channel: self.channel_identifier, + seq_no: self.counter, + identifier: self.identifier, messages, internal, }); }); + local.extend(changes.iter().cloned()); + let payload = (self.source, self.counter, std::mem::take(changes)); let mut to_push = Some(Bincode { payload }); self.pusher.push(&mut to_push); diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 9d2e2e6a3..313cae5fd 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -329,9 +329,9 @@ where // Transmit produced progress updates. self.send_progress(); - // Apply the updates of child scopes now, rather than on a later activation, so - // that their effect on our outputs reaches the parent in this call. Children - // whose frontiers change as a result run on the next activation. + // Apply the updates of child scopes and our own sent updates now, rather than on + // a later activation, so that their effect on our outputs reaches the parent in + // this call. Children whose frontiers change as a result run on the next activation. if !self.final_pointstamp.is_empty() { self.propagate_pointstamps(); } @@ -547,7 +547,7 @@ where }; if must_send { - self.progcaster.send(&mut self.local_pointstamp); + self.progcaster.send(&mut self.local_pointstamp, &mut self.final_pointstamp); } } } diff --git a/timely/src/worker.rs b/timely/src/worker.rs index f24caefca..4dfbb95da 100644 --- a/timely/src/worker.rs +++ b/timely/src/worker.rs @@ -609,6 +609,15 @@ impl Worker { self.allocator.borrow_mut().broadcast(identifier) } + /// Allocates a broadcast channel, where each pushed message is received by all workers except the sender. + pub fn broadcast_peers(&self, identifier: usize, address: Rc<[usize]>) -> (Box>, Box>) { + if address.is_empty() { panic!("Unacceptable address: Length zero"); } + let mut paths = self.paths.borrow_mut(); + paths.insert(identifier, address); + self.temp_channel_ids.borrow_mut().push(identifier); + self.allocator.borrow_mut().broadcast_peers(identifier) + } + /// Construct a new dataflow. /// /// # Examples From 96ea1628f334e51803f5c5f30163ef88110cab41 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 25 Sep 2026 18:00:30 +0200 Subject: [PATCH 4/5] Do not reschedule a scope after draining its progress channel A counting puller records an event for its channel once it has been drained of messages, so that the channel's recipient runs once more. For progress channels the recipient is `Subgraph::schedule`, which applies everything it receives in the same call and activates itself when work remains. The extra activation scheduled the scope and all of its enclosing scopes to find nothing to do. The progress puller is now quiet. Data channels keep the event. Co-Authored-By: Claude Opus 5.5 --- timely/src/progress/broadcast.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/timely/src/progress/broadcast.rs b/timely/src/progress/broadcast.rs index 2ea719901..c69d1680e 100644 --- a/timely/src/progress/broadcast.rs +++ b/timely/src/progress/broadcast.rs @@ -36,7 +36,10 @@ impl Progcaster { let channel_identifier = worker.new_identifier(); // The channel excludes this worker, and `send` applies our own updates directly. - let (pusher, puller) = worker.broadcast_peers(channel_identifier, addr); + let (pusher, mut puller) = worker.broadcast_peers(channel_identifier, addr); + // `Subgraph::schedule` applies everything it receives in the call that drains the channel, + // and activates itself if work remains, so a drained channel need not schedule it again. + puller.quiet(); logging.as_mut().map(|l| l.log(crate::logging::CommChannelsEvent { identifier: channel_identifier, kind: crate::logging::CommChannelKind::Progress, From 6dcf482b6d65fa0d8a134963446630120c235fc6 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 25 Sep 2026 20:32:45 +0200 Subject: [PATCH 5/5] Defer a dataflow root's own progress to its next activation Applying progress before a subgraph returns lets its effect reach the parent in the same call. A dataflow root has no parent, so propagating its own sent updates early only splits one propagation per step into two, one for its own updates and one for those that arrive from other workers. In a loop at the root, `barrier` with four workers took 8% longer than before this series. A dataflow root now leaves the updates for its next activation, which it requests, and applies them together with those it receives by then. Nested scopes still propagate before they return. The root counts unapplied updates as pending work when it reports whether it is complete. Co-Authored-By: Claude Opus 5.5 --- timely/src/progress/subgraph.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 313cae5fd..e318cbb2b 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -332,18 +332,20 @@ where // Apply the updates of child scopes and our own sent updates now, rather than on // a later activation, so that their effect on our outputs reaches the parent in // this call. Children whose frontiers change as a result run on the next activation. - if !self.final_pointstamp.is_empty() { + // A dataflow root has no parent to report to, and applies the updates on its next + // activation, together with those it receives from other workers by then. + if !self.final_pointstamp.is_empty() && self.path.len() > 1 { self.propagate_pointstamps(); } - if !self.temp_active.is_empty() { + if !self.temp_active.is_empty() || !self.final_pointstamp.is_empty() { self.activations.borrow_mut().activate(&self.path[..]); } // A subgraph is incomplete if any child is incomplete, there are outstanding messages, - // or children have yet to observe changes to their frontiers. + // updates remain to be applied, or children have yet to observe changes to their frontiers. let incomplete = self.incomplete_count > 0; let tracking = self.pointstamp_tracker.tracking_anything(); - let pending = !self.temp_active.is_empty(); + let pending = !self.temp_active.is_empty() || !self.final_pointstamp.is_empty(); incomplete || tracking || pending }