Rename Container to WithProgress

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

Author: antiguru

container/src/lib.rs

@@ -4,28 +4,29 @@
 
 use std::collections::VecDeque;
 
-/// A container transferring data through dataflow edges
+/// An object with effects on progress
 ///
-/// A container stores a number of updates and thus is able to describe it count
+/// The object stores a number of updates and thus is able to describe it count
 /// (`update_count()`) and whether it is empty (`is_empty()`). It is empty if the
 /// update count is zero.
 ///
-/// A container must implement default. The default implementation is not required to allocate
-/// memory for variable-length components.
+/// It must implement default for historic reason. The default implementation is not required
+/// to allocate memory for variable-length components.
+// TODO: Remove `Default` requirement in the future.
 // The container is `Default` because `CapacityContainerBuilder` only implements `ContainerBuilder`
 // for containers that implement `Default`, and we use the associated `::Container` all over Timely.
 // We can only access the type if all requirements for the `ContainerBuilder` implementation are
 // satisfied.
-pub trait Container: Default {
-    /// The number of updates in this container
+pub trait WithProgress: Default {
+    /// The number of updates
     ///
     /// This number is used in progress tracking to confirm the receipt of some number
     /// of outstanding updates, and it is highly load bearing. The main restriction is
     /// imposed on the `LengthPreservingContainerBuilder` trait, whose implementors
     /// must preserve the number of items.
     fn update_count(&self) -> i64;
 
-    /// Determine if the container contains any updates, corresponding to `update_count() == 0`.
+    /// Determine if this contains any updates, corresponding to `update_count() == 0`.
     #[inline] fn is_empty(&self) -> bool { self.update_count() == 0 }
 }
 
@@ -51,7 +52,7 @@ pub trait DrainContainer {
 }
 
 /// 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.
@@ -95,7 +96,7 @@ pub trait ContainerBuilder: Default + 'static {
     /// The container type we're building.
     // The container is `Clone` because `Tee` requires it, otherwise we need to repeat it
     // all over Timely. `'static` because we don't want lifetimes everywhere.
-    type Container: Container + Clone + 'static;
+    type Container: WithProgress + Clone + 'static;
     /// Extract assembled containers, potentially leaving unfinished data behind. Can
     /// be called repeatedly, for example while the caller can send data.
     ///
@@ -167,7 +168,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]
@@ -190,9 +191,9 @@ impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuild
     }
 }
 
-impl<C: Container + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
+impl<C: WithProgress + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
 
