Cleanup and improvements for neon driver (#700)

Author: fafhrd91

ntex-io/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.3.1] - 2025-12-18
+
+* Add IoTaskStatus::ready() helper method
+
 ## [3.3.0] - 2025-12-17
 
 * Upgrade to ntex-service v4

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "3.3.0"
+version = "3.3.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for abstracting io streams"
 keywords = ["network", "framework", "async", "futures"]

ntex-io/src/lib.rs

@@ -122,6 +122,14 @@ pub enum IoTaskStatus {
     Stop,
 }
 
+impl IoTaskStatus {
+    #[inline]
+    /// Ready for more io ops
+    pub fn ready(self) -> bool {
+        self == IoTaskStatus::Io
+    }
+}
+
 /// Io status
 #[derive(Debug)]
 pub enum IoStatusUpdate {

ntex-net/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.4.1] - 2025-12-18
+
+* Cleanup and improvements for polling driver
+
 ## [3.4.0] - 2025-12-17
 
 * Upgrade to ntex-service v4

ntex-net/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-net"
-version = "3.4.0"
+version = "3.4.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntexwork utils for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -35,7 +35,7 @@ io-uring-compat = []
 ntex-service = "4"
 ntex-bytes = "1"
 ntex-http = "1"
-ntex-io = "3"
+ntex-io = "3.3.1"
 ntex-rt = "3.3"
 ntex-util = "3"
 

ntex-net/src/helpers.rs

@@ -1,15 +1,7 @@
-use std::{error, io, net, net::SocketAddr, path::Path};
+use std::{io, net, net::SocketAddr, path::Path};
 
-use ntex_util::channel::oneshot::channel;
 use socket2::{Domain, Protocol, SockAddr, Socket, Type};
 
-pub(crate) fn pool_io_err<T, E>(result: std::result::Result<T, E>) -> io::Result<T>
-where
-    E: error::Error + Send + Sync + 'static,
-{
-    result.map_err(|e| io::Error::other(e))
-}
-
 pub(crate) async fn connect(addr: SocketAddr) -> io::Result<Socket> {
     let addr = SockAddr::from(addr);
     let domain = addr.domain();
@@ -28,17 +20,9 @@ async fn connect_inner(
     protocol: Option<Protocol>,
 ) -> io::Result<Socket> {
     let sock = prep_socket(Socket::new(domain, ty, protocol)?)?;
-
-    let (sender, rx) = channel();
-
-    crate::rt_impl::connect::ConnectOps::current().connect(sock, addr, sender)?;
-
-    let sock = rx
+    crate::rt_impl::connect::ConnectOps::current()
+        .connect(sock, addr)
         .await
-        .map_err(|_| io::Error::other("IO Driver is gone"))
-        .and_then(|item| item)?;
-
-    Ok(sock)
 }
 
 pub(crate) fn prep_socket(sock: Socket) -> io::Result<Socket> {

ntex-net/src/lib.rs

@@ -1,6 +1,5 @@
 //! Utility for async runtime abstraction
 #![deny(rust_2018_idioms, unreachable_pub, missing_debug_implementations)]
-#![allow(unused_variables, dead_code)]
 
 mod compat;
 pub mod connect;

ntex-net/src/rt_polling/connect.rs

@@ -2,19 +2,13 @@ use std::{cell::RefCell, io, os::fd::AsRawFd, os::fd::RawFd, rc::Rc};
 
 use ntex_neon::driver::{DriverApi, Event, Handler};
 use ntex_neon::{Runtime, syscall};
-use ntex_util::channel::oneshot::Sender;
+use ntex_util::channel::oneshot::{Sender, channel};
 use slab::Slab;
 use socket2::{SockAddr, Socket};
 
 #[derive(Clone)]
 pub(crate) struct ConnectOps(Rc<ConnectOpsInner>);
 
-#[derive(Debug)]
-enum Change {
-    Event(Event),
-    Error(io::Error),
-}
-
 struct ConnectOpsBatcher {
     inner: Rc<ConnectOpsInner>,
 }
@@ -52,21 +46,25 @@ impl ConnectOps {
         })
     }
 
