Optimize exact-scan MPC threshold primitives

ampc-actor-utils/src/network/mpc/handle/config.rs

@@ -1,4 +1,4 @@
-use std::{cmp, time::Duration};
+use std::time::Duration;
 
 #[derive(Default, Clone, Debug)]
 pub struct MpcConfig {
@@ -11,24 +11,24 @@ pub struct MpcConfig {
 
 impl MpcConfig {
     pub fn new(timeout_duration: Duration, num_connections: usize, num_sessions: usize) -> Self {
-        // don't allow fewer requests than connections...
-        let connection_parallelism = cmp::min(num_connections, num_sessions);
+        assert!(num_connections > 0, "MPC networking requires a connection");
+        assert!(num_sessions > 0, "MPC networking requires a session");
 
         Self {
             timeout_duration,
             num_sessions: num_sessions as u32,
-            num_connections: connection_parallelism as u32,
+            // A logical session is striped over every connection. Do not clamp
+            // this to the session count: batch-size-one requests need multiple
+            // physical flows to exceed a cloud provider's per-flow limit.
+            num_connections: num_connections as u32,
         }
     }
 
     pub fn get_sessions_for_connection(&self, idx: u32) -> u32 {
-        let num_sessions = self.num_sessions;
-        let num_connections = self.num_connections;
-        num_sessions / num_connections
-            + if idx < (num_sessions % num_connections) {
-                1
-            } else {
-                0
-            }
+        if idx < self.num_connections {
+            self.num_sessions
+        } else {
+            0
+        }
     }
 }

ampc-actor-utils/src/network/mpc/handle/control_channel.rs

@@ -150,19 +150,19 @@ impl<T: NetworkConnection> ControlChannel for TcpControlChannel<T> {
     }
 
     async fn sync(&mut self) -> Result<()> {
-        let token = NetworkValue::Bytes(SYNC_TOKEN_BYTES.to_vec());
+        let token = NetworkValue::Bytes(SYNC_TOKEN_BYTES.to_vec().into());
         self.send_next(token.clone()).await?;
         self.send_prev(token).await?;
 
         let next_token = self.recv_next().await?;
         match next_token {
-            NetworkValue::Bytes(ref bytes) if bytes == SYNC_TOKEN_BYTES => {}
+            NetworkValue::Bytes(ref bytes) if &bytes[..] == SYNC_TOKEN_BYTES => {}
             _ => bail!("invalid sync token received from next party"),
         }
 
         let prev_token = self.recv_prev().await?;
         match prev_token {
-            NetworkValue::Bytes(ref bytes) if bytes == SYNC_TOKEN_BYTES => {}
+            NetworkValue::Bytes(ref bytes) if &bytes[..] == SYNC_TOKEN_BYTES => {}
             _ => bail!("invalid sync token received from prev party"),
         }