Cheaper coordination among workers: poll before parking, fewer signals, serialize broadcasts once

communication/src/allocator/counters.rs

@@ -44,9 +44,13 @@ impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
         // }
         // 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<T, P: Push<T>> Push<T> for ArcPusher<T, P> {
         //     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

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<dyn Push<T>>)
+             .enumerate()
+             .map(|(target, ((s,b), sender))| {
+                if target == self.index {
+                    Box::new(local_send.take().expect("self pusher used twice")) as Box<dyn Push<T>>
+                }
+                else {
+                    Box::new(CountPusher::new(s, identifier, sender.clone(), b)) as Box<dyn Push<T>>
+                }
+             })
              .collect::<Vec<_>>();
 
-        let recv = Box::new(CountPuller::new(recv, identifier, Rc::clone(self.inner.events()))) as Box<dyn Pull<T>>;
+        let remote = CountPuller::new(recv, identifier, Rc::clone(self.inner.events()));
+        let puller = Box::new(LocalFirst { local: local_recv, remote }) as Box<dyn Pull<T>>;
 
-        (sends, recv)
+        (sends, puller)
     }
 
     fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
@@ -227,6 +241,20 @@ struct Puller<T> {
     source: Receiver<T>,
 }
 
+/// A puller that drains the worker's own pushes before those of other workers.
+struct LocalFirst<T> {
+    local: ThreadPuller<T>,
+    remote: crate::allocator::counters::Puller<T, Puller<T>>,
+}
+
+impl<T> Pull<T> for LocalFirst<T> {
+    #[inline]
+    fn pull(&mut self) -> &mut Option<T> {
+        let local = self.local.pull();
+        if local.is_some() { local } else { self.remote.pull() }
+    }
+}
+
 impl<T> Pull<T> for Puller<T> {
     #[inline]
     fn pull(&mut self) -> &mut Option<T> {

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<Rc<RefCell<SendEndpoint<MergeQueue>>>>, // sends[x] -> goes to thread x.
     recvs:      Vec<MergeQueue>,                            // recvs[x] <- from thread x.
     to_local:   HashMap<usize, Rc<RefCell<VecDeque<Bytes>>>>,          // 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<Rc<RefCell<SendEndpoint<MergeQueue>>>>,
+}
+
+impl BytesPush for Fanout {
+    fn extend<I: IntoIterator<Item=Bytes>>(&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<T: Exchangeable> {
+    local: ThreadPusher<T>,
+    remote: Option<Pusher<T, Fanout>>,
+}
+
+impl<T: Exchangeable> Push<T> for BroadcastPusher<T> {
+    fn push(&mut self, element: &mut Option<T>) {
+        // 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<T: Exchangeable>(&mut self, identifier: usize) -> (ThreadPusher<T>, Box<dyn Pull<T>>) {
+        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<T: Exchangeable + Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
+
+        // 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::<T>(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) {

communication/src/allocator/zero_copy/bytes_exchange.rs

@@ -227,6 +227,14 @@ impl<P: BytesPush> SendEndpoint<P> {
     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<P: BytesPush> Drop for SendEndpoint<P> {

timely/src/lib.rs

@@ -145,8 +145,11 @@ mod encoding {
     impl<T: Data> Bytesable for Bincode<T> {
         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 }
         }