chmux: minor performance improvements

Author: surban

remoc/src/chmux/msg.rs

@@ -311,8 +311,8 @@ impl MultiplexMsg {
                 let first = flags & MSG_PORT_DATA_FLAG_FIRST != 0;
                 let last = flags & MSG_PORT_DATA_FLAG_LAST != 0;
                 let wait = flags & MSG_PORT_DATA_FLAG_WAIT != 0;
-                let mut ids = (flags & MSG_PORT_DATA_FLAG_IDS != 0).then_some(Vec::new());
-                let mut ports = Vec::new();
+                let mut ids = (flags & MSG_PORT_DATA_FLAG_IDS != 0).then_some(Vec::with_capacity(16));
+                let mut ports = Vec::with_capacity(16);
                 loop {
                     match reader.read_u32::<LE>() {
                         Ok(p) => ports.push(p),
@@ -340,7 +340,7 @@ impl MultiplexMsg {
     }
 
     pub(crate) fn to_vec(&self) -> Vec<u8> {
-        let mut data = Vec::new();
+        let mut data = Vec::with_capacity(MAX_MSG_LENGTH);
         self.write(&mut data).expect("message serialization failed");
         data
     }

remoc/src/chmux/mux.rs

@@ -23,6 +23,7 @@ use tokio::{
     sync::{mpsc, mpsc::Permit, oneshot},
     try_join,
 };
+use tokio_util::sync::ReusableBoxFuture;
 
 use super::{
     AnyStorage, Cfg, ChMuxError, PROTOCOL_VERSION, PROTOCOL_VERSION_PORT_ID, PortReq,
@@ -535,7 +536,7 @@ where
             }
         }
 
-        let mut next_ping = get_next_ping(ping_interval).fuse().boxed();
+        let mut next_ping = ReusableBoxFuture::new(get_next_ping(ping_interval));
         let mut need_flush = false;
 
         loop {
@@ -555,7 +556,7 @@ where
                                 break;
                             }
 
-                            next_ping = get_next_ping(ping_interval).fuse().boxed();
+                            next_ping.set(get_next_ping(ping_interval));
                             need_flush = true;
                         }
                         None => break,
@@ -564,7 +565,7 @@ where
 
                 () = &mut next_ping => {
                     Self::feed_msg(TransportMsg::new(MultiplexMsg::Ping), sink).await?;
-                    next_ping = get_next_ping(ping_interval).fuse().boxed();
+                    next_ping.set(get_next_ping(ping_interval));
                     need_flush = true;
                 }
 
@@ -595,7 +596,8 @@ where
             }
         }
 
-        let mut next_timeout = get_connection_timeout(connection_timeout).fuse().boxed();
+        let mut next_timeout = ReusableBoxFuture::new(get_connection_timeout(connection_timeout));
+
         while let Ok(tx_permit) = tx.reserve().await {
             tokio::select! {
                 biased;
@@ -608,7 +610,7 @@ where
                         break;
                     }
 
-                    next_timeout = get_connection_timeout(connection_timeout).fuse().boxed();
+                    next_timeout.set(get_connection_timeout(connection_timeout));
                 },
 
                 () = &mut next_timeout => return Err(ChMuxError::Timeout),