Return implication changes from propagate_all

This commit makes the reachability tracker directly return changes to
implications from `propagate_all`, rather than stashing them and
expecting callers to access them through a separate `pushed` method.
This API changes allows `propagate_all` to assume that the caller will
consume the changes before the next call to `propagate_all`, which in
turn guarantees that `pushed_changes` only contains changes from the
last `propagate_all` call, rather than possibly changes from previous
calls as well.

Author: teskje

timely/src/progress/reachability.rs

@@ -33,12 +33,9 @@
 //! tracker.update_source(Source::new(0, 0), 17, 1);
 //!
 //! // Propagate changes; until this call updates are simply buffered.
-//! tracker.propagate_all();
+//! let updates = tracker.propagate_all();
 //!
-//! let mut results =
-//! tracker
-//!     .pushed()
-//!     .drain()
+//! let mut results = updates
 //!     .filter(|((location, time), delta)| location.is_target())
 //!     .collect::<Vec<_>>();
 //!
@@ -55,12 +52,9 @@
 //! tracker.update_source(Source::new(0, 0), 17, -1);
 //!
 //! // Propagate changes; until this call updates are simply buffered.
-//! tracker.propagate_all();
+//! let updates = tracker.propagate_all();
 //!
-//! let mut results =
-//! tracker
-//!     .pushed()
-//!     .drain()
+//! let mut results = updates
 //!     .filter(|((location, time), delta)| location.is_target())
 //!     .collect::<Vec<_>>();
 //!
@@ -564,8 +558,10 @@ impl<T:Timestamp> Tracker<T> {
     /// Propagates all pending updates.
     ///
     /// The method drains `self.input_changes` and circulates their implications
-    /// until we cease deriving new implications.
-    pub fn propagate_all(&mut self) {
+    /// until we cease deriving new implications. It returns an iterator over updates
+    /// to implications.
+    pub fn propagate_all(&mut self) -> impl Iterator<Item = ((Location, T), i64)> + '_ {
+        self.pushed_changes.clear();
 
         // Step 0: If logging is enabled, construct and log inbound changes.
         if let Some(logger) = &mut self.logger {
@@ -699,18 +695,15 @@ impl<T:Timestamp> Tracker<T> {
                 };
             }
         }
+
+        self.pushed_changes.drain()
     }
 
     /// Implications of maintained capabilities projected to each output.
     pub fn pushed_output(&mut self) -> &mut [ChangeBatch<T>] {
         &mut self.output_changes[..]
     }
 
-    /// A mutable reference to the pushed results of changes.
-    pub fn pushed(&mut self) -> &mut ChangeBatch<(Location, T)> {
-        &mut self.pushed_changes
-    }
-
     /// Reveals per-operator frontier state.
     pub fn node_state(&self, index: usize) -> &PerOperator<T> {
         &self.per_operator[index]

timely/src/progress/subgraph.rs

@@ -459,10 +459,10 @@ where
         }
 
         // Propagate implications of progress changes.
-        self.pointstamp_tracker.propagate_all();
+        let updates = self.pointstamp_tracker.propagate_all();
 
         // Drain propagated information into shared progress structure.
-        for ((location, time), diff) in self.pointstamp_tracker.pushed().drain() {
+        for ((location, time), diff) in updates {
             self.maybe_shutdown.push(location.node);
             // Targets are actionable, sources are not.
             if let crate::progress::Port::Target(port) = location.port {