Add windows NamedPipe (#1351)

Adds a NamedPipe implementation for windows. It is based on the IOCP API. To
bridge the IOCP model with the readiness model, buffers are maintained
internally. Writes are first written to the internal buffer and the buffer is submitted
to IOCP. The internal read buffer is submitted to IOCP to be filled. Once it is filled
the pipe becomes ready and the user can read from it.

Cargo.toml

@@ -49,6 +49,7 @@ ntapi  = "0.3"
 [dev-dependencies]
 env_logger = { version = "0.6.2", default-features = false }
 net2       = "0.2.33"
+rand = "0.4"
 
 [package.metadata.docs.rs]
 all-features = true

src/lib.rs

@@ -89,6 +89,16 @@ pub mod unix {
     pub use crate::sys::SourceFd;
 }
 
+#[cfg(all(windows, feature = "os-util"))]
+#[cfg_attr(docsrs, doc(cfg(all(windows, feature = "os-util"))))]
+pub mod windows {
+    //! Windows only extensions.
+
+    cfg_os_poll! {
+        pub use crate::sys::named_pipe::NamedPipe;
+    }
+}
+
 // Enable with `cargo doc --features extra-docs`.
 #[cfg(feature = "extra-docs")]
 pub mod features {

src/sys/mod.rs

@@ -80,7 +80,7 @@ cfg_os_poll! {
 #[cfg(windows)]
 cfg_os_poll! {
     mod windows;
-    pub(crate) use self::windows::*;
+    pub use self::windows::*;
 }
 
 cfg_not_os_poll! {

src/sys/windows/afd.rs

@@ -7,9 +7,7 @@ use std::io;
 use std::mem::size_of;
 use std::os::windows::io::AsRawHandle;
 use std::ptr::null_mut;
-use winapi::shared::ntdef::{
-    HANDLE, LARGE_INTEGER, NTSTATUS, PVOID, ULONG,
-};
+use winapi::shared::ntdef::{HANDLE, LARGE_INTEGER, NTSTATUS, PVOID, ULONG};
 use winapi::shared::ntstatus::{STATUS_NOT_FOUND, STATUS_PENDING, STATUS_SUCCESS};
 
 const IOCTL_AFD_POLL: ULONG = 0x00012024;
@@ -196,7 +194,11 @@ cfg_net! {
                     ));
                 }
                 let fd = File::from_raw_handle(afd_helper_handle as RawHandle);
-                let token = NEXT_TOKEN.fetch_add(1, Ordering::Relaxed) + 1;
+                // Increment by 2 to reserve space for other types of handles.
+                // Non-AFD types (currently only NamedPipe), use odd numbered
+                // tokens. This allows the selector to differentate between them
+                // and dispatch events accordingly.
+                let token = NEXT_TOKEN.fetch_add(2, Ordering::Relaxed) + 2;
                 let afd = Afd { fd };
                 cp.add_handle(token, &afd.fd)?;
                 match SetFileCompletionNotificationModes(

src/sys/windows/event.rs

@@ -14,6 +14,35 @@ pub fn token(event: &Event) -> Token {
     Token(event.data as usize)
 }
 
+impl Event {
+    pub(super) fn new(token: Token) -> Event {
+        Event {
+            flags: 0,
+            data: usize::from(token) as u64,
+        }
+    }
+
+    pub(super) fn set_readable(&mut self) {
+        self.flags |= afd::POLL_RECEIVE
+    }
+
+    #[cfg(feature = "os-util")]
+    pub(super) fn set_writable(&mut self) {
+        self.flags |= afd::POLL_SEND;
+    }
+
+    pub(super) fn from_completion_status(status: &CompletionStatus) -> Event {
+        Event {
+            flags: status.bytes_transferred(),
+            data: status.token() as u64,
+        }
+    }
+
+    pub(super) fn to_completion_status(&self) -> CompletionStatus {
+        CompletionStatus::new(self.flags, self.data as usize, std::ptr::null_mut())
+    }
+}
+
 pub(crate) const READABLE_FLAGS: u32 = afd::POLL_RECEIVE
     | afd::POLL_DISCONNECT
     | afd::POLL_ACCEPT

src/sys/windows/mod.rs

@@ -7,6 +7,9 @@ pub use event::{Event, Events};
 mod selector;
 pub use selector::{Selector, SelectorInner, SockState};
 
+mod overlapped;
+use overlapped::Overlapped;
+
 // Macros must be defined before the modules that use them
 cfg_net! {
     /// Helper macro to execute a system call that returns an `io::Result`.
@@ -32,6 +35,9 @@ cfg_udp! {
     pub(crate) mod udp;
 }
 
+#[cfg(feature = "os-util")]
+pub(crate) mod named_pipe;
+
 mod waker;
 pub(crate) use waker::Waker;