Owned streams take 2

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

Author: antiguru

timely/examples/bfs.rs

@@ -48,7 +48,7 @@ fn main() {
 
             // use the stream of edges
             graph.binary_notify(
-                &stream,
+                stream,
                 Exchange::new(|x: &(u32, u32)| u64::from(x.0)),
                 Exchange::new(|x: &(u32, u32)| u64::from(x.0)),
                 "BFS",

timely/examples/hashjoin.rs

@@ -35,7 +35,7 @@ fn main() {
             let exchange2 = Exchange::new(|x: &(u64, u64)| x.0);
 
             stream1
-                .binary(&stream2, exchange1, exchange2, "HashJoin", |_capability, _info| {
+                .binary(stream2, exchange1, exchange2, "HashJoin", |_capability, _info| {
 
                     let mut map1 = HashMap::<u64, Vec<u64>>::new();
                     let mut map2 = HashMap::<u64, Vec<u64>>::new();

timely/examples/loopdemo.rs

@@ -32,9 +32,9 @@ fn main() {
                 .concat(&loop_stream)
                 .map(|x| if x % 2 == 0 { x / 2 } else { 3 * x + 1 })
                 .filter(|x| x > &1);
-
-            step.connect_loop(loop_handle);
-            step.probe_with(&mut probe);
+            step
+                .probe_with(&mut probe)
+                .connect_loop(loop_handle);
         });
 
         let ns_per_request = 1_000_000_000 / rate;
@@ -122,4 +122,4 @@ fn main() {
         }
 
     }).unwrap();
-}
+}

timely/examples/pagerank.rs

@@ -26,7 +26,7 @@ fn main() {
 
             // bring edges and ranks together!
             let changes = edge_stream.binary_frontier(
-                &rank_stream,
+                rank_stream,
                 Exchange::new(|x: &((usize, usize), i64)| (x.0).0 as u64),
                 Exchange::new(|x: &(usize, i64)| x.0 as u64),
                 "PageRank",

timely/examples/unionfind.rs

@@ -51,11 +51,11 @@ fn main() {
 }
 
 trait UnionFind {
-    fn union_find(&self) -> Self;
+    fn union_find(self) -> Self;
 }
 
 impl<G: Scope> UnionFind for Stream<G, (usize, usize)> {
-    fn union_find(&self) -> Stream<G, (usize, usize)> {
+    fn union_find(self) -> Stream<G, (usize, usize)> {
 
         self.unary(Pipeline, "UnionFind", |_,_| {
 

timely/src/dataflow/operators/aggregation/aggregate.rs

@@ -61,7 +61,7 @@ pub trait Aggregate<S: Scope, K: ExchangeData+Hash, V: ExchangeData> {
     /// });
     /// ```
     fn aggregate<R: Data, D: Default+'static, F: Fn(&K, V, &mut D)+'static, E: Fn(K, D)->R+'static, H: Fn(&K)->u64+'static>(
-        &self,
+        self,
         fold: F,
         emit: E,
         hash: H) -> Stream<S, R> where S::Timestamp: Eq;
@@ -70,7 +70,7 @@ pub trait Aggregate<S: Scope, K: ExchangeData+Hash, V: ExchangeData> {
 impl<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> Aggregate<S, K, V> for Stream<S, (K, V)> {
 
     fn aggregate<R: Data, D: Default+'static, F: Fn(&K, V, &mut D)+'static, E: Fn(K, D)->R+'static, H: Fn(&K)->u64+'static>(
-        &self,
+        self,
         fold: F,
         emit: E,
         hash: H) -> Stream<S, R> where S::Timestamp: Eq {

timely/src/dataflow/operators/aggregation/state_machine.rs

@@ -51,7 +51,7 @@ pub trait StateMachine<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> {
         I: IntoIterator<Item=R>,                    // type of output iterator
         F: Fn(&K, V, &mut D)->(bool, I)+'static,    // state update logic
         H: Fn(&K)->u64+'static,                     // "hash" function for keys
-    >(&self, fold: F, hash: H) -> Stream<S, R> where S::Timestamp : Hash+Eq ;
+    >(self, fold: F, hash: H) -> Stream<S, R> where S::Timestamp : Hash+Eq ;
 }
 
 impl<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> StateMachine<S, K, V> for Stream<S, (K, V)> {
@@ -61,7 +61,7 @@ impl<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> StateMachine<S, K, V> f
             I: IntoIterator<Item=R>,                    // type of output iterator
             F: Fn(&K, V, &mut D)->(bool, I)+'static,    // state update logic
             H: Fn(&K)->u64+'static,                     // "hash" function for keys
-        >(&self, fold: F, hash: H) -> Stream<S, R> where S::Timestamp : Hash+Eq {
+        >(self, fold: F, hash: H) -> Stream<S, R> where S::Timestamp : Hash+Eq {
 
         let mut pending: HashMap<_, Vec<(K, V)>> = HashMap::new();   // times -> (keys -> state)
         let mut states = HashMap::new();    // keys -> state

timely/src/dataflow/operators/branch.rs

@@ -29,14 +29,14 @@ pub trait Branch<S: Scope, D: Data> {
     /// });
     /// ```
     fn branch(
-        &self,
+        self,
         condition: impl Fn(&S::Timestamp, &D) -> bool + 'static,
     ) -> (Stream<S, D>, Stream<S, D>);
 }
 
 impl<S: Scope, D: Data> Branch<S, D> for Stream<S, D> {
     fn branch(
-        &self,
+        self,
         condition: impl Fn(&S::Timestamp, &D) -> bool + 'static,
     ) -> (Stream<S, D>, Stream<S, D>) {
         let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());
@@ -91,11 +91,11 @@ pub trait BranchWhen<T>: Sized {
     ///     after_five.inspect(|x| println!("Times 5 and later: {:?}", x));
     /// });
     /// ```
-    fn branch_when(&self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self);
+    fn branch_when(self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self);
 }
 
 impl<S: Scope, C: Container> BranchWhen<S::Timestamp> for StreamCore<S, C> {
-    fn branch_when(&self, condition: impl Fn(&S::Timestamp) -> bool + 'static) -> (Self, Self) {
+    fn branch_when(self, condition: impl Fn(&S::Timestamp) -> bool + 'static) -> (Self, Self) {
         let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());
 
         let mut input = builder.new_input(self, Pipeline);

timely/src/dataflow/operators/broadcast.rs

@@ -18,11 +18,11 @@ pub trait Broadcast<D: ExchangeData> {
     ///            .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```
-    fn broadcast(&self) -> Self;
+    fn broadcast(self) -> Self;
 }
 
 impl<G: Scope, D: ExchangeData> Broadcast<D> for Stream<G, D> {
-    fn broadcast(&self) -> Stream<G, D> {
+    fn broadcast(self) -> Stream<G, D> {
 
         // NOTE: Simplified implementation due to underlying motion
         // in timely dataflow internals. Optimize once they have

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, D: Container>: Sized {
     /// 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, D>+'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, D>> {
         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) {
+    fn capture_into<P: EventPusherCore<S::Timestamp, D>+'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));

timely/src/dataflow/operators/concat.rs

@@ -73,7 +73,7 @@ impl<G: Scope, D: Container> Concatenate<G, D> for G {
         let mut builder = OperatorBuilder::new("Concatenate".to_string(), self.clone());
 
         // create new input handles for each input stream.
-        let mut handles = sources.into_iter().map(|s| builder.new_input(&s, Pipeline)).collect::<Vec<_>>();
+        let mut handles = sources.into_iter().map(|s| builder.new_input(s, Pipeline)).collect::<Vec<_>>();
 
         // create one output handle for the concatenated results.
         let (mut output, result) = builder.new_output();

timely/src/dataflow/operators/count.rs

@@ -9,7 +9,7 @@ use crate::dataflow::{Stream, Scope};
 use crate::dataflow::operators::generic::operator::Operator;
 
 /// Accumulates records within a timestamp.
-pub trait Accumulate<G: Scope, D: Data> {
+pub trait Accumulate<G: Scope, D: Data>: Sized {
     /// Accumulates records within a timestamp.
     ///
     /// # Examples
@@ -27,7 +27,7 @@ pub trait Accumulate<G: Scope, D: Data> {
     /// let extracted = captured.extract();
     /// assert_eq!(extracted, vec![(0, vec![45])]);
     /// ```
-    fn accumulate<A: Data>(&self, default: A, logic: impl Fn(&mut A, RefOrMut<Vec<D>>)+'static) -> Stream<G, A>;
+    fn accumulate<A: Data>(self, default: A, logic: impl Fn(&mut A, RefOrMut<Vec<D>>)+'static) -> Stream<G, A>;
     /// Counts the number of records observed at each time.
     ///
     /// # Examples
@@ -45,13 +45,13 @@ pub trait Accumulate<G: Scope, D: Data> {
     /// let extracted = captured.extract();
     /// assert_eq!(extracted, vec![(0, vec![10])]);
     /// ```
-    fn count(&self) -> Stream<G, usize> {
+    fn count(self) -> Stream<G, usize> {
         self.accumulate(0, |sum, data| *sum += data.len())
     }
 }
 
 impl<G: Scope, D: Data> Accumulate<G, D> for Stream<G, D> {
-    fn accumulate<A: Data>(&self, default: A, logic: impl Fn(&mut A, RefOrMut<Vec<D>>)+'static) -> Stream<G, A> {
+    fn accumulate<A: Data>(self, default: A, logic: impl Fn(&mut A, RefOrMut<Vec<D>>)+'static) -> Stream<G, A> {
 
         let mut accums = HashMap::new();
         self.unary_notify(Pipeline, "Accumulate", vec![], move |input, output, notificator| {

timely/src/dataflow/operators/delay.rs

@@ -36,7 +36,7 @@ pub trait Delay<G: Scope, D: Data> {
     ///            });
     /// });
     /// ```
-    fn delay<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(&self, func: L) -> Self;
+    fn delay<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(self, func: L) -> Self;
 
     /// Advances the timestamp of records using a supplied function.
     ///
@@ -63,7 +63,7 @@ pub trait Delay<G: Scope, D: Data> {
     ///            });
     /// });
     /// ```
-    fn delay_total<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(&self, func: L) -> Self
+    fn delay_total<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(self, func: L) -> Self
     where G::Timestamp: TotalOrder;
 
     /// Advances the timestamp of batches of records using a supplied function.
@@ -91,11 +91,11 @@ pub trait Delay<G: Scope, D: Data> {
     ///            });
     /// });
     /// ```
-    fn delay_batch<L: FnMut(&G::Timestamp)->G::Timestamp+'static>(&self, func: L) -> Self;
+    fn delay_batch<L: FnMut(&G::Timestamp)->G::Timestamp+'static>(self, func: L) -> Self;
 }
 
 impl<G: Scope, D: Data> Delay<G, D> for Stream<G, D> {
-    fn delay<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(&self, mut func: L) -> Self {
+    fn delay<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(self, mut func: L) -> Self {
         let mut elements = HashMap::new();
         let mut vector = Vec::new();
         self.unary_notify(Pipeline, "Delay", vec![], move |input, output, notificator| {
@@ -119,13 +119,13 @@ impl<G: Scope, D: Data> Delay<G, D> for Stream<G, D> {
         })
     }
 
-    fn delay_total<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(&self, func: L) -> Self
+    fn delay_total<L: FnMut(&D, &G::Timestamp)->G::Timestamp+'static>(self, func: L) -> Self
     where G::Timestamp: TotalOrder
     {
         self.delay(func)
     }
 
-    fn delay_batch<L: FnMut(&G::Timestamp)->G::Timestamp+'static>(&self, mut func: L) -> Self {
+    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| {
             input.for_each(|time, data| {

timely/src/dataflow/operators/enterleave.rs

@@ -51,7 +51,7 @@ pub trait Enter<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container> {
     ///     });
     /// });
     /// ```
-    fn enter<'a>(&self, _: &Child<'a, G, T>) -> StreamCore<Child<'a, G, T>, C>;
+    fn enter<'a>(self, _: &Child<'a, G, T>) -> StreamCore<Child<'a, G, T>, C>;
 }
 
 use crate::dataflow::scopes::child::Iterative;
@@ -72,18 +72,18 @@ pub trait EnterAt<G: Scope, T: Timestamp, D: Data> {
     ///     });
     /// });
     /// ```
-    fn enter_at<'a, F:FnMut(&D)->T+'static>(&self, scope: &Iterative<'a, G, T>, initial: F) -> Stream<Iterative<'a, G, T>, D> ;
+    fn enter_at<'a, F:FnMut(&D)->T+'static>(self, scope: &Iterative<'a, G, T>, initial: F) -> Stream<Iterative<'a, G, T>, D> ;
 }
 
 impl<G: Scope, T: Timestamp, D: Data, E: Enter<G, Product<<G as ScopeParent>::Timestamp, T>, Vec<D>>> EnterAt<G, T, D> for E {
-    fn enter_at<'a, F:FnMut(&D)->T+'static>(&self, scope: &Iterative<'a, G, T>, mut initial: F) ->
+    fn enter_at<'a, F:FnMut(&D)->T+'static>(self, scope: &Iterative<'a, G, T>, mut initial: F) ->
         Stream<Iterative<'a, G, T>, D> {
             self.enter(scope).delay(move |datum, time| Product::new(time.clone().to_outer(), initial(datum)))
     }
 }
 
 impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Data+Container> Enter<G, T, C> for StreamCore<G, C> {
-    fn enter<'a>(&self, scope: &Child<'a, G, T>) -> StreamCore<Child<'a, G, T>, C> {
+    fn enter<'a>(self, scope: &Child<'a, G, T>) -> StreamCore<Child<'a, G, T>, C> {
 
         use crate::scheduling::Scheduler;
 
@@ -120,11 +120,11 @@ pub trait Leave<G: Scope, D: Container> {
     ///     });
     /// });
     /// ```
-    fn leave(&self) -> StreamCore<G, D>;
+    fn leave(self) -> StreamCore<G, D>;
 }
 
 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> {
+    fn leave(self) -> StreamCore<G, D> {
 
         let scope = self.scope();
 

timely/src/dataflow/operators/exchange.rs

@@ -23,15 +23,15 @@ pub trait Exchange<D> {
     ///            .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```
-    fn exchange(&self, route: impl FnMut(&D) -> u64 + 'static) -> Self;
+    fn exchange(self, route: impl FnMut(&D) -> u64 + 'static) -> Self;
 }
 
 impl<G: Scope, C> Exchange<C::Item> for StreamCore<G, C>
 where
     C: PushPartitioned + ExchangeData,
     C::Item: ExchangeData,
 {
-    fn exchange(&self, route: impl FnMut(&C::Item) -> u64 + 'static) -> StreamCore<G, C> {
+    fn exchange(self, route: impl FnMut(&C::Item) -> u64 + 'static) -> StreamCore<G, C> {
         let mut container = Default::default();
         self.unary(ExchangeCore::new(route), "Exchange", |_, _| {
             move |input, output| {

timely/src/dataflow/operators/feedback.rs

@@ -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, D>);
 }
 
 impl<G: Scope, D: Container> ConnectLoop<G, D> for StreamCore<G, D> {
-    fn connect_loop(&self, helper: HandleCore<G, D>) {
+    fn connect_loop(self, helper: HandleCore<G, D>) {
 
         let mut builder = helper.builder;
         let summary = helper.summary;

timely/src/dataflow/operators/filter.rs

@@ -19,11 +19,11 @@ pub trait Filter<D: Data> {
     ///            .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```
-    fn filter<P: FnMut(&D)->bool+'static>(&self, predicate: P) -> Self;
+    fn filter<P: FnMut(&D)->bool+'static>(self, predicate: P) -> Self;
 }
 
 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> {
+    fn filter<P: FnMut(&D)->bool+'static>(self, mut predicate: P) -> Stream<G, D> {
         let mut vector = Vec::new();
         self.unary(Pipeline, "Filter", move |_,_| move |input, output| {
             input.for_each(|time, data| {

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

@@ -105,15 +105,15 @@ impl<G: Scope> OperatorBuilder<G> {
     }
 
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
-    pub fn new_input<D: Container, P>(&mut self, stream: &StreamCore<G, D>, pact: P) -> P::Puller
+    pub fn new_input<D: Container, P>(&mut self, stream: StreamCore<G, D>, pact: P) -> P::Puller
         where
             P: ParallelizationContractCore<G::Timestamp, D> {
         let connection = vec![Antichain::from_elem(Default::default()); self.shape.outputs];
         self.new_input_connection(stream, pact, connection)
     }
 
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
-    pub fn new_input_connection<D: 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<D: Container, P>(&mut self, stream: StreamCore<G, D>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> P::Puller
     where
         P: ParallelizationContractCore<G::Timestamp, D> {
 

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

@@ -59,7 +59,7 @@ impl<G: Scope> OperatorBuilder<G> {
     }
 
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
-    pub fn new_input<D: Container, P>(&mut self, stream: &StreamCore<G, D>, pact: P) -> InputHandleCore<G::Timestamp, D, P::Puller>
+    pub fn new_input<D: Container, P>(&mut self, stream: StreamCore<G, D>, pact: P) -> InputHandleCore<G::Timestamp, D, P::Puller>
     where
         P: ParallelizationContractCore<G::Timestamp, D> {
 
@@ -75,7 +75,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_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<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>
         where
             P: ParallelizationContractCore<G::Timestamp, D> {