reducing acks sent

Cargo.toml

@@ -13,8 +13,8 @@ etherparse = "0.16.0"
 local-ip-address = "0.6.3"
 rand = "0.8.5"
 rstest = "0.23.0"
-thiserror = "2.0.6"
-tokio = { version = "1.41.1", features = ["full", "tracing"] }
+thiserror = "2.0.8"
+tokio = { version = "1.42.0", features = ["full", "tracing"] }
 console-subscriber = "0.4.1"
 libc = "0.2.164"
 sysctl = "0.6.0"

src/bin/client.rs

@@ -59,7 +59,7 @@ async fn main() -> BluefinResult<()> {
                 }
             }
             println!("Sent {} bytes", total_bytes);
-            sleep(Duration::from_secs(2)).await;
+            sleep(Duration::from_secs(3)).await;
 
             Ok::<(), BluefinError>(())
         });

src/worker/conn_reader.rs

@@ -5,9 +5,11 @@ use tokio::sync::mpsc::{self};
 use crate::core::error::BluefinError;
 use crate::core::header::PacketType;
 use crate::core::packet::BluefinPacket;
+use crate::net::ack_handler::AckBuffer;
+use crate::net::connection::ConnectionBuffer;
 use crate::net::{ConnectionManagedBuffers, MAX_BLUEFIN_BYTES_IN_UDP_DATAGRAM};
 use crate::utils::common::BluefinResult;
-use std::sync::Arc;
+use std::sync::{Arc, MutexGuard};
 
 pub(crate) struct ConnReaderHandler {
     socket: Arc<UdpSocket>,
@@ -21,7 +23,7 @@ impl ConnReaderHandler {
 
     pub(crate) fn start(&self) -> BluefinResult<()> {
         let (tx, rx) = mpsc::channel::<Vec<BluefinPacket>>(1024);
-        for _ in 0..4 {
+        for _ in 0..2 {
             let tx_cloned = tx.clone();
             let socket_cloned = self.socket.clone();
             spawn(async move {
@@ -61,31 +63,75 @@ impl ConnReaderHandler {
     ) {
         loop {
             if let Some(packets) = rx.recv().await {
-                for p in packets {
-                    let _ = ConnReaderHandler::buffer_in_packet(conn_bufs, p);
-                }
+                let _ = Self::buffer_in_packets(packets, conn_bufs);
             }
         }
     }
 
     #[inline]
-    fn buffer_in_packet(
+    fn buffer_in_packets(
+        packets: Vec<BluefinPacket>,
         conn_bufs: &ConnectionManagedBuffers,
-        packet: BluefinPacket,
     ) -> BluefinResult<()> {
-        if packet.header.type_field == PacketType::Ack {
-            let mut guard = conn_bufs.ack_buff.lock().unwrap();
-            guard.buffer_in_ack_packet(packet)?;
-            guard.wake()?;
-        } else {
-            let mut guard = conn_bufs.conn_buff.lock().unwrap();
-            guard.buffer_in_bytes(packet)?;
-            if let Some(w) = guard.get_waker() {
-                w.wake_by_ref();
-            } else {
-                return Err(BluefinError::NoSuchWakerError);
+        // Nothing to do if empty
+        if packets.is_empty() {
+            return Ok(());
+        }
+
+        // Peek at the first packet and acquire the buffer.
+        let p = packets.first().unwrap();
+        match p.header.type_field {
+            PacketType::Ack => {
+                let guard = conn_bufs.ack_buff.lock().unwrap();
+                Self::buffer_in_ack_packets(guard, packets)
+            }
+            _ => {
+                let guard = conn_bufs.conn_buff.lock().unwrap();
+                Self::buffer_in_data_packets(guard, packets)
+            }
+        }
+    }
+
+    #[inline]
+    fn buffer_in_ack_packets(
+        mut guard: MutexGuard<'_, AckBuffer>,
+        packets: Vec<BluefinPacket>,
+    ) -> BluefinResult<()> {
+        let mut e: Option<BluefinError> = None;
+        for p in packets {
+            if let Err(err) = guard.buffer_in_ack_packet(p) {
+                e = Some(err);
+            }
+        }
+        guard.wake()?;
+
+        if e.is_some() {
+            return Err(e.unwrap());
+        }
+        Ok(())
+    }
+
+    #[inline]
+    fn buffer_in_data_packets(
+        mut guard: MutexGuard<'_, ConnectionBuffer>,
+        packets: Vec<BluefinPacket>,
+    ) -> BluefinResult<()> {
+        let mut e: Option<BluefinError> = None;
+        for p in packets {
+            if let Err(err) = guard.buffer_in_bytes(p) {
+                e = Some(err);
             }
         }
+
+        if let Some(w) = guard.get_waker() {
+            w.wake_by_ref();
+        } else {
+            return Err(BluefinError::NoSuchWakerError);
+        }
+
+        if e.is_some() {
+            return Err(e.unwrap());
+        }
         Ok(())
     }
 }

src/worker/reader.rs

@@ -44,6 +44,8 @@ pub(crate) struct ReaderTxChannel {
 pub(crate) struct ReaderRxChannel {
     future: ReaderRxChannelFuture,
     writer_handler: WriterHandler,
+    packets_consumed: usize,
+    packets_consumed_before_ack: usize,
 }
 
 #[derive(Clone)]
@@ -74,6 +76,8 @@ impl ReaderRxChannel {
         Self {
             future,
             writer_handler,
+            packets_consumed: 0,
+            packets_consumed_before_ack: 100,
         }
     }
 
@@ -90,9 +94,13 @@ impl ReaderRxChannel {
         };
         let num_packets_consumed = consume_res.get_num_packets_consumed();
         let base_packet_num = consume_res.get_base_packet_number();
+        self.packets_consumed += num_packets_consumed;
 
         // We need to send an ack.
-        if num_packets_consumed > 0 && base_packet_num != 0 {
+        if num_packets_consumed > 0
+            && base_packet_num != 0
+            && self.packets_consumed >= self.packets_consumed_before_ack
+        {
             if let Err(e) = self
                 .writer_handler
                 .send_ack(base_packet_num, num_packets_consumed)
@@ -102,6 +110,7 @@ impl ReaderRxChannel {
                     e
                 );
             }
+            self.packets_consumed = 0;
         }
 
         Ok((consume_res.get_bytes_consumed(), addr))