Remove operators that read a time off the capability

mdbook/src/chapter_4/chapter_4_2.md

@@ -14,7 +14,7 @@ We are going to check the [Collatz conjecture](https://en.wikipedia.org/wiki/Col
 extern crate timely;
 
 use timely::dataflow::operators::{Feedback, ToStream, Concat, Inspect, ConnectLoop};
-use timely::dataflow::operators::vec::{Map, Filter, BranchWhen};
+use timely::dataflow::operators::vec::{Map, Filter};
 
 fn main() {
     timely::example(|scope| {
@@ -29,13 +29,12 @@ fn main() {
             .map(|x| if x % 2 == 0 { x / 2 } else { 3 * x + 1 } )
             .inspect(|x| println!("{:?}", x))
             .filter(|x| *x != 1)
-            .branch_when(|t| t < &100).1
             .connect_loop(handle);
     });
 }
 ```
 
-This program first creates a loop variable, using the `feedback` method on scopes. This method comes from the `Feedback` extension trait in `dataflow::operators`, in case you can't find it. When we create a new loop variable, we have to tell timely dataflow by how much we should increment the timestamp each time around the loop. To be more specific, we have to give a path summary which often is just a number that tells us by how much to increment the timestamp. When we later connect the output of an operation back to this loop variable we can specify an upper bound on the number of iterations by using the `branch_when` method.
+This program first creates a loop variable, using the `feedback` method on scopes. This method comes from the `Feedback` extension trait in `dataflow::operators`, in case you can't find it. When we create a new loop variable, we have to tell timely dataflow by how much we should increment the timestamp each time around the loop. To be more specific, we have to give a path summary which often is just a number that tells us by how much to increment the timestamp. The loop here ends on its own, once every number has reached one and been filtered out; a loop that might not end needs its own bound, for example a round counter carried in the data and a `filter` on it.
 
 We start with a stream of the numbers from one through nine, because we have to start somewhere. Our plan is to repeatedly apply the Collatz step, and then discard any numbers equal to one, but we want to apply this not only to our input but also to whatever comes back around our loop variable. So, the very first step is to `concat` our input stream with the feedback stream. Then we can apply the Collatz step, filter out the ones, and then connect the resulting stream as the definition of the feedback stream.
 

mdbook/src/chapter_4/chapter_4_3.md

@@ -54,13 +54,13 @@ but without actually producing the 4,999,950,000 intermediate records all at onc
 
 One way to do this is to build a self-regulating dataflow, into which we can immediately dump all the records, but which will buffer records until it is certain that the work for prior records has drained. We will write this out in all the gory details, but these operators are certainly things that could be packaged up and reused.
 
-The idea here is to take our stream of work, and to use the `delay` operator to assign new timestamps to the records. We will spread the work out so that each timestamp has at most (in this case) 100 numbers. We can write a `binary` operator that will buffer received records until their timestamp is "next", meaning all strictly prior work has drained from the dataflow fragment. How do we do this? We turn our previously unary operator into a binary operator that has a feedback edge connected to its second input. We use the frontier of that feedback input to control when we emit data.
+The idea here is to take our stream of work, and to use a small operator to assign new timestamps to the records, by sending each under a capability delayed to its new time. We will spread the work out so that each timestamp has at most (in this case) 100 numbers. We can write a `binary` operator that will buffer received records until their timestamp is "next", meaning all strictly prior work has drained from the dataflow fragment. How do we do this? We turn our previously unary operator into a binary operator that has a feedback edge connected to its second input. We use the frontier of that feedback input to control when we emit data.
 
 ```rust,no_run
 extern crate timely;
 
 use timely::dataflow::operators::{Feedback, ToStream, Operator, ConnectLoop};
-use timely::dataflow::operators::vec::{Delay, Map, Filter};
+use timely::dataflow::operators::vec::{Map, Filter};
 use timely::dataflow::channels::pact::Pipeline;
 
 fn main() {
@@ -74,8 +74,16 @@ fn main() {
         // Produce all numbers less than each input number.
         (1 .. 100_000u64)
             .to_stream(scope)
+            .container::<Vec<_>>()
             // Assign timestamps to records so that not much work is in each time.
-            .delay(|number, time| *number / 100 )
+            .unary(Pipeline, "Delay", |_capability, _info| move |input, output| {
+                input.for_each_stamp(|cap, data| {
+                    for number in data.flat_map(|d| d.drain(..)) {
+                        output.session(&cap.delayed(&(number / 100), 0)).give(number);
+                    }
+                });
+            })
+            .container::<Vec<_>>()
             // Buffer records until all prior timestamps have completed.
             .binary_frontier(cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
 

timely/examples/pingpong.rs

@@ -1,4 +1,4 @@
-use timely::dataflow::operators::{ToStream, Exchange, Feedback, Concat, ConnectLoop, vec::{Map, BranchWhen}};
+use timely::dataflow::operators::{ToStream, Exchange, Feedback, Concat, ConnectLoop, vec::{Map, Filter}};
 
 fn main() {
 
@@ -11,13 +11,16 @@ fn main() {
         let peers = worker.peers();
         worker.dataflow::<u64,_,_>(move |scope| {
             let (helper, cycle) = scope.feedback(1);
+            // Each record counts the rounds it has made; all start at zero, and each
+            // makes exactly `iterations` trips around the loop.
             (0 .. elements)
-                  .filter(move |&x| (x as usize) % peers == index)
+                  .filter(move |&i| (i as usize) % peers == index)
+                  .map(|_| 0u64)
                   .to_stream(scope)
                   .concat(cycle)
                   .exchange(|&x| x)
                   .map_in_place(|x| *x += 1)
-                  .branch_when(move |t| t < &iterations).1
+                  .filter(move |&x| x <= iterations)
                   .connect_loop(helper);
         });
     }).unwrap();

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

@@ -22,16 +22,17 @@ pub trait Feedback<'scope, T: Timestamp> {
     /// ```
     /// use timely::dataflow::Scope;
     /// use timely::dataflow::operators::{Feedback, ConnectLoop, ToStream, Concat, Inspect};
-    /// use timely::dataflow::operators::vec::BranchWhen;
+    /// use timely::dataflow::operators::vec::{Map, Filter};
     ///
     /// timely::example(|scope| {
-    ///     // circulate 0..10 for 100 iterations.
+    ///     // circulate 0..10, incrementing each, until each reaches 100.
     ///     let (handle, cycle) = scope.feedback(1);
     ///     (0..10).to_stream(scope)
     ///            .container::<Vec<_>>()
     ///            .concat(cycle)
     ///            .inspect(|x| println!("seen: {:?}", x))
-    ///            .branch_when(|t| t < &100).1
+    ///            .map(|x| x + 1)
+    ///            .filter(|x| *x < 100)
     ///            .connect_loop(handle);
     /// });
     /// ```
@@ -94,16 +95,17 @@ pub trait ConnectLoop<'scope, T: Timestamp, C: Container> {
     /// ```
     /// use timely::dataflow::Scope;
     /// use timely::dataflow::operators::{Feedback, ConnectLoop, ToStream, Concat, Inspect};
-    /// use timely::dataflow::operators::vec::BranchWhen;
+    /// use timely::dataflow::operators::vec::{Map, Filter};
     ///
     /// timely::example(|scope| {
-    ///     // circulate 0..10 for 100 iterations.
+    ///     // circulate 0..10, incrementing each, until each reaches 100.
     ///     let (handle, cycle) = scope.feedback(1);
     ///     (0..10).to_stream(scope)
     ///            .container::<Vec<_>>()
     ///            .concat(cycle)
     ///            .inspect(|x| println!("seen: {:?}", x))
-    ///            .branch_when(|t| t < &100).1
+    ///            .map(|x| x + 1)
+    ///            .filter(|x| *x < 100)
     ///            .connect_loop(handle);
     /// });
     /// ```

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

@@ -14,7 +14,6 @@ pub mod ok_err;
 pub mod partition;
 pub mod probe;
 pub mod rc;
-pub mod reclock;
 pub mod to_stream;
 pub mod unordered_input;
 
@@ -31,5 +30,4 @@ pub use ok_err::OkErr;
 pub use partition::Partition;
 pub use probe::Probe;
 pub use to_stream::{ToStream, ToStreamBuilder};
-pub use reclock::Reclock;
 pub use unordered_input::{UnorderedInput, UnorderedHandle};

timely/src/dataflow/operators/mod.rs

@@ -17,7 +17,6 @@ pub use self::exchange::Exchange;
 pub use self::generic::Operator;
 pub use self::generic::{Notificator, FrontierNotificator};
 
-pub use self::reclock::Reclock;
 
 pub mod core;
 pub mod vec;
@@ -36,7 +35,6 @@ pub use self::core::input::Input;
 
 pub mod generic;
 
-pub use self::core::reclock;
 
 // keep "mint" module-private
 mod capability;