Present owned OutputBuilderSession

frankmcsherry · frankmcsherry · commit c27a14ba6db2 · 2025-09-20T18:10:01.000-04:00
diff --git a/mdbook/src/chapter_2/chapter_2_4.md b/mdbook/src/chapter_2/chapter_2_4.md
@@ -19,7 +19,7 @@ fn main() {
             .to_stream(scope)
             .unary(Pipeline, "increment", |capability, info| {
 
-                move |input, output| {
+                move |input, mut output| {
                     input.for_each_time(|time, data| {
                         let mut session = output.session(&time);
                         for datum in data.flat_map(|d| d.drain(..)) {
@@ -81,7 +81,7 @@ fn main() {
             let activator = scope.activator_for(info.address);
 
             let mut cap = Some(capability);
-            move |output| {
+            move |mut output| {
 
                 let mut done = false;
                 if let Some(cap) = cap.as_mut() {
@@ -135,7 +135,7 @@ fn main() {
 
                 let mut maximum = 0;    // define this here; use in the closure
 
-                move |input, output| {
+                move |input, mut output| {
                     input.for_each_time(|time, data| {
                         let mut session = output.session(&time);
                         for datum in data.flat_map(|d| d.drain(..)) {
@@ -187,7 +187,7 @@ fn main() {
             let mut notificator = FrontierNotificator::default();
             let mut stash = HashMap::new();
 
-            move |(input1, frontier1), (input2, frontier2), output| {
+            move |(input1, frontier1), (input2, frontier2), mut output| {
                 input1.for_each_time(|time, data| {
                     stash.entry(time.time().clone())
                          .or_insert(Vec::new())
@@ -237,7 +237,7 @@ fn main() {
 
             let mut stash = HashMap::new();
 
-            move |(input1, frontier1), (input2, frontier2), output| {
+            move |(input1, frontier1), (input2, frontier2), mut output| {
 
                 input1.for_each_time(|time, data| {
                     stash.entry(time.retain())
diff --git a/mdbook/src/chapter_2/chapter_2_5.md b/mdbook/src/chapter_2/chapter_2_5.md
@@ -206,7 +206,7 @@ As before, I'm just going to show you the new code, which now lives just after `
                     let mut counts = HashMap::new();
                     let mut buffer = Vec::new();
 
-                    move |(input, frontier), output| {
+                    move |(input, frontier), mut output| {
 
                         // for each input batch, stash it at `time`.
                         input.for_each_time(|time, data| {
diff --git a/mdbook/src/chapter_4/chapter_4_3.md b/mdbook/src/chapter_4/chapter_4_3.md
@@ -78,7 +78,7 @@ fn main() {
             // Buffer records until all prior timestamps have completed.
             .binary_frontier(&cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
 
-                move |(input1, frontier1), (input2, frontier2), output| {
+                move |(input1, frontier1), (input2, frontier2), mut output| {
 
                     // Stash received data.
                     input1.for_each_time(|time, data| {
diff --git a/timely/examples/bfs.rs b/timely/examples/bfs.rs
@@ -50,7 +50,7 @@ fn main() {
                 Exchange::new(|x: &(u32, u32)| u64::from(x.0)),
                 "BFS",
                 vec![],
-                move |input1, input2, output, notify| {
+                move |input1, input2, mut output, notify| {
 
                     // receive edges, start to sort them
                     input1.for_each_time(|time, data| {
diff --git a/timely/examples/columnar.rs b/timely/examples/columnar.rs
@@ -43,7 +43,7 @@ fn main() {
                     Pipeline,
                     "Split",
                     |_cap, _info| {
-                        move |input, output| {
+                        move |input, mut output| {
                             input.for_each_time(|time, data| {
                                 let mut session = output.session(&time);
                                 for data in data {
@@ -65,7 +65,7 @@ fn main() {
                         let mut queues = HashMap::new();
                         let mut counts = HashMap::new();
 
-                        move |(input, frontier), output| {
+                        move |(input, frontier), mut output| {
                             input.for_each_time(|time, data| {
                                 queues
                                     .entry(time.retain())
diff --git a/timely/examples/distinct.rs b/timely/examples/distinct.rs
@@ -17,7 +17,7 @@ fn main() {
             let mut counts_by_time = HashMap::new();
             scope.input_from(&mut input)
                 .unary(Exchange::new(|x| *x), "Distinct", move |_, _|
-                    move |input, output| {
+                    move |input, mut output| {
                         input.for_each_time(|time, data| {
                             let counts =
                             counts_by_time
diff --git a/timely/examples/hashjoin.rs b/timely/examples/hashjoin.rs
@@ -37,7 +37,7 @@ fn main() {
                     let mut map1 = HashMap::<u64, Vec<u64>>::new();
                     let mut map2 = HashMap::<u64, Vec<u64>>::new();
 
-                    move |input1, input2, output| {
+                    move |input1, input2, mut output| {
 
                         // Drain first input, check second map, update first map.
                         input1.for_each_time(|time, data| {
diff --git a/timely/examples/pagerank.rs b/timely/examples/pagerank.rs
@@ -41,7 +41,7 @@ fn main() {
 
                     let timer = ::std::time::Instant::now();
 
-                    move |(input1, frontier1), (input2, frontier2), output| {
+                    move |(input1, frontier1), (input2, frontier2), mut output| {
 
                         // hold on to edge changes until it is time.
                         input1.for_each_time(|time, data| {
diff --git a/timely/examples/unionfind.rs b/timely/examples/unionfind.rs
@@ -59,7 +59,7 @@ impl<G: Scope> UnionFind for Stream<G, (usize, usize)> {
             let mut roots = vec![];  // u32 works, and is smaller than uint/u64
             let mut ranks = vec![];  // u8 should be large enough (n < 2^256)
 
-            move |input, output| {
+            move |input, mut output| {
 
                 input.for_each_time(|time, data| {
                     let mut session = output.session(&time);
diff --git a/timely/examples/wordcount.rs b/timely/examples/wordcount.rs
@@ -29,7 +29,7 @@ fn main() {
                     let mut queues = HashMap::new();
                     let mut counts = HashMap::new();
 
-                    move |(input, frontier), output| {
+                    move |(input, frontier), mut output| {
                         input.for_each_time(|time, data| {
                             queues.entry(time.retain())
                                   .or_insert(Vec::new())
diff --git a/timely/src/dataflow/operators/aggregation/aggregate.rs b/timely/src/dataflow/operators/aggregation/aggregate.rs
@@ -76,7 +76,7 @@ impl<S: Scope<Timestamp: ::std::hash::Hash>, K: ExchangeData+Hash+Eq, V: Exchang
         hash: H) -> Stream<S, R> where S::Timestamp: Eq {
 
         let mut aggregates = HashMap::new();
-        self.unary_notify(Exchange::new(move |(k, _)| hash(k)), "Aggregate", vec![], move |input, output, notificator| {
+        self.unary_notify(Exchange::new(move |(k, _)| hash(k)), "Aggregate", vec![], move |input, mut output, notificator| {
 
             // read each input, fold into aggregates
             input.for_each_time(|time, data| {
diff --git a/timely/src/dataflow/operators/aggregation/state_machine.rs b/timely/src/dataflow/operators/aggregation/state_machine.rs
@@ -66,7 +66,7 @@ impl<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> StateMachine<S, K, V> f
         let mut pending: HashMap<_, Vec<(K, V)>> = HashMap::new();   // times -> (keys -> state)
         let mut states = HashMap::new();    // keys -> state
 
-        self.unary_notify(Exchange::new(move |(k, _)| hash(k)), "StateMachine", vec![], move |input, output, notificator| {
+        self.unary_notify(Exchange::new(move |(k, _)| hash(k)), "StateMachine", vec![], move |input, mut output, notificator| {
 
             // go through each time with data, process each (key, val) pair.
             notificator.for_each(|time,_,_| {
diff --git a/timely/src/dataflow/operators/capability.rs b/timely/src/dataflow/operators/capability.rs
@@ -427,7 +427,7 @@ impl<T: Timestamp> CapabilitySet<T> {
     ///     vec![()].into_iter().to_stream(scope)
     ///         .unary_frontier(Pipeline, "example", |default_cap, _info| {
     ///             let mut cap = CapabilitySet::from_elem(default_cap);
-    ///             move |(input, frontier), output| {
+    ///             move |(input, frontier), mut output| {
     ///                 cap.downgrade(&frontier.frontier());
     ///                 input.for_each_time(|time, data| {});
     ///                 let a_cap = cap.first();
diff --git a/timely/src/dataflow/operators/core/exchange.rs b/timely/src/dataflow/operators/core/exchange.rs
@@ -42,7 +42,7 @@ where
         for<'a> F: FnMut(&C::Item<'a>) -> u64 + 'static,
     {
         self.unary(ExchangeCore::new(route), "Exchange", |_, _| {
-            move |input, output| {
+            move |input, mut output| {
                 input.for_each_time(|time, data| {
                     output.session(&time).give_containers(data);
                 });
diff --git a/timely/src/dataflow/operators/core/filter.rs b/timely/src/dataflow/operators/core/filter.rs
@@ -28,7 +28,7 @@ where
     for<'a> C: PushInto<C::Item<'a>>
 {
     fn filter<P: FnMut(&C::Item<'_>)->bool+'static>(&self, mut predicate: P) -> StreamCore<G, C> {
-        self.unary(Pipeline, "Filter", move |_,_| move |input, output| {
+        self.unary(Pipeline, "Filter", move |_,_| move |input, mut output| {
             input.for_each_time(|time, data| {
                 output.session(&time)
                       .give_iterator(data.flat_map(|d| d.drain()).filter(&mut predicate));
diff --git a/timely/src/dataflow/operators/core/inspect.rs b/timely/src/dataflow/operators/core/inspect.rs
@@ -133,7 +133,7 @@ impl<G: Scope, C: Container> InspectCore<G, C> for StreamCore<G, C> {
     {
         use crate::progress::timestamp::Timestamp;
         let mut frontier = crate::progress::Antichain::from_elem(G::Timestamp::minimum());
-        self.unary_frontier(Pipeline, "InspectBatch", move |_,_| move |(input, chain), output| {
+        self.unary_frontier(Pipeline, "InspectBatch", move |_,_| move |(input, chain), mut output| {
             if chain.frontier() != frontier.borrow() {
                 frontier.clear();
                 frontier.extend(chain.frontier().iter().cloned());
diff --git a/timely/src/dataflow/operators/core/map.rs b/timely/src/dataflow/operators/core/map.rs
@@ -96,7 +96,7 @@ impl<S: Scope, C: Container + DrainContainer> Map<S, C> for StreamCore<S, C> {
         C2: Container + SizableContainer + PushInto<I::Item>,
         L: FnMut(C::Item<'_>)->I + 'static,
     {
-        self.unary(Pipeline, "FlatMap", move |_,_| move |input, output| {
+        self.unary(Pipeline, "FlatMap", move |_,_| move |input, mut output| {
             input.for_each_time(|time, data| {
                 output.session(&time)
                       .give_iterator(data.flat_map(|d| d.drain()).flat_map(&mut logic));
diff --git a/timely/src/dataflow/operators/core/rc.rs b/timely/src/dataflow/operators/core/rc.rs
@@ -27,7 +27,7 @@ pub trait SharedStream<S: Scope, C> {
 impl<S: Scope, C: Container> SharedStream<S, C> for StreamCore<S, C> {
     fn shared(&self) -> StreamCore<S, Rc<C>> {
         self.unary(Pipeline, "Shared", move |_, _| {
-            move |input, output| {
+            move |input, mut output| {
                 input.for_each_time(|time, data| {
                     let mut session = output.session(&time);
                     for data in data {
@@ -54,14 +54,14 @@ mod test {
             scope
                 .concatenate([
                     shared.unary(Pipeline, "read shared 1", |_, _| {
-                        move |input, output| {
+                        move |input, mut output| {
                             input.for_each_time(|time, data| {
                                 output.session(&time).give_iterator(data.map(|d| d.as_ptr() as usize));
                             });
                         }
                     }),
                     shared.unary(Pipeline, "read shared 2", |_, _| {
-                        move |input, output| {
+                        move |input, mut output| {
                             input.for_each_time(|time, data| {
                                 output.session(&time).give_iterator(data.map(|d| d.as_ptr() as usize));
                             });
diff --git a/timely/src/dataflow/operators/core/reclock.rs b/timely/src/dataflow/operators/core/reclock.rs
@@ -53,7 +53,7 @@ impl<S: Scope, C: Container> Reclock<S> for StreamCore<S, C> {
 
         let mut stash = vec![];
 
-        self.binary_notify(clock, Pipeline, Pipeline, "Reclock", vec![], move |input1, input2, output, notificator| {
+        self.binary_notify(clock, Pipeline, Pipeline, "Reclock", vec![], move |input1, input2, mut output, notificator| {
 
             // stash each data input with its timestamp.
             input1.for_each_time(|cap, data| {
diff --git a/timely/src/dataflow/operators/core/to_stream.rs b/timely/src/dataflow/operators/core/to_stream.rs
@@ -42,7 +42,7 @@ impl<CB: ContainerBuilder, I: IntoIterator+'static> ToStreamBuilder<CB> for I wh
             let mut iterator = self.into_iter().fuse();
             let mut capability = Some(capability);
 
-            move |output| {
+            move |mut output| {
 
                 if let Some(element) = iterator.next() {
                     let mut session = output.session_with_builder(capability.as_ref().unwrap());
diff --git a/timely/src/dataflow/operators/core/unordered_input.rs b/timely/src/dataflow/operators/core/unordered_input.rs
@@ -12,7 +12,7 @@ use crate::progress::Source;
 use crate::progress::ChangeBatch;
 use crate::progress::operate::Connectivity;
 use crate::dataflow::channels::pushers::{Counter, Tee, Output};
-use crate::dataflow::operators::generic::{OutputBuilder, OutputBuffer};
+use crate::dataflow::operators::generic::{OutputBuilder, OutputBuilderSession};
 use crate::dataflow::operators::{ActivateCapability, Capability};
 use crate::dataflow::{Scope, StreamCore};
 
@@ -163,7 +163,7 @@ impl<T: Timestamp, CB: ContainerBuilder> UnorderedHandle<T, CB> {
 
     /// Allocates a new automatically flushing session based on the supplied capability.
     #[inline]
-    pub fn activate(&mut self) -> ActivateOnDrop<OutputBuffer<'_, T, CB>> {
+    pub fn activate(&mut self) -> ActivateOnDrop<OutputBuilderSession<'_, T, CB>> {
         ActivateOnDrop::new(self.output.activate(), Rc::clone(&self.address), Rc::clone(&self.activations))
     }
 }
diff --git a/timely/src/dataflow/operators/count.rs b/timely/src/dataflow/operators/count.rs
@@ -52,7 +52,7 @@ impl<G: Scope<Timestamp: ::std::hash::Hash>, D: Data> Accumulate<G, D> for Strea
     fn accumulate<A: Data>(&self, default: A, logic: impl Fn(&mut A, &mut Vec<D>)+'static) -> Stream<G, A> {
 
         let mut accums = HashMap::new();
-        self.unary_notify(Pipeline, "Accumulate", vec![], move |input, output, notificator| {
+        self.unary_notify(Pipeline, "Accumulate", vec![], move |input, mut output, notificator| {
             input.for_each_time(|time, data| {
                 for data in data {
                     logic(accums.entry(time.time().clone()).or_insert_with(|| default.clone()), data);
diff --git a/timely/src/dataflow/operators/delay.rs b/timely/src/dataflow/operators/delay.rs
@@ -97,7 +97,7 @@ pub trait Delay<G: Scope, D: Data> {
 impl<G: Scope<Timestamp: ::std::hash::Hash>, D: Data> Delay<G, D> for Stream<G, D> {
     fn delay<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(&self, mut func: L) -> Self {
         let mut elements = HashMap::new();
-        self.unary_notify(Pipeline, "Delay", vec![], move |input, output, notificator| {
+        self.unary_notify(Pipeline, "Delay", vec![], move |input, mut output, notificator| {
             input.for_each_time(|time, data| {
                 for datum in data.flat_map(|d| d.drain(..)) {
                     let new_time = func(&datum, &time);
@@ -125,7 +125,7 @@ impl<G: Scope<Timestamp: ::std::hash::Hash>, D: Data> Delay<G, D> for Stream<G,
 
     fn delay_batch<L: FnMut(&G::Timestamp)->G::Timestamp+'static>(&self, mut func: L) -> Self {
         let mut elements = HashMap::new();
-        self.unary_notify(Pipeline, "Delay", vec![], move |input, output, notificator| {
+        self.unary_notify(Pipeline, "Delay", vec![], move |input, mut output, notificator| {
             input.for_each_time(|time, data| {
                 let new_time = func(&time);
                 assert!(time.time().less_equal(&new_time));
diff --git a/timely/src/dataflow/operators/filter.rs b/timely/src/dataflow/operators/filter.rs
@@ -24,7 +24,7 @@ pub trait Filter<D: Data> {
 
 impl<G: Scope, D: Data> Filter<D> for Stream<G, D> {
     fn filter<P: FnMut(&D)->bool+'static>(&self, mut predicate: P) -> Stream<G, D> {
-        self.unary(Pipeline, "Filter", move |_,_| move |input, output| {
+        self.unary(Pipeline, "Filter", move |_,_| move |input, mut output| {
             input.for_each_time(|time, data| {
                 let mut session = output.session(&time);
                 for data in data {
diff --git a/timely/src/dataflow/operators/flow_controlled.rs b/timely/src/dataflow/operators/flow_controlled.rs
@@ -20,8 +20,8 @@ pub struct IteratorSourceInput<T: Clone, D: Data, DI: IntoIterator<Item=D>, I: I
 }
 
 /// Construct a source that repeatedly calls the provided function to ingest input.
-/// 
-/// The function can return `None` to signal the end of the input. 
+///
+/// The function can return `None` to signal the end of the input.
 /// Otherwise, it should return a [`IteratorSourceInput`], where:
 /// * `lower_bound` is a lower bound on timestamps that can be emitted by this input in the future,
 ///   `Default::default()` can be used if this isn't needed (the source will assume that
@@ -85,7 +85,7 @@ pub fn iterator_source<
     source(scope, name, |cap, info| {
         let mut cap = Some(cap);
         let activator = scope.activator_for(info.address);
-        move |output| {
+        move |mut output| {
             cap = cap.take().and_then(|mut cap| {
                 loop {
                     if !probe.less_than(&target) {
diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs
@@ -114,33 +114,35 @@ pub fn new_input_handle<T: Timestamp, C: Accountable, P: Pull<Message<T, C>>>(
     }
 }
 
+use crate::dataflow::channels::pushers;
+
 /// An owning pair of output pusher and container builder.
 pub struct OutputBuilder<T: Timestamp, CB: ContainerBuilder> {
-    output: crate::dataflow::channels::pushers::Output<T, CB::Container>,
+    output: pushers::Output<T, CB::Container>,
     builder: CB,
 }
 
 impl<T: Timestamp, CB: ContainerBuilder> OutputBuilder<T, CB> {
     /// Constructs an output builder from an output and a default container builder.
-    pub fn from(output: crate::dataflow::channels::pushers::Output<T, CB::Container>) -> Self {
+    pub fn from(output: pushers::Output<T, CB::Container>) -> Self {
         Self { output, builder: CB::default() }
     }
     /// An activated output buffer for building containers.
-    pub fn activate<'a>(&'a mut self) -> OutputBuffer<'a, T, CB> {
-        OutputBuffer {
+    pub fn activate<'a>(&'a mut self) -> OutputBuilderSession<'a, T, CB> {
+        OutputBuilderSession {
             session: self.output.activate(),
             builder: &mut self.builder,
         }
     }
 }
 
 /// A wrapper around a live output session, with a container builder to buffer.
-pub struct OutputBuffer<'a, T: Timestamp, CB: ContainerBuilder> {
-    session: crate::dataflow::channels::pushers::OutputSession<'a, T, CB::Container>,
+pub struct OutputBuilderSession<'a, T: Timestamp, CB: ContainerBuilder> {
+    session: pushers::OutputSession<'a, T, CB::Container>,
     builder: &'a mut CB,
 }
 
-impl<'a, T: Timestamp, CB: ContainerBuilder> OutputBuffer<'a, T, CB> {
+impl<'a, T: Timestamp, CB: ContainerBuilder> OutputBuilderSession<'a, T, CB> {
     /// A container-building session associated with a capability.
     ///
     /// This method is the prefered way of sending records that must be accumulated into a container,
@@ -154,7 +156,7 @@ impl<'a, T: Timestamp, CB: ContainerBuilder> OutputBuffer<'a, T, CB> {
     }
 }
 
-impl<'a, T: Timestamp, C: Container> OutputBuffer<'a, T, CapacityContainerBuilder<C>> {
+impl<'a, T: Timestamp, C: Container> OutputBuilderSession<'a, T, CapacityContainerBuilder<C>> {
     /// A container-building session associated with a capability.
     ///
     /// This method is the prefered way of sending records that must be accumulated into a container,
@@ -170,7 +172,7 @@ impl<'a, T: Timestamp, C: Container> OutputBuffer<'a, T, CapacityContainerBuilde
 
 /// An active output building session, which accepts items and builds containers.
 pub struct Session<'a: 'b, 'b, T: Timestamp, CB: ContainerBuilder, CT: CapabilityTrait<T>> {
-    buffer: &'b mut OutputBuffer<'a, T, CB>,
+    buffer: &'b mut OutputBuilderSession<'a, T, CB>,
     capability: &'b CT,
 }
 
diff --git a/timely/src/dataflow/operators/generic/mod.rs b/timely/src/dataflow/operators/generic/mod.rs
@@ -8,7 +8,7 @@ mod handles;
 mod notificator;
 mod operator_info;
 
-pub use self::handles::{InputHandleCore, OutputBuilder, OutputBuffer, Session};
+pub use self::handles::{InputHandleCore, OutputBuilder, OutputBuilderSession, Session};
 pub use self::notificator::{Notificator, FrontierNotificator};
 
 pub use self::operator::{Operator, source};
diff --git a/timely/src/dataflow/operators/generic/notificator.rs b/timely/src/dataflow/operators/generic/notificator.rs
diff --git a/timely/src/dataflow/operators/generic/operator.rs b/timely/src/dataflow/operators/generic/operator.rs
diff --git a/timely/src/dataflow/operators/map.rs b/timely/src/dataflow/operators/map.rs
diff --git a/timely/src/synchronization/sequence.rs b/timely/src/synchronization/sequence.rs
diff --git a/timely/tests/gh_523.rs b/timely/tests/gh_523.rs
