Remove Notificator, FrontierNotificator, unary_notify, and binary_notify

mdbook/src/chapter_2/chapter_2_4.md

@@ -173,70 +173,13 @@ Specifically, each input has a `frontier` method which returns a `&[Timestamp]`,
 
 This frontier information is invaluable for operators that must be sure that their output is correct and final before they send it as output. For our `maximum` example, we will want to wait to apply the new maximum until we are sure that we will not see any more elements at earlier times. That isn't to say we can't do anything with data we receive "early"; in the case of the maximum, each batch at a given time can be reduced down to just its maximum value, as all received values would be applied simultaneously.
 
-To make life easier for you, we've written a helper type called `Notificator` whose job in life is to help you keep track of times that you would like to send outputs, and to tell you when (according to your input frontiers) it is now safe to send the data. In fact, notificators do more by holding on to the *capabilities* for you, so that you can be sure that, even if you *don't* receive any more messages but just an indication that there will be none, you will still retain the ability to send your messages.
-
-Here is a worked example where we use a binary operator that implements the behavior of `concat`, but it puts its inputs in order, buffering its inputs until their associated timestamp is complete, and then sending all data at that time. The operator defines and captures a `HashMap<Time, Vec<Data>>` named `stash` which it uses to buffer received input data that are not yet ready to send.
-
-```rust
-extern crate timely;
-
-use std::collections::HashMap;
-use timely::dataflow::operators::{ToStream, FrontierNotificator};
-use timely::dataflow::operators::generic::operator::Operator;
-use timely::dataflow::channels::pact::Pipeline;
-
-fn main() {
-    timely::example(|scope| {
-
-        let in1 = (0 .. 10).to_stream(scope).container::<Vec<_>>();
-        let in2 = (0 .. 10).to_stream(scope).container::<Vec<_>>();
-
-        in1.binary_frontier(in2, Pipeline, Pipeline, "concat_buffer", |capability, info| {
-
-            let mut notificator = FrontierNotificator::default();
-            let mut stash = HashMap::new();
-
-            move |(input1, frontier1), (input2, frontier2), output| {
-                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_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| {
-                    let mut session = output.session(&time);
-                    if let Some(list) = stash.remove(time.time()) {
-                        for mut vector in list.into_iter() {
-                            session.give_container(&mut vector);
-                        }
-                    }
-                });
-            }
-        });
-    });
-}
-```
-
-As an exercise, this example could be improved in a few ways. How might you change it so that the data are still sent in the order they are received, but messages may be sent as soon as they are received if their time is currently in the frontier? This would avoid buffering messages that are ready to go, and would only buffer messages that are out-of-order, potentially reducing the memory footprint and improving the effective latency.
-
-Before ending the section, let's rewrite this example without the `notificator`, in an attempt to demystify how it works. Whether you use a notificator or not is up to you; they are mostly about staying sane in what can be a confusing setting, and you can totally skip them once you have internalized how capabilities and frontiers work.
+Here is a worked example where we use a binary operator that implements the behavior of `concat`, but it puts its inputs in order, buffering its inputs until their associated timestamp is complete, and then sending all data at that time. The operator defines and captures a `HashMap<Capability, Vec<Data>>` named `stash` which it uses to buffer received input data that are not yet ready to send, keyed by a capability for the time at which they will eventually be sent.
 
 ```rust
 extern crate timely;
 
 use std::collections::HashMap;
-use timely::dataflow::operators::{ToStream, FrontierNotificator};
+use timely::dataflow::operators::ToStream;
 use timely::dataflow::operators::generic::operator::Operator;
 use timely::dataflow::channels::pact::Pipeline;
 
@@ -287,4 +230,6 @@ fn main() {
 }
 ```
 
-Take a moment and check out the differences. Mainly, `stash` is now the one source of truth about `time` and `data`, but we now have to do our own checking of `time` against the input frontiers, and *very importantly* we need to make sure to discard `time` from the `stash` when we are finished with it (otherwise we retain the ability to send at `time`, and the system will not make progress).
+A few things to notice. The `stash` is the one source of truth about times and data: it holds the *capabilities*, which is what allows the operator to send at those times later, even if it receives no more messages but only word that there will be none. We check each capability's time against the input frontiers, and send once neither input can produce data at that time. And *very importantly* we discard entries from the `stash` once we are finished with them: holding a capability holds back the frontier for everyone downstream, and the system will not make progress until it is dropped.
+
+As an exercise, this example could be improved in a few ways. How might you change it so that the data are still sent in the order they are received, but messages may be sent as soon as they are received if their time is currently in the frontier? This would avoid buffering messages that are ready to go, and would only buffer messages that are out-of-order, potentially reducing the memory footprint and improving the effective latency.

