diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index dcd377d7e..088bd8940 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -191,6 +191,12 @@ pub struct SharedProgress { pub internals: Vec>, /// Produced message changes reported by the child operator. pub produceds: Vec>, + /// Outputs on which a scope sent records this activation, before reporting them as produced. + /// + /// 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, } impl SharedProgress { @@ -201,6 +207,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 d8bdf243d..6ed89c9a9 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(); @@ -380,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 @@ -414,12 +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); } 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)); + } } } @@ -590,7 +608,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. @@ -752,7 +770,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(); @@ -781,6 +803,14 @@ impl PerOperatorState { } } } + for (output, pushed) in shared_progress.pushed.iter_mut().enumerate() { + if std::mem::take(pushed) { + for target in self.edges[output].iter() { + if target.node == 0 { outer_pushed[target.port] = true; } + else { temp_active.push(Reverse(target.node)); } + } + } + } } /// Test the validity of `self.shared_progress`.