Fewer scope schedules from progress tracking

communication/src/allocator/counters.rs

@@ -118,6 +118,8 @@ pub struct Puller<T, P: Pull<T>> {
     events: Rc<RefCell<Vec<usize>>>,
     puller: P,
     phantom: ::std::marker::PhantomData<T>,
+    /// Whether to record an event once the channel has been drained.
+    echo: bool,
 }
 
 impl<T, P: Pull<T>>  Puller<T, P> {
@@ -129,6 +131,7 @@ impl<T, P: Pull<T>>  Puller<T, P> {
             events,
             puller,
             phantom: ::std::marker::PhantomData,
+            echo: true,
         }
     }
 }
@@ -138,9 +141,11 @@ impl<T, P: Pull<T>> Pull<T> for Puller<T, P> {
         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<T, P: Pull<T>> Pull<T> for Puller<T, P> {
 
         result
     }
+    fn quiet(&mut self) {
+        self.echo = false;
+        self.puller.quiet();
+    }
 }

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<T: Exchangeable+Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
+        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<T: Exchangeable+Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
         self.broadcast(identifier)
     }
+    fn broadcast_peers<T: Exchangeable+Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
+        self.broadcast_peers(identifier)
+    }
     fn receive(&mut self) { self.receive(); }
     fn release(&mut self) { self.release(); }
     fn events(&self) -> &Rc<RefCell<Vec<usize>>> { self.events() }

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<T: Exchangeable + Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
+        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<T: Exchangeable + Clone>(&mut self, identifier: usize)
+        -> (Box<dyn Push<T>>, Box<dyn Pull<T>>)
+    {
+        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(),

communication/src/allocator/zero_copy/allocator.rs

@@ -149,6 +149,51 @@ pub struct TcpAllocator {
     to_local:   HashMap<usize, Rc<RefCell<VecDeque<Bytes>>>>,   // 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<T: Exchangeable + Clone>(&mut self, identifier: usize, include_self: bool) -> (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);
+
+        // Result list of boxed pushers.
+        // One entry for each process.
+        let mut pushes = Vec::<Box<dyn Push<T>>>::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<T: Exchangeable + Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
+        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::<Box<dyn Push<T>>>::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<T: Exchangeable + Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
+        self.broadcast_to(identifier, false)
     }
 
     // Perform preparatory work, most likely reading binary buffers from self.recv.

communication/src/allocator/zero_copy/allocator_process.rs

@@ -155,16 +155,18 @@ 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<T: Exchangeable> {
-    local: ThreadPusher<T>,
+    local: Option<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);
+        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<T: Exchangeable>(&mut self, identifier: usize, include_self: bool) -> (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))))
+        };
+
+        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<T: Exchangeable + Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
+        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::<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)
+    fn broadcast_peers<T: Exchangeable + Clone>(&mut self, identifier: usize) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
+        self.broadcast_to(identifier, false)
     }
 
     // Perform preparatory work, most likely reading binary buffers from self.recv.

communication/src/allocator/zero_copy/push_pull.rs

@@ -137,4 +137,5 @@ impl<T: Bytesable> Pull<T> for PullerInner<T> {
             &mut self.current
         }
     }
+    fn quiet(&mut self) { self.inner.quiet() }
 }

communication/src/lib.rs

@@ -160,11 +160,18 @@ pub trait Pull<T> {
     /// Takes an `Option<T>` and leaves `None` behind.
     #[inline]
     fn recv(&mut self) -> Option<T> { 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<T, P: ?Sized + Pull<T>> Pull<T> for Box<P> {
     #[inline]
     fn pull(&mut self) -> &mut Option<T> { (**self).pull() }
+    fn quiet(&mut self) { (**self).quiet() }
 }
 
 

timely/src/progress/broadcast.rs

@@ -35,7 +35,11 @@ impl<T:Timestamp+Send> Progcaster<T> {
     pub fn new(worker: &crate::worker::Worker, addr: Rc<[usize]>, identifier: usize, mut logging: Option<Logger>, progress_logging: Option<ProgressLogger<T>>) -> Progcaster<T> {
 
         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<T:Timestamp+Send> Progcaster<T> {
     }
 
     /// 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<T:Timestamp+Send> Progcaster<T> {
                     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);