Broadcast channels for remote processes

Author: frankmcsherry

communication/src/allocator/zero_copy/allocator.rs

@@ -163,7 +163,8 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
                 let header = MessageHeader {
                     channel:    identifier,
                     source:     self.index,
-                    target:     target_index,
+                    target_lower:     target_index,
+                    target_upper:     target_index + 1,
                     length:     0,
                     seqno:      0,
                 };
@@ -187,6 +188,50 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
         (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);
+
+        // 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() {
+            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, send.clone())))
+        }
+
+        let channel =
+        self.to_local
+            .entry(identifier)
+            .or_insert_with(|| Rc::new(RefCell::new(VecDeque::new())))
+            .clone();
+
+        use crate::allocator::counters::Puller as CountPuller;
+        let canary = Canary::new(identifier, self.canaries.clone());
+        let puller = Box::new(CountPuller::new(PullerInner::new(inner_recv, channel, canary), identifier, self.events().clone()));
+
+        let pushes = Box::new(crate::allocator::Broadcaster { spare: None, pushers: pushes });
+        (pushes, puller, )
+    }
+
     // Perform preparatory work, most likely reading binary buffers from self.recv.
     #[inline(never)]
     fn receive(&mut self) {

communication/src/allocator/zero_copy/allocator_process.rs

@@ -135,7 +135,8 @@ impl Allocate for ProcessAllocator {
             let header = MessageHeader {
                 channel:    identifier,
                 source:     self.index,
-                target:     target_index,
+                target_lower:     target_index,
+                target_upper:     target_index+1,
                 length:     0,
                 seqno:      0,
             };

communication/src/allocator/zero_copy/tcp.rs

@@ -94,7 +94,9 @@ where
             });
 
             if header.length > 0 {
-                stageds[header.target - worker_offset].push(bytes);
+                for target in header.target_lower .. header.target_upper {
+                    stageds[target - worker_offset].push(bytes.clone());
+                }
             }
             else {
                 // Shutting down; confirm absence of subsequent data.
@@ -196,7 +198,8 @@ pub fn send_loop<S: Stream>(
     let header = MessageHeader {
         channel:    0,
         source:     0,
-        target:     0,
+        target_lower:     0,
+        target_upper:     0,
         length:     0,
         seqno:      0,
     };

communication/src/networking.rs

@@ -29,8 +29,13 @@ pub struct MessageHeader {
     pub channel:    usize,
     /// index of worker sending message.
     pub source:     usize,
-    /// index of worker receiving message.
-    pub target:     usize,
+    /// lower bound of index of worker receiving message.
+    pub target_lower:     usize,
+    /// upper bound of index of worker receiving message.
+    ///
+    /// This is often `self.target_lower + 1` for point to point messages,
+    /// but can be larger for broadcast messages.
+    pub target_upper:     usize,
     /// number of bytes in message.
     pub length:     usize,
     /// sequence number.
@@ -40,7 +45,7 @@ pub struct MessageHeader {
 impl MessageHeader {
 
     /// The number of `usize` fields in [MessageHeader].
-    const FIELDS: usize = 5;
+    const FIELDS: usize = 6;
 
     /// Returns a header when there is enough supporting data
     #[inline]
@@ -52,9 +57,10 @@ impl MessageHeader {
             // Order must match writing order.
             channel: buffer[0] as usize,
             source: buffer[1] as usize,
-            target: buffer[2] as usize,
-            length: buffer[3] as usize,
-            seqno: buffer[4] as usize,
+            target_lower: buffer[2] as usize,
+            target_upper: buffer[3] as usize,
+            length: buffer[4] as usize,
+            seqno: buffer[5] as usize,
         };
 
         if bytes.len() >= header.required_bytes() {
@@ -72,7 +78,8 @@ impl MessageHeader {
         // Order must match reading order.
         cursor.write_u64::<ByteOrder>(self.channel as u64)?;
         cursor.write_u64::<ByteOrder>(self.source as u64)?;
-        cursor.write_u64::<ByteOrder>(self.target as u64)?;
+        cursor.write_u64::<ByteOrder>(self.target_lower as u64)?;
+        cursor.write_u64::<ByteOrder>(self.target_upper as u64)?;
         cursor.write_u64::<ByteOrder>(self.length as u64)?;
         cursor.write_u64::<ByteOrder>(self.seqno as u64)?;