refactor: DEFAULT_QUEUE_SIZE

Author: Karrq

msg-socket/src/lib.rs

@@ -36,6 +36,9 @@ pub use connection::*;
 /// The default buffer size for a socket.
 pub const DEFAULT_BUFFER_SIZE: usize = 8192;
 
+/// The default queue size for a channel.
+pub const DEFAULT_QUEUE_SIZE: usize = 8192;
+
 /// A request Identifier.
 pub struct RequestId(u32);
 

msg-socket/src/rep/mod.rs

@@ -6,16 +6,14 @@ use msg_transport::Address;
 use thiserror::Error;
 use tokio::sync::oneshot;
 
-use crate::DEFAULT_BUFFER_SIZE;
-
 mod driver;
 mod socket;
-mod stats;
-use crate::{Profile, stats::SocketStats};
 pub use socket::*;
+
+mod stats;
 use stats::RepStats;
 
-const DEFAULT_MIN_COMPRESS_SIZE: usize = 8192;
+use crate::{DEFAULT_BUFFER_SIZE, DEFAULT_QUEUE_SIZE, Profile, stats::SocketStats};
 
 /// Errors that can occur when using a reply socket.
 #[derive(Debug, Error)]
@@ -47,23 +45,29 @@ impl RepError {
 pub struct RepOptions {
     /// The maximum number of concurrent clients.
     pub(crate) max_clients: Option<usize>,
+    /// Minimum payload size in bytes for compression to be used.
+    ///
+    /// If the payload is smaller than this threshold, it will not be compressed.
     pub(crate) min_compress_size: usize,
+    /// The size of the write buffer in bytes.
     pub(crate) write_buffer_size: usize,
+    /// The maximum duration between flushes to the underlying transport
     pub(crate) write_buffer_linger: Option<Duration>,
-    /// High-water mark for pending responses per peer. When this limit is reached,
-    /// new requests will not be read from the underlying connection until pending
-    /// responses are fulfilled.
+    /// High-water mark for pending responses per peer.
+    ///
+    /// When this limit is reached, new requests will not be read from the underlying connection
+    /// until pending responses are fulfilled.
     pub(crate) max_pending_responses: usize,
 }
 
 impl Default for RepOptions {
     fn default() -> Self {
         Self {
             max_clients: None,
-            min_compress_size: DEFAULT_MIN_COMPRESS_SIZE,
-            write_buffer_size: 8192,
+            min_compress_size: DEFAULT_BUFFER_SIZE,
+            write_buffer_size: DEFAULT_BUFFER_SIZE,
             write_buffer_linger: Some(Duration::from_micros(100)),
-            max_pending_responses: DEFAULT_BUFFER_SIZE,
+            max_pending_responses: DEFAULT_QUEUE_SIZE,
         }
     }
 }
@@ -115,6 +119,8 @@ impl RepOptions {
 
     /// Sets the minimum payload size for compression.
     /// If the payload is smaller than this value, it will not be compressed.
+    ///
+    /// Default: [`DEFAULT_BUFFER_SIZE`]
     pub fn with_min_compress_size(mut self, min_compress_size: usize) -> Self {
         self.min_compress_size = min_compress_size;
         self
@@ -123,7 +129,7 @@ impl RepOptions {
     /// Sets the size (max capacity) of the write buffer in bytes. When the buffer is full, it will
     /// be flushed to the underlying transport.
     ///
-    /// Default: 8KiB
+    /// Default: [`DEFAULT_BUFFER_SIZE`]
     pub fn with_write_buffer_size(mut self, size: usize) -> Self {
         self.write_buffer_size = size;
         self
@@ -142,7 +148,7 @@ impl RepOptions {
     /// new requests will not be read from the underlying connection until pending
     /// responses are fulfilled.
     ///
-    /// Default: [`DEFAULT_BUFFER_SIZE`]
+    /// Default: [`DEFAULT_QUEUE_SIZE`]
     pub fn with_max_pending_responses(mut self, hwm: usize) -> Self {
         self.max_pending_responses = hwm;
         self

msg-socket/src/rep/socket.rs

@@ -16,7 +16,7 @@ use tokio_stream::StreamMap;
 use tracing::{debug, warn};
 
 use crate::{
-    Authenticator, DEFAULT_BUFFER_SIZE, RepOptions, Request,
+    Authenticator, DEFAULT_QUEUE_SIZE, RepOptions, Request,
     rep::{RepError, SocketState, driver::RepDriver},
 };
 
@@ -110,8 +110,8 @@ where
 
     /// Binds the socket to the given address. This spawns the socket driver task.
     pub async fn try_bind(&mut self, addresses: Vec<A>) -> Result<(), RepError> {
-        let (to_socket, from_backend) = mpsc::channel(DEFAULT_BUFFER_SIZE);
-        let (control_tx, control_rx) = mpsc::channel(DEFAULT_BUFFER_SIZE);
+        let (to_socket, from_backend) = mpsc::channel(DEFAULT_QUEUE_SIZE);
+        let (control_tx, control_rx) = mpsc::channel(DEFAULT_QUEUE_SIZE);
 
         let mut transport = self.transport.take().expect("transport has been moved already");
 

msg-socket/src/req/mod.rs

@@ -23,7 +23,7 @@ pub use socket::*;
 use crate::{Profile, stats::SocketStats};
 use stats::ReqStats;
 
-use crate::DEFAULT_BUFFER_SIZE;
+use crate::{DEFAULT_BUFFER_SIZE, DEFAULT_QUEUE_SIZE};
 
 pub(crate) static DRIVER_ID: AtomicUsize = AtomicUsize::new(0);
 
@@ -194,6 +194,8 @@ impl ReqOptions {
     /// Sets the minimum payload size in bytes for compression to be used.
     ///
     /// If the payload is smaller than this threshold, it will not be compressed.
+    ///
+    /// Default: [`DEFAULT_BUFFER_SIZE`]
     pub fn with_min_compress_size(mut self, min_compress_size: usize) -> Self {
         self.min_compress_size = min_compress_size;
         self
@@ -202,7 +204,7 @@ impl ReqOptions {
     /// Sets the size (max capacity) of the write buffer in bytes.
     /// When the buffer is full, it will be flushed to the underlying transport.
     ///
-    /// Default: 8KiB
+    /// Default: [`DEFAULT_BUFFER_SIZE`]
     pub fn with_write_buffer_size(mut self, size: usize) -> Self {
         self.write_buffer_size = size;
         self
@@ -221,7 +223,7 @@ impl ReqOptions {
     /// This controls how many requests can be queued, on top of the current pending requests,
     /// before the socket returns [`ReqError::HighWaterMarkReached`].
     ///
-    /// Default: [`DEFAULT_BUFFER_SIZE`]
+    /// Default: [`DEFAULT_QUEUE_SIZE`]
     pub fn with_max_queue_size(mut self, size: usize) -> Self {
         self.max_queue_size = size;
         self
@@ -231,7 +233,7 @@ impl ReqOptions {
     /// will not be processed and will be queued up to [`Self::with_max_queue_size`] elements.
     /// Once both limits are reached, new requests will return [`ReqError::HighWaterMarkReached`].
     ///
-    /// Default: [`DEFAULT_BUFFER_SIZE`]
+    /// Default: [`DEFAULT_QUEUE_SIZE`]
     pub fn with_max_pending_requests(mut self, hwm: usize) -> Self {
         self.max_pending_requests = hwm;
         self
@@ -244,11 +246,11 @@ impl Default for ReqOptions {
             conn: ConnOptions::default(),
             timeout: Duration::from_secs(5),
             blocking_connect: false,
-            min_compress_size: 8192,
-            write_buffer_size: 8192,
+            min_compress_size: DEFAULT_BUFFER_SIZE,
+            write_buffer_size: DEFAULT_BUFFER_SIZE,
             write_buffer_linger: Some(Duration::from_micros(100)),
-            max_queue_size: DEFAULT_BUFFER_SIZE,
-            max_pending_requests: DEFAULT_BUFFER_SIZE,
+            max_queue_size: DEFAULT_QUEUE_SIZE,
+            max_pending_requests: DEFAULT_QUEUE_SIZE,
         }
     }
 }

msg-socket/src/sub/mod.rs

@@ -12,16 +12,14 @@ mod socket;
 pub use socket::*;
 
 mod stats;
-
-use crate::stats::SocketStats;
 use stats::SubStats;
 
 mod stream;
 
 use msg_transport::Address;
 use msg_wire::pubsub;
 
-use crate::DEFAULT_BUFFER_SIZE;
+use crate::{DEFAULT_BUFFER_SIZE, DEFAULT_QUEUE_SIZE, stats::SocketStats};
 
 #[derive(Debug, Error)]
 pub enum SubError {
@@ -61,7 +59,7 @@ pub struct SubOptions {
     auth_token: Option<Bytes>,
     /// The maximum amount of incoming messages that will be buffered before being dropped due to
     /// a slow consumer.
-    ingress_buffer_size: usize,
+    ingress_queue_size: usize,
     /// The read buffer size for each session.
     read_buffer_size: usize,
     /// The initial backoff for reconnecting to a publisher.
@@ -78,15 +76,19 @@ impl SubOptions {
         self
     }
 
-    /// Sets the ingress buffer size. This is the maximum amount of incoming messages that will be
+    /// Sets the ingress queue size. This is the maximum amount of incoming messages that will be
     /// buffered. If the consumer cannot keep up with the incoming messages, messages will start
     /// being dropped.
-    pub fn with_ingress_buffer_size(mut self, ingress_buffer_size: usize) -> Self {
-        self.ingress_buffer_size = ingress_buffer_size;
+    ///
+    /// Default: [`DEFAULT_QUEUE_SIZE`]
+    pub fn with_ingress_queue_size(mut self, ingress_queue_size: usize) -> Self {
+        self.ingress_queue_size = ingress_queue_size;
         self
     }
 
     /// Sets the read buffer size. This sets the size of the read buffer for each session.
+    ///
+    /// Default: [`DEFAULT_BUFFER_SIZE`]
     pub fn with_read_buffer_size(mut self, read_buffer_size: usize) -> Self {
         self.read_buffer_size = read_buffer_size;
         self
@@ -110,7 +112,7 @@ impl Default for SubOptions {
     fn default() -> Self {
         Self {
             auth_token: None,
-            ingress_buffer_size: DEFAULT_BUFFER_SIZE,
+            ingress_queue_size: DEFAULT_QUEUE_SIZE,
             read_buffer_size: 8192,
             initial_backoff: Duration::from_millis(100),
             retry_attempts: Some(24),

msg-socket/src/sub/socket.rs

@@ -17,19 +17,9 @@ use tokio::{
 use msg_common::{IpAddrExt, JoinMap};
 use msg_transport::{Address, Transport};
 
-// ADDED: Import the specific SubStats struct for the API
-use super::stats::SubStats;
-// Import the rest from the parent module (sub/mod.rs)
-use super::{
-    // REMOVED: Old/removed stats structs
-    // Command, PubMessage, SocketState, SocketStats, SocketWideStats, SubDriver, SubError,
-    Command,
-    DEFAULT_BUFFER_SIZE,
-    PubMessage,
-    SocketState,
-    SubDriver,
-    SubError,
-    SubOptions,
+use crate::sub::{
+    Command, DEFAULT_BUFFER_SIZE, PubMessage, SocketState, SubDriver, SubError, SubOptions,
+    stats::SubStats,
 };
 
 /// A subscriber socket. This socket implements [`Stream`] and yields incoming [`PubMessage`]s.
@@ -136,7 +126,7 @@ where
     /// Creates a new subscriber socket with the given transport and options.
     pub fn with_options(transport: T, options: SubOptions) -> Self {
         let (to_driver, from_socket) = mpsc::channel(DEFAULT_BUFFER_SIZE);
-        let (to_socket, from_driver) = mpsc::channel(options.ingress_buffer_size);
+        let (to_socket, from_driver) = mpsc::channel(options.ingress_queue_size);
 
         let options = Arc::new(options);
 

msg-socket/tests/it/reqrep.rs

@@ -1,7 +1,7 @@
 use std::time::Duration;
 
 use bytes::Bytes;
-use msg_socket::{DEFAULT_BUFFER_SIZE, RepSocket, ReqOptions, ReqSocket};
+use msg_socket::{DEFAULT_QUEUE_SIZE, RepSocket, ReqOptions, ReqSocket};
 use msg_transport::{
     tcp::Tcp,
     tcp_tls::{self, TcpTls},
@@ -273,11 +273,11 @@ async fn reqrep_hwm_reached() {
     // Share req via Arc for concurrent access
     let req = std::sync::Arc::new(req);
 
-    const TOTAL_CAPACITY: usize = HWM + DEFAULT_BUFFER_SIZE;
+    const TOTAL_CAPACITY: usize = HWM + DEFAULT_QUEUE_SIZE;
 
     // Send requests until the channel is full (HighWaterMarkReached error)
     // - HWM requests will be moved to pending_requests
-    // - DEFAULT_BUFFER_SIZE requests will be buffered in the channel (driver stops polling at HWM)
+    // - DEFAULT_QUEUE_SIZE requests will be buffered in the channel (driver stops polling at HWM)
     // - The next request will fail with HighWaterMarkReached
     let mut success_receivers = Vec::new();
     let mut sent_count = 0;

msg/benches/pubsub.rs

@@ -159,7 +159,7 @@ fn pubsub_single_thread_tcp(c: &mut Criterion) {
         Tcp::default(),
         SubOptions::default()
             .with_read_buffer_size(buffer_size)
-            .with_ingress_buffer_size(N_REQS * 2),
+            .with_ingress_queue_size(N_REQS * 2),
     );
 
     let mut bench = PairBenchmark {
@@ -201,7 +201,7 @@ fn pubsub_multi_thread_tcp(c: &mut Criterion) {
         Tcp::default(),
         SubOptions::default()
             .with_read_buffer_size(buffer_size)
-            .with_ingress_buffer_size(N_REQS * 2),
+            .with_ingress_queue_size(N_REQS * 2),
     );
 
     let mut bench = PairBenchmark {
@@ -242,7 +242,7 @@ fn pubsub_single_thread_quic(c: &mut Criterion) {
         Quic::default(),
         SubOptions::default()
             .with_read_buffer_size(buffer_size)
-            .with_ingress_buffer_size(N_REQS * 2),
+            .with_ingress_queue_size(N_REQS * 2),
     );
 
     let mut bench = PairBenchmark {
@@ -284,7 +284,7 @@ fn pubsub_multi_thread_quic(c: &mut Criterion) {
         Quic::default(),
         SubOptions::default()
             .with_read_buffer_size(buffer_size)
-            .with_ingress_buffer_size(N_REQS * 2),
+            .with_ingress_queue_size(N_REQS * 2),
     );
 
     let mut bench = PairBenchmark {
@@ -325,7 +325,7 @@ fn pubsub_single_thread_ipc(c: &mut Criterion) {
         Ipc::default(),
         SubOptions::default()
             .with_read_buffer_size(buffer_size)
-            .with_ingress_buffer_size(N_REQS * 2),
+            .with_ingress_queue_size(N_REQS * 2),
     );
 
     let mut bench = PairBenchmark {
@@ -367,7 +367,7 @@ fn pubsub_multi_thread_ipc(c: &mut Criterion) {
         Ipc::default(),
         SubOptions::default()
             .with_read_buffer_size(buffer_size)
-            .with_ingress_buffer_size(N_REQS * 2),
+            .with_ingress_queue_size(N_REQS * 2),
     );
 
     let mut bench = PairBenchmark {

msg/examples/pubsub.rs

@@ -21,13 +21,13 @@ async fn main() {
     // Configure the subscribers with options
     let mut sub1 = SubSocket::with_options(
         Tcp::default(),
-        SubOptions::default().with_ingress_buffer_size(1024),
+        SubOptions::default().with_ingress_queue_size(1024),
     );
 
     let mut sub2 = SubSocket::with_options(
         // TCP transport with blocking connect, usually connection happens in the background.
         Tcp::default(),
-        SubOptions::default().with_ingress_buffer_size(1024),
+        SubOptions::default().with_ingress_queue_size(1024),
     );
 
     tracing::info!("Setting up the sockets...");

msg/examples/quic_vs_tcp.rs

@@ -27,7 +27,7 @@ async fn run_tcp() {
     // Configure the subscribers with options
     let mut sub1 = SubSocket::with_options(
         Tcp::default(),
-        SubOptions::default().with_ingress_buffer_size(1024),
+        SubOptions::default().with_ingress_queue_size(1024),
     );
 
     tracing::info!("Setting up the sockets...");
@@ -60,7 +60,7 @@ async fn run_quic() {
     let mut sub1 = SubSocket::with_options(
         // TCP transport with blocking connect, usually connection happens in the background.
         Quic::default(),
-        SubOptions::default().with_ingress_buffer_size(1024),
+        SubOptions::default().with_ingress_queue_size(1024),
     );
 
     tracing::info!("Setting up the sockets...");