io_uring: move runtime into workspace crate

io-uring/src/ops/accept.rs

@@ -0,0 +1,67 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
+        pending_result::PendingResult,
+        sys::{IORING_OP_ACCEPT, io_uring_sqe},
+    },
+    std::rc::Rc,
+    uapi::{OwnedFd, c},
+};
+
+impl IoUring {
+    pub async fn accept(
+        &self,
+        fd: &Rc<OwnedFd>,
+        flags: c::c_int,
+    ) -> Result<Rc<OwnedFd>, IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut pw = self.ring.cached_accepts.pop().unwrap_or_default();
+            pw.id = id.id;
+            pw.fd = fd.raw() as _;
+            pw.flags = flags as _;
+            pw.data = Some(Data {
+                pr: pr.clone(),
+                _fd: fd.clone(),
+            });
+            self.ring.schedule(pw);
+        }
+        Ok(pr.await.map(OwnedFd::new).map(Rc::new)).merge()
+    }
+}
+
+struct Data {
+    pr: PendingResult,
+    _fd: Rc<OwnedFd>,
+}
+
+#[derive(Default)]
+pub struct AcceptTask {
+    id: IoUringTaskId,
+    fd: i32,
+    flags: u32,
+    data: Option<Data>,
+}
+
+unsafe impl Task for AcceptTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(data) = self.data.take() {
+            data.pr.complete(res);
+        }
+        ring.cached_accepts.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_ACCEPT;
+        sqe.fd = self.fd;
+        sqe.u2.addr = 0;
+        sqe.u1.addr2 = 0;
+        sqe.u3.accept_flags = self.flags;
+    }
+}

io-uring/src/ops/async_cancel.rs

@@ -0,0 +1,48 @@
+use {
+    crate::{
+        IoUringData, IoUringTaskId, Task,
+        sys::{IORING_OP_ASYNC_CANCEL, io_uring_sqe},
+    },
+    jay_utils::errorfmt::ErrorFmt,
+    uapi::c,
+};
+
+#[derive(Default)]
+pub struct AsyncCancelTask {
+    id: IoUringTaskId,
+    target: IoUringTaskId,
+}
+
+impl IoUringData {
+    pub fn cancel_task_in_kernel(&self, target: IoUringTaskId) {
+        let id = self.id_raw();
+        let mut task = self.cached_cancels.pop().unwrap_or_default();
+        task.id = id;
+        task.target = target;
+        self.schedule(task);
+    }
+}
+
+unsafe impl Task for AsyncCancelTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Err(e) = map_err!(res) {
+            if e.0 != c::ENOENT {
+                log::debug!("Could not cancel task: {}", ErrorFmt(e));
+            }
+        }
+        ring.cached_cancels.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_ASYNC_CANCEL;
+        sqe.u2.addr = self.target.raw();
+    }
+
+    fn is_cancel(&self) -> bool {
+        true
+    }
+}

io-uring/src/ops/connect.rs

@@ -0,0 +1,77 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
+        pending_result::PendingResult,
+        sys::{IORING_OP_CONNECT, io_uring_sqe},
+    },
+    std::{ptr, rc::Rc},
+    uapi::{OwnedFd, SockAddr, c},
+};
+
+impl IoUring {
+    pub async fn connect<T: SockAddr>(&self, fd: &Rc<OwnedFd>, t: &T) -> Result<(), IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut pw = self.ring.cached_connects.pop().unwrap_or_default();
+            pw.id = id.id;
+            pw.fd = fd.raw() as _;
+            unsafe {
+                ptr::copy_nonoverlapping(t, &mut pw.sockaddr as *mut _ as *mut _, 1);
+            }
+            pw.addrlen = size_of::<T>() as _;
+            pw.data = Some(Data {
+                pr: pr.clone(),
+                _fd: fd.clone(),
+            });
+            self.ring.schedule(pw);
+        }
+        Ok(pr.await.map(drop)).merge()
+    }
+}
+
+struct Data {
+    pr: PendingResult,
+    _fd: Rc<OwnedFd>,
+}
+
+pub struct ConnectTask {
+    id: IoUringTaskId,
+    fd: i32,
+    sockaddr: c::sockaddr_storage,
+    addrlen: u64,
+    data: Option<Data>,
+}
+
+impl Default for ConnectTask {
+    fn default() -> Self {
+        Self {
+            id: Default::default(),
+            fd: 0,
+            sockaddr: uapi::pod_zeroed(),
+            addrlen: 0,
+            data: None,
+        }
+    }
+}
+
+unsafe impl Task for ConnectTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(data) = self.data.take() {
+            data.pr.complete(res);
+        }
+        ring.cached_connects.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_CONNECT;
+        sqe.fd = self.fd;
+        sqe.u2.addr = &self.sockaddr as *const _ as _;
+        sqe.u1.off = self.addrlen;
+    }
+}

