net/p2p/broadcast: Broadcast messages to peers concurrently.

src/net/p2p.rs

@@ -1,8 +1,8 @@
-use async_std::sync::{Arc, Mutex};
 use std::fmt;
 
 use async_executor::Executor;
-use futures::{select, try_join, FutureExt};
+use async_std::sync::{Arc, Mutex};
+use futures::{select, stream::FuturesUnordered, try_join, FutureExt, StreamExt, TryFutureExt};
 use fxhash::{FxHashMap, FxHashSet};
 use log::{debug, error, warn};
 use rand::Rng;
@@ -307,33 +307,70 @@ impl P2p {
         self.stop_subscriber.notify(()).await
     }
 
-    /// Broadcasts a message across all channels.
+    /// Broadcasts a message concurrently across all channels.
     pub async fn broadcast<M: Message + Clone>(&self, message: M) -> Result<()> {
-        for channel in self.channels.lock().await.values() {
-            if let Err(e) = channel.send(message.clone()).await {
-                error!(
-                    "P2p::broadcast(): Broadcasting message to {} failed: {}",
+        let chans = self.channels.lock().await;
+        let iter = chans.values();
+        let mut futures = FuturesUnordered::new();
+
+        for channel in iter {
+            futures.push(channel.send(message.clone()).map_err(|e| {
+                format!(
+                    "P2P::broadcast: Broadcasting message to {} failed: {}",
                     channel.address(),
                     e
-                );
+                )
+            }));
+        }
+
+        if futures.is_empty() {
+            error!("P2P::broadcast: No connected channels found");
+            return Ok(())
+        }
+
+        while let Some(entry) = futures.next().await {
+            if let Err(e) = entry {
+                error!("{}", e);
             }
         }
+
         Ok(())
     }
 
-    /// Broadcasts a message across all channels.
-    /// exclude channels provided in exclude_list
+    /// Broadcasts a message concurrently across all channels.
+    /// Excludes channels provided in `exclude_list`.
     pub async fn broadcast_with_exclude<M: Message + Clone>(
         &self,
         message: M,
         exclude_list: &[Url],
     ) -> Result<()> {
-        for channel in self.channels.lock().await.values() {
-            if exclude_list.contains(&channel.address()) {
-                continue
+        let chans = self.channels.lock().await;
+        let iter = chans.values();
+        let mut futures = FuturesUnordered::new();
+
+        for channel in iter {
+            if !exclude_list.contains(&channel.address()) {
+                futures.push(channel.send(message.clone()).map_err(|e| {
+                    format!(
+                        "P2P::broadcast_with_exclude: Broadcasting message to {} failed: {}",
+                        channel.address(),
+                        e
+                    )
+                }));
             }
-            channel.send(message.clone()).await?;
         }
+
+        if futures.is_empty() {
+            error!("P2P::broadcast_with_exclude: No connected channels found");
+            return Ok(())
+        }
+
+        while let Some(entry) = futures.next().await {
+            if let Err(e) = entry {
+                error!("{}", e);
+            }
+        }
+
         Ok(())
     }