Adds support for UnixStream and UnixListener on Windows

Cargo.toml

@@ -48,13 +48,17 @@ log = "0.4.8"
 [target.'cfg(unix)'.dependencies]
 libc = "0.2.121"
 
+[target.'cfg(windows)'.dependencies]
+rand = "0.8"
+
 [target.'cfg(windows)'.dependencies.windows-sys]
 version = "0.36"
 features = [
   "Win32_Storage_FileSystem",         # Enables NtCreateFile
   "Win32_Foundation",                 # Basic types eg HANDLE
   "Win32_Networking_WinSock",         # winsock2 types/functions
   "Win32_System_IO",                  # IO types like OVERLAPPED etc
+  "Win32_System_Threading",           # Process utilities
   "Win32_System_WindowsProgramming",  # General future used for various types/funcs
 ]
 

src/net/mod.rs

@@ -33,7 +33,11 @@ mod udp;
 #[cfg(not(target_os = "wasi"))]
 pub use self::udp::UdpSocket;
 
-#[cfg(unix)]
+#[cfg(any(unix, windows))]
 mod uds;
+#[cfg(any(unix, windows))]
+pub use self::uds::{SocketAddr, UnixListener, UnixStream};
 #[cfg(unix)]
-pub use self::uds::{SocketAddr, UnixDatagram, UnixListener, UnixStream};
+pub use self::uds::UnixDatagram;
+#[cfg(windows)]
+pub use self::uds::stdnet;

src/net/uds/listener.rs

@@ -2,8 +2,14 @@ use crate::io_source::IoSource;
 use crate::net::{SocketAddr, UnixStream};
 use crate::{event, sys, Interest, Registry, Token};
 
+#[cfg(unix)]
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+#[cfg(unix)]
 use std::os::unix::net;
+#[cfg(windows)]
+use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
+#[cfg(windows)]
+use crate::sys::uds::{stdnet as net};
 use std::path::Path;
 use std::{fmt, io};
 
@@ -35,7 +41,9 @@ impl UnixListener {
     /// The call is responsible for ensuring that the listening socket is in
     /// non-blocking mode.
     pub fn accept(&self) -> io::Result<(UnixStream, SocketAddr)> {
-        sys::uds::listener::accept(&self.inner)
+        self.inner.do_io(|inner| {
+            sys::uds::listener::accept(&*inner)
+        })
     }
 
     /// Returns the local socket address of this listener.
@@ -79,18 +87,21 @@ impl fmt::Debug for UnixListener {
     }
 }
 
+#[cfg(unix)]
 impl IntoRawFd for UnixListener {
     fn into_raw_fd(self) -> RawFd {
         self.inner.into_inner().into_raw_fd()
     }
 }
 
+#[cfg(unix)]
 impl AsRawFd for UnixListener {
     fn as_raw_fd(&self) -> RawFd {
         self.inner.as_raw_fd()
     }
 }
 
+#[cfg(unix)]
 impl FromRawFd for UnixListener {
     /// Converts a `RawFd` to a `UnixListener`.
     ///
@@ -102,3 +113,24 @@ impl FromRawFd for UnixListener {
         UnixListener::from_std(FromRawFd::from_raw_fd(fd))
     }
 }
+
+#[cfg(windows)]
+impl IntoRawSocket for UnixListener {
+    fn into_raw_socket(self) -> RawSocket {
+        self.inner.into_inner().into_raw_socket()
+    }
+}
+
+#[cfg(windows)]
+impl AsRawSocket for UnixListener {
+    fn as_raw_socket(&self) -> RawSocket {
+        self.inner.as_raw_socket()
+    }
+}
+
+#[cfg(windows)]
+impl FromRawSocket for UnixListener {
+    unsafe fn from_raw_socket(sock: RawSocket) -> Self {
+        UnixListener::from_std(FromRawSocket::from_raw_socket(sock))
+    }
+}

src/net/uds/mod.rs

@@ -1,4 +1,6 @@
+#[cfg(unix)]
 mod datagram;
+#[cfg(unix)]
 pub use self::datagram::UnixDatagram;
 
 mod listener;
@@ -8,3 +10,6 @@ mod stream;
 pub use self::stream::UnixStream;
 
 pub use crate::sys::SocketAddr;
+
+#[cfg(windows)]
+pub use crate::sys::uds::stdnet;

src/net/uds/stream.rs

@@ -4,8 +4,14 @@ use crate::{event, sys, Interest, Registry, Token};
 use std::fmt;
 use std::io::{self, IoSlice, IoSliceMut, Read, Write};
 use std::net::Shutdown;
+#[cfg(unix)]
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+#[cfg(unix)]
 use std::os::unix::net;