io-uring/src/ops/poll.rs

@@ -0,0 +1,70 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
+        ops::TaskResult,
+        pending_result::PendingResult,
+        sys::{IORING_OP_POLL_ADD, io_uring_sqe},
+    },
+    std::rc::Rc,
+    uapi::{OwnedFd, c},
+};
+
+impl IoUring {
+    pub async fn poll(&self, fd: &Rc<OwnedFd>, events: c::c_short) -> TaskResult<c::c_short> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut pw = self.ring.cached_polls.pop().unwrap_or_default();
+            pw.id = id.id;
+            pw.fd = fd.raw() as _;
+            pw.events = events as _;
+            pw.data = Some(Data {
+                pr: pr.clone(),
+                _fd: fd.clone(),
+            });
+            self.ring.schedule(pw);
+        }
+        Ok(pr.await.map(|v| v as c::c_short))
+    }
+
+    pub async fn readable(&self, fd: &Rc<OwnedFd>) -> Result<c::c_short, IoUringError> {
+        self.poll(fd, c::POLLIN).await.merge()
+    }
+
+    pub async fn writable(&self, fd: &Rc<OwnedFd>) -> Result<c::c_short, IoUringError> {
+        self.poll(fd, c::POLLOUT).await.merge()
+    }
+}
+
+struct Data {
+    pr: PendingResult,
+    _fd: Rc<OwnedFd>,
+}
+
+#[derive(Default)]
+pub struct PollTask {
+    id: IoUringTaskId,
+    events: u16,
+    fd: i32,
+    data: Option<Data>,
+}
+
+unsafe impl Task for PollTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(data) = self.data.take() {
+            data.pr.complete(res);
+        }
+        ring.cached_polls.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_POLL_ADD;
+        sqe.fd = self.fd;
+        sqe.u3.poll_events = self.events;
+    }
+}

io-uring/src/ops/poll_external.rs

