Address feedback

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

Author: antiguru

container/src/lib.rs

@@ -4,15 +4,12 @@
 
 use std::collections::VecDeque;
 
-/// An object with effects on progress
+/// An type containing a number of records accounted for by progress tracking.
 ///
 /// 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.
-///
-/// It must implement default for historic reason. The default implementation is not required
-/// to allocate memory for variable-length components.
-pub trait WithProgress {
+pub trait Accountable {
     /// The number of records
     ///
     /// This number is used in progress tracking to confirm the receipt of some number
@@ -22,12 +19,13 @@ pub trait WithProgress {
     fn record_count(&self) -> i64;
 
     /// Determine if this contains any updates, corresponding to `update_count() == 0`.
+    /// It is a correctness error for this to by anything other than `self.record_count() == 0`.
     #[inline] fn is_empty(&self) -> bool { self.record_count() == 0 }
 }
 
-/// A container that allows iteration.
+/// A container that allows iteration morally equivalent to [`IntoIterator`].
 ///
-/// Iterating the container presents items in an implmentation-specific order.
+/// Iterating the container presents items in an implementation-specific order.
 /// The container's contents are not changed.
 pub trait IterContainer {
     /// The type of elements when reading non-destructively from the container.
@@ -54,7 +52,7 @@ pub trait DrainContainer {
 }
 
 /// A container that can be sized and reveals its capacity.
-pub trait SizableContainer: Sized {
+pub trait SizableContainer {
     /// 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.
@@ -65,7 +63,7 @@ pub trait SizableContainer: Sized {
     ///
     /// Assume that the `stash` is in an undefined state, and properly clear it
     /// before re-using it.
-    fn ensure_capacity(&mut self, stash: &mut Option<Self>);
+    fn ensure_capacity(&mut self, stash: &mut Option<Self>) where Self: Sized;
 }
 
 /// A container that can absorb items of a specific type.
@@ -90,8 +88,10 @@ pub trait PushInto<T> {
 ///
 /// The caller should consume the containers returned by [`Self::extract`] and
 /// [`Self::finish`]. Implementations can recycle buffers, but should ensure that they clear
-/// any remaining elements. It is up to the implementation of this trait to ensure that
-/// containers are properly cleared before recycling them.
+/// any remaining elements.
+///
+/// Implementations are allowed to re-use the contents of the mutable references left by the caller,
+/// but they should ensure that they clear the contents before doing so.
 ///
 /// For example, a consolidating builder can aggregate differences in-place, but it has
 /// to ensure that it preserves the intended information.
@@ -102,7 +102,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: WithProgress + Default + Clone + 'static;
+    type Container: Accountable + Default + Clone + 'static;
     /// Extract assembled containers, potentially leaving unfinished data behind. Can
     /// be called repeatedly, for example while the caller can send data.
     ///
@@ -139,7 +139,7 @@ pub trait ContainerBuilder: Default + 'static {
 /// A wrapper trait indicating that the container building will preserve the number of records.
 ///
 /// Specifically, the sum of record counts of all extracted and finished containers must equal the
-/// number of times that `push_into` is called on the container builder.
+/// number of accounted records that are pushed into the container builder.
 /// If you have any questions about this trait you are best off not implementing it.
 pub trait LengthPreservingContainerBuilder : ContainerBuilder { }
 
@@ -175,7 +175,7 @@ impl<T, C: SizableContainer + Default + PushInto<T>> PushInto<T> for CapacityCon
     }
 }
 
-impl<C: WithProgress + Default + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
+impl<C: Accountable + Default + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
     type Container = C;
 
     #[inline]
@@ -198,9 +198,9 @@ impl<C: WithProgress + Default + Clone + 'static> ContainerBuilder for CapacityC
     }
 }
 
-impl<C: WithProgress + SizableContainer + Default + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
+impl<C: Accountable + SizableContainer + Default + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
 
-impl<T> WithProgress for Vec<T> {
+impl<T> Accountable for Vec<T> {
     #[inline] fn record_count(&self) -> i64 { i64::try_from(Vec::len(self)).unwrap() }
     #[inline] fn is_empty(&self) -> bool { Vec::is_empty(self) }
 }
@@ -263,9 +263,9 @@ mod rc {
     use std::ops::Deref;
     use std::rc::Rc;
 
-    use crate::{WithProgress, IterContainer, DrainContainer};
+    use crate::{IterContainer, DrainContainer};
 
-    impl<T: WithProgress> WithProgress for Rc<T> {
+    impl<T: crate::Accountable> crate::Accountable for Rc<T> {
         #[inline] fn record_count(&self) -> i64 { std::ops::Deref::deref(self).record_count() }
         #[inline] fn is_empty(&self) -> bool { std::ops::Deref::deref(self).is_empty() }
     }
@@ -285,9 +285,9 @@ mod arc {
     use std::ops::Deref;
     use std::sync::Arc;
 
-    use crate::{WithProgress, IterContainer, DrainContainer};
+    use crate::{IterContainer, DrainContainer};
 
-    impl<T: WithProgress> WithProgress for Arc<T> {
+    impl<T: crate::Accountable> crate::Accountable for Arc<T> {
         #[inline] fn record_count(&self) -> i64 { std::ops::Deref::deref(self).record_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::WithProgress for Column<C> {
+    impl<C: columnar::ContainerBytes> timely::Accountable for Column<C> {
         #[inline] fn record_count(&self) -> i64 { i64::try_from(self.borrow().len()).unwrap() }
         #[inline] fn is_empty(&self) -> bool { self.borrow().is_empty() }
     }

timely/src/dataflow/channels/pact.rs

@@ -10,7 +10,7 @@
 use std::{fmt::{self, Debug}, marker::PhantomData};
 use std::rc::Rc;
 
-use crate::WithProgress;
+use crate::Accountable;
 use crate::container::{ContainerBuilder, DrainContainer, LengthPreservingContainerBuilder, SizableContainer, CapacityContainerBuilder, PushInto};
 use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
 use crate::communication::{Push, Pull};
@@ -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: Accountable + '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) {
@@ -129,7 +129,7 @@ impl<P> LogPusher<P> {
     }
 }
 
-impl<T, C: WithProgress, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<P> {
+impl<T, C: Accountable, 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 {
@@ -177,7 +177,7 @@ impl<P> LogPuller<P> {
     }
 }
 
-impl<T, C: WithProgress, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<P> {
+impl<T, C: Accountable, 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::WithProgress;
+use crate::Accountable;
 
 /// A wrapper which accounts records pulled past in a shared count map.
 pub struct Counter<T, C, P> {
@@ -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: Accountable, 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, WithProgress};
+use crate::{Container, Accountable};
 
 /// Buffers data sent at the same time, for efficient communication.
 ///

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::WithProgress;
+use crate::Accountable;
 
 /// 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: Accountable, 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/operators/core/enterleave.rs

@@ -26,7 +26,7 @@ use crate::logging::{TimelyLogger, MessagesEvent};
 use crate::progress::Timestamp;
 use crate::progress::timestamp::Refines;
 use crate::progress::{Source, Target};
-use crate::{WithProgress, Container};
+use crate::{Accountable, Container};
 use crate::communication::Push;
 use crate::dataflow::channels::pushers::{Counter, Tee};
 use crate::dataflow::channels::Message;
@@ -207,7 +207,7 @@ impl<P> LogPusher<P> {
 
 impl<T, C, P> Push<Message<T, C>> for LogPusher<P>
 where
-    C: WithProgress,
+    C: Accountable,
     P: Push<Message<T, C>>,
 {
     fn push(&mut self, element: &mut Option<Message<T, C>>) {

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

@@ -10,7 +10,7 @@ use crate::scheduling::{Schedule, Activator};
 use crate::progress::{Operate, operate::SharedProgress, Timestamp, ChangeBatch};
 use crate::progress::Source;
 use crate::progress::operate::Connectivity;
-use crate::{WithProgress, Container};
+use crate::{Accountable, Container};
 use crate::communication::Push;
 use crate::dataflow::{Scope, ScopeParent, StreamCore};
 use crate::dataflow::channels::pushers::{Tee, Counter};

timely/src/dataflow/operators/generic/handles.rs

@@ -15,7 +15,7 @@ use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::{Buffer, Session};
 use crate::dataflow::channels::Message;
 use crate::communication::{Push, Pull};
-use crate::{Container, WithProgress};
+use crate::{Container, Accountable};
 use crate::container::{ContainerBuilder, CapacityContainerBuilder};
 
 use crate::dataflow::operators::InputCapability;
@@ -46,7 +46,7 @@ pub struct FrontieredInputHandleCore<'a, T: Timestamp, C: 'a, P: Pull<Message<T,
 /// Handle to an operator's input stream and frontier, specialized to vectors.
 pub type FrontieredInputHandle<'a, T, D, P> = FrontieredInputHandleCore<'a, T, Vec<D>, P>;
 
-impl<T: Timestamp, C: WithProgress, P: Pull<Message<T, C>>> InputHandleCore<T, C, P> {
+impl<T: Timestamp, C: Accountable, P: Pull<Message<T, C>>> InputHandleCore<T, C, P> {
 
     /// Reads the next input buffer (at some timestamp `t`) and a corresponding capability for `t`.
     /// The timestamp `t` of the input buffer can be retrieved by invoking `.time()` on the capability.
@@ -87,7 +87,7 @@ impl<T: Timestamp, C: WithProgress, P: Pull<Message<T, C>>> InputHandleCore<T, C
 
 }
 
-impl<'a, T: Timestamp, C: WithProgress, P: Pull<Message<T, C>>+'a> FrontieredInputHandleCore<'a, T, C, P> {
+impl<'a, T: Timestamp, C: Accountable, P: Pull<Message<T, C>>+'a> FrontieredInputHandleCore<'a, T, C, P> {
     /// Allocate a new frontiered input handle.
     pub fn new(handle: &'a mut InputHandleCore<T, C, P>, frontier: &'a MutableAntichain<T>) -> Self {
         FrontieredInputHandleCore {
@@ -134,13 +134,13 @@ impl<'a, T: Timestamp, C: WithProgress, P: Pull<Message<T, C>>+'a> FrontieredInp
     }
 }
 
-pub fn _access_pull_counter<T: Timestamp, C: WithProgress, P: Pull<Message<T, C>>>(input: &mut InputHandleCore<T, C, P>) -> &mut PullCounter<T, C, P> {
+pub fn _access_pull_counter<T: Timestamp, C: Accountable, P: Pull<Message<T, C>>>(input: &mut InputHandleCore<T, C, P>) -> &mut PullCounter<T, C, P> {
     &mut input.pull_counter
 }
 
 /// Constructs an input handle.
 /// Declared separately so that it can be kept private when `InputHandle` is re-exported.
-pub fn new_input_handle<T: Timestamp, C: WithProgress, P: Pull<Message<T, C>>>(
+pub fn new_input_handle<T: Timestamp, C: Accountable, P: Pull<Message<T, C>>>(
     pull_counter: PullCounter<T, C, P>,
     internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>,
     summaries: Rc<RefCell<PortConnectivity<T::Summary>>>,

timely/src/lib.rs

@@ -65,7 +65,7 @@ pub use timely_communication::Config as CommunicationConfig;
 pub use worker::Config as WorkerConfig;
 pub use execute::Config as Config;
 
-pub use timely_container::WithProgress;
+pub use timely_container::Accountable;
 /// Re-export of the `timely_container` crate.
 pub mod container {
     pub use timely_container::*;
@@ -107,8 +107,8 @@ impl<T: Clone+'static> Data for T { }
 /// A composite trait for types usable as containers in timely dataflow.
 ///
 /// The `Container` trait is necessary for all containers in timely dataflow channels.
-pub trait Container: WithProgress + Default + Clone + 'static { }
-impl<C: WithProgress + Default + Clone + 'static> Container for C { }
+pub trait Container: Accountable + Default + Clone + 'static { }
+impl<C: Accountable + Default + Clone + 'static> Container for C { }
 
 /// A composite trait for types usable on exchange channels in timely dataflow.
 ///