diff --git a/communication/src/allocator/counters.rs b/communication/src/allocator/counters.rs index 6af216e8f..809829807 100644 --- a/communication/src/allocator/counters.rs +++ b/communication/src/allocator/counters.rs @@ -44,9 +44,13 @@ impl> Push for Pusher { // } // TODO: Version above is less chatty, but can be a bit late in // moving information along. Better, but needs cooperation. - self.events - .borrow_mut() - .push(self.index); + // A `None` is a flush, and the wrapped pusher is unbuffered: nothing + // is enqueued, and so there is nothing to announce. + if element.is_some() { + self.events + .borrow_mut() + .push(self.index); + } self.pusher.push(element) } @@ -91,6 +95,10 @@ impl> Push for ArcPusher { // self.count += 1; // } + // A `None` is a flush, and the wrapped pusher is unbuffered: nothing + // is enqueued, and so there is nothing to announce or to awaken for. + if element.is_none() { return; } + // These three calls should happen in this order, to ensure that // we first enqueue data, second enqueue interest in the channel, // and finally awaken the thread. Other orders are defective when diff --git a/communication/src/allocator/process.rs b/communication/src/allocator/process.rs index 5080d040c..acf43f2c1 100644 --- a/communication/src/allocator/process.rs +++ b/communication/src/allocator/process.rs @@ -8,7 +8,7 @@ use std::time::Duration; use std::collections::{HashMap}; use std::sync::mpsc::{Sender, Receiver}; -use crate::allocator::thread::{ThreadBuilder}; +use crate::allocator::thread::{ThreadBuilder, ThreadPuller}; use crate::allocator::{Allocate, AllocateBuilder, PeerBuilder, Thread}; use crate::{Push, Pull}; use crate::buzzer::Buzzer; @@ -170,16 +170,30 @@ impl Allocate for Process { use crate::allocator::counters::ArcPusher as CountPusher; use crate::allocator::counters::Puller as CountPuller; + // Messages a worker sends to itself take a thread-local queue rather + // than the shared channel, which would cost two cross-thread sends, + // a self-unpark, and two cross-thread receives, all for no reason. + let (local_send, local_recv) = Thread::new_from(identifier, Rc::clone(self.inner.events())); + let mut local_send = Some(local_send); + let sends = sends.into_iter() .zip(self.counters_send.iter()) - .map(|((s,b), sender)| CountPusher::new(s, identifier, sender.clone(), b)) - .map(|s| Box::new(s) as Box>) + .enumerate() + .map(|(target, ((s,b), sender))| { + if target == self.index { + Box::new(local_send.take().expect("self pusher used twice")) as Box> + } + else { + Box::new(CountPusher::new(s, identifier, sender.clone(), b)) as Box> + } + }) .collect::>(); - let recv = Box::new(CountPuller::new(recv, identifier, Rc::clone(self.inner.events()))) as Box>; + let remote = CountPuller::new(recv, identifier, Rc::clone(self.inner.events())); + let puller = Box::new(LocalFirst { local: local_recv, remote }) as Box>; - (sends, recv) + (sends, puller) } fn events(&self) -> &Rc>> { @@ -227,6 +241,20 @@ struct Puller { source: Receiver, } +/// A puller that drains the worker's own pushes before those of other workers. +struct LocalFirst { + local: ThreadPuller, + remote: crate::allocator::counters::Puller>, +} + +impl Pull for LocalFirst { + #[inline] + fn pull(&mut self) -> &mut Option { + let local = self.local.pull(); + if local.is_some() { local } else { self.remote.pull() } + } +} + impl Pull for Puller { #[inline] fn pull(&mut self) -> &mut Option { diff --git a/communication/src/allocator/zero_copy/allocator_process.rs b/communication/src/allocator/zero_copy/allocator_process.rs index 72a1e94b8..7377b00bc 100644 --- a/communication/src/allocator/zero_copy/allocator_process.rs +++ b/communication/src/allocator/zero_copy/allocator_process.rs @@ -10,13 +10,14 @@ use timely_bytes::arc::Bytes; use crate::networking::MessageHeader; use crate::{Allocate, Push, Pull}; -use crate::allocator::{AllocateBuilder, Exchangeable, PeerBuilder}; +use crate::allocator::{AllocateBuilder, Exchangeable, PeerBuilder, Thread}; +use crate::allocator::thread::ThreadPusher; use crate::allocator::canary::Canary; use crate::allocator::zero_copy::bytes_slab::BytesRefill; use crate::allocator::zero_copy::spill::SpillPolicyFn; -use super::bytes_exchange::{BytesPull, SendEndpoint, MergeQueue}; +use super::bytes_exchange::{BytesPush, BytesPull, SendEndpoint, MergeQueue}; -use super::push_pull::{Pusher, Puller}; +use super::push_pull::{Pusher, Puller, PullerInner}; /// Builds an instance of a ProcessAllocator. /// @@ -100,6 +101,7 @@ impl ProcessBuilder { sends, recvs, to_local: HashMap::new(), + refill: self.refill, } } } @@ -130,6 +132,58 @@ pub struct ProcessAllocator { sends: Vec>>>, // sends[x] -> goes to thread x. recvs: Vec, // recvs[x] <- from thread x. to_local: HashMap>>>, // to worker-local typed pullers. + refill: BytesRefill, // for staging buffers allocated after construction. +} + +/// Delivers each pushed `Bytes` to several destinations, sharing the allocation. +/// +/// Each destination receives a clone of the handle, a reference count rather +/// than a copy, through its own endpoint so that ordering and any spill policy +/// toward that destination are unaffected. +struct Fanout { + targets: Vec>>>, +} + +impl BytesPush for Fanout { + fn extend>(&mut self, iterator: I) { + for bytes in iterator { + for target in self.targets.iter() { + target.borrow_mut().push_bytes(bytes.clone()); + } + } + } +} + +/// A pusher that serializes once for all other workers, and hands the element itself to this worker. +struct BroadcastPusher { + local: ThreadPusher, + remote: Option>, +} + +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); + } +} + +impl ProcessAllocator { + /// A thread-local queue for messages to this worker, and the puller that + /// drains it ahead of the bytes other workers send. + /// + /// Used for progress broadcasts, whose messages are small and never worth + /// paging out. Data channels keep the shared byte queue for self-sends, so + /// that a spill policy can apply to them. + fn local_channel(&mut self, identifier: usize) -> (ThreadPusher, Box>) { + let (local_send, local_recv) = Thread::new_from(identifier, Rc::clone(&self.events)); + 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 = PullerInner::new(Box::new(local_recv), channel, canary); + let puller = Box::new(CountPuller::new(puller, identifier, Rc::clone(&self.events))); + (local_send, puller) + } } impl Allocate for ProcessAllocator { @@ -170,6 +224,34 @@ impl Allocate for ProcessAllocator { (pushes, puller) } + fn broadcast(&mut self, identifier: usize) -> (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)))) + }; + + (Box::new(BroadcastPusher { local, remote }), puller) + } + // Perform preparatory work, most likely reading binary buffers from self.recv. #[inline(never)] fn receive(&mut self) { diff --git a/communication/src/allocator/zero_copy/bytes_exchange.rs b/communication/src/allocator/zero_copy/bytes_exchange.rs index bcd0d247c..60d1865de 100644 --- a/communication/src/allocator/zero_copy/bytes_exchange.rs +++ b/communication/src/allocator/zero_copy/bytes_exchange.rs @@ -227,6 +227,14 @@ impl SendEndpoint

{ pub fn publish(&mut self) { self.send_buffer(); } + /// Sends already-formed bytes, after anything staged so far. + /// + /// Staged bytes are sent first, so that the order of messages toward + /// the destination is the order in which they were pushed. + pub fn push_bytes(&mut self, bytes: Bytes) { + self.send_buffer(); + self.send.extend(Some(bytes)); + } } impl Drop for SendEndpoint

{ diff --git a/timely/src/lib.rs b/timely/src/lib.rs index 39625f7e3..8afd4fa31 100644 --- a/timely/src/lib.rs +++ b/timely/src/lib.rs @@ -145,8 +145,11 @@ mod encoding { impl Bytesable for Bincode { fn from_bytes(bytes: Bytes) -> Self { let typed = ::bincode::deserialize(&bytes[..]).expect("bincode::deserialize() failed"); - let typed_size = ::bincode::serialized_size(&typed).expect("bincode::serialized_size() failed") as usize; - assert_eq!(bytes.len(), (typed_size + 7) & !7); + // Measuring the payload again costs a second traversal of it, on every receive. + #[cfg(debug_assertions)] { + let typed_size = ::bincode::serialized_size(&typed).expect("bincode::serialized_size() failed") as usize; + assert_eq!(bytes.len(), (typed_size + 7) & !7); + } Bincode { payload: typed } } diff --git a/timely/src/worker.rs b/timely/src/worker.rs index 2ec4fbeb6..f24caefca 100644 --- a/timely/src/worker.rs +++ b/timely/src/worker.rs @@ -78,15 +78,42 @@ impl FromStr for ProgressMode { } /// Worker configuration. -#[derive(Debug, Default, Clone)] +#[derive(Debug, Clone)] pub struct Config { /// The progress mode to use. pub(crate) progress_mode: ProgressMode, + /// How long an idle worker polls for new events before parking its thread. + /// + /// Parking a thread and waking it again costs several microseconds, which + /// dominates the cost of fine-grained coordination among workers (for + /// example, a barrier per iteration of a loop). A worker with nothing to do + /// first polls its channels for this long, and parks only if nothing arrives. + /// The polling occupies a core, so the duration bounds the CPU an idle worker + /// burns each time it goes idle. + pub(crate) idle_spin: Duration, /// A map from parameter name to typed parameter values. registry: HashMap>, } +impl Default for Config { + fn default() -> Self { + Config { + progress_mode: ProgressMode::default(), + idle_spin: Duration::from_micros(Self::DEFAULT_IDLE_SPIN_MICROS), + registry: HashMap::new(), + } + } +} + impl Config { + /// The default `idle_spin`, in microseconds. + /// + /// Parking and unparking a thread costs a few microseconds on common + /// platforms, so a worker that polls for this long before parking avoids + /// that cost for the short waits typical of tightly coupled workers, at + /// the expense of at most this much CPU each time it goes idle. + pub const DEFAULT_IDLE_SPIN_MICROS: u64 = 10; + /// Installs options into a [getopts::Options] struct that correspond /// to the parameters in the configuration. /// @@ -99,6 +126,7 @@ impl Config { #[cfg(feature = "getopts")] pub fn install_options(opts: &mut getopts::Options) { opts.optopt("", "progress-mode", "progress tracking mode (eager or demand)", "MODE"); + opts.optopt("", "idle-spin", "microseconds an idle worker polls before parking", "MICROS"); } /// Instantiates a configuration based upon the parsed options in `matches`. @@ -113,7 +141,11 @@ impl Config { pub fn from_matches(matches: &getopts::Matches) -> Result { let progress_mode = matches .opt_get_default("progress-mode", ProgressMode::Demand)?; - Ok(Config::default().progress_mode(progress_mode)) + let mut config = Config::default().progress_mode(progress_mode); + if let Some(micros) = matches.opt_get::("idle-spin").map_err(|e| e.to_string())? { + config = config.idle_spin(Duration::from_micros(micros)); + } + Ok(config) } /// Sets the progress mode to `progress_mode`. @@ -122,6 +154,12 @@ impl Config { self } + /// Sets how long an idle worker polls for events before parking. + pub fn idle_spin(mut self, idle_spin: Duration) -> Self { + self.idle_spin = idle_spin; + self + } + /// Sets a typed configuration parameter for the given `key`. /// /// It is recommended to install a single configuration struct using a key @@ -266,41 +304,30 @@ impl Worker { /// ``` pub fn step_or_park(&mut self, duration: Option) -> bool { - { // Process channel events. Activate responders. - let mut allocator = self.allocator.borrow_mut(); - allocator.receive(); - let events = allocator.events(); - let mut borrow = events.borrow_mut(); - let paths = self.paths.borrow(); - borrow.sort_unstable(); - borrow.dedup(); - for channel in borrow.drain(..) { - // Consider tracking whether a channel - // in non-empty, and only activating - // on the basis of non-empty channels. - // TODO: This is a sloppy way to deal - // with channels that may not be alloc'd. - if let Some(path) = paths.get(&channel) { - self.activations - .borrow_mut() - .activate(&path[..]); + // Determine the minimum park duration, where `None` are an absence of a constraint. + let mut delay = self.poll_events(duration); + + // An idle worker polls for a while before parking, as parking and + // unparking a thread costs more than a short wait usually lasts. + if delay != Some(Duration::new(0,0)) { + let budget = match delay { + Some(delay) => std::cmp::min(delay, self.config.idle_spin), + None => self.config.idle_spin, + }; + if budget > Duration::new(0,0) { + let start = Instant::now(); + let mut polls = 0u32; + loop { + std::hint::spin_loop(); + delay = self.poll_events(duration); + if delay == Some(Duration::new(0,0)) { break; } + // Consult the clock only occasionally, as it is not free. + polls = polls.wrapping_add(1); + if polls % 16 == 0 && start.elapsed() >= budget { break; } } } } - // Organize activations. - self.activations - .borrow_mut() - .advance(); - - // Consider parking only if we have no pending events, some dataflows, and a non-zero duration. - let empty_for = self.activations.borrow().empty_for(); - // Determine the minimum park duration, where `None` are an absence of a constraint. - let delay = match (duration, empty_for) { - (Some(x), Some(y)) => Some(std::cmp::min(x,y)), - (x, y) => x.or(y), - }; - if delay != Some(Duration::new(0,0)) { // Log parking and flush log. @@ -346,6 +373,48 @@ impl Worker { !self.dataflows.borrow().is_empty() } + /// Surfaces channel events as activations, and reports how long the worker may idle. + /// + /// Returns the minimum of `duration` and the time until the next scheduled + /// activation, where `None` is an absence of any constraint, and `Some(0)` + /// means there is work to do now. + fn poll_events(&self, duration: Option) -> Option { + + { // Process channel events. Activate responders. + let mut allocator = self.allocator.borrow_mut(); + allocator.receive(); + let events = allocator.events(); + let mut borrow = events.borrow_mut(); + let paths = self.paths.borrow(); + borrow.sort_unstable(); + borrow.dedup(); + for channel in borrow.drain(..) { + // Consider tracking whether a channel + // in non-empty, and only activating + // on the basis of non-empty channels. + // TODO: This is a sloppy way to deal + // with channels that may not be alloc'd. + if let Some(path) = paths.get(&channel) { + self.activations + .borrow_mut() + .activate(&path[..]); + } + } + } + + // Organize activations. + self.activations + .borrow_mut() + .advance(); + + // Consider parking only if we have no pending events, some dataflows, and a non-zero duration. + let empty_for = self.activations.borrow().empty_for(); + match (duration, empty_for) { + (Some(x), Some(y)) => Some(std::cmp::min(x,y)), + (x, y) => x.or(y), + } + } + /// Calls `self.step()` as long as `func` evaluates to `true`. /// /// This method will continually execute even if there is not work