p2p: simplify and make safe API, and thereby fix the possibility for any deadlock to happen when using the API.

src/event_graph/mod.rs

@@ -195,7 +195,7 @@ impl EventGraph {
         //   from the beginning
 
         // Get references to all our peers.
-        let channels = self.p2p.channels().lock().await.clone();
+        let channels = self.p2p.channels().await;
         let mut communicated_peers = channels.len();
         info!(
             target: "event_graph::dag_sync()",
@@ -207,7 +207,9 @@ impl EventGraph {
 
         // Let's first ask all of our peers for their tips and collect them
         // in our hashmap above.
-        for (url, channel) in channels.iter() {
+        for channel in channels.iter() {
+            let url = channel.address();
+
             let tip_rep_sub = match channel.subscribe_msg::<TipRep>().await {
                 Ok(v) => v,
                 Err(e) => {
@@ -294,7 +296,9 @@ impl EventGraph {
             for parent_id in missing_parents.clone().iter() {
                 let mut found_event = false;
 
-                for (url, channel) in channels.iter() {
+                for channel in channels.iter() {
+                    let url = channel.address();
+
                     debug!(
                         target: "event_graph::dag_sync()",
                         "Requesting {} from {}...", parent_id, url,

src/net/channel.rs

@@ -161,6 +161,10 @@ impl Channel {
         Ok(sub)
     }
 
+    pub fn is_stopped(&self) -> bool {
+        self.stopped.load(SeqCst)
+    }
+
     /// Sends a message across a channel. Calls `send_message` that creates
     /// a new payload and sends it over the network transport as a packet.
     /// Returns an error if something goes wrong.

src/net/p2p.rs

@@ -173,63 +173,42 @@ impl P2p {
         self.broadcast_with_exclude(message, &[]).await
     }
 
-    /// Broadcast a message concurrently to all given peers.
-    pub async fn broadcast_to<M: Message>(&self, message: &M, peer_list: &[ChannelPtr]) {
-        let mut futures = FuturesUnordered::new();
-
-        for channel in peer_list {
-            futures.push(channel.send(message).map_err(|e| {
-                (
-                    format!("[P2P] Broadcasting message to {} failed: {}", channel.address(), e),
-                    channel.clone(),
-                )
-            }));
-        }
-
-        if futures.is_empty() {
-            warn!(target: "net::p2p::broadcast()", "[P2P] No connected channels found for broadcast");
-            return
-        }
-
-        while let Some(entry) = futures.next().await {
-            if let Err((e, chan)) = entry {
-                error!(target: "net::p2p::broadcast()", "{}", e);
-                self.remove(chan).await;
-            }
-        }
-    }
-
     /// Broadcasts a message concurrently across active channels, excluding
     /// the ones provided in `exclude_list`.
     pub async fn broadcast_with_exclude<M: Message>(&self, message: &M, exclude_list: &[Url]) {
-        let chans = self.channels.lock().await;
-        let iter = chans.values();
-        let mut futures = FuturesUnordered::new();
-
-        for channel in iter {
+        let mut channels = Vec::new();
+        for channel in self.channels().await {
             if exclude_list.contains(channel.address()) {
                 continue
             }
-
-            futures.push(channel.send(message).map_err(|e| {
-                (
-                    format!("[P2P] Broadcasting message to {} failed: {}", channel.address(), e),
-                    channel.clone(),
-                )
-            }));
+            channels.push(channel);
         }
+        self.broadcast_to(message, &channels).await
+    }
 
-        if futures.is_empty() {
+    /// Broadcast a message concurrently to all given peers.
+    pub async fn broadcast_to<M: Message>(&self, message: &M, channel_list: &[ChannelPtr]) {
+        if channel_list.is_empty() {
             warn!(target: "net::p2p::broadcast()", "[P2P] No connected channels found for broadcast");
             return
         }
 
-        while let Some(entry) = futures.next().await {
-            if let Err((e, chan)) = entry {
-                error!(target: "net::p2p::broadcast()", "{}", e);
-                self.remove(chan).await;
-            }
+        let futures = FuturesUnordered::new();
+
+        for channel in channel_list {
+            futures.push(channel.send(message).map_err(|e| {
+                error!(
+                    target: "net::p2p::broadcast()",
+                    "[P2P] Broadcasting message to {} failed: {}",
+                    channel.address(), e
+                );
+                // If the channel is stopped then it should automatically die
+                // and the session will remove it from p2p.
+                assert!(channel.is_stopped());
+            }));
         }
+
+        let _results: Vec<_> = futures.collect().await;
     }
 
     /// Check whether we're connected to a given address
@@ -260,19 +239,19 @@ impl P2p {
         self.pending.lock().await.remove(addr);
     }
 
-    /// Return reference to connected channels map
-    pub fn channels(&self) -> &ConnectedChannels {
-        &self.channels
+    /// Return all connected channels
+    pub async fn channels(&self) -> Vec<ChannelPtr> {
+        self.channels.lock().await.values().cloned().collect()
     }
 
     /// Retrieve a random connected channel from the
     pub async fn random_channel(&self) -> Option<ChannelPtr> {
-        let channels = self.channels().lock().await;
+        let channels = self.channels.lock().await;
         channels.values().choose(&mut OsRng).cloned()
     }
 
     pub async fn is_connected(&self) -> bool {
-        !self.channels().lock().await.is_empty()
+        !self.channels.lock().await.is_empty()
     }
 
     /// Return an atomic pointer to the set network settings

src/rpc/p2p_method.rs

@@ -28,7 +28,7 @@ use crate::net;
 pub trait HandlerP2p: Sync + Send {
     async fn p2p_get_info(&self, id: u16, _params: JsonValue) -> JsonResult {
         let mut channels = Vec::new();
-        for (url, channel) in self.p2p().channels().lock().await.iter() {
+        for channel in self.p2p().channels().await {
             let session = match channel.session_type_id() {
                 net::session::SESSION_INBOUND => "inbound",
                 net::session::SESSION_OUTBOUND => "outbound",
@@ -37,7 +37,7 @@ pub trait HandlerP2p: Sync + Send {
                 _ => panic!("invalid result from channel.session_type_id()"),
             };
             channels.push(json_map([
-                ("url", JsonStr(url.clone().into())),
+                ("url", JsonStr(channel.address().clone().into())),
                 ("session", json_str(session)),
                 ("id", JsonNum(channel.info.id.into())),
             ]));