Rework container traits

container/src/lib.rs

@@ -4,46 +4,34 @@
 
 use std::collections::VecDeque;
 
-/// A container transferring data through dataflow edges
+/// A type representing progress, with an update count.
 ///
-/// 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`).
+/// It describes its update count (`count()`) and 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)
-    }
-
+/// We require [`Default`] for convenience purposes.
+pub trait WithProgress: Default {
     /// The number of elements in this container
     ///
     /// This number is used in progress tracking to confirm the receipt of some number
     /// of outstanding records, and it is highly load bearing. The main restriction is
-    /// imposed on the `LengthPreservingContainerBuilder` trait, whose implementors
+    /// imposed on the [`CountPreservingContainerBuilder`] trait, whose implementors
     /// must preserve the number of items.
-    fn len(&self) -> usize;
+    fn count(&self) -> usize;
 
-    /// Determine if the container contains any elements, corresponding to `len() == 0`.
+    /// Determine if the container contains any elements, corresponding to `count() == 0`.
+    #[inline(always)]
     fn is_empty(&self) -> bool {
-        self.len() == 0
+        self.count() == 0
     }
+}
 
-    /// Remove all contents from `self` while retaining allocated memory.
-    /// After calling `clear`, `is_empty` must return `true` and `len` 0.
-    fn clear(&mut self);
+/// A container that can reveal its contents through iterating by reference and draining.
+pub trait IterableContainer: WithProgress {
+    /// 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;
 
     /// Iterator type when reading from the container.
     type Iter<'a>: Iterator<Item=Self::ItemRef<'a>> where Self: 'a;
@@ -60,7 +48,7 @@ pub trait Container: Default {
 }
 
 /// A container that can be sized and reveals its capacity.
-pub trait SizableContainer: Container {
+pub trait SizableContainer: WithProgress {
     /// Indicates that the container is "full" and should be shipped.
     fn at_capacity(&self) -> bool;
     /// Restores `self` to its desired capacity, if it has one.
@@ -102,7 +90,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: WithProgress + Default + Clone + 'static;
     /// Extract assembled containers, potentially leaving unfinished data behind. Can
     /// be called repeatedly, for example while the caller can send data.
     ///
@@ -116,14 +104,14 @@ pub trait ContainerBuilder: Default + 'static {
     /// Partitions `container` among `builders`, using the function `index` to direct items.
     fn partition<I>(container: &mut Self::Container, builders: &mut [Self], mut index: I)
     where
-        Self: for<'a> PushInto<<Self::Container as Container>::Item<'a>>,
-        I: for<'a> FnMut(&<Self::Container as Container>::Item<'a>) -> usize,
+        Self: for<'a> PushInto<<Self::Container as IterableContainer>::Item<'a>>,
+        I: for<'a> FnMut(&<Self::Container as IterableContainer>::Item<'a>) -> usize,
+        Self::Container: IterableContainer,
     {
         for datum in container.drain() {
             let index = index(&datum);
             builders[index].push_into(datum);
         }
-        container.clear();
     }
 
     /// Indicates a good moment to release resources.
@@ -140,7 +128,34 @@ pub trait ContainerBuilder: Default + 'static {
 /// Specifically, the sum of lengths of all extracted and finished containers must equal the
 /// number of times that `push_into` is called on the container builder.
 /// If you have any questions about this trait you are best off not implementing it.
-pub trait LengthPreservingContainerBuilder : ContainerBuilder { }
+pub trait CountPreservingContainerBuilder: 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: WithProgress + Clone + 'static> ContainerBuilder for PassthroughContainerBuilder<C>
+{
+    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.
 ///
@@ -165,7 +180,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() {
@@ -174,7 +189,7 @@ impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuil
     }
 }
 
-impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
+impl<C: WithProgress + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
     type Container = C;
 
     #[inline]
@@ -197,22 +212,11 @@ impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuild
     }
 }
 
