Probe only retains weak handle to Rc

The probe operator writes into a shared frontier, but it does so even if
there is no other reference to the shared frontier. This change adjusts
the behavior to only write to the shared frontier while there exists
another reference.

Signed-off-by: Moritz Hoffmann <[email protected]>

timely/src/dataflow/operators/probe.rs

@@ -76,7 +76,7 @@ pub trait Probe<G: Scope, D: Container> {
     ///     }
     /// }).unwrap();
     /// ```
-    fn probe_with(&self, handle: &mut Handle<G::Timestamp>) -> StreamCore<G, D>;
+    fn probe_with(&self, handle: &Handle<G::Timestamp>) -> StreamCore<G, D>;
 }
 
 impl<G: Scope, D: Container> Probe<G, D> for StreamCore<G, D> {
@@ -87,14 +87,14 @@ impl<G: Scope, D: Container> Probe<G, D> for StreamCore<G, D> {
         self.probe_with(&mut handle);
         handle
     }
-    fn probe_with(&self, handle: &mut Handle<G::Timestamp>) -> StreamCore<G, D> {
+    fn probe_with(&self, handle: &Handle<G::Timestamp>) -> StreamCore<G, D> {
 
         let mut builder = OperatorBuilder::new("Probe".to_owned(), self.scope());
         let mut input = PullCounter::new(builder.new_input(self, Pipeline));
         let (tee, stream) = builder.new_output();
         let mut output = PushBuffer::new(PushCounter::new(tee));
 
-        let shared_frontier = handle.frontier.clone();
+        let shared_frontier = Rc::downgrade(&handle.frontier);
         let mut started = false;
 
         let mut vector = Default::default();
@@ -103,8 +103,10 @@ impl<G: Scope, D: Container> Probe<G, D> for StreamCore<G, D> {
             move |progress| {
 
                 // surface all frontier changes to the shared frontier.
-                let mut borrow = shared_frontier.borrow_mut();
-                borrow.update_iter(progress.frontiers[0].drain());
+                if let Some(shared_frontier) = shared_frontier.upgrade() {
+                    let mut borrow = shared_frontier.borrow_mut();
+                    borrow.update_iter(progress.frontiers[0].drain());
+                }
 
                 if !started {
                     // discard initial capability.