@@ -0,0 +1,113 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task,
+        sys::{IORING_OP_POLL_ADD, io_uring_sqe},
+    },
+    jay_utils::oserror::OsError,
+    std::{cell::Cell, rc::Rc},
+    uapi::{OwnedFd, c},
+};
+
+pub trait PollCallback {
+    fn completed(self: Rc<Self>, res: Result<c::c_short, OsError>);
+}
+
+pub struct PendingPoll {
+    data: Rc<IoUringData>,
+    shared: Rc<PollExternalTaskShared>,
+    id: IoUringTaskId,
+}
+
+impl Drop for PendingPoll {
+    fn drop(&mut self) {
+        if self.shared.id.get() != self.id {
+            return;
+        }
+        self.shared.callback.take();
+        self.data.cancel_task(self.id);
+    }
+}
+
+impl IoUring {
+    pub fn poll_external(
+        &self,
+        fd: &Rc<OwnedFd>,
+        events: c::c_short,
+        callback: Rc<dyn PollCallback>,
+    ) -> Result<PendingPoll, IoUringError> {
+        self.ring.check_destroyed()?;
+        let mut pw = self.ring.cached_polls_external.pop().unwrap_or_default();
+        pw.shared.id.set(self.ring.id_raw());
+        pw.shared.callback.set(Some(callback));
+        pw.fd = fd.raw() as _;
+        pw.events = events as _;
+        pw.data = Some(Data { _fd: fd.clone() });
+        let pending = PendingPoll {
+            data: self.ring.clone(),
+            shared: pw.shared.clone(),
+            id: pw.shared.id.get(),
+        };
+        self.ring.schedule(pw);
+        Ok(pending)
+    }
+
+    pub fn readable_external(
+        &self,
+        fd: &Rc<OwnedFd>,
+        callback: Rc<dyn PollCallback>,
+    ) -> Result<PendingPoll, IoUringError> {
+        self.poll_external(fd, c::POLLIN, callback)
+    }
+
+    pub fn writable_external(
+        &self,
+        fd: &Rc<OwnedFd>,
+        callback: Rc<dyn PollCallback>,
+    ) -> Result<PendingPoll, IoUringError> {
+        self.poll_external(fd, c::POLLOUT, callback)
+    }
+}
+
+struct Data {
+    _fd: Rc<OwnedFd>,
+}
+
+#[derive(Default)]
+struct PollExternalTaskShared {
+    id: Cell<IoUringTaskId>,
+    callback: Cell<Option<Rc<dyn PollCallback>>>,
+}
+
+#[derive(Default)]
+pub struct PollExternalTask {
+    shared: Rc<PollExternalTaskShared>,
+    events: u16,
+    fd: i32,
+    data: Option<Data>,
+}
+
+unsafe impl Task for PollExternalTask {
+    fn id(&self) -> IoUringTaskId {
+        self.shared.id.get()
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        self.data.take();
+        self.shared.id.set(Default::default());
+        if let Some(cb) = self.shared.callback.take() {
+            let res = if res < 0 {
+                Err(OsError::from(-res as c::c_int))
+            } else {
+                Ok(res as _)
+            };
+            cb.completed(res)
+        }
+        ring.cached_polls_external.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_POLL_ADD;
+        sqe.fd = self.fd;
+        sqe.u3.poll_events = self.events;
+    }
+}

io-uring/src/ops/read_write.rs

@@ -0,0 +1,100 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
+        pending_result::PendingResult,
+        sys::{IORING_OP_READ, IORING_OP_WRITE, io_uring_sqe},
+    },
+    jay_time::Time,
+    jay_utils::buf::Buf,
+    std::rc::Rc,
+    uapi::{OwnedFd, c},
+};
+
+impl IoUring {
+    pub async fn read(&self, fd: &Rc<OwnedFd>, buf: Buf) -> Result<usize, IoUringError> {
+        self.perform(fd, buf, None, IORING_OP_READ).await
+    }
+
+    pub async fn write(
+        &self,
+        fd: &Rc<OwnedFd>,
+        buf: Buf,
+        timeout: Option<Time>,
+    ) -> Result<usize, IoUringError> {
+        self.perform(fd, buf, timeout, IORING_OP_WRITE).await
+    }
+
+    async fn perform(
+        &self,
+        fd: &Rc<OwnedFd>,
+        buf: Buf,
+        timeout: Option<Time>,
+        opcode: u8,
+    ) -> Result<usize, IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut pw = self.ring.cached_read_writes.pop().unwrap_or_default();
+            pw.opcode = opcode;
+            pw.id = id.id;
+            pw.has_timeout = timeout.is_some();
+            pw.fd = fd.raw();
+            pw.buf = buf.as_ptr() as _;
+            pw.len = buf.len();
+            pw.data = Some(ReadWriteTaskData {
+                _fd: fd.clone(),
+                _buf: buf,
+                res: pr.clone(),
+            });
+            self.ring.schedule(pw);
+            if let Some(time) = timeout {
+                self.schedule_timeout_link(time);
+            }
+        }
+        Ok(pr.await.map(|v| v as usize)).merge()
+    }
+}
+
+struct ReadWriteTaskData {
+    _fd: Rc<OwnedFd>,
+    _buf: Buf,
+    res: PendingResult,
+}
+
+#[derive(Default)]
+pub struct ReadWriteTask {
+    id: IoUringTaskId,
+    has_timeout: bool,
+    fd: c::c_int,
+    buf: usize,
+    len: usize,
+    data: Option<ReadWriteTaskData>,
+    opcode: u8,
+}
+
+unsafe impl Task for ReadWriteTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(data) = self.data.take() {
+            data.res.complete(res);
+        }
+        ring.cached_read_writes.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = self.opcode;
+        sqe.fd = self.fd as _;
+        sqe.u1.off = !0;
+        sqe.u2.addr = self.buf as _;
+        sqe.u3.rw_flags = 0;
+        sqe.len = self.len as _;
+    }
+
+    fn has_timeout(&self) -> bool {
+        self.has_timeout
+    }
+}

