container: Preserve public API

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

Author: antiguru

kafkaesque/src/kafka_source.rs

@@ -96,7 +96,7 @@ where
     D: Data,
     L: Fn(&[u8],
           &mut Capability<G::Timestamp>,
-          &mut OutputHandle<G::Timestamp, Vec<D>, Tee<G::Timestamp, Vec<D>>>) -> bool+'static,
+          &mut OutputHandle<G::Timestamp, D, Tee<G::Timestamp, D>>) -> bool+'static,
 {
     use timely::dataflow::operators::generic::source;
     source(scope, name, move |capability, info| {

timely/src/dataflow/channels/pact.rs

@@ -19,8 +19,8 @@ use super::{Bundle, Message};
 
 use crate::logging::{TimelyLogger as Logger, MessagesEvent};
 
-/// A `ParallelizationContract` allocates paired `Push` and `Pull` implementors.
-pub trait ParallelizationContract<T: 'static, D: 'static> {
+/// A `ParallelizationContractCore` allocates paired `Push` and `Pull` implementors.
+pub trait ParallelizationContractCore<T: 'static, D: 'static> {
     /// Type implementing `Push` produced by this pact.
     type Pusher: Push<Bundle<T, D>>+'static;
     /// Type implementing `Pull` produced by this pact.
@@ -29,11 +29,15 @@ pub trait ParallelizationContract<T: 'static, D: 'static> {
     fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: &[usize], logging: Option<Logger>) -> (Self::Pusher, Self::Puller);
 }
 
+/// A `ParallelizationContractCore` specialized for `Vec` containers
+/// TODO: Use trait aliases once stable.
+pub trait ParallelizationContract<T: 'static, D: 'static>: ParallelizationContractCore<T, Vec<D>> { }
+
 /// A direct connection
 #[derive(Debug)]
 pub struct Pipeline;
 
-impl<T: 'static, D: Container+'static> ParallelizationContract<T, D> for Pipeline {
+impl<T: 'static, D: Container+'static> ParallelizationContractCore<T, D> for Pipeline {
     type Pusher = LogPusher<T, D, ThreadPusher<Bundle<T, D>>>;
     type Puller = LogPuller<T, D, ThreadPuller<Bundle<T, D>>>;
     fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: &[usize], logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
@@ -45,6 +49,8 @@ impl<T: 'static, D: Container+'static> ParallelizationContract<T, D> for Pipelin
     }
 }
 
+impl<T: 'static, D: Clone+'static> ParallelizationContract<T, D> for Pipeline { }
+
 /// An exchange between multiple observers by data
 pub struct Exchange<C, D, F> { hash_func: F, phantom: PhantomData<(C, D)> }
 
@@ -59,7 +65,7 @@ impl<'a, C: Container, D: Data, F: FnMut(&D)->u64+'static> Exchange<C, D, F> {
 }
 
 // Exchange uses a `Box<Pushable>` because it cannot know what type of pushable will return from the allocator.
-impl<'a, T: Eq+Data+Clone, C: Container<Inner=D>+ExchangeContainer+'static, D: Data+Clone, F: FnMut(&D)->u64+'static> ParallelizationContract<T, C> for Exchange<C, D, F>
+impl<'a, T: Eq+Data+Clone, C: Container<Inner=D>+ExchangeContainer+'static, D: Data+Clone, F: FnMut(&D)->u64+'static> ParallelizationContractCore<T, C> for Exchange<C, D, F>
     where for<'b> &'b mut C: DrainContainer<Inner=D>,
 {
     // TODO: The closure in the type prevents us from naming it.
@@ -73,6 +79,8 @@ impl<'a, T: Eq+Data+Clone, C: Container<Inner=D>+ExchangeContainer+'static, D: D
     }
 }
 
+impl<T: Eq+Data+Clone, D: Data+Clone, F: FnMut(&D)->u64+'static> ParallelizationContract<T, D> for Exchange<Vec<D>, D, F> { }
+
 impl<C, D, F> Debug for Exchange<C, D, F> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Exchange").finish()

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

@@ -27,7 +27,8 @@ impl<T: ::std::fmt::Debug, C: Container, P: Push<Bundle<T, C>>+::std::fmt::Debug
         let mut debug = f.debug_struct("UnorderedHandle");
         debug.field("time", &self.time);
         debug.field("pusher", &self.pusher);
-        debug.finish_non_exhaustive()
+        debug.field("buffer", &"...");
+        debug.finish()
     }
 }
 

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

@@ -1,4 +1,4 @@
-pub use self::tee::{Tee, TeeHelper};
+pub use self::tee::{Tee, TeeCore, TeeHelper};
 pub use self::exchange::Exchange;
 pub use self::counter::Counter;
 

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

@@ -12,12 +12,15 @@ use crate::communication::Push;
 type PushList<T, D> = Rc<RefCell<Vec<Box<dyn Push<Bundle<T, D>>>>>>;
 
 /// Wraps a shared list of `Box<Push>` to forward pushes to. Owned by `Stream`.
-pub struct Tee<T: 'static, D: 'static> {
+pub struct TeeCore<T: 'static, D: 'static> {
     buffer: D,
     shared: PushList<T, D>,
 }
 
-impl<T: Data, D: Container> Push<Bundle<T, D>> for Tee<T, D> {
+/// [TeeCore] specialized to `Vec`-based container.
+pub type Tee<T, D> = TeeCore<T, Vec<D>>;
+
+impl<T: Data, D: Container> Push<Bundle<T, D>> for TeeCore<T, D> {
     #[inline]
     fn push(&mut self, message: &mut Option<Bundle<T, D>>) {
         let mut pushers = self.shared.borrow_mut();
@@ -41,11 +44,11 @@ impl<T: Data, D: Container> Push<Bundle<T, D>> for Tee<T, D> {
     }
 }
 
-impl<T, D: Container> Tee<T, D> {
+impl<T, D: Container> TeeCore<T, D> {
     /// Allocates a new pair of `Tee` and `TeeHelper`.
-    pub fn new() -> (Tee<T, D>, TeeHelper<T, D>) {
+    pub fn new() -> (TeeCore<T, D>, TeeHelper<T, D>) {
         let shared = Rc::new(RefCell::new(Vec::new()));
-        let port = Tee {
+        let port = TeeCore {
             buffer: D::Builder::with_capacity(D::default_length()).build(),
             shared: shared.clone(),
         };
@@ -54,7 +57,7 @@ impl<T, D: Container> Tee<T, D> {
     }
 }
 
-impl<T, D> Debug for Tee<T, D>
+impl<T, D> Debug for TeeCore<T, D>
 where
     D: Debug,
 {

timely/src/dataflow/operators/enterleave.rs

@@ -27,7 +27,7 @@ use crate::progress::{Source, Target};
 use crate::order::Product;
 use crate::{Container, ContainerBuilder, Data};
 use crate::communication::Push;
-use crate::dataflow::channels::pushers::{Counter, Tee};
+use crate::dataflow::channels::pushers::{Counter, TeeCore};
 use crate::dataflow::channels::{Bundle, Message};
 
 use crate::worker::AsWorker;
@@ -87,7 +87,7 @@ impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container<Inner=D>, D> Ent
 
         use crate::scheduling::Scheduler;
 
-        let (targets, registrar) = Tee::<T, C>::new();
+        let (targets, registrar) = TeeCore::<T, C>::new();
         let ingress = IngressNub {
             targets: Counter::new(targets),
             phantom: ::std::marker::PhantomData,
@@ -129,7 +129,7 @@ impl<'a, G: Scope, D: Container, T: Timestamp+Refines<G::Timestamp>> Leave<G, D>
         let scope = self.scope();
 
         let output = scope.subgraph.borrow_mut().new_output();
-        let (targets, registrar) = Tee::<G::Timestamp, D>::new();
+        let (targets, registrar) = TeeCore::<G::Timestamp, D>::new();
         let channel_id = scope.clone().new_identifier();
         self.connect_to(Target::new(0, output.port), EgressNub { targets, phantom: PhantomData }, channel_id);
 
@@ -143,7 +143,7 @@ impl<'a, G: Scope, D: Container, T: Timestamp+Refines<G::Timestamp>> Leave<G, D>
 
 
 struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> {
-    targets: Counter<TInner, TData, Tee<TInner, TData>>,
+    targets: Counter<TInner, TData, TeeCore<TInner, TData>>,
     phantom: ::std::marker::PhantomData<TOuter>,
     activator: crate::scheduling::Activator,
     active: bool,
@@ -175,7 +175,7 @@ impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> Pus
 
 
 struct EgressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> {
-    targets: Tee<TOuter, TData>,
+    targets: TeeCore<TOuter, TData>,
     phantom: PhantomData<TInner>,
 }
 

timely/src/dataflow/operators/feedback.rs

@@ -4,7 +4,7 @@ use crate::progress::{Timestamp, PathSummary};
 use crate::progress::frontier::Antichain;
 use crate::order::Product;
 
-use crate::dataflow::channels::pushers::Tee;
+use crate::dataflow::channels::pushers::TeeCore;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{CoreStream, Scope, Stream};
 use crate::dataflow::scopes::child::Iterative;
@@ -35,7 +35,7 @@ pub trait Feedback<G: Scope> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn feedback<D: Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, Vec<D>>, Stream<G, D>);
+    fn feedback<D: Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, D>, Stream<G, D>);
 
     /// Creates a `CoreStream` and a `Handle` to later bind the source of that `Stream`.
     ///
@@ -58,7 +58,7 @@ pub trait Feedback<G: Scope> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn feedback_core<D: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, D>, CoreStream<G, D>);
+    fn feedback_core<D: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, D>, CoreStream<G, D>);
 }
 
 /// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
@@ -86,25 +86,25 @@ pub trait LoopVariable<'a, G: Scope, T: Timestamp> {
     ///     });
     /// });
     /// ```
-    fn loop_variable<D: Container>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, D>, CoreStream<Iterative<'a, G, T>, D>);
+    fn loop_variable<D: Container>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, D>, CoreStream<Iterative<'a, G, T>, D>);
 }
 
 impl<G: Scope> Feedback<G> for G {
-    fn feedback<D: Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, Vec<D>>, Stream<G, D>) {
+    fn feedback<D: Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, D>, Stream<G, D>) {
         self.feedback_core(summary)
     }
 
-    fn feedback_core<D: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, D>, CoreStream<G, D>) {
+    fn feedback_core<D: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, D>, CoreStream<G, D>) {
 
         let mut builder = OperatorBuilder::new("Feedback".to_owned(), self.clone());
         let (output, stream) = builder.new_output();
 
-        (Handle { builder, summary, output }, stream)
+        (HandleCore { builder, summary, output }, stream)
     }
 }
 
 impl<'a, G: Scope, T: Timestamp> LoopVariable<'a, G, T> for Iterative<'a, G, T> {
-    fn loop_variable<D: Container>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, D>, CoreStream<Iterative<'a, G, T>, D>) {
+    fn loop_variable<D: Container>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, D>, CoreStream<Iterative<'a, G, T>, D>) {
         self.feedback_core(Product::new(Default::default(), summary))
     }
 }
@@ -128,11 +128,11 @@ pub trait ConnectLoop<G: Scope, D: Container> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn connect_loop(&self, _: Handle<G, D>);
+    fn connect_loop(&self, _: HandleCore<G, D>);
 }
 
 impl<G: Scope, D: Container> ConnectLoop<G, D> for CoreStream<G, D> {
-    fn connect_loop(&self, helper: Handle<G, D>) {
+    fn connect_loop(&self, helper: HandleCore<G, D>) {
 
         let mut builder = helper.builder;
         let summary = helper.summary;
@@ -158,8 +158,11 @@ impl<G: Scope, D: Container> ConnectLoop<G, D> for CoreStream<G, D> {
 
 /// A handle used to bind the source of a loop variable.
 #[derive(Debug)]
-pub struct Handle<G: Scope, D: Container> {
+pub struct HandleCore<G: Scope, D: Container> {
     builder: OperatorBuilder<G>,
     summary: <G::Timestamp as Timestamp>::Summary,
-    output: OutputWrapper<G::Timestamp, D, Tee<G::Timestamp, D>>,
+    output: OutputWrapper<G::Timestamp, D, TeeCore<G::Timestamp, D>>,
 }
+
+/// A `HandleCore` specialized for using `Vec` as container
+pub type Handle<G, D> = HandleCore<G, Vec<D>>;

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

@@ -16,8 +16,8 @@ use crate::progress::{Source, Target};
 use crate::progress::{Timestamp, Operate, operate::SharedProgress, Antichain};
 
 use crate::dataflow::{CoreStream, Scope};
-use crate::dataflow::channels::pushers::Tee;
-use crate::dataflow::channels::pact::ParallelizationContract;
+use crate::dataflow::channels::pushers::TeeCore;
+use crate::dataflow::channels::pact::ParallelizationContractCore;
 use crate::dataflow::operators::generic::operator_info::OperatorInfo;
 use crate::Container;
 
@@ -109,15 +109,15 @@ impl<G: Scope> OperatorBuilder<G> {
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
     pub fn new_input<D: Data, P>(&mut self, stream: &CoreStream<G, D>, pact: P) -> P::Puller
         where
-            P: ParallelizationContract<G::Timestamp, D> {
+            P: ParallelizationContractCore<G::Timestamp, D> {
         let connection = vec![Antichain::from_elem(Default::default()); self.shape.outputs];
         self.new_input_connection(stream, pact, connection)
     }
 
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
     pub fn new_input_connection<D: Data, P>(&mut self, stream: &CoreStream<G, D>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> P::Puller
     where
-        P: ParallelizationContract<G::Timestamp, D> {
+        P: ParallelizationContractCore<G::Timestamp, D> {
 
         let channel_id = self.scope.new_identifier();
         let logging = self.scope.logging();
@@ -133,16 +133,16 @@ impl<G: Scope> OperatorBuilder<G> {
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
-    pub fn new_output<D: Container>(&mut self) -> (Tee<G::Timestamp, D>, CoreStream<G, D>) {
+    pub fn new_output<D: Container>(&mut self) -> (TeeCore<G::Timestamp, D>, CoreStream<G, D>) {
 
         let connection = vec![Antichain::from_elem(Default::default()); self.shape.inputs];
         self.new_output_connection(connection)
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
-    pub fn new_output_connection<D: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (Tee<G::Timestamp, D>, CoreStream<G, D>) {
+    pub fn new_output_connection<D: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (TeeCore<G::Timestamp, D>, CoreStream<G, D>) {
 
-        let (targets, registrar) = Tee::<G::Timestamp,D>::new();
+        let (targets, registrar) = TeeCore::<G::Timestamp,D>::new();
         let source = Source::new(self.index, self.shape.outputs);
         let stream = CoreStream::new(source, registrar, self.scope.clone());
 

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

@@ -11,10 +11,10 @@ use crate::progress::operate::SharedProgress;
 use crate::progress::frontier::{Antichain, MutableAntichain};
 
 use crate::dataflow::{Scope, CoreStream};
-use crate::dataflow::channels::pushers::Tee;
+use crate::dataflow::channels::pushers::TeeCore;
 use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::Buffer as PushBuffer;
-use crate::dataflow::channels::pact::ParallelizationContract;
+use crate::dataflow::channels::pact::ParallelizationContractCore;
 use crate::dataflow::channels::pullers::Counter as PullCounter;
 use crate::dataflow::operators::capability::Capability;
 use crate::dataflow::operators::generic::handles::{InputHandle, new_input_handle, OutputWrapper};
@@ -60,7 +60,7 @@ impl<G: Scope> OperatorBuilder<G> {
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
     pub fn new_input<D: Container, P>(&mut self, stream: &CoreStream<G, D>, pact: P) -> InputHandle<G::Timestamp, D, P::Puller>
     where
-        P: ParallelizationContract<G::Timestamp, D> {
+        P: ParallelizationContractCore<G::Timestamp, D> {
 
         let connection = vec![Antichain::from_elem(Default::default()); self.builder.shape().outputs()];
         self.new_input_connection(stream, pact, connection)
@@ -76,7 +76,7 @@ impl<G: Scope> OperatorBuilder<G> {
     /// antichain indicating that there is no connection from the input to the output.
     pub fn new_input_connection<D: Data, P>(&mut self, stream: &CoreStream<G, D>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> InputHandle<G::Timestamp, D, P::Puller>
         where
-            P: ParallelizationContract<G::Timestamp, D> {
+            P: ParallelizationContractCore<G::Timestamp, D> {
 
         let puller = self.builder.new_input_connection(stream, pact, connection);
 
@@ -88,7 +88,7 @@ impl<G: Scope> OperatorBuilder<G> {
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
-    pub fn new_output<D: Container>(&mut self) -> (OutputWrapper<G::Timestamp, D, Tee<G::Timestamp, D>>, CoreStream<G, D>) {
+    pub fn new_output<D: Container>(&mut self) -> (OutputWrapper<G::Timestamp, D, TeeCore<G::Timestamp, D>>, CoreStream<G, D>) {
         let connection = vec![Antichain::from_elem(Default::default()); self.builder.shape().inputs()];
         self.new_output_connection(connection)
     }
@@ -101,7 +101,7 @@ impl<G: Scope> OperatorBuilder<G> {
     ///
     /// Commonly the connections are either the unit summary, indicating the same timestamp might be produced as output, or an empty
     /// antichain indicating that there is no connection from the input to the output.
-    pub fn new_output_connection<D: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (OutputWrapper<G::Timestamp, D, Tee<G::Timestamp, D>>, CoreStream<G, D>) {
+    pub fn new_output_connection<D: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (OutputWrapper<G::Timestamp, D, TeeCore<G::Timestamp, D>>, CoreStream<G, D>) {
 
         let (tee, stream) = self.builder.new_output_connection(connection);
 

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

@@ -179,8 +179,8 @@ impl<T: Timestamp, D: Container, P: Push<Bundle<T, D>>> OutputWrapper<T, D, P> {
     ///
     /// This method ensures that the only access to the push buffer is through the `OutputHandle`
     /// type which ensures the use of capabilities, and which calls `cease` when it is dropped.
-    pub fn activate(&mut self) -> OutputHandle<T, D, P> {
-        OutputHandle {
+    pub fn activate(&mut self) -> OutputHandleCore<T, D, P> {
+        OutputHandleCore {
             push_buffer: &mut self.push_buffer,
             internal_buffer: &self.internal_buffer,
         }
@@ -189,12 +189,15 @@ impl<T: Timestamp, D: Container, P: Push<Bundle<T, D>>> OutputWrapper<T, D, P> {
 
 
 /// Handle to an operator's output stream.
-pub struct OutputHandle<'a, T: Timestamp, C: Container+'a, P: Push<Bundle<T, C>>+'a> {
+pub struct OutputHandleCore<'a, T: Timestamp, C: Container+'a, P: Push<Bundle<T, C>>+'a> {
     push_buffer: &'a mut Buffer<T, C, PushCounter<T, C, P>>,
     internal_buffer: &'a Rc<RefCell<ChangeBatch<T>>>,
 }
 
-impl<'a, T: Timestamp, C: Container, P: Push<Bundle<T, C>>> OutputHandle<'a, T, C, P> {
+/// Handle specialized to `Vec`-based container.
+pub type OutputHandle<'a, T, D, P> = OutputHandleCore<'a, T, Vec<D>, P>;
+
+impl<'a, T: Timestamp, C: Container, P: Push<Bundle<T, C>>> OutputHandleCore<'a, T, C, P> {
     /// Obtains a session that can send data at the timestamp associated with capability `cap`.
     ///
     /// In order to send data at a future timestamp, obtain a capability for the new timestamp
@@ -223,7 +226,7 @@ impl<'a, T: Timestamp, C: Container, P: Push<Bundle<T, C>>> OutputHandle<'a, T,
     }
 }
 
-impl<'a, T: Timestamp, C: Container, P: Push<Bundle<T, C>>> Drop for OutputHandle<'a, T, C, P> {
+impl<'a, T: Timestamp, C: Container, P: Push<Bundle<T, C>>> Drop for OutputHandleCore<'a, T, C, P> {
     fn drop(&mut self) {
         self.push_buffer.cease();
     }