feat(net,win): make socket & shutdown sync (#789)

Author: Berrysoft

compio-driver/src/op.rs

@@ -4,7 +4,7 @@
 //! The operation itself doesn't perform anything.
 //! You need to pass them to [`crate::Proactor`], and poll the driver.
 
-use std::{io, marker::PhantomPinned, mem::ManuallyDrop, net::Shutdown};
+use std::{io, marker::PhantomPinned, mem::ManuallyDrop};
 
 use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut, IoVectoredBuf, SetLen};
 use pin_project_lite::pin_project;
@@ -19,8 +19,8 @@ pub use crate::sys::op::{
 #[cfg(unix)]
 pub use crate::sys::op::{
     AcceptMulti, CreateDir, CreateSocket, CurrentDir, FileStat, HardLink, Interest, OpenFile,
-    PathStat, PollOnce, ReadVectored, ReadVectoredAt, Rename, Stat, Symlink, TruncateFile, Unlink,
-    WriteVectored, WriteVectoredAt,
+    PathStat, PollOnce, ReadVectored, ReadVectoredAt, Rename, ShutdownSocket, Stat, Symlink,
+    TruncateFile, Unlink, WriteVectored, WriteVectoredAt,
 };
 #[cfg(windows)]
 pub use crate::sys::op::{ConnectNamedPipe, DeviceIoControl};
@@ -446,19 +446,6 @@ impl<S> Sync<S> {
     }
 }
 