io-uring/src/ops/read_write_no_cancel.rs

@@ -0,0 +1,135 @@
+#[cfg(test)]
+mod tests;
+
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
+        pending_result::PendingResult,
+        sys::{IORING_OP_READ, IORING_OP_WRITE, io_uring_sqe},
+    },
+    jay_time::Time,
+    run_on_drop::on_drop,
+    uapi::{Fd, c},
+};
+
+impl IoUring {
+    pub async fn read_no_cancel(
+        &self,
+        fd: Fd,
+        offset: usize,
+        buf: &mut [u8],
+        cancel: impl FnOnce(IoUringTaskId),
+    ) -> Result<usize, IoUringError> {
+        self.perform_no_cancel(
+            fd,
+            offset,
+            buf.as_mut_ptr(),
+            buf.len(),
+            None,
+            IORING_OP_READ,
+            cancel,
+        )
+        .await
+    }
+
+    pub async fn write_no_cancel(
+        &self,
+        fd: Fd,
+        offset: usize,
+        buf: &[u8],
+        timeout: Option<Time>,
+        cancel: impl FnOnce(IoUringTaskId),
+    ) -> Result<usize, IoUringError> {
+        self.perform_no_cancel(
+            fd,
+            offset,
+            buf.as_ptr() as _,
+            buf.len(),
+            timeout,
+            IORING_OP_WRITE,
+            cancel,
+        )
+        .await
+    }
+
+    async fn perform_no_cancel(
+        &self,
+        fd: Fd,
+        offset: usize,
+        buf: *mut u8,
+        len: usize,
+        timeout: Option<Time>,
+        opcode: u8,
+        cancel: impl FnOnce(IoUringTaskId),
+    ) -> Result<usize, IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut pw = self
+                .ring
+                .cached_read_writes_no_cancel
+                .pop()
+                .unwrap_or_default();
+            pw.opcode = opcode;
+            pw.id = id.id;
+            pw.has_timeout = timeout.is_some();
+            pw.fd = fd.raw();
+            pw.offset = offset;
+            pw.buf = buf as _;
+            pw.len = len;
+            pw.data = Some(ReadWriteTaskData { res: pr.clone() });
+            self.ring.schedule(pw);
+            if let Some(time) = timeout {
+                self.schedule_timeout_link(time);
+            }
+        }
+        let panic = on_drop(|| panic!("Operation cannot be cancelled from userspace"));
+        cancel(id.id);
+        let res = Ok(pr.await.map(|v| v as usize)).merge();
+        panic.forget();
+        res
+    }
+}
+
+struct ReadWriteTaskData {
+    res: PendingResult,
+}
+
+#[derive(Default)]
+pub struct ReadWriteNoCancelTask {
+    id: IoUringTaskId,
+    has_timeout: bool,
+    fd: c::c_int,
+    offset: usize,
+    buf: usize,
+    len: usize,
+    data: Option<ReadWriteTaskData>,
+    opcode: u8,
+}
+
+unsafe impl Task for ReadWriteNoCancelTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(data) = self.data.take() {
+            data.res.complete(res);
+        }
+        ring.cached_read_writes_no_cancel.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = self.opcode;
+        sqe.fd = self.fd as _;
+        sqe.u1.off = self.offset as _;
+        sqe.u2.addr = self.buf as _;
+        sqe.u3.rw_flags = 0;
+        sqe.len = self.len as _;
+    }
+
+    fn has_timeout(&self) -> bool {
+        self.has_timeout
+    }
+}

io-uring/src/ops/read_write_no_cancel/tests.rs

