diff --git a/communication/src/allocator/counters.rs b/communication/src/allocator/counters.rs index 809829807..afb105695 100644 --- a/communication/src/allocator/counters.rs +++ b/communication/src/allocator/counters.rs @@ -118,6 +118,8 @@ pub struct Puller> { events: Rc>>, puller: P, phantom: ::std::marker::PhantomData, + /// Whether to record an event once the channel has been drained. + echo: bool, } impl> Puller { @@ -129,6 +131,7 @@ impl> Puller { events, puller, phantom: ::std::marker::PhantomData, + echo: true, } } } @@ -138,9 +141,11 @@ impl> Pull for Puller { let result = self.puller.pull(); if result.is_none() { if self.count != 0 { - self.events - .borrow_mut() - .push(self.index); + if self.echo { + self.events + .borrow_mut() + .push(self.index); + } self.count = 0; } } @@ -150,4 +155,8 @@ impl> Pull for Puller { result } + fn quiet(&mut self) { + self.echo = false; + self.puller.quiet(); + } } diff --git a/communication/src/allocator/generic.rs b/communication/src/allocator/generic.rs index a3f68b12e..03506499e 100644 --- a/communication/src/allocator/generic.rs +++ b/communication/src/allocator/generic.rs @@ -56,6 +56,14 @@ impl Allocator { Allocator::Tcp(z) => z.broadcast(identifier), } } + /// Constructs a broadcast channel that does not deliver to the sender. + pub fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + match self { + Allocator::Thread(t) => t.broadcast_peers(identifier), + Allocator::Process(p) => p.broadcast_peers(identifier), + Allocator::Tcp(z) => z.broadcast_peers(identifier), + } + } /// Perform work before scheduling operators. pub fn receive(&mut self) { match self { @@ -111,6 +119,9 @@ impl Allocate for Allocator { fn broadcast(&mut self, identifier: usize) -> (Box>, Box>) { self.broadcast(identifier) } + fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_peers(identifier) + } fn receive(&mut self) { self.receive(); } fn release(&mut self) { self.release(); } fn events(&self) -> &Rc>> { self.events() } diff --git a/communication/src/allocator/mod.rs b/communication/src/allocator/mod.rs index 8c7cf6893..5c47b0ee9 100644 --- a/communication/src/allocator/mod.rs +++ b/communication/src/allocator/mod.rs @@ -92,6 +92,16 @@ pub(crate) trait Allocate { let (pushers, pull) = self.allocate(identifier); (Box::new(Broadcaster { spare: None, pushers }), pull) } + + /// Allocates a broadcast channel, where each pushed message is received by all workers except the sender. + /// + /// The puller still receives the messages that other workers push. + fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + let index = self.index(); + let (mut pushers, pull) = self.allocate(identifier); + pushers.remove(index); + (Box::new(Broadcaster { spare: None, pushers }), pull) + } } /// An adapter to broadcast any pushed element. @@ -208,6 +218,14 @@ impl Process { Process::Bytes(pb) => pb.broadcast(identifier), } } + pub(crate) fn broadcast_peers(&mut self, identifier: usize) + -> (Box>, Box>) + { + match self { + Process::Typed(p) => p.broadcast_peers(identifier), + Process::Bytes(pb) => pb.broadcast_peers(identifier), + } + } pub(crate) fn receive(&mut self) { match self { Process::Typed(p) => p.receive(), diff --git a/communication/src/allocator/zero_copy/allocator.rs b/communication/src/allocator/zero_copy/allocator.rs index 59ce805a1..e3076fea7 100644 --- a/communication/src/allocator/zero_copy/allocator.rs +++ b/communication/src/allocator/zero_copy/allocator.rs @@ -149,6 +149,51 @@ pub struct TcpAllocator { to_local: HashMap>>>, // to worker-local typed pullers. } +impl TcpAllocator { + /// Allocates a broadcast channel that delivers to this worker only if `include_self` is set. + fn broadcast_to(&mut self, identifier: usize, include_self: bool) -> (Box>, Box>) { + + // Assume and enforce in-order identifier allocation. + if let Some(bound) = self.channel_id_bound { + assert!(bound < identifier); + } + self.channel_id_bound = Some(identifier); + + // Result list of boxed pushers. + // One entry for each process. + let mut pushes = Vec::>>::with_capacity(self.sends.len() + 1); + + // Inner exchange allocations. + let inner_peers = self.inner.peers(); + let (inner_send, inner_recv) = if include_self { self.inner.broadcast(identifier) } else { self.inner.broadcast_peers(identifier) }; + + pushes.push(inner_send); + for (mut index, send) in self.sends.iter().enumerate() { + // The span of worker indexes jumps by `inner_peers` as we skip our own process. + // We bump `index` by one as we pass `self.index/inner_peers` to effect this. + if index >= self.index/inner_peers { index += 1; } + let header = MessageHeader { + channel: identifier, + source: self.index, + target_lower: index * inner_peers, + target_upper: index * inner_peers + inner_peers, + length: 0, + seqno: 0, + }; + pushes.push(Box::new(Pusher::new(header, Rc::clone(send)))) + } + + let channel = Rc::clone(self.to_local.entry(identifier).or_default()); + + use crate::allocator::counters::Puller as CountPuller; + let canary = Canary::new(identifier, Rc::clone(&self.canaries)); + let puller = Box::new(CountPuller::new(PullerInner::new(inner_recv, channel, canary), identifier, Rc::clone(self.events()))); + + let pushes = Box::new(crate::allocator::Broadcaster { spare: None, pushers: pushes }); + (pushes, puller, ) + } +} + impl Allocate for TcpAllocator { fn index(&self) -> usize { self.index } fn peers(&self) -> usize { self.peers } @@ -202,45 +247,11 @@ impl Allocate for TcpAllocator { } fn broadcast(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_to(identifier, true) + } - // Assume and enforce in-order identifier allocation. - if let Some(bound) = self.channel_id_bound { - assert!(bound < identifier); - } - self.channel_id_bound = Some(identifier); - - // Result list of boxed pushers. - // One entry for each process. - let mut pushes = Vec::>>::with_capacity(self.sends.len() + 1); - - // Inner exchange allocations. - let inner_peers = self.inner.peers(); - let (inner_send, inner_recv) = self.inner.broadcast(identifier); - - pushes.push(inner_send); - for (mut index, send) in self.sends.iter().enumerate() { - // The span of worker indexes jumps by `inner_peers` as we skip our own process. - // We bump `index` by one as we pass `self.index/inner_peers` to effect this. - if index >= self.index/inner_peers { index += 1; } - let header = MessageHeader { - channel: identifier, - source: self.index, - target_lower: index * inner_peers, - target_upper: index * inner_peers + inner_peers, - length: 0, - seqno: 0, - }; - pushes.push(Box::new(Pusher::new(header, Rc::clone(send)))) - } - - let channel = Rc::clone(self.to_local.entry(identifier).or_default()); - - use crate::allocator::counters::Puller as CountPuller; - let canary = Canary::new(identifier, Rc::clone(&self.canaries)); - let puller = Box::new(CountPuller::new(PullerInner::new(inner_recv, channel, canary), identifier, Rc::clone(self.events()))); - - let pushes = Box::new(crate::allocator::Broadcaster { spare: None, pushers: pushes }); - (pushes, puller, ) + fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_to(identifier, false) } // Perform preparatory work, most likely reading binary buffers from self.recv. diff --git a/communication/src/allocator/zero_copy/allocator_process.rs b/communication/src/allocator/zero_copy/allocator_process.rs index 7377b00bc..132263476 100644 --- a/communication/src/allocator/zero_copy/allocator_process.rs +++ b/communication/src/allocator/zero_copy/allocator_process.rs @@ -155,8 +155,10 @@ impl BytesPush for Fanout { } /// A pusher that serializes once for all other workers, and hands the element itself to this worker. +/// +/// Without a local pusher, the element is left with the caller. struct BroadcastPusher { - local: ThreadPusher, + local: Option>, remote: Option>, } @@ -164,7 +166,7 @@ impl Push for BroadcastPusher { fn push(&mut self, element: &mut Option) { // The serializing pusher reads the element and leaves it in place. if let Some(remote) = self.remote.as_mut() { remote.push(element); } - self.local.push(element); + if let Some(local) = self.local.as_mut() { local.push(element); } } } @@ -184,6 +186,36 @@ impl ProcessAllocator { let puller = Box::new(CountPuller::new(puller, identifier, Rc::clone(&self.events))); (local_send, puller) } + + /// Allocates a broadcast channel that delivers to this worker only if `include_self` is set. + fn broadcast_to(&mut self, identifier: usize, include_self: bool) -> (Box>, Box>) { + + // Assume and enforce in-order identifier allocation. + if let Some(bound) = self.channel_id_bound { + assert!(bound < identifier); + } + self.channel_id_bound = Some(identifier); + + let (local, puller) = self.local_channel::(identifier); + + // Serialize once, and hand every other worker a reference to the bytes. + let targets: Vec<_> = (0 .. self.peers).filter(|&target| target != self.index).map(|target| Rc::clone(&self.sends[target])).collect(); + let remote = if targets.is_empty() { None } else { + let header = MessageHeader { + channel: identifier, + source: self.index, + target_lower: 0, + target_upper: self.peers, + length: 0, + seqno: 0, + }; + let endpoint = SendEndpoint::new(Fanout { targets }, self.refill.clone()); + Some(Pusher::new(header, Rc::new(RefCell::new(endpoint)))) + }; + + let local = if include_self { Some(local) } else { None }; + (Box::new(BroadcastPusher { local, remote }), puller) + } } impl Allocate for ProcessAllocator { @@ -225,31 +257,11 @@ impl Allocate for ProcessAllocator { } fn broadcast(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_to(identifier, true) + } - // Assume and enforce in-order identifier allocation. - if let Some(bound) = self.channel_id_bound { - assert!(bound < identifier); - } - self.channel_id_bound = Some(identifier); - - let (local, puller) = self.local_channel::(identifier); - - // Serialize once, and hand every other worker a reference to the bytes. - let targets: Vec<_> = (0 .. self.peers).filter(|&target| target != self.index).map(|target| Rc::clone(&self.sends[target])).collect(); - let remote = if targets.is_empty() { None } else { - let header = MessageHeader { - channel: identifier, - source: self.index, - target_lower: 0, - target_upper: self.peers, - length: 0, - seqno: 0, - }; - let endpoint = SendEndpoint::new(Fanout { targets }, self.refill.clone()); - Some(Pusher::new(header, Rc::new(RefCell::new(endpoint)))) - }; - - (Box::new(BroadcastPusher { local, remote }), puller) + fn broadcast_peers(&mut self, identifier: usize) -> (Box>, Box>) { + self.broadcast_to(identifier, false) } // Perform preparatory work, most likely reading binary buffers from self.recv. diff --git a/communication/src/allocator/zero_copy/push_pull.rs b/communication/src/allocator/zero_copy/push_pull.rs index 0e7ada2d3..adbf4bf36 100644 --- a/communication/src/allocator/zero_copy/push_pull.rs +++ b/communication/src/allocator/zero_copy/push_pull.rs @@ -137,4 +137,5 @@ impl Pull for PullerInner { &mut self.current } } + fn quiet(&mut self) { self.inner.quiet() } } diff --git a/communication/src/lib.rs b/communication/src/lib.rs index 3daac7b43..d6699d1aa 100644 --- a/communication/src/lib.rs +++ b/communication/src/lib.rs @@ -160,11 +160,18 @@ pub trait Pull { /// Takes an `Option` and leaves `None` behind. #[inline] fn recv(&mut self) -> Option { self.pull().take() } + /// Stops the puller from announcing its channel again once it has been drained. + /// + /// By default, draining a channel that yielded messages records an event for the channel, + /// which schedules its recipient once more. A recipient that finishes its work in the call + /// that drains the channel can opt out. + fn quiet(&mut self) { } } impl> Pull for Box

{ #[inline] fn pull(&mut self) -> &mut Option { (**self).pull() } + fn quiet(&mut self) { (**self).quiet() } } diff --git a/timely/src/progress/broadcast.rs b/timely/src/progress/broadcast.rs index 2c77f223f..c69d1680e 100644 --- a/timely/src/progress/broadcast.rs +++ b/timely/src/progress/broadcast.rs @@ -35,7 +35,11 @@ impl Progcaster { pub fn new(worker: &crate::worker::Worker, addr: Rc<[usize]>, identifier: usize, mut logging: Option, progress_logging: Option>) -> Progcaster { let channel_identifier = worker.new_identifier(); - let (pusher, puller) = worker.broadcast(channel_identifier, addr); + // The channel excludes this worker, and `send` applies our own updates directly. + let (pusher, mut puller) = worker.broadcast_peers(channel_identifier, addr); + // `Subgraph::schedule` applies everything it receives in the call that drains the channel, + // and activates itself if work remains, so a drained channel need not schedule it again. + puller.quiet(); logging.as_mut().map(|l| l.log(crate::logging::CommChannelsEvent { identifier: channel_identifier, kind: crate::logging::CommChannelKind::Progress, @@ -53,7 +57,10 @@ impl Progcaster { } /// Sends pointstamp changes to all workers. - pub fn send(&mut self, changes: &mut ChangeBatch<(Location, T)>) { + /// + /// Other workers receive the changes through the channel, and this worker's copy is added to `local`. + /// Each worker's changes reach every worker in the order sent, which is all that `recv` relies on. + pub fn send(&mut self, changes: &mut ChangeBatch<(Location, T)>, local: &mut ChangeBatch<(Location, T)>) { changes.compact(); if !changes.is_empty() { @@ -83,11 +90,23 @@ impl Progcaster { channel: self.channel_identifier, seq_no: self.counter, identifier: self.identifier, + messages: messages.clone(), + internal: internal.clone(), + }); + // Log the local delivery as a receive, as it would appear had it used the channel. + l.log(crate::logging::TimelyProgressEvent { + is_send: false, + source: self.source, + channel: self.channel_identifier, + seq_no: self.counter, + identifier: self.identifier, messages, internal, }); }); + local.extend(changes.iter().cloned()); + let payload = (self.source, self.counter, std::mem::take(changes)); let mut to_push = Some(Bincode { payload }); self.pusher.push(&mut to_push); diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index d863132fb..e318cbb2b 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -329,16 +329,25 @@ where // Transmit produced progress updates. self.send_progress(); - // If child scopes surface more final pointstamp updates we must re-execute. - if !self.final_pointstamp.is_empty() { + // Apply the updates of child scopes and our own sent updates now, rather than on + // a later activation, so that their effect on our outputs reaches the parent in + // this call. Children whose frontiers change as a result run on the next activation. + // A dataflow root has no parent to report to, and applies the updates on its next + // activation, together with those it receives from other workers by then. + if !self.final_pointstamp.is_empty() && self.path.len() > 1 { + self.propagate_pointstamps(); + } + if !self.temp_active.is_empty() || !self.final_pointstamp.is_empty() { self.activations.borrow_mut().activate(&self.path[..]); } - // A subgraph is incomplete if any child is incomplete, or there are outstanding messages. + // A subgraph is incomplete if any child is incomplete, there are outstanding messages, + // updates remain to be applied, or children have yet to observe changes to their frontiers. let incomplete = self.incomplete_count > 0; let tracking = self.pointstamp_tracker.tracking_anything(); + let pending = !self.temp_active.is_empty() || !self.final_pointstamp.is_empty(); - incomplete || tracking + incomplete || tracking || pending } } @@ -540,7 +549,7 @@ where }; if must_send { - self.progcaster.send(&mut self.local_pointstamp); + self.progcaster.send(&mut self.local_pointstamp, &mut self.final_pointstamp); } } } diff --git a/timely/src/worker.rs b/timely/src/worker.rs index f24caefca..4dfbb95da 100644 --- a/timely/src/worker.rs +++ b/timely/src/worker.rs @@ -609,6 +609,15 @@ impl Worker { self.allocator.borrow_mut().broadcast(identifier) } + /// Allocates a broadcast channel, where each pushed message is received by all workers except the sender. + pub fn broadcast_peers(&self, identifier: usize, address: Rc<[usize]>) -> (Box>, Box>) { + if address.is_empty() { panic!("Unacceptable address: Length zero"); } + let mut paths = self.paths.borrow_mut(); + paths.insert(identifier, address); + self.temp_channel_ids.borrow_mut().push(identifier); + self.allocator.borrow_mut().broadcast_peers(identifier) + } + /// Construct a new dataflow. /// /// # Examples diff --git a/timely/tests/nested_progress.rs b/timely/tests/nested_progress.rs new file mode 100644 index 000000000..5233a385b --- /dev/null +++ b/timely/tests/nested_progress.rs @@ -0,0 +1,126 @@ +//! Progress tracking through nested regions and an iterative scope. +//! +//! Records pass through several levels of regions, each holding them until the input frontier +//! passes their time, and then circulate through a loop for a fixed number of rounds. The test +//! checks that every round completes, that all records arrive exactly once, and that the +//! dataflow shuts down once its input closes. + +use std::cell::RefCell; +use std::collections::BTreeMap; +use std::rc::Rc; + +use timely::dataflow::channels::pact::Exchange; +use timely::dataflow::operators::generic::Operator; +use timely::dataflow::operators::vec::{Filter, Map}; +use timely::dataflow::operators::{CapabilitySet, Concat, ConnectLoop, Enter, Feedback, Input, Inspect, Leave, Probe}; +use timely::dataflow::{InputHandle, ProbeHandle, Stream}; +use timely::order::Product; +use timely::progress::Timestamp; + +/// Holds records until the input frontier passes every time in their stamp. +fn hold<'s, T: Timestamp>(stream: Stream<'s, T, Vec<(u64, u64)>>) -> Stream<'s, T, Vec<(u64, u64)>> { + stream.unary_frontier(Exchange::new(|x: &(u64, u64)| x.0), "Hold", |_cap, _info| { + let mut stash: BTreeMap, (CapabilitySet, Vec<(u64, u64)>)> = BTreeMap::new(); + move |(input, frontier), output| { + input.for_each_stamp(|cap, data| { + let key = cap.stamp().elements().to_vec(); + let entry = stash.entry(key).or_insert_with(|| (cap.retain_stamp(0), Vec::new())); + for d in data { entry.1.extend(d.drain(..)); } + }); + let ready: Vec> = stash.keys().filter(|s| !s.iter().any(|t| frontier.less_equal(t))).cloned().collect(); + for s in ready { + let (caps, data) = stash.remove(&s).unwrap(); + output.session(&caps).give_iterator(data.into_iter()); + } + } + }) +} + +fn nest<'s, T: Timestamp>(stream: Stream<'s, T, Vec<(u64, u64)>>, depth: usize) -> Stream<'s, T, Vec<(u64, u64)>> { + let outer = stream.scope(); + outer.region(|inner| { + let s = hold(stream.enter(inner)); + let s = if depth > 1 { + nest(s, depth - 1) + } + else { + inner.iterative::(|it| { + let (handle, cycle) = it.feedback(Product::new(Default::default(), 1)); + // The low two bits count down the remaining rounds. + let body = hold(s.enter(it).map(|(k, v)| (k, v * 4 + 3)).concat(cycle)); + body.clone().filter(|x| x.1 % 4 != 0).map(|(k, v)| (k, v - 1)).connect_loop(handle); + body.filter(|x| x.1 % 4 == 0).map(|(k, v)| (k, v / 4)).leave(inner) + }) + }; + hold(s).leave(outer) + }) +} + +fn run(config: timely::Config) { + let results = timely::execute(config, |worker| { + let index = worker.index(); + let mut input = InputHandle::new(); + let probe = ProbeHandle::new(); + let seen = Rc::new(RefCell::new(Vec::new())); + let seen2 = Rc::clone(&seen); + worker.dataflow::(|scope| { + nest(scope.input_from(&mut input).container::>(), 4) + .inspect(move |x| seen2.borrow_mut().push(*x)) + .probe_with(&probe); + }); + for round in 0 .. 20u64 { + if index == 0 { + for i in 0 .. 3 { input.send((round * 3 + i, round * 3 + i)); } + } + input.advance_to(round + 1); + let mut steps = 0; + while probe.less_than(input.time()) { + worker.step(); + steps += 1; + assert!(steps < 100_000, "round {round} did not complete"); + } + } + drop(input); + let mut steps = 0; + while worker.step() { + steps += 1; + assert!(steps < 100_000, "dataflow did not shut down"); + } + // The probe must observe the final frontier before the dataflow shuts down. + assert!(probe.done()); + seen.take() + }).unwrap().join(); + let mut all: Vec<(u64, u64)> = results.into_iter().flat_map(|r| r.unwrap()).collect(); + all.sort(); + let expected: Vec<(u64, u64)> = (0 .. 60).map(|x| (x, x)).collect(); + assert_eq!(all, expected); +} + +#[test] +fn nested_progress_thread() { run(timely::Config::thread()); } + +#[test] +fn nested_progress_process() { run(timely::Config::process(4)); } + +/// A worker's own final update must reach operators before the dataflow completes. +fn probe_sees_final_frontier(config: timely::Config) { + timely::execute(config, |worker| { + let (mut input, probe) = worker.dataflow::(|scope| { + let (input, stream) = scope.new_input::>(); + (input, stream.probe().0) + }); + for round in 0 .. 5 { + input.advance_to(round + 1); + worker.step(); + } + drop(input); + while worker.step() { } + assert!(probe.done()); + }).unwrap(); +} + +#[test] +fn probe_sees_final_frontier_thread() { probe_sees_final_frontier(timely::Config::thread()); } + +#[test] +fn probe_sees_final_frontier_process() { probe_sees_final_frontier(timely::Config::process(2)); }