timely/examples/barrier.rs

@@ -9,18 +9,22 @@ fn main() {
 
     timely::execute_from_args(std::env::args().skip(2), move |worker| {
 
-        worker.dataflow(move |scope| {
+        worker.dataflow::<usize,_,_>(move |scope| {
             let (handle, stream) = scope.feedback::<Vec<usize>>(1);
-            stream.unary_notify::<CapacityContainerBuilder<_>, _, _>(
+            stream.unary_frontier::<CapacityContainerBuilder<_>, _, _, _>(
                 Pipeline,
                 "Barrier",
-                vec![0],
-                move |_, _, notificator| {
-                    while let Some((cap, _count)) = notificator.next() {
-                        let time = *cap.time() + 1;
-                        if time < iterations {
-                            notificator.notify_at(cap.delayed(&time));
-                        }
+                move |capability, _info| {
+                    // A capability for the current round; advanced once the round is complete.
+                    let mut caps = vec![capability.delayed(&0)];
+                    move |(input, frontier), _output| {
+                        input.for_each_stamp(|_, _| { });
+                        caps = std::mem::take(&mut caps).into_iter().filter_map(|cap| {
+                            if frontier.frontier().less_equal(cap.time()) { Some(cap) } else {
+                                let time = *cap.time() + 1;
+                                if time < iterations { Some(cap.delayed(&time)) } else { None }
+                            }
+                        }).collect();
                     }
                 }
             )

timely/examples/bfs.rs

@@ -1,6 +1,6 @@
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 
-use timely::dataflow::operators::{ToStream, Concat, Feedback, ConnectLoop};
+use timely::dataflow::operators::{ToStream, Concat, Feedback, ConnectLoop, Capability};
 use timely::dataflow::operators::generic::operator::Operator;
 use timely::dataflow::channels::pact::Exchange;
 
@@ -46,35 +46,38 @@ fn main() {
             let (handle, stream) = scope.feedback(1usize);
 
             // use the stream of edges
-            graph.binary_notify(
+            graph.binary_frontier(
                 stream,
                 Exchange::new(|x: &(u32, u32)| u64::from(x.0)),
                 Exchange::new(|x: &(u32, u32)| u64::from(x.0)),
                 "BFS",
-                vec![],
-                move |input1, input2, output, notify| {
+                move |_capability, _info| {
+
+                    // capabilities for times with pending work, in time order.
+                    let mut pending: BTreeMap<usize, Capability<usize>> = BTreeMap::new();
+
+                    move |(input1, frontier1), (input2, frontier2), output| {
 
                     // receive edges, start to sort them
                     input1.for_each_stamp(|cap, data| {
                         if let Some(cap) = cap.retain_least(output.output_index()) {
-                            notify.notify_at(cap);
+                            pending.entry(*cap.time()).or_insert(cap);
                             edge_list.extend(data.map(std::mem::take));
                         }
                     });
 
                     // receive (node, worker) pairs, note any new ones.
                     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));
+                            node_lists.entry(*cap.time()).or_insert_with(Vec::new).extend(data.map(std::mem::take));
+                            pending.entry(*cap.time()).or_insert(cap);
                         }
                     });
 
-                    notify.for_each(|cap, _num, _notify| {
+                    // process each time once neither input can produce data at it, in time order.
+                    while let Some(time) = pending.keys().next().copied() {
+                        if frontier1.frontier().less_equal(&time) || frontier2.frontier().less_equal(&time) { break; }
+                        let cap = pending.remove(&time).unwrap();
 
                         // maybe process the graph
                         if *cap.time() == 0 {
@@ -133,7 +136,8 @@ fn main() {
                                 }
                             }
                         }
-                    });
+                    }
+                    }
                 }
             )
             .concat((0..1).map(|x| (x,x)).to_stream(scope))

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

@@ -5,11 +5,9 @@ pub mod builder_rc;
 pub mod builder_raw;
 // pub mod builder_ref;
 mod handles;
-mod notificator;
 mod operator_info;
 
 pub use self::handles::{InputHandleCore, OutputBuilder, OutputBuilderSession, Session};
-pub use self::notificator::{Notificator, FrontierNotificator};
 
 pub use self::operator::{Operator, source};
 pub use self::operator_info::OperatorInfo;