@@ -0,0 +1,46 @@
+use {
+    crate::{IoUring, IoUringError},
+    jay_async_engine::AsyncEngine,
+    jay_utils::{oserror::OsError, queue::AsyncQueue},
+    std::rc::Rc,
+    uapi::c::ECANCELED,
+};
+
+fn cancel(timeout: bool) {
+    let eng = AsyncEngine::new();
+    let ring = IoUring::new(&eng, 32).unwrap();
+    let ring2 = ring.clone();
+    let ring3 = ring.clone();
+    let queue = Rc::new(AsyncQueue::new());
+    let queue2 = queue.clone();
+    let _fut1 = eng.spawn("", async move {
+        let (read, _write) = uapi::pipe().unwrap();
+        let mut buf = [10];
+        let res = ring
+            .read_no_cancel(read.borrow(), !0, &mut buf, |id| queue.push(id))
+            .await;
+        assert!(matches!(
+            res.unwrap_err(),
+            IoUringError::OsError(OsError(ECANCELED))
+        ));
+        ring.stop();
+    });
+    let _fut2 = eng.spawn("", async move {
+        let id = queue2.pop().await;
+        if timeout {
+            ring2.timeout(1).await.unwrap();
+        }
+        ring2.cancel(id);
+    });
+    ring3.run().unwrap();
+}
+
+#[test]
+fn cancel_in_kernel() {
+    cancel(true);
+}
+
+#[test]
+fn cancel_in_userspace() {
+    cancel(true);
+}

io-uring/src/ops/recvmsg.rs

@@ -0,0 +1,129 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task,
+        pending_result::PendingResult,
+        sys::{IORING_OP_RECVMSG, io_uring_sqe},
+    },
+    jay_utils::buf::Buf,
+    std::{cell::Cell, collections::VecDeque, mem::MaybeUninit, rc::Rc},
+    uapi::{OwnedFd, c},
+};
+
+impl IoUring {
+    pub async fn recvmsg(
+        &self,
+        fd: &Rc<OwnedFd>,
+        bufs: &mut [Buf],
+        fds: &mut VecDeque<Rc<OwnedFd>>,
+    ) -> Result<usize, IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        let mut cmsg = self.ring.cmsg_buf();
+        let cmsg_len;
+        {
+            let mut rm = self.ring.cached_recvmsg.pop().unwrap_or_default();
+            rm.iovecs.clear();
+            for buf in bufs {
+                rm.bufs.push(buf.clone());
+                rm.iovecs.push(c::iovec {
+                    iov_base: buf.as_ptr() as _,
+                    iov_len: buf.len() as _,
+                });
+            }
+            rm.id = id.id;
+            rm.fd = fd.raw();
+            rm.msghdr.msg_control = cmsg.as_ptr() as _;
+            rm.msghdr.msg_controllen = cmsg.len() as _;
+            rm.msghdr.msg_iov = rm.iovecs.as_mut_ptr();
+            rm.msghdr.msg_iovlen = rm.iovecs.len() as _;
+            rm.data = Some(Data {
+                _cmsg: cmsg.clone(),
+                _fd: fd.clone(),
+                pr: pr.clone(),
+            });
+            cmsg_len = rm.cmsg_len.clone();
+            self.ring.schedule(rm);
+        }
+        macro_rules! return_cmsg {
+            () => {
+                self.ring.cached_cmsg_bufs.push(cmsg);
+            };
+        }
+        match pr.await {
+            Ok(n) => {
+                let mut cmsg_data = &cmsg[..cmsg_len.get()];
+                while cmsg_data.len() > 0 {
+                    let (_, hdr, data) = match uapi::cmsg_read(&mut cmsg_data) {
+                        Ok(m) => m,
+                        Err(_) => {
+                            return_cmsg!();
+                            return Err(IoUringError::InvalidCmsgData);
+                        }
+                    };
+                    if (hdr.cmsg_level, hdr.cmsg_type) == (c::SOL_SOCKET, c::SCM_RIGHTS) {
+                        fds.extend(uapi::pod_iter(data).unwrap().map(Rc::new));
+                    }
+                }
+                return_cmsg!();
+                Ok(n as _)
+            }
+            Err(e) => {
+                return_cmsg!();
+                Err(IoUringError::OsError(e))
+            }
+        }
+    }
+}
+
+struct Data {
+    _cmsg: Buf,
+    _fd: Rc<OwnedFd>,
+    pr: PendingResult,
+}
+
+pub struct RecvmsgTask {
+    id: IoUringTaskId,
+    fd: c::c_int,
+    bufs: Vec<Buf>,
+    iovecs: Vec<c::iovec>,
+    msghdr: c::msghdr,
+    cmsg_len: Rc<Cell<usize>>,
+    data: Option<Data>,
+}
+
+impl Default for RecvmsgTask {
+    fn default() -> Self {
+        RecvmsgTask {
+            id: Default::default(),
+            fd: 0,
+            bufs: vec![],
+            iovecs: vec![],
+            msghdr: unsafe { MaybeUninit::zeroed().assume_init() },
+            cmsg_len: Rc::new(Cell::new(0)),
+            data: None,
+        }
+    }
+}
+
+unsafe impl Task for RecvmsgTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        self.cmsg_len.set(self.msghdr.msg_controllen as _);
+        self.bufs.clear();
+        if let Some(data) = self.data.take() {
+            data.pr.complete(res);
+        }
+        ring.cached_recvmsg.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_RECVMSG;
+        sqe.fd = self.fd as _;
+        sqe.u2.addr = &self.msghdr as *const _ as _;
+        sqe.u3.msg_flags = c::MSG_CMSG_CLOEXEC as _;
+    }
+}

