wip

ntex-io/src/tasks.rs

@@ -447,6 +447,21 @@ impl IoContext {
         }
     }
 
+    /// Get read buffer
+    pub fn get_read_buf(&self) -> BytesVec {
+        let inner = &self.0 .0;
+        if inner.flags.get().is_read_buf_ready() {
+            // read buffer is still not read by dispatcher
+            // we cannot touch it
+            inner.pool.get().get_read_buf()
+        } else {
+            inner
+                .buffer
+                .get_read_source()
+                .unwrap_or_else(|| inner.pool.get().get_read_buf())
+        }
+    }
+
     /// Get read buffer
     pub fn with_read_buf<F>(&self, f: F) -> Poll<()>
     where

ntex-neon/src/driver/uring/mod.rs

@@ -41,7 +41,7 @@ pub trait OpCode {
 #[derive(Debug)]
 enum Change {
     Submit { entry: SEntry },
-    Cancel { user_data: u64 },
+    Cancel { user_data: u64, op_id: u64 },
 }
 
 pub struct DriverApi {
@@ -57,25 +57,22 @@ impl DriverApi {
             user_data,
             entry,
         );
-        self.change(Change::Submit {
+        self.changes.borrow_mut().push_back(Change::Submit {
             entry: entry.user_data(user_data as u64 | self.batch),
         });
     }
 
-    pub fn cancel(&self, user_data: u32) {
+    pub fn cancel(&self, user_data: u32, op_id: u32) {
         log::debug!(
             "Cancel operation batch: {:?} user-data: {:?}",
             self.batch,
             user_data
         );
-        self.change(Change::Cancel {
+        self.changes.borrow_mut().push_back(Change::Cancel {
+            op_id: op_id as u64 | self.batch,
             user_data: user_data as u64 | self.batch,
         });
     }
-
-    fn change(&self, change: Change) {
-        self.changes.borrow_mut().push_back(change);
-    }
 }
 
 /// Low-level driver of io-uring.
@@ -91,8 +88,7 @@ pub(crate) struct Driver {
 }
 
 impl Driver {
-    const CANCEL: u64 = u64::MAX;
-    const NOTIFY: u64 = u64::MAX - 1;
+    const NOTIFY: u64 = u64::MAX;
     const BATCH_MASK: u64 = 0xFFFF << 48;
     const DATA_MASK: u64 = 0xFFFF >> 16;
 
@@ -200,10 +196,10 @@ impl Driver {
                         break;
                     }
                 }
-                Change::Cancel { user_data } => {
-                    let entry = AsyncCancel::new(user_data).build().user_data(Self::CANCEL);
+                Change::Cancel { user_data, op_id } => {
+                    let entry = AsyncCancel::new(op_id).build().user_data(user_data);
                     if unsafe { squeue.push(&entry.user_data(user_data)) }.is_err() {
-                        changes.push_front(Change::Cancel { user_data });
+                        changes.push_front(Change::Cancel { user_data, op_id });
                         break;
                     }
                 }
@@ -263,7 +259,6 @@ impl Driver {
         for entry in cqueue {
             let user_data = entry.user_data();
             match user_data {
-                Self::CANCEL => {}
                 Self::NOTIFY => {
                     let flags = entry.flags();
                     debug_assert!(more(flags));
@@ -288,9 +283,6 @@ impl Driver {
                 }
             }
         }
-        for handler in handlers.iter_mut() {
-            handler.commit();
-        }
         self.handlers.set(Some(handlers));
         true
     }

ntex-neon/src/driver/uring/op.rs

@@ -8,9 +8,6 @@ use crate::{driver::op::*, syscall};
 pub trait Handler {
     /// Operation is completed
     fn completed(&mut self, user_data: usize, flags: u32, result: io::Result<i32>);
-
-    /// All events are processed, process all updates
-    fn commit(&mut self);
 }
 
 impl<D, F> OpCode for Asyncify<F, D>

ntex-net/Cargo.toml

@@ -28,7 +28,7 @@ compio = ["ntex-rt/compio", "ntex-compio"]
 neon = ["ntex-rt/neon", "ntex-neon/polling", "slab", "socket2"]
 
 # neon io-uring runtime
-neon-uring = ["ntex-rt/neon", "ntex-neon/io-uring", "slab", "socket2"]
+neon-uring = ["ntex-rt/neon", "ntex-neon/io-uring", "io-uring", "slab", "socket2"]
 
 [dependencies]
 ntex-service = "3.3"
@@ -49,6 +49,10 @@ thiserror = { workspace = true }
 slab = { workspace = true, optional = true }
 socket2 = { workspace = true, optional = true }
 
+# Linux specific dependencies
+[target.'cfg(target_os = "linux")'.dependencies]
+io-uring = { workspace = true, optional = true }
+
 [dev-dependencies]
 ntex = "2"
 env_logger = "0.11"

ntex-net/src/compat.rs

@@ -17,6 +17,17 @@ pub use crate::rt_polling::{
     unix_connect_in,
 };
 
+#[cfg(all(
+    feature = "neon-uring",
+    not(feature = "neon"),
+    not(feature = "tokio"),
+    not(feature = "compio")
+))]
+pub use crate::rt_uring::{
+    from_tcp_stream, from_unix_stream, tcp_connect, tcp_connect_in, unix_connect,
+    unix_connect_in,
+};
+
 #[cfg(all(
     feature = "compio",
     not(feature = "tokio"),
@@ -104,5 +115,10 @@ mod no_rt {
     }
 }
 
-#[cfg(all(not(feature = "tokio"), not(feature = "compio"), not(feature = "neon")))]
+#[cfg(all(
+    not(feature = "tokio"),
+    not(feature = "compio"),
+    not(feature = "neon"),
+    not(feature = "neon-uring")
+))]
 pub use no_rt::*;

