Optimized broadcast #309

examples/network/messaging.rs

@@ -2,6 +2,7 @@
 use crossbeam::thread::{Scope, ScopedJoinHandle};
 use crossbeam_channel::{self, bounded, select, unbounded, Receiver, Sender};
 use hbbft::{SourcedMessage, Target, TargetedMessage};
+use std::collections::BTreeSet;
 
 /// The queue functionality for messages sent between algorithm instances.
 /// The messaging struct allows for targeted message exchange between comms
@@ -107,7 +108,7 @@ impl<M: Send> Messaging<M> {
                 select! {
                     recv(rx_from_algo) -> tm => {
                         if let Ok(tm) = tm {
-                            match tm.target {
+                            match &tm.target {
                                 Target::All => {
                                     // Send the message to all remote nodes, stopping at the first
                                     // error.
@@ -120,9 +121,26 @@ impl<M: Send> Messaging<M> {
                                             }
                                         }).map_err(Error::from);
                                 },
+                                Target::AllExcept(exclude) => {
+                                    // Send the message to all remote nodes not in `exclude`, stopping at the first
+                                    // error.
+                                    let filtered_txs: Vec<_> = (0..txs_to_comms.len())
+                                        .collect::<BTreeSet<_>>()
+                                        .difference(exclude)
+                                        .cloned()
+                                        .collect();
+                                    result = filtered_txs.iter()
+                                        .fold(Ok(()), |result, i| {
+                                            if result.is_ok() {
+                                                txs_to_comms[*i].send(tm.message.clone())
+                                            } else {
+                                                result
+                                            }
+                                        }).map_err(Error::from);
+                                },
                                 Target::Node(i) => {
-                                    result = if i < txs_to_comms.len() {
-                                        txs_to_comms[i].send(tm.message)
+                                    result = if *i < txs_to_comms.len() {
+                                        txs_to_comms[*i].send(tm.message)
                                             .map_err(Error::from)
                                     } else {
                                         Err(Error::NoSuchTarget)

examples/simulation.rs

@@ -273,14 +273,21 @@ where
         Q: IntoIterator<Item = TimestampedMessage<D>>,
     {
         for ts_msg in msgs {
-            match ts_msg.target {
+            match &ts_msg.target {
                 Target::All => {
                     for node in self.nodes.values_mut() {
                         if node.id != ts_msg.sender_id {
                             node.add_message(ts_msg.clone())
                         }
                     }
                 }
+                Target::AllExcept(exclude) => {
+                    for node in self.nodes.values_mut().filter(|n| !exclude.contains(&n.id)) {
+                        if node.id != ts_msg.sender_id {
+                            node.add_message(ts_msg.clone())
+                        }
+                    }
+                }
                 Target::Node(to_id) => {
                     if let Some(node) = self.nodes.get_mut(&to_id) {
                         node.add_message(ts_msg);

hbbft_testing/src/lib.rs

@@ -257,6 +257,22 @@ where
                         message_count = message_count.saturating_add(1);
                     }
 
+                    dest.push_back(NetworkMessage::new(
+                        stepped_id.clone(),
+                        tmsg.message.clone(),
+                        to.clone(),
+                    ));
+                }
+            }
+            hbbft::Target::AllExcept(exclude) => {
+                for to in nodes
+                    .keys()
+                    .filter(|&to| to != &stepped_id && !exclude.contains(to))
+                {
+                    if !faulty {
+                        message_count = message_count.saturating_add(1);
+                    }
+
                     dest.push_back(NetworkMessage::new(
                         stepped_id.clone(),
                         tmsg.message.clone(),