io-uring/src/ops/sendmsg.rs

@@ -0,0 +1,139 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task,
+        pending_result::PendingResult,
+        sys::{IORING_OP_SENDMSG, io_uring_sqe},
+    },
+    jay_time::Time,
+    jay_utils::{buf::Buf, compat::IovLength, vec_ext::UninitVecExt},
+    std::{mem::MaybeUninit, ptr, rc::Rc},
+    uapi::{OwnedFd, c},
+};
+
+impl IoUring {
+    pub async fn sendmsg_one(
+        &self,
+        fd: &Rc<OwnedFd>,
+        buf: Buf,
+        fds: Vec<Rc<OwnedFd>>,
+        timeout: Option<Time>,
+    ) -> Result<usize, IoUringError> {
+        self.sendmsg(fd, &mut [buf], fds, timeout).await
+    }
+
+    pub async fn sendmsg(
+        &self,
+        fd: &Rc<OwnedFd>,
+        bufs: &mut [Buf],
+        fds: Vec<Rc<OwnedFd>>,
+        timeout: Option<Time>,
+    ) -> Result<usize, IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut st = self.ring.cached_sendmsg.pop().unwrap_or_default();
+            st.fds = fds;
+            if st.fds.len() > 0 {
+                let mut fd_ids = self.ring.fd_ids_scratch.borrow_mut();
+                fd_ids.clear();
+                fd_ids.extend(st.fds.iter().map(|f| f.raw()));
+                let space = uapi::cmsg_space(size_of_val(&fd_ids[..]));
+                st.cmsg.clear();
+                st.cmsg.reserve(space);
+                st.cmsg.set_len_safe(space);
+                let mut hdr: c::cmsghdr = uapi::pod_zeroed();
+                hdr.cmsg_level = c::SOL_SOCKET;
+                hdr.cmsg_type = c::SCM_RIGHTS;
+                uapi::cmsg_write(&mut &mut st.cmsg[..], hdr, &fd_ids[..]).unwrap();
+                st.msghdr.msg_control = st.cmsg.as_ptr() as _;
+                st.msghdr.msg_controllen = st.cmsg.len() as _;
+            } else {
+                st.msghdr.msg_control = ptr::null_mut();
+                st.msghdr.msg_controllen = 0;
+            }
+            st.id = id.id;
+            st.fd = fd.raw();
+            st.bufs.clear();
+            st.bufs.extend(bufs.iter_mut().map(|b| b.clone()));
+            st.iovecs.clear();
+            st.iovecs.extend(bufs.iter().map(|b| c::iovec {
+                iov_base: b.as_ptr() as _,
+                iov_len: b.len(),
+            }));
+            st.msghdr.msg_iov = st.iovecs.as_ptr() as _;
+            st.msghdr.msg_iovlen = st.iovecs.len() as IovLength;
+            st.data = Some(SendmsgTaskData {
+                _fd: fd.clone(),
+                res: pr.clone(),
+            });
+            st.has_timeout = timeout.is_some();
+            self.ring.schedule(st);
+            if let Some(timeout) = timeout {
+                self.schedule_timeout_link(timeout);
+            }
+        }
+        Ok(pr.await? as _)
+    }
+}
+
+struct SendmsgTaskData {
+    _fd: Rc<OwnedFd>,
+    res: PendingResult,
+}
+
+pub struct SendmsgTask {
+    id: IoUringTaskId,
+    iovecs: Vec<c::iovec>,
+    msghdr: c::msghdr,
+    bufs: Vec<Buf>,
+    fd: i32,
+    has_timeout: bool,
+    fds: Vec<Rc<OwnedFd>>,
+    cmsg: Vec<MaybeUninit<u8>>,
+    data: Option<SendmsgTaskData>,
+}
+
+impl Default for SendmsgTask {
+    fn default() -> Self {
+        unsafe {
+            SendmsgTask {
+                id: Default::default(),
+                iovecs: vec![],
+                msghdr: MaybeUninit::zeroed().assume_init(),
+                bufs: vec![],
+                fd: 0,
+                has_timeout: false,
+                fds: vec![],
+                cmsg: vec![],
+                data: None,
+            }
+        }
+    }
+}
+
+unsafe impl Task for SendmsgTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        self.fds.clear();
+        self.bufs.clear();
+        if let Some(data) = self.data.take() {
+            data.res.complete(res);
+        }
+        ring.cached_sendmsg.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_SENDMSG;
+        sqe.fd = self.fd;
+        sqe.u2.addr = &self.msghdr as *const _ as _;
+        sqe.u3.msg_flags = c::MSG_NOSIGNAL as _;
+    }
+
+    fn has_timeout(&self) -> bool {
+        self.has_timeout
+    }
+}

