Rework the pipeline: batch allocation + deterministic backoff

Cargo.toml

@@ -92,7 +92,6 @@ console-subscriber = "0.4.0"
 const_format = "0.2.33"
 crc = "3.2.1"
 criterion = "0.5"
-crossbeam-utils = "0.8.20"
 crossbeam-queue = "0.3.12"
 derive_more = { version = "1.0.0", features = ["as_ref"] }
 derive-new = "0.7.0"

DEFAULT_CONFIG.json5

@@ -471,14 +471,14 @@
           /// then amount of memory being allocated for each queue is SIZE_XXX * LINK_MTU.
           /// If qos is false, then only the DATA priority will be allocated.
           size: {
-            control: 2,
-            real_time: 2,
-            interactive_high: 2,
-            interactive_low: 2,
+            control: 1,
+            real_time: 1,
+            interactive_high: 1,
+            interactive_low: 1,
             data_high: 2,
-            data: 2,
-            data_low: 2,
-            background: 2,
+            data: 4,
+            data_low: 4,
+            background: 4,
           },
           /// Congestion occurs when the queue is empty (no available batch).
           congestion_control: {
@@ -506,12 +506,6 @@
             /// The maximum time limit (in ms) a message should be retained for batching when back-pressure happens.
             time_limit: 1,
           },
-          allocation: {
-            /// Mode for memory allocation of batches in the priority queues. 
-            /// - "init": batches are allocated at queue initialization time. 
-            /// - "lazy": batches are allocated when needed up to the maximum number of batches configured in the size configuration parameter.
-            mode: "lazy",
-          },
         },
       },
       /// Configure the zenoh RX parameters of a link

commons/zenoh-config/src/defaults.rs

@@ -274,14 +274,14 @@ impl QueueSizeConf {
 impl Default for QueueSizeConf {
     fn default() -> Self {
         Self {
-            control: 2,
-            real_time: 2,
-            interactive_low: 2,
-            interactive_high: 2,
+            control: 1,
+            real_time: 1,
+            interactive_low: 1,
+            interactive_high: 1,
             data_high: 2,
-            data: 2,
+            data: 4,
             data_low: 2,
-            background: 2,
+            background: 1,
         }
     }
 }

commons/zenoh-config/src/lib.rs

@@ -508,13 +508,6 @@ validated_struct::validator! {
                             /// The maximum time limit (in ms) a message should be retained for batching when back-pressure happens.
                             time_limit: u64,
                         },
-                        /// Perform lazy memory allocation of batches in the prioritiey queues. If set to false all batches are initialized at
-                        /// initialization time. If set to true the batches will be allocated when needed up to the maximum number of batches
-                        /// configured in the size configuration parameter.
-                        pub allocation: #[derive(Default, Copy, PartialEq, Eq)]
-                        QueueAllocConf {
-                            pub mode: QueueAllocMode,
-                        },
                     },
                     // Number of threads used for TX
                     threads: usize,
@@ -659,14 +652,6 @@ validated_struct::validator! {
     }
 }
 
-#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum QueueAllocMode {
-    Init,
-    #[default]
-    Lazy,
-}
-
 #[derive(Default, Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum ShmInitMode {

io/zenoh-transport/Cargo.toml

@@ -44,28 +44,28 @@ transport_ws = ["zenoh-link/transport_ws"]
 transport_serial = ["zenoh-link/transport_serial"]
 transport_compression = []
 transport_unixpipe = ["zenoh-link/transport_unixpipe"]
-transport_vsock= ["zenoh-link/transport_vsock"]
+transport_vsock = ["zenoh-link/transport_vsock"]
 stats = ["zenoh-protocol/stats"]
 test = []
 unstable = []
 default = ["test", "transport_multilink"]
 
 [dependencies]
 async-trait = { workspace = true }
-crossbeam-utils = { workspace = true }
 tokio = { workspace = true, features = [
-  "sync",
-  "fs",
-  "time",
-  "macros",
-  "rt-multi-thread",
-  "io-util",
-  "net",
+    "sync",
+    "fs",
+    "time",
+    "macros",
+    "rt-multi-thread",
+    "io-util",
+    "net",
 ] }
 lazy_static = { workspace = true }
-tokio-util = { workspace = true, features = ["rt"]}
+tokio-util = { workspace = true, features = ["rt"] }
 flume = { workspace = true }
-tracing = {workspace = true}
+futures = { workspace = true }
+tracing = { workspace = true }
 lz4_flex = { workspace = true }
 paste = { workspace = true }
 rand = { workspace = true, features = ["default"] }
@@ -90,8 +90,6 @@ zenoh-task = { workspace = true }
 
 
 [dev-dependencies]
-futures-util = { workspace = true }
-zenoh-util = {workspace = true }
+zenoh-util = { workspace = true }
 zenoh-protocol = { workspace = true, features = ["test"] }
-futures = { workspace = true }
 zenoh-link-commons = { workspace = true }

io/zenoh-transport/src/common/batch.rs

@@ -202,7 +202,7 @@ pub struct WBatch {
     #[cfg(feature = "stats")]
     pub stats: WBatchStats,
     // an ephemeral batch will not be recycled in the pipeline
-    // it can be used to push a stop fragment when no batch are available
+    // it can be used to push a drop fragment when no batch are available
     pub ephemeral: bool,
 }
 
@@ -212,9 +212,9 @@ impl WBatch {
             buffer: BBuf::with_capacity(config.mtu as usize),
             codec: Zenoh080Batch::new(),
             config,
-            ephemeral: false,
             #[cfg(feature = "stats")]
             stats: WBatchStats::default(),
+            ephemeral: false,
         };
 
         // Bring the batch in a clear state