Add cease to output handles (#496)

In certain situations, a handle survives longer than we would like to wait
for a flush to happen. In this cases, an explicit call to cease can help
to indicate to the rest of the system that no more data follows
immediately, which is equivalent to dropping the handle.

Specifically, in async code the handle can be long-lived and survive await
points, which makes it more important to signal momentary completion to the
system.

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

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

Author: antiguru

timely/src/dataflow/operators/generic/handles.rs

@@ -221,6 +221,11 @@ impl<'a, T: Timestamp, C: Container, P: Push<BundleCore<T, C>>> OutputHandleCore
         assert!(cap.valid_for_output(&self.internal_buffer), "Attempted to open output session with invalid capability");
         self.push_buffer.session(cap.time())
     }
+
+    /// Flushes all pending data and indicate that no more data immediately follows.
+    pub fn cease(&mut self) {
+        self.push_buffer.cease();
+    }
 }
 
 impl<'a, T: Timestamp, C: Container, P: Push<BundleCore<T, C>>> Drop for OutputHandleCore<'a, T, C, P> {