Split Container into Container, IterContainer, DrainContainer

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

Author: antiguru

container/src/lib.rs

@@ -20,12 +20,6 @@ use std::collections::VecDeque;
 // We can only access the type if all requirements for the `ContainerBuilder` implementation are
 // satisfied.
 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;
-
     /// The number of elements in this container
     ///
     /// This number is used in progress tracking to confirm the receipt of some number
@@ -38,16 +32,24 @@ pub trait Container: Default {
     fn is_empty(&self) -> bool {
         self.len() == 0
     }
+}
 
+/// TODO
+pub trait IterContainer {
+    /// The type of elements when reading non-destructively from the container.
+    type ItemRef<'a> where Self: 'a;
     /// Iterator type when reading from the container.
     type Iter<'a>: Iterator<Item=Self::ItemRef<'a>> where Self: 'a;
-
     /// Returns an iterator that reads the contents of this container.
     fn iter(&self) -> Self::Iter<'_>;
+}
 
+/// TODO
+pub trait DrainContainer {
+    /// The type of elements when draining the container.
+    type Item<'a> where Self: 'a;
     /// Iterator type when draining the container.
     type DrainIter<'a>: Iterator<Item=Self::Item<'a>> where Self: 'a;
-
     /// Returns an iterator that drains the contents of this container.
     /// Drain leaves the container in an undefined state.
     fn drain(&mut self) -> Self::DrainIter<'_>;
@@ -112,8 +114,9 @@ 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::Container: DrainContainer,
+        Self: for<'a> PushInto<<Self::Container as DrainContainer>::Item<'a>>,
+        I: for<'a> FnMut(&<Self::Container as DrainContainer>::Item<'a>) -> usize,
     {
         for datum in container.drain() {
             let index = index(&datum);
@@ -195,25 +198,25 @@ impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuild
 impl<C: Container + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
 
 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)
     }
+}
 
+impl<T> IterContainer for Vec<T> {
+    type ItemRef<'a> = &'a T where T: 'a;
     type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;
-
     fn iter(&self) -> Self::Iter<'_> {
         self.as_slice().iter()
     }
+}
 
+impl<T> DrainContainer for Vec<T> {
+    type Item<'a> = T where T: 'a;
     type DrainIter<'a> = std::vec::Drain<'a, T> where Self: 'a;
-
     fn drain(&mut self) -> Self::DrainIter<'_> {
         self.drain(..)
     }
@@ -261,63 +264,47 @@ mod rc {
     use std::ops::Deref;
     use std::rc::Rc;
 
-    use crate::Container;
+    use crate::{Container, IterContainer, DrainContainer};
 
     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()
         }
-
+    }
+    impl<T: IterContainer> IterContainer for Rc<T> {
+        type ItemRef<'a> = T::ItemRef<'a> where Self: 'a;
         type Iter<'a> = T::Iter<'a> where Self: 'a;
-
-        fn iter(&self) -> Self::Iter<'_> {
-            self.deref().iter()
-        }
-
+        fn iter(&self) -> Self::Iter<'_> { self.deref().iter() }
+    }
+    impl<T: IterContainer> DrainContainer for Rc<T> {
+        type Item<'a> = T::ItemRef<'a> where Self: 'a;
         type DrainIter<'a> = T::Iter<'a> where Self: 'a;
-
-        fn drain(&mut self) -> Self::DrainIter<'_> {
-            self.iter()
-        }
+        fn drain(&mut self) -> Self::DrainIter<'_> { self.iter() }
     }
 }
 
 mod arc {
     use std::ops::Deref;
     use std::sync::Arc;
 
-    use crate::Container;
+    use crate::{Container, IterContainer, DrainContainer};
 
     impl<T: Container> Container for Arc<T> {
+        fn len(&self) -> usize { std::ops::Deref::deref(self).len() }
+        fn is_empty(&self) -> bool { std::ops::Deref::deref(self).is_empty() }
+    }
+    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;
-
-        fn len(&self) -> usize {
-            std::ops::Deref::deref(self).len()
-        }
-
-        fn is_empty(&self) -> bool {
-            std::ops::Deref::deref(self).is_empty()
-        }
-
         type Iter<'a> = T::Iter<'a> where Self: 'a;
-
-        fn iter(&self) -> Self::Iter<'_> {
-            self.deref().iter()
-        }
-
+        fn iter(&self) -> Self::Iter<'_> { self.deref().iter() }
+    }
+    impl<T: IterContainer> DrainContainer for Arc<T> {
+        type Item<'a> = T::ItemRef<'a> where Self: 'a;
         type DrainIter<'a> = T::Iter<'a> where Self: 'a;
-
-        fn drain(&mut self) -> Self::DrainIter<'_> {
-            self.iter()
-        }
+        fn drain(&mut self) -> Self::DrainIter<'_> { self.iter() }
     }
 }
 

