feat: add socket buffer size configuration (#22)

Author: wemeetagain

Cargo.toml

@@ -21,6 +21,7 @@ rustls = { version = "0.23.23", features = ["ring"] }
 tokio = { version = "1.43.0", features = ["full"] }
 # included to satisfy napi dependencies
 ctor = "0.3.6"
+socket2 = "0.5.8"
 
 [build-dependencies]
 napi-build = "2.1.4"

rust/config.rs

@@ -64,25 +64,55 @@ pub struct Config {
   /// Timeout for the initial handshake when establishing a connection.
   /// The actual timeout is the minimum of this and the [`Config::max_idle_timeout`].
   pub handshake_timeout: u32,
+
   /// Maximum duration of inactivity in ms to accept before timing out the connection.
   pub max_idle_timeout: u32,
+
   /// Period of inactivity before sending a keep-alive packet.
   /// Must be set lower than the idle_timeout of both
   /// peers to be effective.
   ///
   /// See [`quinn::TransportConfig::keep_alive_interval`] for more
   /// info.
   pub keep_alive_interval: u32,
+
   /// Maximum number of incoming bidirectional streams that may be open
   /// concurrently by the remote peer.
   pub max_concurrent_stream_limit: u32,
 
-  /// Max unacknowledged data in bytes that may be sent on a single stream.
+  /// Maximum number of bytes the peer may transmit without acknowledgement on any one stream
+  /// before becoming blocked.
+  ///
+  /// This should be set to at least the expected connection latency multiplied by the maximum
+  /// desired throughput. Setting this smaller than `max_connection_data` helps ensure that a single
+  /// stream doesn't monopolize receive buffers, which may otherwise occur if the application
+  /// chooses not to read from a large stream for a time while still requiring data on other
+  /// streams.
   pub max_stream_data: u32,
 
-  /// Max unacknowledged data in bytes that may be sent in total on all streams
-  /// of a connection.
+  /// Maximum number of bytes the peer may transmit across all streams of a connection before
+  /// becoming blocked.
+  ///
+  /// This should be set to at least the expected connection latency multiplied by the maximum
+  /// desired throughput. Larger values can be useful to allow maximum throughput within a
+  /// stream while another is blocked.
   pub max_connection_data: u32,
+
+  /// OS socket receive buffer size.
+  ///
+  /// If this is set higher than the OS maximum, it will be clamped to the maximum allowed size.
+  pub receive_buffer_size: u32,
+
+  /// OS socket send buffer size.
+  ///
+  /// If this is set higher than the OS maximum, it will be clamped to the maximum allowed size.
+  pub send_buffer_size: u32,
+}
+
+#[derive(Clone, Copy)]
+pub struct SocketConfig {
+  pub receive_buffer_size: u32,
+  pub send_buffer_size: u32,
 }
 
 /// Configuration used by the QUIC library
@@ -92,6 +122,7 @@ pub struct QuinnConfig {
   pub(crate) client_config: quinn::ClientConfig,
   pub(crate) server_config: quinn::ServerConfig,
   pub(crate) endpoint_config: quinn::EndpointConfig,
+  pub(crate) socket_config: SocketConfig,
 }
 
 #[napi]
@@ -113,20 +144,25 @@ impl TryFrom<Config> for QuinnConfig {
       max_connection_data,
       max_stream_data,
       handshake_timeout: _,
+      receive_buffer_size,
+      send_buffer_size,
     } = config;
 
     let keypair = libp2p_identity::Keypair::from_protobuf_encoding(&private_key_proto)
       .map_err(|e| ConfigError::InvalidPrivateKey(e))?;
 
     let mut transport = quinn::TransportConfig::default();
+
+    // Disable features we don't use/want
     // Disable uni-directional streams.
     transport.max_concurrent_uni_streams(0u32.into());
-    transport.max_concurrent_bidi_streams(max_concurrent_stream_limit.into());
     // Disable datagrams.
     transport.datagram_receive_buffer_size(None);
+    transport.allow_spin(false);
+
+    transport.max_concurrent_bidi_streams(max_concurrent_stream_limit.into());
     transport.keep_alive_interval(Some(Duration::from_millis(keep_alive_interval.into())));
     transport.max_idle_timeout(Some(quinn::VarInt::from_u32(max_idle_timeout).into()));
-    transport.allow_spin(false);
     transport.stream_receive_window(max_stream_data.into());
     transport.receive_window(max_connection_data.into());
     transport.mtu_discovery_config(Default::default());
@@ -170,6 +206,10 @@ impl TryFrom<Config> for QuinnConfig {
       client_config,
       server_config,
       endpoint_config,
+      socket_config: SocketConfig {
+        receive_buffer_size,
+        send_buffer_size,
+      },
     })
   }
 }

rust/lib.rs

