fix(driver): repost entry from other IOCP

Author: Berrysoft

compio-driver/src/iocp/cp/global.rs

@@ -74,7 +74,7 @@ fn iocp_start() -> io::Result<()> {
     let port = iocp_port()?;
     std::thread::spawn(move || {
         loop {
-            for (driver, entry) in port.port.poll(None)? {
+            for (driver, entry) in port.port.poll(None, None)? {
                 port.push(driver, entry);
             }
         }

compio-driver/src/iocp/cp/mod.rs

@@ -86,9 +86,12 @@ impl CompletionPort {
         Ok(())
     }
 
+    // If current_driver is specified, any entry that doesn't belong the driver will
+    // be reposted. The driver id will be used as IOCP handle.
     pub fn poll(
         &self,
         timeout: Option<Duration>,
+        current_driver: Option<usize>,
     ) -> io::Result<impl Iterator<Item = (usize, Entry)>> {
         const DEFAULT_CAPACITY: usize = 1024;
 
@@ -112,12 +115,26 @@ impl CompletionPort {
         trace!("recv_count: {recv_count}");
         unsafe { entries.set_len(recv_count as _) };
 
-        Ok(entries.into_iter().map(|entry| {
+        Ok(entries.into_iter().map(move |entry| {
             let transferred = entry.dwNumberOfBytesTransferred;
             trace!("entry transferred: {transferred}");
             // Any thin pointer is OK because we don't use the type of opcode.
             let overlapped_ptr: *mut Overlapped<()> = entry.lpOverlapped.cast();
             let overlapped = unsafe { &*overlapped_ptr };
+            if let Some(current_driver) = current_driver {
+                if overlapped.driver != current_driver {
+                    syscall!(
+                        BOOL,
+                        PostQueuedCompletionStatus(
+                            overlapped.driver as _,
+                            entry.dwNumberOfBytesTransferred,
+                            entry.lpCompletionKey,
+                            entry.lpOverlapped
+                        )
+                    )
+                    .ok();
+                }
+            }
             let res = if matches!(
                 overlapped.base.Internal as NTSTATUS,
                 STATUS_SUCCESS | STATUS_PENDING

compio-driver/src/iocp/cp/multi.rs

@@ -28,7 +28,7 @@ impl Port {
 
     pub fn poll(&self, timeout: Option<Duration>) -> io::Result<impl Iterator<Item = Entry> + '_> {
         let current_id = self.id();
-        self.port.poll(timeout).map(move |it| {
+        self.port.poll(timeout, Some(current_id)).map(move |it| {
             it.filter_map(
                 move |(id, entry)| {
                     if id == current_id { Some(entry) } else { None }

compio/Cargo.toml

@@ -73,7 +73,6 @@ libc = { workspace = true }
 default = ["runtime", "io-uring"]
 io-uring = ["compio-driver/io-uring"]
 polling = ["compio-driver/polling"]
-iocp-global = ["compio-driver/iocp-global"]
 io = ["dep:compio-io"]
 io-compat = ["io", "compio-io/compat"]
 runtime = ["dep:compio-runtime", "dep:compio-fs", "dep:compio-net", "io"]

compio/tests/runtime.rs

@@ -7,12 +7,11 @@ use compio::{
     fs::File,
     io::{AsyncReadAt, AsyncReadExt, AsyncWriteAt, AsyncWriteExt},
     net::{TcpListener, TcpStream},
+    runtime::TryClone,
 };
-use compio_runtime::TryClone;
 use tempfile::NamedTempFile;
 
 #[compio_macros::test]
-#[cfg(any(not(windows), feature = "iocp-global"))]
 async fn multi_threading() {
     const DATA: &str = "Hello world!";
 
@@ -29,16 +28,13 @@ async fn multi_threading() {
     assert_eq!(n, buffer.len());
     assert_eq!(DATA, String::from_utf8(buffer).unwrap());
 
-    if let Err(e) = std::thread::spawn(move || {
+    compio::runtime::spawn_blocking(move || {
         compio::runtime::Runtime::new().unwrap().block_on(async {
             let ((), buffer) = rx.read_exact(Vec::with_capacity(DATA.len())).await.unwrap();
             assert_eq!(DATA, String::from_utf8(buffer).unwrap());
         });
     })
-    .join()
-    {
-        std::panic::resume_unwind(e)
-    }
+    .await
 }
 
 #[compio_macros::test]