timely/examples/columnar.rs

@@ -1,13 +1,12 @@
 //! Wordcount based on the `columnar` crate.
 
-use {
-    std::collections::HashMap,
-    timely::{Container, container::CapacityContainerBuilder},
-    timely::dataflow::channels::pact::{ExchangeCore, Pipeline},
-    timely::dataflow::InputHandleCore,
-    timely::dataflow::operators::{Inspect, Operator, Probe},
-    timely::dataflow::ProbeHandle,
-};
+use std::collections::HashMap;
+
+use timely::container::{IterContainer, CapacityContainerBuilder};
+use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
+use timely::dataflow::InputHandleCore;
+use timely::dataflow::operators::{Inspect, Operator, Probe};
+use timely::dataflow::ProbeHandle;
 
 // Creates `WordCountContainer` and `WordCountReference` structs,
 // as well as various implementations relating them to `WordCount`.
@@ -179,10 +178,13 @@ mod container {
 
     impl<C: columnar::ContainerBytes> timely::Container for Column<C> {
         fn len(&self) -> usize { self.borrow().len() }
+    }
+    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() }
-
+    }
+    impl<C: columnar::ContainerBytes> timely::container::DrainContainer for Column<C> {
         type Item<'a> = C::Ref<'a>;
         type DrainIter<'a> = IterOwn<C::Borrowed<'a>>;
         fn drain<'a>(&'a mut self) -> Self::DrainIter<'a> { self.borrow().into_index_iter() }

timely/src/dataflow/channels/pact.rs

@@ -10,7 +10,8 @@
 use std::{fmt::{self, Debug}, marker::PhantomData};
 use std::rc::Rc;
 
-use crate::{Container, container::{ContainerBuilder, LengthPreservingContainerBuilder, SizableContainer, CapacityContainerBuilder, PushInto}};
+use crate::Container;
+use crate::container::{ContainerBuilder, DrainContainer, LengthPreservingContainerBuilder, SizableContainer, CapacityContainerBuilder, PushInto};
 use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
 use crate::communication::{Push, Pull};
 use crate::dataflow::channels::pushers::Exchange as ExchangePusher;
@@ -53,7 +54,8 @@ pub type Exchange<D, F> = ExchangeCore<CapacityContainerBuilder<Vec<D>>, F>;
 impl<CB, F> ExchangeCore<CB, F>
 where
     CB: LengthPreservingContainerBuilder,
