feat(driver,net)!: more nonempty

Cargo.toml

@@ -75,7 +75,7 @@ widestring = "1.0.2"
 windows-sys = "0.61.0"
 thiserror = "2.0.3"
 smallvec = "1.13.2"
-synchrony = "0.1.7"
+synchrony = "0.1.8"
 thin-cell = "0.2.1"
 slotmap = "1.1.1"
 crossfire = "3.1.5"

compio-driver/src/sys/driver/mod.rs

@@ -21,3 +21,11 @@ cfg_if::cfg_if! {
 
 crate::assert_not_impl!(Driver, Send);
 crate::assert_not_impl!(Driver, Sync);
+
+/// An operation that can be optimized by making use of the "poll-first"
+/// feature.
+pub trait PollFirst {
+    /// Poll first before syscall. This is only meaningful for io-uring. It sets
+    /// `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio` of the SQE.
+    fn poll_first(&mut self);
+}

compio-driver/src/sys/extra/mod.rs

@@ -96,8 +96,7 @@ impl Extra {
         ///
         /// # Behaviour
         ///
-        /// This method must be used only on the flags for any of the `receive`
-        /// variants supported by `IO_URING`. The driver will try to check whether
+        /// This method must be used only on `IO_URING`. The driver will try to check whether
         /// the `IORING_CQE_F_SOCK_NONEMPTY` flag was set by the kernel for the CQE.
         /// On other platforms, this will always return the [`Unsupported`] error.
         ///

compio-driver/src/sys/op/managed/fallback.rs

@@ -5,7 +5,7 @@ use rustix::net::RecvFlags;
 use socket2::SockAddr;
 
 use crate::{
-    AsFd, BufferPool, BufferRef,
+    AsFd, BufferPool, BufferRef, PollFirst,
     op::{RecvMsg, TakeBuffer},
     sys::op::{Read, ReadAt, Recv, RecvFrom},
 };
@@ -69,11 +69,12 @@ impl<S> RecvManaged<S> {
             op: Recv::new(fd, pool.pop()?.with_capacity(len), flags),
         })
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    // This method has been added here for the sake of API compatibility.
-    pub fn poll_first(&mut self) {}
+impl<S> PollFirst for RecvManaged<S> {
+    fn poll_first(&mut self) {
+        self.op.poll_first();
+    }
 }
 
 impl<S> TakeBuffer for RecvManaged<S> {
@@ -96,11 +97,12 @@ impl<S: AsFd> RecvFromManaged<S> {
             op: RecvFrom::new(fd, pool.pop()?.with_capacity(len), flags),
         })
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    // This method has been added here for the sake of API compatibility.
-    pub fn poll_first(&mut self) {}
+impl<S: AsFd> PollFirst for RecvFromManaged<S> {
+    fn poll_first(&mut self) {
+        self.op.poll_first();
+    }
 }
 
 impl<S: AsFd> TakeBuffer for RecvFromManaged<S> {
@@ -129,11 +131,12 @@ impl<C: IoBufMut, S: AsFd> RecvMsgManaged<C, S> {
             op: RecvMsg::new(fd, [pool.pop()?.with_capacity(len)], control, flags),
         })
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    // This method has been added here for the sake of API compatibility.
-    pub fn poll_first(&mut self) {}
+impl<C: IoBufMut, S: AsFd> PollFirst for RecvMsgManaged<C, S> {
+    fn poll_first(&mut self) {
+        self.op.poll_first();
+    }
 }
 
 impl<C: IoBufMut, S: AsFd> TakeBuffer for RecvMsgManaged<C, S> {

compio-driver/src/sys/op/managed/fusion.rs

@@ -3,7 +3,9 @@ use rustix::net::RecvFlags;
 use socket2::SockAddr;
 
 use super::{fallback, iour};
-use crate::{BufferPool, BufferRef, IourOpCode, OpEntry, OpType, PollOpCode, sys::pal::*};
+use crate::{
+    BufferPool, BufferRef, IourOpCode, OpEntry, OpType, PollFirst, PollOpCode, sys::pal::*,
+};
 
 macro_rules! mop {
     (<$($ty:ident: $trait:ident),* $(,)?> $name:ident( $($arg:ident: $arg_t:ty),* $(,)? ) with $pool:ident) => {
@@ -131,32 +133,26 @@ mop!(<S: AsFd> RecvMulti(fd: S, pool: &BufferPool, len: usize, flags: RecvFlags)
 mop!(<S: AsFd> RecvFromMulti(fd: S, pool: &BufferPool, flags: RecvFlags) with pool; RecvFromMultiResult);
 mop!(<S: AsFd> RecvMsgMulti(fd: S, pool: &BufferPool, control_len: usize, flags: RecvFlags) with pool; RecvMsgMultiResult);
 
-impl<S: AsFd> RecvManaged<S> {
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<S: AsFd> PollFirst for RecvManaged<S> {
+    fn poll_first(&mut self) {
         match self.inner {
             RecvManagedInner::Poll(ref mut i) => i.poll_first(),
             RecvManagedInner::IoUring(ref mut i) => i.poll_first(),
         }
     }
 }
 
-impl<S: AsFd> RecvFromManaged<S> {
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<S: AsFd> PollFirst for RecvFromManaged<S> {
+    fn poll_first(&mut self) {
         match self.inner {
             RecvFromManagedInner::Poll(ref mut i) => i.poll_first(),
             RecvFromManagedInner::IoUring(ref mut i) => i.poll_first(),
         }
     }
 }
 
-impl<C: IoBufMut, S: AsFd> RecvMsgManaged<C, S> {
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<C: IoBufMut, S: AsFd> PollFirst for RecvMsgManaged<C, S> {
+    fn poll_first(&mut self) {
         match self.inner {
             RecvMsgManagedInner::Poll(ref mut i) => i.poll_first(),
             RecvMsgManagedInner::IoUring(ref mut i) => i.poll_first(),

compio-driver/src/sys/op/managed/iour.rs

@@ -12,7 +12,7 @@ use rustix::net::RecvFlags;
 use socket2::{SockAddr, SockAddrStorage, socklen_t};
 
 use crate::{
-    BufferPool, BufferRef, Extra, IourOpCode as OpCode, OpEntry,
+    BufferPool, BufferRef, Extra, IourOpCode as OpCode, OpEntry, PollFirst,
     op::TakeBuffer,
     sys::pal::{is_kernel_at_least, set_poll_first},
 };
@@ -162,10 +162,10 @@ impl<S> RecvManaged<S> {
             poll_first: false,
         })
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<S> PollFirst for RecvManaged<S> {
+    fn poll_first(&mut self) {
         self.poll_first = true;
     }
 }
@@ -250,10 +250,10 @@ impl<S> RecvFromManaged<S> {
             poll_first: false,
         })
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<S> PollFirst for RecvFromManaged<S> {
+    fn poll_first(&mut self) {
         self.poll_first = true;
     }
 }
@@ -330,10 +330,10 @@ impl<C: IoBufMut, S: AsFd> RecvMsgManaged<C, S> {
             control_len: 0,
         })
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<C: IoBufMut, S: AsFd> PollFirst for RecvMsgManaged<C, S> {
+    fn poll_first(&mut self) {
         self.op.poll_first();
     }
 }

compio-driver/src/sys/op/socket/iour.rs

@@ -94,14 +94,15 @@ unsafe impl<S: AsFd> OpCode for Accept<S> {
     type Control = ();
 
     fn create_entry(&mut self, _: &mut Self::Control) -> OpEntry {
-        opcode::Accept::new(
+        let entry = opcode::Accept::new(
             Fd(self.fd.as_fd().as_raw_fd()),
             unsafe { self.buffer.view_as::<libc::sockaddr>() },
             &raw mut self.addr_len,
         )
         .flags(libc::SOCK_CLOEXEC)
-        .build()
-        .into()
+        .build();
+        let entry = set_poll_first(entry, self.poll_first);
+        entry.into()
     }
 
     unsafe fn set_result(&mut self, _: &mut Self::Control, res: &io::Result<usize>, _: &Extra) {

compio-driver/src/sys/op/socket/mod.rs

@@ -15,7 +15,7 @@ mod_use![stub];
 
 use rustix::net::{RecvFlags, SendFlags};
 
-use crate::sys::prelude::*;
+use crate::{PollFirst, sys::prelude::*};
 
 /// Connect to a remote address.
 pub struct Connect<S> {
@@ -261,10 +261,10 @@ impl<T: IoVectoredBufMut, C: IoBufMut, S> RecvMsg<T, C, S> {
             poll_first: false,
         }
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<T: IoVectoredBufMut, C: IoBufMut, S> PollFirst for RecvMsg<T, C, S> {
+    fn poll_first(&mut self) {
         self.poll_first = true;
     }
 }
@@ -291,10 +291,10 @@ impl<T: IoBufMut, S> Recv<T, S> {
             poll_first: false,
         }
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<T: IoBufMut, S> PollFirst for Recv<T, S> {
+    fn poll_first(&mut self) {
         self.poll_first = true;
     }
 }
@@ -317,10 +317,10 @@ impl<T: IoVectoredBufMut, S> RecvVectored<T, S> {
             poll_first: false,
         }
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<T: IoVectoredBufMut, S> PollFirst for RecvVectored<T, S> {
+    fn poll_first(&mut self) {
         self.poll_first = true;
     }
 }
@@ -359,10 +359,10 @@ impl<T: IoVectoredBufMut, S> RecvFromVectored<T, S> {
             buffer,
         }
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<T: IoVectoredBufMut, S> PollFirst for RecvFromVectored<T, S> {
+    fn poll_first(&mut self) {
         self.header.poll_first = true;
     }
 }
@@ -384,10 +384,10 @@ impl<T: IoBufMut, S> RecvFrom<T, S> {
             buffer,
         }
     }
+}
 
-    /// This method sets the `IORING_RECVSEND_POLL_FIRST` flag in the `ioprio`
-    /// of the SQE on the IO_URING driver.
-    pub fn poll_first(&mut self) {
+impl<T: IoBufMut, S> PollFirst for RecvFrom<T, S> {
+    fn poll_first(&mut self) {
         self.header.poll_first = true;
     }
 }

compio-driver/src/sys/op/socket/unix.rs

@@ -9,7 +9,7 @@ use rustix::{
     },
 };
 
-use crate::sys::op::*;
+use crate::{PollFirst, sys::op::*};
 
 impl<S: AsFd> Accept<S> {
     pub(crate) fn call(&mut self) -> io::Result<usize> {
@@ -299,6 +299,7 @@ pub struct Accept<S> {
     pub(crate) buffer: SockAddrStorage,
     pub(crate) addr_len: socklen_t,
     pub(crate) accepted_fd: Option<Socket2>,
+    pub(crate) poll_first: bool,
 }
 
 impl<S> Accept<S> {
@@ -311,10 +312,17 @@ impl<S> Accept<S> {
             buffer,
             addr_len,
             accepted_fd: None,
+            poll_first: false,
         }
     }
 }
 
+impl<S> PollFirst for Accept<S> {
+    fn poll_first(&mut self) {
+        self.poll_first = true;
+    }
+}
+
 impl<S> IntoInner for Accept<S> {
     type Inner = (Socket2, SockAddr);
 

compio-driver/src/sys/pal/iour/mod.rs

@@ -57,7 +57,11 @@ pub fn is_kernel_at_least(v: impl Into<KernelVersion>) -> bool {
 }
 
 pub(crate) fn set_poll_first(mut entry: Entry, flag: bool) -> Entry {
-    if flag && is_kernel_at_least((5, 19)) {
+    let version = match entry.get_opcode() as u8 {
+        io_uring::opcode::Accept::CODE => (6, 10),
+        _ => (5, 19),
+    };
+    if flag && is_kernel_at_least(version) {
         let sqe = &raw mut entry as *mut io_uring_sqe;
         unsafe {
             (*sqe).ioprio |= IORING_RECVSEND_POLL_FIRST as u16;