make tests work

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

Author: antiguru

mdbook/src/chapter_2/chapter_2_3.md

@@ -126,12 +126,12 @@ use timely::dataflow::operators::{ToStream, Partition, Inspect};
 
 fn main() {
     timely::example(|scope| {
-        let streams = (0..10).to_stream(scope)
+        let mut streams = (0..10).to_stream(scope)
                              .partition(3, |x| (x % 3, x));
 
-        streams[0].inspect(|x| println!("seen 0: {:?}", x));
-        streams[1].inspect(|x| println!("seen 1: {:?}", x));
-        streams[2].inspect(|x| println!("seen 2: {:?}", x));
+        streams.pop().unwrap().inspect(|x| println!("seen 2: {:?}", x));
+        streams.pop().unwrap().inspect(|x| println!("seen 1: {:?}", x));
+        streams.pop().unwrap().inspect(|x| println!("seen 0: {:?}", x));
     });
 }
 ```
@@ -147,11 +147,11 @@ use timely::dataflow::operators::{ToStream, Partition, Concat, Inspect};
 
 fn main() {
     timely::example(|scope| {
-        let streams = (0..10).to_stream(scope)
+        let mut streams = (0..10).to_stream(scope)
                              .partition(3, |x| (x % 3, x));
-        streams[0]
-            .concat(&streams[1])
-            .concat(&streams[2])
+        streams.pop().unwrap()
+            .concat(streams.pop().unwrap())
+            .concat(streams.pop().unwrap())
             .inspect(|x| println!("seen: {:?}", x));
     });
 }

mdbook/src/chapter_2/chapter_2_4.md

@@ -183,7 +183,7 @@ fn main() {
         let in1 = (0 .. 10).to_stream(scope);
         let in2 = (0 .. 10).to_stream(scope);
 
-        in1.binary_frontier(&in2, Pipeline, Pipeline, "concat_buffer", |capability, info| {
+        in1.binary_frontier(in2, Pipeline, Pipeline, "concat_buffer", |capability, info| {
 
             let mut notificator = FrontierNotificator::new();
             let mut stash = HashMap::new();
@@ -234,7 +234,7 @@ fn main() {
         let in1 = (0 .. 10).to_stream(scope);
         let in2 = (0 .. 10).to_stream(scope);
 
-        in1.binary_frontier(&in2, Pipeline, Pipeline, "concat_buffer", |capability, info| {
+        in1.binary_frontier(in2, Pipeline, Pipeline, "concat_buffer", |capability, info| {
 
             let mut stash = HashMap::new();
 

mdbook/src/chapter_4/chapter_4_2.md

@@ -24,7 +24,7 @@ fn main() {
         // circulate numbers, Collatz stepping each time.
         (1 .. 10)
             .to_stream(scope)
-            .concat(&stream)
+            .concat(stream)
             .map(|x| if x % 2 == 0 { x / 2 } else { 3 * x + 1 } )
             .inspect(|x| println!("{:?}", x))
             .filter(|x| *x != 1)
@@ -63,17 +63,17 @@ fn main() {
         let results1 = stream1.map(|x| 3 * x + 1);
 
         // partition the input and feedback streams by even-ness.
-        let parts =
+        let mut parts =
             (1 .. 10)
                 .to_stream(scope)
-                .concat(&results0)
-                .concat(&results1)
+                .concat(results0)
+                .concat(results1)
                 .inspect(|x| println!("{:?}", x))
                 .partition(2, |x| (x % 2, x));
 
         // connect each part appropriately.
-        parts[0].connect_loop(handle0);
-        parts[1].connect_loop(handle1);
+        parts.pop().unwrap().connect_loop(handle1);
+        parts.pop().unwrap().connect_loop(handle0);
     });
 }
 ```
