Drop implementation for Tracker

This allows the tracker to log reachability information when dropping
dataflows.

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>

Author: antiguru

timely/src/progress/frontier.rs

@@ -528,6 +528,14 @@ impl<T> MutableAntichain<T> {
             .map(|td| td.1)
             .sum()
     }
+
+    pub(crate) fn updates(&mut self) -> impl Iterator<Item=&(T, i64)>
+    where
+        T: Clone + PartialOrder + Ord,
+    {
+        self.rebuild();
+        self.updates.iter()
+    }
 }
 
 impl<T> Default for MutableAntichain<T> {

timely/src/progress/reachability.rs

@@ -897,4 +897,36 @@ pub mod logging {
     impl From<TargetUpdate> for TrackerEvent {
         fn from(v: TargetUpdate) -> TrackerEvent { TrackerEvent::TargetUpdate(v) }
     }
-}
+}
+
+impl<T: Timestamp> Drop for Tracker<T> {
+    fn drop(&mut self) {
+        let Some(logger) = &mut self.logger else { return; };
+        for (index, per_operator) in self.per_operator.iter_mut().enumerate() {
+            let target_changes = per_operator.targets
+                .iter_mut()
+                .enumerate()
+                .flat_map(|(port, target)| {
+                    target.pointstamps
+                        .updates()
+                        .map(move |(time, diff)| (index, port, time.clone(), -diff))
+                })
+                .collect::<Vec<_>>();
+            if !target_changes.is_empty() {
+                logger.log_target_updates(Box::new(target_changes));
+            }
+            let source_changes = per_operator.sources
+                .iter_mut()
+                .enumerate()
+                .flat_map(|(port, source)| {
+                    source.pointstamps
+                        .updates()
+                        .map(move |(time, diff)| (index, port, time.clone(), -diff))
+                })
+                .collect::<Vec<_>>();
+            if !source_changes.is_empty() {
+                logger.log_source_updates(Box::new(source_changes));
+            }
+        }
+    }
+}