Add FilterCtx::io() (#712)

Author: fafhrd91

ntex-io/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.3.2] - 2026-01-08
+
+* Add FilterCtx::io()
+
 ## [3.3.1] - 2025-12-18
 
 * Add IoTaskStatus::ready() helper method

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "3.3.1"
+version = "3.3.2"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for abstracting io streams"
 keywords = ["network", "framework", "async", "futures"]
@@ -27,5 +27,5 @@ log = { workspace = true }
 pin-project-lite = { workspace = true }
 
 [dev-dependencies]
-ntex = "3.0.0-pre.8"
+ntex = "3.0.0-pre.10"
 rand = { workspace = true }

ntex-io/src/buf.rs

@@ -247,11 +247,19 @@ impl<'a> FilterCtx<'a> {
     }
 
     #[inline]
+    /// Get io
+    pub fn io(&self) -> &IoRef {
+        self.io
+    }
+
+    #[inline]
+    /// Get io tag
     pub fn tag(&self) -> &'static str {
         self.io.tag()
     }
 
     #[inline]
+    /// Get filter ctx for next filter in chain
     pub fn next(&self) -> Self {
         Self {
             io: self.io,
@@ -261,6 +269,7 @@ impl<'a> FilterCtx<'a> {
     }
 
     #[inline]
+    /// Get current read buffer
     pub fn read_buf<F, R>(&self, nbytes: usize, f: F) -> R
     where
         F: FnOnce(&ReadBuf<'_>) -> R,
@@ -278,6 +287,7 @@ impl<'a> FilterCtx<'a> {
     }
 
     #[inline]
+    /// Get current write buffer
     pub fn write_buf<F, R>(&self, f: F) -> R
     where
         F: FnOnce(&WriteBuf<'_>) -> R,

ntex-net/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.5.1] - 2026-01-08
+
+* Remove changes queue checks for io-uring driver
+
 ## [3.5.0] - 2026-01-03
 
 * Refactor io driver

ntex-net/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-net"
-version = "3.5.0"
+version = "3.5.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntexwork utils for ntex framework"
 keywords = ["network", "framework", "async", "futures"]

ntex-net/src/polling/connect.rs

@@ -163,16 +163,18 @@ impl Handler for ConnectOpsBatcher {
 
     fn error(&mut self, id: usize, err: io::Error) {
         let mut connects = self.inner.connects.borrow_mut();
-        log::trace!(
-            "Connect {id:?} is failed {err:?}, has-con: {}",
-            connects.contains(id)
-        );
 
         if connects.contains(id) {
-            let Item { sock, sender, .. } = connects.remove(id);
+            let Item {
+                sock, sender, cfg, ..
+            } = connects.remove(id);
+            log::trace!("{}: Connect {id:?} is failed {err:?}", cfg.tag());
+
             let _ = sender.send(Err(err));
             self.inner.api.detach(sock.as_raw_fd(), id as u32);
             crate::helpers::close_socket(sock);
+        } else {
+            log::error!("Connect {id:?} is failed {err:?}");
         }
     }
 

ntex-net/src/uring/driver.rs

@@ -1,8 +1,6 @@
+use std::cell::{Cell, UnsafeCell};
 use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
-use std::{
-    cell::Cell, cell::RefCell, cmp, collections::VecDeque, fmt, io, mem, net, rc::Rc,
-    sync::Arc,
-};
+use std::{cmp, collections::VecDeque, fmt, io, mem, net, ptr, rc::Rc, sync::Arc};
 
 #[cfg(unix)]
 use std::os::unix::net::UnixStream as OsUnixStream;
@@ -49,14 +47,15 @@ impl DriverApi {
     where
         F: FnOnce(&mut SEntry),
     {
-        let mut changes = self.inner.changes.borrow_mut();
-        let sq = self.inner.ring.submission();
-        if !changes.is_empty() || sq.is_full() {
-            let mut entry = Default::default();
-            f(&mut entry);
-            changes.push_back(entry);
-        } else {
-            unsafe {
+        unsafe {
+            let changes = &mut *self.inner.changes.get();
+            let sq = self.inner.ring.submission();
+            if !changes.is_empty() || sq.is_full() {
+                changes.push_back(mem::MaybeUninit::uninit());
+                let entry = changes.back_mut().unwrap();
+                ptr::write_bytes(entry.as_mut_ptr(), 0, 1);
+                f(entry.assume_init_mut());
+            } else {
                 sq.push_inline(f).expect("Queue size is checked");
             }
         }
@@ -135,7 +134,7 @@ struct DriverInner {
     probe: Probe,
     flags: Cell<Flags>,
     ring: IoUring<SEntry, CEntry>,
-    changes: RefCell<VecDeque<SEntry>>,
+    changes: UnsafeCell<VecDeque<mem::MaybeUninit<SEntry>>>,
 }
 
 impl Driver {
@@ -189,7 +188,7 @@ impl Driver {
             ring,
             probe,
             flags: Cell::new(if new { Flags::NEW } else { Flags::empty() }),
-            changes: RefCell::new(VecDeque::with_capacity(32)),
+            changes: UnsafeCell::new(VecDeque::with_capacity(32)),
         });
 
         Ok(Self {
@@ -225,24 +224,35 @@ impl Driver {
     }
 
     fn apply_changes(&self, sq: SubmissionQueue<'_, SEntry>) -> bool {
-        let mut changes = self.inner.changes.borrow_mut();
-        if changes.is_empty() {
-            false
-        } else {
-            let num = cmp::min(changes.len(), sq.capacity() - sq.len());
-            let (s1, s2) = changes.as_slices();
-            let s1_num = cmp::min(s1.len(), num);
-            if s1_num > 0 {
-                unsafe { sq.push_multiple(&s1[0..s1_num]) }.unwrap();
-            } else if !s2.is_empty() {
-                let s2_num = cmp::min(s2.len(), num - s1_num);
-                if s2_num > 0 {
-                    unsafe { sq.push_multiple(&s2[0..s2_num]) }.unwrap();
+        unsafe {
+            let changes = &mut *self.inner.changes.get();
+            if changes.is_empty() {
+                false
+            } else {
+                let num = cmp::min(changes.len(), sq.capacity() - sq.len());
+                let (s1, s2) = changes.as_slices();
+                let s1_num = cmp::min(s1.len(), num);
+                if s1_num > 0 {
+                    // safety: "changes" contains only initialized entries
+                    sq.push_multiple(mem::transmute::<
+                        &[mem::MaybeUninit<SEntry>],
+                        &[SEntry],
+                    >(&s1[0..s1_num]))
+                        .unwrap();
+                } else if !s2.is_empty() {
+                    let s2_num = cmp::min(s2.len(), num - s1_num);
+                    if s2_num > 0 {
+                        sq.push_multiple(mem::transmute::<
+                            &[mem::MaybeUninit<SEntry>],
+                            &[SEntry],
+                        >(&s2[0..s2_num]))
+                            .unwrap();
+                    }
                 }
-            }
-            changes.drain(0..num);
+                changes.drain(0..num);
 
-            !changes.is_empty()
+                !changes.is_empty()
+            }
         }
     }
 

ntex-net/src/uring/stream.rs

@@ -137,7 +137,7 @@ impl StreamOps {
     }
 
     pub(crate) fn register(
-        &self,
+        self,
         io: Socket,
         ctx: IoContext,
         zc: bool,
@@ -150,7 +150,7 @@ impl StreamOps {
             flags: if zc { self.0.default_flags } else { Flags::NO_ZC },
         };
 
-        let id = self.0.with(move |st| {
+        let id = self.0.with(|st| {
             // handle RDHUP event
             let op = opcode::PollAdd::new(item.fd(), libc::POLLRDHUP as u32).build();
             let id = st.streams.insert(item);

ntex-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-rt"
-version = "3.4.0"
+version = "3.4.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex runtime"
 keywords = ["network", "framework", "async", "futures"]

ntex-rt/src/rt.rs

@@ -212,8 +212,6 @@ impl RunnableQueue {
     }
 
     fn run(&self) -> bool {
-        self.idle.set(false);
-
         for _ in 0..self.event_interval {
             let task = unsafe { (*self.local_queue.get()).pop_front() };
             if let Some(task) = task {
@@ -238,11 +236,15 @@ impl RunnableQueue {
             }
             break;
         }
-        self.idle.set(true);
 
-        !unsafe { (*self.local_queue.get()).is_empty() }
+        let more_tasks = !unsafe { (*self.local_queue.get()).is_empty() }
             || !self.sync_fixed_queue.is_empty()
-            || !self.sync_queue.is_empty()
+            || !self.sync_queue.is_empty();
+
+        if !more_tasks {
+            self.idle.set(true);
+        }
+        more_tasks
     }
 
     fn clear(&self) {