Remove Container: Clone + 'static (#540)

Remove the requirement that all Container implementations are Clone and
'static. This makes implementing types simpler that depend on Container,
and requires us to explicitly mark various places as `Data` such that they
comply with Timely's type requirements.

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

Author: antiguru

container/src/columnation.rs

@@ -339,7 +339,7 @@ mod container {
 
     use crate::columnation::{Columnation, TimelyStack};
 
-    impl<T: Columnation + 'static> Container for TimelyStack<T> {
+    impl<T: Columnation> Container for TimelyStack<T> {
         type ItemRef<'a> = &'a T where Self: 'a;
         type Item<'a> = &'a T where Self: 'a;
 
@@ -355,20 +355,20 @@ mod container {
             TimelyStack::clear(self)
         }
 
-        type Iter<'a> = std::slice::Iter<'a, T>;
+        type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;
 
         fn iter(&self) -> Self::Iter<'_> {
             self.deref().iter()
         }
 
-        type DrainIter<'a> = std::slice::Iter<'a, T>;
+        type DrainIter<'a> = std::slice::Iter<'a, T> where Self: 'a;
 
         fn drain(&mut self) -> Self::DrainIter<'_> {
             (*self).iter()
         }
     }
 
-    impl<T: Columnation + 'static> SizableContainer for TimelyStack<T> {
+    impl<T: Columnation> SizableContainer for TimelyStack<T> {
         fn capacity(&self) -> usize {
             self.capacity()
         }

container/src/flatcontainer.rs

@@ -3,7 +3,7 @@
 pub use flatcontainer::*;
 use crate::{buffer, Container, SizableContainer, PushInto};
 
-impl<R: Region + Clone + 'static> Container for FlatStack<R> {
+impl<R: Region> Container for FlatStack<R> {
     type ItemRef<'a> = R::ReadItem<'a>  where Self: 'a;
     type Item<'a> = R::ReadItem<'a> where Self: 'a;
 
@@ -15,20 +15,20 @@ impl<R: Region + Clone + 'static> Container for FlatStack<R> {
         self.clear()
     }
 
-    type Iter<'a> = <&'a Self as IntoIterator>::IntoIter;
+    type Iter<'a> = <&'a Self as IntoIterator>::IntoIter where Self: 'a;
 
     fn iter(&self) -> Self::Iter<'_> {
         IntoIterator::into_iter(self)
     }
 
-    type DrainIter<'a> = Self::Iter<'a>;
+    type DrainIter<'a> = Self::Iter<'a> where Self: 'a;
 
     fn drain(&mut self) -> Self::DrainIter<'_> {
         IntoIterator::into_iter(&*self)
     }
 }
 
-impl<R: Region + Clone + 'static> SizableContainer for FlatStack<R> {
+impl<R: Region> SizableContainer for FlatStack<R> {
     fn capacity(&self) -> usize {
         self.capacity()
     }

container/src/lib.rs

@@ -18,8 +18,7 @@ pub mod flatcontainer;
 /// 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`.)
-/// TODO: Don't require `Container: Clone`
-pub trait Container: Default + Clone + 'static {
+pub trait Container: Default {
     /// The type of elements when reading non-destructively from the container.
     type ItemRef<'a> where Self: 'a;
 
@@ -50,13 +49,13 @@ pub trait Container: Default + Clone + 'static {
     fn clear(&mut self);
 
     /// Iterator type when reading from the container.
-    type Iter<'a>: Iterator<Item=Self::ItemRef<'a>>;
+    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<'_>;
 
     /// Iterator type when draining the container.
-    type DrainIter<'a>: Iterator<Item=Self::Item<'a>>;
+    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.
@@ -104,7 +103,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;
+    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.
     ///
@@ -160,7 +159,7 @@ impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuil
     }
 }
 
-impl<C: Container> ContainerBuilder for CapacityContainerBuilder<C> {
+impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
     type Container = C;
 
     #[inline]
@@ -204,7 +203,7 @@ impl<C: Container> CapacityContainerBuilder<C> {
     }
 }
 
-impl<T: Clone + 'static> Container for Vec<T> {
+impl<T> Container for Vec<T> {
     type ItemRef<'a> = &'a T where T: 'a;
     type Item<'a> = T where T: 'a;
 
@@ -218,20 +217,20 @@ impl<T: Clone + 'static> Container for Vec<T> {
 
     fn clear(&mut self) { Vec::clear(self) }
 
-    type Iter<'a> = std::slice::Iter<'a, T>;
+    type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;
 
     fn iter(&self) -> Self::Iter<'_> {
         self.as_slice().iter()
     }
 
-    type DrainIter<'a> = std::vec::Drain<'a, T>;
+    type DrainIter<'a> = std::vec::Drain<'a, T> where Self: 'a;
 
     fn drain(&mut self) -> Self::DrainIter<'_> {
         self.drain(..)
     }
 }
 
-impl<T: Clone + 'static> SizableContainer for Vec<T> {
+impl<T> SizableContainer for Vec<T> {
     fn capacity(&self) -> usize {
         self.capacity()
     }
@@ -294,13 +293,13 @@ mod rc {
             }
         }
 
-        type Iter<'a> = T::Iter<'a>;
+        type Iter<'a> = T::Iter<'a> where Self: 'a;
 
         fn iter(&self) -> Self::Iter<'_> {
             self.deref().iter()
         }
 
-        type DrainIter<'a> = T::Iter<'a>;
+        type DrainIter<'a> = T::Iter<'a> where Self: 'a;
 
         fn drain(&mut self) -> Self::DrainIter<'_> {
             self.iter()
@@ -335,13 +334,13 @@ mod arc {
             }
         }
 
