Restore clear, cleanup

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

Author: antiguru

container/src/lib.rs

@@ -6,21 +6,25 @@ 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 {
+pub trait Container: 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 [`CountPreservingContainerBuilder`] trait, whose implementors
+    /// imposed on the [`LengthPreservingContainerBuilder`] trait, whose implementors
     /// must preserve the number of items.
     fn count(&self) -> usize;
+
+    /// 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 {
+pub trait IterContainer: Container {
     /// The type of elements when reading non-destructively from the container.
     type ItemRef<'a> where Self: 'a;
 
@@ -42,7 +46,7 @@ pub trait IterableContainer: WithProgress {
 }
 
 /// A container that can be sized and reveals its capacity.
-pub trait SizableContainer: WithProgress {
+pub trait SizableContainer: Container {
     /// 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.
@@ -84,7 +88,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: Container + Clone + 'static;
     /// Extract assembled containers, potentially leaving unfinished data behind. Can
     /// be called repeatedly, for example while the caller can send data.
     ///
@@ -98,14 +102,15 @@ 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 IterableContainer>::Item<'a>>,
-        I: for<'a> FnMut(&<Self::Container as IterableContainer>::Item<'a>) -> usize,
-        Self::Container: IterableContainer,
+        Self: for<'a> PushInto<<Self::Container as IterContainer>::Item<'a>>,
+        I: for<'a> FnMut(&<Self::Container as IterContainer>::Item<'a>) -> usize,
+        Self::Container: IterContainer,
     {
         for datum in container.drain() {
             let index = index(&datum);
             builders[index].push_into(datum);
         }
+        container.clear();
     }
 
     /// Indicates a good moment to release resources.
@@ -122,7 +127,7 @@ 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 CountPreservingContainerBuilder: ContainerBuilder { }
+pub trait LengthPreservingContainerBuilder : ContainerBuilder { }
 
 /// A container builder that never produces any outputs, and can be used to pass through data in
 /// operators.
@@ -136,7 +141,7 @@ impl<C> Default for PassthroughContainerBuilder<C> {
     }
 }
 
-impl<C: WithProgress + Clone + 'static> ContainerBuilder for PassthroughContainerBuilder<C>
+impl<C: Container + Clone + 'static> ContainerBuilder for PassthroughContainerBuilder<C>
 {
     type Container = C;
 
@@ -183,7 +188,7 @@ impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuil
     }
 }
 
-impl<C: WithProgress + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
+impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
     type Container = C;
 
     #[inline]
@@ -206,9 +211,14 @@ impl<C: WithProgress + Clone + 'static> ContainerBuilder for CapacityContainerBu
     }
 }
 
-impl<C: WithProgress + SizableContainer + Clone + 'static> CountPreservingContainerBuilder for CapacityContainerBuilder<C> { }
+impl<C: Container + SizableContainer + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
+
+impl<T> Container for Vec<T> {
+    #[inline(always)] fn count(&self) -> usize { Vec::len(self) }
+    #[inline(always)] fn clear(&mut self) { Vec::clear(self) }
+}
 