-    for<'a> F: FnMut(&<CB::Container as Container>::Item<'a>)->u64
+    CB::Container: DrainContainer,
+    for<'a> F: FnMut(&<CB::Container as DrainContainer>::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
     pub fn new_core(func: F) -> ExchangeCore<CB, F> {
@@ -66,7 +68,7 @@ where
 
 impl<C, F> ExchangeCore<CapacityContainerBuilder<C>, F>
 where
-    C: SizableContainer,
+    C: SizableContainer + DrainContainer,
     for<'a> F: FnMut(&C::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
@@ -82,9 +84,10 @@ where
 impl<T: Timestamp, CB, H: 'static> ParallelizationContract<T, CB::Container> for ExchangeCore<CB, H>
 where
     CB: ContainerBuilder,
-    CB: for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
+    CB::Container: DrainContainer,
+    CB: for<'a> PushInto<<CB::Container as DrainContainer>::Item<'a>>,
     CB::Container: Data + Send + crate::dataflow::channels::ContainerBytes,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as DrainContainer>::Item<'a>) -> u64
 {
     type Pusher = ExchangePusher<T, CB, LogPusher<Box<dyn Push<Message<T, CB::Container>>>>, H>;
     type Puller = LogPuller<Box<dyn Pull<Message<T, CB::Container>>>>;

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

@@ -3,15 +3,17 @@
 use crate::communication::Push;
 use crate::container::{ContainerBuilder, PushInto};
 use crate::dataflow::channels::Message;
-use crate::{Container, Data};
+use crate::Data;
+use crate::container::DrainContainer;
 
 // TODO : Software write combining
 /// Distributes records among target pushees according to a distribution function.
 pub struct Exchange<T, CB, P, H>
 where
     CB: ContainerBuilder,
+    CB::Container: DrainContainer,
     P: Push<Message<T, CB::Container>>,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as DrainContainer>::Item<'a>) -> u64
 {
     pushers: Vec<P>,
     builders: Vec<CB>,
@@ -22,8 +24,9 @@ where
 impl<T: Clone, CB, P, H>  Exchange<T, CB, P, H>
 where
     CB: ContainerBuilder,
+    CB::Container: DrainContainer,
     P: Push<Message<T, CB::Container>>,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as DrainContainer>::Item<'a>) -> u64
 {
     /// Allocates a new `Exchange` from a supplied set of pushers and a distribution function.
     pub fn new(pushers: Vec<P>, key: H) -> Exchange<T, CB, P, H> {
@@ -48,9 +51,10 @@ where
 impl<T: Eq+Data, CB, P, H> Push<Message<T, CB::Container>> for Exchange<T, CB, P, H>
 where
     CB: ContainerBuilder,
-    CB: for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
+    CB::Container: DrainContainer,
+    CB: for<'a> PushInto<<CB::Container as DrainContainer>::Item<'a>>,
     P: Push<Message<T, CB::Container>>,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as DrainContainer>::Item<'a>) -> u64
 {
     #[inline(never)]
     fn push(&mut self, message: &mut Option<Message<T, CB::Container>>) {

timely/src/dataflow/operators/core/capture/extract.rs

@@ -1,7 +1,7 @@
 //! Traits and types for extracting captured timely dataflow streams.
 
 use super::Event;
-use crate::{container::{SizableContainer, PushInto}};
+use crate::container::{SizableContainer, DrainContainer, PushInto};
 
 /// Supports extracting a sequence of timestamp and data.
 pub trait Extract<T, C> {
@@ -48,10 +48,10 @@ pub trait Extract<T, C> {
     fn extract(self) -> Vec<(T, C)>;
 }
 
-impl<T: Ord, C: SizableContainer> Extract<T, C> for ::std::sync::mpsc::Receiver<Event<T, C>>
+impl<T, C> Extract<T, C> for ::std::sync::mpsc::Receiver<Event<T, C>>
 where
-    for<'a> C: PushInto<C::Item<'a>>,
-    for<'a> C::Item<'a>: Ord,
+    for<'a> C: SizableContainer + DrainContainer<Item<'a>: Ord> + PushInto<C::Item<'a>>,
+    T: Ord,
 {
     fn extract(self) -> Vec<(T, C)> {
         let mut staged = std::collections::BTreeMap::new();

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

@@ -1,13 +1,13 @@
 //! Exchange records between workers.
 
 use crate::ExchangeData;
-use crate::container::{Container, SizableContainer, PushInto};
+use crate::container::{DrainContainer, SizableContainer, PushInto};
 use crate::dataflow::channels::pact::ExchangeCore;
 use crate::dataflow::operators::generic::operator::Operator;
 use crate::dataflow::{Scope, StreamCore};
 
 /// Exchange records between workers.
-pub trait Exchange<C: Container> {
+pub trait Exchange<C: DrainContainer> {
     /// Exchange records between workers.
     ///
     /// The closure supplied should map a reference to a record to a `u64`,
@@ -30,9 +30,11 @@ pub trait Exchange<C: Container> {
 
 impl<G: Scope, C> Exchange<C> for StreamCore<G, C>
 where
-    C: SizableContainer + ExchangeData + crate::dataflow::channels::ContainerBytes,
-    C: for<'a> PushInto<C::Item<'a>>,
-
+    C: SizableContainer
+        + DrainContainer
+        + ExchangeData
+        + crate::dataflow::channels::ContainerBytes
+        + for<'a> PushInto<C::Item<'a>>,
 {
     fn exchange<F>(&self, route: F) -> StreamCore<G, C>
     where

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

@@ -1,12 +1,12 @@
 //! Filters a stream by a predicate.
-use crate::container::{Container, SizableContainer, PushInto};
+use crate::container::{DrainContainer, SizableContainer, PushInto};
 use crate::Data;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::operators::generic::operator::Operator;
 
 /// Extension trait for filtering.
-pub trait Filter<C: Container> {
+pub trait Filter<C: DrainContainer> {
     /// Returns a new instance of `self` containing only records satisfying `predicate`.
     ///
     /// # Examples
@@ -23,7 +23,7 @@ pub trait Filter<C: Container> {
     fn filter<P: FnMut(&C::Item<'_>)->bool+'static>(&self, predicate: P) -> Self;
 }
 
-impl<G: Scope, C: SizableContainer + Data> Filter<C> for StreamCore<G, C>
+impl<G: Scope, C: SizableContainer + DrainContainer + Data> Filter<C> for StreamCore<G, C>
 where
     for<'a> C: PushInto<C::Item<'a>>
 {