Return lifetimes to Scope and friends (#785)

* Add scope lifetime to foundational types

* Batch2 operators; those that return Self

* Batch3: operator traits that must reflect lifetimes

* Batch4: infrastructural operators/traits

* Further tidying

* Simplifications

* Relax bounds on enter/leave

Author: frankmcsherry

mdbook/src/chapter_4/chapter_4_4.md

@@ -56,7 +56,7 @@ One nice aspect of `capture_into` is that it really does reveal everything that
 At *its* core, `replay_into` takes some sequence of `Event<T, D>` items and reproduces the stream, as it was recorded. It is also fairly simple, and we can just look at its implementation as well:
 
 ```rust,ignore
-    fn replay_into<T: Timestamp>(self, scope: &mut Scope<T>) -> Stream<T, C>{
+    fn replay_into<'scope, T: Timestamp>(self, scope: &mut Scope<'scope, T>) -> Stream<'scope, T, C>{
 
         let mut builder = OperatorBuilder::new("Replay".to_owned(), scope.clone());
         let (targets, stream) = builder.new_output();

timely/examples/unionfind.rs

@@ -54,8 +54,8 @@ trait UnionFind {
     fn union_find(self) -> Self;
 }
 
-impl<T: Timestamp> UnionFind for StreamVec<T, (usize, usize)> {
-    fn union_find(self) -> StreamVec<T, (usize, usize)> {
+impl<'scope, T: Timestamp> UnionFind for StreamVec<'scope, T, (usize, usize)> {
+    fn union_find(self) -> Self {
 
         self.unary(Pipeline, "UnionFind", |_,_| {
 

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

@@ -115,7 +115,7 @@ pub trait Capture<T: Timestamp, C: Container> : Sized {
     }
 }
 
-impl<T: Timestamp, C: Container> Capture<T, C> for Stream<T, C> {
+impl<T: Timestamp, C: Container> Capture<T, C> for Stream<'_, T, C> {
     fn capture_into<P: EventPusher<T, C>+'static>(self, mut event_pusher: P) {
 
         let mut builder = OperatorBuilder::new("Capture".to_owned(), self.scope());

timely/src/dataflow/operators/core/capture/replay.rs

@@ -51,24 +51,24 @@ use crate::dataflow::channels::Message;
 
 /// 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<T, C>`.
-    fn replay_into(self, scope: &mut Scope<T>) -> Stream<T, C> {
+    /// Replays `self` into the provided scope, as a `Stream<'scope, T, C>`.
+    fn replay_into<'scope>(self, scope: &mut Scope<'scope, T>) -> Stream<'scope, T, C> {
         self.replay_core(scope, Some(std::time::Duration::new(0, 0)))
     }
-    /// Replays `self` into the provided scope, as a `Stream<T, C>`.
+    /// Replays `self` into the provided scope, as a `Stream<'scope, T, 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
     /// re-activate itself.
