InputCapability::time() requires TotalOrder and returns Option

mdbook/src/chapter_1/chapter_1_2.md

@@ -26,7 +26,7 @@ Let's change the program to print out the timestamp with each record. This shoul
 # }
 ```
 
-with a slightly more complicated operator, `inspect_batch`.
+with a slightly more complicated operator, `inspect_core`.
 
 ```rust
 # extern crate timely;
@@ -36,34 +36,36 @@ with a slightly more complicated operator, `inspect_batch`.
 #     let index = scope.index();
 #     (0..10u64).to_stream(scope)
 #         .container::<Vec<_>>()
-.inspect_batch(move |t,xs| {
-    for x in xs.iter() {
-        println!("worker {}:\thello {} @ {:?}", index, x, t)
+.inspect_core(move |event| {
+    if let Ok((stamp, xs)) = event {
+        for x in xs.iter() {
+            println!("worker {}:\thello {} @ {:?}", index, x, stamp.elements())
+        }
     }
 })
 # ;
 # });
 # }
 ```
 
-The `inspect_batch` operator gets lower-level access to data in timely dataflow, in particular access to batches of records with the same timestamp. It is intended for diagnosing system-level details, but we can also use it to see what timestamps accompany the data.
+The `inspect_core` operator gets lower-level access to data in timely dataflow, in particular access to batches of records that travel together, along with the stamp of timestamps that accompanies them (usually just one timestamp). It also reports changes to the frontier, which we ignore here. It is intended for diagnosing system-level details, but we can also use it to see what timestamps accompany the data.
 
 The output we get with two workers is now:
 
 ```ignore
     Echidnatron% cargo run --example hello -- -w2
         Finished dev [unoptimized + debuginfo] target(s) in 0.0 secs
         Running `target/debug/examples/hello -w2`
-    worker 1:	hello 1 @ 1
-    worker 1:	hello 3 @ 3
-    worker 1:	hello 5 @ 5
-    worker 0:	hello 0 @ 0
-    worker 0:	hello 2 @ 2
-    worker 0:	hello 4 @ 4
-    worker 0:	hello 6 @ 6
-    worker 0:	hello 8 @ 8
-    worker 1:	hello 7 @ 7
-    worker 1:	hello 9 @ 9
+    worker 1:	hello 1 @ [1]
+    worker 1:	hello 3 @ [3]
+    worker 1:	hello 5 @ [5]
+    worker 0:	hello 0 @ [0]
+    worker 0:	hello 2 @ [2]
+    worker 0:	hello 4 @ [4]
+    worker 0:	hello 6 @ [6]
+    worker 0:	hello 8 @ [8]
+    worker 1:	hello 7 @ [7]
+    worker 1:	hello 9 @ [9]
     Echidnatron%
 ```
 

mdbook/src/chapter_2/chapter_2_2.md

@@ -25,7 +25,7 @@ This simple example turns the sequence zero through nine into a stream and then
 
 ## Inspecting Batches
 
-The `inspect` operator has a big sibling, `inspect_batch`, whose closure gets access to whole batches of records at a time, just like the underlying operator. More precisely, `inspect_batch` takes a closure of two parameters: first, the timestamp of a batch, and second a reference to the batch itself. The `inspect_batch` operator can be especially helpful if you want to process the outputs more efficiently.
+The `inspect` operator has a big sibling, `inspect_core`, whose closure gets access to whole batches of records at a time, just like the underlying operator. More precisely, `inspect_core` takes a closure of one parameter, a `Result` that is either `Ok` with the stamp of a batch (the timestamps it travels under, usually just one) and a reference to the batch itself, or `Err` with a new frontier. The `inspect_core` operator can be especially helpful if you want to process the outputs more efficiently.
 
 ```rust
 extern crate timely;
@@ -38,7 +38,7 @@ fn main() {
             (0 .. 10)
                 .to_stream(scope)
                 .container::<Vec<_>>()
-                .inspect_batch(|t, xs| println!("hello: {:?} @ {:?}", xs, t));
+                .inspect_core(|event| if let Ok((stamp, xs)) = event { println!("hello: {:?} @ {:?}", xs, stamp.elements()) });
         });
     }).unwrap();
 }

mdbook/src/chapter_2/chapter_2_4.md

