Batch p2p messages in one round instead of flushing immediately

Signed-off-by: maurges <nikita@dfns.co>

Author: maurges

Cargo.lock

@@ -28,6 +28,8 @@ serde = { workspace = true, features = ["derive"] }
 serde_with = { workspace = true }
 hex = { workspace = true, features = ["serde"] }
 
+futures = { workspace = true }
+
 displaydoc = { workspace = true }
 thiserror = { workspace = true, optional = true }
 

cggmp21-keygen/Cargo.toml

@@ -277,15 +277,16 @@ where
         .await
         .map_err(IoError::send_message)?;
 
-    for j in utils::iter_peers(i, n) {
+    let messages = utils::iter_peers(i, n).map(|j| {
         let message = MsgRound2Uni {
             sigma: sigmas[usize::from(j)],
         };
-        outgoings
-            .send(Outgoing::p2p(j, Msg::Round2Uni(message)))
-            .await
-            .map_err(IoError::send_message)?;
-    }
+        Outgoing::p2p(j, Msg::Round2Uni(message))
+    });
+    outgoings
+        .send_all(&mut futures::stream::iter(messages.map(Ok)))
+        .await
+        .map_err(IoError::send_message)?;
     tracer.msg_sent();
 
     // Round 3

cggmp21-keygen/src/threshold.rs

@@ -715,13 +715,12 @@ where
         )
         .map_err(|e| Bug::PiEnc(BugSource::psi0, e))?;
 
-        tracer.send_msg();
         outgoings
-            .send(Outgoing::p2p(j, Msg::Round1b(MsgRound1b { psi0 })))
+            .feed(Outgoing::p2p(j, Msg::Round1b(MsgRound1b { psi0 })))
             .await
             .map_err(IoError::send_message)?;
-        tracer.msg_sent();
     }
+    outgoings.flush().await.map_err(IoError::send_message)?;
 
     // Round 2
     tracer.round_begins();
@@ -950,9 +949,8 @@ where
         .map_err(|e| Bug::PiLog(BugSource::psi_prime, e))?;
         runtime.yield_now().await;
 
-        tracer.send_msg();
         outgoings
-            .send(Outgoing::p2p(
+            .feed(Outgoing::p2p(
                 j,
                 Msg::Round2(MsgRound2 {
                     Gamma: Gamma_i,
@@ -967,8 +965,8 @@ where
             ))
             .await
             .map_err(IoError::send_message)?;
-        tracer.msg_sent();
     }
+    outgoings.flush().await.map_err(IoError::send_message)?;
 
     // Round 3
     tracer.round_begins();
@@ -1124,9 +1122,8 @@ where
         )
         .map_err(|e| Bug::PiLog(BugSource::psi_prime_prime, e))?;
 
-        tracer.send_msg();
         outgoings
-            .send(Outgoing::p2p(
+            .feed(Outgoing::p2p(
                 j,
                 Msg::Round3(MsgRound3 {
                     delta: delta_i,
@@ -1136,8 +1133,8 @@ where
             ))
             .await
             .map_err(IoError::send_message)?;
-        tracer.msg_sent();
     }
+    outgoings.flush().await.map_err(IoError::send_message)?;
 
     // Output
     tracer.named_round_begins("Presig output");