-impl<T> IterableContainer for Vec<T> {
+impl<T> IterContainer for Vec<T> {
     type ItemRef<'a> = &'a T where T: 'a;
     type Item<'a> = T where T: 'a;
     type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;
@@ -266,9 +276,22 @@ mod rc {
     use std::ops::Deref;
     use std::rc::Rc;
 
-    use crate::IterableContainer;
+    use crate::{Container, IterContainer};
+
+    impl<T: Container> Container for Rc<T> {
+        #[inline(always)] fn count(&self) -> usize { self.as_ref().count() }
+        #[inline(always)] 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();
+            }
+        }
+    }
+
 
-    impl<T: IterableContainer> IterableContainer for Rc<T> {
+    impl<T: IterContainer> IterContainer for Rc<T> {
         type ItemRef<'a> = T::ItemRef<'a> where Self: 'a;
         type Item<'a> = T::ItemRef<'a> where Self: 'a;
         type Iter<'a> = T::Iter<'a> where Self: 'a;
@@ -289,9 +312,21 @@ mod arc {
     use std::ops::Deref;
     use std::sync::Arc;
 
-    use crate::IterableContainer;
+    use crate::{Container, IterContainer};
+
+    impl<T: Container> Container for std::sync::Arc<T> {
+        #[inline(always)] fn count(&self) -> usize { self.as_ref().count() }
+        #[inline(always)] 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();
+            }
+        }
+    }
 
-    impl<T: IterableContainer> IterableContainer for Arc<T> {
+    impl<T: IterContainer> IterContainer for Arc<T> {
         type ItemRef<'a> = T::ItemRef<'a> where Self: 'a;
         type Item<'a> = T::ItemRef<'a> where Self: 'a;
         type Iter<'a> = T::Iter<'a> where Self: 'a;
@@ -328,15 +363,3 @@ pub mod buffer {
         }
     }
 }
-
-impl<T> WithProgress for Vec<T> {
-    #[inline(always)] fn count(&self) -> usize { self.len() }
-}
-
-impl<T: WithProgress> WithProgress for std::rc::Rc<T> {
-    #[inline(always)] fn count(&self) -> usize { self.as_ref().count() }
-}
-
-impl<T: WithProgress> WithProgress for std::sync::Arc<T> {
-    #[inline(always)] fn count(&self) -> usize { self.as_ref().count() }
-}

timely/examples/columnar.rs

@@ -2,7 +2,7 @@
 
 use {
     std::collections::HashMap,
-    timely::container::{CapacityContainerBuilder, IterableContainer},
+    timely::container::{CapacityContainerBuilder, IterContainer},
     timely::dataflow::channels::pact::{ExchangeCore, Pipeline},
     timely::dataflow::InputHandleCore,
     timely::dataflow::operators::{Inspect, Operator, Probe},
@@ -165,11 +165,19 @@ mod container {
         }
     }
 
-    impl<C: columnar::ContainerBytes> timely::container::WithProgress for Column<C> {
+    impl<C: columnar::ContainerBytes> timely::container::Container for Column<C> {
+        #[inline(always)] fn count(&self) -> usize { self.borrow().len() }
+        // This sets `self` to be an empty `Typed` variant, appropriate for pushing into.
         #[inline(always)]
-        fn count(&self) -> usize { self.borrow().len() }
+        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::IterableContainer for Column<C> {
+    impl<C: columnar::ContainerBytes> timely::container::IterContainer 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() }
@@ -278,7 +286,7 @@ mod builder {
         }
     }
 
-    use timely::container::{ContainerBuilder, CountPreservingContainerBuilder};
+    use timely::container::{ContainerBuilder, LengthPreservingContainerBuilder};
     impl<C: columnar::ContainerBytes> ContainerBuilder for ColumnBuilder<C> {
         type Container = Column<C>;
 
@@ -311,5 +319,5 @@ mod builder {
         }
     }
 
-    impl<C: columnar::ContainerBytes> CountPreservingContainerBuilder for ColumnBuilder<C> { }
+    impl<C: columnar::ContainerBytes> LengthPreservingContainerBuilder for ColumnBuilder<C> { }
 }

timely/src/dataflow/channels/mod.rs

@@ -2,7 +2,7 @@
 
 use serde::{Deserialize, Serialize};
 use crate::communication::Push;
-use crate::container::WithProgress;
+use crate::Container;
 
 /// A collection of types that may be pushed at.
 pub mod pushers;
@@ -32,15 +32,14 @@ impl<T, C> Message<T, C> {
     }
 }
 
-impl<T, C: WithProgress> Message<T, C> {
+impl<T, C: Container> 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's contents are left in an
-    /// undefined state, specifically the caller cannot rely on this function clearing the buffer.
+    /// leaves in place, or the container's default element. The buffer is cleared.
     #[inline]
     pub fn push_at<P: Push<Message<T, C>>>(buffer: &mut C, time: T, pusher: &mut P) {
 
@@ -52,6 +51,7 @@ impl<T, C: WithProgress> 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::{ContainerBuilder, CountPreservingContainerBuilder, IterableContainer, WithProgress, SizableContainer, CapacityContainerBuilder, PushInto};
+use crate::container::{ContainerBuilder, LengthPreservingContainerBuilder, IterContainer, Container, 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: WithProgress + 'static> ParallelizationContract<T, C> for Pipeline {
+impl<T: 'static, C: Container + '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,8 +52,8 @@ pub type Exchange<D, F> = ExchangeCore<CapacityContainerBuilder<Vec<D>>, F>;
 
 impl<CB, F> ExchangeCore<CB, F>
 where
-    CB: CountPreservingContainerBuilder<Container: IterableContainer>,
-    for<'a> F: FnMut(&<CB::Container as IterableContainer>::Item<'a>)->u64
+    CB: LengthPreservingContainerBuilder<Container: IterContainer>,
+    for<'a> F: FnMut(&<CB::Container as IterContainer>::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
     pub fn new_core(func: F) -> ExchangeCore<CB, F> {
@@ -66,7 +66,7 @@ where
 
 impl<C, F> ExchangeCore<CapacityContainerBuilder<C>, F>
 where
-    C: SizableContainer + IterableContainer,
+    C: SizableContainer + IterContainer,
     for<'a> F: FnMut(&C::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
@@ -81,10 +81,10 @@ 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<Container: IterableContainer>,
-    CB: for<'a> PushInto<<CB::Container as IterableContainer>::Item<'a>>,
+    CB: ContainerBuilder<Container: IterContainer>,
+    CB: for<'a> PushInto<<CB::Container as IterContainer>::Item<'a>>,
     CB::Container: Data + Send + crate::dataflow::channels::ContainerBytes,
-    for<'a> H: FnMut(&<CB::Container as IterableContainer>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as IterContainer>::Item<'a>) -> u64
 {
     type Pusher = ExchangePusher<T, CB, LogPusher<T, CB::Container, Box<dyn Push<Message<T, CB::Container>>>>, H>;
     type Puller = LogPuller<T, CB::Container, Box<dyn Pull<Message<T, CB::Container>>>>;
@@ -129,7 +129,7 @@ impl<T, C, P: Push<Message<T, C>>> LogPusher<T, C, P> {
     }
 }
 
-impl<T, C: WithProgress, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<T, C, P> {
+impl<T, C: Container, 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: WithProgress, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<T, C, P> {
+impl<T, C: Container, 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::WithProgress;
+use crate::Container;
 
 /// 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: WithProgress, P: Pull<Message<T, C>>> Counter<T, C, P> {
+impl<T:Ord+Clone+'static, C: Container, 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,11 +2,11 @@
 //! with the performance of batched sends.
 
 use crate::communication::Push;
-use crate::container::{ContainerBuilder, CapacityContainerBuilder, WithProgress, PushInto, SizableContainer};
+use crate::container::{ContainerBuilder, CapacityContainerBuilder, PushInto};
 use crate::dataflow::channels::Message;
 use crate::dataflow::operators::Capability;
 use crate::progress::Timestamp;
-use crate::Data;
+use crate::{Container, Data};
 
 /// Buffers data sent at the same time, for efficient communication.
 ///
@@ -44,7 +44,7 @@ impl<T, CB: Default, P> Buffer<T, CB, P> {
     }
 }
 
-impl<T, C: SizableContainer + Data, P: Push<Message<T, C>>> Buffer<T, CapacityContainerBuilder<C>, P> where T: Eq+Clone {
+impl<T, C: Container + Data, P: Push<Message<T, C>>> Buffer<T, CapacityContainerBuilder<C>, P> where T: Eq+Clone {
     /// Returns a `Session`, which accepts data to send at the associated time
     #[inline]
     pub fn session(&mut self, time: &T) -> Session<'_, T, CapacityContainerBuilder<C>, P> {

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::WithProgress;
+use crate::Container;
 
 /// 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: WithProgress, P> Push<Message<T, C>> for Counter<T, C, P> where P: Push<Message<T, C>> {
+impl<T: Timestamp, C: Container, 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 {