Handle overflown cqes when IORING_FEAT_NODROP supported (#349)

* fix: handle overflown cqes when IORING_FEAT_NODROP supported

* test: add test for cq overflow

* fix: remove duplicated deps

* fix: create buffer per read

Author: yw-furiosa

io-uring-test/src/main.rs

@@ -15,6 +15,11 @@ fn main() -> anyhow::Result<()> {
     let entries = 8;
 
     test::<squeue::Entry, cqueue::Entry>(IoUring::new(entries)?)?;
+    test_sqpoll::<squeue::Entry, cqueue::Entry>(
+        IoUring::builder()
+            .setup_sqpoll(1000) // Enable SQPOLL with 1000ms idle time
+            .build(entries)?,
+    )?;
 
     #[cfg(not(feature = "ci"))]
     {
@@ -181,3 +186,45 @@ fn test<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
 
     Ok(())
 }
+
+fn test_sqpoll<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    mut ring: IoUring<S, C>,
+) -> anyhow::Result<()> {
+    if !ring.params().is_setup_sqpoll() {
+        return Err(anyhow::anyhow!(
+            "IORING_SETUP_SQPOLL must be enabled to run this test"
+        ));
+    }
+
+    let mut probe = Probe::new();
+
+    if ring.submitter().register_probe(&mut probe).is_err() {
+        eprintln!("No probe supported");
+    }
+
+    println!();
+    println!(
+        "ring type: IoUring<{}, {}>",
+        std::any::type_name::<S>()
+            .strip_prefix("io_uring::")
+            .unwrap(),
+        std::any::type_name::<C>()
+            .strip_prefix("io_uring::")
+            .unwrap(),
+    );
+    println!("params: {:#?}", ring.params());
+    println!("probe: {:?}", probe);
+    println!();
+
+    let test = Test {
+        probe,
+        target: std::env::args().nth(1),
+        count: Cell::new(0),
+    };
+
+    tests::sqpoll::test_sqpoll_cq_overflow(&mut ring, &test)?;
+
+    println!("Test count: {}", test.count.get());
+
+    Ok(())
+}

io-uring-test/src/tests/mod.rs

@@ -12,3 +12,5 @@ pub mod register_buffers;
 pub mod register_sync_cancel;
 pub mod regression;
 pub mod timeout;
+
+pub mod sqpoll;

io-uring-test/src/tests/sqpoll.rs

@@ -0,0 +1,109 @@
+use io_uring::{cqueue, opcode, squeue, types, IoUring};
+use std::fs::File;
+use std::io::Write;
+use std::os::unix::io::AsRawFd;
+use std::path::Path;
+use tempfile::tempdir;
+
+use crate::Test;
+
+/// Test to reproduce SQPOLL CQ overflow issue
+///
+/// This test demonstrates the issue when:
+/// 1) SQPOLL is enabled
+/// 2) IORING_FEAT_NODROP feature is supported
+/// 3) We submit more concurrent I/O requests than the CQ can handle
+pub fn test_sqpoll_cq_overflow<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    if !ring.params().is_feature_nodrop() {
+        println!("IORING_FEAT_NODROP is not supported by the kernel, skip");
+        return Ok(());
+    }
+
+    require! {
+        test;
+    }
+
+    println!("test sqpoll_cq_overflow");
+
+    let idle = ring.params().sq_thread_idle();
+    let entries = ring.params().sq_entries() as usize;
+    let num_requests = entries * 4;
+
+    // Create a temporary directory for test files
+    let temp_dir = tempdir().expect("Failed to create temp directory");
+    let test_dir = temp_dir.path();
+
+    // Create test files for I/O operations
+    let test_files = create_test_files(test_dir, num_requests);
+    let mut bufs = (0..num_requests)
+        .map(|_| vec![0u8; 1024])
+        .collect::<Vec<_>>();
+
+    let start = std::time::Instant::now();
+
+    test_files
+        .iter()
+        .zip(bufs.iter_mut())
+        .for_each(|(file, buf)| {
+            let fd = file.as_raw_fd();
+            let entry = opcode::Read::new(types::Fd(fd), buf.as_mut_ptr(), buf.len() as _)
+                .build()
+                .into();
+            while unsafe { ring.submission().push(&entry).is_err() } {
+                ring.submit().expect("Failed to submit");
+            }
+        });
+
+    let mut completed_count = 0;
+    // Try to collect all completions
+    while completed_count < num_requests {
+        while ring.completion().next().is_some() {
+            completed_count += 1;
+        }
+
+        if ring.submission().cq_overflow() {
+            // Call `io_uring_enter` to make the kernel flush overflowed completions
+            println!("CQ overflow, syscall to flush");
+            ring.submit().expect("Failed to submit");
+        }
+    }
+
+    let end = start.elapsed();
+
+    assert_eq!(completed_count, num_requests);
+    // After idle time, `IORING_SQ_NEED_WAKEUP` flag will be set, and `io_uring_enter` will be called.
+    // That call will flush the overflowed completions eventually.
+    // To distinguish this case from the case where the fix of PR #349 is applied,
+    // we need to check the elapsed time.
+    assert!(
+        end.as_millis() < idle as _,
+        "Overflown completions should be flushed within the idle time"
+    );
+
+    Ok(())
+}
+
+/// Create test files for I/O operations
+fn create_test_files(dir: &Path, count: usize) -> Vec<File> {
+    let mut files = Vec::new();
+
+    for i in 0..count {
+        let file_path = dir.join(format!("test_file_{}.txt", i));
+        let mut file = File::create(&file_path).expect("Failed to create test file");
+
+        // Write some data to the file
+        let content = format!("Test content for file {}", i);
+        file.write_all(content.as_bytes())
+            .expect("Failed to write to test file");
+        file.flush().expect("Failed to flush file");
+
+        // Reopen for reading
+        let read_file = File::open(&file_path).expect("Failed to open test file for reading");
+        files.push(read_file);
+    }
+
+    files
+}

