Progress container with length

Splits the Container trait into a progress container and a container. The
progress container only exposes a length, which is enough for most parts of
Timely. The container trait extends the progress container and carries the
usual functions to iterate and clear the contents.

I'm open to renaming and moving types, and primarily wanted to check
whether it's possible to split the trait.

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

Author: antiguru

container/src/lib.rs

@@ -7,27 +7,11 @@ use std::collections::VecDeque;
 /// A container transferring data through dataflow edges
 ///
 /// A container stores a number of elements and thus is able to describe it length (`len()`) and
-/// whether it is empty (`is_empty()`). It supports removing all elements (`clear`).
+/// whether it is empty (`is_empty()`).
 ///
 /// A container must implement default. The default implementation is not required to allocate
 /// memory for variable-length components.
-///
-/// We require the container to be cloneable to enable efficient copies when providing references
-/// of containers to operators. Care must be taken that the type's `clone_from` implementation
-/// is efficient (which is not necessarily the case when deriving `Clone`.)
-pub trait Container: Default {
-    /// The type of elements when reading non-destructively from the container.
-    type ItemRef<'a> where Self: 'a;
-
-    /// The type of elements when draining the container.
-    type Item<'a> where Self: 'a;
-
-    /// Push `item` into self
-    #[inline]
-    fn push<T>(&mut self, item: T) where Self: PushInto<T> {
-        self.push_into(item)
-    }
-
+pub trait ProgressContainer: Default {
     /// The number of elements in this container
     ///
     /// This number is used in progress tracking to confirm the receipt of some number
@@ -37,9 +21,27 @@ pub trait Container: Default {
     fn len(&self) -> usize;
 
     /// Determine if the container contains any elements, corresponding to `len() == 0`.
+    #[inline(always)]
     fn is_empty(&self) -> bool {
         self.len() == 0
     }
+}
+
+/// A container that can reveal its contents through iterating by reference and draining.
+///
+/// It supports removing all elements (`clear`).
+pub trait Container: ProgressContainer {
+    /// The type of elements when reading non-destructively from the container.
+    type ItemRef<'a> where Self: 'a;
+
+    /// The type of elements when draining the container.
+    type Item<'a> where Self: 'a;
+
+    /// Push `item` into self
+    #[inline]
+    fn push<T>(&mut self, item: T) where Self: PushInto<T> {
+        self.push_into(item)
+    }
 
     /// Remove all contents from `self` while retaining allocated memory.
     /// After calling `clear`, `is_empty` must return `true` and `len` 0.
@@ -102,7 +104,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: Container + Clone + 'static;
+    type Container: ProgressContainer + Clone + 'static;
     /// Extract assembled containers, potentially leaving unfinished data behind. Can
     /// be called repeatedly, for example while the caller can send data.
     ///
@@ -118,6 +120,7 @@ pub trait ContainerBuilder: Default + 'static {
     where
         Self: for<'a> PushInto<<Self::Container as Container>::Item<'a>>,
         I: for<'a> FnMut(&<Self::Container as Container>::Item<'a>) -> usize,
+        Self::Container: Container,
     {
         for datum in container.drain() {
             let index = index(&datum);
@@ -142,6 +145,35 @@ pub trait ContainerBuilder: Default + 'static {
 /// If you have any questions about this trait you are best off not implementing it.
 pub trait LengthPreservingContainerBuilder : ContainerBuilder { }
 
+/// A container builder that never produces any outputs, and can be used to pass through data in
+/// operators.
+#[derive(Debug, Clone)]
+pub struct PassthroughContainerBuilder<C>(std::marker::PhantomData<C>);
+
+impl<C> Default for PassthroughContainerBuilder<C> {
+    #[inline(always)]
+    fn default() -> Self {
+        PassthroughContainerBuilder(std::marker::PhantomData)
+    }
+}
+
+impl<C> ContainerBuilder for PassthroughContainerBuilder<C>
+where
+    C: ProgressContainer + Clone + 'static,
+{
+    type Container = C;
+
+    #[inline(always)]
+    fn extract(&mut self) -> Option<&mut Self::Container> {
+        None
+    }
+
+    #[inline(always)]
+    fn finish(&mut self) -> Option<&mut Self::Container> {
+        None
+    }
+}
+
 /// A default container builder that uses length and preferred capacity to chunk data.
 ///
 /// Maintains a single empty allocation between [`Self::push_into`] and [`Self::extract`], but not
@@ -165,7 +197,7 @@ impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuil
         self.current.ensure_capacity(&mut self.empty);
 
         // Push item
-        self.current.push(item);
+        self.current.push_into(item);
 
         // Maybe flush
         if self.current.at_capacity() {
@@ -199,18 +231,18 @@ impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuild
 
 impl<C: Container + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
 
+impl<T> ProgressContainer for Vec<T> {
+    #[inline(always)]
+    fn len(&self) -> usize { Vec::len(self) }
+
+    #[inline(always)]
+    fn is_empty(&self) -> bool { Vec::is_empty(self) }
+}
+
 impl<T> Container for Vec<T> {
     type ItemRef<'a> = &'a T where T: 'a;
     type Item<'a> = T where T: 'a;
 
-    fn len(&self) -> usize {
-        Vec::len(self)
-    }
-
-    fn is_empty(&self) -> bool {
-        Vec::is_empty(self)
-    }
-
     fn clear(&mut self) { Vec::clear(self) }
 
     type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;
@@ -268,20 +300,20 @@ mod rc {
     use std::ops::Deref;
     use std::rc::Rc;
 
-    use crate::Container;
+    use crate::{Container, ProgressContainer};
+
+    impl<T: ProgressContainer> ProgressContainer for Rc<T> {
+        #[inline(always)]
+        fn len(&self) -> usize { std::ops::Deref::deref(self).len() }
+
+        #[inline(always)]
+        fn is_empty(&self) -> bool { std::ops::Deref::deref(self).is_empty() }
+    }
 
     impl<T: Container> Container for Rc<T> {
         type ItemRef<'a> = T::ItemRef<'a> where Self: 'a;
         type Item<'a> = T::ItemRef<'a> where Self: 'a;
 
-        fn len(&self) -> usize {
-            std::ops::Deref::deref(self).len()
-        }
-
-        fn is_empty(&self) -> bool {
-            std::ops::Deref::deref(self).is_empty()
-        }
-
         fn clear(&mut self) {
             // Try to reuse the allocation if possible
             if let Some(inner) = Rc::get_mut(self) {
@@ -309,20 +341,20 @@ mod arc {
     use std::ops::Deref;
     use std::sync::Arc;
 
-    use crate::Container;
+    use crate::{Container, ProgressContainer};
+
+    impl<T: ProgressContainer> ProgressContainer for Arc<T> {
+        #[inline(always)]
+        fn len(&self) -> usize { std::ops::Deref::deref(self).len() }
+
+        #[inline(always)]
+        fn is_empty(&self) -> bool { std::ops::Deref::deref(self).is_empty() }
+    }
 
     impl<T: Container> Container for Arc<T> {
         type ItemRef<'a> = T::ItemRef<'a> where Self: 'a;
         type Item<'a> = T::ItemRef<'a> where Self: 'a;
 
-        fn len(&self) -> usize {
-            std::ops::Deref::deref(self).len()
-        }
-
-        fn is_empty(&self) -> bool {
-            std::ops::Deref::deref(self).is_empty()
-        }
-
         fn clear(&mut self) {
             // Try to reuse the allocation if possible
             if let Some(inner) = Arc::get_mut(self) {

timely/examples/columnar.rs

@@ -165,8 +165,11 @@ mod container {
         }
     }
 
-    impl<C: columnar::ContainerBytes> timely::Container for Column<C> {
+    impl<C: columnar::ContainerBytes> timely::container::ProgressContainer for Column<C> {
+        #[inline(always)]
         fn len(&self) -> usize { self.borrow().len() }
+    }
+    impl<C: columnar::ContainerBytes> timely::Container for Column<C> {
         // This sets `self` to be an empty `Typed` variant, appropriate for pushing into.
         fn clear(&mut self) {
             match self {

timely/src/dataflow/channels/mod.rs

@@ -2,7 +2,7 @@
 
 use serde::{Deserialize, Serialize};
 use crate::communication::Push;
-use crate::Container;
+use crate::container::ProgressContainer;
 
 /// A collection of types that may be pushed at.
 pub mod pushers;
@@ -32,14 +32,15 @@ impl<T, C> Message<T, C> {
     }
 }
 
-impl<T, C: Container> Message<T, C> {
+impl<T, C: ProgressContainer> Message<T, C> {
     /// Creates a new message instance from arguments.
     pub fn new(time: T, data: C, from: usize, seq: usize) -> Self {
         Message { time, data, from, seq }
     }
 
     /// Forms a message, and pushes contents at `pusher`. Replaces `buffer` with what the pusher
-    /// leaves in place, or the container's default element. The buffer is cleared.
+    /// leaves in place, or the container's default element. The buffer's contents are left in an
+    /// undefined state, specifically the caller cannot rely on this function clearing the buffer.
     #[inline]
     pub fn push_at<P: Push<Message<T, C>>>(buffer: &mut C, time: T, pusher: &mut P) {
 
@@ -51,7 +52,6 @@ impl<T, C: Container> Message<T, C> {
 
         if let Some(message) = bundle {
             *buffer = message.data;
-            buffer.clear();
         }
     }
 }

timely/src/dataflow/channels/pact.rs

@@ -10,7 +10,7 @@
 use std::{fmt::{self, Debug}, marker::PhantomData};
 use std::rc::Rc;
 
-use crate::{Container, container::{ContainerBuilder, LengthPreservingContainerBuilder, SizableContainer, CapacityContainerBuilder, PushInto}};
+use crate::{Container, container::{ContainerBuilder, LengthPreservingContainerBuilder, ProgressContainer, SizableContainer, CapacityContainerBuilder, PushInto}};
 use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
 use crate::communication::{Push, Pull};
 use crate::dataflow::channels::pushers::Exchange as ExchangePusher;
@@ -34,7 +34,7 @@ pub trait ParallelizationContract<T, C> {
 #[derive(Debug)]
 pub struct Pipeline;
 
-impl<T: 'static, C: Container + 'static> ParallelizationContract<T, C> for Pipeline {
+impl<T: 'static, C: ProgressContainer + 'static> ParallelizationContract<T, C> for Pipeline {
     type Pusher = LogPusher<T, C, ThreadPusher<Message<T, C>>>;
     type Puller = LogPuller<T, C, ThreadPuller<Message<T, C>>>;
     fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: Rc<[usize]>, logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
@@ -52,7 +52,7 @@ pub type Exchange<D, F> = ExchangeCore<CapacityContainerBuilder<Vec<D>>, F>;
 
 impl<CB, F> ExchangeCore<CB, F>
 where
-    CB: LengthPreservingContainerBuilder,
+    CB: LengthPreservingContainerBuilder<Container: Container>,
     for<'a> F: FnMut(&<CB::Container as Container>::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
@@ -81,7 +81,7 @@ where
 // Exchange uses a `Box<Pushable>` because it cannot know what type of pushable will return from the allocator.
 impl<T: Timestamp, CB, H: 'static> ParallelizationContract<T, CB::Container> for ExchangeCore<CB, H>
 where
-    CB: ContainerBuilder,
+    CB: ContainerBuilder<Container: Container>,
     CB: for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
     CB::Container: Data + Send + crate::dataflow::channels::ContainerBytes,
     for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
@@ -129,7 +129,7 @@ impl<T, C, P: Push<Message<T, C>>> LogPusher<T, C, P> {
     }
 }
 
-impl<T, C: Container, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<T, C, P> {
+impl<T, C: ProgressContainer, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<T, C, P> {
     #[inline]
     fn push(&mut self, pair: &mut Option<Message<T, C>>) {
         if let Some(bundle) = pair {
@@ -179,7 +179,7 @@ impl<T, C, P: Pull<Message<T, C>>> LogPuller<T, C, P> {
     }
 }
 
-impl<T, C: Container, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<T, C, P> {
+impl<T, C: ProgressContainer, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<T, C, P> {
     #[inline]
     fn pull(&mut self) -> &mut Option<Message<T, C>> {
         let result = self.puller.pull();

timely/src/dataflow/channels/pullers/counter.rs

@@ -6,7 +6,7 @@ use std::cell::RefCell;
 use crate::dataflow::channels::Message;
 use crate::progress::ChangeBatch;
 use crate::communication::Pull;
-use crate::Container;
+use crate::container::ProgressContainer;
 
 /// A wrapper which accounts records pulled past in a shared count map.
 pub struct Counter<T: Ord+Clone+'static, C, P: Pull<Message<T, C>>> {
@@ -36,7 +36,7 @@ impl<T:Ord+Clone+'static> Drop for ConsumedGuard<T> {
     }
 }
 
-impl<T:Ord+Clone+'static, C: Container, P: Pull<Message<T, C>>> Counter<T, C, P> {
+impl<T:Ord+Clone+'static, C: ProgressContainer, P: Pull<Message<T, C>>> Counter<T, C, P> {
     /// Retrieves the next timestamp and batch of data.
     #[inline]
     pub fn next(&mut self) -> Option<&mut Message<T, C>> {

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};
+use crate::container::{ContainerBuilder, CapacityContainerBuilder, ProgressContainer, PushInto};
 use crate::dataflow::channels::Message;
 use crate::dataflow::operators::Capability;
 use crate::progress::Timestamp;

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

@@ -7,7 +7,7 @@ use std::cell::RefCell;
 use crate::progress::{ChangeBatch, Timestamp};
 use crate::dataflow::channels::Message;
 use crate::communication::Push;
-use crate::Container;
+use crate::container::ProgressContainer;
 
 /// A wrapper which updates shared `produced` based on the number of records pushed.
 #[derive(Debug)]
@@ -17,7 +17,7 @@ pub struct Counter<T, C, P: Push<Message<T, C>>> {
     phantom: PhantomData<C>,
 }
 
-impl<T: Timestamp, C: Container, P> Push<Message<T, C>> for Counter<T, C, P> where P: Push<Message<T, C>> {
+impl<T: Timestamp, C: ProgressContainer, P> Push<Message<T, C>> for Counter<T, C, P> where P: Push<Message<T, C>> {
     #[inline]
     fn push(&mut self, message: &mut Option<Message<T, C>>) {
         if let Some(message) = message {

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

@@ -9,7 +9,7 @@ use crate::{Container, Data};
 /// Distributes records among target pushees according to a distribution function.
 pub struct Exchange<T, CB, P, H>
 where
-    CB: ContainerBuilder,
+    CB: ContainerBuilder<Container: Container>,
     P: Push<Message<T, CB::Container>>,
     for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
 {
@@ -21,7 +21,7 @@ where
 
 impl<T: Clone, CB, P, H>  Exchange<T, CB, P, H>
 where
-    CB: ContainerBuilder,
+    CB: ContainerBuilder<Container: Container>,
     P: Push<Message<T, CB::Container>>,
     for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
 {
@@ -50,7 +50,7 @@ where
 
 impl<T: Eq+Data, CB, P, H> Push<Message<T, CB::Container>> for Exchange<T, CB, P, H>
 where
-    CB: ContainerBuilder,
+    CB: ContainerBuilder<Container: Container>,
     CB: for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
     P: Push<Message<T, CB::Container>>,
     for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64