io-uring/src/ops/timeout.rs

@@ -0,0 +1,63 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task,
+        pending_result::PendingResult,
+        sys::{IORING_OP_TIMEOUT, IORING_TIMEOUT_ABS, io_uring_sqe},
+    },
+    uapi::c,
+};
+
+#[repr(C)]
+#[derive(Default)]
+pub(super) struct timespec64 {
+    pub tv_sec: i64,
+    pub tv_nsec: c::c_long,
+}
+
+#[derive(Default)]
+pub struct TimeoutTask {
+    id: IoUringTaskId,
+    timespec: timespec64,
+    pr: Option<PendingResult>,
+}
+
+impl IoUring {
+    pub async fn timeout(&self, timeout_nsec: u64) -> Result<(), IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut pw = self.ring.cached_timeouts.pop().unwrap_or_default();
+            pw.id = id.id;
+            pw.timespec = timespec64 {
+                tv_sec: (timeout_nsec / 1_000_000_000) as _,
+                tv_nsec: (timeout_nsec % 1_000_000_000) as _,
+            };
+            pw.pr = Some(pr.clone());
+            self.ring.schedule(pw);
+        }
+        let _ = pr.await;
+        Ok(())
+    }
+}
+
+unsafe impl Task for TimeoutTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(pr) = self.pr.take() {
+            pr.complete(res);
+        }
+        ring.cached_timeouts.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_TIMEOUT;
+        sqe.u2.addr = &self.timespec as *const _ as _;
+        sqe.len = 1;
+        sqe.u3.timeout_flags = IORING_TIMEOUT_ABS;
+        sqe.u1.off = 0;
+    }
+}

io-uring/src/ops/timeout_external.rs

