Remove WithProgress::is_empty

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

Author: antiguru

container/src/lib.rs

@@ -6,7 +6,7 @@ use std::collections::VecDeque;
 
 /// A type representing progress, with an update count.
 ///
-/// It describes its update count (`count()`).
+/// It describes its update count (`count()`) and whether it is empty (`is_empty()`).
 ///
 /// We require [`Default`] for convenience purposes.
 pub trait WithProgress: Default {
@@ -84,7 +84,7 @@ pub trait PushInto<T> {
 /// decide to represent a push order for `extract` and `finish`, or not.
 pub trait ContainerBuilder: Default + 'static {
     /// The container type we're building.
-    type Container: WithProgress + Clone + 'static;
+    type Container: WithProgress + Default + Clone + 'static;
     /// Extract assembled containers, potentially leaving unfinished data behind. Can
     /// be called repeatedly, for example while the caller can send data.
     ///

timely/src/dataflow/channels/pushers/buffer.rs

@@ -2,7 +2,7 @@
 //! with the performance of batched sends.
 
 use crate::communication::Push;
-use crate::container::{ContainerBuilder, CapacityContainerBuilder, PushInto, SizableContainer};
+use crate::container::{ContainerBuilder, CapacityContainerBuilder, WithProgress, PushInto, SizableContainer};
 use crate::dataflow::channels::Message;
 use crate::dataflow::operators::Capability;
 use crate::progress::Timestamp;
@@ -109,9 +109,11 @@ impl<T, CB: ContainerBuilder, P: Push<Message<T, CB::Container>>> Buffer<T, CB,
     // buffer always requires a container builder. We could expose the buffer's underlying pusher
     // directly, but this would bypass the buffer's time tracking.
     fn give_container(&mut self, container: &mut CB::Container) {
-        self.flush();
-        let time = self.time.as_ref().unwrap().clone();
-        Message::push_at(container, time, &mut self.pusher);
+        if container.count() > 0 {
+            self.flush();
+            let time = self.time.as_ref().unwrap().clone();
+            Message::push_at(container, time, &mut self.pusher);
+        }
     }
 
     /// An internal implementation of push that should only be called by sessions.

timely/src/dataflow/operators/core/input.rs

@@ -3,7 +3,7 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 
-use crate::container::{CapacityContainerBuilder, ContainerBuilder, SizableContainer, PushInto};
+use crate::container::{CapacityContainerBuilder, ContainerBuilder, SizableContainer, WithProgress, PushInto};
 
 use crate::scheduling::{Schedule, Activator};
 
@@ -439,9 +439,11 @@ impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
     /// });
     /// ```
     pub fn send_batch(&mut self, buffer: &mut CB::Container) {
-        // flush buffered elements to ensure local fifo.
-        self.flush();
-        Self::send_container(buffer, &mut self.buffer, &mut self.pushers, &self.now_at);
+        if !buffer.count() > 0 {
+            // flush buffered elements to ensure local fifo.
+            self.flush();
+            Self::send_container(buffer, &mut self.buffer, &mut self.pushers, &self.now_at);
+        }
     }
 
     /// Advances the current epoch to `next`.