+#[cfg(windows)]
+use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
+#[cfg(windows)]
+use crate::sys::uds::{stdnet as net};
 use std::path::Path;
 
 /// A non-blocking Unix stream socket.
@@ -86,7 +92,8 @@ impl UnixStream {
     ///
     /// # Examples
     ///
-    /// ```
+    #[cfg_attr(unix, doc = "```")]
+    #[cfg_attr(windows, doc = "```ignore")]
     /// # use std::error::Error;
     /// #
     /// # fn main() -> Result<(), Box<dyn Error>> {
@@ -134,6 +141,73 @@ impl UnixStream {
     /// # Ok(())
     /// # }
     /// ```
+    ///
+    #[cfg_attr(windows, doc = "```")]
+    #[cfg_attr(unix, doc = "```ignore")]
+    /// # use std::error::Error;
+    /// #
+    /// # fn main() -> Result<(), Box<dyn Error>> {
+    /// use std::io;
+    /// use std::os::windows::io::AsRawSocket;
+    /// use std::os::raw::c_int;
+    /// use mio::net::UnixStream;
+    /// use windows_sys::Win32::Networking::WinSock;
+    /// use std::convert::TryInto;
+    ///
+    /// let (stream1, stream2) = UnixStream::pair()?;
+    ///
+    /// // Wait until the stream is writable...
+    ///
+    /// // Write to the stream using a direct WinSock call, of course the
+    /// // `io::Write` implementation would be easier to use.
+    /// let buf = b"hello";
+    /// let n = stream1.try_io(|| {
+    ///     let res = unsafe {
+    ///         WinSock::send(
+    ///             stream1.as_raw_socket().try_into().unwrap(),
+    ///             &buf as *const _ as *const _,
+    ///             buf.len() as c_int,
+    ///             0
+    ///         )
+    ///     };
+    ///     if res != WinSock::SOCKET_ERROR {
+    ///         Ok(res as usize)
+    ///     } else {
+    ///         // If EAGAIN or EWOULDBLOCK is set by WinSock::send, the closure
+    ///         // should return `WouldBlock` error.
+    ///         Err(io::Error::from_raw_os_error(unsafe {
+    ///             WinSock::WSAGetLastError()
+    ///         }))
+    ///     }
+    /// })?;
+    /// eprintln!("write {} bytes", n);
+    ///
+    /// // Wait until the stream is readable...
+    ///
+    /// // Read from the stream using a direct WinSock call, of course the
+    /// // `io::Read` implementation would be easier to use.
+    /// let mut buf = [0; 512];
+    /// let n = stream2.try_io(|| {
+    ///     let res = unsafe {
+    ///         WinSock::recv(
+    ///             stream2.as_raw_socket().try_into().unwrap(),
+    ///             &mut buf as *mut _ as *mut _,
+    ///             buf.len() as c_int,
+    ///             0
+    ///         )
+    ///     };
+    ///     if res != WinSock::SOCKET_ERROR {
+    ///         Ok(res as usize)
+    ///     } else {
+    ///         // If EAGAIN or EWOULDBLOCK is set by WinSock::recv, the closure
+    ///         // should return `WouldBlock` error.
+    ///         Err(io::Error::last_os_error())
+    ///     }
+    /// })?;
+    /// eprintln!("read {} bytes", n);
+    /// # Ok(())
+    /// # }
+    /// ```
     pub fn try_io<F, T>(&self, f: F) -> io::Result<T>
     where
         F: FnOnce() -> io::Result<T>,
@@ -220,18 +294,21 @@ impl fmt::Debug for UnixStream {
     }
 }
 
+#[cfg(unix)]
 impl IntoRawFd for UnixStream {
     fn into_raw_fd(self) -> RawFd {
         self.inner.into_inner().into_raw_fd()
     }
 }
 
+#[cfg(unix)]
 impl AsRawFd for UnixStream {
     fn as_raw_fd(&self) -> RawFd {
         self.inner.as_raw_fd()
     }
 }
 
+#[cfg(unix)]
 impl FromRawFd for UnixStream {
     /// Converts a `RawFd` to a `UnixStream`.
     ///
@@ -243,3 +320,24 @@ impl FromRawFd for UnixStream {
         UnixStream::from_std(FromRawFd::from_raw_fd(fd))
     }
 }
+
+#[cfg(windows)]
+impl IntoRawSocket for UnixStream {
+    fn into_raw_socket(self) -> RawSocket {
+        self.inner.into_inner().into_raw_socket()
+    }
+}
+
+#[cfg(windows)]
+impl AsRawSocket for UnixStream {
+    fn as_raw_socket(&self) -> RawSocket {
+        self.inner.as_raw_socket()
+    }
+}
+
+#[cfg(windows)]
+impl FromRawSocket for UnixStream {
+    unsafe fn from_raw_socket(sock: RawSocket) -> Self {
+        UnixStream::from_std(FromRawSocket::from_raw_socket(sock))
+    }
+}

src/sys/mod.rs

@@ -59,7 +59,7 @@ cfg_os_poll! {
 
 #[cfg(windows)]
 cfg_os_poll! {
-    mod windows;
+    pub mod windows;
     pub use self::windows::*;
 }
 
@@ -83,4 +83,9 @@ cfg_not_os_poll! {
     cfg_net! {
         pub use self::unix::SocketAddr;
     }
+
+    #[cfg(windows)]
+    cfg_net! {
+        pub use self::windows::SocketAddr;
+    }
 }

src/sys/windows/mod.rs

@@ -31,10 +31,28 @@ cfg_net! {
         }};
     }
 
