basic support IOCP Operator

Author: loongs-zhang

.github/workflows/ci-preemptive.sh

@@ -34,3 +34,13 @@ if [ "${TARGET}" = "x86_64-unknown-linux-gnu" ]; then
     "${CARGO}" test --target "${TARGET}" --no-default-features --features io_uring,preemptive,ci
     "${CARGO}" test --target "${TARGET}" --no-default-features --features io_uring,preemptive,ci --release
 fi
+
+# test IOCP
+if [ "${OS}" = "windows-latest" ]; then
+    cd "${PROJECT_DIR}"/core
+    "${CARGO}" test --target "${TARGET}" --no-default-features --features iocp,preemptive,ci
+    "${CARGO}" test --target "${TARGET}" --no-default-features --features iocp,preemptive,ci --release
+    cd "${PROJECT_DIR}"/open-coroutine
+    "${CARGO}" test --target "${TARGET}" --no-default-features --features iocp,preemptive,ci
+    "${CARGO}" test --target "${TARGET}" --no-default-features --features iocp,preemptive,ci --release
+fi

.github/workflows/ci.sh

@@ -34,3 +34,13 @@ if [ "${TARGET}" = "x86_64-unknown-linux-gnu" ]; then
     "${CARGO}" test --target "${TARGET}" --no-default-features --features io_uring,ci
     "${CARGO}" test --target "${TARGET}" --no-default-features --features io_uring,ci --release
 fi
+
+# test IOCP
+if [ "${OS}" = "windows-latest" ]; then
+    cd "${PROJECT_DIR}"/core
+    "${CARGO}" test --target "${TARGET}" --no-default-features --features iocp,ci
+    "${CARGO}" test --target "${TARGET}" --no-default-features --features iocp,ci --release
+    cd "${PROJECT_DIR}"/open-coroutine
+    "${CARGO}" test --target "${TARGET}" --no-default-features --features iocp,ci
+    "${CARGO}" test --target "${TARGET}" --no-default-features --features iocp,ci --release
+fi

core/Cargo.toml

@@ -58,6 +58,7 @@ windows-sys = { workspace = true, features = [
     "Win32_Networking_WinSock",
     "Win32_System_SystemInformation",
     "Win32_System_Diagnostics_Debug",
+    "Win32_System_WindowsProgramming",
 ] }
 polling = { workspace = true, optional = true }
 
@@ -95,5 +96,11 @@ net = ["korosensei", "polling", "mio", "crossbeam-utils", "core_affinity"]
 # Provide io_uring adaptation, this feature only works in linux.
 io_uring = ["net", "io-uring"]
 
+# Provide IOCP adaptation, this feature only works in windows.
+iocp = ["net"]
+
+# Provide completion IO adaptation
+completion_io = ["io_uring", "iocp"]
+
 # Provide syscall implementation.
 syscall = ["net"]

core/src/net/event_loop.rs

@@ -24,16 +24,35 @@ cfg_if::cfg_if! {
     }
 }
 
+cfg_if::cfg_if! {
+    if #[cfg(all(windows, feature = "iocp"))] {
+        use dashmap::DashMap;
+        use std::ffi::{c_longlong, c_uint};
+        use windows_sys::core::{PCSTR, PSTR};
+        use windows_sys::Win32::Networking::WinSock::{
+            setsockopt, LPWSAOVERLAPPED_COMPLETION_ROUTINE, SEND_RECV_FLAGS, SOCKADDR, SOCKET, SOL_SOCKET,
+            SO_UPDATE_ACCEPT_CONTEXT, WSABUF,
+        };
+        use windows_sys::Win32::System::IO::OVERLAPPED;
+    }
+}
+
 #[repr(C)]
 #[derive(Debug)]
 pub(crate) struct EventLoop<'e> {
     stop: Arc<(Mutex<bool>, Condvar)>,
     shared_stop: Arc<(Mutex<AtomicUsize>, Condvar)>,
     cpu: usize,
-    #[cfg(all(target_os = "linux", feature = "io_uring"))]
+    #[cfg(any(
+        all(target_os = "linux", feature = "io_uring"),
+        all(windows, feature = "iocp")
+    ))]
     operator: crate::net::operator::Operator<'e>,
     #[allow(clippy::type_complexity)]
-    #[cfg(all(target_os = "linux", feature = "io_uring"))]
+    #[cfg(any(
+        all(target_os = "linux", feature = "io_uring"),
+        all(windows, feature = "iocp")
+    ))]
     syscall_wait_table: DashMap<usize, Arc<(Mutex<Option<c_longlong>>, Condvar)>>,
     selector: Poller,
     pool: CoroutinePool<'e>,