@@ -30,7 +30,8 @@ impl Server {
   pub fn new(config: &config::QuinnConfig, ip: String, port: u16) -> Result<Self> {
     let ip_addr = ip.parse::<IpAddr>().map_err(to_err)?;
     let socket_addr = SocketAddr::new(ip_addr, port);
-    let socket = std::net::UdpSocket::bind(socket_addr)?;
+    let socket = socket::create_socket(config.socket_config, socket_addr)?;
+
     let socket = block_on(async move{
       socket::UdpSocket::wrap_udp_socket(socket)
     })?;
@@ -93,7 +94,7 @@ impl Client {
       SocketFamily::Ipv6 => SocketAddr::new(std::net::Ipv6Addr::UNSPECIFIED.into(), 0),
     };
     let mut endpoint = block_on(async move {
-      let socket = std::net::UdpSocket::bind(bind_addr)?;
+      let socket = socket::create_socket(config.socket_config,bind_addr)?;
       let endpoint = quinn::Endpoint::new(
         config.endpoint_config.clone(),
         None,

rust/socket.rs

@@ -23,6 +23,19 @@ use std::{
 use quinn::udp;
 use quinn::{AsyncUdpSocket, UdpPoller};
 use tokio::{io::Interest, sync::RwLock};
+use socket2::Socket;
+
+use crate::config;
+
+pub fn create_socket(config: config::SocketConfig, socket_addr: std::net::SocketAddr) -> napi::Result<std::net::UdpSocket> {
+    let socket = std::net::UdpSocket::bind(socket_addr)?;
+
+    let socket = Socket::from(socket);
+    socket.set_send_buffer_size(config.send_buffer_size as usize).unwrap();
+    socket.set_recv_buffer_size(config.receive_buffer_size as usize).unwrap();
+    let socket = std::net::UdpSocket::from(socket);
+    Ok(socket)
+}
 
 #[derive(Debug)]
 pub struct UdpSocket {
@@ -244,3 +257,36 @@ impl<MakeFut, Fut> std::fmt::Debug for UdpPollHelper<MakeFut, Fut> {
     f.debug_struct("UdpPollHelper").finish_non_exhaustive()
   }
 }
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  /// Check how buffer sizes are set. On my (linux, ubuntu 24.04.2) machine, this is the output:
+  ///
+  /// send buffer size: 212992
+  /// receive buffer size: 212992
+  /// updating buffer sizes: 1000000
+  /// send buffer size: 425984
+  /// receive buffer size: 425984
+  #[test]
+  fn check_default_buffer_sizes() {
+    let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+    let socket = Socket::from(socket);
+    let send_buffer_size = socket.send_buffer_size().unwrap();
+    let receive_buffer_size = socket.recv_buffer_size().unwrap();
+    println!("send buffer size: {}", send_buffer_size);
+    println!("receive buffer size: {}", receive_buffer_size);
+
+    let new_size: usize = 1_000_000;
+    println!("updating buffer sizes: {}", new_size);
+
+    socket.set_send_buffer_size(new_size).unwrap();
+    socket.set_recv_buffer_size(new_size).unwrap();
+
+    let send_buffer_size = socket.send_buffer_size().unwrap();
+    let receive_buffer_size = socket.recv_buffer_size().unwrap();
+    println!("send buffer size: {}", send_buffer_size);
+    println!("receive buffer size: {}", receive_buffer_size);
+  }
+}

src/index.ts

@@ -51,5 +51,7 @@ export const defaultOptions: QuicOptions = {
   keepAliveInterval: 5_000,
   maxConcurrentStreamLimit: 256,
   maxStreamData: 10_000_000,
-  maxConnectionData: 15_000_000
+  maxConnectionData: 15_000_000,
+  receiveBufferSize: 500_000,
+  sendBufferSize: 500_000,
 }

src/napi.d.ts

@@ -382,13 +382,38 @@ export interface Config {
    * concurrently by the remote peer.
    */
   maxConcurrentStreamLimit: number
-  /** Max unacknowledged data in bytes that may be sent on a single stream. */
+  /**
+   * Maximum number of bytes the peer may transmit without acknowledgement on any one stream
+   * before becoming blocked.
+   *
+   * This should be set to at least the expected connection latency multiplied by the maximum
+   * desired throughput. Setting this smaller than `max_connection_data` helps ensure that a single
+   * stream doesn't monopolize receive buffers, which may otherwise occur if the application
+   * chooses not to read from a large stream for a time while still requiring data on other
+   * streams.
+   */
   maxStreamData: number
   /**
-   * Max unacknowledged data in bytes that may be sent in total on all streams
-   * of a connection.
+   * Maximum number of bytes the peer may transmit across all streams of a connection before
+   * becoming blocked.
+   *
+   * This should be set to at least the expected connection latency multiplied by the maximum
+   * desired throughput. Larger values can be useful to allow maximum throughput within a
+   * stream while another is blocked.
    */
   maxConnectionData: number
+  /**
+   * OS socket receive buffer size.
+   *
+   * If this is set higher than the OS maximum, it will be clamped to the maximum allowed size.
+   */
+  receiveBufferSize: number
+  /**
+   * OS socket send buffer size.
+   *
+   * If this is set higher than the OS maximum, it will be clamped to the maximum allowed size.
+   */
+  sendBufferSize: number
 }
 
 export declare const enum SocketFamily {