- Optimize bufer reclamation

- Optimize memory fragmentation
- Optimize main read loop
- Optimnize zbuf reader

Author: yellowhatter

commons/zenoh-buffers/src/zbuf.rs

@@ -95,6 +95,17 @@ impl Buffer for ZBuf {
             .iter()
             .fold(0, |len, slice| len + slice.len())
     }
+
+    #[inline(always)]
+    fn is_empty(&self) -> bool {
+        // optimize compared to default implementation by avoiding the walkthouh
+        for slice in self.slices.as_ref() {
+            if !slice.is_empty() {
+                return false;
+            }
+        }
+        true
+    }
 }
 
 // SplitBuffer

commons/zenoh-uring/src/linux/api/reader/mod.rs

@@ -88,6 +88,8 @@ impl Reader {
     }
 
     pub fn new(batch_size: usize, batch_count: BufferCount) -> ZResult<Self> {
+        let batch_size = batch_size + batch_size / 2; // add some headroom to reduce ENOBUFS errors
+
         // create eventfd to wake io_uring on demand by producing read events
         let waker = Arc::new(nix::sys::eventfd::EventFd::from_value_and_flags(
             0,
@@ -184,17 +186,12 @@ impl Reader {
             }
 
             loop {
+                #[cfg(feature = "uring_trace")]
+                let mut i = 0;
+
                 while let Some(e) = unsafe { ring.completion_shared() }.next() {
                     let mut sq = unsafe { ring.submission_shared() };
 
-                    roll_cmds(
-                        &receiver,
-                        &mut context_storage,
-                        &arena,
-                        &mut sq,
-                        batch_count,
-                    )?;
-
                     match e.user_data() {
                         IndexGeneration::INVALID_MIN => {
                             tracing::debug!("Zero-user-data entry: {:?}", e);
@@ -205,9 +202,16 @@ impl Reader {
                             unsafe { sq.push(&waker_read)? };
                         }
                         index => {
+                            #[cfg(feature = "uring_trace")]
+                            {
+                                i += 1;
+                            }
+
                             let index = unsafe { IndexGeneration::new_unchecked(index) };
-                            if Reader::multi(&context_storage, &e, index, &arena, &mut sq)? {
-                                let len = sq.len() as u32;
+                            let to_submit =
+                                Reader::multi(&context_storage, &e, index, &arena, &mut sq)?;
+                            let len = sq.len() as u32;
+                            if to_submit || len >= (batch_count / 2) as u32 {
                                 drop(sq);
                                 //ring.submit()?;
                                 unsafe {
@@ -223,6 +227,9 @@ impl Reader {
                     }
                 }
 
+                #[cfg(feature = "uring_trace")]
+                tracing::info!("Processed {} completion entries", i);
+
                 // receive external submissions
                 let mut sq = unsafe { ring.submission_shared() };
                 roll_cmds(
@@ -354,16 +361,17 @@ impl Reader {
     ) -> ZResult<bool> {
         let mut need_submit = false;
         if e.result() < 0 {
-            tracing::trace!("Error entry: {:?}", e);
+            tracing::debug!("Error entry: {:?}", e);
 
             match e.result().neg() {
                 libc::ENOBUFS => {
                     // We are out of buffers
                     tracing::debug!("ENOBUFS: Restart multishot receive for task {:?}", index);
 
-                    let recv = opcode::RecvMulti::new(types::Fd(context.fd), context.buffer_group().id())
+                    let recv =
+                        opcode::RecvMulti::new(types::Fd(context.fd), context.buffer_group().id())
                             .build()
-                            //.flags(io_uring::squeue::Flags::ASYNC)
+                            .flags(io_uring::squeue::Flags::ASYNC)
                             .user_data(index.into());
 
                     unsafe { sq.push(&recv)? };
@@ -393,7 +401,7 @@ impl Reader {
                             context.buffer_group().id(),
                         )
                         .build()
-                        //.flags(io_uring::squeue::Flags::ASYNC)
+                        .flags(io_uring::squeue::Flags::ASYNC)
                         .user_data(index.into());
 
                         unsafe { sq.push(&recv)? };

commons/zenoh-uring/src/linux/reader/buffer_group.rs

@@ -139,9 +139,20 @@ impl BufferGroup {
         buf_len: usize,
         sq: &mut SubmissionQueue<'_>,
     ) -> ZResult<RxBuffer> {
+        let count = self
+            .buffers_missing
+            .load(std::sync::atomic::Ordering::Relaxed)
+            + 1;
+        let leftover = self
+            .arena
+            .arena
+            .provide_batches_to_group(self.id, count, sq)?;
         self.buffers_missing
-            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        self.ensure_batches_to_ring(sq)?;
+            .store(leftover, std::sync::atomic::Ordering::Relaxed);
+
+        //self.buffers_missing
+        //    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        //self.ensure_batches_to_ring(sq)?;
 
         let data = &mut unsafe {
             self.arena

commons/zenoh-uring/src/linux/reader/reservable_arena.rs

@@ -14,11 +14,14 @@
 
 use std::sync::Arc;
 
+use io_uring::{opcode, squeue::Flags, SubmissionQueue};
+use zenoh_result::ZResult;
+
 use crate::{
     api::types::BufferCount,
     batch_arena::{BatchArena, Batches},
     reader::submission::SubmissionIface,
-    types::BufferId,
+    types::{BufferGroupId, BufferId},
 };
 
 pub(crate) struct ReservableArenaInner {
@@ -54,6 +57,70 @@ impl ReservableArenaInner {
         self.arena.batch_size()
     }
 
+    pub fn provide_batches_to_group(
+        &self,
+        group_id: BufferGroupId,
+        mut count: BufferCount,
+        sq: &mut SubmissionQueue<'_>,
+    ) -> ZResult<BufferCount> {
+        // recycle batches from the recycled_batches queue first
+        while let Some(buf_id) = self.recycled_batches.pop() {
+            let data = unsafe { self.arena.index_mut_unchecked(buf_id as usize) };
+
+            let entry = opcode::ProvideBuffers::new(
+                data.as_mut_ptr(),
+                self.arena.batch_size() as i32,
+                1,
+                group_id,
+                buf_id as BufferId,
+            )
+            .build()
+            .flags(Flags::SKIP_SUCCESS);
+
+            unsafe {
+                sq.push(&entry)?;
+            }
+
+            count -= 1;
+
+            if count == 0 {
+                break;
+            }
+        }
+
+        // allocate more memory if needed
+        if count > 0 {
+            if let Some(additional_batches) = self.arena.allocate_more_batches() {
+                let (primary, to_recycle) =
+                    additional_batches.split(count as BufferCount, self.arena.batch_size());
+
+                // push the primary batch to the result
+                let entry = opcode::ProvideBuffers::new(
+                    primary.addr,
+                    self.arena.batch_size() as i32,
+                    primary.nbufs,
+                    group_id,
+                    primary.start_bid as BufferId,
+                )
+                .build()
+                .flags(Flags::SKIP_SUCCESS);
+
+                unsafe {
+                    sq.push(&entry)?;
+                }
+
+                // recycle the leftover batches
+                if let Some(to_recycle) = to_recycle {
+                    for buf_id in to_recycle.start_bid..to_recycle.start_bid + to_recycle.nbufs {
+                        self.recycle_batch(buf_id);
+                    }
+                }
+            }
+        }
+
+        Ok(count)
+    }
+
     pub fn pop_batches(&self, count: BufferCount) -> Vec<Batches> {
         let mut result = Vec::with_capacity(count as usize);