-impl<T> Container for Vec<T> {
+impl<T> WithProgress for Vec<T> {
     #[inline] fn update_count(&self) -> i64 { i64::try_from(Vec::len(self)).unwrap() }
     #[inline] fn is_empty(&self) -> bool { Vec::is_empty(self) }
 }
@@ -255,9 +256,9 @@ mod rc {
     use std::ops::Deref;
     use std::rc::Rc;
 
-    use crate::{Container, IterContainer, DrainContainer};
+    use crate::{WithProgress, IterContainer, DrainContainer};
 
-    impl<T: Container> Container for Rc<T> {
+    impl<T: WithProgress> WithProgress for Rc<T> {
         #[inline] fn update_count(&self) -> i64 { std::ops::Deref::deref(self).update_count() }
         #[inline] fn is_empty(&self) -> bool { std::ops::Deref::deref(self).is_empty() }
     }
@@ -277,9 +278,9 @@ mod arc {
     use std::ops::Deref;
     use std::sync::Arc;
 
-    use crate::{Container, IterContainer, DrainContainer};
+    use crate::{WithProgress, IterContainer, DrainContainer};
 
-    impl<T: Container> Container for Arc<T> {
+    impl<T: WithProgress> WithProgress for Arc<T> {
         #[inline] fn update_count(&self) -> i64 { std::ops::Deref::deref(self).update_count() }
         #[inline] fn is_empty(&self) -> bool { std::ops::Deref::deref(self).is_empty() }
     }

timely/examples/columnar.rs

@@ -176,7 +176,7 @@ mod container {
         }
     }
 
-    impl<C: columnar::ContainerBytes> timely::Container for Column<C> {
+    impl<C: columnar::ContainerBytes> timely::WithProgress for Column<C> {
         #[inline] fn update_count(&self) -> i64 { i64::try_from(self.borrow().len()).unwrap() }
         #[inline] fn is_empty(&self) -> bool { self.borrow().is_empty() }
     }

timely/src/dataflow/channels/mod.rs

@@ -2,7 +2,7 @@
 
 use serde::{Deserialize, Serialize};
 use crate::communication::Push;
-use crate::Container;
+use crate::WithProgress;
 
 /// A collection of types that may be pushed at.
 pub mod pushers;
@@ -32,7 +32,7 @@ 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 }

timely/src/dataflow/channels/pact.rs

@@ -10,7 +10,7 @@
 use std::{fmt::{self, Debug}, marker::PhantomData};
 use std::rc::Rc;
 
-use crate::Container;
+use crate::WithProgress;
 use crate::container::{ContainerBuilder, DrainContainer, LengthPreservingContainerBuilder, SizableContainer, CapacityContainerBuilder, PushInto};
 use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
 use crate::communication::{Push, Pull};
@@ -35,7 +35,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: WithProgress + 'static> ParallelizationContract<T, C> for Pipeline {
     type Pusher = LogPusher<ThreadPusher<Message<T, C>>>;
     type Puller = LogPuller<ThreadPuller<Message<T, C>>>;
     fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: Rc<[usize]>, logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
@@ -130,7 +130,7 @@ impl<P> LogPusher<P> {
     }
 }
 
-impl<T, C: Container, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<P> {
+impl<T, C: WithProgress, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<P> {
     #[inline]
     fn push(&mut self, pair: &mut Option<Message<T, C>>) {
         if let Some(bundle) = pair {
@@ -178,7 +178,7 @@ impl<P> LogPuller<P> {
     }
 }
 
-impl<T, C: Container, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<P> {
+impl<T, C: WithProgress, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<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::WithProgress;
 
 /// 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: WithProgress, 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

@@ -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, Data};
+use crate::{WithProgress, 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 + Data, P: Push<Message<T, C>>> Buffer<T, CapacityContainerBuilder<C>, P> where T: Eq+Clone {
+impl<T, C: WithProgress + 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;
+use crate::WithProgress;
 
 /// 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: WithProgress, 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/tee.rs

@@ -7,7 +7,7 @@ use std::rc::Rc;
 use crate::dataflow::channels::Message;
 
 use crate::communication::Push;
-use crate::{Container, Data};
+use crate::{WithProgress, Data};
 
 type PushList<T, C> = Rc<RefCell<Vec<Box<dyn Push<Message<T, C>>>>>>;
 
@@ -17,7 +17,7 @@ pub struct Tee<T, C> {
     shared: PushList<T, C>,
 }
 
-impl<T: Data, C: Container + Data> Push<Message<T, C>> for Tee<T, C> {
+impl<T: Data, C: WithProgress + 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();
@@ -39,7 +39,7 @@ impl<T: Data, C: Container + Data> Push<Message<T, C>> for Tee<T, C> {
     }
 }
 
-impl<T, C: Container> Tee<T, C> {
+impl<T, C: WithProgress> Tee<T, C> {
     /// Allocates a new pair of `Tee` and `TeeHelper`.
     pub fn new() -> (Tee<T, C>, TeeHelper<T, C>) {
         let shared = Rc::new(RefCell::new(Vec::new()));
@@ -52,7 +52,7 @@ impl<T, C: Container> Tee<T, C> {
     }
 }
 
-impl<T, C: Container> Clone for Tee<T, C> {
+impl<T, C: WithProgress> Clone for Tee<T, C> {
     fn clone(&self) -> Self {
         Self {
             buffer: Default::default(),

timely/src/dataflow/operators/branch.rs

@@ -3,7 +3,7 @@
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
 use crate::dataflow::{Scope, Stream, StreamCore};
-use crate::{Container, Data};
+use crate::{WithProgress, Data};
 
 /// Extension trait for `Stream`.
 pub trait Branch<S: Scope, D: Data> {
@@ -93,7 +93,7 @@ pub trait BranchWhen<T>: Sized {
     fn branch_when(&self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self);
 }
 
-impl<S: Scope, C: Container + Data> BranchWhen<S::Timestamp> for StreamCore<S, C> {
+impl<S: Scope, C: WithProgress + 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());
         builder.set_notify(false);

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, Data};
+use crate::{WithProgress, 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 + Data> {
+pub trait Capture<T: Timestamp, C: WithProgress + Data> {
     /// Captures a stream of timestamped data for later replay.
     ///
     /// # Examples
@@ -113,7 +113,7 @@ pub trait Capture<T: Timestamp, C: Container + Data> {
     }
 }
 
-impl<S: Scope, C: Container + Data> Capture<S::Timestamp, C> for StreamCore<S, C> {
+impl<S: Scope, C: WithProgress + 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());