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()`) and whether it is empty (`is_empty()`).
+/// It describes its update count (`count()`).
 ///
 /// We require [`Default`] for convenience purposes.
 pub trait WithProgress: Default {
@@ -17,12 +17,6 @@ pub trait WithProgress: Default {
     /// imposed on the [`CountPreservingContainerBuilder`] trait, whose implementors
     /// must preserve the number of items.
     fn count(&self) -> usize;
-
-    /// Determine if the container contains any elements, corresponding to `count() == 0`.
-    #[inline(always)]
-    fn is_empty(&self) -> bool {
-        self.count() == 0
-    }
 }
 
 /// A container that can reveal its contents through iterating by reference and draining.
@@ -90,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 + Default + Clone + 'static;
+    type Container: WithProgress + Clone + 'static;
     /// Extract assembled containers, potentially leaving unfinished data behind. Can
     /// be called repeatedly, for example while the caller can send data.
     ///
@@ -204,7 +198,7 @@ impl<C: WithProgress + Clone + 'static> ContainerBuilder for CapacityContainerBu
 
     #[inline]
     fn finish(&mut self) -> Option<&mut C> {
-        if !self.current.is_empty() {
+        if self.current.count() > 0 {
             self.pending.push_back(std::mem::take(&mut self.current));
         }
         self.empty = self.pending.pop_front();
@@ -337,15 +331,12 @@ pub mod buffer {
 
 impl<T> WithProgress for Vec<T> {
     #[inline(always)] fn count(&self) -> usize { self.len() }
-    #[inline(always)] fn is_empty(&self) -> bool { Vec::is_empty(self) }
 }
 
 impl<T: WithProgress> WithProgress for std::rc::Rc<T> {
     #[inline(always)] fn count(&self) -> usize { self.as_ref().count() }
-    #[inline(always)] fn is_empty(&self) -> bool { self.as_ref().is_empty() }
 }
 
 impl<T: WithProgress> WithProgress for std::sync::Arc<T> {
     #[inline(always)] fn count(&self) -> usize { self.as_ref().count() }
-    #[inline(always)] fn is_empty(&self) -> bool { self.as_ref().is_empty() }
 }

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, WithProgress, PushInto, SizableContainer};
+use crate::container::{ContainerBuilder, CapacityContainerBuilder, PushInto, SizableContainer};
 use crate::dataflow::channels::Message;
 use crate::dataflow::operators::Capability;
 use crate::progress::Timestamp;
@@ -109,11 +109,9 @@ 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) {
-        if !container.is_empty() {
-            self.flush();
-            let time = self.time.as_ref().unwrap().clone();
-            Message::push_at(container, time, &mut self.pusher);
-        }
+        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/capture/extract.rs

@@ -74,9 +74,7 @@ where
             for datum in to_sort.into_iter() {
                 sorted.push_into(datum);
             }
-            if !sorted.is_empty() {
-                result.push((time, sorted));
-            }
+            result.push((time, sorted));
         }
         result
     }

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

@@ -30,9 +30,7 @@ where
     fn filter<P: FnMut(&C::Item<'_>)->bool+'static>(&self, mut predicate: P) -> StreamCore<G, C> {
         self.unary(Pipeline, "Filter", move |_,_| move |input, output| {
             input.for_each(|time, data| {
-                if !data.is_empty() {
-                    output.session(&time).give_iterator(data.drain().filter(&mut predicate));
-                }
+                output.session(&time).give_iterator(data.drain().filter(&mut predicate));
             });
         })
     }

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, WithProgress, PushInto};
+use crate::container::{CapacityContainerBuilder, ContainerBuilder, SizableContainer, PushInto};
 
 use crate::scheduling::{Schedule, Activator};
 
@@ -439,11 +439,9 @@ impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
     /// });
     /// ```
     pub fn send_batch(&mut self, buffer: &mut CB::Container) {
-        if !buffer.is_empty() {
-            // flush buffered elements to ensure local fifo.
-            self.flush();
-            Self::send_container(buffer, &mut self.buffer, &mut self.pushers, &self.now_at);
-        }
+        // 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`.