@@ -87,9 +106,15 @@ impl<'e> EventLoop<'e> {
             stop: Arc::new((Mutex::new(false), Condvar::new())),
             shared_stop,
             cpu,
-            #[cfg(all(target_os = "linux", feature = "io_uring"))]
+            #[cfg(any(
+                all(target_os = "linux", feature = "io_uring"),
+                all(windows, feature = "iocp")
+            ))]
             operator: crate::net::operator::Operator::new(cpu)?,
-            #[cfg(all(target_os = "linux", feature = "io_uring"))]
+            #[cfg(any(
+                all(target_os = "linux", feature = "io_uring"),
+                all(windows, feature = "iocp")
+            ))]
             syscall_wait_table: DashMap::new(),
             selector: Poller::new()?,
             pool: CoroutinePool::new(name, stack_size, min_size, max_size, keep_alive_time),
@@ -222,6 +247,8 @@ impl<'e> EventLoop<'e> {
         cfg_if::cfg_if! {
             if #[cfg(all(target_os = "linux", feature = "io_uring"))] {
                 left_time = self.adapt_io_uring(left_time)?;
+            } else if #[cfg(all(windows, feature = "iocp"))] {
+                left_time = self.adapt_iocp(left_time)?;
             }
         }
 
@@ -267,6 +294,52 @@ impl<'e> EventLoop<'e> {
         Ok(left_time)
     }
 
+    #[cfg(all(windows, feature = "iocp"))]
+    fn adapt_iocp(&self, mut left_time: Option<Duration>) -> std::io::Result<Option<Duration>> {
+        // use IOCP
+        let (count, mut cq, left) = self.operator.select(left_time, 0)?;
+        if count > 0 {
+            for cqe in &mut cq {
+                let token = cqe.token;
+                let bytes_transferred = cqe.bytes_transferred;
+                // resolve completed read/write tasks
+                // todo refactor IOCP impl
+                let result = match cqe.syscall_name {
+                    SyscallName::accept => unsafe {
+                        if setsockopt(
+                            cqe.socket,
+                            SOL_SOCKET,
+                            SO_UPDATE_ACCEPT_CONTEXT,
+                            std::ptr::from_ref(&cqe.from_fd).cast(),
+                            c_int::try_from(size_of::<SOCKET>()).expect("overflow"),
+                        ) == 0
+                        {
+                            cqe.socket.try_into().expect("result overflow")
+                        } else {
+                            -c_longlong::from(windows_sys::Win32::Foundation::GetLastError())
+                        }
+                    },
+                    SyscallName::recv
+                    | SyscallName::WSARecv
+                    | SyscallName::send
+                    | SyscallName::WSASend => bytes_transferred.into(),
+                    _ => panic!("unsupported"),
+                };
+                if let Some((_, pair)) = self.syscall_wait_table.remove(&token) {
+                    let (lock, cvar) = &*pair;
+                    let mut pending = lock.lock().expect("lock failed");
+                    *pending = Some(result);
+                    cvar.notify_one();
+                }
+                unsafe { self.resume(token) };
+            }
+        }
+        if left != left_time {
+            left_time = Some(left.unwrap_or(Duration::ZERO));
+        }
+        Ok(left_time)
+    }
+
     unsafe fn resume(&self, token: usize) {
         if COROUTINE_TOKENS.remove(&token).is_none() {
             return;
@@ -446,6 +519,34 @@ impl_io_uring!(mkdirat(dirfd: c_int, pathname: *const c_char, mode: mode_t) -> c
 impl_io_uring!(renameat(olddirfd: c_int, oldpath: *const c_char, newdirfd: c_int, newpath: *const c_char) -> c_int);
 impl_io_uring!(renameat2(olddirfd: c_int, oldpath: *const c_char, newdirfd: c_int, newpath: *const c_char, flags: c_uint) -> c_int);
 
+macro_rules! impl_iocp {
+    ( $syscall: ident($($arg: ident : $arg_type: ty),*) -> $result: ty ) => {
+        #[cfg(all(windows, feature = "iocp"))]
+        impl EventLoop<'_> {
+            #[allow(non_snake_case, clippy::too_many_arguments)]
+            pub(super) fn $syscall(
+                &self,
+                $($arg: $arg_type),*
+            ) -> std::io::Result<Arc<(Mutex<Option<c_longlong>>, Condvar)>> {
+                let token = EventLoop::token(SyscallName::$syscall);
+                self.operator.$syscall(token, $($arg, )*)?;
+                let arc = Arc::new((Mutex::new(None), Condvar::new()));
+                assert!(
+                    self.syscall_wait_table.insert(token, arc.clone()).is_none(),
+                    "The previous token was not retrieved in a timely manner"
+                );
+                Ok(arc)
+            }
+        }
+    }
+}
+
+impl_iocp!(accept(fd: SOCKET, addr: *mut SOCKADDR, len: *mut c_int) -> c_int);
+impl_iocp!(recv(fd: SOCKET, buf: PSTR, len: c_int, flags: SEND_RECV_FLAGS) -> c_int);
+impl_iocp!(WSARecv(fd: SOCKET, buf: *const WSABUF, dwbuffercount: c_uint, lpnumberofbytesrecvd: *mut c_uint, lpflags : *mut c_uint, lpoverlapped: *mut OVERLAPPED, lpcompletionroutine : LPWSAOVERLAPPED_COMPLETION_ROUTINE) -> c_int);
+impl_iocp!(send(fd: SOCKET, buf: PCSTR, len: c_int, flags: SEND_RECV_FLAGS) -> c_int);
+impl_iocp!(WSASend(fd: SOCKET, buf: *const WSABUF, dwbuffercount: c_uint, lpnumberofbytesrecvd: *mut c_uint, dwflags : c_uint, lpoverlapped: *mut OVERLAPPED, lpcompletionroutine : LPWSAOVERLAPPED_COMPLETION_ROUTINE) -> c_int);
+
 #[cfg(all(test, not(all(unix, feature = "preemptive"))))]
 mod tests {
     use crate::net::event_loop::EventLoop;

core/src/net/mod.rs

@@ -18,13 +18,27 @@ cfg_if::cfg_if! {
     }
 }
 
+cfg_if::cfg_if! {
+    if #[cfg(all(windows, feature = "iocp"))] {
+        use std::ffi::c_uint;
+        use windows_sys::core::{PCSTR, PSTR};
+        use windows_sys::Win32::Networking::WinSock::{
+            LPWSAOVERLAPPED_COMPLETION_ROUTINE, SEND_RECV_FLAGS, SOCKADDR, SOCKET, WSABUF,
+        };
+        use windows_sys::Win32::System::IO::OVERLAPPED;
+    }
+}
+
 /// 做C兼容时会用到
 pub type UserFunc = extern "C" fn(usize) -> usize;
 
 mod selector;
 
 #[allow(clippy::too_many_arguments)]
-#[cfg(all(target_os = "linux", feature = "io_uring"))]
+#[cfg(any(
+    all(target_os = "linux", feature = "io_uring"),
+    all(windows, feature = "iocp")
+))]
 mod operator;
 
 #[allow(missing_docs)]
