Merge pull request #356 from mintlayer/p2p-unbounded-channels

p2p: Start using `tokio::unbounded_channel`

Author: altonen

p2p/src/constants.rs

@@ -27,6 +27,3 @@ pub const PING_MAX_RETRIES: u32 = 3;
 
 /// Maximum message size
 pub const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024;
-
-// TODO: think about channel sizes
-pub const CHANNEL_SIZE: usize = 64;

p2p/src/lib.rs

@@ -38,9 +38,6 @@ pub mod sync;
 /// Result type with P2P errors
 pub type Result<T> = core::result::Result<T, P2pError>;
 
-// TODO: figure out proper channel sizes
-const CHANNEL_SIZE: usize = 64;
-
 pub struct P2pInterface<T: NetworkingService> {
     p2p: P2P<T>,
 }
@@ -65,7 +62,6 @@ where
                 })?,
                 tx,
             ))
-            .await
             .map_err(|_| P2pError::ChannelClosed)?;
         rx.await.map_err(P2pError::from)?
     }
@@ -83,7 +79,6 @@ where
         self.p2p
             .tx_swarm
             .send(event::SwarmEvent::Disconnect(peer_id, tx))
-            .await
             .map_err(|_| P2pError::ChannelClosed)?;
         rx.await.map_err(P2pError::from)?
     }
@@ -93,7 +88,6 @@ where
         self.p2p
             .tx_swarm
             .send(event::SwarmEvent::GetPeerCount(tx))
-            .await
             .map_err(P2pError::from)?;
         rx.await.map_err(P2pError::from)
     }
@@ -103,7 +97,6 @@ where
         self.p2p
             .tx_swarm
             .send(event::SwarmEvent::GetBindAddress(tx))
-            .await
             .map_err(P2pError::from)?;
         rx.await.map_err(P2pError::from)
     }
@@ -113,7 +106,6 @@ where
         self.p2p
             .tx_swarm
             .send(event::SwarmEvent::GetPeerId(tx))
-            .await
             .map_err(P2pError::from)?;
         rx.await.map_err(P2pError::from)
     }
@@ -123,7 +115,6 @@ where
         self.p2p
             .tx_swarm
             .send(event::SwarmEvent::GetConnectedPeers(tx))
-            .await
             .map_err(P2pError::from)?;
         rx.await.map_err(P2pError::from)
     }
@@ -132,10 +123,10 @@ where
 struct P2P<T: NetworkingService> {
     // TODO: add abstration for channels
     /// TX channel for sending swarm control events
-    pub tx_swarm: mpsc::Sender<event::SwarmEvent<T>>,
+    pub tx_swarm: mpsc::UnboundedSender<event::SwarmEvent<T>>,
 
     /// TX channel for sending syncing/pubsub events
-    pub _tx_sync: mpsc::Sender<event::SyncEvent>,
+    pub _tx_sync: mpsc::UnboundedSender<event::SyncEvent>,
 }
 
 impl<T> P2P<T>
@@ -177,10 +168,10 @@ where
         //
         // The difference between these types is that enums that contain the events *can* have
         // a `oneshot::channel` object that must be used to send the response.