-        type Iter<'a> = T::Iter<'a>;
+        type Iter<'a> = T::Iter<'a> where Self: 'a;
 
         fn iter(&self) -> Self::Iter<'_> {
             self.deref().iter()
         }
 
-        type DrainIter<'a> = T::Iter<'a>;
+        type DrainIter<'a> = T::Iter<'a> where Self: 'a;
 
         fn drain(&mut self) -> Self::DrainIter<'_> {
             self.iter()

timely/src/dataflow/channels/pact.rs

@@ -35,7 +35,7 @@ pub trait ParallelizationContract<T, C> {
 #[derive(Debug)]
 pub struct Pipeline;
 
-impl<T: 'static, C: Container> 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) {

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

@@ -6,7 +6,7 @@ use crate::container::{ContainerBuilder, CapacityContainerBuilder, PushInto};
 use crate::dataflow::channels::Message;
 use crate::dataflow::operators::Capability;
 use crate::progress::Timestamp;
-use crate::Container;
+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: Container, 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> {
@@ -133,7 +133,7 @@ pub struct Session<'a, T, CB, P> {
     buffer: &'a mut Buffer<T, CB, P>,
 }
 
-impl<'a, T, C: Container, P> Session<'a, T, CapacityContainerBuilder<C>, P>
+impl<'a, T, C: Container + Data, P> Session<'a, T, CapacityContainerBuilder<C>, P>
 where
     T: Eq + Clone + 'a,
     P: Push<Message<T, C>> + 'a,

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

@@ -17,7 +17,7 @@ pub struct Tee<T, C> {
     shared: PushList<T, C>,
 }
 
-impl<T: Data, C: Container> Push<Message<T, C>> for Tee<T, C> {
+impl<T: Data, C: Container + Data> Push<Message<T, C>> for Tee<T, C> {
     #[inline]
     fn push(&mut self, message: &mut Option<Message<T, C>>) {
         let mut pushers = self.shared.borrow_mut();

timely/src/dataflow/operators/branch.rs

@@ -92,7 +92,7 @@ pub trait BranchWhen<T>: Sized {
     fn branch_when(&self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self);
 }
 
-impl<S: Scope, C: Container> BranchWhen<S::Timestamp> for StreamCore<S, C> {
+impl<S: Scope, C: Container + Data> BranchWhen<S::Timestamp> for StreamCore<S, C> {
     fn branch_when(&self, condition: impl Fn(&S::Timestamp) -> bool + 'static) -> (Self, Self) {
         let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());
 

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

@@ -10,14 +10,14 @@ use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::channels::pullers::Counter as PullCounter;
 use crate::dataflow::operators::generic::builder_raw::OperatorBuilder;
 
-use crate::Container;
+use crate::{Container, Data};
 use crate::progress::ChangeBatch;
 use crate::progress::Timestamp;
 
 use super::{Event, EventPusher};
 
 /// Capture a stream of timestamped data for later replay.
-pub trait Capture<T: Timestamp, C: Container> {
+pub trait Capture<T: Timestamp, C: Container + Data> {
     /// Captures a stream of timestamped data for later replay.
     ///
     /// # Examples
@@ -113,7 +113,7 @@ pub trait Capture<T: Timestamp, C: Container> {
     }
 }
 
-impl<S: Scope, C: Container> Capture<S::Timestamp, C> for StreamCore<S, C> {
+impl<S: Scope, C: Container + Data> Capture<S::Timestamp, C> for StreamCore<S, C> {
     fn capture_into<P: EventPusher<S::Timestamp, C>+'static>(&self, mut event_pusher: P) {
 
         let mut builder = OperatorBuilder::new("Capture".to_owned(), self.scope());

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

@@ -62,7 +62,7 @@ pub trait Replay<T: Timestamp, C> : Sized {
     fn replay_core<S: Scope<Timestamp=T>>(self, scope: &mut S, period: Option<std::time::Duration>) -> StreamCore<S, C>;
 }
 
-impl<T: Timestamp, C: Container, I> Replay<T, C> for I
+impl<T: Timestamp, C: Container + Clone + 'static, I> Replay<T, C> for I
 where
     I : IntoIterator,
     <I as IntoIterator>::Item: EventIterator<T, C>+'static,

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

@@ -1,7 +1,7 @@
 //! Merges the contents of multiple streams.
 
 
-use crate::Container;
+use crate::{Container, Data};
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{StreamCore, Scope};
 
@@ -23,7 +23,7 @@ pub trait Concat<G: Scope, C: Container> {
     fn concat(&self, _: &StreamCore<G, C>) -> StreamCore<G, C>;
 }
 
-impl<G: Scope, C: Container> Concat<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container + Data> Concat<G, C> for StreamCore<G, C> {
     fn concat(&self, other: &StreamCore<G, C>) -> StreamCore<G, C> {
         self.scope().concatenate([self.clone(), other.clone()])
     }
@@ -52,7 +52,7 @@ pub trait Concatenate<G: Scope, C: Container> {
         I: IntoIterator<Item=StreamCore<G, C>>;
 }
 
-impl<G: Scope, C: Container> Concatenate<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container + Data> Concatenate<G, C> for StreamCore<G, C> {
     fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
     where
         I: IntoIterator<Item=StreamCore<G, C>>
@@ -62,7 +62,7 @@ impl<G: Scope, C: Container> Concatenate<G, C> for StreamCore<G, C> {
     }
 }
 
-impl<G: Scope, C: Container> Concatenate<G, C> for G {
+impl<G: Scope, C: Container + Data> Concatenate<G, C> for G {
     fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
     where
         I: IntoIterator<Item=StreamCore<G, C>>