@@ -280,3 +294,24 @@ impl_io_uring!(fsync(fd: c_int) -> c_int);
 impl_io_uring!(mkdirat(dirfd: c_int, pathname: *const c_char, mode: mode_t) -> c_int);
 impl_io_uring!(renameat(olddirfd: c_int, oldpath: *const c_char, newdirfd: c_int, newpath: *const c_char) -> c_int);
 impl_io_uring!(renameat2(olddirfd: c_int, oldpath: *const c_char, newdirfd: c_int, newpath: *const c_char, flags: c_uint) -> c_int);
+
+macro_rules! impl_iocp {
+    ( $syscall: ident($($arg: ident : $arg_type: ty),*) -> $result: ty ) => {
+        #[allow(non_snake_case)]
+        #[cfg(all(windows, feature = "iocp"))]
+        impl EventLoops {
+            #[allow(missing_docs)]
+            pub fn $syscall(
+                $($arg: $arg_type),*
+            ) -> std::io::Result<Arc<(Mutex<Option<c_longlong>>, Condvar)>> {
+                Self::event_loop().$syscall($($arg, )*)
+            }
+        }
+    }
+}
+
+impl_iocp!(accept(fd: SOCKET, addr: *mut SOCKADDR, len: *mut c_int) -> c_int);
+impl_iocp!(recv(fd: SOCKET, buf: PSTR, len: c_int, flags: SEND_RECV_FLAGS) -> c_int);
+impl_iocp!(WSARecv(fd: SOCKET, buf: *const WSABUF, dwbuffercount: c_uint, lpnumberofbytesrecvd: *mut c_uint, lpflags : *mut c_uint, lpoverlapped: *mut OVERLAPPED, lpcompletionroutine : LPWSAOVERLAPPED_COMPLETION_ROUTINE) -> c_int);
+impl_iocp!(send(fd: SOCKET, buf: PCSTR, len: c_int, flags: SEND_RECV_FLAGS) -> c_int);
+impl_iocp!(WSASend(fd: SOCKET, buf: *const WSABUF, dwbuffercount: c_uint, lpnumberofbytesrecvd: *mut c_uint, dwflags : c_uint, lpoverlapped: *mut OVERLAPPED, lpcompletionroutine : LPWSAOVERLAPPED_COMPLETION_ROUTINE) -> c_int);

core/src/net/operator/mod.rs

@@ -2,3 +2,9 @@
 mod linux;
 #[cfg(all(target_os = "linux", feature = "io_uring"))]
 pub(crate) use linux::*;
+
+#[allow(non_snake_case)]
+#[cfg(all(windows, feature = "iocp"))]
+mod windows;
+#[cfg(all(windows, feature = "iocp"))]
+pub(crate) use windows::*;