diff --git a/timely/src/dataflow/channels/mod.rs b/timely/src/dataflow/channels/mod.rs
index 5bced98cd..aab1e0b03 100644
--- a/timely/src/dataflow/channels/mod.rs
+++ b/timely/src/dataflow/channels/mod.rs
@@ -11,39 +11,39 @@ pub mod pullers;
 pub mod pact;
 
 /// The input to and output from timely dataflow communication channels.
-pub type Bundle<T, D> = crate::communication::Message<Message<T, D>>;
+pub type Bundle<T, C> = crate::communication::Message<Message<T, C>>;
 
 /// A serializable representation of timestamped data.
 #[derive(Clone, Abomonation, Serialize, Deserialize)]
-pub struct Message<T, D> {
+pub struct Message<T, C> {
     /// The timestamp associated with the message.
     pub time: T,
     /// The data in the message.
-    pub data: D,
+    pub data: C,
     /// The source worker.
     pub from: usize,
     /// A sequence number for this worker-to-worker stream.
     pub seq: usize,
 }
 
-impl<T, D> Message<T, D> {
+impl<T, C> Message<T, C> {
     /// Default buffer size.
     #[deprecated = "Use timely::buffer::default_capacity instead"]
     pub fn default_length() -> usize {
-        crate::container::buffer::default_capacity::<D>()
+        crate::container::buffer::default_capacity::<C>()
     }
 }
 
-impl<T, D: Container> Message<T, D> {
+impl<T, C: Container> Message<T, C> {
     /// Creates a new message instance from arguments.
-    pub fn new(time: T, data: D, from: usize, seq: usize) -> Self {
+    pub fn new(time: T, data: C, from: usize, seq: usize) -> Self {
         Message { time, data, from, seq }
     }
 
     /// Forms a message, and pushes contents at `pusher`. Replaces `buffer` with what the pusher
     /// leaves in place, or the container's default element.
     #[inline]
-    pub fn push_at<P: Push<Bundle<T, D>>>(buffer: &mut D, time: T, pusher: &mut P) {
+    pub fn push_at<P: Push<Bundle<T, C>>>(buffer: &mut C, time: T, pusher: &mut P) {
 
         let data = ::std::mem::take(buffer);
         let message = Message::new(time, data, 0, 0);
diff --git a/timely/src/dataflow/channels/pact.rs b/timely/src/dataflow/channels/pact.rs
index 3e0ba21f7..abf6fef3a 100644
--- a/timely/src/dataflow/channels/pact.rs
+++ b/timely/src/dataflow/channels/pact.rs
@@ -20,11 +20,11 @@ use crate::progress::Timestamp;
 use crate::worker::AsWorker;
 
 /// A `ParallelizationContract` allocates paired `Push` and `Pull` implementors.
-pub trait ParallelizationContract<T, D> {
+pub trait ParallelizationContract<T, C> {
     /// Type implementing `Push` produced by this pact.
-    type Pusher: Push<Bundle<T, D>>+'static;
+    type Pusher: Push<Bundle<T, C>>+'static;
     /// Type implementing `Pull` produced by this pact.
-    type Puller: Pull<Bundle<T, D>>+'static;
+    type Puller: Pull<Bundle<T, C>>+'static;
     /// Allocates a matched pair of push and pull endpoints implementing the pact.
     fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: &[usize], logging: Option<Logger>) -> (Self::Pusher, Self::Puller);
 }
@@ -33,13 +33,11 @@ pub trait ParallelizationContract<T, D> {
 #[derive(Debug)]
 pub struct Pipeline;
 
-impl<T: 'static, D: Container> ParallelizationContract<T, D> for Pipeline {
-    type Pusher = LogPusher<T, D, ThreadPusher<Bundle<T, D>>>;
-    type Puller = LogPuller<T, D, ThreadPuller<Bundle<T, D>>>;
+impl<T: 'static, C: Container> ParallelizationContract<T, C> for Pipeline {
+    type Pusher = LogPusher<T, C, ThreadPusher<Bundle<T, C>>>;
+    type Puller = LogPuller<T, C, ThreadPuller<Bundle<T, C>>>;
     fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: &[usize], logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
-        let (pusher, puller) = allocator.pipeline::<Message<T, D>>(identifier, address);
-        // // ignore `&mut A` and use thread allocator
-        // let (pusher, puller) = Thread::new::<Bundle<T, Vec<D>>>();
+        let (pusher, puller) = allocator.pipeline::<Message<T, C>>(identifier, address);
         (LogPusher::new(pusher, allocator.index(), allocator.index(), identifier, logging.clone()),
          LogPuller::new(puller, allocator.index(), identifier, logging))
     }
@@ -89,17 +87,17 @@ impl<C, F> Debug for ExchangeCore<C, F> {
 
 /// Wraps a `Message<T,D>` pusher to provide a `Push<(T, Content<D>)>`.
 #[derive(Debug)]
-pub struct LogPusher<T, D, P: Push<Bundle<T, D>>> {
+pub struct LogPusher<T, C, P: Push<Bundle<T, C>>> {
     pusher: P,
     channel: usize,
     counter: usize,
     source: usize,
     target: usize,
-    phantom: PhantomData<(T, D)>,
+    phantom: PhantomData<(T, C)>,
     logging: Option<Logger>,
 }
 
-impl<T, D, P: Push<Bundle<T, D>>> LogPusher<T, D, P> {
+impl<T, C, P: Push<Bundle<T, C>>> LogPusher<T, C, P> {
     /// Allocates a new pusher.
     pub fn new(pusher: P, source: usize, target: usize, channel: usize, logging: Option<Logger>) -> Self {
         LogPusher {
@@ -114,9 +112,9 @@ impl<T, D, P: Push<Bundle<T, D>>> LogPusher<T, D, P> {
     }
 }
 
-impl<T, D: Container, P: Push<Bundle<T, D>>> Push<Bundle<T, D>> for LogPusher<T, D, P> {
+impl<T, C: Container, P: Push<Bundle<T, C>>> Push<Bundle<T, C>> for LogPusher<T, C, P> {
     #[inline]
-    fn push(&mut self, pair: &mut Option<Bundle<T, D>>) {
+    fn push(&mut self, pair: &mut Option<Bundle<T, C>>) {
         if let Some(bundle) = pair {
             self.counter += 1;
 
@@ -145,15 +143,15 @@ impl<T, D: Container, P: Push<Bundle<T, D>>> Push<Bundle<T, D>> for LogPusher<T,
 
 /// Wraps a `Message<T,D>` puller to provide a `Pull<(T, Content<D>)>`.
 #[derive(Debug)]
-pub struct LogPuller<T, D, P: Pull<Bundle<T, D>>> {
+pub struct LogPuller<T, C, P: Pull<Bundle<T, C>>> {
     puller: P,
     channel: usize,
     index: usize,
-    phantom: PhantomData<(T, D)>,
+    phantom: PhantomData<(T, C)>,
     logging: Option<Logger>,
 }
 
-impl<T, D, P: Pull<Bundle<T, D>>> LogPuller<T, D, P> {
+impl<T, C, P: Pull<Bundle<T, C>>> LogPuller<T, C, P> {
     /// Allocates a new `Puller`.
     pub fn new(puller: P, index: usize, channel: usize, logging: Option<Logger>) -> Self {
         LogPuller {
@@ -166,9 +164,9 @@ impl<T, D, P: Pull<Bundle<T, D>>> LogPuller<T, D, P> {
     }
 }
 
-impl<T, D: Container, P: Pull<Bundle<T, D>>> Pull<Bundle<T, D>> for LogPuller<T, D, P> {
+impl<T, C: Container, P: Pull<Bundle<T, C>>> Pull<Bundle<T, C>> for LogPuller<T, C, P> {
     #[inline]
-    fn pull(&mut self) -> &mut Option<Bundle<T,D>> {
+    fn pull(&mut self) -> &mut Option<Bundle<T, C>> {
         let result = self.puller.pull();
         if let Some(bundle) = result {
             let channel = self.channel;
diff --git a/timely/src/dataflow/channels/pullers/counter.rs b/timely/src/dataflow/channels/pullers/counter.rs
index 848b9a8f5..8efba9318 100644
--- a/timely/src/dataflow/channels/pullers/counter.rs
+++ b/timely/src/dataflow/channels/pullers/counter.rs
@@ -9,10 +9,10 @@ use crate::communication::Pull;
 use crate::Container;
 
 /// A wrapper which accounts records pulled past in a shared count map.
-pub struct Counter<T: Ord+Clone+'static, D, P: Pull<Bundle<T, D>>> {
+pub struct Counter<T: Ord+Clone+'static, C, P: Pull<Bundle<T, C>>> {
     pullable: P,
     consumed: Rc<RefCell<ChangeBatch<T>>>,
-    phantom: ::std::marker::PhantomData<D>,
+    phantom: ::std::marker::PhantomData<C>,
 }
 
 /// A guard type that updates the change batch counts on drop
@@ -36,15 +36,15 @@ impl<T:Ord+Clone+'static> Drop for ConsumedGuard<T> {
     }
 }
 
-impl<T:Ord+Clone+'static, D: Container, P: Pull<Bundle<T, D>>> Counter<T, D, P> {
+impl<T:Ord+Clone+'static, C: Container, P: Pull<Bundle<T, C>>> Counter<T, C, P> {
     /// Retrieves the next timestamp and batch of data.
     #[inline]
-    pub fn next(&mut self) -> Option<&mut Bundle<T, D>> {
+    pub fn next(&mut self) -> Option<&mut Bundle<T, C>> {
         self.next_guarded().map(|(_guard, bundle)| bundle)
     }
 
     #[inline]
-    pub(crate) fn next_guarded(&mut self) -> Option<(ConsumedGuard<T>, &mut Bundle<T, D>)> {
+    pub(crate) fn next_guarded(&mut self) -> Option<(ConsumedGuard<T>, &mut Bundle<T, C>)> {
         if let Some(message) = self.pullable.pull() {
             let guard = ConsumedGuard {
                 consumed: Rc::clone(&self.consumed),
@@ -57,7 +57,7 @@ impl<T:Ord+Clone+'static, D: Container, P: Pull<Bundle<T, D>>> Counter<T, D, P>
     }
 }
 
-impl<T:Ord+Clone+'static, D, P: Pull<Bundle<T, D>>> Counter<T, D, P> {
+impl<T:Ord+Clone+'static, C, P: Pull<Bundle<T, C>>> Counter<T, C, P> {
     /// Allocates a new `Counter` from a boxed puller.
     pub fn new(pullable: P) -> Self {
         Counter {
diff --git a/timely/src/dataflow/channels/pushers/buffer.rs b/timely/src/dataflow/channels/pushers/buffer.rs
index 27e95377f..5262d36c9 100644
--- a/timely/src/dataflow/channels/pushers/buffer.rs
+++ b/timely/src/dataflow/channels/pushers/buffer.rs
@@ -13,11 +13,11 @@ use crate::{Container, Data};
 /// The `Buffer` type should be used by calling `session` with a time, which checks whether
 /// data must be flushed and creates a `Session` object which allows sending at the given time.
 #[derive(Debug)]
-pub struct Buffer<T, D: Container, P: Push<Bundle<T, D>>> {
+pub struct Buffer<T, C: Container, P: Push<Bundle<T, C>>> {
     /// the currently open time, if it is open
     time: Option<T>,
     /// a buffer for records, to send at self.time
-    buffer: D,
+    buffer: C,
     pusher: P,
 }
 
diff --git a/timely/src/dataflow/channels/pushers/counter.rs b/timely/src/dataflow/channels/pushers/counter.rs
index c40e3ddec..11b72946a 100644
--- a/timely/src/dataflow/channels/pushers/counter.rs
+++ b/timely/src/dataflow/channels/pushers/counter.rs
@@ -11,15 +11,15 @@ use crate::Container;
 
 /// A wrapper which updates shared `produced` based on the number of records pushed.
 #[derive(Debug)]
-pub struct Counter<T, D, P: Push<Bundle<T, D>>> {
+pub struct Counter<T, C, P: Push<Bundle<T, C>>> {
     pushee: P,
     produced: Rc<RefCell<ChangeBatch<T>>>,
-    phantom: PhantomData<D>,
+    phantom: PhantomData<C>,
 }
 
-impl<T: Timestamp, D: Container, P> Push<Bundle<T, D>> for Counter<T, D, P> where P: Push<Bundle<T, D>> {
+impl<T: Timestamp, C: Container, P> Push<Bundle<T, C>> for Counter<T, C, P> where P: Push<Bundle<T, C>> {
     #[inline]
-    fn push(&mut self, message: &mut Option<Bundle<T, D>>) {
+    fn push(&mut self, message: &mut Option<Bundle<T, C>>) {
         if let Some(message) = message {
             self.produced.borrow_mut().update(message.time.clone(), message.data.len() as i64);
         }
@@ -31,9 +31,9 @@ impl<T: Timestamp, D: Container, P> Push<Bundle<T, D>> for Counter<T, D, P> wher
     }
 }
 
-impl<T, D, P: Push<Bundle<T, D>>> Counter<T, D, P> where T : Ord+Clone+'static {
+impl<T, C, P: Push<Bundle<T, C>>> Counter<T, C, P> where T : Ord+Clone+'static {
     /// Allocates a new `Counter` from a pushee and shared counts.
-    pub fn new(pushee: P) -> Counter<T, D, P> {
+    pub fn new(pushee: P) -> Counter<T, C, P> {
         Counter {
             pushee,
             produced: Rc::new(RefCell::new(ChangeBatch::new())),
diff --git a/timely/src/dataflow/channels/pushers/tee.rs b/timely/src/dataflow/channels/pushers/tee.rs
index 210c9d149..f5cbfb226 100644
--- a/timely/src/dataflow/channels/pushers/tee.rs
+++ b/timely/src/dataflow/channels/pushers/tee.rs
@@ -9,17 +9,17 @@ use crate::dataflow::channels::{Bundle, Message};
 use crate::communication::Push;
 use crate::{Container, Data};
 
-type PushList<T, D> = Rc<RefCell<Vec<Box<dyn Push<Bundle<T, D>>>>>>;
+type PushList<T, C> = Rc<RefCell<Vec<Box<dyn Push<Bundle<T, C>>>>>>;
 
 /// Wraps a shared list of `Box<Push>` to forward pushes to. Owned by `Stream`.
-pub struct Tee<T, D> {
-    buffer: D,
-    shared: PushList<T, D>,
+pub struct Tee<T, C> {
+    buffer: C,
+    shared: PushList<T, C>,
 }
 
-impl<T: Data, D: Container> Push<Bundle<T, D>> for Tee<T, D> {
+impl<T: Data, C: Container> Push<Bundle<T, C>> for Tee<T, C> {
     #[inline]
-    fn push(&mut self, message: &mut Option<Bundle<T, D>>) {
+    fn push(&mut self, message: &mut Option<Bundle<T, C>>) {
         let mut pushers = self.shared.borrow_mut();
         if let Some(message) = message {
             for index in 1..pushers.len() {
@@ -39,9 +39,9 @@ impl<T: Data, D: Container> Push<Bundle<T, D>> for Tee<T, D> {
     }
 }
 
-impl<T, D: Container> Tee<T, D> {
+impl<T, C: Container> Tee<T, C> {
     /// Allocates a new pair of `Tee` and `TeeHelper`.
-    pub fn new() -> (Tee<T, D>, TeeHelper<T, D>) {
+    pub fn new() -> (Tee<T, C>, TeeHelper<T, C>) {
         let shared = Rc::new(RefCell::new(Vec::new()));
         let port = Tee {
             buffer: Default::default(),
@@ -52,7 +52,7 @@ impl<T, D: Container> Tee<T, D> {
     }
 }
 
-impl<T, D: Container> Clone for Tee<T, D> {
+impl<T, C: Container> Clone for Tee<T, C> {
     fn clone(&self) -> Self {
         Self {
             buffer: Default::default(),
@@ -61,9 +61,9 @@ impl<T, D: Container> Clone for Tee<T, D> {
     }
 }
 
-impl<T, D> Debug for Tee<T, D>
+impl<T, C> Debug for Tee<T, C>
 where
-    D: Debug,
+    C: Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let mut debug = f.debug_struct("Tee");
@@ -80,18 +80,18 @@ where
 }
 
 /// A shared list of `Box<Push>` used to add `Push` implementors.
-pub struct TeeHelper<T, D> {
-    shared: PushList<T, D>,
+pub struct TeeHelper<T, C> {
+    shared: PushList<T, C>,
 }
 
-impl<T, D> TeeHelper<T, D> {
+impl<T, C> TeeHelper<T, C> {
     /// Adds a new `Push` implementor to the list of recipients shared with a `Stream`.
-    pub fn add_pusher<P: Push<Bundle<T, D>>+'static>(&self, pusher: P) {
+    pub fn add_pusher<P: Push<Bundle<T, C>>+'static>(&self, pusher: P) {
         self.shared.borrow_mut().push(Box::new(pusher));
     }
 }
 
-impl<T, D> Clone for TeeHelper<T, D> {
+impl<T, C> Clone for TeeHelper<T, C> {
     fn clone(&self) -> Self {
         TeeHelper {
             shared: self.shared.clone(),
@@ -99,7 +99,7 @@ impl<T, D> Clone for TeeHelper<T, D> {
     }
 }
 
-impl<T, D> Debug for TeeHelper<T, D> {
+impl<T, C> Debug for TeeHelper<T, C> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let mut debug = f.debug_struct("TeeHelper");
 
diff --git a/timely/src/dataflow/operators/capture/capture.rs b/timely/src/dataflow/operators/capture/capture.rs
index c62b95417..9aa3827de 100644
--- a/timely/src/dataflow/operators/capture/capture.rs
+++ b/timely/src/dataflow/operators/capture/capture.rs
@@ -17,7 +17,7 @@ use crate::progress::Timestamp;
 use super::{EventCore, EventPusherCore};
 
 /// Capture a stream of timestamped data for later replay.
-pub trait Capture<T: Timestamp, D: Container> {
+pub trait Capture<T: Timestamp, C: Container> {
     /// Captures a stream of timestamped data for later replay.
     ///
     /// # Examples
@@ -103,18 +103,18 @@ pub trait Capture<T: Timestamp, D: Container> {
     ///
     /// assert_eq!(recv0.extract()[0].1, (0..10).collect::<Vec<_>>());
     /// ```
-    fn capture_into<P: EventPusherCore<T, D>+'static>(&self, pusher: P);
+    fn capture_into<P: EventPusherCore<T, C>+'static>(&self, pusher: P);
 
     /// Captures a stream using Rust's MPSC channels.
-    fn capture(&self) -> ::std::sync::mpsc::Receiver<EventCore<T, D>> {
+    fn capture(&self) -> ::std::sync::mpsc::Receiver<EventCore<T, C>> {
         let (send, recv) = ::std::sync::mpsc::channel();
         self.capture_into(send);
         recv
     }
 }
 
-impl<S: Scope, D: Container> Capture<S::Timestamp, D> for StreamCore<S, D> {
-    fn capture_into<P: EventPusherCore<S::Timestamp, D>+'static>(&self, mut event_pusher: P) {
+impl<S: Scope, C: Container> Capture<S::Timestamp, C> for StreamCore<S, C> {
+    fn capture_into<P: EventPusherCore<S::Timestamp, C>+'static>(&self, mut event_pusher: P) {
 
         let mut builder = OperatorBuilder::new("Capture".to_owned(), self.scope());
         let mut input = PullCounter::new(builder.new_input(self, Pipeline));
diff --git a/timely/src/dataflow/operators/capture/event.rs b/timely/src/dataflow/operators/capture/event.rs
index 252b7361c..a5b5f07b8 100644
--- a/timely/src/dataflow/operators/capture/event.rs
+++ b/timely/src/dataflow/operators/capture/event.rs
@@ -6,25 +6,25 @@
 
 /// Data and progress events of the captured stream.
 #[derive(Debug, Clone, Abomonation, Hash, Ord, PartialOrd, Eq, PartialEq, Deserialize, Serialize)]
-pub enum EventCore<T, D> {
+pub enum EventCore<T, C> {
     /// Progress received via `push_external_progress`.
     Progress(Vec<(T, i64)>),
     /// Messages received via the data stream.
-    Messages(T, D),
+    Messages(T, C),
 }
 
 /// Data and progress events of the captured stream, specialized to vector-based containers.
 pub type Event<T, D> = EventCore<T, Vec<D>>;
 
-/// Iterates over contained `EventCore<T, D>`.
+/// Iterates over contained `EventCore<T, C>`.
 ///
 /// The `EventIterator` trait describes types that can iterate over references to events,
 /// and which can be used to replay a stream into a new timely dataflow computation.
 ///
 /// This method is not simply an iterator because of the lifetime in the result.
-pub trait EventIteratorCore<T, D> {
-    /// Iterates over references to `EventCore<T, D>` elements.
-    fn next(&mut self) -> Option<&EventCore<T, D>>;
+pub trait EventIteratorCore<T, C> {
+    /// Iterates over references to `EventCore<T, C>` elements.
+    fn next(&mut self) -> Option<&EventCore<T, C>>;
 }
 
 /// A [EventIteratorCore] specialized to vector-based containers.
@@ -40,10 +40,10 @@ impl<T, D, E: EventIteratorCore<T, Vec<D>>> EventIterator<T, D> for E {
 }
 
 
-/// Receives `EventCore<T, D>` events.
-pub trait EventPusherCore<T, D> {
+/// Receives `EventCore<T, C>` events.
+pub trait EventPusherCore<T, C> {
     /// Provides a new `Event<T, D>` to the pusher.
-    fn push(&mut self, event: EventCore<T, D>);
+    fn push(&mut self, event: EventCore<T, C>);
 }
 
 /// A [EventPusherCore] specialized to vector-based containers.
@@ -53,8 +53,8 @@ impl<T, D, E: EventPusherCore<T, Vec<D>>> EventPusher<T, D> for E {}
 
 
 // implementation for the linked list behind a `Handle`.
-impl<T, D> EventPusherCore<T, D> for ::std::sync::mpsc::Sender<EventCore<T, D>> {
-    fn push(&mut self, event: EventCore<T, D>) {
+impl<T, C> EventPusherCore<T, C> for ::std::sync::mpsc::Sender<EventCore<T, C>> {
+    fn push(&mut self, event: EventCore<T, C>) {
         // NOTE: An Err(x) result just means "data not accepted" most likely
         //       because the receiver is gone. No need to panic.
         let _ = self.send(event);
@@ -69,38 +69,38 @@ pub mod link {
 
     use super::{EventCore, EventPusherCore, EventIteratorCore};
 
-    /// A linked list of EventCore<T, D>.
-    pub struct EventLinkCore<T, D> {
+    /// A linked list of EventCore<T, C>.
+    pub struct EventLinkCore<T, C> {
         /// An event, if one exists.
         ///
         /// An event might not exist, if either we want to insert a `None` and have the output iterator pause,
         /// or in the case of the very first linked list element, which has no event when constructed.
-        pub event: Option<EventCore<T, D>>,
+        pub event: Option<EventCore<T, C>>,
         /// The next event, if it exists.
-        pub next: RefCell<Option<Rc<EventLinkCore<T, D>>>>,
+        pub next: RefCell<Option<Rc<EventLinkCore<T, C>>>>,
     }
 
     /// A [EventLinkCore] specialized to vector-based containers.
     pub type EventLink<T, D> = EventLinkCore<T, Vec<D>>;
 
-    impl<T, D> EventLinkCore<T, D> {
+    impl<T, C> EventLinkCore<T, C> {
         /// Allocates a new `EventLink`.
-        pub fn new() -> EventLinkCore<T, D> {
+        pub fn new() -> EventLinkCore<T, C> {
             EventLinkCore { event: None, next: RefCell::new(None) }
         }
     }
 
     // implementation for the linked list behind a `Handle`.
-    impl<T, D> EventPusherCore<T, D> for Rc<EventLinkCore<T, D>> {
-        fn push(&mut self, event: EventCore<T, D>) {
+    impl<T, C> EventPusherCore<T, C> for Rc<EventLinkCore<T, C>> {
+        fn push(&mut self, event: EventCore<T, C>) {
             *self.next.borrow_mut() = Some(Rc::new(EventLinkCore { event: Some(event), next: RefCell::new(None) }));
             let next = self.next.borrow().as_ref().unwrap().clone();
             *self = next;
         }
     }
 
-    impl<T, D> EventIteratorCore<T, D> for Rc<EventLinkCore<T, D>> {
-        fn next(&mut self) -> Option<&EventCore<T, D>> {
+    impl<T, C> EventIteratorCore<T, C> for Rc<EventLinkCore<T, C>> {
+        fn next(&mut self) -> Option<&EventCore<T, C>> {
             let is_some = self.next.borrow().is_some();
             if is_some {
                 let next = self.next.borrow().as_ref().unwrap().clone();
@@ -114,7 +114,7 @@ pub mod link {
     }
 
     // Drop implementation to prevent stack overflow through naive drop impl.
-    impl<T, D> Drop for EventLinkCore<T, D> {
+    impl<T, C> Drop for EventLinkCore<T, C> {
         fn drop(&mut self) {
             while let Some(link) = self.next.replace(None) {
                 if let Ok(head) = Rc::try_unwrap(link) {
@@ -124,7 +124,7 @@ pub mod link {
         }
     }
 
-    impl<T, D> Default for EventLinkCore<T, D> {
+    impl<T, C> Default for EventLinkCore<T, C> {
         fn default() -> Self {
             Self::new()
         }
@@ -147,16 +147,16 @@ pub mod binary {
     use abomonation::Abomonation;
     use super::{EventCore, EventPusherCore, EventIteratorCore};
 
-    /// A wrapper for `W: Write` implementing `EventPusherCore<T, D>`.
-    pub struct EventWriterCore<T, D, W: ::std::io::Write> {
+    /// A wrapper for `W: Write` implementing `EventPusherCore<T, C>`.
+    pub struct EventWriterCore<T, C, W: ::std::io::Write> {
         stream: W,
-        phant: ::std::marker::PhantomData<(T,D)>,
+        phant: ::std::marker::PhantomData<(T, C)>,
     }
 
     /// [EventWriterCore] specialized to vector-based containers.
     pub type EventWriter<T, D, W> = EventWriterCore<T, Vec<D>, W>;
 
-    impl<T, D, W: ::std::io::Write> EventWriterCore<T, D, W> {
+    impl<T, C, W: ::std::io::Write> EventWriterCore<T, C, W> {
         /// Allocates a new `EventWriter` wrapping a supplied writer.
         pub fn new(w: W) -> Self {
             Self {
@@ -166,28 +166,28 @@ pub mod binary {
         }
     }
 
-    impl<T: Abomonation, D: Abomonation, W: ::std::io::Write> EventPusherCore<T, D> for EventWriterCore<T, D, W> {
-        fn push(&mut self, event: EventCore<T, D>) {
+    impl<T: Abomonation, C: Abomonation, W: ::std::io::Write> EventPusherCore<T, C> for EventWriterCore<T, C, W> {
+        fn push(&mut self, event: EventCore<T, C>) {
             // TODO: `push` has no mechanism to report errors, so we `unwrap`.
             unsafe { ::abomonation::encode(&event, &mut self.stream).expect("Event abomonation/write failed"); }
         }
     }
 
     /// A Wrapper for `R: Read` implementing `EventIterator<T, D>`.
-    pub struct EventReaderCore<T, D, R: ::std::io::Read> {
+    pub struct EventReaderCore<T, C, R: ::std::io::Read> {
         reader: R,
         bytes: Vec<u8>,
         buff1: Vec<u8>,
         buff2: Vec<u8>,
         consumed: usize,
         valid: usize,
-        phant: ::std::marker::PhantomData<(T,D)>,
+        phant: ::std::marker::PhantomData<(T, C)>,
     }
 
     /// [EventReaderCore] specialized to vector-based containers.
     pub type EventReader<T, D, R> = EventReaderCore<T, Vec<D>, R>;
 
-    impl<T, D, R: ::std::io::Read> EventReaderCore<T, D, R> {
+    impl<T, C, R: ::std::io::Read> EventReaderCore<T, C, R> {
         /// Allocates a new `EventReader` wrapping a supplied reader.
         pub fn new(r: R) -> Self {
             Self {
@@ -202,12 +202,12 @@ pub mod binary {
         }
     }
 
-    impl<T: Abomonation, D: Abomonation, R: ::std::io::Read> EventIteratorCore<T, D> for EventReaderCore<T, D, R> {
-        fn next(&mut self) -> Option<&EventCore<T, D>> {
+    impl<T: Abomonation, C: Abomonation, R: ::std::io::Read> EventIteratorCore<T, C> for EventReaderCore<T, C, R> {
+        fn next(&mut self) -> Option<&EventCore<T, C>> {
 
             // if we can decode something, we should just return it! :D
-            if unsafe { ::abomonation::decode::<EventCore<T,D>>(&mut self.buff1[self.consumed..]) }.is_some() {
-                let (item, rest) = unsafe { ::abomonation::decode::<EventCore<T,D>>(&mut self.buff1[self.consumed..]) }.unwrap();
+            if unsafe { ::abomonation::decode::<EventCore<T, C>>(&mut self.buff1[self.consumed..]) }.is_some() {
+                let (item, rest) = unsafe { ::abomonation::decode::<EventCore<T, C>>(&mut self.buff1[self.consumed..]) }.unwrap();
                 self.consumed = self.valid - rest.len();
                 return Some(item);
             }
diff --git a/timely/src/dataflow/operators/concat.rs b/timely/src/dataflow/operators/concat.rs
index 449609f5b..f69fd98a7 100644
--- a/timely/src/dataflow/operators/concat.rs
+++ b/timely/src/dataflow/operators/concat.rs
@@ -6,7 +6,7 @@ use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{StreamCore, Scope};
 
 /// Merge the contents of two streams.
-pub trait Concat<G: Scope, D: Container> {
+pub trait Concat<G: Scope, C: Container> {
     /// Merge the contents of two streams.
     ///
     /// # Examples
@@ -20,17 +20,17 @@ pub trait Concat<G: Scope, D: Container> {
     ///           .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```
-    fn concat(&self, _: &StreamCore<G, D>) -> StreamCore<G, D>;
+    fn concat(&self, _: &StreamCore<G, C>) -> StreamCore<G, C>;
 }
 
-impl<G: Scope, D: Container> Concat<G, D> for StreamCore<G, D> {
-    fn concat(&self, other: &StreamCore<G, D>) -> StreamCore<G, D> {
+impl<G: Scope, C: Container> Concat<G, C> for StreamCore<G, C> {
+    fn concat(&self, other: &StreamCore<G, C>) -> StreamCore<G, C> {
         self.scope().concatenate([self.clone(), other.clone()])
     }
 }
 
 /// Merge the contents of multiple streams.
-pub trait Concatenate<G: Scope, D: Container> {
+pub trait Concatenate<G: Scope, C: Container> {
     /// Merge the contents of multiple streams.
     ///
     /// # Examples
@@ -47,25 +47,25 @@ pub trait Concatenate<G: Scope, D: Container> {
     ///          .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```
-    fn concatenate<I>(&self, sources: I) -> StreamCore<G, D>
+    fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
     where
-        I: IntoIterator<Item=StreamCore<G, D>>;
+        I: IntoIterator<Item=StreamCore<G, C>>;
 }
 
-impl<G: Scope, D: Container> Concatenate<G, D> for StreamCore<G, D> {
-    fn concatenate<I>(&self, sources: I) -> StreamCore<G, D>
+impl<G: Scope, C: Container> Concatenate<G, C> for StreamCore<G, C> {
+    fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
     where
-        I: IntoIterator<Item=StreamCore<G, D>>
+        I: IntoIterator<Item=StreamCore<G, C>>
     {
         let clone = self.clone();
         self.scope().concatenate(Some(clone).into_iter().chain(sources))
     }
 }
 
-impl<G: Scope, D: Container> Concatenate<G, D> for G {
-    fn concatenate<I>(&self, sources: I) -> StreamCore<G, D>
+impl<G: Scope, C: Container> Concatenate<G, C> for G {
+    fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
     where
-        I: IntoIterator<Item=StreamCore<G, D>>
+        I: IntoIterator<Item=StreamCore<G, C>>
     {
 
         // create an operator builder.
diff --git a/timely/src/dataflow/operators/enterleave.rs b/timely/src/dataflow/operators/enterleave.rs
index af2eaf00f..97ada7ae2 100644
--- a/timely/src/dataflow/operators/enterleave.rs
+++ b/timely/src/dataflow/operators/enterleave.rs
@@ -115,7 +115,7 @@ impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Data+Container> Enter<G, T
 }
 
 /// Extension trait to move a `Stream` to the parent of its current `Scope`.
-pub trait Leave<G: Scope, D: Container> {
+pub trait Leave<G: Scope, C: Container> {
     /// Moves a `Stream` to the parent of its current `Scope`.
     ///
     /// # Examples
@@ -130,17 +130,17 @@ pub trait Leave<G: Scope, D: Container> {
     ///     });
     /// });
     /// ```
-    fn leave(&self) -> StreamCore<G, D>;
+    fn leave(&self) -> StreamCore<G, C>;
 }
 
-impl<'a, G: Scope, D: Clone+Container, T: Timestamp+Refines<G::Timestamp>> Leave<G, D> for StreamCore<Child<'a, G, T>, D> {
-    fn leave(&self) -> StreamCore<G, D> {
+impl<'a, G: Scope, C: Clone+Container, T: Timestamp+Refines<G::Timestamp>> Leave<G, C> for StreamCore<Child<'a, G, T>, C> {
+    fn leave(&self) -> StreamCore<G, C> {
 
         let scope = self.scope();
 
         let output = scope.subgraph.borrow_mut().new_output();
         let target = Target::new(0, output.port);
-        let (targets, registrar) = Tee::<G::Timestamp, D>::new();
+        let (targets, registrar) = Tee::<G::Timestamp, C>::new();
         let egress = EgressNub { targets, phantom: PhantomData };
         let channel_id = scope.clone().new_identifier();
 
@@ -160,15 +160,15 @@ impl<'a, G: Scope, D: Clone+Container, T: Timestamp+Refines<G::Timestamp>> Leave
 }
 
 
-struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> {
-    targets: Counter<TInner, TData, Tee<TInner, TData>>,
+struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> {
+    targets: Counter<TInner, TContainer, Tee<TInner, TContainer>>,
     phantom: ::std::marker::PhantomData<TOuter>,
     activator: crate::scheduling::Activator,
     active: bool,
 }
 
-impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> Push<Bundle<TOuter, TData>> for IngressNub<TOuter, TInner, TData> {
-    fn push(&mut self, element: &mut Option<Bundle<TOuter, TData>>) {
+impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> Push<Bundle<TOuter, TContainer>> for IngressNub<TOuter, TInner, TContainer> {
+    fn push(&mut self, element: &mut Option<Bundle<TOuter, TContainer>>) {
         if let Some(message) = element {
             let outer_message = message.as_mut();
             let data = ::std::mem::take(&mut outer_message.data);
@@ -192,14 +192,14 @@ impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> Pus
 }
 
 
-struct EgressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Data> {
-    targets: Tee<TOuter, TData>,
+struct EgressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Data> {
+    targets: Tee<TOuter, TContainer>,
     phantom: PhantomData<TInner>,
 }
 
-impl<TOuter, TInner, TData: Container> Push<Bundle<TInner, TData>> for EgressNub<TOuter, TInner, TData>
-where TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Data {
-    fn push(&mut self, message: &mut Option<Bundle<TInner, TData>>) {
+impl<TOuter, TInner, TContainer: Container> Push<Bundle<TInner, TContainer>> for EgressNub<TOuter, TInner, TContainer>
+where TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Data {
+    fn push(&mut self, message: &mut Option<Bundle<TInner, TContainer>>) {
         if let Some(message) = message {
             let inner_message = message.as_mut();
             let data = ::std::mem::take(&mut inner_message.data);
@@ -241,12 +241,12 @@ impl<P> LogPusher<P> {
     }
 }
 
-impl<T, D, P> Push<Bundle<T, D>> for LogPusher<P>
+impl<T, C, P> Push<Bundle<T, C>> for LogPusher<P>
 where
-    D: Container,
-    P: Push<Bundle<T, D>>,
+    C: Container,
+    P: Push<Bundle<T, C>>,
 {
-    fn push(&mut self, element: &mut Option<Bundle<T, D>>) {
+    fn push(&mut self, element: &mut Option<Bundle<T, C>>) {
         if let Some(bundle) = element {
             let send_event = MessagesEvent {
                 is_send: true,
diff --git a/timely/src/dataflow/operators/feedback.rs b/timely/src/dataflow/operators/feedback.rs
index 038ca7e09..80e623df1 100644
--- a/timely/src/dataflow/operators/feedback.rs
+++ b/timely/src/dataflow/operators/feedback.rs
@@ -59,7 +59,7 @@ pub trait Feedback<G: Scope> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn feedback_core<D: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, D>, StreamCore<G, D>);
+    fn feedback_core<C: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, C>, StreamCore<G, C>);
 }
 
 /// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
@@ -87,7 +87,7 @@ pub trait LoopVariable<'a, G: Scope, T: Timestamp> {
     ///     });
     /// });
     /// ```
-    fn loop_variable<D: Container>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, D>, StreamCore<Iterative<'a, G, T>, D>);
+    fn loop_variable<C: Container>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>);
 }
 
 impl<G: Scope> Feedback<G> for G {
@@ -95,7 +95,7 @@ impl<G: Scope> Feedback<G> for G {
         self.feedback_core(summary)
     }
 
-    fn feedback_core<D: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, D>, StreamCore<G, D>) {
+    fn feedback_core<C: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, C>, StreamCore<G, C>) {
 
         let mut builder = OperatorBuilder::new("Feedback".to_owned(), self.clone());
         let (output, stream) = builder.new_output();
@@ -105,13 +105,13 @@ impl<G: Scope> Feedback<G> for G {
 }
 
 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) -> (HandleCore<Iterative<'a, G, T>, D>, StreamCore<Iterative<'a, G, T>, D>) {
+    fn loop_variable<C: Container>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>) {
         self.feedback_core(Product::new(Default::default(), summary))
     }
 }
 
 /// Connect a `Stream` to the input of a loop variable.
-pub trait ConnectLoop<G: Scope, D: Container> {
+pub trait ConnectLoop<G: Scope, C: Container> {
     /// Connect a `Stream` to be the input of a loop variable.
     ///
     /// # Examples
@@ -129,11 +129,11 @@ pub trait ConnectLoop<G: Scope, D: Container> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn connect_loop(&self, _: HandleCore<G, D>);
+    fn connect_loop(&self, _: HandleCore<G, C>);
 }
 
-impl<G: Scope, D: Container> ConnectLoop<G, D> for StreamCore<G, D> {
-    fn connect_loop(&self, helper: HandleCore<G, D>) {
+impl<G: Scope, C: Container> ConnectLoop<G, C> for StreamCore<G, C> {
+    fn connect_loop(&self, helper: HandleCore<G, C>) {
 
         let mut builder = helper.builder;
         let summary = helper.summary;
@@ -159,10 +159,10 @@ impl<G: Scope, D: Container> ConnectLoop<G, D> for StreamCore<G, D> {
 
 /// A handle used to bind the source of a loop variable.
 #[derive(Debug)]
-pub struct HandleCore<G: Scope, D: Container> {
+pub struct HandleCore<G: Scope, C: Container> {
     builder: OperatorBuilder<G>,
     summary: <G::Timestamp as Timestamp>::Summary,
-    output: OutputWrapper<G::Timestamp, D, Tee<G::Timestamp, D>>,
+    output: OutputWrapper<G::Timestamp, C, Tee<G::Timestamp, C>>,
 }
 
 /// A `HandleCore` specialized for using `Vec` as container
diff --git a/timely/src/dataflow/operators/generic/builder_raw.rs b/timely/src/dataflow/operators/generic/builder_raw.rs
index 10899b9be..9522b6f34 100644
--- a/timely/src/dataflow/operators/generic/builder_raw.rs
+++ b/timely/src/dataflow/operators/generic/builder_raw.rs
@@ -105,17 +105,17 @@ 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: &StreamCore<G, D>, pact: P) -> P::Puller
+    pub fn new_input<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P) -> P::Puller
         where
-            P: ParallelizationContract<G::Timestamp, D> {
+            P: ParallelizationContract<G::Timestamp, C> {
         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: Container, P>(&mut self, stream: &StreamCore<G, D>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> P::Puller
+    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> P::Puller
     where
-        P: ParallelizationContract<G::Timestamp, D> {
+        P: ParallelizationContract<G::Timestamp, C> {
 
         let channel_id = self.scope.new_identifier();
         let logging = self.scope.logging();
@@ -131,16 +131,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>, StreamCore<G, D>) {
+    pub fn new_output<C: Container>(&mut self) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
 
         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>, StreamCore<G, D>) {
+    pub fn new_output_connection<C: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
 
-        let (targets, registrar) = Tee::<G::Timestamp,D>::new();
+        let (targets, registrar) = Tee::<G::Timestamp,C>::new();
         let source = Source::new(self.index, self.shape.outputs);
         let stream = StreamCore::new(source, registrar, self.scope.clone());
 
diff --git a/timely/src/dataflow/operators/generic/builder_rc.rs b/timely/src/dataflow/operators/generic/builder_rc.rs
index f753ef09a..3d2ebbe29 100644
--- a/timely/src/dataflow/operators/generic/builder_rc.rs
+++ b/timely/src/dataflow/operators/generic/builder_rc.rs
@@ -59,9 +59,9 @@ 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: &StreamCore<G, D>, pact: P) -> InputHandleCore<G::Timestamp, D, P::Puller>
+    pub fn new_input<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P) -> InputHandleCore<G::Timestamp, C, P::Puller>
     where
-        P: ParallelizationContract<G::Timestamp, D> {
+        P: ParallelizationContract<G::Timestamp, C> {
 
         let connection = vec![Antichain::from_elem(Default::default()); self.builder.shape().outputs()];
         self.new_input_connection(stream, pact, connection)
@@ -75,9 +75,9 @@ 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_input_connection<D: Container, P>(&mut self, stream: &StreamCore<G, D>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> InputHandleCore<G::Timestamp, D, P::Puller>
+    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> InputHandleCore<G::Timestamp, C, P::Puller>
         where
-            P: ParallelizationContract<G::Timestamp, D> {
+            P: ParallelizationContract<G::Timestamp, C> {
 
         let puller = self.builder.new_input_connection(stream, pact, connection.clone());
 
@@ -92,7 +92,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>>, StreamCore<G, D>) {
+    pub fn new_output<C: Container>(&mut self) -> (OutputWrapper<G::Timestamp, C, Tee<G::Timestamp, C>>, StreamCore<G, C>) {
         let connection = vec![Antichain::from_elem(Default::default()); self.builder.shape().inputs()];
         self.new_output_connection(connection)
     }
@@ -105,7 +105,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>>, StreamCore<G, D>) {
+    pub fn new_output_connection<C: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (OutputWrapper<G::Timestamp, C, Tee<G::Timestamp, C>>, StreamCore<G, C>) {
 
         let (tee, stream) = self.builder.new_output_connection(connection.clone());
 
diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs
index b5157f4ce..ee4059bb4 100644
--- a/timely/src/dataflow/operators/generic/handles.rs
+++ b/timely/src/dataflow/operators/generic/handles.rs
@@ -22,8 +22,8 @@ use crate::dataflow::operators::InputCapability;
 use crate::dataflow::operators::capability::CapabilityTrait;
 
 /// Handle to an operator's input stream.
-pub struct InputHandleCore<T: Timestamp, D: Container, P: Pull<Bundle<T, D>>> {
-    pull_counter: PullCounter<T, D, P>,
+pub struct InputHandleCore<T: Timestamp, C: Container, P: Pull<Bundle<T, C>>> {
+    pull_counter: PullCounter<T, C, P>,
     internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>,
     /// Timestamp summaries from this input to each output.
     ///
@@ -37,9 +37,9 @@ pub struct InputHandleCore<T: Timestamp, D: Container, P: Pull<Bundle<T, D>>> {
 pub type InputHandle<T, D, P> = InputHandleCore<T, Vec<D>, P>;
 
 /// Handle to an operator's input stream and frontier.
-pub struct FrontieredInputHandleCore<'a, T: Timestamp, D: Container+'a, P: Pull<Bundle<T, D>>+'a> {
+pub struct FrontieredInputHandleCore<'a, T: Timestamp, C: Container+'a, P: Pull<Bundle<T, C>>+'a> {
     /// The underlying input handle.
-    pub handle: &'a mut InputHandleCore<T, D, P>,
+    pub handle: &'a mut InputHandleCore<T, C, P>,
     /// The frontier as reported by timely progress tracking.
     pub frontier: &'a MutableAntichain<T>,
 }
@@ -47,13 +47,13 @@ pub struct FrontieredInputHandleCore<'a, T: Timestamp, D: Container+'a, P: Pull<
 /// 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<'a, T: Timestamp, D: Container, P: Pull<Bundle<T, D>>> InputHandleCore<T, D, P> {
+impl<'a, T: Timestamp, C: Container, P: Pull<Bundle<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.
     /// Returns `None` when there's no more data available.
     #[inline]
-    pub fn next(&mut self) -> Option<(InputCapability<T>, RefOrMut<D>)> {
+    pub fn next(&mut self) -> Option<(InputCapability<T>, RefOrMut<C>)> {
         let internal = &self.internal;
         let summaries = &self.summaries;
         self.pull_counter.next_guarded().map(|(guard, bundle)| {
@@ -87,7 +87,7 @@ impl<'a, T: Timestamp, D: Container, P: Pull<Bundle<T, D>>> InputHandleCore<T, D
     /// });
     /// ```
     #[inline]
-    pub fn for_each<F: FnMut(InputCapability<T>, RefOrMut<D>)>(&mut self, mut logic: F) {
+    pub fn for_each<F: FnMut(InputCapability<T>, RefOrMut<C>)>(&mut self, mut logic: F) {
         let mut logging = self.logging.take();
         while let Some((cap, data)) = self.next() {
             logging.as_mut().map(|l| l.log(crate::logging::GuardedMessageEvent { is_start: true }));
@@ -99,9 +99,9 @@ impl<'a, T: Timestamp, D: Container, P: Pull<Bundle<T, D>>> InputHandleCore<T, D
 
 }
 
-impl<'a, T: Timestamp, D: Container, P: Pull<Bundle<T, D>>+'a> FrontieredInputHandleCore<'a, T, D, P> {
+impl<'a, T: Timestamp, C: Container, P: Pull<Bundle<T, C>>+'a> FrontieredInputHandleCore<'a, T, C, P> {
     /// Allocate a new frontiered input handle.
-    pub fn new(handle: &'a mut InputHandleCore<T, D, P>, frontier: &'a MutableAntichain<T>) -> Self {
+    pub fn new(handle: &'a mut InputHandleCore<T, C, P>, frontier: &'a MutableAntichain<T>) -> Self {
         FrontieredInputHandleCore {
             handle,
             frontier,
@@ -112,7 +112,7 @@ impl<'a, T: Timestamp, D: Container, P: Pull<Bundle<T, D>>+'a> FrontieredInputHa
     /// The timestamp `t` of the input buffer can be retrieved by invoking `.time()` on the capability.
     /// Returns `None` when there's no more data available.
     #[inline]
-    pub fn next(&mut self) -> Option<(InputCapability<T>, RefOrMut<D>)> {
+    pub fn next(&mut self) -> Option<(InputCapability<T>, RefOrMut<C>)> {
         self.handle.next()
     }
 
@@ -135,7 +135,7 @@ impl<'a, T: Timestamp, D: Container, P: Pull<Bundle<T, D>>+'a> FrontieredInputHa
     /// });
     /// ```
     #[inline]
-    pub fn for_each<F: FnMut(InputCapability<T>, RefOrMut<D>)>(&mut self, logic: F) {
+    pub fn for_each<F: FnMut(InputCapability<T>, RefOrMut<C>)>(&mut self, logic: F) {
         self.handle.for_each(logic)
     }
 
@@ -146,18 +146,18 @@ impl<'a, T: Timestamp, D: Container, P: Pull<Bundle<T, D>>+'a> FrontieredInputHa
     }
 }
 
-pub fn _access_pull_counter<T: Timestamp, D: Container, P: Pull<Bundle<T, D>>>(input: &mut InputHandleCore<T, D, P>) -> &mut PullCounter<T, D, P> {
+pub fn _access_pull_counter<T: Timestamp, C: Container, P: Pull<Bundle<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, D: Container, P: Pull<Bundle<T, D>>>(
-    pull_counter: PullCounter<T, D, P>, 
+pub fn new_input_handle<T: Timestamp, C: Container, P: Pull<Bundle<T, C>>>(
+    pull_counter: PullCounter<T, C, P>, 
     internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>, 
     summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>, 
     logging: Option<Logger>
-) -> InputHandleCore<T, D, P> {
+) -> InputHandleCore<T, C, P> {
     InputHandleCore {
         pull_counter,
         internal,
@@ -172,14 +172,14 @@ pub fn new_input_handle<T: Timestamp, D: Container, P: Pull<Bundle<T, D>>>(
 /// than with an `OutputHandle`, whose methods ensure that capabilities are used and that the
 /// pusher is flushed (via the `cease` method) once it is no longer used.
 #[derive(Debug)]
-pub struct OutputWrapper<T: Timestamp, D: Container, P: Push<Bundle<T, D>>> {
-    push_buffer: Buffer<T, D, PushCounter<T, D, P>>,
+pub struct OutputWrapper<T: Timestamp, C: Container, P: Push<Bundle<T, C>>> {
+    push_buffer: Buffer<T, C, PushCounter<T, C, P>>,
     internal_buffer: Rc<RefCell<ChangeBatch<T>>>,
 }
 
-impl<T: Timestamp, D: Container, P: Push<Bundle<T, D>>> OutputWrapper<T, D, P> {
+impl<T: Timestamp, C: Container, P: Push<Bundle<T, C>>> OutputWrapper<T, C, P> {
     /// Creates a new output wrapper from a push buffer.
-    pub fn new(push_buffer: Buffer<T, D, PushCounter<T, D, P>>, internal_buffer: Rc<RefCell<ChangeBatch<T>>>) -> Self {
+    pub fn new(push_buffer: Buffer<T, C, PushCounter<T, C, P>>, internal_buffer: Rc<RefCell<ChangeBatch<T>>>) -> Self {
         OutputWrapper {
             push_buffer,
             internal_buffer,
@@ -189,7 +189,7 @@ 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) -> OutputHandleCore<T, D, P> {
+    pub fn activate(&mut self) -> OutputHandleCore<T, C, P> {
         OutputHandleCore {
             push_buffer: &mut self.push_buffer,
             internal_buffer: &self.internal_buffer,
diff --git a/timely/src/dataflow/operators/generic/operator.rs b/timely/src/dataflow/operators/generic/operator.rs
index 5ba6a211f..45571153a 100644
--- a/timely/src/dataflow/operators/generic/operator.rs
+++ b/timely/src/dataflow/operators/generic/operator.rs
@@ -15,7 +15,7 @@ use crate::dataflow::operators::generic::notificator::{Notificator, FrontierNoti
 use crate::Container;
 
 /// Methods to construct generic streaming and blocking operators.
-pub trait Operator<G: Scope, D1: Container> {
+pub trait Operator<G: Scope, C1: Container> {
     /// Creates a new dataflow operator that partitions its input stream by a parallelization
     /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
     /// `logic` can read from the input stream, write to the output stream, and inspect the frontier at the input.
@@ -55,13 +55,13 @@ pub trait Operator<G: Scope, D1: Container> {
     ///     });
     /// }
     /// ```
-    fn unary_frontier<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, D2>
+    fn unary_frontier<C2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, C2>
     where
-        D2: Container,
+        C2: Container,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P::Puller>,
-                 &mut OutputHandleCore<G::Timestamp, D2, Tee<G::Timestamp, D2>>)+'static,
-        P: ParallelizationContract<G::Timestamp, D1>;
+        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, C1, P::Puller>,
+                 &mut OutputHandleCore<G::Timestamp, C2, Tee<G::Timestamp, C2>>)+'static,
+        P: ParallelizationContract<G::Timestamp, C1>;
 
     /// Creates a new dataflow operator that partitions its input stream by a parallelization
     /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
@@ -92,12 +92,12 @@ pub trait Operator<G: Scope, D1: Container> {
     ///     });
     /// }
     /// ```
-    fn unary_notify<D2: Container,
-            L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>,
-                     &mut OutputHandleCore<G::Timestamp, D2, Tee<G::Timestamp, D2>>,
+    fn unary_notify<C2: Container,
+            L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P::Puller>,
+                     &mut OutputHandleCore<G::Timestamp, C2, Tee<G::Timestamp, C2>>,
                      &mut Notificator<G::Timestamp>)+'static,
-             P: ParallelizationContract<G::Timestamp, D1>>
-             (&self, pact: P, name: &str, init: impl IntoIterator<Item=G::Timestamp>, logic: L) -> StreamCore<G, D2>;
+             P: ParallelizationContract<G::Timestamp, C1>>
+             (&self, pact: P, name: &str, init: impl IntoIterator<Item=G::Timestamp>, logic: L) -> StreamCore<G, C2>;
 
     /// Creates a new dataflow operator that partitions its input stream by a parallelization
     /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
@@ -127,13 +127,13 @@ pub trait Operator<G: Scope, D1: Container> {
     ///         });
     /// });
     /// ```
-    fn unary<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, D2>
+    fn unary<C2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, C2>
     where
-        D2: Container,
+        C2: Container,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-        L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>,
-                 &mut OutputHandleCore<G::Timestamp, D2, Tee<G::Timestamp, D2>>)+'static,
-        P: ParallelizationContract<G::Timestamp, D1>;
+        L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P::Puller>,
+                 &mut OutputHandleCore<G::Timestamp, C2, Tee<G::Timestamp, C2>>)+'static,
+        P: ParallelizationContract<G::Timestamp, C1>;
 
     /// Creates a new dataflow operator that partitions its input streams by a parallelization
     /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
@@ -185,16 +185,16 @@ pub trait Operator<G: Scope, D1: Container> {
     ///    }
     /// }).unwrap();
     /// ```
-    fn binary_frontier<D2, D3, B, L, P1, P2>(&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, D3>
+    fn binary_frontier<C2, C3, B, L, P1, P2>(&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, C3>
     where
-        D2: Container,
-        D3: Container,
+        C2: Container,
+        C3: Container,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P1::Puller>,
-                 &mut FrontieredInputHandleCore<G::Timestamp, D2, P2::Puller>,
-                 &mut OutputHandleCore<G::Timestamp, D3, Tee<G::Timestamp, D3>>)+'static,
-        P1: ParallelizationContract<G::Timestamp, D1>,
-        P2: ParallelizationContract<G::Timestamp, D2>;
+        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, C1, P1::Puller>,
+                 &mut FrontieredInputHandleCore<G::Timestamp, C2, P2::Puller>,
+                 &mut OutputHandleCore<G::Timestamp, C3, Tee<G::Timestamp, C3>>)+'static,
+        P1: ParallelizationContract<G::Timestamp, C1>,
+        P2: ParallelizationContract<G::Timestamp, C2>;
 
     /// Creates a new dataflow operator that partitions its input streams by a parallelization
     /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
@@ -241,15 +241,15 @@ pub trait Operator<G: Scope, D1: Container> {
     ///    }
     /// }).unwrap();
     /// ```
-    fn binary_notify<D2: Container,
-              D3: Container,
-              L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>,
-                       &mut InputHandleCore<G::Timestamp, D2, P2::Puller>,
-                       &mut OutputHandleCore<G::Timestamp, D3, Tee<G::Timestamp, D3>>,
+    fn binary_notify<C2: Container,
+              C3: Container,
+              L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P1::Puller>,
+                       &mut InputHandleCore<G::Timestamp, C2, P2::Puller>,
+                       &mut OutputHandleCore<G::Timestamp, C3, Tee<G::Timestamp, C3>>,
                        &mut Notificator<G::Timestamp>)+'static,
-              P1: ParallelizationContract<G::Timestamp, D1>,
-              P2: ParallelizationContract<G::Timestamp, D2>>
-            (&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator<Item=G::Timestamp>, logic: L) -> StreamCore<G, D3>;
+              P1: ParallelizationContract<G::Timestamp, C1>,
+              P2: ParallelizationContract<G::Timestamp, C2>>
+            (&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator<Item=G::Timestamp>, logic: L) -> StreamCore<G, C3>;
 
     /// Creates a new dataflow operator that partitions its input streams by a parallelization
     /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
@@ -285,16 +285,16 @@ pub trait Operator<G: Scope, D1: Container> {
     ///         }).inspect(|x| println!("{:?}", x));
     /// });
     /// ```
-    fn binary<D2, D3, B, L, P1, P2>(&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, D3>
+    fn binary<C2, C3, B, L, P1, P2>(&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, C3>
     where
-        D2: Container,
-        D3: Container,
+        C2: Container,
+        C3: Container,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-        L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>,
-                 &mut InputHandleCore<G::Timestamp, D2, P2::Puller>,
-                 &mut OutputHandleCore<G::Timestamp, D3, Tee<G::Timestamp, D3>>)+'static,
-        P1: ParallelizationContract<G::Timestamp, D1>,
-        P2: ParallelizationContract<G::Timestamp, D2>;
+        L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P1::Puller>,
+                 &mut InputHandleCore<G::Timestamp, C2, P2::Puller>,
+                 &mut OutputHandleCore<G::Timestamp, C3, Tee<G::Timestamp, C3>>)+'static,
+        P1: ParallelizationContract<G::Timestamp, C1>,
+        P2: ParallelizationContract<G::Timestamp, C2>;
 
     /// Creates a new dataflow operator that partitions its input stream by a parallelization
     /// strategy `pact`, and repeatedly invokes the function `logic` which can read from the input stream
@@ -321,19 +321,19 @@ pub trait Operator<G: Scope, D1: Container> {
     /// ```
     fn sink<L, P>(&self, pact: P, name: &str, logic: L)
     where
-        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P::Puller>)+'static,
-        P: ParallelizationContract<G::Timestamp, D1>;
+        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, C1, P::Puller>)+'static,
+        P: ParallelizationContract<G::Timestamp, C1>;
 }
 
-impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
+impl<G: Scope, C1: Container> Operator<G, C1> for StreamCore<G, C1> {
 
-    fn unary_frontier<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, D2>
+    fn unary_frontier<C2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, C2>
     where
-        D2: Container,
+        C2: Container,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P::Puller>,
-                 &mut OutputHandleCore<G::Timestamp, D2, Tee<G::Timestamp, D2>>)+'static,
-        P: ParallelizationContract<G::Timestamp, D1> {
+        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, C1, P::Puller>,
+                 &mut OutputHandleCore<G::Timestamp, C2, Tee<G::Timestamp, C2>>)+'static,
+        P: ParallelizationContract<G::Timestamp, C1> {
 
         let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
         let operator_info = builder.operator_info();
@@ -355,12 +355,12 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
         stream
     }
 
-    fn unary_notify<D2: Container,
-            L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>,
-                     &mut OutputHandleCore<G::Timestamp, D2, Tee<G::Timestamp, D2>>,
+    fn unary_notify<C2: Container,
+            L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P::Puller>,
+                     &mut OutputHandleCore<G::Timestamp, C2, Tee<G::Timestamp, C2>>,
                      &mut Notificator<G::Timestamp>)+'static,
-             P: ParallelizationContract<G::Timestamp, D1>>
-             (&self, pact: P, name: &str, init: impl IntoIterator<Item=G::Timestamp>, mut logic: L) -> StreamCore<G, D2> {
+             P: ParallelizationContract<G::Timestamp, C1>>
+             (&self, pact: P, name: &str, init: impl IntoIterator<Item=G::Timestamp>, mut logic: L) -> StreamCore<G, C2> {
 
         self.unary_frontier(pact, name, move |capability, _info| {
             let mut notificator = FrontierNotificator::new();
@@ -377,13 +377,13 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
         })
     }
 
-    fn unary<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, D2>
+    fn unary<C2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, C2>
     where
-        D2: Container,
+        C2: Container,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-        L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>,
-                 &mut OutputHandleCore<G::Timestamp, D2, Tee<G::Timestamp, D2>>)+'static,
-        P: ParallelizationContract<G::Timestamp, D1> {
+        L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P::Puller>,
+                 &mut OutputHandleCore<G::Timestamp, C2, Tee<G::Timestamp, C2>>)+'static,
+        P: ParallelizationContract<G::Timestamp, C1> {
 
         let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
         let operator_info = builder.operator_info();
@@ -405,16 +405,16 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
         stream
     }
 
-    fn binary_frontier<D2, D3, B, L, P1, P2>(&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, D3>
+    fn binary_frontier<C2, C3, B, L, P1, P2>(&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, C3>
     where
-        D2: Container,
-        D3: Container,
+        C2: Container,
+        C3: Container,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P1::Puller>,
-                 &mut FrontieredInputHandleCore<G::Timestamp, D2, P2::Puller>,
-                 &mut OutputHandleCore<G::Timestamp, D3, Tee<G::Timestamp, D3>>)+'static,
-        P1: ParallelizationContract<G::Timestamp, D1>,
-        P2: ParallelizationContract<G::Timestamp, D2> {
+        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, C1, P1::Puller>,
+                 &mut FrontieredInputHandleCore<G::Timestamp, C2, P2::Puller>,
+                 &mut OutputHandleCore<G::Timestamp, C3, Tee<G::Timestamp, C3>>)+'static,
+        P1: ParallelizationContract<G::Timestamp, C1>,
+        P2: ParallelizationContract<G::Timestamp, C2> {
 
         let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
         let operator_info = builder.operator_info();
@@ -438,15 +438,15 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
         stream
     }
 
-    fn binary_notify<D2: Container,
-              D3: Container,
-              L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>,
-                       &mut InputHandleCore<G::Timestamp, D2, P2::Puller>,
-                       &mut OutputHandleCore<G::Timestamp, D3, Tee<G::Timestamp, D3>>,
+    fn binary_notify<C2: Container,
+              C3: Container,
+              L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P1::Puller>,
+                       &mut InputHandleCore<G::Timestamp, C2, P2::Puller>,
+                       &mut OutputHandleCore<G::Timestamp, C3, Tee<G::Timestamp, C3>>,
                        &mut Notificator<G::Timestamp>)+'static,
-              P1: ParallelizationContract<G::Timestamp, D1>,
-              P2: ParallelizationContract<G::Timestamp, D2>>
-            (&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator<Item=G::Timestamp>, mut logic: L) -> StreamCore<G, D3> {
+              P1: ParallelizationContract<G::Timestamp, C1>,
+              P2: ParallelizationContract<G::Timestamp, C2>>
+            (&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator<Item=G::Timestamp>, mut logic: L) -> StreamCore<G, C3> {
 
         self.binary_frontier(other, pact1, pact2, name, |capability, _info| {
             let mut notificator = FrontierNotificator::new();
@@ -465,16 +465,16 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
     }
 
 
-    fn binary<D2, D3, B, L, P1, P2>(&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, D3>
+    fn binary<C2, C3, B, L, P1, P2>(&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, C3>
     where
-        D2: Container,
-        D3: Container,
+        C2: Container,
+        C3: Container,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-        L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>,
-                 &mut InputHandleCore<G::Timestamp, D2, P2::Puller>,
-                 &mut OutputHandleCore<G::Timestamp, D3, Tee<G::Timestamp, D3>>)+'static,
-        P1: ParallelizationContract<G::Timestamp, D1>,
-        P2: ParallelizationContract<G::Timestamp, D2> {
+        L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P1::Puller>,
+                 &mut InputHandleCore<G::Timestamp, C2, P2::Puller>,
+                 &mut OutputHandleCore<G::Timestamp, C3, Tee<G::Timestamp, C3>>)+'static,
+        P1: ParallelizationContract<G::Timestamp, C1>,
+        P2: ParallelizationContract<G::Timestamp, C2> {
 
         let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
         let operator_info = builder.operator_info();
@@ -499,8 +499,8 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
 
     fn sink<L, P>(&self, pact: P, name: &str, mut logic: L)
     where
-        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P::Puller>)+'static,
-        P: ParallelizationContract<G::Timestamp, D1> {
+        L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, C1, P::Puller>)+'static,
+        P: ParallelizationContract<G::Timestamp, C1> {
 
         let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
         let mut input = builder.new_input(self, pact);
@@ -555,11 +555,11 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
 ///     .inspect(|x| println!("number: {:?}", x));
 /// });
 /// ```
-pub fn source<G: Scope, D, B, L>(scope: &G, name: &str, constructor: B) -> StreamCore<G, D>
+pub fn source<G: Scope, C, B, L>(scope: &G, name: &str, constructor: B) -> StreamCore<G, C>
 where
-    D: Container,
+    C: Container,
     B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
-    L: FnMut(&mut OutputHandleCore<G::Timestamp, D, Tee<G::Timestamp, D>>)+'static {
+    L: FnMut(&mut OutputHandleCore<G::Timestamp, C, Tee<G::Timestamp, C>>)+'static {
 
     let mut builder = OperatorBuilder::new(name.to_owned(), scope.clone());
     let operator_info = builder.operator_info();
@@ -599,7 +599,7 @@ where
 ///
 /// });
 /// ```
-pub fn empty<G: Scope, D: Container>(scope: &G) -> StreamCore<G, D> {
+pub fn empty<G: Scope, C: Container>(scope: &G) -> StreamCore<G, C> {
     source(scope, "Empty", |_capability, _info| |_output| {
         // drop capability, do nothing
     })
diff --git a/timely/src/dataflow/operators/input.rs b/timely/src/dataflow/operators/input.rs
index 3be59db31..90eb45810 100644
--- a/timely/src/dataflow/operators/input.rs
+++ b/timely/src/dataflow/operators/input.rs
@@ -93,7 +93,7 @@ pub trait Input : Scope {
     ///     }
     /// });
     /// ```
-    fn new_input_core<D: Container>(&mut self) -> (HandleCore<<Self as ScopeParent>::Timestamp, D>, StreamCore<Self, D>);
+    fn new_input_core<C: Container>(&mut self) -> (HandleCore<<Self as ScopeParent>::Timestamp, C>, StreamCore<Self, C>);
 
     /// Create a new stream from a supplied interactive handle.
     ///
@@ -157,7 +157,7 @@ pub trait Input : Scope {
     ///     }
     /// });
     /// ```
-    fn input_from_core<D: Container>(&mut self, handle: &mut HandleCore<<Self as ScopeParent>::Timestamp, D>) -> StreamCore<Self, D>;
+    fn input_from_core<C: Container>(&mut self, handle: &mut HandleCore<<Self as ScopeParent>::Timestamp, C>) -> StreamCore<Self, C>;
 }
 
 use crate::order::TotalOrder;
@@ -170,14 +170,14 @@ impl<G: Scope> Input for G where <G as ScopeParent>::Timestamp: TotalOrder {
         self.input_from_core(handle)
     }
 
-    fn new_input_core<D: Container>(&mut self) -> (HandleCore<<G as ScopeParent>::Timestamp, D>, StreamCore<G, D>) {
+    fn new_input_core<C: Container>(&mut self) -> (HandleCore<<G as ScopeParent>::Timestamp, C>, StreamCore<G, C>) {
         let mut handle = HandleCore::new();
         let stream = self.input_from_core(&mut handle);
         (handle, stream)
     }
 
-    fn input_from_core<D: Container>(&mut self, handle: &mut HandleCore<<G as ScopeParent>::Timestamp, D>) -> StreamCore<G, D> {
-        let (output, registrar) = Tee::<<G as ScopeParent>::Timestamp, D>::new();
+    fn input_from_core<C: Container>(&mut self, handle: &mut HandleCore<<G as ScopeParent>::Timestamp, C>) -> StreamCore<G, C> {
+        let (output, registrar) = Tee::<<G as ScopeParent>::Timestamp, C>::new();
         let counter = Counter::new(output);
         let produced = counter.produced().clone();
 
@@ -258,7 +258,7 @@ pub struct HandleCore<T: Timestamp, C: Container> {
 /// A handle specialized to vector-based containers.
 pub type Handle<T, D> = HandleCore<T, Vec<D>>;
 
-impl<T: Timestamp, D: Container> HandleCore<T, D> {
+impl<T: Timestamp, C: Container> HandleCore<T, C> {
     /// Allocates a new input handle, from which one can create timely streams.
     ///
     /// # Examples
@@ -322,7 +322,7 @@ impl<T: Timestamp, D: Container> HandleCore<T, D> {
     ///     }
     /// });
     /// ```
-    pub fn to_stream<G: Scope>(&mut self, scope: &mut G) -> StreamCore<G, D>
+    pub fn to_stream<G: Scope>(&mut self, scope: &mut G) -> StreamCore<G, C>
     where
         T: TotalOrder,
         G: ScopeParent<Timestamp=T>,
@@ -332,7 +332,7 @@ impl<T: Timestamp, D: Container> HandleCore<T, D> {
 
     fn register(
         &mut self,
-        pusher: Counter<T, D, Tee<T, D>>,
+        pusher: Counter<T, C, Tee<T, C>>,
         progress: Rc<RefCell<ChangeBatch<T>>>,
     ) {
         // flush current contents, so new registrant does not see existing data.
@@ -407,7 +407,7 @@ impl<T: Timestamp, D: Container> HandleCore<T, D> {
     ///     }
     /// });
     /// ```
-    pub fn send_batch(&mut self, buffer: &mut D) {
+    pub fn send_batch(&mut self, buffer: &mut C) {
 
         if !buffer.is_empty() {
             // flush buffered elements to ensure local fifo.
diff --git a/timely/src/dataflow/operators/probe.rs b/timely/src/dataflow/operators/probe.rs
index 982303a59..592899a22 100644
--- a/timely/src/dataflow/operators/probe.rs
+++ b/timely/src/dataflow/operators/probe.rs
@@ -16,7 +16,7 @@ use crate::dataflow::{StreamCore, Scope};
 use crate::Container;
 
 /// Monitors progress at a `Stream`.
-pub trait Probe<G: Scope, D: Container> {
+pub trait Probe<G: Scope, C: Container> {
     /// Constructs a progress probe which indicates which timestamps have elapsed at the operator.
     ///
     /// # Examples
@@ -76,10 +76,10 @@ pub trait Probe<G: Scope, D: Container> {
     ///     }
     /// }).unwrap();
     /// ```
-    fn probe_with(&self, handle: &Handle<G::Timestamp>) -> StreamCore<G, D>;
+    fn probe_with(&self, handle: &Handle<G::Timestamp>) -> StreamCore<G, C>;
 }
 
-impl<G: Scope, D: Container> Probe<G, D> for StreamCore<G, D> {
+impl<G: Scope, C: Container> Probe<G, C> for StreamCore<G, C> {
     fn probe(&self) -> Handle<G::Timestamp> {
 
         // the frontier is shared state; scope updates, handle reads.
@@ -87,7 +87,7 @@ impl<G: Scope, D: Container> Probe<G, D> for StreamCore<G, D> {
         self.probe_with(&mut handle);
         handle
     }
-    fn probe_with(&self, handle: &Handle<G::Timestamp>) -> StreamCore<G, D> {
+    fn probe_with(&self, handle: &Handle<G::Timestamp>) -> StreamCore<G, C> {
 
         let mut builder = OperatorBuilder::new("Probe".to_owned(), self.scope());
         let mut input = PullCounter::new(builder.new_input(self, Pipeline));
diff --git a/timely/src/dataflow/operators/unordered_input.rs b/timely/src/dataflow/operators/unordered_input.rs
index ea6084770..86c6b191c 100644
--- a/timely/src/dataflow/operators/unordered_input.rs
+++ b/timely/src/dataflow/operators/unordered_input.rs
@@ -142,14 +142,14 @@ pub trait UnorderedInputCore<G: Scope> {
     ///     assert_eq!(extract[i], (i, vec![i]));
     /// }
     /// ```
-    fn new_unordered_input_core<D: Container>(&mut self) -> ((UnorderedHandleCore<G::Timestamp, D>, ActivateCapability<G::Timestamp>), StreamCore<G, D>);
+    fn new_unordered_input_core<C: Container>(&mut self) -> ((UnorderedHandleCore<G::Timestamp, C>, ActivateCapability<G::Timestamp>), StreamCore<G, C>);
 }
 
 
 impl<G: Scope> UnorderedInputCore<G> for G {
-    fn new_unordered_input_core<D: Container>(&mut self) -> ((UnorderedHandleCore<G::Timestamp, D>, ActivateCapability<G::Timestamp>), StreamCore<G, D>) {
+    fn new_unordered_input_core<C: Container>(&mut self) -> ((UnorderedHandleCore<G::Timestamp, C>, ActivateCapability<G::Timestamp>), StreamCore<G, C>) {
 
-        let (output, registrar) = Tee::<G::Timestamp, D>::new();
+        let (output, registrar) = Tee::<G::Timestamp, C>::new();
         let internal = Rc::new(RefCell::new(ChangeBatch::new()));
         // let produced = Rc::new(RefCell::new(ChangeBatch::new()));
         let cap = Capability::new(G::Timestamp::minimum(), internal.clone());
@@ -215,19 +215,19 @@ impl<T:Timestamp> Operate<T> for UnorderedOperator<T> {
 
 /// A handle to an input [StreamCore], used to introduce data to a timely dataflow computation.
 #[derive(Debug)]
-pub struct UnorderedHandleCore<T: Timestamp, D: Container> {
-    buffer: PushBuffer<T, D, PushCounter<T, D, Tee<T, D>>>,
+pub struct UnorderedHandleCore<T: Timestamp, C: Container> {
+    buffer: PushBuffer<T, C, PushCounter<T, C, Tee<T, C>>>,
 }
 
-impl<T: Timestamp, D: Container> UnorderedHandleCore<T, D> {
-    fn new(pusher: PushCounter<T, D, Tee<T, D>>) -> UnorderedHandleCore<T, D> {
+impl<T: Timestamp, C: Container> UnorderedHandleCore<T, C> {
+    fn new(pusher: PushCounter<T, C, Tee<T, C>>) -> UnorderedHandleCore<T, C> {
         UnorderedHandleCore {
             buffer: PushBuffer::new(pusher),
         }
     }
 
     /// Allocates a new automatically flushing session based on the supplied capability.
-    pub fn session<'b>(&'b mut self, cap: ActivateCapability<T>) -> ActivateOnDrop<AutoflushSessionCore<'b, T, D, PushCounter<T, D, Tee<T, D>>>> {
+    pub fn session<'b>(&'b mut self, cap: ActivateCapability<T>) -> ActivateOnDrop<AutoflushSessionCore<'b, T, C, PushCounter<T, C, Tee<T, C>>>> {
         ActivateOnDrop::new(self.buffer.autoflush_session(cap.capability.clone()), cap.address.clone(), cap.activations.clone())
     }
 }
diff --git a/timely/src/dataflow/stream.rs b/timely/src/dataflow/stream.rs
index e69520729..8eafd5a4b 100644
--- a/timely/src/dataflow/stream.rs
+++ b/timely/src/dataflow/stream.rs
@@ -15,29 +15,29 @@ use crate::Container;
 
 // use dataflow::scopes::root::loggers::CHANNELS_Q;
 
-/// Abstraction of a stream of `D: Container` records timestamped with `S::Timestamp`.
+/// Abstraction of a stream of `C: Container` records timestamped with `S::Timestamp`.
 ///
 /// Internally `Stream` maintains a list of data recipients who should be presented with data
 /// produced by the source of the stream.
 #[derive(Clone)]
-pub struct StreamCore<S: Scope, D> {
+pub struct StreamCore<S: Scope, C> {
     /// The progress identifier of the stream's data source.
     name: Source,
     /// The `Scope` containing the stream.
     scope: S,
-    /// Maintains a list of Push<Bundle<T, Vec<D>>> interested in the stream's output.
-    ports: TeeHelper<S::Timestamp, D>,
+    /// Maintains a list of Push<Bundle<T, C>> interested in the stream's output.
+    ports: TeeHelper<S::Timestamp, C>,
 }
 
 /// A stream batching data in vectors.
 pub type Stream<S, D> = StreamCore<S, Vec<D>>;
 
-impl<S: Scope, D: Container> StreamCore<S, D> {
+impl<S: Scope, C: Container> StreamCore<S, C> {
     /// Connects the stream to a destination.
     ///
     /// The destination is described both by a `Target`, for progress tracking information, and a `P: Push` where the
     /// records should actually be sent. The identifier is unique to the edge and is used only for logging purposes.
-    pub fn connect_to<P: Push<Bundle<S::Timestamp, D>>+'static>(&self, target: Target, pusher: P, identifier: usize) {
+    pub fn connect_to<P: Push<Bundle<S::Timestamp, C>>+'static>(&self, target: Target, pusher: P, identifier: usize) {
 
         let mut logging = self.scope().logging();
         logging.as_mut().map(|l| l.log(crate::logging::ChannelsEvent {
@@ -51,7 +51,7 @@ impl<S: Scope, D: Container> StreamCore<S, D> {
         self.ports.add_pusher(pusher);
     }
     /// Allocates a `Stream` from a supplied `Source` name and rendezvous point.
-    pub fn new(source: Source, output: TeeHelper<S::Timestamp, D>, scope: S) -> Self {
+    pub fn new(source: Source, output: TeeHelper<S::Timestamp, C>, scope: S) -> Self {
         Self { name: source, ports: output, scope }
     }
     /// The name of the stream's source operator.
@@ -60,7 +60,7 @@ impl<S: Scope, D: Container> StreamCore<S, D> {
     pub fn scope(&self) -> S { self.scope.clone() }
 }
 
-impl<S, D> Debug for StreamCore<S, D>
+impl<S, C> Debug for StreamCore<S, C>
 where
     S: Scope,
 {