Merge pull request #200 from Berrysoft/refactor/simplify-attacher

refactor: simplify attach logic

Author: Berrysoft

.github/workflows/ci_test.yml

@@ -43,6 +43,9 @@ jobs:
             target: "x86_64-pc-windows-gnu"
           - os: "windows-latest"
             target: "i686-pc-windows-msvc"
+          - os: "windows-latest"
+            target: "x86_64-pc-windows-msvc"
+            features: "iocp-global"
           - os: "macos-12"
           - os: "macos-13"
           - os: "macos-14"

compio-dispatcher/tests/listener.rs

@@ -1,10 +1,10 @@
 use std::{num::NonZeroUsize, panic::resume_unwind};
 
-use compio_buf::{arrayvec::ArrayVec, IntoInner};
+use compio_buf::arrayvec::ArrayVec;
 use compio_dispatcher::Dispatcher;
 use compio_io::{AsyncReadExt, AsyncWriteExt};
 use compio_net::{TcpListener, TcpStream};
-use compio_runtime::{spawn, Unattached};
+use compio_runtime::spawn;
 use futures_util::{stream::FuturesUnordered, StreamExt};
 
 #[compio_macros::test]
@@ -27,15 +27,11 @@ async fn listener_dispatch() {
     });
     let mut handles = FuturesUnordered::new();
     for _i in 0..CLIENT_NUM {
-        let (srv, _) = listener.accept().await.unwrap();
-        let srv = Unattached::new(srv).unwrap();
+        let (mut srv, _) = listener.accept().await.unwrap();
         let handle = dispatcher
-            .dispatch(move || {
-                let mut srv = srv.into_inner();
-                async move {
-                    let (_, buf) = srv.read_exact(ArrayVec::<u8, 12>::new()).await.unwrap();
-                    assert_eq!(buf.as_slice(), b"Hello world!");
-                }
+            .dispatch(move || async move {
+                let (_, buf) = srv.read_exact(ArrayVec::<u8, 12>::new()).await.unwrap();
+                assert_eq!(buf.as_slice(), b"Hello world!");
             })
             .unwrap();
         handles.push(handle.join());

compio-driver/Cargo.toml

@@ -70,6 +70,7 @@ polling = "3.3.0"
 os_pipe = { workspace = true }
 
 [target.'cfg(unix)'.dependencies]
+crossbeam-channel = { workspace = true }
 crossbeam-queue = { workspace = true }
 libc = { workspace = true }
 
@@ -83,6 +84,8 @@ polling = ["dep:polling", "dep:os_pipe"]
 io-uring-sqe128 = []
 io-uring-cqe32 = []
 
+iocp-global = []
+
 # Nightly features
 once_cell_try = []
 nightly = ["once_cell_try"]

compio-driver/src/fusion/mod.rs

@@ -132,6 +132,13 @@ impl Driver {
         }
     }
 
+    pub fn create_op<T: OpCode + 'static>(&self, user_data: usize, op: T) -> RawOp {
+        match &self.fuse {
+            FuseDriver::Poll(driver) => driver.create_op(user_data, op),
+            FuseDriver::IoUring(driver) => driver.create_op(user_data, op),
+        }
+    }
+
     pub fn attach(&mut self, fd: RawFd) -> io::Result<()> {
         match &mut self.fuse {
             FuseDriver::Poll(driver) => driver.attach(fd),

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

@@ -0,0 +1,149 @@
+#[cfg(feature = "once_cell_try")]
+use std::sync::OnceLock;
+use std::{
+    io,
+    os::windows::io::{AsRawHandle, RawHandle},
+    time::Duration,
+};
+
+use compio_log::*;
+#[cfg(not(feature = "once_cell_try"))]
+use once_cell::sync::OnceCell as OnceLock;
+use windows_sys::Win32::System::IO::PostQueuedCompletionStatus;
+
+use super::CompletionPort;
+use crate::{syscall, Entry, Overlapped, RawFd};
+
+struct GlobalPort {
+    port: CompletionPort,
+}
+
+impl GlobalPort {
+    pub fn new() -> io::Result<Self> {
+        Ok(Self {
+            port: CompletionPort::new()?,
+        })
+    }
+
+    pub fn attach(&self, fd: RawFd) -> io::Result<()> {
+        self.port.attach(fd)
+    }
+
+    pub fn post<T: ?Sized>(
+        &self,
+        res: io::Result<usize>,
+        optr: *mut Overlapped<T>,
+    ) -> io::Result<()> {
+        self.port.post(res, optr)
+    }
+
+    pub fn post_raw<T: ?Sized>(&self, optr: *const Overlapped<T>) -> io::Result<()> {
+        self.port.post_raw(optr)
+    }
+}
+
+impl AsRawHandle for GlobalPort {
+    fn as_raw_handle(&self) -> RawHandle {
+        self.port.as_raw_handle()
+    }
+}
+
+static IOCP_PORT: OnceLock<GlobalPort> = OnceLock::new();
+
+#[inline]
+fn iocp_port() -> io::Result<&'static GlobalPort> {
+    IOCP_PORT.get_or_try_init(GlobalPort::new)
+}
+
+fn iocp_start() -> io::Result<()> {
+    let port = iocp_port()?;
+    std::thread::spawn(move || {
+        instrument!(compio_log::Level::TRACE, "iocp_start");
+        loop {
+            for entry in port.port.poll_raw(None)? {
+                // 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 Err(_e) = syscall!(
+                    BOOL,
+                    PostQueuedCompletionStatus(
+                        overlapped.driver as _,
+                        entry.dwNumberOfBytesTransferred,
+                        entry.lpCompletionKey,
+                        entry.lpOverlapped,
+                    )
+                ) {
+                    error!(
+                        "fail to dispatch entry ({}, {}, {:p}) to driver {:p}: {:?}",
+                        entry.dwNumberOfBytesTransferred,
+                        entry.lpCompletionKey,
+                        entry.lpOverlapped,
+                        overlapped.driver,
+                        _e
+                    );
+                }
+            }
+        }
+        #[allow(unreachable_code)]
+        io::Result::Ok(())
+    });
+    Ok(())
+}
+
+static IOCP_INIT_ONCE: OnceLock<()> = OnceLock::new();
+
+pub struct Port {
+    port: CompletionPort,
+    global_port: &'static GlobalPort,
+}
+
+impl Port {
+    pub fn new() -> io::Result<Self> {
+        IOCP_INIT_ONCE.get_or_try_init(iocp_start)?;
+
+        Ok(Self {
+            port: CompletionPort::new()?,
+            global_port: iocp_port()?,
+        })
+    }
+
+    pub fn attach(&mut self, fd: RawFd) -> io::Result<()> {
+        self.global_port.attach(fd)
+    }
+
+    pub fn handle(&self) -> PortHandle {
+        PortHandle::new(self.global_port)
+    }
+
+    pub fn poll(&self, timeout: Option<Duration>) -> io::Result<impl Iterator<Item = Entry> + '_> {
+        self.port.poll(timeout, None)
+    }
+}
+
+impl AsRawHandle for Port {
+    fn as_raw_handle(&self) -> RawHandle {
+        self.port.as_raw_handle()
+    }
+}
+
+pub struct PortHandle {
+    port: &'static GlobalPort,
+}
+
+impl PortHandle {
+    fn new(port: &'static GlobalPort) -> Self {
+        Self { port }
+    }
+
+    pub fn post<T: ?Sized>(
+        &self,
+        res: io::Result<usize>,
+        optr: *mut Overlapped<T>,
+    ) -> io::Result<()> {
+        self.port.post(res, optr)
+    }
+
+    pub fn post_raw<T: ?Sized>(&self, optr: *const Overlapped<T>) -> io::Result<()> {
+        self.port.post_raw(optr)
+    }
+}