@@ -19,8 +19,8 @@ fn main() {
             .container::<Vec<_>>()
             .unary(Pipeline, "increment", |capability, info| {
                 move |input, output| {
-                    input.for_each_time(|time, data| {
-                        let mut session = output.session(&time);
+                    input.for_each_stamp(|cap, data| {
+                        let mut session = output.session(&cap);
                         for datum in data.flat_map(|d| d.drain(..)) {
                             session.give(datum + 1);
                         }
@@ -38,7 +38,7 @@ Most of what is interesting lies in the closure, so let's first tidy up some loo
 
 The heart of the logic lies in the closure that binds `input` and `output`. These two are handles respectively to the operator's input (from which it can read records) and the operator's output (to which it can send records).
 
-The input handle `input` has one primary method, `for_each_time`, which is invoked for each distinct timestamp with a `CapabilityRef<Timestamp>` and a list of batches of data.
+The input handle `input` has one primary method, `for_each_stamp`, which is invoked for each distinct stamp (the set of timestamps under which a message travels, usually just one) with an `InputCapability<Timestamp>` and a list of batches of data.
 This method should be called each scheduling invocation, if nothing else to move the batches of data from the input to some operator-local storage.
 
 The output handle `output` has one primary method, `session`, which starts up an output session at the indicated time. The resulting session can be given data in various ways: (i) an element at a time with `give`, (ii) an iterator at a time with `give_iterator`, and (iii) a container at a time with `give_container`. Internally it is buffering up the output and flushing automatically when the session goes out of scope, which happens above when we go around the `while` loop.
@@ -145,8 +145,8 @@ fn main() {
                 let mut maximum = 0;    // define this here; use in the closure
 
                 move |input, output| {
-                    input.for_each_time(|time, data| {
-                        let mut session = output.session(&time);
+                    input.for_each_stamp(|cap, data| {
+                        let mut session = output.session(&cap);
                         for datum in data.flat_map(|d| d.drain(..)) {
                             if datum > maximum {
                                 session.give(datum + 1);
@@ -197,17 +197,21 @@ fn main() {
             let mut stash = HashMap::new();
 
             move |(input1, frontier1), (input2, frontier2), output| {
-                input1.for_each_time(|time, data| {
-                    stash.entry(time.time().clone())
-                         .or_insert(Vec::new())
-                         .extend(data.map(std::mem::take));
-                    notificator.notify_at(time.retain(output.output_index()));
+                input1.for_each_stamp(|cap, data| {
+                    if let Some(cap) = cap.retain_least(output.output_index()) {
+                        stash.entry(cap.time().clone())
+                             .or_insert(Vec::new())
+                             .extend(data.map(std::mem::take));
+                        notificator.notify_at(cap);
+                    }
                 });
-                input2.for_each_time(|time, data| {
-                    stash.entry(time.time().clone())
-                         .or_insert(Vec::new())
-                         .extend(data.map(std::mem::take));
-                    notificator.notify_at(time.retain(output.output_index()));
+                input2.for_each_stamp(|cap, data| {
+                    if let Some(cap) = cap.retain_least(output.output_index()) {
+                        stash.entry(cap.time().clone())
+                             .or_insert(Vec::new())
+                             .extend(data.map(std::mem::take));
+                        notificator.notify_at(cap);
+                    }
                 });
 
                 notificator.for_each(&[frontier1, frontier2], |time, notificator| {
@@ -248,15 +252,19 @@ fn main() {
 
             move |(input1, frontier1), (input2, frontier2), output| {
 
-                input1.for_each_time(|time, data| {
-                    stash.entry(time.retain(output.output_index()))
-                         .or_insert(Vec::new())
-                         .extend(data.map(std::mem::take));
+                input1.for_each_stamp(|cap, data| {
+                    if let Some(cap) = cap.retain_least(output.output_index()) {
+                        stash.entry(cap)
+                             .or_insert(Vec::new())
+                             .extend(data.map(std::mem::take));
+                    }
                 });
-                input2.for_each_time(|time, data| {
-                    stash.entry(time.retain(output.output_index()))
-                         .or_insert(Vec::new())
-                         .extend(data.map(std::mem::take));
+                input2.for_each_stamp(|cap, data| {
+                    if let Some(cap) = cap.retain_least(output.output_index()) {
+                        stash.entry(cap)
+                             .or_insert(Vec::new())
+                             .extend(data.map(std::mem::take));
+                    }
                 });
 
                 // consider sending everything in `stash`.

mdbook/src/chapter_2/chapter_2_5.md

@@ -236,10 +236,12 @@ As before, I'm just going to show you the new code, which now lives just after `
                     move |(input, frontier), output| {
 
                         // for each input batch, stash it at `time`.
-                        input.for_each_time(|time, data| {
-                            queues.entry(time.retain(output.output_index()))
-                                  .or_insert(Vec::new())
-                                  .extend(data.flat_map(|d| d.drain(..)));
+                        input.for_each_stamp(|cap, data| {
+                            if let Some(cap) = cap.retain_least(output.output_index()) {
+                                queues.entry(cap)
+                                      .or_insert(Vec::new())
+                                      .extend(data.flat_map(|d| d.drain(..)));
+                            }
                         });
 
                         // enable each stashed time if ready.
@@ -321,14 +323,16 @@ Inside the closure, we do two things: (i) read inputs and (ii) update counts and
 
 ```rust,ignore
         // for each input batch, stash it at `time`.
-        while let Some((time, data)) = input.next() {
-            queues.entry(time.retain(output.output_index()))
-                  .or_insert(Vec::new())
-                  .extend(std::mem::take(data));
-        }
+        input.for_each_stamp(|cap, data| {
+            if let Some(cap) = cap.retain_least(output.output_index()) {
+                queues.entry(cap)
+                      .or_insert(Vec::new())
+                      .extend(data.flat_map(|d| d.drain(..)));
+            }
+        });
 ```
 
-The `input` handle has a `next` method, and it optionally returns a pair of `time` and `data`, representing a timely dataflow timestamp and a hunk of data bearing that timestamp, respectively. Our plan is to iterate through all available input (the `next()` method doesn't block, it just returns `None` when it runs out of data), accepting it from the timely dataflow system and moving it into our `queue` hash map.
+The `input` handle has a `for_each_stamp` method, which calls our closure with a capability `cap` and the batches of `data` received under it, once for each distinct stamp (the set of timestamps under which a message travels, usually just one). Our plan is to accept all available input (the method doesn't block; it just stops when it runs out of data) and move it into our `queue` hash map, keyed by a capability for the least time of the stamp. A message sent under no capabilities has no such time, and we let it go.
 
 Why do we do this? Because this is a streaming system, we could be getting data out of order. Our goal is to update the counts in time order, and to do this we'll need to enqueue what we get until we also get word that the associated `time` is complete. That happens in the next few hunks of code
 

mdbook/src/chapter_4/chapter_4_1.md

@@ -58,7 +58,7 @@ fn main() {
         // Create a new scope with the same (u64) timestamp.
         let result = scope.scoped::<u64,_,_>("SubScope", |subscope| {
             stream.enter(subscope)
-                  .inspect_batch(|t, xs| println!("{:?}, {:?}", t, xs))
+                  .inspect_core(|event| if let Ok((stamp, xs)) = event { println!("{:?}, {:?}", stamp.elements(), xs) })
                   .leave(scope)
         });
 
@@ -90,7 +90,7 @@ fn main() {
         // Create a new scope with the same (u64) timestamp.
         let result = scope.region(|subscope| {
             stream.enter(subscope)
-                  .inspect_batch(|t, xs| println!("{:?}, {:?}", t, xs))
+                  .inspect_core(|event| if let Ok((stamp, xs)) = event { println!("{:?}, {:?}", stamp.elements(), xs) })
                   .leave(scope)
         });
 
@@ -122,7 +122,7 @@ fn main() {
         // Create a new scope with a (u64, u32) timestamp.
         let result = scope.iterative::<u32,_,_>(|subscope| {
             stream.enter(subscope)
-                  .inspect_batch(|t, xs| println!("{:?}, {:?}", t, xs))
+                  .inspect_core(|event| if let Ok((stamp, xs)) = event { println!("{:?}, {:?}", stamp.elements(), xs) })
                   .leave(scope)
         });
 

mdbook/src/chapter_4/chapter_4_3.md

@@ -82,10 +82,12 @@ fn main() {
                 move |(input1, frontier1), (input2, frontier2), output| {
 
                     // Stash received data.
-                    input1.for_each_time(|time, data| {
-                        stash.entry(time.retain(output.output_index()))
-                             .or_insert(Vec::new())
-                             .extend(data.flat_map(|d| d.drain(..)));
+                    input1.for_each_stamp(|cap, data| {
+                        if let Some(cap) = cap.retain_least(output.output_index()) {
+                            stash.entry(cap)
+                                 .or_insert(Vec::new())
+                                 .extend(data.flat_map(|d| d.drain(..)));
+                        }
                     });
 
                     // Consider sending stashed data.

timely/examples/bfs.rs

@@ -55,25 +55,29 @@ fn main() {
                 move |input1, input2, output, notify| {
 
                     // receive edges, start to sort them
-                    input1.for_each_time(|time, data| {
-                        notify.notify_at(time.retain(output.output_index()));
-                        edge_list.extend(data.map(std::mem::take));
+                    input1.for_each_stamp(|cap, data| {
+                        if let Some(cap) = cap.retain_least(output.output_index()) {
+                            notify.notify_at(cap);
+                            edge_list.extend(data.map(std::mem::take));
+                        }
                     });
 
                     // receive (node, worker) pairs, note any new ones.
-                    input2.for_each_time(|time, data| {
-                        node_lists.entry(*time.time())
-                                  .or_insert_with(|| {
-                                      notify.notify_at(time.retain(output.output_index()));
-                                      Vec::new()
-                                  })
-                                  .extend(data.map(std::mem::take));
+                    input2.for_each_stamp(|cap, data| {
+                        if let Some(cap) = cap.retain_least(output.output_index()) {
+                            node_lists.entry(*cap.time())
+                                      .or_insert_with(|| {
+                                          notify.notify_at(cap);
+                                          Vec::new()
+                                      })
+                                      .extend(data.map(std::mem::take));
+                        }
                     });
 
-                    notify.for_each(|time, _num, _notify| {
+                    notify.for_each(|cap, _num, _notify| {
 
                         // maybe process the graph
-                        if *time == 0 {
+                        if *cap.time() == 0 {
 
                             // print some diagnostic timing information
                             if index == 0 { println!("{:?}:\tsorting", start.elapsed()); }
@@ -107,10 +111,10 @@ fn main() {
                         }
 
                         // print some diagnostic timing information
-                        if index == 0 { println!("{:?}:\ttime: {:?}", start.elapsed(), time.time()); }
+                        if index == 0 { println!("{:?}:\ttime: {:?}", start.elapsed(), cap.time()); }
 
-                        if let Some(mut todo) = node_lists.remove(&time) {
-                            let mut session = output.session(&time);
+                        if let Some(mut todo) = node_lists.remove(cap.time()) {
+                            let mut session = output.session(&cap);
 
                             // we could sort these, or not (previously: radix sorted).
                             // todo.sort();

timely/examples/columnar.rs

@@ -7,7 +7,7 @@ use timely::Accountable;
 use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
 use timely::dataflow::InputHandle;
-use timely::dataflow::operators::{InspectCore, Operator, Probe};
+use timely::dataflow::operators::{Inspect, Operator, Probe};
 use timely::dataflow::ProbeHandle;
 
 // Creates `WordCountContainer` and `WordCountReference` structs,
@@ -44,8 +44,8 @@ fn main() {
                     "Split",
                     |_cap, _info| {
                         move |input, output| {
-                            input.for_each_time(|time, data| {
-                                let mut session = output.session(&time);
+                            input.for_each_stamp(|cap, data| {
+                                let mut session = output.session(&cap);
                                 for data in data {
                                     for wordcount in data.borrow().into_index_iter().flat_map(|wordcount| {
                                         wordcount.text.split(|b| b.is_ascii_whitespace()).filter(|s| !s.is_empty()).map(move |text| WordCountReference { text, diff: wordcount.diff })
@@ -66,11 +66,13 @@ fn main() {
                         let mut counts = HashMap::new();
 
                         move |(input, frontier), output| {
-                            input.for_each_time(|time, data| {
-                                queues
-                                    .entry(time.retain(output.output_index()))
-                                    .or_insert(Vec::new())
-                                    .extend(data.map(std::mem::take));
+                            input.for_each_stamp(|cap, data| {
+                                if let Some(cap) = cap.retain_least(output.output_index()) {
+                                    queues
+                                        .entry(cap)
+                                        .or_insert(Vec::new())
+                                        .extend(data.map(std::mem::take));
+                                }
 
                             });
 
@@ -99,7 +101,7 @@ fn main() {
                     },
                 )
                 .container::<Container>()
-                .inspect_container(|x| {
+                .inspect_core(|x| {
                     match x {
                         Ok((time, data)) => {
                             println!("seen at: {:?}\t{:?} records", time, data.record_count());