-/// Shutdown a socket.
-pub struct ShutdownSocket<S> {
-    pub(crate) fd: S,
-    pub(crate) how: Shutdown,
-}
-
-impl<S> ShutdownSocket<S> {
-    /// Create [`ShutdownSocket`].
-    pub fn new(fd: S, how: Shutdown) -> Self {
-        Self { fd, how }
-    }
-}
-
 /// Close socket fd.
 pub struct CloseSocket {
     pub(crate) fd: ManuallyDrop<OwnedFd>,

compio-driver/src/sys/iocp/op.rs

@@ -3,7 +3,6 @@ use std::sync::OnceLock;
 use std::{
     io,
     marker::PhantomPinned,
-    net::Shutdown,
     os::windows::io::AsRawSocket,
     pin::Pin,
     ptr::{null, null_mut, read_unaligned},
@@ -24,11 +23,11 @@ use windows_sys::{
         },
         Networking::WinSock::{
             CMSGHDR, LPFN_ACCEPTEX, LPFN_CONNECTEX, LPFN_GETACCEPTEXSOCKADDRS, LPFN_WSARECVMSG,
-            SD_BOTH, SD_RECEIVE, SD_SEND, SIO_GET_EXTENSION_FUNCTION_POINTER,
-            SO_UPDATE_ACCEPT_CONTEXT, SO_UPDATE_CONNECT_CONTEXT, SOCKADDR, SOCKADDR_STORAGE,
-            SOL_SOCKET, WSAID_ACCEPTEX, WSAID_CONNECTEX, WSAID_GETACCEPTEXSOCKADDRS,
-            WSAID_WSARECVMSG, WSAIoctl, WSAMSG, WSARecv, WSARecvFrom, WSASend, WSASendMsg,
-            WSASendTo, closesocket, setsockopt, shutdown, socklen_t,
+            SIO_GET_EXTENSION_FUNCTION_POINTER, SO_UPDATE_ACCEPT_CONTEXT,
+            SO_UPDATE_CONNECT_CONTEXT, SOCKADDR, SOCKADDR_STORAGE, SOL_SOCKET, WSAID_ACCEPTEX,
+            WSAID_CONNECTEX, WSAID_GETACCEPTEXSOCKADDRS, WSAID_WSARECVMSG, WSAIoctl, WSAMSG,
+            WSARecv, WSARecvFrom, WSASend, WSASendMsg, WSASendTo, closesocket, setsockopt,
+            socklen_t,
         },
         Storage::FileSystem::{FlushFileBuffers, ReadFile, WriteFile},
         System::{
@@ -320,23 +319,6 @@ unsafe impl<S: AsFd> OpCode for Sync<S> {
     }
 }
 
-unsafe impl<S: AsFd> OpCode for ShutdownSocket<S> {
-    fn op_type(&self) -> OpType {
-        OpType::Blocking
-    }
-
-    unsafe fn operate(self: Pin<&mut Self>, _optr: *mut OVERLAPPED) -> Poll<io::Result<usize>> {
-        let how = match self.how {
-            Shutdown::Write => SD_SEND,
-            Shutdown::Read => SD_RECEIVE,
-            Shutdown::Both => SD_BOTH,
-        };
-        Poll::Ready(Ok(
-            syscall!(SOCKET, shutdown(self.fd.as_fd().as_raw_fd() as _, how))? as _,
-        ))
-    }
-}
-
 unsafe impl OpCode for CloseSocket {
     fn op_type(&self) -> OpType {
         OpType::Blocking

compio-driver/src/sys/unix_op.rs

@@ -504,6 +504,19 @@ impl IntoInner for CreateSocket {
     }
 }
 
+/// Shutdown a socket.
+pub struct ShutdownSocket<S> {
+    pub(crate) fd: S,
+    pub(crate) how: Shutdown,
+}
+
+impl<S> ShutdownSocket<S> {
+    /// Create [`ShutdownSocket`].
+    pub fn new(fd: S, how: Shutdown) -> Self {
+        Self { fd, how }
+    }
+}
+
 impl<S: AsFd> ShutdownSocket<S> {
     pub(crate) fn how(&self) -> i32 {
         match self.how {

compio-net/src/socket.rs

@@ -6,14 +6,14 @@ use std::{
 
 use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut, IoVectoredBuf, IoVectoredBufMut, buf_try};
 #[cfg(unix)]
-use compio_driver::op::CreateSocket;
+use compio_driver::op::{CreateSocket, ShutdownSocket};
 use compio_driver::{
     AsRawFd, OpCode, ToSharedFd, impl_raw_fd,
     op::{
         Accept, BufResultExt, CloseSocket, Connect, Recv, RecvFrom, RecvFromManaged,
         RecvFromVectored, RecvManaged, RecvMsg, RecvResultExt, RecvVectored, ResultTakeBuffer,
         Send, SendMsg, SendMsgZc, SendTo, SendToVectored, SendToVectoredZc, SendToZc, SendVectored,
-        SendVectoredZc, SendZc, ShutdownSocket, VecBufResultExt,
+        SendVectoredZc, SendZc, VecBufResultExt,
     },
     syscall,
 };
@@ -68,12 +68,7 @@ impl Socket {
 
     #[cfg(windows)]
     pub async fn new(domain: Domain, ty: Type, protocol: Option<Protocol>) -> io::Result<Self> {
-        use std::panic::resume_unwind;
-
-        let socket = compio_runtime::spawn_blocking(move || Socket2::new(domain, ty, protocol))
-            .await
-            .unwrap_or_else(|e| resume_unwind(e))?;
-        Self::from_socket2(socket)
+        Self::from_socket2(Socket2::new(domain, ty, protocol)?)
     }
 
     #[cfg(unix)]
@@ -121,16 +116,10 @@ impl Socket {
 
     #[cfg(windows)]
     pub async fn accept(&self) -> io::Result<(Self, SockAddr)> {
-        use std::panic::resume_unwind;
-
         let domain = self.local_addr()?.domain();
-        // We should allow users sending this accepted socket to a new thread.
         let ty = self.socket.r#type()?;
         let protocol = self.socket.protocol()?;
-        let accept_sock =
-            compio_runtime::spawn_blocking(move || Socket2::new(domain, ty, protocol))
-                .await
-                .unwrap_or_else(|e| resume_unwind(e))?;
+        let accept_sock = Socket2::new(domain, ty, protocol)?;
         let op = Accept::new(self.to_shared_fd(), accept_sock);
         let (_, op) = buf_try!(@try compio_runtime::submit(op).await);
         op.update_context()?;
@@ -161,12 +150,19 @@ impl Socket {
         }
     }
 
+    #[cfg(unix)]
     pub async fn shutdown(&self) -> io::Result<()> {
         let op = ShutdownSocket::new(self.to_shared_fd(), std::net::Shutdown::Write);
         compio_runtime::submit(op).await.0?;
         Ok(())
     }
 
+    #[cfg(windows)]
+    pub async fn shutdown(&self) -> io::Result<()> {
+        self.socket.shutdown(std::net::Shutdown::Write)?;
+        Ok(())
+    }
+
     pub async fn recv<B: IoBufMut>(&self, buffer: B, flags: i32) -> BufResult<usize, B> {
         let fd = self.to_shared_fd();
         let op = Recv::new(fd, buffer, flags);