-    pub(crate) fn connect(
-        &self,
-        sock: Socket,
-        addr: SockAddr,
-        sender: Sender<io::Result<Socket>>,
-    ) -> io::Result<usize> {
+    pub(crate) async fn connect(&self, sock: Socket, addr: SockAddr) -> io::Result<Socket> {
         let result = syscall!(
             break libc::connect(sock.as_raw_fd(), addr.as_ptr().cast(), addr.len())
         )?;
-        let fd = sock.as_raw_fd();
-        let item = Item { sock, sender };
-        let id = self.0.connects.borrow_mut().insert(item);
-        self.0.api.attach(fd, id as u32, Event::writable(0));
-
-        Ok(id)
+        if result.is_ready() {
+            // socket is connected
+            Ok(sock)
+        } else {
+            // connect is async
+            let (sender, rx) = channel();
+            let fd = sock.as_raw_fd();
+            let item = Item { sock, sender };
+            let id = self.0.connects.borrow_mut().insert(item);
+            self.0.api.attach(fd, id as u32, Event::writable(0));
+
+            rx.await
+                .map_err(|_| io::Error::other("IO Driver is gone"))
+                .and_then(|item| item)
+        }
     }
 }
 

ntex-net/src/rt_polling/driver.rs

@@ -50,8 +50,6 @@ struct StreamOpsInner {
     delayed_drop: Cell<bool>,
     delayed_feed: Cell<Option<Vec<IdType>>>,
     streams: Cell<Option<Box<Slab<StreamItem>>>>,
-    lw: usize,
-    hw: usize,
 }
 
 impl StreamOps {
@@ -65,8 +63,6 @@ impl StreamOps {
                     delayed_drop: Cell::new(false),
                     delayed_feed: Cell::new(Some(Vec::new())),
                     streams: Cell::new(Some(Box::new(Slab::new()))),
-                    lw: 1024,
-                    hw: 1024 * 16,
                 });
                 inner = Some(ops.clone());
                 Box::new(StreamOpsHandler { inner: ops })
