diff --git a/mdbook/src/chapter_2/chapter_2_4.md b/mdbook/src/chapter_2/chapter_2_4.md index 15b0f1018..ff3623b08 100644 --- a/mdbook/src/chapter_2/chapter_2_4.md +++ b/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>` 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::>(); - let in2 = (0 .. 10).to_stream(scope).container::>(); - - 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>` 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. diff --git a/timely/examples/barrier.rs b/timely/examples/barrier.rs index e9d6ec871..49c69b1a7 100644 --- a/timely/examples/barrier.rs +++ b/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::(move |scope| { let (handle, stream) = scope.feedback::>(1); - stream.unary_notify::, _, _>( + stream.unary_frontier::, _, _, _>( 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(); } } ) diff --git a/timely/examples/bfs.rs b/timely/examples/bfs.rs index 8ad7c6060..414553a24 100644 --- a/timely/examples/bfs.rs +++ b/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,18 +46,22 @@ 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> = 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)); } }); @@ -65,16 +69,15 @@ fn main() { // 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)) diff --git a/timely/src/dataflow/operators/generic/mod.rs b/timely/src/dataflow/operators/generic/mod.rs index 59a4c1a33..8188f4272 100644 --- a/timely/src/dataflow/operators/generic/mod.rs +++ b/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; diff --git a/timely/src/dataflow/operators/generic/notificator.rs b/timely/src/dataflow/operators/generic/notificator.rs deleted file mode 100644 index 748afbbf9..000000000 --- a/timely/src/dataflow/operators/generic/notificator.rs +++ /dev/null @@ -1,449 +0,0 @@ -use crate::progress::frontier::{AntichainRef, MutableAntichain}; -use crate::progress::Timestamp; -use crate::dataflow::operators::Capability; - -/// Tracks requests for notification and delivers available notifications. -/// -/// A `Notificator` represents a dynamic set of notifications and a fixed notification frontier. -/// One can interact with one by requesting notification with `notify_at`, and retrieving notifications -/// with `for_each` and `next`. The next notification to be delivered will be the available notification -/// with the least timestamp, with the implication that the notifications will be non-decreasing as long -/// as you do not request notifications at times prior to those that have already been delivered. -/// -/// Notification requests persist across uses of `Notificator`, and it may help to think of `Notificator` -/// as a notification *session*. However, idiomatically it seems you mostly want to restrict your usage -/// to such sessions, which is why this is the main notificator type. -#[derive(Debug)] -pub struct Notificator<'a, T: Timestamp> { - frontiers: &'a [&'a MutableAntichain], - inner: &'a mut FrontierNotificator, -} - -impl<'a, T: Timestamp> Notificator<'a, T> { - /// Allocates a new `Notificator`. - /// - /// This is more commonly accomplished using `input.monotonic(frontiers)`. - pub fn new( - frontiers: &'a [&'a MutableAntichain], - inner: &'a mut FrontierNotificator, - ) -> Self { - - inner.make_available(frontiers); - - Notificator { - frontiers, - inner, - } - } - - /// Reveals the elements in the frontier of the indicated input. - pub fn frontier(&self, input: usize) -> AntichainRef<'_, T> { - self.frontiers[input].frontier() - } - - /// Requests a notification at the time associated with capability `cap`. - /// - /// In order to request a notification at future timestamp, obtain a capability for the new - /// timestamp first, as show in the example. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::ToStream; - /// use timely::dataflow::operators::generic::Operator; - /// use timely::dataflow::channels::pact::Pipeline; - /// - /// timely::example(|scope| { - /// (0..10).to_stream(scope) - /// .container::>() - /// .unary_notify(Pipeline, "example", Some(0), |input, output, notificator| { - /// input.for_each_stamp(|cap, data| { - /// output.session(&cap).give_containers(data); - /// if let Some(time) = cap.least().map(|t| t + 1) { - /// notificator.notify_at(cap.delayed(&time, output.output_index())); - /// } - /// }); - /// notificator.for_each(|cap, count, _| { - /// println!("done with cap: {:?}, requested {} times", cap.time(), count); - /// assert!(*cap.time() == 0 && count == 2 || count == 1); - /// }); - /// }); - /// }); - /// ``` - #[inline] - pub fn notify_at(&mut self, cap: Capability) { - self.inner.notify_at_frontiered(cap, self.frontiers); - } - - /// Repeatedly calls `logic` until exhaustion of the available notifications. - /// - /// `logic` receives a capability for `t`, the timestamp being notified and a `count` - /// representing how many capabilities were requested for that specific timestamp. - #[inline] - pub fn for_each, u64, &mut Notificator)>(&mut self, mut logic: F) { - while let Some((cap, count)) = self.next() { - logic(cap, count, self); - } - } -} - -impl Iterator for Notificator<'_, T> { - type Item = (Capability, u64); - - /// Retrieve the next available notification. - /// - /// Returns `None` if no notification is available. Returns `Some(cap, count)` otherwise: - /// `cap` is a capability for `t`, the timestamp being notified and, `count` represents - /// how many notifications (out of those requested) are being delivered for that specific - /// timestamp. - #[inline] - fn next(&mut self) -> Option<(Capability, u64)> { - self.inner.next_count(self.frontiers) - } -} - -#[test] -fn notificator_delivers_notifications_in_topo_order() { - use std::rc::Rc; - use std::cell::RefCell; - use crate::progress::ChangeBatch; - use crate::progress::frontier::MutableAntichain; - use crate::order::Product; - use crate::dataflow::operators::capability::Capability; - - let mut frontier = MutableAntichain::from_elem(Product::new(0, 0)); - - let root_capability = Capability::new(Product::new(0,0), Rc::new(RefCell::new(ChangeBatch::new()))); - - // notificator.update_frontier_from_cm(&mut vec![ChangeBatch::new_from(ts_from_tuple((0, 0)), 1)]); - let times = [ - Product::new(3, 5), - Product::new(5, 4), - Product::new(1, 2), - Product::new(1, 1), - Product::new(1, 1), - Product::new(5, 4), - Product::new(6, 0), - Product::new(6, 2), - Product::new(5, 8), - ]; - - // create a raw notificator with pending notifications at the times above. - let mut frontier_notificator = FrontierNotificator::from(times.iter().map(|t| root_capability.delayed(t))); - - // the frontier is initially (0,0), and so we should deliver no notifications. - assert!(frontier_notificator.monotonic(&[&frontier]).next().is_none()); - - // advance the frontier to [(5,7), (6,0)], opening up some notifications. - frontier.update_iter(vec![(Product::new(0,0),-1), (Product::new(5,7), 1), (Product::new(6,1), 1)]); - - { - let frontiers = [&frontier]; - let mut notificator = frontier_notificator.monotonic(&frontiers); - - // we should deliver the following available notifications, in this order. - assert_eq!(notificator.next().unwrap().0.time(), &Product::new(1,1)); - assert_eq!(notificator.next().unwrap().0.time(), &Product::new(1,2)); - assert_eq!(notificator.next().unwrap().0.time(), &Product::new(3,5)); - assert_eq!(notificator.next().unwrap().0.time(), &Product::new(5,4)); - assert_eq!(notificator.next().unwrap().0.time(), &Product::new(6,0)); - assert_eq!(notificator.next(), None); - } - - // advance the frontier to [(6,10)] opening up all remaining notifications. - frontier.update_iter(vec![(Product::new(5,7), -1), (Product::new(6,1), -1), (Product::new(6,10), 1)]); - - { - let frontiers = [&frontier]; - let mut notificator = frontier_notificator.monotonic(&frontiers); - - // the first available notification should be (5,8). Note: before (6,0) in the total order, but not - // in the partial order. We don't make the promise that we respect the total order. - assert_eq!(notificator.next().unwrap().0.time(), &Product::new(5, 8)); - - // add a new notification, mid notification session. - notificator.notify_at(root_capability.delayed(&Product::new(5,9))); - - // we expect to see (5,9) before we see (6,2) before we see None. - assert_eq!(notificator.next().unwrap().0.time(), &Product::new(5,9)); - assert_eq!(notificator.next().unwrap().0.time(), &Product::new(6,2)); - assert_eq!(notificator.next(), None); - } -} - -/// Tracks requests for notification and delivers available notifications. -/// -/// `FrontierNotificator` is meant to manage the delivery of requested notifications in the -/// presence of inputs that may have outstanding messages to deliver. -/// The notificator inspects the frontiers, as presented from the outside, for each input. -/// Requested notifications can be served only once there are no frontier elements less-or-equal -/// to them, and there are no other pending notification requests less than them. Each will be -/// less-or-equal to itself, so we want to dodge that corner case. -/// -/// # Examples -/// ``` -/// use std::collections::HashMap; -/// use timely::dataflow::operators::{Input, Inspect, FrontierNotificator}; -/// use timely::dataflow::operators::generic::operator::Operator; -/// use timely::dataflow::channels::pact::Pipeline; -/// -/// timely::execute(timely::Config::thread(), |worker| { -/// let (mut in1, mut in2) = worker.dataflow::(|scope| { -/// let (in1_handle, in1) = scope.new_input::>(); -/// let (in2_handle, in2) = scope.new_input::>(); -/// in1.binary_frontier(in2, Pipeline, Pipeline, "example", |mut _default_cap, _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.flat_map(|d| d.drain(..))); -/// 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.flat_map(|d| d.drain(..))); -/// notificator.notify_at(cap); -/// } -/// }); -/// notificator.for_each(&[frontier1, frontier2], |cap, _| { -/// if let Some(mut vec) = stash.remove(cap.time()) { -/// output.session(&cap).give_iterator(vec.drain(..)); -/// } -/// }); -/// } -/// }) -/// .container::>() -/// .inspect_core(|e| if let Ok((s, x)) = e { println!("{:?} -> {:?}", s, x) }); -/// -/// (in1_handle, in2_handle) -/// }); -/// -/// for i in 1..10 { -/// in1.send(i - 1); -/// in1.advance_to(i); -/// in2.send(i - 1); -/// in2.advance_to(i); -/// } -/// in1.close(); -/// in2.close(); -/// }).unwrap(); -/// ``` -#[derive(Debug)] -pub struct FrontierNotificator { - pending: Vec<(Capability, u64)>, - available: ::std::collections::BinaryHeap>, -} - -impl Default for FrontierNotificator { - fn default() -> Self { - FrontierNotificator { - pending: Vec::new(), - available: ::std::collections::BinaryHeap::new(), - } - } -} - -impl FrontierNotificator { - /// Allocates a new `FrontierNotificator` with initial capabilities. - pub fn from>>(iter: I) -> Self { - FrontierNotificator { - pending: iter.into_iter().map(|x| (x,1)).collect(), - available: ::std::collections::BinaryHeap::new(), - } - } - - /// Requests a notification at the time associated with capability `cap`. Takes ownership of - /// the capability. - /// - /// In order to request a notification at future timestamp, obtain a capability for the new - /// timestamp first, as shown in the example. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, FrontierNotificator}; - /// use timely::dataflow::operators::generic::operator::Operator; - /// use timely::dataflow::channels::pact::Pipeline; - /// - /// timely::example(|scope| { - /// (0..10).to_stream(scope) - /// .container::>() - /// .unary_frontier(Pipeline, "example", |_, _| { - /// let mut notificator = FrontierNotificator::default(); - /// move |(input, frontier), output| { - /// input.for_each_stamp(|cap, data| { - /// output.session(&cap).give_containers(data); - /// if let Some(time) = cap.least().map(|t| t + 1) { - /// notificator.notify_at(cap.delayed(&time, output.output_index())); - /// } - /// }); - /// notificator.for_each(&[frontier], |cap, _| { - /// println!("done with cap: {:?}", cap.time()); - /// }); - /// } - /// }); - /// }); - /// ``` - #[inline] - pub fn notify_at(&mut self, cap: Capability) { - self.pending.push((cap,1)); - } - - /// Requests a notification at the time associated with capability `cap`. - /// - /// The method takes list of frontiers from which it determines if the capability is immediately available. - /// When used with the same frontier as `make_available`, this method can ensure that notifications are - /// non-decreasing. Simply using `notify_at` will only insert new notifications into the list of pending - /// notifications, which are only re-examine with calls to `make_available`. - #[inline] - pub fn notify_at_frontiered<'a>(&mut self, cap: Capability, frontiers: &'a [&'a MutableAntichain]) { - if frontiers.iter().all(|f| !f.less_equal(cap.time())) { - self.available.push(OrderReversed::new(cap, 1)); - } - else { - self.pending.push((cap,1)); - } - } - - /// Enables pending notifications not in advance of any element of `frontiers`. - pub fn make_available<'a>(&mut self, frontiers: &'a [&'a MutableAntichain]) { - - // By invariant, nothing in self.available is greater_equal anything in self.pending. - // It should be safe to append any ordered subset of self.pending to self.available, - // in that the sequence of capabilities in self.available will remain non-decreasing. - - if !self.pending.is_empty() { - - self.pending.sort_unstable_by(|x,y| x.0.time().cmp(y.0.time())); - for i in 0 .. self.pending.len() - 1 { - if self.pending[i].0.time() == self.pending[i+1].0.time() { - self.pending[i+1].1 += self.pending[i].1; - self.pending[i].1 = 0; - } - } - self.pending.retain(|x| x.1 > 0); - - // Move available capabilities rather than cloning them, which would - // tour a spurious increment and decrement through progress tracking. - // `available` is a heap, so the order disruption is harmless. - let mut index = 0; - while index < self.pending.len() { - if frontiers.iter().all(|f| !f.less_equal(&self.pending[index].0)) { - let (capability, count) = self.pending.swap_remove(index); - self.available.push(OrderReversed::new(capability, count)); - } - else { - index += 1; - } - } - } - } - - /// Returns the next available capability with respect to the supplied frontiers, if one exists, - /// and the count of how many instances are found. - /// - /// In the interest of efficiency, this method may yield capabilities in decreasing order, in certain - /// circumstances. If you want to iterate through capabilities with an in-order guarantee, either (i) - /// use `for_each`, or (ii) call `make_available` first. - #[inline] - pub fn next_count<'a>(&mut self, frontiers: &'a [&'a MutableAntichain]) -> Option<(Capability, u64)> { - if self.available.is_empty() { - self.make_available(frontiers); - } - self.available.pop().map(|front| { - let mut count = front.value; - while self.available.peek() == Some(&front) { - count += self.available.pop().unwrap().value; - } - (front.element, count) - }) - } - - /// Returns the next available capability with respect to the supplied frontiers, if one exists. - /// - /// In the interest of efficiency, this method may yield capabilities in decreasing order, in certain - /// circumstances. If you want to iterate through capabilities with an in-order guarantee, either (i) - /// use `for_each`, or (ii) call `make_available` first. - #[inline] - pub fn next<'a>(&mut self, frontiers: &'a [&'a MutableAntichain]) -> Option> { - self.next_count(frontiers).map(|(cap, _)| cap) - } - - /// Repeatedly calls `logic` till exhaustion of the notifications made available by inspecting - /// the frontiers. - /// - /// `logic` receives a capability for `t`, the timestamp being notified. - #[inline] - pub fn for_each<'a, F: FnMut(Capability, &mut FrontierNotificator)>(&mut self, frontiers: &'a [&'a MutableAntichain], mut logic: F) { - self.make_available(frontiers); - while let Some(cap) = self.next(frontiers) { - logic(cap, self); - } - } - - /// Creates a notificator session in which delivered notification will be non-decreasing. - /// - /// This implementation can be emulated with judicious use of `make_available` and `notify_at_frontiered`, - /// in the event that `Notificator` provides too restrictive an interface. - #[inline] - pub fn monotonic<'a>(&'a mut self, frontiers: &'a [&'a MutableAntichain]) -> Notificator<'a, T> { - Notificator::new(frontiers, self) - } - - /// Iterates over pending capabilities and their count. The count represents how often a - /// capability has been requested. - /// - /// To make sure all pending capabilities are above the frontier, use `for_each` or exhaust - /// `next` to consume all available capabilities. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, FrontierNotificator}; - /// use timely::dataflow::operators::generic::operator::Operator; - /// use timely::dataflow::channels::pact::Pipeline; - /// - /// timely::example(|scope| { - /// (0..10).to_stream(scope) - /// .container::>() - /// .unary_frontier(Pipeline, "example", |_, _| { - /// let mut notificator = FrontierNotificator::default(); - /// move |(input, frontier), output| { - /// input.for_each_stamp(|cap, data| { - /// output.session(&cap).give_containers(data); - /// if let Some(time) = cap.least().map(|t| t + 1) { - /// notificator.notify_at(cap.delayed(&time, output.output_index())); - /// assert_eq!(notificator.pending().filter(|t| t.0.time() == &time).count(), 1); - /// } - /// }); - /// notificator.for_each(&[frontier], |cap, _| { - /// println!("done with cap: {:?}", cap.time()); - /// }); - /// } - /// }); - /// }); - /// ``` - pub fn pending(&self) -> ::std::slice::Iter<'_, (Capability, u64)> { - self.pending.iter() - } -} - -#[derive(Debug, PartialEq, Eq)] -struct OrderReversed { - element: Capability, - value: u64, -} - -impl OrderReversed { - fn new(element: Capability, value: u64) -> Self { OrderReversed { element, value} } -} - -impl PartialOrd for OrderReversed { - fn partial_cmp(&self, other: &Self) -> Option<::std::cmp::Ordering> { - Some(self.cmp(other)) - } -} -impl Ord for OrderReversed { - fn cmp(&self, other: &Self) -> ::std::cmp::Ordering { - other.element.time().cmp(self.element.time()) - } -} diff --git a/timely/src/dataflow/operators/generic/operator.rs b/timely/src/dataflow/operators/generic/operator.rs index b3fea38f9..93409c48d 100644 --- a/timely/src/dataflow/operators/generic/operator.rs +++ b/timely/src/dataflow/operators/generic/operator.rs @@ -12,7 +12,6 @@ use crate::dataflow::{Scope, Stream}; use super::builder_rc::OperatorBuilder; use crate::dataflow::operators::generic::OperatorInfo; -use crate::dataflow::operators::generic::notificator::{Notificator, FrontierNotificator}; use crate::{Container, ContainerBuilder}; use crate::container::CapacityContainerBuilder; @@ -25,7 +24,7 @@ pub trait Operator<'scope, T: Timestamp, C1> { /// # Examples /// ``` /// use std::collections::HashMap; - /// use timely::dataflow::operators::{ToStream, FrontierNotificator}; + /// use timely::dataflow::operators::ToStream; /// use timely::dataflow::operators::generic::Operator; /// use timely::dataflow::channels::pact::Pipeline; /// @@ -35,22 +34,24 @@ pub trait Operator<'scope, T: Timestamp, C1> { /// .container::>() /// .unary_frontier(Pipeline, "example", |default_cap, _info| { /// let mut cap = Some(default_cap.delayed(&12)); - /// let mut notificator = FrontierNotificator::default(); /// let mut stash = HashMap::new(); /// move |(input, frontier), output| { /// if let Some(ref c) = cap.take() { /// output.session(&c).give(12); /// } + /// // Stash data under a capability for its least time. /// input.for_each_stamp(|cap, data| { - /// if let Some(t) = cap.least() { - /// stash.entry(t.clone()) + /// if let Some(cap) = cap.retain_least(output.output_index()) { + /// stash.entry(cap) /// .or_insert(Vec::new()) /// .extend(data.flat_map(|d| d.drain(..))); /// } /// }); - /// notificator.for_each(&[frontier], |cap, _not| { - /// if let Some(mut vec) = stash.remove(cap.time()) { - /// output.session(&cap).give_iterator(vec.drain(..)); + /// // Send stashed data whose time the input frontier has passed. + /// stash.retain(|cap, vec| { + /// if frontier.frontier().less_equal(cap.time()) { true } else { + /// output.session(cap).give_iterator(vec.drain(..)); + /// false /// } /// }); /// } @@ -66,40 +67,6 @@ pub trait Operator<'scope, T: Timestamp, C1> { &mut OutputBuilderSession<'_, T, CB>)+'static, P: ParallelizationContract; - /// Creates a new dataflow operator that partitions its input stream by a parallelization strategy `pact`, - /// and repeatedly invokes the closure supplied as `logic`, which can read from the input stream, write to - /// the output stream, and inspect the frontier at the input. - /// - /// # Examples - /// ``` - /// use std::collections::HashMap; - /// use timely::dataflow::operators::{ToStream, FrontierNotificator}; - /// use timely::dataflow::operators::generic::Operator; - /// use timely::dataflow::channels::pact::Pipeline; - /// - /// timely::example(|scope| { - /// (0u64..10) - /// .to_stream(scope) - /// .container::>() - /// .unary_notify(Pipeline, "example", None, move |input, output, notificator| { - /// input.for_each_stamp(|cap, data| { - /// output.session(&cap).give_containers(data); - /// if let Some(cap) = cap.retain_least(output.output_index()) { - /// notificator.notify_at(cap); - /// } - /// }); - /// notificator.for_each(|cap, _cnt, _not| { - /// println!("notified at {:?}", cap); - /// }); - /// }); - /// }); - /// ``` - fn unary_notify, - &mut OutputBuilderSession<'_, T, CB>, - &mut Notificator)+'static, - P: ParallelizationContract> - (self, pact: P, name: &str, init: impl IntoIterator, logic: L) -> Stream<'scope, T, CB::Container>; /// Creates a new dataflow operator that partitions its input stream by a parallelization /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`. @@ -107,7 +74,7 @@ pub trait Operator<'scope, T: Timestamp, C1> { /// /// # Examples /// ``` - /// use timely::dataflow::operators::{ToStream, FrontierNotificator}; + /// use timely::dataflow::operators::ToStream; /// use timely::dataflow::operators::generic::operator::Operator; /// use timely::dataflow::channels::pact::Pipeline; /// use timely::dataflow::Scope; @@ -144,7 +111,7 @@ pub trait Operator<'scope, T: Timestamp, C1> { /// # Examples /// ``` /// use std::collections::HashMap; - /// use timely::dataflow::operators::{Input, Inspect, FrontierNotificator}; + /// use timely::dataflow::operators::{Input, Inspect}; /// use timely::dataflow::operators::generic::operator::Operator; /// use timely::dataflow::channels::pact::Pipeline; /// @@ -153,24 +120,24 @@ pub trait Operator<'scope, T: Timestamp, C1> { /// let (in1_handle, in1) = scope.new_input::>(); /// let (in2_handle, in2) = scope.new_input::>(); /// in1.binary_frontier(in2, Pipeline, Pipeline, "example", |mut _default_cap, _info| { - /// let mut notificator = FrontierNotificator::default(); /// let mut stash = HashMap::new(); /// move |(input1, frontier1), (input2, frontier2), output| { + /// // Stash data from either input under a capability for its least time. /// 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.flat_map(|d| d.drain(..))); - /// notificator.notify_at(cap); + /// stash.entry(cap).or_insert(Vec::new()).extend(data.flat_map(|d| d.drain(..))); /// } /// }); /// 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.flat_map(|d| d.drain(..))); - /// notificator.notify_at(cap); + /// stash.entry(cap).or_insert(Vec::new()).extend(data.flat_map(|d| d.drain(..))); /// } /// }); - /// notificator.for_each(&[frontier1, frontier2], |cap, _not| { - /// if let Some(mut vec) = stash.remove(cap.time()) { - /// output.session(&cap).give_iterator(vec.drain(..)); + /// // Send stashed data whose time both input frontiers have passed. + /// stash.retain(|cap, vec| { + /// if frontier1.frontier().less_equal(cap.time()) || frontier2.frontier().less_equal(cap.time()) { true } else { + /// output.session(cap).give_iterator(vec.drain(..)); + /// false /// } /// }); /// } @@ -200,60 +167,6 @@ pub trait Operator<'scope, T: Timestamp, C1> { P1: ParallelizationContract, P2: ParallelizationContract; - /// Creates a new dataflow operator that partitions its input stream by a parallelization strategy `pact`, - /// and repeatedly invokes the closure supplied as `logic`, which can read from the input streams, write to - /// the output stream, and inspect the frontier at the inputs. - /// - /// # Examples - /// ``` - /// use std::collections::HashMap; - /// use timely::dataflow::operators::{Input, Inspect, FrontierNotificator}; - /// use timely::dataflow::operators::generic::operator::Operator; - /// use timely::dataflow::channels::pact::Pipeline; - /// - /// timely::execute(timely::Config::thread(), |worker| { - /// let (mut in1, mut in2) = worker.dataflow::(|scope| { - /// let (in1_handle, in1) = scope.new_input::>(); - /// let (in2_handle, in2) = scope.new_input::>(); - /// - /// in1.binary_notify(in2, Pipeline, Pipeline, "example", None, move |input1, input2, output, notificator| { - /// input1.for_each_stamp(|cap, data| { - /// output.session(&cap).give_containers(data); - /// if let Some(cap) = cap.retain_least(output.output_index()) { - /// notificator.notify_at(cap); - /// } - /// }); - /// input2.for_each_stamp(|cap, data| { - /// output.session(&cap).give_containers(data); - /// if let Some(cap) = cap.retain_least(output.output_index()) { - /// notificator.notify_at(cap); - /// } - /// }); - /// notificator.for_each(|cap, _cnt, _not| { - /// println!("notified at {:?}", cap); - /// }); - /// }); - /// - /// (in1_handle, in2_handle) - /// }); - /// - /// for i in 1..10 { - /// in1.send(i - 1); - /// in1.advance_to(i); - /// in2.send(i - 1); - /// in2.advance_to(i); - /// } - /// }).unwrap(); - /// ``` - fn binary_notify, - &mut InputHandleCore, - &mut OutputBuilderSession<'_, T, CB>, - &mut Notificator)+'static, - P1: ParallelizationContract, - P2: ParallelizationContract> - (self, other: Stream<'scope, T, C2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator, logic: L) -> Stream<'scope, T, CB::Container>; /// Creates a new dataflow operator that partitions its input streams by a parallelization /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`. @@ -261,7 +174,7 @@ pub trait Operator<'scope, T: Timestamp, C1> { /// /// # Examples /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect, FrontierNotificator}; + /// use timely::dataflow::operators::{ToStream, Inspect}; /// use timely::dataflow::operators::generic::operator::Operator; /// use timely::dataflow::channels::pact::Pipeline; /// use timely::dataflow::Scope; @@ -299,7 +212,7 @@ pub trait Operator<'scope, T: Timestamp, C1> { /// /// # Examples /// ``` - /// use timely::dataflow::operators::{ToStream, FrontierNotificator}; + /// use timely::dataflow::operators::ToStream; /// use timely::dataflow::operators::generic::operator::Operator; /// use timely::dataflow::channels::pact::Pipeline; /// use timely::dataflow::Scope; @@ -353,26 +266,6 @@ impl<'scope, T: Timestamp, C1: Container> Operator<'scope, T, C1> for Stream<'sc stream } - fn unary_notify, - &mut OutputBuilderSession<'_, T, CB>, - &mut Notificator)+'static, - P: ParallelizationContract> - (self, pact: P, name: &str, init: impl IntoIterator, mut logic: L) -> Stream<'scope, T, CB::Container> { - - self.unary_frontier(pact, name, move |capability, _info| { - let mut notificator = FrontierNotificator::default(); - for time in init { - notificator.notify_at(capability.delayed(&time)); - } - - move |(input, frontier), output| { - let frontiers = &[frontier]; - let notificator = &mut Notificator::new(frontiers, &mut notificator); - logic(input, output, notificator); - } - }) - } fn unary(self, pact: P, name: &str, constructor: B) -> Stream<'scope, T, CB::Container> where @@ -432,30 +325,6 @@ impl<'scope, T: Timestamp, C1: Container> Operator<'scope, T, C1> for Stream<'sc stream } - fn binary_notify, - &mut InputHandleCore, - &mut OutputBuilderSession<'_, T, CB>, - &mut Notificator)+'static, - P1: ParallelizationContract, - P2: ParallelizationContract> - (self, other: Stream<'scope, T, C2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator, mut logic: L) -> Stream<'scope, T, CB::Container> { - - self.binary_frontier(other, pact1, pact2, name, |capability, _info| { - let mut notificator = FrontierNotificator::default(); - for time in init { - notificator.notify_at(capability.delayed(&time)); - } - - move |(input1, frontier1), (input2, frontier2), output| { - let frontiers = &[frontier1, frontier2]; - let notificator = &mut Notificator::new(frontiers, &mut notificator); - logic(input1, input2, output, notificator); - } - }) - - } fn binary(self, other: Stream<'scope, T, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> Stream<'scope, T, CB::Container> diff --git a/timely/src/dataflow/operators/mod.rs b/timely/src/dataflow/operators/mod.rs index 99c1fad9c..18d9fc50a 100644 --- a/timely/src/dataflow/operators/mod.rs +++ b/timely/src/dataflow/operators/mod.rs @@ -15,7 +15,6 @@ pub use self::inspect::Inspect; pub use self::exchange::Exchange; pub use self::generic::Operator; -pub use self::generic::{Notificator, FrontierNotificator}; pub mod core; diff --git a/timely/tests/barrier.rs b/timely/tests/barrier.rs index 79491af49..25114c394 100644 --- a/timely/tests/barrier.rs +++ b/timely/tests/barrier.rs @@ -15,22 +15,26 @@ fn barrier_sync_helper(comm_config: ::timely::CommunicationConfig) { worker: WorkerConfig::default(), }; timely::execute(config, move |worker| { - worker.dataflow(move |scope| { + worker.dataflow::(move |scope| { let (handle, stream) = scope.feedback::>(1); - stream.unary_notify::, _, _>( + stream.unary_frontier::, _, _, _>( Pipeline, "Barrier", - vec![0, 1], - move |_, _, notificator| { - let mut count = 0; - while let Some((cap, _count)) = notificator.next() { - count += 1; - let time = *cap.time() + 1; - if time < 100 { - notificator.notify_at(cap.delayed(&time)); - } + move |capability, _info| { + // Capabilities for rounds in flight; each advances once its round is complete. + let mut caps = vec![capability.delayed(&0), capability.delayed(&1)]; + move |(input, frontier), _output| { + input.for_each_stamp(|_, _| { }); + let mut times = std::collections::BTreeSet::new(); + caps = std::mem::take(&mut caps).into_iter().filter_map(|cap| { + if frontier.frontier().less_equal(cap.time()) { Some(cap) } else { + times.insert(*cap.time()); + let time = *cap.time() + 1; + if time < 100 { Some(cap.delayed(&time)) } else { None } + } + }).collect(); + assert!(times.len() <= 1); } - assert!(count <= 1); } ) .connect_loop(handle);