-impl<C: Container + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
+impl<C: WithProgress + SizableContainer + Clone + 'static> CountPreservingContainerBuilder for CapacityContainerBuilder<C> { }
 
-impl<T> Container for Vec<T> {
+impl<T> IterableContainer 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;
 
     fn iter(&self) -> Self::Iter<'_> {
@@ -268,29 +272,11 @@ mod rc {
     use std::ops::Deref;
     use std::rc::Rc;
 
-    use crate::Container;
+    use crate::IterableContainer;
 
-    impl<T: Container> Container for Rc<T> {
+    impl<T: IterableContainer> IterableContainer 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) {
-                inner.clear();
-            } else {
-                *self = Self::default();
-            }
-        }
-
         type Iter<'a> = T::Iter<'a> where Self: 'a;
 
         fn iter(&self) -> Self::Iter<'_> {
@@ -309,29 +295,11 @@ mod arc {
     use std::ops::Deref;
     use std::sync::Arc;
 
-    use crate::Container;
+    use crate::IterableContainer;
 
-    impl<T: Container> Container for Arc<T> {
+    impl<T: IterableContainer> IterableContainer 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) {
-                inner.clear();
-            } else {
-                *self = Self::default();
-            }
-        }
-
         type Iter<'a> = T::Iter<'a> where Self: 'a;
 
         fn iter(&self) -> Self::Iter<'_> {
@@ -366,3 +334,18 @@ 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/examples/columnar.rs

@@ -2,7 +2,7 @@
 
 use {
     std::collections::HashMap,
-    timely::{Container, container::CapacityContainerBuilder},
+    timely::container::{CapacityContainerBuilder, IterableContainer},
     timely::dataflow::channels::pact::{ExchangeCore, Pipeline},
     timely::dataflow::InputHandleCore,
     timely::dataflow::operators::{Inspect, Operator, Probe},
@@ -165,17 +165,11 @@ mod container {
         }
     }
 
-    impl<C: columnar::ContainerBytes> timely::Container for Column<C> {
-        fn len(&self) -> usize { self.borrow().len() }
-        // This sets `self` to be an empty `Typed` variant, appropriate for pushing into.
-        fn clear(&mut self) {
-            match self {
-                Column::Typed(t) => t.clear(),
-                Column::Bytes(_) => *self = Column::Typed(Default::default()),
-                Column::Align(_) => *self = Column::Typed(Default::default()),
-            }
-        }
-
+    impl<C: columnar::ContainerBytes> timely::container::WithProgress for Column<C> {
+        #[inline(always)]
+        fn count(&self) -> usize { self.borrow().len() }
+    }
+    impl<C: columnar::ContainerBytes> timely::container::IterableContainer for Column<C> {
         type ItemRef<'a> = C::Ref<'a>;
         type Iter<'a> = IterOwn<C::Borrowed<'a>>;
         fn iter<'a>(&'a self) -> Self::Iter<'a> { self.borrow().into_index_iter() }
@@ -284,7 +278,7 @@ mod builder {
         }
     }
 
-    use timely::container::{ContainerBuilder, LengthPreservingContainerBuilder};
+    use timely::container::{ContainerBuilder, CountPreservingContainerBuilder};
     impl<C: columnar::ContainerBytes> ContainerBuilder for ColumnBuilder<C> {
         type Container = Column<C>;
 
@@ -317,5 +311,5 @@ mod builder {
         }
     }
 
-    impl<C: columnar::ContainerBytes> LengthPreservingContainerBuilder for ColumnBuilder<C> { }
+    impl<C: columnar::ContainerBytes> CountPreservingContainerBuilder for ColumnBuilder<C> { }
 }

timely/src/dataflow/channels/mod.rs

@@ -2,7 +2,7 @@
 
 use serde::{Deserialize, Serialize};
 use crate::communication::Push;
-use crate::Container;
+use crate::container::WithProgress;
 
 /// 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: WithProgress> 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();
         }
     }
 }