@@ -0,0 +1,94 @@
+use {
+    crate::{
+        IoUring, IoUringData, IoUringError, IoUringTaskId, Task, ops::timeout::timespec64,
+        sys::{IORING_OP_TIMEOUT, IORING_TIMEOUT_ABS, io_uring_sqe},
+    },
+    jay_utils::oserror::OsError,
+    std::{cell::Cell, rc::Rc},
+    uapi::c,
+};
+
+pub trait TimeoutCallback {
+    fn completed(self: Rc<Self>, res: Result<(), OsError>, data: u64);
+}
+
+pub struct PendingTimeout {
+    data: Rc<IoUringData>,
+    shared: Rc<TimeoutExternalTaskShared>,
+    id: IoUringTaskId,
+}
+
+impl Drop for PendingTimeout {
+    fn drop(&mut self) {
+        if self.shared.id.get() != self.id {
+            return;
+        }
+        self.shared.callback.take();
+        self.data.cancel_task(self.id);
+    }
+}
+
+#[derive(Default)]
+struct TimeoutExternalTaskShared {
+    id: Cell<IoUringTaskId>,
+    callback: Cell<Option<Rc<dyn TimeoutCallback>>>,
+}
+
+#[derive(Default)]
+pub struct TimeoutExternalTask {
+    timespec: timespec64,
+    shared: Rc<TimeoutExternalTaskShared>,
+    data: u64,
+}
+
+impl IoUring {
+    pub fn timeout_external(
+        &self,
+        timeout_nsec: u64,
+        callback: Rc<dyn TimeoutCallback>,
+        data: u64,
+    ) -> Result<PendingTimeout, IoUringError> {
+        self.ring.check_destroyed()?;
+        let mut pw = self.ring.cached_timeouts_external.pop().unwrap_or_default();
+        pw.shared.id.set(self.ring.id_raw());
+        pw.shared.callback.set(Some(callback));
+        pw.timespec = timespec64 {
+            tv_sec: (timeout_nsec / 1_000_000_000) as _,
+            tv_nsec: (timeout_nsec % 1_000_000_000) as _,
+        };
+        pw.data = data;
+        let pending = PendingTimeout {
+            data: self.ring.clone(),
+            shared: pw.shared.clone(),
+            id: pw.shared.id.get(),
+        };
+        self.ring.schedule(pw);
+        Ok(pending)
+    }
+}
+
+unsafe impl Task for TimeoutExternalTask {
+    fn id(&self) -> IoUringTaskId {
+        self.shared.id.get()
+    }
+
+    fn complete(self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(pr) = self.shared.callback.take() {
+            let res = if res == -c::ETIME {
+                Ok(())
+            } else {
+                map_err!(res).map(drop)
+            };
+            pr.completed(res, self.data);
+        }
+        ring.cached_timeouts_external.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_TIMEOUT;
+        sqe.u2.addr = &self.timespec as *const _ as _;
+        sqe.len = 1;
+        sqe.u3.timeout_flags = IORING_TIMEOUT_ABS;
+        sqe.u1.off = 0;
+    }
+}

io-uring/src/ops/timeout_link.rs

@@ -0,0 +1,41 @@
+use crate::{
+    IoUring, IoUringData, IoUringTaskId, Task, ops::timeout::timespec64,
+    sys::{IORING_OP_LINK_TIMEOUT, IORING_TIMEOUT_ABS, io_uring_sqe},
+};
+use jay_time::Time;
+
+#[derive(Default)]
+pub struct TimeoutLinkTask {
+    id: IoUringTaskId,
+    timespec: timespec64,
+}
+
+impl IoUring {
+    pub(super) fn schedule_timeout_link(&self, timeout: Time) {
+        let id = self.ring.id_raw();
+        {
+            let mut to = self.ring.cached_timeout_links.pop().unwrap_or_default();
+            to.id = id;
+            to.timespec.tv_sec = timeout.0.tv_sec as _;
+            to.timespec.tv_nsec = timeout.0.tv_nsec as _;
+            self.ring.schedule(to);
+        }
+    }
+}
+
+unsafe impl Task for TimeoutLinkTask {
+    fn id(&self) -> IoUringTaskId {
+        self.id
+    }
+
+    fn complete(self: Box<Self>, ring: &IoUringData, _res: i32) {
+        ring.cached_timeout_links.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_LINK_TIMEOUT;
+        sqe.u2.addr = &self.timespec as *const _ as _;
+        sqe.len = 1;
+        sqe.u3.timeout_flags = IORING_TIMEOUT_ABS;
+    }
+}