From 6c5567799fe01fef0b6067943730c8ef59d5e26b Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 23 Mar 2024 03:13:49 -0400
Subject: [PATCH 1/3] Remove non-Core capture logic

---
 timely/examples/capture_recv.rs               |   2 +-
 timely/examples/logging-recv.rs               |   2 +-
 .../src/dataflow/operators/capture/capture.rs |  18 +--
 .../src/dataflow/operators/capture/event.rs   | 109 ++++++-----------
 .../src/dataflow/operators/capture/extract.rs | 112 +++++-------------
 timely/src/dataflow/operators/capture/mod.rs  |  16 +--
 .../src/dataflow/operators/capture/replay.rs  |  18 +--
 timely/src/dataflow/operators/core/rc.rs      |   1 +
 timely/src/dataflow/operators/core/reclock.rs |   3 +-
 .../src/dataflow/operators/core/to_stream.rs  |   2 +-
 timely/src/execute.rs                         |   8 +-
 timely/src/logging.rs                         |   6 +-
 12 files changed, 110 insertions(+), 187 deletions(-)

diff --git a/timely/examples/capture_recv.rs b/timely/examples/capture_recv.rs
index d6f7291c11..5cb48b055f 100644
--- a/timely/examples/capture_recv.rs
+++ b/timely/examples/capture_recv.rs
@@ -17,7 +17,7 @@ fn main() {
             .collect::<Vec<_>>()
             .into_iter()
             .map(|l| l.incoming().next().unwrap().unwrap())
-            .map(|r| EventReader::<_,u64,_>::new(r))
+            .map(|r| EventReader::<_,Vec<u64>,_>::new(r))
             .collect::<Vec<_>>();
 
         worker.dataflow::<u64,_,_>(|scope| {
diff --git a/timely/examples/logging-recv.rs b/timely/examples/logging-recv.rs
index 1c2a5ccda8..4608cad3da 100644
--- a/timely/examples/logging-recv.rs
+++ b/timely/examples/logging-recv.rs
@@ -20,7 +20,7 @@ fn main() {
             .collect::<Vec<_>>()
             .into_iter()
             .map(|l| l.incoming().next().unwrap().unwrap())
-            .map(|r| EventReader::<Duration,(Duration,TimelySetup,TimelyEvent),_>::new(r))
+            .map(|r| EventReader::<Duration,Vec<(Duration,TimelySetup,TimelyEvent)>,_>::new(r))
             .collect::<Vec<_>>();
 
         worker.dataflow(|scope| {
diff --git a/timely/src/dataflow/operators/capture/capture.rs b/timely/src/dataflow/operators/capture/capture.rs
index 9aa3827de2..b14d3a6b56 100644
--- a/timely/src/dataflow/operators/capture/capture.rs
+++ b/timely/src/dataflow/operators/capture/capture.rs
@@ -14,7 +14,7 @@ use crate::Container;
 use crate::progress::ChangeBatch;
 use crate::progress::Timestamp;
 
-use super::{EventCore, EventPusherCore};
+use super::{Event, EventPusher};
 
 /// Capture a stream of timestamped data for later replay.
 pub trait Capture<T: Timestamp, C: Container> {
@@ -30,7 +30,7 @@ pub trait Capture<T: Timestamp, C: Container> {
     /// use std::sync::{Arc, Mutex};
     /// use timely::dataflow::Scope;
     /// use timely::dataflow::operators::{Capture, ToStream, Inspect};
-    /// use timely::dataflow::operators::capture::{EventLinkCore, Replay, Extract};
+    /// use timely::dataflow::operators::capture::{EventLink, Replay, Extract};
     ///
     /// // get send and recv endpoints, wrap send to share
     /// let (send, recv) = ::std::sync::mpsc::channel();
@@ -42,7 +42,7 @@ pub trait Capture<T: Timestamp, C: Container> {
     ///     let send = send.lock().unwrap().clone();
     ///
     ///     // these are to capture/replay the stream.
-    ///     let handle1 = Rc::new(EventLinkCore::new());
+    ///     let handle1 = Rc::new(EventLink::new());
     ///     let handle2 = Some(handle1.clone());
     ///
     ///     worker.dataflow::<u64,_,_>(|scope1|
@@ -95,7 +95,7 @@ pub trait Capture<T: Timestamp, C: Container> {
     ///     );
     ///
     ///     worker.dataflow::<u64,_,_>(|scope2| {
-    ///         Some(EventReader::<_,u64,_>::new(recv))
+    ///         Some(EventReader::<_,Vec<u64>,_>::new(recv))
     ///             .replay_into(scope2)
     ///             .capture_into(send0)
     ///     });
@@ -103,10 +103,10 @@ pub trait Capture<T: Timestamp, C: Container> {
     ///
     /// assert_eq!(recv0.extract()[0].1, (0..10).collect::<Vec<_>>());
     /// ```
-    fn capture_into<P: EventPusherCore<T, C>+'static>(&self, pusher: P);
+    fn capture_into<P: EventPusher<T, C>+'static>(&self, pusher: P);
 
     /// Captures a stream using Rust's MPSC channels.
-    fn capture(&self) -> ::std::sync::mpsc::Receiver<EventCore<T, C>> {
+    fn capture(&self) -> ::std::sync::mpsc::Receiver<Event<T, C>> {
         let (send, recv) = ::std::sync::mpsc::channel();
         self.capture_into(send);
         recv
@@ -114,7 +114,7 @@ pub trait Capture<T: Timestamp, C: Container> {
 }
 
 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) {
+    fn capture_into<P: EventPusher<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));
@@ -131,7 +131,7 @@ impl<S: Scope, C: Container> Capture<S::Timestamp, C> for StreamCore<S, C> {
                 if !progress.frontiers[0].is_empty() {
                     // transmit any frontier progress.
                     let to_send = ::std::mem::replace(&mut progress.frontiers[0], ChangeBatch::new());
-                    event_pusher.push(EventCore::Progress(to_send.into_inner()));
+                    event_pusher.push(Event::Progress(to_send.into_inner()));
                 }
 
                 use crate::communication::message::RefOrMut;
@@ -143,7 +143,7 @@ impl<S: Scope, C: Container> Capture<S::Timestamp, C> for StreamCore<S, C> {
                         RefOrMut::Mut(reference) => (&reference.time, RefOrMut::Mut(&mut reference.data)),
                     };
                     let vector = data.replace(Default::default());
-                    event_pusher.push(EventCore::Messages(time.clone(), vector));
+                    event_pusher.push(Event::Messages(time.clone(), vector));
                 }
                 input.consumed().borrow_mut().drain_into(&mut progress.consumeds[0]);
                 false
diff --git a/timely/src/dataflow/operators/capture/event.rs b/timely/src/dataflow/operators/capture/event.rs
index a5b5f07b8c..3f7961e288 100644
--- a/timely/src/dataflow/operators/capture/event.rs
+++ b/timely/src/dataflow/operators/capture/event.rs
@@ -6,55 +6,33 @@
 
 /// Data and progress events of the captured stream.
 #[derive(Debug, Clone, Abomonation, Hash, Ord, PartialOrd, Eq, PartialEq, Deserialize, Serialize)]
-pub enum EventCore<T, C> {
+pub enum Event<T, C> {
     /// Progress received via `push_external_progress`.
     Progress(Vec<(T, i64)>),
     /// Messages received via the data stream.
     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, C>`.
+/// Iterates over contained `Event<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, C> {
-    /// Iterates over references to `EventCore<T, C>` elements.
-    fn next(&mut self) -> Option<&EventCore<T, C>>;
-}
-
-/// A [EventIteratorCore] specialized to vector-based containers.
-// TODO: use trait aliases once stable.
-pub trait EventIterator<T, D>: EventIteratorCore<T, Vec<D>> {
-    /// Iterates over references to `Event<T, D>` elements.
-    fn next(&mut self) -> Option<&Event<T, D>>;
+pub trait EventIterator<T, C> {
+    /// Iterates over references to `Event<T, C>` elements.
+    fn next(&mut self) -> Option<&Event<T, C>>;
 }
-impl<T, D, E: EventIteratorCore<T, Vec<D>>> EventIterator<T, D> for E {
-    fn next(&mut self) -> Option<&Event<T, D>> {
-        <Self as EventIteratorCore<_, _>>::next(self)
-    }
-}
-
 
-/// Receives `EventCore<T, C>` events.
-pub trait EventPusherCore<T, C> {
+/// Receives `Event<T, C>` events.
+pub trait EventPusher<T, C> {
     /// Provides a new `Event<T, D>` to the pusher.
-    fn push(&mut self, event: EventCore<T, C>);
+    fn push(&mut self, event: Event<T, C>);
 }
 
-/// A [EventPusherCore] specialized to vector-based containers.
-// TODO: use trait aliases once stable.
-pub trait EventPusher<T, D>: EventPusherCore<T, Vec<D>> {}
-impl<T, D, E: EventPusherCore<T, Vec<D>>> EventPusher<T, D> for E {}
-
-
 // implementation for the linked list behind a `Handle`.
-impl<T, C> EventPusherCore<T, C> for ::std::sync::mpsc::Sender<EventCore<T, C>> {
-    fn push(&mut self, event: EventCore<T, C>) {
+impl<T, C> EventPusher<T, C> for ::std::sync::mpsc::Sender<Event<T, C>> {
+    fn push(&mut self, event: Event<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);
@@ -67,40 +45,37 @@ pub mod link {
     use std::rc::Rc;
     use std::cell::RefCell;
 
-    use super::{EventCore, EventPusherCore, EventIteratorCore};
+    use super::{Event, EventPusher, EventIterator};
 
-    /// A linked list of EventCore<T, C>.
-    pub struct EventLinkCore<T, C> {
+    /// A linked list of Event<T, C>.
+    pub struct EventLink<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, C>>,
+        pub event: Option<Event<T, C>>,
         /// The next event, if it exists.
-        pub next: RefCell<Option<Rc<EventLinkCore<T, C>>>>,
+        pub next: RefCell<Option<Rc<EventLink<T, C>>>>,
     }
 
-    /// A [EventLinkCore] specialized to vector-based containers.
-    pub type EventLink<T, D> = EventLinkCore<T, Vec<D>>;
-
-    impl<T, C> EventLinkCore<T, C> {
+    impl<T, C> EventLink<T, C> {
         /// Allocates a new `EventLink`.
-        pub fn new() -> EventLinkCore<T, C> {
-            EventLinkCore { event: None, next: RefCell::new(None) }
+        pub fn new() -> EventLink<T, C> {
+            EventLink { event: None, next: RefCell::new(None) }
         }
     }
 
     // implementation for the linked list behind a `Handle`.
-    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) }));
+    impl<T, C> EventPusher<T, C> for Rc<EventLink<T, C>> {
+        fn push(&mut self, event: Event<T, C>) {
+            *self.next.borrow_mut() = Some(Rc::new(EventLink { event: Some(event), next: RefCell::new(None) }));
             let next = self.next.borrow().as_ref().unwrap().clone();
             *self = next;
         }
     }
 
-    impl<T, C> EventIteratorCore<T, C> for Rc<EventLinkCore<T, C>> {
-        fn next(&mut self) -> Option<&EventCore<T, C>> {
+    impl<T, C> EventIterator<T, C> for Rc<EventLink<T, C>> {
+        fn next(&mut self) -> Option<&Event<T, C>> {
             let is_some = self.next.borrow().is_some();
             if is_some {
                 let next = self.next.borrow().as_ref().unwrap().clone();
@@ -114,7 +89,7 @@ pub mod link {
     }
 
     // Drop implementation to prevent stack overflow through naive drop impl.
-    impl<T, C> Drop for EventLinkCore<T, C> {
+    impl<T, C> Drop for EventLink<T, C> {
         fn drop(&mut self) {
             while let Some(link) = self.next.replace(None) {
                 if let Ok(head) = Rc::try_unwrap(link) {
@@ -124,7 +99,7 @@ pub mod link {
         }
     }
 
-    impl<T, C> Default for EventLinkCore<T, C> {
+    impl<T, C> Default for EventLink<T, C> {
         fn default() -> Self {
             Self::new()
         }
@@ -132,10 +107,10 @@ pub mod link {
 
     #[test]
     fn avoid_stack_overflow_in_drop() {
-        let mut event1 = Rc::new(EventLinkCore::<(),()>::new());
+        let mut event1 = Rc::new(EventLink::<(),()>::new());
         let _event2 = event1.clone();
         for _ in 0 .. 1_000_000 {
-            event1.push(EventCore::Progress(vec![]));
+            event1.push(Event::Progress(vec![]));
         }
     }
 }
@@ -145,18 +120,15 @@ pub mod binary {
 
     use std::io::Write;
     use abomonation::Abomonation;
-    use super::{EventCore, EventPusherCore, EventIteratorCore};
+    use super::{Event, EventPusher, EventIterator};
 
-    /// A wrapper for `W: Write` implementing `EventPusherCore<T, C>`.
-    pub struct EventWriterCore<T, C, W: ::std::io::Write> {
+    /// A wrapper for `W: Write` implementing `EventPusher<T, C>`.
+    pub struct EventWriter<T, C, W: ::std::io::Write> {
         stream: W,
         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, C, W: ::std::io::Write> EventWriterCore<T, C, W> {
+    impl<T, C, W: ::std::io::Write> EventWriter<T, C, W> {
         /// Allocates a new `EventWriter` wrapping a supplied writer.
         pub fn new(w: W) -> Self {
             Self {
@@ -166,15 +138,15 @@ pub mod binary {
         }
     }
 
-    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>) {
+    impl<T: Abomonation, C: Abomonation, W: ::std::io::Write> EventPusher<T, C> for EventWriter<T, C, W> {
+        fn push(&mut self, event: Event<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, C, R: ::std::io::Read> {
+    pub struct EventReader<T, C, R: ::std::io::Read> {
         reader: R,
         bytes: Vec<u8>,
         buff1: Vec<u8>,
@@ -184,10 +156,7 @@ pub mod binary {
         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, C, R: ::std::io::Read> EventReaderCore<T, C, R> {
+    impl<T, C, R: ::std::io::Read> EventReader<T, C, R> {
         /// Allocates a new `EventReader` wrapping a supplied reader.
         pub fn new(r: R) -> Self {
             Self {
@@ -202,12 +171,12 @@ pub mod binary {
         }
     }
 
-    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>> {
+    impl<T: Abomonation, C: Abomonation, R: ::std::io::Read> EventIterator<T, C> for EventReader<T, C, R> {
+        fn next(&mut self) -> Option<&Event<T, C>> {
 
             // if we can decode something, we should just return it! :D
-            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();
+            if unsafe { ::abomonation::decode::<Event<T, C>>(&mut self.buff1[self.consumed..]) }.is_some() {
+                let (item, rest) = unsafe { ::abomonation::decode::<Event<T, C>>(&mut self.buff1[self.consumed..]) }.unwrap();
                 self.consumed = self.valid - rest.len();
                 return Some(item);
             }
diff --git a/timely/src/dataflow/operators/capture/extract.rs b/timely/src/dataflow/operators/capture/extract.rs
index dcf57ae3bb..a496202855 100644
--- a/timely/src/dataflow/operators/capture/extract.rs
+++ b/timely/src/dataflow/operators/capture/extract.rs
@@ -1,14 +1,13 @@
 //! Traits and types for extracting captured timely dataflow streams.
 
-use super::EventCore;
-use crate::Container;
-use crate::Data;
+use super::Event;
+use crate::{container::{PushContainer, PushInto}};
 
 /// Supports extracting a sequence of timestamp and data.
-pub trait Extract<T: Ord, D: Ord> {
+pub trait Extract<T, C> {
     /// Converts `self` into a sequence of timestamped data.
     ///
-    /// Currently this is only implemented for `Receiver<Event<T, Vec<D>>>`, and is used only
+    /// Currently this is only implemented for `Receiver<Event<T, C>>`, and is used only
     /// to easily pull data out of a timely dataflow computation once it has completed.
     ///
     /// # Examples
@@ -18,7 +17,7 @@ pub trait Extract<T: Ord, D: Ord> {
     /// use std::sync::{Arc, Mutex};
     /// use timely::dataflow::Scope;
     /// use timely::dataflow::operators::{Capture, ToStream, Inspect};
-    /// use timely::dataflow::operators::capture::{EventLinkCore, Replay, Extract};
+    /// use timely::dataflow::operators::capture::{EventLink, Replay, Extract};
     ///
     /// // get send and recv endpoints, wrap send to share
     /// let (send, recv) = ::std::sync::mpsc::channel();
@@ -30,7 +29,7 @@ pub trait Extract<T: Ord, D: Ord> {
     ///     let send = send.lock().unwrap().clone();
     ///
     ///     // these are to capture/replay the stream.
-    ///     let handle1 = Rc::new(EventLinkCore::new());
+    ///     let handle1 = Rc::new(EventLink::new());
     ///     let handle2 = Some(handle1.clone());
     ///
     ///     worker.dataflow::<u64,_,_>(|scope1|
@@ -44,88 +43,39 @@ pub trait Extract<T: Ord, D: Ord> {
     ///     });
     /// }).unwrap();
     ///
-    /// assert_eq!(recv.extract()[0].1, (0..10).collect::<Vec<_>>());
+    /// assert_eq!(recv.extract().into_iter().flat_map(|x| x.1).collect::<Vec<_>>(), (0..10).collect::<Vec<_>>());
     /// ```
-    fn extract(self) -> Vec<(T, Vec<D>)>;
+    fn extract(self) -> Vec<(T, C)>;
 }
 
-impl<T: Ord, D: Ord+Data> Extract<T,D> for ::std::sync::mpsc::Receiver<EventCore<T, Vec<D>>> {
-    fn extract(self) -> Vec<(T, Vec<D>)> {
-        let mut result = self.extract_core();
-
-        let mut current = 0;
-        for i in 1 .. result.len() {
-            if result[current].0 == result[i].0 {
-                let dataz = ::std::mem::replace(&mut result[i].1, Vec::new());
-                result[current].1.extend(dataz);
-            }
-            else {
-                current = i;
+impl<T: Ord, C: PushContainer> Extract<T, C> for ::std::sync::mpsc::Receiver<Event<T, C>>
+where
+    for<'a> C::Item<'a>: PushInto<C> + Ord,
+{
+    fn extract(self) -> Vec<(T, C)> {
+        let mut staged = std::collections::BTreeMap::new();
+        for event in self {
+            if let Event::Messages(time, data) = event {
+                staged.entry(time)
+                      .or_insert_with(Vec::new)
+                      .push(data);
             }
         }
-
-        for &mut (_, ref mut data) in &mut result {
-            data.sort();
-        }
-        result.retain(|x| !x.1.is_empty());
-        result
-    }
-}
-
-/// Supports extracting a sequence of timestamp and data.
-pub trait ExtractCore<T, C> {
-    /// Converts `self` into a sequence of timestamped data.
-    ///
-    /// Currently this is only implemented for `Receiver<Event<T, C>>`, and is used only
-    /// to easily pull data out of a timely dataflow computation once it has completed.
-    ///
-    /// # Examples
-    ///
-    /// ```rust
-    /// use std::rc::Rc;
-    /// use std::sync::{Arc, Mutex};
-    /// use timely::dataflow::Scope;
-    /// use timely::dataflow::operators::{Capture, ToStream, Inspect};
-    /// use timely::dataflow::operators::capture::{EventLinkCore, Replay, ExtractCore};
-    ///
-    /// // get send and recv endpoints, wrap send to share
-    /// let (send, recv) = ::std::sync::mpsc::channel();
-    /// let send = Arc::new(Mutex::new(send));
-    ///
-    /// timely::execute(timely::Config::thread(), move |worker| {
-    ///
-    ///     // this is only to validate the output.
-    ///     let send = send.lock().unwrap().clone();
-    ///
-    ///     // these are to capture/replay the stream.
-    ///     let handle1 = Rc::new(EventLinkCore::new());
-    ///     let handle2 = Some(handle1.clone());
-    ///
-    ///     worker.dataflow::<u64,_,_>(|scope1|
-    ///         (0..10).to_stream(scope1)
-    ///                .capture_into(handle1)
-    ///     );
-    ///
-    ///     worker.dataflow(|scope2| {
-    ///         handle2.replay_into(scope2)
-    ///                .capture_into(send)
-    ///     });
-    /// }).unwrap();
-    ///
-    /// assert_eq!(recv.extract_core().into_iter().flat_map(|x| x.1).collect::<Vec<_>>(), (0..10).collect::<Vec<_>>());
-    /// ```
-    fn extract_core(self) -> Vec<(T, C)>;
-}
-
-impl<T, C: Container> ExtractCore<T, C> for ::std::sync::mpsc::Receiver<EventCore<T, C>> {
-    fn extract_core(self) -> Vec<(T, C)> {
         let mut result = Vec::new();
-        for event in self {
-            if let EventCore::Messages(time, data) = event {
-                result.push((time, data));
+        for (time, mut dataz) in staged.into_iter() {
+            let mut to_sort = Vec::new();
+            for data in dataz.iter_mut() {
+                to_sort.extend(data.drain());
+            }
+            to_sort.sort();
+            let mut sorted = C::default();
+            for datum in to_sort.into_iter() {
+                sorted.push(datum);
+            }
+            if !sorted.is_empty() {
+                result.push((time, sorted));
             }
         }
-        result.retain(|x| !x.1.is_empty());
         result
     }
 }
diff --git a/timely/src/dataflow/operators/capture/mod.rs b/timely/src/dataflow/operators/capture/mod.rs
index 22d332ea02..84fb68aab9 100644
--- a/timely/src/dataflow/operators/capture/mod.rs
+++ b/timely/src/dataflow/operators/capture/mod.rs
@@ -22,10 +22,10 @@
 //! use std::rc::Rc;
 //! use timely::dataflow::Scope;
 //! use timely::dataflow::operators::{Capture, ToStream, Inspect};
-//! use timely::dataflow::operators::capture::{EventLinkCore, Replay};
+//! use timely::dataflow::operators::capture::{EventLink, Replay};
 //!
 //! timely::execute(timely::Config::thread(), |worker| {
-//!     let handle1 = Rc::new(EventLinkCore::new());
+//!     let handle1 = Rc::new(EventLink::new());
 //!     let handle2 = Some(handle1.clone());
 //!
 //!     worker.dataflow::<u64,_,_>(|scope1|
@@ -66,7 +66,7 @@
 //!     );
 //!
 //!     worker.dataflow::<u64,_,_>(|scope2| {
-//!         Some(EventReader::<_,u64,_>::new(recv))
+//!         Some(EventReader::<_,Vec<u64>,_>::new(recv))
 //!             .replay_into(scope2)
 //!             .inspect(|x| println!("replayed: {:?}", x));
 //!     })
@@ -75,11 +75,11 @@
 
 pub use self::capture::Capture;
 pub use self::replay::Replay;
-pub use self::extract::{Extract, ExtractCore};
-pub use self::event::{Event, EventCore, EventPusher, EventPusherCore};
-pub use self::event::link::{EventLink, EventLinkCore};
-pub use self::event::binary::{EventReader, EventReaderCore};
-pub use self::event::binary::{EventWriter, EventWriterCore};
+pub use self::extract::Extract;
+pub use self::event::{Event, EventPusher};
+pub use self::event::link::EventLink;
+pub use self::event::binary::EventReader;
+pub use self::event::binary::EventWriter;
 
 pub mod capture;
 pub mod replay;
diff --git a/timely/src/dataflow/operators/capture/replay.rs b/timely/src/dataflow/operators/capture/replay.rs
index 5fdb237040..bdf6558a07 100644
--- a/timely/src/dataflow/operators/capture/replay.rs
+++ b/timely/src/dataflow/operators/capture/replay.rs
@@ -44,17 +44,17 @@ use crate::dataflow::channels::pushers::buffer::Buffer as PushBuffer;
 use crate::dataflow::operators::generic::builder_raw::OperatorBuilder;
 use crate::progress::Timestamp;
 
-use super::EventCore;
-use super::event::EventIteratorCore;
+use super::Event;
+use super::event::EventIterator;
 use crate::Container;
 
 /// Replay a capture stream into a scope with the same timestamp.
 pub trait Replay<T: Timestamp, C> : Sized {
-    /// Replays `self` into the provided scope, as a `Stream<S, D>`.
+    /// Replays `self` into the provided scope, as a `StreamCore<S,CD>`.
     fn replay_into<S: Scope<Timestamp=T>>(self, scope: &mut S) -> StreamCore<S, C> {
         self.replay_core(scope, Some(std::time::Duration::new(0, 0)))
     }
-    /// Replays `self` into the provided scope, as a `Stream<S, D>'.
+    /// Replays `self` into the provided scope, as a `StreamCore<S, C>'.
     ///
     /// The `period` argument allows the specification of a re-activation period, where the operator
     /// will re-activate itself every so often. The `None` argument instructs the operator not to
@@ -63,8 +63,10 @@ pub trait Replay<T: Timestamp, C> : Sized {
 }
 
 impl<T: Timestamp, C: Container, I> Replay<T, C> for I
-where I : IntoIterator,
-      <I as IntoIterator>::Item: EventIteratorCore<T, C>+'static {
+where
+    I : IntoIterator,
+    <I as IntoIterator>::Item: EventIterator<T, C>+'static,
+{
     fn replay_core<S: Scope<Timestamp=T>>(self, scope: &mut S, period: Option<std::time::Duration>) -> StreamCore<S, C>{
 
         let mut builder = OperatorBuilder::new("Replay".to_owned(), scope.clone());
@@ -93,10 +95,10 @@ where I : IntoIterator,
                 for event_stream in event_streams.iter_mut() {
                     while let Some(event) = event_stream.next() {
                         match event {
-                            EventCore::Progress(vec) => {
+                            Event::Progress(vec) => {
                                 progress.internals[0].extend(vec.iter().cloned());
                             },
-                            EventCore::Messages(ref time, data) => {
+                            Event::Messages(ref time, data) => {
                                 allocation.clone_from(data);
                                 output.session(time).give_container(&mut allocation);
                             }
diff --git a/timely/src/dataflow/operators/core/rc.rs b/timely/src/dataflow/operators/core/rc.rs
index fcdc646cf2..c00108a7c3 100644
--- a/timely/src/dataflow/operators/core/rc.rs
+++ b/timely/src/dataflow/operators/core/rc.rs
@@ -72,6 +72,7 @@ mod test {
                         }
                     }),
                 ])
+                .container::<Vec<_>>()
                 .capture()
         });
         let output = &mut output.extract()[0].1;
diff --git a/timely/src/dataflow/operators/core/reclock.rs b/timely/src/dataflow/operators/core/reclock.rs
index 9510e87223..ec9617fb7f 100644
--- a/timely/src/dataflow/operators/core/reclock.rs
+++ b/timely/src/dataflow/operators/core/reclock.rs
@@ -19,7 +19,7 @@ pub trait Reclock<S: Scope> {
     /// # Examples
     ///
     /// ```
-    /// use timely::dataflow::operators::{ToStream, Delay, Map, Reclock, Capture};
+    /// use timely::dataflow::operators::{Inspect, ToStream, Delay, Map, Reclock, Capture};
     /// use timely::dataflow::operators::capture::Extract;
     ///
     /// let captured = timely::example(|scope| {
@@ -36,6 +36,7 @@ pub trait Reclock<S: Scope> {
     ///
     ///     // reclock the data.
     ///     data.reclock(&clock)
+    ///         .inspect_batch(|t,x| println!("SEEN: {:?}, {:?}", t, x))
     ///         .capture()
     /// });
     ///
diff --git a/timely/src/dataflow/operators/core/to_stream.rs b/timely/src/dataflow/operators/core/to_stream.rs
index 3775e0267b..848adf3c1d 100644
--- a/timely/src/dataflow/operators/core/to_stream.rs
+++ b/timely/src/dataflow/operators/core/to_stream.rs
@@ -12,7 +12,7 @@ pub trait ToStream<C: Container> {
     /// # Examples
     ///
     /// ```
-    /// use timely::dataflow::operators::core::ToStream;
+    /// use timely::dataflow::operators::ToStream;
     /// use timely::dataflow::operators::Capture;
     /// use timely::dataflow::operators::capture::Extract;
     ///
diff --git a/timely/src/execute.rs b/timely/src/execute.rs
index 3564d35928..9a8340dbfe 100644
--- a/timely/src/execute.rs
+++ b/timely/src/execute.rs
@@ -235,12 +235,12 @@ where
 
                 use ::std::net::TcpStream;
                 use crate::logging::BatchLogger;
-                use crate::dataflow::operators::capture::EventWriterCore;
+                use crate::dataflow::operators::capture::EventWriter;
 
                 eprintln!("enabled COMM logging to {}", addr);
 
                 if let Ok(stream) = TcpStream::connect(&addr) {
-                    let writer = EventWriterCore::new(stream);
+                    let writer = EventWriter::new(stream);
                     let mut logger = BatchLogger::new(writer);
                     result = Some(crate::logging_core::Logger::new(
                         ::std::time::Instant::now(),
@@ -269,10 +269,10 @@ where
 
             use ::std::net::TcpStream;
             use crate::logging::{BatchLogger, TimelyEvent};
-            use crate::dataflow::operators::capture::EventWriterCore;
+            use crate::dataflow::operators::capture::EventWriter;
 
             if let Ok(stream) = TcpStream::connect(&addr) {
-                let writer = EventWriterCore::new(stream);
+                let writer = EventWriter::new(stream);
                 let mut logger = BatchLogger::new(writer);
                 worker.log_register()
                     .insert::<TimelyEvent,_>("timely", move |time, data|
diff --git a/timely/src/logging.rs b/timely/src/logging.rs
index ca9868fbe5..39944ade05 100644
--- a/timely/src/logging.rs
+++ b/timely/src/logging.rs
@@ -13,14 +13,14 @@ use std::time::Duration;
 use crate::dataflow::operators::capture::{Event, EventPusher};
 
 /// Logs events as a timely stream, with progress statements.
-pub struct BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E, T)> {
+pub struct BatchLogger<T, E, P> where P: EventPusher<Duration, Vec<(Duration, E, T)>> {
     // None when the logging stream is closed
     time: Duration,
     event_pusher: P,
     _phantom: ::std::marker::PhantomData<(E, T)>,
 }
 
-impl<T, E, P> BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E, T)> {
+impl<T, E, P> BatchLogger<T, E, P> where P: EventPusher<Duration, Vec<(Duration, E, T)>> {
     /// Creates a new batch logger.
     pub fn new(event_pusher: P) -> Self {
         BatchLogger {
@@ -42,7 +42,7 @@ impl<T, E, P> BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E,
         self.time = time;
     }
 }
-impl<T, E, P> Drop for BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E, T)> {
+impl<T, E, P> Drop for BatchLogger<T, E, P> where P: EventPusher<Duration, Vec<(Duration, E, T)>> {
     fn drop(&mut self) {
         self.event_pusher.push(Event::Progress(vec![(self.time, -1)]));
     }

From fea3459c92f6866c079e6f95fab790a76015c784 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 23 Mar 2024 03:15:52 -0400
Subject: [PATCH 2/3] Update Capture

---
 timely/src/dataflow/operators/{ => core}/capture/capture.rs | 0
 timely/src/dataflow/operators/{ => core}/capture/event.rs   | 0
 timely/src/dataflow/operators/{ => core}/capture/extract.rs | 0
 timely/src/dataflow/operators/{ => core}/capture/mod.rs     | 0
 timely/src/dataflow/operators/{ => core}/capture/replay.rs  | 0
 timely/src/dataflow/operators/core/mod.rs                   | 2 ++
 timely/src/dataflow/operators/mod.rs                        | 3 +--
 7 files changed, 3 insertions(+), 2 deletions(-)
 rename timely/src/dataflow/operators/{ => core}/capture/capture.rs (100%)
 rename timely/src/dataflow/operators/{ => core}/capture/event.rs (100%)
 rename timely/src/dataflow/operators/{ => core}/capture/extract.rs (100%)
 rename timely/src/dataflow/operators/{ => core}/capture/mod.rs (100%)
 rename timely/src/dataflow/operators/{ => core}/capture/replay.rs (100%)

diff --git a/timely/src/dataflow/operators/capture/capture.rs b/timely/src/dataflow/operators/core/capture/capture.rs
similarity index 100%
rename from timely/src/dataflow/operators/capture/capture.rs
rename to timely/src/dataflow/operators/core/capture/capture.rs
diff --git a/timely/src/dataflow/operators/capture/event.rs b/timely/src/dataflow/operators/core/capture/event.rs
similarity index 100%
rename from timely/src/dataflow/operators/capture/event.rs
rename to timely/src/dataflow/operators/core/capture/event.rs
diff --git a/timely/src/dataflow/operators/capture/extract.rs b/timely/src/dataflow/operators/core/capture/extract.rs
similarity index 100%
rename from timely/src/dataflow/operators/capture/extract.rs
rename to timely/src/dataflow/operators/core/capture/extract.rs
diff --git a/timely/src/dataflow/operators/capture/mod.rs b/timely/src/dataflow/operators/core/capture/mod.rs
similarity index 100%
rename from timely/src/dataflow/operators/capture/mod.rs
rename to timely/src/dataflow/operators/core/capture/mod.rs
diff --git a/timely/src/dataflow/operators/capture/replay.rs b/timely/src/dataflow/operators/core/capture/replay.rs
similarity index 100%
rename from timely/src/dataflow/operators/capture/replay.rs
rename to timely/src/dataflow/operators/core/capture/replay.rs
diff --git a/timely/src/dataflow/operators/core/mod.rs b/timely/src/dataflow/operators/core/mod.rs
index d2ae145d97..19cba0afdc 100644
--- a/timely/src/dataflow/operators/core/mod.rs
+++ b/timely/src/dataflow/operators/core/mod.rs
@@ -1,6 +1,7 @@
 //! Extension traits for `StreamCore` implementing various operators that
 //! are independent of specific container types.
 
+pub mod capture;
 pub mod concat;
 pub mod enterleave;
 pub mod exchange;
@@ -16,6 +17,7 @@ pub mod reclock;
 pub mod to_stream;
 pub mod unordered_input;
 
+pub use capture::Capture;
 pub use concat::{Concat, Concatenate};
 pub use enterleave::{Enter, Leave};
 pub use exchange::Exchange;
diff --git a/timely/src/dataflow/operators/mod.rs b/timely/src/dataflow/operators/mod.rs
index 53b504d3ba..2644e7ce06 100644
--- a/timely/src/dataflow/operators/mod.rs
+++ b/timely/src/dataflow/operators/mod.rs
@@ -17,7 +17,6 @@ pub use self::filter::Filter;
 pub use self::delay::Delay;
 pub use self::exchange::Exchange;
 pub use self::broadcast::Broadcast;
-pub use self::capture::Capture;
 pub use self::branch::{Branch, BranchWhen};
 pub use self::result::ResultStream;
 pub use self::to_stream::ToStream;
@@ -46,7 +45,7 @@ pub use self::core::exchange;
 pub mod broadcast;
 pub use self::core::probe::{self, Probe};
 pub mod to_stream;
-pub mod capture;
+pub use self::core::capture::{self, Capture};
 pub mod branch;
 pub use self::core::ok_err::{self, OkErr};
 pub use self::core::rc;

From 0f42c3e243cff231e591dbf34e141a515ef6e581 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 23 Mar 2024 03:23:43 -0400
Subject: [PATCH 3/3] comment/doctest clean-up

---
 timely/src/dataflow/operators/core/capture/replay.rs | 2 +-
 timely/src/dataflow/operators/core/reclock.rs        | 3 +--
 timely/src/dataflow/operators/core/to_stream.rs      | 9 ++++-----
 3 files changed, 6 insertions(+), 8 deletions(-)

diff --git a/timely/src/dataflow/operators/core/capture/replay.rs b/timely/src/dataflow/operators/core/capture/replay.rs
index bdf6558a07..2ab843f860 100644
--- a/timely/src/dataflow/operators/core/capture/replay.rs
+++ b/timely/src/dataflow/operators/core/capture/replay.rs
@@ -50,7 +50,7 @@ use crate::Container;
 
 /// Replay a capture stream into a scope with the same timestamp.
 pub trait Replay<T: Timestamp, C> : Sized {
-    /// Replays `self` into the provided scope, as a `StreamCore<S,CD>`.
+    /// Replays `self` into the provided scope, as a `StreamCore<S, C>`.
     fn replay_into<S: Scope<Timestamp=T>>(self, scope: &mut S) -> StreamCore<S, C> {
         self.replay_core(scope, Some(std::time::Duration::new(0, 0)))
     }
diff --git a/timely/src/dataflow/operators/core/reclock.rs b/timely/src/dataflow/operators/core/reclock.rs
index ec9617fb7f..9510e87223 100644
--- a/timely/src/dataflow/operators/core/reclock.rs
+++ b/timely/src/dataflow/operators/core/reclock.rs
@@ -19,7 +19,7 @@ pub trait Reclock<S: Scope> {
     /// # Examples
     ///
     /// ```
-    /// use timely::dataflow::operators::{Inspect, ToStream, Delay, Map, Reclock, Capture};
+    /// use timely::dataflow::operators::{ToStream, Delay, Map, Reclock, Capture};
     /// use timely::dataflow::operators::capture::Extract;
     ///
     /// let captured = timely::example(|scope| {
@@ -36,7 +36,6 @@ pub trait Reclock<S: Scope> {
     ///
     ///     // reclock the data.
     ///     data.reclock(&clock)
-    ///         .inspect_batch(|t,x| println!("SEEN: {:?}, {:?}", t, x))
     ///         .capture()
     /// });
     ///
diff --git a/timely/src/dataflow/operators/core/to_stream.rs b/timely/src/dataflow/operators/core/to_stream.rs
index 848adf3c1d..a7d874be4e 100644
--- a/timely/src/dataflow/operators/core/to_stream.rs
+++ b/timely/src/dataflow/operators/core/to_stream.rs
@@ -12,13 +12,12 @@ pub trait ToStream<C: Container> {
     /// # Examples
     ///
     /// ```
-    /// use timely::dataflow::operators::ToStream;
-    /// use timely::dataflow::operators::Capture;
-    /// use timely::dataflow::operators::capture::Extract;
+    /// use timely::dataflow::operators::core::{ToStream, Capture};
+    /// use timely::dataflow::operators::core::capture::Extract;
     ///
     /// let (data1, data2) = timely::example(|scope| {
-    ///     let data1 = (0..3).to_stream(scope).capture();
-    ///     let data2 = vec![0,1,2].to_stream(scope).capture();
+    ///     let data1 = (0..3).to_stream(scope).container::<Vec<_>>().capture();
+    ///     let data2 = vec![0,1,2].to_stream(scope).container::<Vec<_>>().capture();
     ///     (data1, data2)
     /// });
     ///