@@ -103,7 +103,7 @@ fn main() {
 
             input
                 .enter(subscope)
-                .concat(&stream)
+                .concat(stream)
                 .map(|x| if x % 2 == 0 { x / 2 } else { 3 * x + 1 } )
                 .inspect(|x| println!("{:?}", x))
                 .filter(|x| *x != 1)

mdbook/src/chapter_4/chapter_4_3.md

@@ -76,7 +76,7 @@ fn main() {
             // Assign timestamps to records so that not much work is in each time.
             .delay(|number, time| number / 100 )
             // Buffer records until all prior timestamps have completed.
-            .binary_frontier(&cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
+            .binary_frontier(cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
 
                 let mut vector = Vec::new();
 

timely/src/dataflow/operators/concat.rs

@@ -16,7 +16,7 @@ pub trait Concat<G: Scope, D: Container, S: StreamLike<G, D>> {
     /// timely::example(|scope| {
     ///
     ///     let stream = (0..10).to_stream(scope).tee();
-    ///     stream.clone().concat(stream)
+    ///     stream.concat(&stream)
     ///           .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```

timely/src/dataflow/operators/enterleave.rs

@@ -28,13 +28,13 @@ use crate::progress::{Source, Target};
 use crate::order::Product;
 use crate::{Container, Data};
 use crate::communication::Push;
-use crate::dataflow::channels::pushers::{CounterCore, TeeCore};
+use crate::dataflow::channels::pushers::{CounterCore, PushOwned};
 use crate::dataflow::channels::{BundleCore, Message};
 
 use crate::worker::AsWorker;
-use crate::dataflow::{StreamCore, Scope};
-use crate::dataflow::channels::pushers::PushOwned;
+use crate::dataflow::{OwnedStream, StreamLike, Scope};
 use crate::dataflow::scopes::{Child, ScopeParent};
+use crate::dataflow::scopes::child::Iterative;
 use crate::dataflow::operators::delay::Delay;
 
 /// Extension trait to move a `Stream` into a child of its current `Scope`.
@@ -53,12 +53,9 @@ 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>) -> OwnedStream<Child<'a, G, T>, C>;
 }
 
-use crate::dataflow::scopes::child::Iterative;
-use crate::dataflow::stream::{OwnedStream, StreamLike};
-
 /// Extension trait to move a `Stream` into a child of its current `Scope` setting the timestamp for each element.
 pub trait EnterAt<G: Scope, T: Timestamp, D: Data> {
     /// Moves the `Stream` argument into a child of its current `Scope` setting the timestamp for each element by `initial`.
@@ -86,11 +83,11 @@ impl<G: Scope, T: Timestamp, D: Data, E: Enter<G, Product<<G as ScopeParent>::Ti
 }
 
 impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container+Data, S: StreamLike<G, C>> Enter<G, T, C> for S {
-    fn enter<'a>(self, scope: &Child<'a, G, T>) -> StreamCore<Child<'a, G, T>, C> {
+    fn enter<'a>(self, scope: &Child<'a, G, T>) -> OwnedStream<Child<'a, G, T>, C> {
 
         use crate::scheduling::Scheduler;
 
-        let (targets, registrar) = TeeCore::<T, C>::new();
+        let (targets, registrar) = PushOwned::<T, C>::new();
         let ingress = IngressNub {
             targets: CounterCore::new(targets),
             phantom: PhantomData,
@@ -108,7 +105,7 @@ impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container+Data, S: StreamL
             self.connect_to(input, ingress, channel_id);
         }
 
-        StreamCore::new(
+        OwnedStream::new(
             Source::new(0, input.port),
             registrar,
             scope.clone(),
@@ -169,7 +166,7 @@ where
 
 
 struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container+Data> {
-    targets: CounterCore<TInner, TData, TeeCore<TInner, TData>>,
+    targets: CounterCore<TInner, TData, PushOwned<TInner, TData>>,
     phantom: ::std::marker::PhantomData<TOuter>,
     activator: crate::scheduling::Activator,
     active: bool,

timely/src/dataflow/operators/rc.rs

@@ -51,7 +51,6 @@ mod test {
     fn test_shared() {
         let output = crate::example(|scope| {
             let shared = vec![Ok(0), Err(())].to_stream(scope).shared().tee();
-            let shared2 = shared.clone();
             scope
                 .concatenate([
                     shared.unary(Pipeline, "read shared 1", |_, _| {
@@ -63,7 +62,7 @@ mod test {
                             });
                         }
                     }),
-                    shared2.unary(Pipeline, "read shared 2", |_, _| {
+                    shared.unary(Pipeline, "read shared 2", |_, _| {
                         let mut container = Default::default();
                         move |input, output| {
                             input.for_each(|time, data| {

timely/src/dataflow/stream.rs

@@ -69,17 +69,7 @@ impl<S: Scope, D: Container> OwnedStream<S, D> {
 /// A stream batching data in vectors.
 pub type Stream<S, D> = StreamCore<S, Vec<D>>;
 
-impl<S: Scope, D: Container> StreamLike<S, D> for StreamCore<S, D> {
-    fn connect_to<P: Push<BundleCore<S::Timestamp, D>> + 'static>(self, target: Target, pusher: P, identifier: usize) {
-        self.connect_to(target, pusher, identifier)
-    }
-
-    fn scope(&self) -> S {
-        self.scope()
-    }
-}
-
-impl<S: Scope, D: Container> StreamLike<S, D> for OwnedStream<S, D> {
+impl<S: Scope, D: Container> StreamLike<S, D> for &StreamCore<S, D> {
     fn connect_to<P: Push<BundleCore<S::Timestamp, D>> + 'static>(self, target: Target, pusher: P, identifier: usize) {
         let mut logging = self.scope().logging();
         logging.as_mut().map(|l| l.log(crate::logging::ChannelsEvent {
@@ -90,21 +80,16 @@ impl<S: Scope, D: Container> StreamLike<S, D> for OwnedStream<S, D> {
         }));
 
         self.scope.add_edge(self.name, target);
-        self.port.set(pusher);
+        self.ports.add_pusher(pusher);
     }
 
     fn scope(&self) -> S {
         self.scope.clone()
     }
 }
 
-impl<S: Scope, D: Container> StreamCore<S, D> {
-    /// Connects the stream to a destination.
-    ///
-    /// The destination is described both by a `Target`, for progress tracking information, and a `P: Push` where the
-    /// records should actually be sent. The identifier is unique to the edge and is used only for logging purposes.
-    pub fn connect_to<P: Push<BundleCore<S::Timestamp, D>>+'static>(self, target: Target, pusher: P, identifier: usize) {
-
+impl<S: Scope, D: Container> StreamLike<S, D> for OwnedStream<S, D> {
+    fn connect_to<P: Push<BundleCore<S::Timestamp, D>> + 'static>(self, target: Target, pusher: P, identifier: usize) {
         let mut logging = self.scope().logging();
         logging.as_mut().map(|l| l.log(crate::logging::ChannelsEvent {
             id: identifier,
@@ -114,16 +99,21 @@ impl<S: Scope, D: Container> StreamCore<S, D> {
         }));
 
         self.scope.add_edge(self.name, target);
-        self.ports.add_pusher(pusher);
+        self.port.set(pusher);
     }
+
+    fn scope(&self) -> S {
+        self.scope.clone()
+    }
+}
+
+impl<S: Scope, D: Container> StreamCore<S, D> {
     /// Allocates a `Stream` from a supplied `Source` name and rendezvous point.
     pub fn new(source: Source, output: TeeHelper<S::Timestamp, D>, scope: S) -> Self {
         Self { name: source, ports: output, scope }
     }
     /// The name of the stream's source operator.
     pub fn name(&self) -> &Source { &self.name }
-    /// The scope immediately containing the stream.
-    pub fn scope(&self) -> S { self.scope.clone() }
 }
 
 impl<S, D> Debug for StreamCore<S, D>