ntex-net/src/lib.rs

@@ -1,5 +1,6 @@
 //! Utility for async runtime abstraction
 #![deny(rust_2018_idioms, unreachable_pub, missing_debug_implementations)]
+#![allow(unused_variables, dead_code)]
 
 mod compat;
 pub mod connect;
@@ -9,5 +10,18 @@ pub use ntex_rt::{spawn, spawn_blocking};
 
 pub use self::compat::*;
 
-#[cfg(all(feature = "neon", not(feature = "tokio"), not(feature = "compio")))]
+#[cfg(all(
+    feature = "neon",
+    not(feature = "neon-uring"),
+    not(feature = "tokio"),
+    not(feature = "compio")
+))]
 mod rt_polling;
+
+#[cfg(all(
+    feature = "neon-uring",
+    not(feature = "neon"),
+    not(feature = "tokio"),
+    not(feature = "compio")
+))]
+mod rt_uring;

ntex-net/src/rt_polling/driver.rs

@@ -19,10 +19,10 @@ bitflags::bitflags! {
 
 pub(crate) struct StreamCtl<T> {
     id: usize,
-    inner: Rc<CompioOpsInner<T>>,
+    inner: Rc<StreamOpsInner<T>>,
 }
 
-struct TcpStreamItem<T> {
+struct StreamItem<T> {
     io: Option<T>,
     fd: RawFd,
     context: IoContext,
@@ -30,7 +30,7 @@ struct TcpStreamItem<T> {
     ref_count: usize,
 }
 
-pub(crate) struct CompioOps<T>(Rc<CompioOpsInner<T>>);
+pub(crate) struct StreamOps<T>(Rc<StreamOpsInner<T>>);
 
 #[derive(Debug)]
 enum Change {
@@ -39,18 +39,18 @@ enum Change {
     Error(io::Error),
 }
 
-struct CompioOpsBatcher<T> {
+struct StreamOpsHandler<T> {
     feed: VecDeque<(usize, Change)>,
-    inner: Rc<CompioOpsInner<T>>,
+    inner: Rc<StreamOpsInner<T>>,
 }
 
-struct CompioOpsInner<T> {
+struct StreamOpsInner<T> {
     api: DriverApi,
     feed: Cell<Option<VecDeque<usize>>>,
-    streams: Cell<Option<Box<Slab<TcpStreamItem<T>>>>>,
+    streams: Cell<Option<Box<Slab<StreamItem<T>>>>>,
 }
 
-impl<T: AsRawFd + 'static> CompioOps<T> {
+impl<T: AsRawFd + 'static> StreamOps<T> {
     pub(crate) fn current() -> Self {
         Runtime::with_current(|rt| {
             if let Some(s) = rt.get::<Self>() {
@@ -58,19 +58,19 @@ impl<T: AsRawFd + 'static> CompioOps<T> {
             } else {
                 let mut inner = None;
                 rt.driver().register_handler(|api| {
-                    let ops = Rc::new(CompioOpsInner {
+                    let ops = Rc::new(StreamOpsInner {
                         api,
                         feed: Cell::new(Some(VecDeque::new())),
                         streams: Cell::new(Some(Box::new(Slab::new()))),
                     });
                     inner = Some(ops.clone());
-                    Box::new(CompioOpsBatcher {
+                    Box::new(StreamOpsHandler {
                         inner: ops,
                         feed: VecDeque::new(),
                     })
                 });
 
-                let s = CompioOps(inner.unwrap());
+                let s = StreamOps(inner.unwrap());
                 rt.insert(s.clone());
                 s
             }
@@ -78,7 +78,7 @@ impl<T: AsRawFd + 'static> CompioOps<T> {
     }
 
     pub(crate) fn register(&self, io: T, context: IoContext) -> StreamCtl<T> {
-        let item = TcpStreamItem {
+        let item = StreamItem {
             context,
             fd: io.as_raw_fd(),
             io: Some(io),
@@ -96,7 +96,7 @@ impl<T: AsRawFd + 'static> CompioOps<T> {
 
     fn with<F, R>(&self, f: F) -> R
     where
-        F: FnOnce(&mut Slab<TcpStreamItem<T>>) -> R,
+        F: FnOnce(&mut Slab<StreamItem<T>>) -> R,
     {
         let mut inner = self.0.streams.take().unwrap();
         let result = f(&mut inner);
@@ -105,13 +105,13 @@ impl<T: AsRawFd + 'static> CompioOps<T> {
     }
 }
 
-impl<T> Clone for CompioOps<T> {
+impl<T> Clone for StreamOps<T> {
     fn clone(&self) -> Self {
         Self(self.0.clone())
     }
 }
 
-impl<T> Handler for CompioOpsBatcher<T> {
+impl<T> Handler for StreamOpsHandler<T> {
     fn readable(&mut self, id: usize) {
         log::debug!("FD is readable {:?}", id);
         self.feed.push_back((id, Change::Readable));
@@ -301,7 +301,7 @@ impl<T> StreamCtl<T> {
 
     fn with<F, R>(&self, f: F) -> R
     where
-        F: FnOnce(&mut Slab<TcpStreamItem<T>>) -> R,
+        F: FnOnce(&mut Slab<StreamItem<T>>) -> R,
     {
         let mut inner = self.inner.streams.take().unwrap();
         let result = f(&mut inner);

ntex-net/src/rt_polling/io.rs

@@ -5,13 +5,13 @@ use ntex_io::{
 };
 use ntex_neon::{net::TcpStream, net::UnixStream, spawn};
 
-use super::driver::{Closable, CompioOps, StreamCtl};
+use super::driver::{Closable, StreamCtl, StreamOps};
 
 impl IoStream for super::TcpStream {
     fn start(self, read: ReadContext, _: WriteContext) -> Option<Box<dyn Handle>> {
         let io = self.0;
         let context = read.context();
-        let ctl = CompioOps::current().register(io, context.clone());
+        let ctl = StreamOps::current().register(io, context.clone());
         let ctl2 = ctl.clone();
         spawn(async move { run(ctl, context).await }).detach();
 
@@ -23,7 +23,7 @@ impl IoStream for super::UnixStream {
     fn start(self, read: ReadContext, _: WriteContext) -> Option<Box<dyn Handle>> {
         let io = self.0;
         let context = read.context();
-        let ctl = CompioOps::current().register(io, context.clone());
+        let ctl = StreamOps::current().register(io, context.clone());
         spawn(async move { run(ctl, context).await }).detach();
 
         None

ntex-net/src/rt_polling/mod.rs

@@ -1,4 +1,3 @@
-#![allow(clippy::type_complexity)]
 use std::{io::Result, net, net::SocketAddr};
 
 use ntex_bytes::PoolRef;
@@ -8,10 +7,10 @@ mod connect;
 mod driver;
 mod io;
 
-/// Tcp stream wrapper for compio TcpStream
+/// Tcp stream wrapper for neon TcpStream
 struct TcpStream(ntex_neon::net::TcpStream);
 
-/// Tcp stream wrapper for compio UnixStream
+/// Tcp stream wrapper for neon UnixStream
 struct UnixStream(ntex_neon::net::UnixStream);
 
 /// Opens a TCP connection to a remote host.

ntex-net/src/rt_uring/connect.rs

@@ -0,0 +1,196 @@
+use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6};
+use std::{cell::RefCell, collections::VecDeque, io, path::Path, rc::Rc, task::Poll};
+
+use ntex_neon::driver::op::{Handler, Interest};
+use ntex_neon::driver::{AsRawFd, DriverApi, RawFd};
+use ntex_neon::net::{Socket, TcpStream, UnixStream};
+use ntex_neon::{syscall, Runtime};
+use ntex_util::channel::oneshot::{channel, Sender};
+use slab::Slab;
+use socket2::{Protocol, SockAddr, Type};
+
+pub(crate) async fn connect(addr: SocketAddr) -> io::Result<TcpStream> {
+    let addr = SockAddr::from(addr);
+    let socket = if cfg!(windows) {
+        let bind_addr = if addr.is_ipv4() {
+            SockAddr::from(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0))
+        } else if addr.is_ipv6() {
+            SockAddr::from(SocketAddrV6::new(Ipv6Addr::UNSPECIFIED, 0, 0, 0))
+        } else {
+            return Err(io::Error::new(
+                io::ErrorKind::AddrNotAvailable,
+                "Unsupported address domain.",
+            ));
+        };
+        Socket::bind(&bind_addr, Type::STREAM, Some(Protocol::TCP)).await?
+    } else {
+        Socket::new(addr.domain(), Type::STREAM, Some(Protocol::TCP)).await?
+    };
+
+    let (sender, rx) = channel();
+
+    ConnectOps::current().connect(socket.as_raw_fd(), addr, sender)?;
+
+    rx.await
+        .map_err(|_| io::Error::new(io::ErrorKind::Other, "IO Driver is gone"))
+        .and_then(|item| item)?;
+
+    Ok(TcpStream::from_socket(socket))
+}
+
+pub(crate) async fn connect_unix(path: impl AsRef<Path>) -> io::Result<UnixStream> {
+    let addr = SockAddr::unix(path)?;
+
+    #[cfg(windows)]
+    let socket = {
+        let new_addr = empty_unix_socket();
+        Socket::bind(&new_addr, Type::STREAM, None).await?
+    };
+    #[cfg(unix)]
+    let socket = {
+        use socket2::Domain;
+        Socket::new(Domain::UNIX, Type::STREAM, None).await?
+    };
+
+    let (sender, rx) = channel();
+
+    ConnectOps::current().connect(socket.as_raw_fd(), addr, sender)?;
+
+    rx.await
+        .map_err(|_| io::Error::new(io::ErrorKind::Other, "IO Driver is gone"))
+        .and_then(|item| item)?;
+
+    Ok(UnixStream::from_socket(socket))
+}
+
+#[derive(Clone)]
+pub(crate) struct ConnectOps(Rc<ConnectOpsInner>);
+
+#[derive(Debug)]
+enum Change {
+    Readable,
+    Writable,
+    Error(io::Error),
+}
+
+struct ConnectOpsBatcher {
+    feed: VecDeque<(usize, Change)>,
+    inner: Rc<ConnectOpsInner>,
+}
+
+struct Item {
+    fd: RawFd,
+    sender: Sender<io::Result<()>>,
+}
+
+struct ConnectOpsInner {
+    api: DriverApi,
+    connects: RefCell<Slab<Item>>,
+}
+
+impl ConnectOps {
+    pub(crate) fn current() -> Self {
+        Runtime::with_current(|rt| {
+            if let Some(s) = rt.get::<Self>() {
+                s
+            } else {
+                let mut inner = None;
+                rt.driver().register_handler(|api| {
+                    let ops = Rc::new(ConnectOpsInner {
+                        api,
+                        connects: RefCell::new(Slab::new()),
+                    });
+                    inner = Some(ops.clone());
+                    Box::new(ConnectOpsBatcher {
+                        inner: ops,
+                        feed: VecDeque::new(),
+                    })
+                });
+
+                let s = ConnectOps(inner.unwrap());
+                rt.insert(s.clone());
+                s
+            }
+        })
+    }
+
+    pub(crate) fn connect(
+        &self,
+        fd: RawFd,
+        addr: SockAddr,
+        sender: Sender<io::Result<()>>,
+    ) -> io::Result<usize> {
+        let result = syscall!(break libc::connect(fd, addr.as_ptr(), addr.len()));
+
+        if let Poll::Ready(res) = result {
+            res?;
+        }
+
+        let item = Item { fd, sender };
+        let id = self.0.connects.borrow_mut().insert(item);
+
+        self.0.api.register(fd, id, Interest::Writable);
+
+        Ok(id)
+    }
+}
+
+impl Handler for ConnectOpsBatcher {
+    fn readable(&mut self, id: usize) {
+        log::debug!("ConnectFD is readable {:?}", id);
+        self.feed.push_back((id, Change::Readable));
+    }
+
+    fn writable(&mut self, id: usize) {
+        log::debug!("ConnectFD is writable {:?}", id);
+        self.feed.push_back((id, Change::Writable));
+    }
+
+    fn error(&mut self, id: usize, err: io::Error) {
+        self.feed.push_back((id, Change::Error(err)));
+    }
+
+    fn commit(&mut self) {
+        if self.feed.is_empty() {
+            return;
+        }
+        log::debug!("Commit connect driver changes, num: {:?}", self.feed.len());
+
+        let mut connects = self.inner.connects.borrow_mut();
+
+        for (id, change) in self.feed.drain(..) {
+            if connects.contains(id) {
+                let item = connects.remove(id);
+                match change {
+                    Change::Readable => unreachable!(),
+                    Change::Writable => {
+                        let mut err: libc::c_int = 0;
+                        let mut err_len =
+                            std::mem::size_of::<libc::c_int>() as libc::socklen_t;
+
+                        let res = syscall!(libc::getsockopt(
+                            item.fd.as_raw_fd(),
+                            libc::SOL_SOCKET,
+                            libc::SO_ERROR,
+                            &mut err as *mut _ as *mut _,
+                            &mut err_len
+                        ));
+
+                        let res = if err == 0 {
+                            res.map(|_| ())
+                        } else {
+                            Err(io::Error::from_raw_os_error(err))
+                        };
+
+                        self.inner.api.unregister_all(item.fd);
+                        let _ = item.sender.send(res);
+                    }
+                    Change::Error(err) => {
+                        let _ = item.sender.send(Err(err));
+                        self.inner.api.unregister_all(item.fd);
+                    }
+                }
+            }
+        }
+    }
+}

ntex-net/src/rt_uring/driver.rs

@@ -0,0 +1,279 @@
+use std::{cell::RefCell, collections::VecDeque, fmt, io, num::NonZeroU32, rc::Rc};
+
+use io_uring::{opcode, types::Fd};
+use ntex_neon::driver::op::Handler;
+use ntex_neon::driver::{AsRawFd, DriverApi};
+use ntex_neon::Runtime;
+use ntex_util::channel::oneshot;
+use slab::Slab;
+
+use ntex_bytes::{BufMut, BytesVec};
+use ntex_io::IoContext;
+
+pub(crate) struct StreamCtl<T> {
+    id: usize,
+    inner: Rc<StreamOpsInner<T>>,
+}
+
+struct StreamItem<T> {
+    io: Option<T>,
+    fd: Fd,
+    context: IoContext,
+    ref_count: usize,
+    rd_op: Option<NonZeroU32>,
+    wr_op: Option<NonZeroU32>,
+}
+
+enum Operation {
+    Recv {
+        id: usize,
+        buf: BytesVec,
+        context: IoContext,
+    },
+    Send {
+        id: usize,
+        buf: BytesVec,
+        context: IoContext,
+    },
+    Close {
+        tx: Option<oneshot::Sender<io::Result<()>>>,
+    },
+    Cancel {
+        id: u32,
+    },
+    Nop,
+}
+
+pub(crate) struct StreamOps<T>(Rc<StreamOpsInner<T>>);
+
+struct StreamOpsHandler<T> {
+    inner: Rc<StreamOpsInner<T>>,
+}
+
+struct StreamOpsInner<T> {
+    api: DriverApi,
+    feed: RefCell<VecDeque<usize>>,
+    storage: RefCell<(Slab<StreamItem<T>>, Slab<Operation>)>,
+}
+
+impl<T: AsRawFd + 'static> StreamOps<T> {
+    pub(crate) fn current() -> Self {
+        Runtime::with_current(|rt| {
+            if let Some(s) = rt.get::<Self>() {
+                s
+            } else {
+                let mut inner = None;
+                rt.driver().register_handler(|api| {
+                    let mut storage = Slab::new();
+                    storage.insert(Operation::Nop);
+
+                    let ops = Rc::new(StreamOpsInner {
+                        api,
+                        feed: RefCell::new(VecDeque::new()),
+                        storage: RefCell::new((Slab::new(), Slab::new())),
+                    });
+                    inner = Some(ops.clone());
+                    Box::new(StreamOpsHandler { inner: ops })
+                });
+
+                let s = StreamOps(inner.unwrap());
+                rt.insert(s.clone());
+                s
+            }
+        })
+    }
+
+    pub(crate) fn register(&self, io: T, context: IoContext) -> StreamCtl<T> {
+        let item = StreamItem {
+            context,
+            fd: Fd(io.as_raw_fd()),
+            io: Some(io),
+            ref_count: 1,
+            rd_op: None,
+            wr_op: None,
+        };
+        self.with(|streams| {
+            let id = streams.0.insert(item);
+            StreamCtl {
+                id,
+                inner: self.0.clone(),
+            }
+        })
+    }
+
+    fn with<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(&mut (Slab<StreamItem<T>>, Slab<Operation>)) -> R,
+    {
+        f(&mut *self.0.storage.borrow_mut())
+    }
+}
+
+impl<T> Clone for StreamOps<T> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}
+
+impl<T> Handler for StreamOpsHandler<T> {
+    fn completed(&mut self, user_data: usize, flags: u32, result: io::Result<i32>) {
+        log::debug!("Op is completed {:?} result: {:?}", user_data, result);
+
+        // let mut storage = self.inner.storage.borrow_mut();
+        // for (id, flags, result) in self.feed.drain(..) {}
+
+        // // extra
+        // for id in self.inner.feed.borrow_mut().drain(..) {
+        //     log::debug!("Drop io ({}), {:?}", id, storage.0[id].fd);
+
+        //     storage.0[id].ref_count -= 1;
+        //     if storage.0[id].ref_count == 0 {
+        //         let item = storage.0.remove(id);
+        //         if item.io.is_some() {
+        //             // self.inner.api.unregister_all(item.fd);
+        //         }
+        //     }
+        // }
+    }
+}
+
+impl<T> StreamCtl<T> {
+    pub(crate) async fn close(self) -> io::Result<()> {
+        let result = self.with(|streams| {
+            let item = &mut streams.0[self.id];
+            if let Some(io) = item.io.take() {
+                let (tx, rx) = oneshot::channel();
+                let id = streams.1.insert(Operation::Close { tx: Some(tx) });
+                assert!(id < u32::MAX as usize);
+                self.inner
+                    .api
+                    .submit(id as u32, opcode::Close::new(item.fd).build());
+                Some(rx)
+            } else {
+                None
+            }
+        });
+
+        if let Some(rx) = result {
+            rx.await
+                .map_err(|_| io::Error::new(io::ErrorKind::Other, "gone"))
+                .and_then(|item| item)
+        } else {
+            Ok(())
+        }
+    }
+
+    pub(crate) fn with_io<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(Option<&T>) -> R,
+    {
+        self.with(|streams| f(streams.0[self.id].io.as_ref()))
+    }
+
+    pub(crate) fn resume_read(&self) {
+        self.with(|streams| {
+            let item = &mut streams.0[self.id];
+
+            if item.rd_op.is_none() {
+                log::debug!("Resume io read ({}), {:?}", self.id, item.fd);
+                let mut buf = item.context.get_read_buf();
+                let slice = buf.chunk_mut();
+                let op = opcode::Recv::new(item.fd, slice.as_mut_ptr(), slice.len() as u32)
+                    .build();
+
+                let id = streams.1.insert(Operation::Recv {
+                    buf,
+                    id: self.id,
+                    context: item.context.clone(),
+                });
+                assert!(id < u32::MAX as usize);
+
+                self.inner.api.submit(id as u32, op);
+            }
+        })
+    }
+
+    pub(crate) fn resume_write(&self) {
+        self.with(|streams| {
+            let item = &mut streams.0[self.id];
+
+            if item.wr_op.is_none() {
+                log::debug!("Resume io write ({}), {:?}", self.id, item.fd);
+                //self.inner.api.unregister(item.fd, Interest::Readable);
+            }
+        })
+    }
+
+    pub(crate) fn pause_read(&self) {
+        self.with(|streams| {
+            let item = &mut streams.0[self.id];
+
+            if let Some(rd_op) = item.rd_op {
+                log::debug!("Pause io read ({}), {:?}", self.id, item.fd);
+                let id = streams.1.insert(Operation::Cancel { id: rd_op.get() });
+                assert!(id < u32::MAX as usize);
+                self.inner.api.cancel(id as u32, rd_op.get());
+            }
+        })
+    }
+
+    fn with<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(&mut (Slab<StreamItem<T>>, Slab<Operation>)) -> R,
+    {
+        let mut storage = self.inner.storage.borrow_mut();
+        f(&mut *storage)
+    }
+}
+
+impl<T> Clone for StreamCtl<T> {
+    fn clone(&self) -> Self {
+        self.with(|streams| {
+            streams.0[self.id].ref_count += 1;
+            Self {
+                id: self.id,
+                inner: self.inner.clone(),
+            }
+        })
+    }
+}
+
+impl<T> Drop for StreamCtl<T> {
+    fn drop(&mut self) {
+        if let Ok(storage) = &mut self.inner.storage.try_borrow_mut() {
+            log::debug!("Drop io ({}), {:?}", self.id, storage.0[self.id].fd);
+
+            storage.0[self.id].ref_count -= 1;
+            if storage.0[self.id].ref_count == 0 {
+                let item = storage.0.remove(self.id);
+                if item.io.is_some() {
+                    let id = storage.1.insert(Operation::Close { tx: None });
+                    assert!(id < u32::MAX as usize);
+                    self.inner
+                        .api
+                        .submit(id as u32, opcode::Close::new(item.fd).build());
+                }
+            }
+        } else {
+            self.inner.feed.borrow_mut().push_back(self.id);
+        }
+    }
+}
+
+impl<T> PartialEq for StreamCtl<T> {
+    #[inline]
+    fn eq(&self, other: &StreamCtl<T>) -> bool {
+        self.id == other.id && std::ptr::eq(&self.inner, &other.inner)
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for StreamCtl<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.with(|streams| {
+            f.debug_struct("StreamCtl")
+                .field("id", &self.id)
+                .field("io", &streams.0[self.id].io)
+                .finish()
+        })
+    }
+}

ntex-net/src/rt_uring/io.rs

@@ -0,0 +1,107 @@
+use std::{any, future::poll_fn, io, task::Poll};
+
+use ntex_io::{
+    types, Handle, IoContext, IoStream, ReadContext, ReadStatus, WriteContext, WriteStatus,
+};
+use ntex_neon::{net::TcpStream, net::UnixStream, spawn};
+
+use super::driver::{Closable, StreamOps, StreamCtl};
+
+impl IoStream for super::TcpStream {
+    fn start(self, read: ReadContext, _: WriteContext) -> Option<Box<dyn Handle>> {
+        let io = self.0;
+        let context = read.context();
+        let ctl = StreamOps::current().register(io, context.clone());
+        let ctl2 = ctl.clone();
+        spawn(async move { run(ctl, context).await }).detach();
+
+        Some(Box::new(HandleWrapper(ctl2)))
+    }
+}
+
+impl IoStream for super::UnixStream {
+    fn start(self, read: ReadContext, _: WriteContext) -> Option<Box<dyn Handle>> {
+        let io = self.0;
+        let context = read.context();
+        let ctl = StreamOps::current().register(io, context.clone());
+        spawn(async move { run(ctl, context).await }).detach();
+
+        None
+    }
+}
+
+struct HandleWrapper(StreamCtl<TcpStream>);
+
+impl Handle for HandleWrapper {
+    fn query(&self, id: any::TypeId) -> Option<Box<dyn any::Any>> {
+        if id == any::TypeId::of::<types::PeerAddr>() {
+            let addr = self.0.with_io(|io| io.and_then(|io| io.peer_addr().ok()));
+            if let Some(addr) = addr {
+                return Some(Box::new(types::PeerAddr(addr)));
+            }
+        }
+        None
+    }
+}
+
+impl Closable for TcpStream {
+    async fn close(self) -> io::Result<()> {
+        TcpStream::close(self).await
+    }
+}
+
+impl Closable for UnixStream {
+    async fn close(self) -> io::Result<()> {
+        UnixStream::close(self).await
+    }
+}
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+enum Status {
+    Shutdown,
+    Terminate,
+}
+
+async fn run<T: Closable>(ctl: StreamCtl<T>, context: IoContext) {
+    // Handle io read readiness
+    let st = poll_fn(|cx| {
+        let read = match context.poll_read_ready(cx) {
+            Poll::Ready(ReadStatus::Ready) => {
+                ctl.resume_read();
+                Poll::Pending
+            }
+            Poll::Ready(ReadStatus::Terminate) => Poll::Ready(()),
+            Poll::Pending => {
+                ctl.pause_read();
+                Poll::Pending
+            }
+        };
+
+        let write = match context.poll_write_ready(cx) {
+            Poll::Ready(WriteStatus::Ready) => {
+                ctl.resume_write();
+                Poll::Pending
+            }
+            Poll::Ready(WriteStatus::Shutdown) => Poll::Ready(Status::Shutdown),
+            Poll::Ready(WriteStatus::Terminate) => Poll::Ready(Status::Terminate),
+            Poll::Pending => Poll::Pending,
+        };
+
+        if read.is_pending() && write.is_pending() {
+            Poll::Pending
+        } else if write.is_ready() {
+            write
+        } else {
+            Poll::Ready(Status::Terminate)
+        }
+    })
+    .await;
+
+    ctl.resume_write();
+    context.shutdown(st == Status::Shutdown).await;
+
+    ctl.pause_all();
+    let result = ctl.close().await;
+
+    context.stopped(result.err());
+}

ntex-net/src/rt_uring/mod.rs

@@ -0,0 +1,65 @@
+use std::{io::Result, net, net::SocketAddr};
+
+use ntex_bytes::PoolRef;
+use ntex_io::Io;
+
+//mod connect;
+mod driver;
+//mod io;
+
+/// Tcp stream wrapper for neon TcpStream
+struct TcpStream(ntex_neon::net::TcpStream);
+
+/// Tcp stream wrapper for neon UnixStream
+struct UnixStream(ntex_neon::net::UnixStream);
+
+/// Opens a TCP connection to a remote host.
+pub async fn tcp_connect(addr: SocketAddr) -> Result<Io> {
+    //let sock = connect::connect(addr).await?;
+    //Ok(Io::new(TcpStream(sock)))
+    todo!()
+}
+
+/// Opens a TCP connection to a remote host and use specified memory pool.
+pub async fn tcp_connect_in(addr: SocketAddr, pool: PoolRef) -> Result<Io> {
+    //let sock = connect::connect(addr).await?;
+    //Ok(Io::with_memory_pool(TcpStream(sock), pool))
+    todo!()
+}
+
+/// Opens a unix stream connection.
+pub async fn unix_connect<'a, P>(addr: P) -> Result<Io>
+where
+    P: AsRef<std::path::Path> + 'a,
+{
+    //let sock = connect::connect_unix(addr).await?;
+    //Ok(Io::new(UnixStream(sock)))
+    todo!()
+}
+
+/// Opens a unix stream connection and specified memory pool.
+pub async fn unix_connect_in<'a, P>(addr: P, pool: PoolRef) -> Result<Io>
+where
+    P: AsRef<std::path::Path> + 'a,
+{
+    //let sock = connect::connect_unix(addr).await?;
+    //Ok(Io::with_memory_pool(UnixStream(sock), pool))
+    todo!()
+}
+
+/// Convert std TcpStream to tokio's TcpStream
+pub fn from_tcp_stream(stream: net::TcpStream) -> Result<Io> {
+    //stream.set_nodelay(true)?;
+    //Ok(Io::new(TcpStream(ntex_neon::net::TcpStream::from_std(
+    //stream,
+    //)?)))
+    todo!()
+}
+
+/// Convert std UnixStream to tokio's UnixStream
+pub fn from_unix_stream(stream: std::os::unix::net::UnixStream) -> Result<Io> {
+    //Ok(Io::new(UnixStream(ntex_neon::net::UnixStream::from_std(
+    //    stream,
+    //)?)))
+    todo!()
+}