+    macro_rules! wsa_syscall {
+        ($fn: ident ( $($arg: expr),* $(,)* ), $err_test: path, $err_value: expr) => {{
+            let res = unsafe { $fn($($arg, )*) };
+            if $err_test(&res, &$err_value) {
+                use windows_sys::Win32::Networking::WinSock::WSAGetLastError;
+                Err(io::Error::from_raw_os_error(unsafe {
+                    WSAGetLastError()
+                }))
+            } else {
+                Ok(res)
+            }
+        }};
+    }
+
     mod net;
 
     pub(crate) mod tcp;
     pub(crate) mod udp;
+    pub mod uds;
+    pub use self::uds::SocketAddr;
+    #[cfg(all(windows, test))]
+    pub use self::uds::stdnet;
 }
 
 cfg_os_ext! {
@@ -84,15 +102,11 @@ cfg_io_source! {
             F: FnOnce(&T) -> io::Result<R>,
         {
             let result = f(io);
-            if let Err(ref e) = result {
-                if e.kind() == io::ErrorKind::WouldBlock {
-                    self.inner.as_ref().map_or(Ok(()), |state| {
-                        state
-                            .selector
-                            .reregister(state.sock_state.clone(), state.token, state.interests)
-                    })?;
-                }
-            }
+            self.inner.as_ref().map_or(Ok(()), |state| {
+                state
+                    .selector
+                    .reregister(state.sock_state.clone(), state.token, state.interests)
+            })?;
             result
         }
 

src/sys/windows/uds/listener.rs

@@ -0,0 +1,22 @@
+use std::io;
+use std::os::windows::io::AsRawSocket;
+use std::path::Path;
+
+use super::{stdnet as net};
+use crate::net::{SocketAddr, UnixStream};
+
+pub(crate) fn bind(path: &Path) -> io::Result<net::UnixListener> {
+    let listener = net::UnixListener::bind(path)?;
+    listener.set_nonblocking(true)?;
+    Ok(listener)
+}
+
+pub(crate) fn accept(listener: &net::UnixListener) -> io::Result<(UnixStream, SocketAddr)> {
+    listener.set_nonblocking(true)?;
+    let es = listener.accept().map(|(stream, addr)| (UnixStream::from_std(stream), addr));
+    es
+}
+
+pub(crate) fn local_addr(listener: &net::UnixListener) -> io::Result<SocketAddr> {
+    super::local_addr(listener.as_raw_socket())
+}

src/sys/windows/uds/mod.rs

@@ -0,0 +1,36 @@
+pub mod stdnet;
+pub use self::stdnet::SocketAddr;
+
+cfg_os_poll! {
+    use std::convert::TryInto;
+    use windows_sys::Win32::Networking::WinSock::{
+        getsockname,
+        getpeername,
+        SOCKET_ERROR
+    };
+    use std::os::windows::io::RawSocket;
+    use std::io;
+
+    pub(crate) mod listener;
+    pub(crate) mod stream;
+
+    pub(crate) fn local_addr(socket: RawSocket) -> io::Result<SocketAddr> {
+        SocketAddr::new(|sockaddr, socklen| {
+            wsa_syscall!(
+                getsockname(socket.try_into().unwrap(), sockaddr, socklen),
+                PartialEq::eq,
+                SOCKET_ERROR
+            )
+        })
+    }
+
+    pub(crate) fn peer_addr(socket: RawSocket) -> io::Result<SocketAddr> {
+        SocketAddr::new(|sockaddr, socklen| {
+            wsa_syscall!(
+                getpeername(socket.try_into().unwrap(), sockaddr, socklen),
+                PartialEq::eq,
+                SOCKET_ERROR
+            )
+        })
+    }
+}