use try_send on connToMain channel

Author: andrewjstone

trust-quorum/src/connection_manager.rs

@@ -446,19 +446,16 @@ impl ConnMgr {
             );
 
             // Inform the main task that accepted connection is established
-            if let Err(e) = main_tx
-                .send(ConnToMainMsg {
-                    task_id: task::id(),
-                    msg: ConnToMainMsgInner::Accepted {
-                        addr,
-                        peer_id: baseboard_id,
-                    },
-                })
-                .await
-            {
-                // The system is shutting down
+            if let Err(_) = main_tx.try_send(ConnToMainMsg {
+                task_id: task::id(),
+                msg: ConnToMainMsgInner::Accepted {
+                    addr,
+                    peer_id: baseboard_id,
+                },
+            }) {
+                // The system is shutting down or we've overloaded the main channel
                 // Just bail from this task
-                warn!(log, "Failed to send 'accepted' msg to main task: {e:?}");
+                warn!(log, "Failed to send 'accepted' msg to main task");
             } else {
                 conn.run().await;
             }
@@ -663,22 +660,16 @@ impl ConnMgr {
             );
             // Inform the main task that the client connection is
             // established.
-            if let Err(e) = main_tx
-                .send(ConnToMainMsg {
-                    task_id: task::id(),
-                    msg: ConnToMainMsgInner::Connected {
-                        addr,
-                        peer_id: baseboard_id,
-                    },
-                })
-                .await
-            {
-                // The system is shutting down
+            if let Err(_) = main_tx.try_send(ConnToMainMsg {
+                task_id: task::id(),
+                msg: ConnToMainMsgInner::Connected {
+                    addr,
+                    peer_id: baseboard_id,
+                },
+            }) {
+                // The system is shutting down or we've overloaded the main channel
                 // Just bail from this task
-                error!(
-                    log,
-                    "Failed to send 'connected' msg to main task: {e:?}"
-                );
+                error!(log, "Failed to send 'connected' msg to main task");
             } else {
                 conn.run().await;
             }

trust-quorum/src/established_conn.rs

@@ -155,17 +155,13 @@ impl EstablishedConn {
     }
 
     async fn close(&mut self) {
-        if let Err(e) = self
-            .main_tx
-            .send(ConnToMainMsg {
-                task_id: self.task_id,
-                msg: ConnToMainMsgInner::Disconnected {
-                    peer_id: self.peer_id.clone(),
-                },
-            })
-            .await
-        {
-            warn!(self.log, "Failed to send to main task: {e:?}");
+        if let Err(_) = self.main_tx.try_send(ConnToMainMsg {
+            task_id: self.task_id,
+            msg: ConnToMainMsgInner::Disconnected {
+                peer_id: self.peer_id.clone(),
+            },
+        }) {
+            warn!(self.log, "Failed to send to main task");
         }
         let _ = self.writer.shutdown().await;
     }
@@ -202,21 +198,18 @@ impl EstablishedConn {
             debug!(self.log, "Received {msg:?}");
             match msg {
                 WireMsg::Tq(msg) => {
-                    if let Err(e) = self
-                        .main_tx
-                        .send(ConnToMainMsg {
-                            task_id: self.task_id,
-                            msg: ConnToMainMsgInner::Received {
-                                from: self.peer_id.clone(),
-                                msg,
-                            },
-                        })
-                        .await
-                    {
-                        warn!(
+                    if let Err(_) = self.main_tx.try_send(ConnToMainMsg {
+                        task_id: self.task_id,
+                        msg: ConnToMainMsgInner::Received {
+                            from: self.peer_id.clone(),
+                            msg,
+                        },
+                    }) {
+                        error!(
                             self.log,
-                            "Failed to send received fsm msg to main task: {e:?}"
+                            "Failed to send received fsm msg to main task"
                         );
+                        panic!("Connection to main task channel full");
                     }
                 }
                 WireMsg::Ping => {
@@ -225,22 +218,19 @@ impl EstablishedConn {
                 }
                 WireMsg::NetworkConfig(config) => {
                     let generation = config.generation;
-                    if let Err(e) = self
-                        .main_tx
-                        .send(ConnToMainMsg {
-                            task_id: self.task_id,
-                            msg: ConnToMainMsgInner::ReceivedNetworkConfig {
-                                from: self.peer_id.clone(),
-                                config,
-                            },
-                        })
-                        .await
-                    {
+                    if let Err(_) = self.main_tx.try_send(ConnToMainMsg {
+                        task_id: self.task_id,
+                        msg: ConnToMainMsgInner::ReceivedNetworkConfig {
+                            from: self.peer_id.clone(),
+                            config,
+                        },
+                    }) {
                         warn!(
                             self.log,
                             "Failed to send received NetworkConfig with
-                             generation {generation} to main task: {e:?}"
+                             generation {generation} to main task"
                         );
+                        panic!("Connection to main task channnel full");
                     }
                 }
             }

trust-quorum/src/task.rs

@@ -18,6 +18,18 @@ use tokio::sync::oneshot::error::RecvError;
 use tokio::sync::{mpsc, oneshot};
 use trust_quorum_protocol::{BaseboardId, Node, NodeCtx};
 
+/// We only expect a handful of messages at a time.
+const API_CHANNEL_BOUND: usize = 32;
+
+/// We size this bound large enough that it should never be hit. Up to 31
+/// `EstablishedConn` tasks can send messages to the main task simultaneously when
+/// messages are received.
+///
+/// We use `try_send.unwrap()` when sending to the main task to prevent deadlock
+/// and inform us via panic that something has gone seriously wrong. This is
+/// similar to using an unbounded channel but will not use all possible memory.
+const CONN_TO_MAIN_CHANNEL_BOUND: usize = 1024;
+
 #[derive(Debug, Clone)]
 pub struct Config {
     pub baseboard_id: BaseboardId,
@@ -130,13 +142,11 @@ impl NodeTask {
             "component" => "trust-quorum",
             "baseboard_id" => config.baseboard_id.to_string()
         ));
-        // We only expect one outstanding request at a time for `Init_` or
-        // `LoadRackSecret` requests, We can have one of those requests in
-        // flight while allowing `PeerAddresses` updates. We also allow status
-        // requests in parallel. Just leave some room.
-        let (tx, rx) = mpsc::channel(10);
 
-        let (conn_mgr_tx, conn_mgr_rx) = mpsc::channel(100);
+        let (tx, rx) = mpsc::channel(API_CHANNEL_BOUND);
+
+        let (conn_mgr_tx, conn_mgr_rx) =
+            mpsc::channel(CONN_TO_MAIN_CHANNEL_BOUND);
 
         let baseboard_id = config.baseboard_id.clone();