-        let (tx_swarm, rx_swarm) = mpsc::channel(CHANNEL_SIZE);
-        let (tx_p2p_sync, rx_p2p_sync) = mpsc::channel(CHANNEL_SIZE);
-        let (_tx_sync, _rx_sync) = mpsc::channel(CHANNEL_SIZE);
-        let (tx_pubsub, rx_pubsub) = mpsc::channel(CHANNEL_SIZE);
+        let (tx_swarm, rx_swarm) = mpsc::unbounded_channel();
+        let (tx_p2p_sync, rx_p2p_sync) = mpsc::unbounded_channel();
+        let (_tx_sync, _rx_sync) = mpsc::unbounded_channel();
+        let (tx_pubsub, rx_pubsub) = mpsc::unbounded_channel();
 
         {
             let chain_config = Arc::clone(&chain_config);

p2p/src/net/libp2p/backend.rs

@@ -42,16 +42,16 @@ pub struct Libp2pBackend {
     pub(super) swarm: Swarm<behaviour::Libp2pBehaviour>,
 
     /// Receiver for incoming commands
-    cmd_rx: mpsc::Receiver<types::Command>,
+    cmd_rx: mpsc::UnboundedReceiver<types::Command>,
 
     /// Sender for outgoing connectivity events
-    pub(super) conn_tx: mpsc::Sender<types::ConnectivityEvent>,
+    pub(super) conn_tx: mpsc::UnboundedSender<types::ConnectivityEvent>,
 
     /// Sender for outgoing gossipsub events
-    pub(super) gossip_tx: mpsc::Sender<types::PubSubEvent>,
+    pub(super) gossip_tx: mpsc::UnboundedSender<types::PubSubEvent>,
 
     /// Sender for outgoing syncing events
-    pub(super) sync_tx: mpsc::Sender<types::SyncingEvent>,
+    pub(super) sync_tx: mpsc::UnboundedSender<types::SyncingEvent>,
 
     /// Active listen address of the backend
     // TODO: cache this inside `Libp2pConnectivityHandle`?
@@ -61,10 +61,10 @@ pub struct Libp2pBackend {
 impl Libp2pBackend {
     pub fn new(
         swarm: Swarm<behaviour::Libp2pBehaviour>,
-        cmd_rx: mpsc::Receiver<types::Command>,
-        conn_tx: mpsc::Sender<types::ConnectivityEvent>,
-        gossip_tx: mpsc::Sender<types::PubSubEvent>,
-        sync_tx: mpsc::Sender<types::SyncingEvent>,
+        cmd_rx: mpsc::UnboundedReceiver<types::Command>,
+        conn_tx: mpsc::UnboundedSender<types::ConnectivityEvent>,
+        gossip_tx: mpsc::UnboundedSender<types::PubSubEvent>,
+        sync_tx: mpsc::UnboundedSender<types::SyncingEvent>,
     ) -> Self {
         Self {
             swarm,
@@ -90,13 +90,13 @@ impl Libp2pBackend {
                         self.swarm.behaviour_mut().connmgr.handle_banned_peer(peer_id);
                     }
                     SwarmEvent::Behaviour(Libp2pBehaviourEvent::Connectivity(event)) => {
-                        self.conn_tx.send(event).await.map_err(P2pError::from)?;
+                        self.conn_tx.send(event).map_err(P2pError::from)?;
                     }
                     SwarmEvent::Behaviour(Libp2pBehaviourEvent::Syncing(event)) => {
-                        self.sync_tx.send(event).await.map_err(P2pError::from)?;
+                        self.sync_tx.send(event).map_err(P2pError::from)?;
                     }
                     SwarmEvent::Behaviour(Libp2pBehaviourEvent::PubSub(event)) => {
-                        self.gossip_tx.send(event).await.map_err(P2pError::from)?;
+                        self.gossip_tx.send(event).map_err(P2pError::from)?;
                     }
                     SwarmEvent::Behaviour(Libp2pBehaviourEvent::Control(
                         ControlEvent::CloseConnection { peer_id })
@@ -408,21 +408,19 @@ mod tests {
     #[tokio::test]
     async fn test_command_listen_success() {
         let swarm = make_swarm().await;
-        let (cmd_tx, cmd_rx) = mpsc::channel(16);
-        let (gossip_tx, _) = mpsc::channel(64);
-        let (conn_tx, _) = mpsc::channel(64);
-        let (sync_tx, _) = mpsc::channel(64);
+        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
+        let (gossip_tx, _) = mpsc::unbounded_channel();
+        let (conn_tx, _) = mpsc::unbounded_channel();
+        let (sync_tx, _) = mpsc::unbounded_channel();
         let mut backend = Libp2pBackend::new(swarm, cmd_rx, conn_tx, gossip_tx, sync_tx);
 
         tokio::spawn(async move { backend.run().await });
 
         let (tx, rx) = oneshot::channel();
-        let res = cmd_tx
-            .send(types::Command::Listen {
-                addr: p2p_test_utils::make_libp2p_addr(),
-                response: tx,
-            })
-            .await;
+        let res = cmd_tx.send(types::Command::Listen {
+            addr: p2p_test_utils::make_libp2p_addr(),
+            response: tx,
+        });
         assert!(res.is_ok());
 
         let res = rx.await;
@@ -435,21 +433,19 @@ mod tests {
     #[tokio::test]
     async fn test_command_listen_addrinuse() {
         let swarm = make_swarm().await;
-        let (cmd_tx, cmd_rx) = mpsc::channel(16);
-        let (gossip_tx, _) = mpsc::channel(64);
-        let (conn_tx, _) = mpsc::channel(64);
-        let (sync_tx, _) = mpsc::channel(64);
+        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
+        let (gossip_tx, _) = mpsc::unbounded_channel();
+        let (conn_tx, _) = mpsc::unbounded_channel();
+        let (sync_tx, _) = mpsc::unbounded_channel();
         let mut backend = Libp2pBackend::new(swarm, cmd_rx, conn_tx, gossip_tx, sync_tx);
 
         tokio::spawn(async move { backend.run().await });
 
         let (tx, rx) = oneshot::channel();
-        let res = cmd_tx
-            .send(types::Command::Listen {
-                addr: p2p_test_utils::make_libp2p_addr(),
-                response: tx,
-            })
-            .await;
+        let res = cmd_tx.send(types::Command::Listen {
+            addr: p2p_test_utils::make_libp2p_addr(),
+            response: tx,
+        });
         assert!(res.is_ok());
 
         let res = rx.await;
@@ -458,12 +454,10 @@ mod tests {
 
         // try to bind to the same interface again
         let (tx, rx) = oneshot::channel();
-        let res = cmd_tx
-            .send(types::Command::Listen {
-                addr: p2p_test_utils::make_libp2p_addr(),
-                response: tx,
-            })
-            .await;
+        let res = cmd_tx.send(types::Command::Listen {
+            addr: p2p_test_utils::make_libp2p_addr(),
+            response: tx,
+        });
         assert!(res.is_ok());
 
         let res = rx.await;
@@ -476,10 +470,10 @@ mod tests {
     #[tokio::test]
     async fn test_drop_command_tx() {
         let swarm = make_swarm().await;
-        let (cmd_tx, cmd_rx) = mpsc::channel(16);
-        let (gossip_tx, _) = mpsc::channel(64);
-        let (conn_tx, _) = mpsc::channel(64);
-        let (sync_tx, _) = mpsc::channel(64);
+        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
+        let (gossip_tx, _) = mpsc::unbounded_channel();
+        let (conn_tx, _) = mpsc::unbounded_channel();
+        let (sync_tx, _) = mpsc::unbounded_channel();
         let mut backend = Libp2pBackend::new(swarm, cmd_rx, conn_tx, gossip_tx, sync_tx);
 
         drop(cmd_tx);

p2p/src/net/libp2p/constants.rs

@@ -33,5 +33,3 @@ pub const PING_MAX_RETRIES: u32 = 3;
 /// Request-response configuration
 pub const REQ_RESP_TIMEOUT: Duration = Duration::from_secs(10);
 pub const MESSAGE_MAX_SIZE: usize = 10 * 1024 * 1024;
-// TODO: think about channel sizes
-pub const CHANNEL_SIZE: usize = 64;

p2p/src/net/libp2p/mod.rs

@@ -103,11 +103,10 @@ impl NetworkingService for Libp2pService {
         )
         .build();
 
-        // TODO: unbounded
-        let (cmd_tx, cmd_rx) = mpsc::channel(constants::CHANNEL_SIZE);
-        let (gossip_tx, gossip_rx) = mpsc::channel(constants::CHANNEL_SIZE);
-        let (conn_tx, conn_rx) = mpsc::channel(constants::CHANNEL_SIZE);
-        let (sync_tx, sync_rx) = mpsc::channel(constants::CHANNEL_SIZE);
+        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
+        let (gossip_tx, gossip_rx) = mpsc::unbounded_channel();
+        let (conn_tx, conn_rx) = mpsc::unbounded_channel();
+        let (sync_tx, sync_rx) = mpsc::unbounded_channel();
 
         // run the libp2p backend in a background task
         tokio::spawn(async move {
@@ -119,12 +118,10 @@ impl NetworkingService for Libp2pService {
         // send listen command to the libp2p backend and if it succeeds,
         // create a multiaddress for local peer and return the Libp2pService object
         let (tx, rx) = oneshot::channel();
-        cmd_tx
-            .send(types::Command::Listen {
-                addr: bind_addr.clone(),
-                response: tx,
-            })
-            .await?;
+        cmd_tx.send(types::Command::Listen {
+            addr: bind_addr.clone(),
+            response: tx,
+        })?;
         rx.await?
             .map_err(|_| P2pError::DialError(DialError::IoError(std::io::ErrorKind::AddrInUse)))?;
 

p2p/src/net/libp2p/service/connectivity.rs

@@ -31,18 +31,18 @@ pub struct Libp2pConnectivityHandle<T: NetworkingService> {
     peer_id: PeerId,
 
     /// Channel for sending commands to libp2p backend
-    cmd_tx: mpsc::Sender<types::Command>,
+    cmd_tx: mpsc::UnboundedSender<types::Command>,
 
     /// Channel for receiving connectivity events from libp2p backend
-    conn_rx: mpsc::Receiver<types::ConnectivityEvent>,
+    conn_rx: mpsc::UnboundedReceiver<types::ConnectivityEvent>,
     _marker: std::marker::PhantomData<fn() -> T>,
 }
 
 impl<T: NetworkingService> Libp2pConnectivityHandle<T> {
     pub fn new(
         peer_id: PeerId,
-        cmd_tx: mpsc::Sender<types::Command>,
-        conn_rx: mpsc::Receiver<types::ConnectivityEvent>,
+        cmd_tx: mpsc::UnboundedSender<types::Command>,
+        conn_rx: mpsc::UnboundedReceiver<types::ConnectivityEvent>,
     ) -> Self {
         Self {
             peer_id,
@@ -177,13 +177,11 @@ where
 
         // try to connect to remote peer
         let (tx, rx) = oneshot::channel();
-        self.cmd_tx
-            .send(types::Command::Connect {
-                peer_id,
-                peer_addr: addr.clone(),
-                response: tx,
-            })
-            .await?;
+        self.cmd_tx.send(types::Command::Connect {
+            peer_id,
+            peer_addr: addr.clone(),
+            response: tx,
+        })?;
 
         rx.await.map_err(P2pError::from)?.map_err(P2pError::from)
     }
@@ -192,18 +190,16 @@ where
         log::debug!("disconnect peer {:?}", peer_id);
 
         let (tx, rx) = oneshot::channel();
-        self.cmd_tx
-            .send(types::Command::Disconnect {
-                peer_id,
-                response: tx,
-            })
-            .await?;
+        self.cmd_tx.send(types::Command::Disconnect {
+            peer_id,
+            response: tx,
+        })?;
         rx.await.map_err(P2pError::from)?.map_err(P2pError::from)
     }
 
     async fn local_addr(&self) -> crate::Result<Option<T::Address>> {
         let (tx, rx) = oneshot::channel();
-        self.cmd_tx.send(types::Command::ListenAddress { response: tx }).await?;
+        self.cmd_tx.send(types::Command::ListenAddress { response: tx })?;
         rx.await.map_err(P2pError::from)
     }
 
@@ -215,12 +211,10 @@ where
         log::debug!("ban peer {}", peer_id);
 
         let (tx, rx) = oneshot::channel();
-        self.cmd_tx
-            .send(types::Command::BanPeer {
-                peer_id,
-                response: tx,
-            })
-            .await?;
+        self.cmd_tx.send(types::Command::BanPeer {
+            peer_id,
+            response: tx,
+        })?;
         rx.await.map_err(P2pError::from)?.map_err(P2pError::from)
     }