@@ -131,21 +127,18 @@ impl Handler for StreamOpsHandler {
             log::trace!("{}: Event ({:?}): {ev:?} {:?}", io.tag(), io.fd(), io.flags);
 
             if ev.readable {
-                match io.read(id as u32, &self.inner.api, self.inner.lw, self.inner.hw) {
-                    IoTaskStatus::Io => {
-                        renew.readable = true;
-                        io.flags.insert(Flags::RD);
-                    }
-                    IoTaskStatus::Pause | IoTaskStatus::Stop => {
-                        io.flags.remove(Flags::RD);
-                    }
+                if io.read().ready() {
+                    renew.readable = true;
+                    io.flags.insert(Flags::RD);
+                } else {
+                    io.flags.remove(Flags::RD);
                 }
             } else if io.flags.contains(Flags::RD) {
                 renew.readable = true;
             }
 
             if ev.writable {
-                if io.write(id as u32, &self.inner.api) == IoTaskStatus::Io {
+                if io.write().ready() {
                     renew.writable = true;
                     io.flags.insert(Flags::WR);
                 } else {
@@ -248,7 +241,6 @@ impl StreamOpsInner {
         if let Some(mut streams) = self.streams.take() {
             let idx = id as usize;
             let item = &mut streams[idx];
-            let fd = item.fd();
 
             if item.flags.contains(Flags::DROPPED_PRI) {
                 let item = streams.remove(idx);
@@ -298,7 +290,7 @@ impl StreamCtl {
             })
             .await
             .map_err(io::Error::other)
-            .and_then(crate::helpers::pool_io_err)
+            .and_then(|res| res.map_err(io::Error::other))
     }
 
     /// Modify poll interest for the stream
@@ -318,9 +310,7 @@ impl StreamCtl {
                 if io.flags.contains(Flags::RD) {
                     event.readable = true;
                     want_update_read = false;
-                } else if io.read(self.id, &self.inner.api, self.inner.lw, self.inner.hw)
-                    == IoTaskStatus::Io
-                {
+                } else if io.read().ready() {
                     event.readable = true;
                     io.flags.insert(Flags::RD);
                 } else {
@@ -337,7 +327,7 @@ impl StreamCtl {
                 if io.flags.contains(Flags::WR) {
                     event.writable = true;
                     want_update_write = false;
-                } else if io.write(self.id, &self.inner.api) == IoTaskStatus::Io {
+                } else if io.write().ready() {
                     event.writable = true;
                     io.flags.insert(Flags::WR);
                 } else {
@@ -393,7 +383,7 @@ impl StreamItem {
         self.ctx.tag()
     }
 
-    fn write(&mut self, id: u32, api: &DriverApi) -> IoTaskStatus {
+    fn write(&mut self) -> IoTaskStatus {
         if let Some(buf) = self.ctx.get_write_buf() {
             let fd = self.fd();
             log::trace!("{}: Write ({fd:?}), buf: {:?}", self.ctx.tag(), buf.len());
@@ -403,7 +393,7 @@ impl StreamItem {
         IoTaskStatus::Pause
     }
 
-    fn read(&mut self, id: u32, api: &DriverApi, lw: usize, hw: usize) -> IoTaskStatus {
+    fn read(&mut self) -> IoTaskStatus {
         let mut buf = self.ctx.get_read_buf();
 
         let fd = self.fd();

ntex-net/src/rt_uring/connect.rs

@@ -2,20 +2,13 @@ use std::{cell::RefCell, io, os::fd::AsRawFd, rc::Rc};
 
 use ntex_neon::driver::io_uring::{opcode, types::Fd};
 use ntex_neon::{Runtime, driver::DriverApi, driver::Handler};
-use ntex_util::channel::oneshot::Sender;
+use ntex_util::channel::oneshot::{Sender, channel};
 use slab::Slab;
 use socket2::{SockAddr, Socket};
 
 #[derive(Clone)]
 pub(crate) struct ConnectOps(Rc<ConnectOpsInner>);
 
-#[derive(Debug)]
-enum Change {
-    Readable,
-    Writable,
-    Error(io::Error),
-}
-
 struct ConnectOpsHandler {
     inner: Rc<ConnectOpsInner>,
 }
@@ -47,27 +40,26 @@ impl ConnectOps {
         })
     }
 
-    pub(crate) fn connect(
-        &self,
-        sock: Socket,
-        addr: SockAddr,
-        sender: Sender<io::Result<Socket>>,
-    ) -> io::Result<()> {
-        let addr2 = addr.clone();
-        let mut ops = self.0.ops.borrow_mut();
-
-        // addr must be stable, neon submits ops at the end of rt turn
-        let addr = Box::new(addr);
-        let (addr_ptr, addr_len) = (addr.as_ref().as_ptr().cast(), addr.len());
-
-        let fd = sock.as_raw_fd();
-        let id = ops.insert((addr, sock, sender));
-        self.0.api.submit(
-            id as u32,
-            opcode::Connect::new(Fd(fd), addr_ptr, addr_len).build(),
-        );
-
-        Ok(())
+    pub(crate) async fn connect(&self, sock: Socket, addr: SockAddr) -> io::Result<Socket> {
+        {
+            let mut ops = self.0.ops.borrow_mut();
+
+            // addr must be stable, neon submits ops at the end of rt turn
+            let addr = Box::new(addr);
+            let (addr_ptr, addr_len) = (addr.as_ref().as_ptr().cast(), addr.len());
+
+            let (sender, rx) = channel();
+            let fd = sock.as_raw_fd();
+            let id = ops.insert((addr, sock, sender));
+            self.0.api.submit(
+                id as u32,
+                opcode::Connect::new(Fd(fd), addr_ptr, addr_len).build(),
+            );
+            rx
+        }
+        .await
+        .map_err(|_| io::Error::other("IO Driver is gone"))
+        .and_then(|item| item)
     }
 }
 
@@ -77,7 +69,7 @@ impl Handler for ConnectOpsHandler {
         self.inner.ops.borrow_mut().remove(user_data);
     }
 
-    fn completed(&mut self, user_data: usize, flags: u32, result: io::Result<usize>) {
+    fn completed(&mut self, user_data: usize, _: u32, result: io::Result<usize>) {
         let (addr, sock, tx) = self.inner.ops.borrow_mut().remove(user_data);
         log::trace!(
             "connect-op is completed {:?} result: {:?}, addr: {:?}",