src/lib.rs

@@ -602,6 +602,11 @@ impl Parameters {
         self.0.sq_entries
     }
 
+    /// The idle time of the SQ poll thread in milliseconds.
+    pub fn sq_thread_idle(&self) -> u32 {
+        self.0.sq_thread_idle
+    }
+
     /// The number of completion queue entries allocated.
     pub fn cq_entries(&self) -> u32 {
         self.0.cq_entries

src/submit.rs

@@ -123,7 +123,18 @@ impl<'a> Submitter<'a> {
         // the IORING_ENTER_SQ_WAKEUP bit is required in all paths where sqpoll
         // is setup when consolidating the reads.
 
-        if want > 0 || self.params.is_setup_iopoll() || self.sq_cq_overflow() {
+        let sq_cq_overflow = self.sq_cq_overflow();
+
+        // When IORING_FEAT_NODROP is enabled and CQ overflows, the kernel buffers
+        // completion events internally but doesn't automatically flush them when
+        // CQ space becomes available. We must explicitly call io_uring_enter()
+        // to flush these buffered events, even with SQPOLL enabled.
+        //
+        // Without this, completions remain stuck in kernel's internal buffer
+        // after draining CQ, causing missing completion notifications.
+        let need_syscall_for_overflow = sq_cq_overflow && self.params.is_feature_nodrop();
+
+        if want > 0 || self.params.is_setup_iopoll() || sq_cq_overflow {
             flags |= sys::IORING_ENTER_GETEVENTS;
         }
 
@@ -132,9 +143,12 @@ impl<'a> Submitter<'a> {
             atomic::fence(atomic::Ordering::SeqCst);
             if self.sq_need_wakeup() {
                 flags |= sys::IORING_ENTER_SQ_WAKEUP;
-            } else if want == 0 {
+            } else if want == 0 && !need_syscall_for_overflow {
                 // The kernel thread is polling and hasn't fallen asleep, so we don't need to tell
                 // it to process events or wake it up
+
+                // However, if the CQ ring is overflown, we need to tell the kernel to process events
+                // by calling io_uring_enter with the IORING_ENTER_GETEVENTS flag.
                 return Ok(len);
             }
         }
@@ -155,7 +169,10 @@ impl<'a> Submitter<'a> {
         let len = self.sq_len();
         let mut flags = sys::IORING_ENTER_EXT_ARG;
 
-        if want > 0 || self.params.is_setup_iopoll() || self.sq_cq_overflow() {
+        let sq_cq_overflow = self.sq_cq_overflow();
+        let need_syscall = sq_cq_overflow & self.params.is_feature_nodrop();
+
+        if want > 0 || self.params.is_setup_iopoll() || sq_cq_overflow {
             flags |= sys::IORING_ENTER_GETEVENTS;
         }
 
@@ -164,7 +181,7 @@ impl<'a> Submitter<'a> {
             atomic::fence(atomic::Ordering::SeqCst);
             if self.sq_need_wakeup() {
                 flags |= sys::IORING_ENTER_SQ_WAKEUP;
-            } else if want == 0 {
+            } else if want == 0 && !need_syscall {
                 // The kernel thread is polling and hasn't fallen asleep, so we don't need to tell
                 // it to process events or wake it up
                 return Ok(len);