Merge pull request #193 from Berrysoft/io-uring-large-qe

feat(driver,iour): add large sqe & cqe support

Author: Berrysoft

azure-pipelines.yml

@@ -36,22 +36,24 @@ jobs:
       matrix:
         focal:
           image: ubuntu-20.04
+          features:
         jammy:
           image: ubuntu-22.04
+          features: io-uring-sqe128,io-uring-cqe32
     pool:
       vmImage: $(image)
 
     steps:
       - script: |
           rustup toolchain install nightly
-          cargo +nightly test --workspace --features all,nightly
+          cargo +nightly test --workspace --features all,nightly,$(features)
         displayName: TestNightly
       - script: |
           rustup toolchain install beta
-          cargo +beta test --workspace --features all
+          cargo +beta test --workspace --features all,$(features)
         displayName: TestBeta
       # - script: |
-      #     cargo test --workspace --features all
+      #     cargo test --workspace --features all,$(features)
       #   displayName: TestStable
 
       - script: |

compio-driver/Cargo.toml

@@ -78,6 +78,9 @@ compio-buf = { workspace = true, features = ["arrayvec"] }
 default = ["io-uring"]
 polling = ["dep:polling"]
 
+io-uring-sqe128 = []
+io-uring-cqe32 = []
+
 # Nightly features
 once_cell_try = []
 nightly = ["once_cell_try"]

compio-driver/src/fusion/mod.rs

@@ -85,6 +85,9 @@ mod driver_type {
             OpenAt::CODE,
             Close::CODE,
             Shutdown::CODE,
+            // Linux kernel 5.19
+            #[cfg(any(feature = "io-uring-seq128", feature = "io-uring-cqe32"))]
+            Socket::CODE,
         ];
 
         Ok(())

compio-driver/src/iour/mod.rs

@@ -8,10 +8,22 @@ use std::{
 
 use compio_log::{instrument, trace};
 use crossbeam_queue::SegQueue;
+cfg_if::cfg_if! {
+    if #[cfg(feature = "io-uring-cqe32")] {
+        use io_uring::cqueue::Entry32 as CEntry;
+    } else {
+        use io_uring::cqueue::Entry as CEntry;
+    }
+}
+cfg_if::cfg_if! {
+    if #[cfg(feature = "io-uring-sqe128")] {
+        use io_uring::squeue::Entry128 as SEntry;
+    } else {
+        use io_uring::squeue::Entry as SEntry;
+    }
+}
 use io_uring::{
-    cqueue,
     opcode::{AsyncCancel, Read},
-    squeue,
     types::{Fd, SubmitArgs, Timespec},
     IoUring,
 };
@@ -26,17 +38,27 @@ pub(crate) use crate::unix::RawOp;
 /// The created entry of [`OpCode`].
 pub enum OpEntry {
     /// This operation creates an io-uring submission entry.
-    Submission(squeue::Entry),
+    Submission(io_uring::squeue::Entry),
+    #[cfg(feature = "io-uring-sqe128")]
+    /// This operation creates an 128-bit io-uring submission entry.
+    Submission128(io_uring::squeue::Entry128),
     /// This operation is a blocking one.
     Blocking,
 }
 
-impl From<squeue::Entry> for OpEntry {
-    fn from(value: squeue::Entry) -> Self {
+impl From<io_uring::squeue::Entry> for OpEntry {
+    fn from(value: io_uring::squeue::Entry) -> Self {
         Self::Submission(value)
     }
 }
 
+#[cfg(feature = "io-uring-sqe128")]
+impl From<io_uring::squeue::Entry128> for OpEntry {
+    fn from(value: io_uring::squeue::Entry128) -> Self {
+        Self::Submission128(value)
+    }
+}
+
 /// Abstraction of io-uring operations.
 pub trait OpCode {
     /// Create submission entry.
@@ -51,8 +73,8 @@ pub trait OpCode {
 
 /// Low-level driver of io-uring.
 pub(crate) struct Driver {
-    inner: IoUring,
-    squeue: VecDeque<squeue::Entry>,
+    inner: IoUring<SEntry, CEntry>,
+    squeue: VecDeque<SEntry>,
     notifier: Notifier,
     notifier_registered: bool,
     pool: AsyncifyPool,
@@ -67,7 +89,7 @@ impl Driver {
         instrument!(compio_log::Level::TRACE, "new", ?builder);
         trace!("new iour driver");
         Ok(Self {
-            inner: IoUring::new(builder.capacity)?,
+            inner: IoUring::builder().build(builder.capacity)?,
             squeue: VecDeque::with_capacity(builder.capacity as usize),
             notifier: Notifier::new()?,
             notifier_registered: false,
@@ -170,24 +192,38 @@ impl Driver {
     pub fn cancel(&mut self, user_data: usize, _registry: &mut Slab<RawOp>) {
         instrument!(compio_log::Level::TRACE, "cancel", user_data);
         trace!("cancel RawOp");
+        #[allow(clippy::useless_conversion)]
         self.squeue.push_back(
             AsyncCancel::new(user_data as _)
                 .build()
-                .user_data(Self::CANCEL),
+                .user_data(Self::CANCEL)
+                .into(),
         );
     }
 
     pub fn push(&mut self, user_data: usize, op: &mut RawOp) -> Poll<io::Result<usize>> {
         instrument!(compio_log::Level::TRACE, "push", user_data);
         let op_pin = op.as_pin();
         trace!("push RawOp");
-        if let OpEntry::Submission(entry) = op_pin.create_entry() {
-            self.squeue.push_back(entry.user_data(user_data as _));
-            Poll::Pending
-        } else if self.push_blocking(user_data, op)? {
-            Poll::Pending
-        } else {
-            Poll::Ready(Err(io::Error::from_raw_os_error(libc::EBUSY)))
+        match op_pin.create_entry() {
+            OpEntry::Submission(entry) => {
+                #[allow(clippy::useless_conversion)]
+                self.squeue
+                    .push_back(entry.user_data(user_data as _).into());
+                Poll::Pending
+            }
+            #[cfg(feature = "io-uring-sqe128")]
+            OpEntry::Submission128(_entry) => {
+                self.squeue.push_back(_entry.user_data(user_data as _));
+                Poll::Pending
+            }
+            OpEntry::Blocking => {
+                if self.push_blocking(user_data, op)? {
+                    Poll::Pending
+                } else {
+                    Poll::Ready(Err(io::Error::from_raw_os_error(libc::EBUSY)))
+                }
+            }
         }
     }
 
@@ -223,10 +259,12 @@ impl Driver {
         if !self.notifier_registered {
             let fd = self.notifier.as_raw_fd();
             let dst = self.notifier.dst();
+            #[allow(clippy::useless_conversion)]
             self.squeue.push_back(
                 Read::new(Fd(fd), dst.as_mut_ptr(), dst.len() as _)
                     .build()
-                    .user_data(Self::NOTIFY),
+                    .user_data(Self::NOTIFY)
+                    .into(),
             );
             trace!("registered notifier");
             self.notifier_registered = true
@@ -259,7 +297,7 @@ impl AsRawFd for Driver {
     }
 }
 
-fn create_entry(entry: cqueue::Entry) -> Entry {
+fn create_entry(entry: CEntry) -> Entry {
     let result = entry.result();
     let result = if result < 0 {
         let result = if result == -libc::ECANCELED {