-    fn replay_core(self, scope: &mut Scope<T>, period: Option<std::time::Duration>) -> Stream<T, C>;
+    fn replay_core<'scope>(self, scope: &mut Scope<'scope, T>, period: Option<std::time::Duration>) -> Stream<'scope, T, C>;
 }
 
 impl<T: Timestamp, C: Container+Clone, I> Replay<T, C> for I
 where
     I : IntoIterator,
     <I as IntoIterator>::Item: EventIterator<T, C>+'static,
 {
-    fn replay_core(self, scope: &mut Scope<T>, period: Option<std::time::Duration>) -> Stream<T, C>{
+    fn replay_core<'scope>(self, scope: &mut Scope<'scope, T>, period: Option<std::time::Duration>) -> Stream<'scope, T, C>{
 
         let mut builder = OperatorBuilder::new("Replay".to_owned(), scope.clone());
 

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

@@ -6,7 +6,7 @@ use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{Stream, Scope};
 
 /// Merge the contents of two streams.
-pub trait Concat<T: Timestamp, C> {
+pub trait Concat{
     /// Merge the contents of two streams.
     ///
     /// # Examples
@@ -22,17 +22,17 @@ pub trait Concat<T: Timestamp, C> {
     ///           .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```
-    fn concat(self, other: Stream<T, C>) -> Stream<T, C>;
+    fn concat(self, other: Self) -> Self;
 }
 
-impl<T: Timestamp, C: Container> Concat<T, C> for Stream<T, C> {
-    fn concat(self, other: Stream<T, C>) -> Stream<T, C> {
+impl<'scope, T: Timestamp, C: Container> Concat for Stream<'scope, T, C> {
+    fn concat(self, other: Self) -> Self {
         self.scope().concatenate([self, other])
     }
 }
 
 /// Merge the contents of multiple streams.
-pub trait Concatenate<T: Timestamp, C> {
+pub trait Concatenate<'scope, T: Timestamp> {
     /// Merge the contents of multiple streams.
     ///
     /// # Examples
@@ -50,15 +50,15 @@ pub trait Concatenate<T: Timestamp, C> {
     ///          .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```
-    fn concatenate<I>(&self, sources: I) -> Stream<T, C>
+    fn concatenate<I, C: Container>(&self, sources: I) -> Stream<'scope, T, C>
     where
-        I: IntoIterator<Item=Stream<T, C>>;
+        I: IntoIterator<Item=Stream<'scope, T, C>>;
 }
 
-impl<T: Timestamp, C: Container> Concatenate<T, C> for Scope<T> {
-    fn concatenate<I>(&self, sources: I) -> Stream<T, C>
+impl<'scope, T: Timestamp> Concatenate<'scope, T> for Scope<'scope, T> {
+    fn concatenate<I, C: Container>(&self, sources: I) -> Stream<'scope, T, C>
     where
-        I: IntoIterator<Item=Stream<T, C>>
+        I: IntoIterator<Item=Stream<'scope, T, C>>
     {
 
         // create an operator builder.

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

@@ -34,7 +34,7 @@ use crate::worker::AsWorker;
 use crate::dataflow::{Stream, Scope};
 
 /// Extension trait to move a `Stream` into a child of its current `Scope`.
-pub trait Enter<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, C> {
+pub trait Enter<'outer, TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, C> {
     /// Moves the `Stream` argument into a child of its current `Scope`.
     ///
     /// The destination scope must be a child of the stream's scope.
@@ -51,16 +51,16 @@ pub trait Enter<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, C> {
     ///     });
     /// });
     /// ```
-    fn enter(self, inner: &Scope<TInner>) -> Stream<TInner, C>;
+    fn enter<'inner>(self, inner: &Scope<'inner, TInner>) -> Stream<'inner, TInner, C>;
 }
 
-impl<TOuter, TInner, C> Enter<TOuter, TInner, C> for Stream<TOuter, C>
+impl<'outer, TOuter, TInner, C> Enter<'outer, TOuter, TInner, C> for Stream<'outer, TOuter, C>
 where
     TOuter: Timestamp,
     TInner: Timestamp + Refines<TOuter>,
     C: Container,
 {
-    fn enter(self, inner: &Scope<TInner>) -> Stream<TInner, C> {
+    fn enter<'inner>(self, inner: &Scope<'inner, TInner>) -> Stream<'inner, TInner, C> {
 
         use crate::scheduling::Scheduler;
 
@@ -103,7 +103,7 @@ where
 }
 
 /// Extension trait to move a `Stream` to the parent of its current `Scope`.
-pub trait Leave<TOuter: Timestamp, C> {
+pub trait Leave<'inner, TInner: Timestamp, C> {
     /// Moves a `Stream` to the parent of its current `Scope`.
     ///
     /// The parent scope must be supplied as an argument.
@@ -122,16 +122,15 @@ pub trait Leave<TOuter: Timestamp, C> {
     ///     });
     /// });
     /// ```
-    fn leave(self, outer: &Scope<TOuter>) -> Stream<TOuter, C>;
+    fn leave<'outer, TOuter: Timestamp>(self, outer: &Scope<'outer, TOuter>) -> Stream<'outer, TOuter, C> where TInner: Refines<TOuter>;
 }
 
-impl<TOuter, TInner, C> Leave<TOuter, C> for Stream<TInner, C>
+impl<'inner, TInner, C> Leave<'inner, TInner, C> for Stream<'inner, TInner, C>
 where
-    TOuter: Timestamp,
-    TInner: Timestamp + Refines<TOuter>,
+    TInner: Timestamp,
     C: Container,
 {
-    fn leave(self, outer: &Scope<TOuter>) -> Stream<TOuter, C> {
+    fn leave<'outer, TOuter: Timestamp>(self, outer: &Scope<'outer, TOuter>) -> Stream<'outer, TOuter, C> where TInner: Refines<TOuter> {
 
         let scope = self.scope();
 

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

@@ -30,7 +30,7 @@ pub trait Exchange<C: DrainContainer> {
         for<'a> F: FnMut(&C::Item<'a>) -> u64 + 'static;
 }
 
-impl<T: Timestamp, C> Exchange<C> for Stream<T, C>
+impl<T: Timestamp, C> Exchange<C> for Stream<'_, T, C>
 where
     C: Container
         + SizableContainer
@@ -39,7 +39,7 @@ where
         + crate::dataflow::channels::ContainerBytes
         + for<'a> PushInto<C::Item<'a>>,
 {
-    fn exchange<F>(self, route: F) -> Stream<T, C>
+    fn exchange<F>(self, route: F) -> Self
     where
         for<'a> F: FnMut(&C::Item<'a>) -> u64 + 'static,
     {

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

@@ -10,7 +10,7 @@ use crate::progress::frontier::Antichain;
 use crate::progress::{Timestamp, PathSummary};
 
 /// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
-pub trait Feedback<T: Timestamp> {
+pub trait Feedback<'scope, T: Timestamp> {
 
     /// Creates a [Stream] and a [Handle] to later bind the source of that `Stream`.
     ///
@@ -35,11 +35,11 @@ pub trait Feedback<T: Timestamp> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn feedback<C: Container>(&mut self, summary: <T as Timestamp>::Summary) -> (Handle<T, C>, Stream<T, C>);
+    fn feedback<C: Container>(&mut self, summary: <T as Timestamp>::Summary) -> (Handle<'scope, T, C>, Stream<'scope, T, C>);
 }
 
 /// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
-pub trait LoopVariable<TOuter: Timestamp, TInner: Timestamp> {
+pub trait LoopVariable<'scope, TOuter: Timestamp, TInner: Timestamp> {
     /// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
     ///
     /// The resulting `Stream` will have its data defined by a future call to `connect_loop` with
@@ -65,12 +65,12 @@ pub trait LoopVariable<TOuter: Timestamp, TInner: Timestamp> {
     ///     });
     /// });
     /// ```
-    fn loop_variable<C: Container>(&mut self, summary: TInner::Summary) -> (Handle<Product<TOuter, TInner>, C>, Stream<Product<TOuter, TInner>, C>);
+    fn loop_variable<C: Container>(&mut self, summary: TInner::Summary) -> (Handle<'scope, Product<TOuter, TInner>, C>, Stream<'scope, Product<TOuter, TInner>, C>);
 }
 
-impl<T: Timestamp> Feedback<T> for Scope<T> {
+impl<'scope, T: Timestamp> Feedback<'scope, T> for Scope<'scope, T> {
 
-    fn feedback<C: Container>(&mut self, summary: <T as Timestamp>::Summary) -> (Handle<T, C>, Stream<T, C>) {
+    fn feedback<C: Container>(&mut self, summary: <T as Timestamp>::Summary) -> (Handle<'scope, T, C>, Stream<'scope, T, C>) {
 
         let mut builder = OperatorBuilder::new("Feedback".to_owned(), self.clone());
         let (output, stream) = builder.new_output();
@@ -79,14 +79,14 @@ impl<T: Timestamp> Feedback<T> for Scope<T> {
     }
 }
 
-impl<TOuter: Timestamp, TInner: Timestamp> LoopVariable<TOuter, TInner> for Iterative<TOuter, TInner> {
-    fn loop_variable<C: Container>(&mut self, summary: TInner::Summary) -> (Handle<Product<TOuter, TInner>, C>, Stream<Product<TOuter, TInner>, C>) {
+impl<'scope, TOuter: Timestamp, TInner: Timestamp> LoopVariable<'scope, TOuter, TInner> for Iterative<'scope, TOuter, TInner> {
+    fn loop_variable<C: Container>(&mut self, summary: TInner::Summary) -> (Handle<'scope, Product<TOuter, TInner>, C>, Stream<'scope, Product<TOuter, TInner>, C>) {
         self.feedback(Product::new(Default::default(), summary))
     }
 }
 
 /// Connect a `Stream` to the input of a loop variable.
-pub trait ConnectLoop<T: Timestamp, C: Container> {
+pub trait ConnectLoop<'scope, T: Timestamp, C: Container> {
     /// Connect a `Stream` to be the input of a loop variable.
     ///
     /// # Examples
@@ -106,11 +106,11 @@ pub trait ConnectLoop<T: Timestamp, C: Container> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn connect_loop(self, handle: Handle<T, C>);
+    fn connect_loop(self, handle: Handle<'scope, T, C>);
 }
 
-impl<T: Timestamp, C: Container> ConnectLoop<T, C> for Stream<T, C> {
-    fn connect_loop(self, handle: Handle<T, C>) {
+impl<'scope, T: Timestamp, C: Container> ConnectLoop<'scope, T, C> for Stream<'scope, T, C> {
+    fn connect_loop(self, handle: Handle<'scope, T, C>) {
 
         let mut builder = handle.builder;
         let summary = handle.summary;
@@ -133,8 +133,8 @@ impl<T: Timestamp, C: Container> ConnectLoop<T, C> for Stream<T, C> {
 
 /// A handle used to bind the source of a loop variable.
 #[derive(Debug)]
-pub struct Handle<T: Timestamp, C: Container> {
-    builder: OperatorBuilder<T>,
+pub struct Handle<'scope, T: Timestamp, C: Container> {
+    builder: OperatorBuilder<'scope, T>,
     summary: <T as Timestamp>::Summary,
     output: crate::dataflow::channels::pushers::Output<T, C>,
 }

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

@@ -24,11 +24,11 @@ pub trait Filter<C: DrainContainer> {
     fn filter<P: FnMut(&C::Item<'_>)->bool+'static>(self, predicate: P) -> Self;
 }
 
-impl<T: Timestamp, C: Container + SizableContainer + DrainContainer> Filter<C> for Stream<T, C>
+impl<T: Timestamp, C: Container + SizableContainer + DrainContainer> Filter<C> for Stream<'_, T, C>
 where
     for<'a> C: PushInto<C::Item<'a>>
 {
-    fn filter<P: FnMut(&C::Item<'_>)->bool+'static>(self, mut predicate: P) -> Stream<T, C> {
+    fn filter<P: FnMut(&C::Item<'_>)->bool+'static>(self, mut predicate: P) -> Self {
         self.unary(Pipeline, "Filter", move |_,_| move |input, output| {
             input.for_each_time(|time, data| {
                 output.session(&time)

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

@@ -26,7 +26,7 @@ use crate::dataflow::channels::Message;
 // NOTE : Might be able to fix with another lifetime parameter, say 'c: 'a.
 
 /// Create a new `Stream` and `Handle` through which to supply input.
-pub trait Input {
+pub trait Input<'scope> {
     /// The timestamp at which this input scope operates.
     type Timestamp: Timestamp;
 
@@ -63,7 +63,7 @@ pub trait Input {
     ///     }
     /// });
     /// ```
-    fn new_input<C: Container+Clone>(&mut self) -> (Handle<Self::Timestamp, CapacityContainerBuilder<C>>, Stream<Self::Timestamp, C>);
+    fn new_input<C: Container+Clone>(&mut self) -> (Handle<Self::Timestamp, CapacityContainerBuilder<C>>, Stream<'scope, Self::Timestamp, C>);
 
     /// Create a new [Stream] and [Handle] through which to supply input.
     ///
@@ -100,7 +100,7 @@ pub trait Input {
     ///     }
     /// });
     /// ```
-    fn new_input_with_builder<CB: ContainerBuilder<Container: Clone>>(&mut self) -> (Handle<Self::Timestamp, CB>, Stream<Self::Timestamp, CB::Container>);
+    fn new_input_with_builder<CB: ContainerBuilder<Container: Clone>>(&mut self) -> (Handle<Self::Timestamp, CB>, Stream<'scope, Self::Timestamp, CB::Container>);
 
     /// Create a new stream from a supplied interactive handle.
     ///
@@ -133,25 +133,25 @@ pub trait Input {
     ///     }
     /// });
     /// ```
-    fn input_from<CB: ContainerBuilder<Container: Clone>>(&mut self, handle: &mut Handle<Self::Timestamp, CB>) -> Stream<Self::Timestamp, CB::Container>;
+    fn input_from<CB: ContainerBuilder<Container: Clone>>(&mut self, handle: &mut Handle<Self::Timestamp, CB>) -> Stream<'scope, Self::Timestamp, CB::Container>;
 }
 
 use crate::order::TotalOrder;
-impl<T: Timestamp + TotalOrder> Input for Scope<T> {
+impl<'scope, T: Timestamp + TotalOrder> Input<'scope> for Scope<'scope, T> {
     type Timestamp = T;
-    fn new_input<C: Container+Clone>(&mut self) -> (Handle<T, CapacityContainerBuilder<C>>, Stream<T, C>) {
+    fn new_input<C: Container+Clone>(&mut self) -> (Handle<T, CapacityContainerBuilder<C>>, Stream<'scope, T, C>) {
         let mut handle = Handle::new();
         let stream = self.input_from(&mut handle);
         (handle, stream)
     }
 
-    fn new_input_with_builder<CB: ContainerBuilder<Container: Clone>>(&mut self) -> (Handle<T, CB>, Stream<T, CB::Container>) {
+    fn new_input_with_builder<CB: ContainerBuilder<Container: Clone>>(&mut self) -> (Handle<T, CB>, Stream<'scope, T, CB::Container>) {
         let mut handle = Handle::new_with_builder();
         let stream = self.input_from(&mut handle);
         (handle, stream)
     }
 
-    fn input_from<CB: ContainerBuilder<Container: Clone>>(&mut self, handle: &mut Handle<T, CB>) -> Stream<T, CB::Container> {
+    fn input_from<CB: ContainerBuilder<Container: Clone>>(&mut self, handle: &mut Handle<T, CB>) -> Stream<'scope, T, CB::Container> {
         let (output, registrar) = Tee::<T, CB::Container>::new();
         let counter = Counter::new(output);
         let produced = Rc::clone(counter.produced());
@@ -338,7 +338,7 @@ impl<T: Timestamp, CB: ContainerBuilder<Container: Clone>> Handle<T, CB> {
     ///     }
     /// });
     /// ```
-    pub fn to_stream(&mut self, scope: &mut Scope<T>) -> Stream<T, CB::Container>
+    pub fn to_stream<'scope>(&mut self, scope: &mut Scope<'scope, T>) -> Stream<'scope, T, CB::Container>
     where
         T: TotalOrder,
     {