Repository: tokio-rs/mio
Pull request #1343 (2020-07-31 12:37:56)

Read directly into Vec in tcp_server example
There was a question about this on Discord chat.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index bc9a8ef..5827f0c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,72 @@
+# 0.7.4
+
+## Fixes
+
+* lost "socket closed" events on windows
+  (https://github.com/tokio-rs/mio/commit/50c299aca56c4a26e5ed20c283007239fbe6a7a7)
+
+## Added
+
+* `TcpSocket::set_linger()` configures SO_LINGER
+  (https://github.com/tokio-rs/mio/commit/3b4096565c1a879f651b8f8282ecdcbdbd5c92d3)
+
+# 0.7.3
+
+## Added
+
+* `TcpSocket` for configuring a TCP socket before connecting or listening.
+  (https://github.com/tokio-rs/mio/commit/5b09e60d0f64419b989bda88c86a3147334a03b3)
+
+# 0.7.2
+
+## Added
+
+* Windows named pipe support.
+  (https://github.com/tokio-rs/mio/commit/52e8c2220e87696d20f13561402bcaabba4136ed)
+
+# 0.7.1
+
+## Reduced support for 32-bit Apple targets
+
+In January 2020 Rust reduced its support for 32-bit Apple targets
+(https://blog.rust-lang.org/2020/01/03/reducing-support-for-32-bit-apple-targets.html).
+Starting with v0.7.1 Mio will do the same as we're no longer checking 32 bit
+iOS/macOS on our CI.
+
+## Added
+
+* Support for illumos
+  (https://github.com/tokio-rs/mio/commit/976f2354d0e8fbbb64fba3bf017d7131f9c369a0).
+* Report `epoll(2)`'s `EPOLLERR` event as `Event::is_write_closed` if it's the
+  only event
+  (https://github.com/tokio-rs/mio/commit/0c77b5712d675eeb9bd43928b5dd7d22b2c7ac0c).
+* Optimised event::Iter::{size_hint, count}
+  (https://github.com/tokio-rs/mio/commit/40df934a11b05233a7796c4de19a4ee06bc4e03e).
+
+## Fixed
+
+* Work around Linux kernel < 2.6.37 bug on 32-bits making timeouts longer then
+  ~30 minutes effectively infinite
+  (https://github.com/tokio-rs/mio/commit/d555991f5ee81f6c1eec0fe481557d3d5b8d5ff4).
+* Set `SO_NOSIGPIPE` on all sockets (not just UDP) on for Apple targets
+  (https://github.com/tokio-rs/mio/commit/b8bbdcb0d3236f4c4acb257996d42a88dc9987d9).
+* Properly handle `POLL_ABORT` on Windows
+  (https://github.com/tokio-rs/mio/commit/a98da62b3ed1eeed1770aaca12f46d647e4fa749).
+* Improved error handling around failing `SIO_BASE_HANDLE` calls on Windows
+  (https://github.com/tokio-rs/mio/commit/b15fc18458a79ef8a51f73effa92548650f4e5dc).
+
+## Changed
+
+* On NetBSD we now use `accept4(2)`
+  (https://github.com/tokio-rs/mio/commit/4e306addc7144f2e02a7e8397c220b179a006a19).
+* The package uploaded to crates.io should be slightly smaller
+  (https://github.com/tokio-rs/mio/commit/eef8d3b9500bc0db957cd1ac68ee128ebc68351f).
+
+## Removed
+
+* Dependency on `lazy_static` on Windows
+  (https://github.com/tokio-rs/mio/commit/57e4c2a8ac153bc7bb87829e22cf0a21e3927e8a).
+
 # 0.7.0
 
 Version 0.7 of Mio contains various major changes compared to version 0.6.
diff --git a/Cargo.toml b/Cargo.toml
index 85d3fd5..e1a2b20 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,11 +6,11 @@ name = "mio"
 # - Update CHANGELOG.md.
 # - Update doc URL.
 # - Create git tag
-version       = "0.7.0"
+version       = "0.7.4"
 license       = "MIT"
 authors       = ["Carl Lerche <me@carllerche.com>"]
 description   = "Lightweight non-blocking IO"
-documentation = "https://docs.rs/mio/0.7.0"
+documentation = "https://docs.rs/mio/0.7.4"
 homepage      = "https://github.com/tokio-rs/mio"
 repository    = "https://github.com/tokio-rs/mio"
 readme        = "README.md"
@@ -24,7 +24,6 @@ include       = [
   "src/**/*.rs",
   "examples/**/*.rs",
 ]
-publish = false
 
 [features]
 default = []
@@ -48,7 +47,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
diff --git a/examples/tcp_server.rs b/examples/tcp_server.rs
index dc871f3..42426ee 100644
--- a/examples/tcp_server.rs
+++ b/examples/tcp_server.rs
@@ -128,18 +128,23 @@ fn handle_connection_event(
 
     if event.is_readable() {
         let mut connection_closed = false;
-        let mut received_data = Vec::with_capacity(4096);
+        let mut received_data = vec![0; 4096];
+        let mut bytes_read = 0;
         // We can (maybe) read from the connection.
         loop {
-            let mut buf = [0; 256];
-            match connection.read(&mut buf) {
+            match connection.read(&mut received_data[bytes_read..]) {
                 Ok(0) => {
                     // Reading 0 bytes means the other side has closed the
                     // connection or is done writing, then so are we.
                     connection_closed = true;
                     break;
                 }
-                Ok(n) => received_data.extend_from_slice(&buf[..n]),
+                Ok(n) => {
+                    bytes_read += n;
+                    if bytes_read == received_data.len() {
+                        received_data.resize(received_data.len() + 1024, 0);
+                    }
+                }
                 // Would block "errors" are the OS's way of saying that the
                 // connection is not actually ready to perform this I/O operation.
                 Err(ref err) if would_block(err) => break,
@@ -149,10 +154,13 @@ fn handle_connection_event(
             }
         }
 
-        if let Ok(str_buf) = from_utf8(&received_data) {
-            println!("Received data: {}", str_buf.trim_end());
-        } else {
-            println!("Received (none UTF-8) data: {:?}", &received_data);
+        if bytes_read != 0 {
+            let received_data = &received_data[..bytes_read];
+            if let Ok(str_buf) = from_utf8(received_data) {
+                println!("Received data: {}", str_buf.trim_end());
+            } else {
+                println!("Received (none UTF-8) data: {:?}", received_data);
+            }
         }
 
         if connection_closed {
diff --git a/src/event/event.rs b/src/event/event.rs
index a9919d7..959affd 100644
--- a/src/event/event.rs
+++ b/src/event/event.rs
@@ -108,6 +108,7 @@ impl Event {
     /// | [OS selector] | Flag(s) checked |
     /// |---------------|-----------------|
     /// | [epoll]       | `EPOLLHUP`, or  |
+    /// |               | only `EPOLLERR`, or |
     /// |               | `EPOLLOUT` and `EPOLLERR` |
     /// | [kqueue]      | `EV_EOF`        |
     ///
diff --git a/src/interest.rs b/src/interest.rs
index 6bc5929..9032d9a 100644
--- a/src/interest.rs
+++ b/src/interest.rs
@@ -70,6 +70,30 @@ impl Interest {
         Interest(unsafe { NonZeroU8::new_unchecked(self.0.get() | other.0.get()) })
     }
 
+    /// Removes `other` `Interest` from `self`.
+    ///
+    /// Returns `None` if the set would be empty after removing `other`.
+    ///
+    /// ```
+    /// use mio::Interest;
+    ///
+    /// const RW_INTERESTS: Interest = Interest::READABLE.add(Interest::WRITABLE);
+    ///
+    /// // As long a one interest remain this will return `Some`.
+    /// let w_interest = RW_INTERESTS.remove(Interest::READABLE).unwrap();
+    /// assert!(!w_interest.is_readable());
+    /// assert!(w_interest.is_writable());
+    ///
+    /// // Removing all interests from the set will return `None`.
+    /// assert_eq!(w_interest.remove(Interest::WRITABLE), None);
+    ///
+    /// // Its also possible to remove multiple interests at once.
+    /// assert_eq!(RW_INTERESTS.remove(RW_INTERESTS), None);
+    /// ```
+    pub fn remove(self, other: Interest) -> Option<Interest> {
+        NonZeroU8::new(self.0.get() & !other.0.get()).map(Interest)
+    }
+
     /// Returns true if the value includes readable readiness.
     pub const fn is_readable(self) -> bool {
         (self.0.get() & READABLE) != 0
diff --git a/src/lib.rs b/src/lib.rs
index 1853582..6e98c87 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/mio/0.7.0")]
+#![doc(html_root_url = "https://docs.rs/mio/0.7.4")]
 #![deny(
     missing_docs,
     missing_debug_implementations,
@@ -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 {
diff --git a/src/net/mod.rs b/src/net/mod.rs
index c3ecabd..91804ec 100644
--- a/src/net/mod.rs
+++ b/src/net/mod.rs
@@ -9,7 +9,7 @@
 
 cfg_tcp! {
     mod tcp;
-    pub use self::tcp::{TcpListener, TcpStream};
+    pub use self::tcp::{TcpListener, TcpSocket, TcpStream};
 }
 
 cfg_udp! {
diff --git a/src/net/tcp/listener.rs b/src/net/tcp/listener.rs
index ba7bdbd..b6f5736 100644
--- a/src/net/tcp/listener.rs
+++ b/src/net/tcp/listener.rs
@@ -5,7 +5,7 @@ use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
 use std::{fmt, io};
 
-use super::TcpStream;
+use super::{TcpSocket, TcpStream};
 use crate::io_source::IoSource;
 use crate::{event, sys, Interest, Registry, Token};
 
@@ -49,7 +49,20 @@ impl TcpListener {
     /// 3. Bind the socket to the specified address.
     /// 4. Calls `listen` on the socket to prepare it to receive new connections.
     pub fn bind(addr: SocketAddr) -> io::Result<TcpListener> {
-        sys::tcp::bind(addr).map(TcpListener::from_std)
+        let socket = TcpSocket::new_for_addr(addr)?;
+
+        // On platforms with Berkeley-derived sockets, this allows to quickly
+        // rebind a socket, without needing to wait for the OS to clean up the
+        // previous one.
+        //
+        // On Windows, this allows rebinding sockets which are actively in use,
+        // which allows “socket hijacking”, so we explicitly don't set it here.
+        // https://docs.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse
+        #[cfg(not(windows))]
+        socket.set_reuseaddr(true)?;
+
+        socket.bind(addr)?;
+        socket.listen(1024)
     }
 
     /// Creates a new `TcpListener` from a standard `net::TcpListener`.
diff --git a/src/net/tcp/mod.rs b/src/net/tcp/mod.rs
index 94af5c1..b39b909 100644
--- a/src/net/tcp/mod.rs
+++ b/src/net/tcp/mod.rs
@@ -1,5 +1,8 @@
 mod listener;
 pub use self::listener::TcpListener;
 
+mod socket;
+pub use self::socket::TcpSocket;
+
 mod stream;
 pub use self::stream::TcpStream;
diff --git a/src/net/tcp/socket.rs b/src/net/tcp/socket.rs
new file mode 100644
index 0000000..a91f665
--- /dev/null
+++ b/src/net/tcp/socket.rs
@@ -0,0 +1,151 @@
+use crate::net::{TcpStream, TcpListener};
+use crate::sys;
+
+use std::io;
+use std::mem;
+use std::net::SocketAddr;
+use std::time::Duration;
+#[cfg(unix)]
+use std::os::unix::io::{AsRawFd, RawFd, FromRawFd};
+#[cfg(windows)]
+use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
+
+/// A non-blocking TCP socket used to configure a stream or listener.
+///
+/// The `TcpSocket` type wraps the operating-system's socket handle. This type
+/// is used to configure the socket before establishing a connection or start
+/// listening for inbound connections.
+///
+/// The socket will be closed when the value is dropped.
+#[derive(Debug)]
+pub struct TcpSocket {
+    sys: sys::tcp::TcpSocket,
+}
+
+impl TcpSocket {
+    /// Create a new IPv4 TCP socket.
+    ///
+    /// This calls `socket(2)`.
+    pub fn new_v4() -> io::Result<TcpSocket> {
+        sys::tcp::new_v4_socket().map(|sys| TcpSocket {
+            sys
+        })
+    }
+
+    /// Create a new IPv6 TCP socket.
+    ///
+    /// This calls `socket(2)`.
+    pub fn new_v6() -> io::Result<TcpSocket> {
+        sys::tcp::new_v6_socket().map(|sys| TcpSocket {
+            sys
+        })
+    }
+
+    pub(crate) fn new_for_addr(addr: SocketAddr) -> io::Result<TcpSocket> {
+        if addr.is_ipv4() {
+            TcpSocket::new_v4()
+        } else {
+            TcpSocket::new_v6()
+        }
+    }
+
+    /// Bind `addr` to the TCP socket.
+    pub fn bind(&self, addr: SocketAddr) -> io::Result<()> {
+        sys::tcp::bind(self.sys, addr)
+    }
+
+    /// Connect the socket to `addr`.
+    ///
+    /// This consumes the socket and performs the connect operation. Once the
+    /// connection completes, the socket is now a non-blocking `TcpStream` and
+    /// can be used as such.
+    pub fn connect(self, addr: SocketAddr) -> io::Result<TcpStream> {
+        let stream = sys::tcp::connect(self.sys, addr)?;
+
+        // Don't close the socket
+        mem::forget(self);
+        Ok(TcpStream::from_std(stream))
+    }
+
+    /// Listen for inbound connections, converting the socket to a
+    /// `TcpListener`.
+    pub fn listen(self, backlog: u32) -> io::Result<TcpListener> {
+        let listener = sys::tcp::listen(self.sys, backlog)?;
+
+        // Don't close the socket
+        mem::forget(self);
+        Ok(TcpListener::from_std(listener))
+    }
+
+    /// Sets the value of `SO_REUSEADDR` on this socket.
+    pub fn set_reuseaddr(&self, reuseaddr: bool) -> io::Result<()> {
+        sys::tcp::set_reuseaddr(self.sys, reuseaddr)
+    }
+
+    /// Sets the value of `SO_LINGER` on this socket.
+    pub fn set_linger(&self, dur: Option<Duration>) -> io::Result<()> {
+        sys::tcp::set_linger(self.sys, dur)
+    }
+}
+
+impl Drop for TcpSocket {
+    fn drop(&mut self) {
+        sys::tcp::close(self.sys);
+    }
+}
+
+#[cfg(unix)]
+impl AsRawFd for TcpSocket {
+    fn as_raw_fd(&self) -> RawFd {
+        self.sys
+    }
+}
+
+#[cfg(unix)]
+impl FromRawFd for TcpSocket {
+    /// Converts a `RawFd` to a `TcpSocket`.
+    ///
+    /// # Notes
+    ///
+    /// The caller is responsible for ensuring that the socket is in
+    /// non-blocking mode.
+    unsafe fn from_raw_fd(fd: RawFd) -> TcpSocket {
+        TcpSocket { sys: fd }
+    }
+}
+
+#[cfg(windows)]
+impl IntoRawSocket for TcpSocket {
+    fn into_raw_socket(self) -> RawSocket {
+        // The winapi crate defines `SOCKET` as `usize`. The Rust std
+        // conditionally defines `RawSocket` as a fixed size unsigned integer
+        // matching the pointer width. These end up being the same type but we
+        // must cast between them.
+        let ret = self.sys as RawSocket;
+
+        // Avoid closing the socket
+        mem::forget(self);
+
+        ret
+    }
+}
+
+#[cfg(windows)]
+impl AsRawSocket for TcpSocket {
+    fn as_raw_socket(&self) -> RawSocket {
+        self.sys as RawSocket
+    }
+}
+
+#[cfg(windows)]
+impl FromRawSocket for TcpSocket {
+    /// Converts a `RawSocket` to a `TcpSocket`.
+    ///
+    /// # Notes
+    ///
+    /// The caller is responsible for ensuring that the socket is in
+    /// non-blocking mode.
+    unsafe fn from_raw_socket(socket: RawSocket) -> TcpSocket {
+        TcpSocket { sys: socket as sys::tcp::TcpSocket }
+    }
+}
diff --git a/src/net/tcp/stream.rs b/src/net/tcp/stream.rs
index 00dd66f..86f674c 100644
--- a/src/net/tcp/stream.rs
+++ b/src/net/tcp/stream.rs
@@ -7,7 +7,8 @@ use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
 
 use crate::io_source::IoSource;
-use crate::{event, sys, Interest, Registry, Token};
+use crate::{event, Interest, Registry, Token};
+use crate::net::TcpSocket;
 
 /// A non-blocking TCP stream between a local socket and a remote socket.
 ///
@@ -48,7 +49,8 @@ impl TcpStream {
     /// Create a new TCP stream and issue a non-blocking connect to the
     /// specified address.
     pub fn connect(addr: SocketAddr) -> io::Result<TcpStream> {
-        sys::tcp::connect(addr).map(TcpStream::from_std)
+        let socket = TcpSocket::new_for_addr(addr)?;
+        socket.connect(addr)
     }
 
     /// Creates a new `TcpStream` from a standard `net::TcpStream`.
diff --git a/src/sys/mod.rs b/src/sys/mod.rs
index 1359181..8852333 100644
--- a/src/sys/mod.rs
+++ b/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! {
diff --git a/src/sys/shell/tcp.rs b/src/sys/shell/tcp.rs
index f72dcca..de1520b 100644
--- a/src/sys/shell/tcp.rs
+++ b/src/sys/shell/tcp.rs
@@ -1,11 +1,38 @@
 use std::io;
 use std::net::{self, SocketAddr};
+use std::time::Duration;
 
-pub fn connect(_: SocketAddr) -> io::Result<net::TcpStream> {
+pub(crate) type TcpSocket = i32;
+
+pub(crate) fn new_v4_socket() -> io::Result<TcpSocket> {
+    os_required!();
+}
+
+pub(crate) fn new_v6_socket() -> io::Result<TcpSocket> {
+    os_required!();
+}
+
+pub(crate) fn bind(_socket: TcpSocket, _addr: SocketAddr) -> io::Result<()> {
+    os_required!();
+}
+
+pub(crate) fn connect(_: TcpSocket, _addr: SocketAddr) -> io::Result<net::TcpStream> {
+    os_required!();
+}
+
+pub(crate) fn listen(_: TcpSocket, _: u32) -> io::Result<net::TcpListener> {
+    os_required!();
+}
+
+pub(crate) fn close(_: TcpSocket) {
+    os_required!();
+}
+
+pub(crate) fn set_reuseaddr(_: TcpSocket, _: bool) -> io::Result<()> {
     os_required!();
 }
 
-pub fn bind(_: SocketAddr) -> io::Result<net::TcpListener> {
+pub(crate) fn set_linger(_: TcpSocket, _: Option<Duration>) -> io::Result<()> {
     os_required!();
 }
 
diff --git a/src/sys/unix/net.rs b/src/sys/unix/net.rs
index bd25ecd..2671b42 100644
--- a/src/sys/unix/net.rs
+++ b/src/sys/unix/net.rs
@@ -1,7 +1,7 @@
 #[cfg(all(feature = "os-poll", any(feature = "tcp", feature = "udp")))]
 use std::net::SocketAddr;
 
-#[cfg(all(feature = "os-poll", any(feature = "tcp", feature = "udp")))]
+#[cfg(all(feature = "os-poll", any(feature = "udp")))]
 pub(crate) fn new_ip_socket(
     addr: SocketAddr,
     socket_type: libc::c_int,
diff --git a/src/sys/unix/selector/epoll.rs b/src/sys/unix/selector/epoll.rs
index 4728717..13f1617 100644
--- a/src/sys/unix/selector/epoll.rs
+++ b/src/sys/unix/selector/epoll.rs
@@ -46,8 +46,16 @@ impl Selector {
     }
 
     pub fn select(&self, events: &mut Events, timeout: Option<Duration>) -> io::Result<()> {
+        // A bug in kernels < 2.6.37 makes timeouts larger than LONG_MAX / CONFIG_HZ
+        // (approx. 30 minutes with CONFIG_HZ=1200) effectively infinite on 32 bits
+        // architectures. The magic number is the same constant used by libuv.
+        #[cfg(target_pointer_width = "32")]
+        const MAX_SAFE_TIMEOUT: u128 = 1789569;
+        #[cfg(not(target_pointer_width = "32"))]
+        const MAX_SAFE_TIMEOUT: u128 = libc::c_int::max_value() as u128;
+
         let timeout = timeout
-            .map(|to| cmp::min(to.as_millis(), libc::c_int::max_value() as u128) as libc::c_int)
+            .map(|to| cmp::min(to.as_millis(), MAX_SAFE_TIMEOUT) as libc::c_int)
             .unwrap_or(-1);
 
         events.clear();
@@ -164,6 +172,8 @@ pub mod event {
             // Unix pipe write end has closed
             || (event.events as libc::c_int & libc::EPOLLOUT != 0
                 && event.events as libc::c_int & libc::EPOLLERR != 0)
+            // The other side (read end) of a Unix pipe has closed.
+            || event.events as libc::c_int == libc::EPOLLERR
     }
 
     pub fn is_priority(event: &Event) -> bool {
diff --git a/src/sys/unix/tcp.rs b/src/sys/unix/tcp.rs
index 08a28c0..81b371b 100644
--- a/src/sys/unix/tcp.rs
+++ b/src/sys/unix/tcp.rs
@@ -1,54 +1,75 @@
 use std::io;
 use std::mem::{size_of, MaybeUninit};
 use std::net::{self, SocketAddr};
+use std::time::Duration;
 use std::os::unix::io::{AsRawFd, FromRawFd};
 
-use crate::sys::unix::net::{new_ip_socket, socket_addr, to_socket_addr};
-
-pub fn connect(addr: SocketAddr) -> io::Result<net::TcpStream> {
-    new_ip_socket(addr, libc::SOCK_STREAM)
-        .and_then(|socket| {
-            let (raw_addr, raw_addr_length) = socket_addr(&addr);
-            syscall!(connect(socket, raw_addr, raw_addr_length))
-                .or_else(|err| match err {
-                    // Connect hasn't finished, but that is fine.
-                    ref err if err.raw_os_error() == Some(libc::EINPROGRESS) => Ok(0),
-                    err => Err(err),
-                })
-                .map(|_| socket)
-                .map_err(|err| {
-                    // Close the socket if we hit an error, ignoring the error
-                    // from closing since we can't pass back two errors.
-                    let _ = unsafe { libc::close(socket) };
-                    err
-                })
-        })
-        .map(|socket| unsafe { net::TcpStream::from_raw_fd(socket) })
+use crate::sys::unix::net::{new_socket, socket_addr, to_socket_addr};
+
+pub type TcpSocket = libc::c_int;
+
+pub(crate) fn new_v4_socket() -> io::Result<TcpSocket> {
+    new_socket(libc::AF_INET, libc::SOCK_STREAM)
 }
 
-pub fn bind(addr: SocketAddr) -> io::Result<net::TcpListener> {
-    new_ip_socket(addr, libc::SOCK_STREAM).and_then(|socket| {
-        // Set SO_REUSEADDR (mirrors what libstd does).
-        syscall!(setsockopt(
-            socket,
-            libc::SOL_SOCKET,
-            libc::SO_REUSEADDR,
-            &1 as *const libc::c_int as *const libc::c_void,
-            size_of::<libc::c_int>() as libc::socklen_t,
-        ))
-        .and_then(|_| {
-            let (raw_addr, raw_addr_length) = socket_addr(&addr);
-            syscall!(bind(socket, raw_addr, raw_addr_length))
-        })
-        .and_then(|_| syscall!(listen(socket, 1024)))
-        .map_err(|err| {
-            // Close the socket if we hit an error, ignoring the error
-            // from closing since we can't pass back two errors.
-            let _ = unsafe { libc::close(socket) };
-            err
-        })
-        .map(|_| unsafe { net::TcpListener::from_raw_fd(socket) })
-    })
+pub(crate) fn new_v6_socket() -> io::Result<TcpSocket> {
+    new_socket(libc::AF_INET6, libc::SOCK_STREAM)
+}
+
+pub(crate) fn bind(socket: TcpSocket, addr: SocketAddr) -> io::Result<()> {
+    let (raw_addr, raw_addr_length) = socket_addr(&addr);
+    syscall!(bind(socket, raw_addr, raw_addr_length))?;
+    Ok(())
+}
+
+pub(crate) fn connect(socket: TcpSocket, addr: SocketAddr) -> io::Result<net::TcpStream> {
+    let (raw_addr, raw_addr_length) = socket_addr(&addr);
+
+    match syscall!(connect(socket, raw_addr, raw_addr_length)) {
+        Err(err) if err.raw_os_error() != Some(libc::EINPROGRESS) => {
+            Err(err)
+        }
+        _ => {
+            Ok(unsafe { net::TcpStream::from_raw_fd(socket) })
+        }
+    }
+}
+
+pub(crate) fn listen(socket: TcpSocket, backlog: u32) -> io::Result<net::TcpListener> {
+    use std::convert::TryInto;
+
+    let backlog = backlog.try_into().unwrap_or(i32::max_value());
+    syscall!(listen(socket, backlog))?;
+    Ok(unsafe { net::TcpListener::from_raw_fd(socket) })
+}
+
+pub(crate) fn close(socket: TcpSocket) {
+    let _ = unsafe { net::TcpStream::from_raw_fd(socket) };
+}
+
+pub(crate) fn set_reuseaddr(socket: TcpSocket, reuseaddr: bool) -> io::Result<()> {
+    let val: libc::c_int = if reuseaddr { 1 } else { 0 };
+    syscall!(setsockopt(
+        socket,
+        libc::SOL_SOCKET,
+        libc::SO_REUSEADDR,
+        &val as *const libc::c_int as *const libc::c_void,
+        size_of::<libc::c_int>() as libc::socklen_t,
+    )).map(|_| ())
+}
+
+pub(crate) fn set_linger(socket: TcpSocket, dur: Option<Duration>) -> io::Result<()> {
+    let val: libc::linger = libc::linger {
+        l_onoff: if dur.is_some() { 1 } else { 0 },
+        l_linger: dur.map(|dur| dur.as_secs() as libc::c_int).unwrap_or_default(),
+    };
+    syscall!(setsockopt(
+        socket,
+        libc::SOL_SOCKET,
+        libc::SO_LINGER,
+        &val as *const libc::linger as *const libc::c_void,
+        size_of::<libc::linger>() as libc::socklen_t,
+    )).map(|_| ())
 }
 
 pub fn accept(listener: &net::TcpListener) -> io::Result<(net::TcpStream, SocketAddr)> {
diff --git a/src/sys/unix/uds/socketaddr.rs b/src/sys/unix/uds/socketaddr.rs
index 69f311a..ddfa2f0 100644
--- a/src/sys/unix/uds/socketaddr.rs
+++ b/src/sys/unix/uds/socketaddr.rs
@@ -74,6 +74,8 @@ cfg_os_poll! {
         /// Documentation reflected in [`SocketAddr`]
         ///
         /// [`SocketAddr`]: std::os::unix::net::SocketAddr
+        // FIXME: The matches macro requires rust 1.42.0 and we still support 1.39.0
+        #[allow(clippy::match_like_matches_macro)]
         pub fn is_unnamed(&self) -> bool {
             if let AddressKind::Unnamed = self.address() {
                 true
diff --git a/src/sys/windows/afd.rs b/src/sys/windows/afd.rs
index 43d443c..82c8e9e 100644
--- a/src/sys/windows/afd.rs
+++ b/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(
diff --git a/src/sys/windows/event.rs b/src/sys/windows/event.rs
index b341255..235074a 100644
--- a/src/sys/windows/event.rs
+++ b/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
diff --git a/src/sys/windows/mod.rs b/src/sys/windows/mod.rs
index e1f4803..7bba6dd 100644
--- a/src/sys/windows/mod.rs
+++ b/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;
 
diff --git a/src/sys/windows/named_pipe.rs b/src/sys/windows/named_pipe.rs
new file mode 100644
index 0000000..a5688ce
--- /dev/null
+++ b/src/sys/windows/named_pipe.rs
@@ -0,0 +1,709 @@
+use crate::{poll, Registry};
+use crate::event::Source;
+use crate::sys::windows::{Event, Overlapped};
+use winapi::um::minwinbase::OVERLAPPED_ENTRY;
+
+use std::ffi::OsStr;
+use std::fmt;
+use std::io::{self, Read, Write};
+use std::mem;
+use std::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
+use std::slice;
+use std::sync::atomic::{AtomicUsize, AtomicBool};
+use std::sync::atomic::Ordering::{Relaxed, SeqCst};
+use std::sync::{Arc, Mutex};
+
+use crate::{Interest, Token};
+use miow::iocp::{CompletionPort, CompletionStatus};
+use miow::pipe;
+use winapi::shared::winerror::{ERROR_BROKEN_PIPE, ERROR_PIPE_LISTENING};
+use winapi::um::ioapiset::CancelIoEx;
+
+/// # Safety
+///
+/// Only valid if the strict is annotated with `#[repr(C)]`. This is only used
+/// with `Overlapped` and `Inner`, which are correctly annotated.
+macro_rules! offset_of {
+    ($t:ty, $($field:ident).+) => (
+        &(*(0 as *const $t)).$($field).+ as *const _ as usize
+    )
+}
+
+macro_rules! overlapped2arc {
+    ($e:expr, $t:ty, $($field:ident).+) => ({
+        let offset = offset_of!($t, $($field).+);
+        debug_assert!(offset < mem::size_of::<$t>());
+        Arc::from_raw(($e as usize - offset) as *mut $t)
+    })
+}
+
+/// Non-blocking windows named pipe.
+///
+/// This structure internally contains a `HANDLE` which represents the named
+/// pipe, and also maintains state associated with the mio event loop and active
+/// I/O operations that have been scheduled to translate IOCP to a readiness
+/// model.
+///
+/// Note, IOCP is a *completion* based model whereas mio is a *readiness* based
+/// model. To bridge this, `NamedPipe` performs internal buffering. Writes are
+/// written to an internal buffer and the buffer is submitted to IOCP. IOCP
+/// reads are submitted using internal buffers and `NamedPipe::read` reads from
+/// this internal buffer.
+///
+/// # Trait implementations
+///
+/// The `Read` and `Write` traits are implemented for `NamedPipe` and for
+/// `&NamedPipe`. This represents that a named pipe can be concurrently read and
+/// written to and also can be read and written to at all. Typically a named
+/// pipe needs to be connected to a client before it can be read or written,
+/// however.
+///
+/// Note that for I/O operations on a named pipe to succeed then the named pipe
+/// needs to be associated with an event loop. Until this happens all I/O
+/// operations will return a "would block" error.
+///
+/// # Managing connections
+///
+/// The `NamedPipe` type supports a `connect` method to connect to a client and
+/// a `disconnect` method to disconnect from that client. These two methods only
+/// work once a named pipe is associated with an event loop.
+///
+/// The `connect` method will succeed asynchronously and a completion can be
+/// detected once the object receives a writable notification.
+///
+/// # Named pipe clients
+///
+/// Currently to create a client of a named pipe server then you can use the
+/// `OpenOptions` type in the standard library to create a `File` that connects
+/// to a named pipe. Afterwards you can use the `into_raw_handle` method coupled
+/// with the `NamedPipe::from_raw_handle` method to convert that to a named pipe
+/// that can operate asynchronously. Don't forget to pass the
+/// `FILE_FLAG_OVERLAPPED` flag when opening the `File`.
+pub struct NamedPipe {
+    inner: Arc<Inner>,
+}
+
+#[repr(C)]
+struct Inner {
+    handle: pipe::NamedPipe,
+
+    connect: Overlapped,
+    connecting: AtomicBool,
+
+    read: Overlapped,
+    write: Overlapped,
+
+    io: Mutex<Io>,
+
+    pool: Mutex<BufferPool>,
+}
+
+struct Io {
+    // Uniquely identifies the selector associated with this named pipe
+    cp: Option<Arc<CompletionPort>>,
+    // Token used to identify events
+    token: Option<Token>,
+    read: State,
+    read_interest: bool,
+    write: State,
+    write_interest: bool,
+    connect_error: Option<io::Error>,
+}
+
+#[derive(Debug)]
+enum State {
+    None,
+    Pending(Vec<u8>, usize),
+    Ok(Vec<u8>, usize),
+    Err(io::Error),
+}
+
+// Odd tokens are for named pipes
+static NEXT_TOKEN: AtomicUsize = AtomicUsize::new(1);
+
+fn would_block() -> io::Error {
+    io::ErrorKind::WouldBlock.into()
+}
+
+impl NamedPipe {
+    /// Creates a new named pipe at the specified `addr` given a "reasonable
+    /// set" of initial configuration options.
+    pub fn new<A: AsRef<OsStr>>(
+        addr: A,
+    ) -> io::Result<NamedPipe> {
+        let pipe = pipe::NamedPipe::new(addr)?;
+        // Safety: nothing actually unsafe about this. The trait fn includes
+        // `unsafe`.
+        Ok(unsafe { NamedPipe::from_raw_handle(pipe.into_raw_handle()) })
+    }
+
+    /// Attempts to call `ConnectNamedPipe`, if possible.
+    ///
+    /// This function will attempt to connect this pipe to a client in an
+    /// asynchronous fashion. If the function immediately establishes a
+    /// connection to a client then `Ok(())` is returned. Otherwise if a
+    /// connection attempt was issued and is now in progress then a "would
+    /// block" error is returned.
+    ///
+    /// When the connection is finished then this object will be flagged as
+    /// being ready for a write, or otherwise in the writable state.
+    ///
+    /// # Errors
+    ///
+    /// This function will return a "would block" error if the pipe has not yet
+    /// been registered with an event loop, if the connection operation has
+    /// previously been issued but has not yet completed, or if the connect
+    /// itself was issued and didn't finish immediately.
+    ///
+    /// Normal I/O errors from the call to `ConnectNamedPipe` are returned
+    /// immediately.
+    pub fn connect(&self) -> io::Result<()> {
+        // "Acquire the connecting lock" or otherwise just make sure we're the
+        // only operation that's using the `connect` overlapped instance.
+        if self.inner.connecting.swap(true, SeqCst) {
+            return Err(would_block());
+        }
+
+        // Now that we've flagged ourselves in the connecting state, issue the
+        // connection attempt. Afterwards interpret the return value and set
+        // internal state accordingly.
+        let res = unsafe {
+            let overlapped = self.inner.connect.as_ptr() as *mut _;
+            self.inner.handle.connect_overlapped(overlapped)
+        };
+
+        match res {
+            // The connection operation finished immediately, so let's schedule
+            // reads/writes and such.
+            Ok(true) => {
+                self.inner.connecting.store(false, SeqCst);
+                Inner::post_register(&self.inner, None);
+                Ok(())
+            }
+
+            // If the overlapped operation was successful and didn't finish
+            // immediately then we forget a copy of the arc we hold
+            // internally. This ensures that when the completion status comes
+            // in for the I/O operation finishing it'll have a reference
+            // associated with it and our data will still be valid. The
+            // `connect_done` function will "reify" this forgotten pointer to
+            // drop the refcount on the other side.
+            Ok(false) => {
+                mem::forget(self.inner.clone());
+                Err(would_block())
+            }
+
+            Err(e) => {
+                self.inner.connecting.store(false, SeqCst);
+                Err(e)
+            }
+        }
+    }
+
+    /// Takes any internal error that has happened after the last I/O operation
+    /// which hasn't been retrieved yet.
+    ///
+    /// This is particularly useful when detecting failed attempts to `connect`.
+    /// After a completed `connect` flags this pipe as writable then callers
+    /// must invoke this method to determine whether the connection actually
+    /// succeeded. If this function returns `None` then a client is connected,
+    /// otherwise it returns an error of what happened and a client shouldn't be
+    /// connected.
+    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
+        Ok(self.inner.io.lock().unwrap().connect_error.take())
+    }
+
+    /// Disconnects this named pipe from a connected client.
+    ///
+    /// This function will disconnect the pipe from a connected client, if any,
+    /// transitively calling the `DisconnectNamedPipe` function.
+    ///
+    /// After a `disconnect` is issued, then a `connect` may be called again to
+    /// connect to another client.
+    pub fn disconnect(&self) -> io::Result<()> {
+        self.inner.handle.disconnect()
+    }
+}
+
+impl FromRawHandle for NamedPipe {
+    unsafe fn from_raw_handle(
+        handle: RawHandle,
+    ) -> NamedPipe {
+        NamedPipe {
+            inner: Arc::new(Inner {
+                // Safety: not really unsafe
+                handle: pipe::NamedPipe::from_raw_handle(handle),
+                // transmutes to straddle winapi versions (mio 0.6 is on an
+                // older winapi)
+                connect: Overlapped::new(connect_done),
+                connecting: AtomicBool::new(false),
+                read: Overlapped::new(read_done),
+                write: Overlapped::new(write_done),
+                io: Mutex::new(Io {
+                    cp: None,
+                    token: None,
+                    read: State::None,
+                    read_interest: false,
+                    write: State::None,
+                    write_interest: false,
+                    connect_error: None,
+                }),
+                pool: Mutex::new(BufferPool::with_capacity(2)),
+            }),
+        }
+    }
+}
+
+impl Read for NamedPipe {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        <&NamedPipe as Read>::read(&mut &*self, buf)
+    }
+}
+
+impl Write for NamedPipe {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        <&NamedPipe as Write>::write(&mut &*self, buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        <&NamedPipe as Write>::flush(&mut &*self)
+    }
+}
+
+impl<'a> Read for &'a NamedPipe {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        let mut state = self.inner.io.lock().unwrap();
+
+        if state.token.is_none() {
+            return Err(would_block());
+        }
+
+        match mem::replace(&mut state.read, State::None) {
+            // In theory not possible with `token` checked above,
+            // but return would block for now.
+            State::None => {
+                Err(would_block())
+            }
+
+            // A read is in flight, still waiting for it to finish
+            State::Pending(buf, amt) => {
+                state.read = State::Pending(buf, amt);
+                Err(would_block())
+            }
+
+            // We previously read something into `data`, try to copy out some
+            // data. If we copy out all the data schedule a new read and
+            // otherwise store the buffer to get read later.
+            State::Ok(data, cur) => {
+                let n = {
+                    let mut remaining = &data[cur..];
+                    remaining.read(buf)?
+                };
+                let next = cur + n;
+                if next != data.len() {
+                    state.read = State::Ok(data, next);
+                } else {
+                    self.inner.put_buffer(data);
+                    Inner::schedule_read(&self.inner, &mut state, None);
+                }
+                Ok(n)
+            }
+
+            // Looks like an in-flight read hit an error, return that here while
+            // we schedule a new one.
+            State::Err(e) => {
+                Inner::schedule_read(&self.inner, &mut state, None);
+                if e.raw_os_error() == Some(ERROR_BROKEN_PIPE as i32) {
+                    Ok(0)
+                } else {
+                    Err(e)
+                }
+            }
+        }
+    }
+}
+
+impl<'a> Write for &'a NamedPipe {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {        
+        // Make sure there's no writes pending
+        let mut io = self.inner.io.lock().unwrap();
+
+        if io.token.is_none() {
+            return Err(would_block());
+        }
+
+        match io.write {
+            State::None => {}
+            _ => {
+                return Err(would_block());
+            }
+        }
+
+        // Move `buf` onto the heap and fire off the write
+        let mut owned_buf = self.inner.get_buffer();
+        owned_buf.extend(buf);
+        Inner::schedule_write(&self.inner, owned_buf, 0, &mut io, None);
+        Ok(buf.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {      
+        Ok(())  
+    }
+}
+
+impl Source for NamedPipe {
+    fn register(&mut self, registry: &Registry, token: Token, interest: Interest) -> io::Result<()> {
+        let mut io = self.inner.io.lock().unwrap();
+
+        io.check_association(registry, false)?;
+
+        if io.token.is_some() {
+            return Err(io::Error::new(
+                io::ErrorKind::AlreadyExists,
+                "I/O source already registered with a `Registry`",
+            ));
+        }
+
+        if io.cp.is_none() {
+            io.cp = Some(poll::selector(registry).clone_port());
+
+            let inner_token = NEXT_TOKEN.fetch_add(2, Relaxed) + 2;
+            poll::selector(registry).inner.cp.add_handle(inner_token, &self.inner.handle)?;
+        }
+
+        io.token = Some(token);
+        io.read_interest = interest.is_readable();
+        io.write_interest = interest.is_writable();
+        drop(io);
+
+        Inner::post_register(&self.inner, None);
+
+        Ok(())
+    }
+
+    fn reregister(&mut self, registry: &Registry, token: Token, interest: Interest) -> io::Result<()> { 
+        let mut io = self.inner.io.lock().unwrap();
+
+        io.check_association(registry, true)?;
+
+        io.token = Some(token);
+        io.read_interest = interest.is_readable();
+        io.write_interest = interest.is_writable();
+        drop(io);
+
+        Inner::post_register(&self.inner, None);
+
+        Ok(())
+    }
+
+    fn deregister(&mut self, registry: &Registry) -> io::Result<()> {
+        let mut io = self.inner.io.lock().unwrap();
+
+        io.check_association(registry, true)?;
+
+        if io.token.is_none() {
+            return Err(io::Error::new(
+                io::ErrorKind::NotFound,
+                "I/O source not registered with `Registry`",
+            ));
+        }
+
+        io.token = None;
+        Ok(())
+    }
+}
+
+impl AsRawHandle for NamedPipe {
+    fn as_raw_handle(&self) -> RawHandle {
+        self.inner.handle.as_raw_handle()
+    }
+}
+
+impl fmt::Debug for NamedPipe {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.inner.handle.fmt(f)
+    }
+}
+
+impl Drop for NamedPipe {
+    fn drop(&mut self) {
+        // Cancel pending reads/connects, but don't cancel writes to ensure that
+        // everything is flushed out.
+        unsafe {
+            if self.inner.connecting.load(SeqCst) {
+                drop(cancel(&self.inner.handle, &self.inner.connect));
+            }
+
+            let io = self.inner.io.lock().unwrap();
+
+            match io.read {
+                State::Pending(..) => {
+                    drop(cancel(&self.inner.handle, &self.inner.read));
+                }
+                _ => {}
+            }
+        }
+    }
+}
+
+impl Inner {
+    /// Schedules a read to happen in the background, executing an overlapped
+    /// operation.
+    ///
+    /// This function returns `true` if a normal error happens or if the read
+    /// is scheduled in the background. If the pipe is no longer connected
+    /// (ERROR_PIPE_LISTENING) then `false` is returned and no read is
+    /// scheduled.
+    fn schedule_read(me: &Arc<Inner>, io: &mut Io, events: Option<&mut Vec<Event>>) -> bool {
+        // Check to see if a read is already scheduled/completed
+        match io.read {
+            State::None => {}
+            _ => return true,
+        }
+
+        // Allocate a buffer and schedule the read.
+        let mut buf = me.get_buffer();
+        let e = unsafe {
+            let overlapped = me.read.as_ptr() as *mut _;
+            let slice = slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.capacity());
+            me.handle.read_overlapped(slice, overlapped)
+        };
+
+        match e {
+            // See `NamedPipe::connect` above for the rationale behind `forget`
+            Ok(_) => {
+                io.read = State::Pending(buf, 0); // 0 is ignored on read side
+                mem::forget(me.clone());
+                true
+            }
+
+            // If ERROR_PIPE_LISTENING happens then it's not a real read error,
+            // we just need to wait for a connect.
+            Err(ref e) if e.raw_os_error() == Some(ERROR_PIPE_LISTENING as i32) => false,
+
+            // If some other error happened, though, we're now readable to give
+            // out the error.
+            Err(e) => {
+                io.read = State::Err(e);
+                io.notify_readable(events);
+                true
+            }
+        }
+    }
+
+    fn schedule_write(me: &Arc<Inner>, buf: Vec<u8>, pos: usize, io: &mut Io, events: Option<&mut Vec<Event>>) {
+        // Very similar to `schedule_read` above, just done for the write half.
+        let e = unsafe {
+            let overlapped = me.write.as_ptr() as *mut _;
+            me.handle.write_overlapped(&buf[pos..], overlapped)
+        };
+
+        match e {
+            // See `connect` above for the rationale behind `forget`
+            Ok(_) => {
+                io.write = State::Pending(buf, pos);
+                mem::forget(me.clone())
+            }
+            Err(e) => {
+                io.write = State::Err(e);
+                io.notify_writable(events);
+            }
+        }
+    }
+
+    fn post_register(me: &Arc<Inner>, mut events: Option<&mut Vec<Event>>) {
+        let mut io = me.io.lock().unwrap();
+        if Inner::schedule_read(&me, &mut io, events.as_mut().map(|ptr| &mut **ptr)) {
+            if let State::None = io.write {
+                io.notify_writable(events);
+            }
+        }
+    }
+
+    fn get_buffer(&self) -> Vec<u8> {
+        self.pool.lock().unwrap().get(4 * 1024)
+    }
+
+    fn put_buffer(&self, buf: Vec<u8>) {
+        self.pool.lock().unwrap().put(buf)
+    }
+}
+
+unsafe fn cancel<T: AsRawHandle>(handle: &T, overlapped: &Overlapped) -> io::Result<()> {
+    let ret = CancelIoEx(handle.as_raw_handle(), overlapped.as_ptr() as *mut _);
+    // `CancelIoEx` returns 0 on error:
+    // https://docs.microsoft.com/en-us/windows/win32/fileio/cancelioex-func
+    if ret == 0 {
+        Err(io::Error::last_os_error())
+    } else {
+        Ok(())
+    }
+}
+
+fn connect_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec<Event>>) {
+    let status = CompletionStatus::from_entry(status);
+
+    // Acquire the `Arc<Inner>`. Note that we should be guaranteed that
+    // the refcount is available to us due to the `mem::forget` in
+    // `connect` above.
+    let me = unsafe { overlapped2arc!(status.overlapped(), Inner, connect) };
+
+    // Flag ourselves as no longer using the `connect` overlapped instances.
+    let prev = me.connecting.swap(false, SeqCst);
+    assert!(prev, "NamedPipe was not previously connecting");
+
+    // Stash away our connect error if one happened
+    debug_assert_eq!(status.bytes_transferred(), 0);
+    unsafe {
+        match me.handle.result(status.overlapped()) {
+            Ok(n) => debug_assert_eq!(n, 0),
+            Err(e) => me.io.lock().unwrap().connect_error = Some(e),
+        }
+    }
+
+    // We essentially just finished a registration, so kick off a
+    // read and register write readiness.
+    Inner::post_register(&me, events);
+}
+
+fn read_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec<Event>>) {
+    let status = CompletionStatus::from_entry(status);
+
+    // Acquire the `FromRawArc<Inner>`. Note that we should be guaranteed that
+    // the refcount is available to us due to the `mem::forget` in
+    // `schedule_read` above.
+    let me = unsafe { overlapped2arc!(status.overlapped(), Inner, read) };
+
+    // Move from the `Pending` to `Ok` state.
+    let mut io = me.io.lock().unwrap();
+    let mut buf = match mem::replace(&mut io.read, State::None) {
+        State::Pending(buf, _) => buf,
+        _ => unreachable!(),
+    };
+    unsafe {
+        match me.handle.result(status.overlapped()) {
+            Ok(n) => {
+                debug_assert_eq!(status.bytes_transferred() as usize, n);
+                buf.set_len(status.bytes_transferred() as usize);
+                io.read = State::Ok(buf, 0);
+            }
+            Err(e) => {
+                debug_assert_eq!(status.bytes_transferred(), 0);
+                io.read = State::Err(e);
+            }
+        }
+    }
+
+    // Flag our readiness that we've got data.
+    io.notify_readable(events);
+}
+
+fn write_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec<Event>>) {
+    let status = CompletionStatus::from_entry(status);
+
+    // Acquire the `Arc<Inner>`. Note that we should be guaranteed that
+    // the refcount is available to us due to the `mem::forget` in
+    // `schedule_write` above.
+    let me = unsafe { overlapped2arc!(status.overlapped(), Inner, write) };
+
+    // Make the state change out of `Pending`. If we wrote the entire buffer
+    // then we're writable again and otherwise we schedule another write.
+    let mut io = me.io.lock().unwrap();
+    let (buf, pos) = match mem::replace(&mut io.write, State::None) {
+        State::Pending(buf, pos) => (buf, pos),
+        _ => unreachable!(),
+    };
+
+    unsafe {
+        match me.handle.result(status.overlapped()) {
+            Ok(n) => {
+                debug_assert_eq!(status.bytes_transferred() as usize, n);
+                let new_pos = pos + (status.bytes_transferred() as usize);
+                if new_pos == buf.len() {
+                    me.put_buffer(buf);
+                    io.notify_writable(events);
+                } else {
+                    Inner::schedule_write(&me, buf, new_pos, &mut io, events);
+                }
+            }
+            Err(e) => {
+                debug_assert_eq!(status.bytes_transferred(), 0);
+                io.write = State::Err(e);
+                io.notify_writable(events);
+            }
+        }
+    }
+}
+
+impl Io {
+    fn check_association(&self, registry: &Registry, required: bool) -> io::Result<()> {
+        match self.cp {
+            Some(ref cp) if !poll::selector(registry).same_port(cp) => {
+                Err(io::Error::new(
+                    io::ErrorKind::AlreadyExists,
+                    "I/O source already registered with a different `Registry`"
+                ))
+            }
+            None if required => {
+                Err(io::Error::new(
+                    io::ErrorKind::NotFound,
+                    "I/O source not registered with `Registry`"
+                ))
+            }
+            _ => Ok(()),
+        }
+    }
+
+    fn notify_readable(&self, events: Option<&mut Vec<Event>>) {
+        if let Some(token) = self.token {
+            let mut ev = Event::new(token);
+            ev.set_readable();
+
+            if let Some(events) = events {
+                events.push(ev);
+            } else {
+                let _ = self.cp.as_ref().unwrap().post(ev.to_completion_status());
+            }
+        }
+    }
+
+    fn notify_writable(&self, events: Option<&mut Vec<Event>>) {
+        if let Some(token) = self.token {
+            let mut ev = Event::new(token);
+            ev.set_writable();
+
+            if let Some(events) = events {
+                events.push(ev);
+            } else {
+                let _ = self.cp.as_ref().unwrap().post(ev.to_completion_status());
+            }
+        }
+    }
+}
+
+struct BufferPool {
+    pool: Vec<Vec<u8>>,
+}
+
+impl BufferPool {
+    fn with_capacity(cap: usize) -> BufferPool {
+        BufferPool {
+            pool: Vec::with_capacity(cap),
+        }
+    }
+
+    fn get(&mut self, default_cap: usize) -> Vec<u8> {
+        self.pool
+            .pop()
+            .unwrap_or_else(|| Vec::with_capacity(default_cap))
+    }
+
+    fn put(&mut self, mut buf: Vec<u8>) {
+        if self.pool.len() < self.pool.capacity() {
+            unsafe {
+                buf.set_len(0);
+            }
+            self.pool.push(buf);
+        }
+    }
+}
diff --git a/src/sys/windows/net.rs b/src/sys/windows/net.rs
index 7e55c06..f825ee3 100644
--- a/src/sys/windows/net.rs
+++ b/src/sys/windows/net.rs
@@ -1,14 +1,12 @@
 use std::io;
 use std::mem::size_of_val;
 use std::net::SocketAddr;
-#[cfg(all(feature = "os-poll", feature = "tcp"))]
-use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
 use std::sync::Once;
 
 use winapi::ctypes::c_int;
 use winapi::shared::ws2def::SOCKADDR;
 use winapi::um::winsock2::{
-    ioctlsocket, socket, FIONBIO, INVALID_SOCKET, PF_INET, PF_INET6, SOCKET,
+    ioctlsocket, socket, FIONBIO, INVALID_SOCKET, SOCKET,
 };
 
 /// Initialise the network stack for Windows.
@@ -23,12 +21,19 @@ pub(crate) fn init() {
 }
 
 /// Create a new non-blocking socket.
-pub(crate) fn new_socket(addr: SocketAddr, socket_type: c_int) -> io::Result<SOCKET> {
+#[cfg(feature = "udp")]
+pub(crate) fn new_ip_socket(addr: SocketAddr, socket_type: c_int) -> io::Result<SOCKET> {
+    use winapi::um::winsock2::{PF_INET, PF_INET6};
+
     let domain = match addr {
         SocketAddr::V4(..) => PF_INET,
         SocketAddr::V6(..) => PF_INET6,
     };
 
+    new_socket(domain, socket_type)
+}
+
+pub(crate) fn new_socket(domain: c_int, socket_type: c_int) -> io::Result<SOCKET> {
     syscall!(
         socket(domain, socket_type, 0),
         PartialEq::eq,
@@ -51,19 +56,3 @@ pub(crate) fn socket_addr(addr: &SocketAddr) -> (*const SOCKADDR, c_int) {
         ),
     }
 }
-
-#[cfg(all(feature = "os-poll", feature = "tcp"))]
-pub(crate) fn inaddr_any(other: SocketAddr) -> SocketAddr {
-    match other {
-        SocketAddr::V4(..) => {
-            let any = Ipv4Addr::new(0, 0, 0, 0);
-            let addr = SocketAddrV4::new(any, 0);
-            SocketAddr::V4(addr)
-        }
-        SocketAddr::V6(..) => {
-            let any = Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0);
-            let addr = SocketAddrV6::new(any, 0, 0, 0);
-            SocketAddr::V6(addr)
-        }
-    }
-}
diff --git a/src/sys/windows/overlapped.rs b/src/sys/windows/overlapped.rs
new file mode 100644
index 0000000..3708f9e
--- /dev/null
+++ b/src/sys/windows/overlapped.rs
@@ -0,0 +1,37 @@
+use crate::sys::windows::Event;
+
+use std::cell::UnsafeCell;
+use std::fmt;
+
+use winapi::um::minwinbase::OVERLAPPED_ENTRY;
+#[cfg(feature = "os-util")]
+use winapi::um::minwinbase::OVERLAPPED;
+
+#[repr(C)]
+pub(crate) struct Overlapped {
+    inner: UnsafeCell<miow::Overlapped>,
+    pub(crate) callback: fn(&OVERLAPPED_ENTRY, Option<&mut Vec<Event>>),
+}
+
+#[cfg(feature = "os-util")]
+impl Overlapped {
+    pub(crate) fn new(cb: fn(&OVERLAPPED_ENTRY, Option<&mut Vec<Event>>)) -> Overlapped {
+        Overlapped {
+            inner: UnsafeCell::new(miow::Overlapped::zero()),
+            callback: cb,
+        }
+    }
+
+    pub(crate) fn as_ptr(&self) -> *const OVERLAPPED {
+        unsafe { (*self.inner.get()).raw() }
+    }
+}
+
+impl fmt::Debug for Overlapped {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Overlapped").finish()
+    }
+}
+
+unsafe impl Send for Overlapped {}
+unsafe impl Sync for Overlapped {}
diff --git a/src/sys/windows/selector.rs b/src/sys/windows/selector.rs
index b1395ac..4a38300 100644
--- a/src/sys/windows/selector.rs
+++ b/src/sys/windows/selector.rs
@@ -1,14 +1,16 @@
 use super::afd::{self, Afd, AfdPollInfo};
 use super::io_status_block::IoStatusBlock;
 use super::Event;
-use crate::sys::event::{
-    ERROR_FLAGS, READABLE_FLAGS, READ_CLOSED_FLAGS, WRITABLE_FLAGS, WRITE_CLOSED_FLAGS,
-};
 use crate::sys::Events;
-use crate::Interest;
+
+cfg_net! {
+    use crate::sys::event::{
+        ERROR_FLAGS, READABLE_FLAGS, READ_CLOSED_FLAGS, WRITABLE_FLAGS, WRITE_CLOSED_FLAGS,
+    };
+    use crate::Interest;
+}
 
 use miow::iocp::{CompletionPort, CompletionStatus};
-use miow::Overlapped;
 use std::collections::VecDeque;
 use std::marker::PhantomPinned;
 use std::os::windows::io::RawSocket;
@@ -18,18 +20,13 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
-use std::{io, ptr};
+use std::io;
 use winapi::shared::ntdef::NT_SUCCESS;
 use winapi::shared::ntdef::{HANDLE, PVOID};
 use winapi::shared::ntstatus::STATUS_CANCELLED;
 use winapi::shared::winerror::{ERROR_INVALID_HANDLE, ERROR_IO_PENDING, WAIT_TIMEOUT};
 use winapi::um::minwinbase::OVERLAPPED;
 
-/// Overlapped value to indicate a `Waker` event.
-//
-// Note: this must be null, `SelectorInner::feed_events` depends on it.
-pub const WAKER_OVERLAPPED: *mut Overlapped = ptr::null_mut();
-
 #[derive(Debug)]
 struct AfdGroup {
     cp: Arc<CompletionPort>,
@@ -232,15 +229,7 @@ impl SockState {
         // In mio, we have to simulate Edge-triggered behavior to match API usage.
         // The strategy here is to intercept all read/write from user that could cause WouldBlock usage,
         // then reregister the socket to reset the interests.
-
-        // Reset readable event
-        if (afd_events & interests_to_afd_flags(Interest::READABLE)) != 0 {
-            self.user_evts &= !(interests_to_afd_flags(Interest::READABLE));
-        }
-        // Reset writable event
-        if (afd_events & interests_to_afd_flags(Interest::WRITABLE)) != 0 {
-            self.user_evts &= !interests_to_afd_flags(Interest::WRITABLE);
-        }
+        self.user_evts &= !afd_events;
 
         Some(Event {
             data: self.user_data,
@@ -339,7 +328,7 @@ pub struct Selector {
     #[cfg(debug_assertions)]
     id: usize,
 
-    inner: Arc<SelectorInner>,
+    pub(super) inner: Arc<SelectorInner>,
 }
 
 impl Selector {
@@ -374,6 +363,11 @@ impl Selector {
     pub(super) fn clone_port(&self) -> Arc<CompletionPort> {
         self.inner.cp.clone()
     }
+
+    #[cfg(feature = "os-util")]
+    pub(super) fn same_port(&self, other: &Arc<CompletionPort>) -> bool {
+        Arc::ptr_eq(&self.inner.cp, other)
+    }
 }
 
 cfg_net! {
@@ -408,7 +402,7 @@ cfg_net! {
 
 #[derive(Debug)]
 pub struct SelectorInner {
-    cp: Arc<CompletionPort>,
+    pub(super) cp: Arc<CompletionPort>,
     update_queue: Mutex<VecDeque<Pin<Arc<Mutex<SockState>>>>>,
     afd_group: AfdGroup,
     is_polling: AtomicBool,
@@ -499,13 +493,17 @@ impl SelectorInner {
         let mut update_queue = self.update_queue.lock().unwrap();
         for iocp_event in iocp_events.iter() {
             if iocp_event.overlapped().is_null() {
-                // `Waker` event, we'll add a readable event to match the other platforms.
-                events.push(Event {
-                    flags: afd::POLL_RECEIVE,
-                    data: iocp_event.token() as u64,
-                });
+                events.push(Event::from_completion_status(iocp_event));
                 n += 1;
                 continue;
+            } else if iocp_event.token() % 2 == 1 {
+                // Handle is a named pipe. This could be extended to be any non-AFD event.
+                let callback = (*(iocp_event.overlapped() as *mut super::Overlapped)).callback;
+    
+                let len = events.len();
+                callback(iocp_event.entry(), Some(events));
+                n += events.len() - len;
+                continue;
             }
 
             let sock_state = from_overlapped(iocp_event.overlapped());
@@ -696,7 +694,16 @@ impl Drop for SelectorInner {
                 Ok(iocp_events) => {
                     events_num = iocp_events.iter().len();
                     for iocp_event in iocp_events.iter() {
-                        if !iocp_event.overlapped().is_null() {
+                        if iocp_event.overlapped().is_null() {
+                            // Custom event
+                        } else if iocp_event.token() % 2 == 1 {
+                            // Named pipe, dispatch the event so it can release resources
+                            let callback = unsafe {
+                                (*(iocp_event.overlapped() as *mut super::Overlapped)).callback
+                            };
+                
+                            callback(iocp_event.entry(), None);
+                        } else {
                             // drain sock state to release memory of Arc reference
                             let _sock_state = from_overlapped(iocp_event.overlapped());
                         }
@@ -718,16 +725,18 @@ impl Drop for SelectorInner {
     }
 }
 
-fn interests_to_afd_flags(interests: Interest) -> u32 {
-    let mut flags = 0;
+cfg_net! {
+    fn interests_to_afd_flags(interests: Interest) -> u32 {
+        let mut flags = 0;
 
-    if interests.is_readable() {
-        flags |= READABLE_FLAGS | READ_CLOSED_FLAGS | ERROR_FLAGS;
-    }
+        if interests.is_readable() {
+            flags |= READABLE_FLAGS | READ_CLOSED_FLAGS | ERROR_FLAGS;
+        }
 
-    if interests.is_writable() {
-        flags |= WRITABLE_FLAGS | WRITE_CLOSED_FLAGS | ERROR_FLAGS;
-    }
+        if interests.is_writable() {
+            flags |= WRITABLE_FLAGS | WRITE_CLOSED_FLAGS | ERROR_FLAGS;
+        }
 
-    flags
-}
+        flags
+    }
+}
\ No newline at end of file
diff --git a/src/sys/windows/tcp.rs b/src/sys/windows/tcp.rs
index 1c2aa3f..46ac1ac 100644
--- a/src/sys/windows/tcp.rs
+++ b/src/sys/windows/tcp.rs
@@ -1,71 +1,111 @@
 use std::io;
+use std::mem::size_of;
 use std::net::{self, SocketAddr};
+use std::time::Duration;
 use std::os::windows::io::FromRawSocket;
 use std::os::windows::raw::SOCKET as StdSocket; // winapi uses usize, stdlib uses u32/u64.
 
+use winapi::ctypes::{c_char, c_int, c_ushort};
+use winapi::shared::minwindef::{BOOL, TRUE, FALSE};
 use winapi::um::winsock2::{
-    bind as win_bind, closesocket, connect as win_connect, listen, SOCKET_ERROR, SOCK_STREAM,
+    self, closesocket, linger, setsockopt, PF_INET, PF_INET6, SOCKET, SOCKET_ERROR,
+    SOCK_STREAM, SOL_SOCKET, SO_LINGER, SO_REUSEADDR,
 };
 
-use crate::sys::windows::net::{inaddr_any, init, new_socket, socket_addr};
+use crate::sys::windows::net::{init, new_socket, socket_addr};
 
-pub fn connect(addr: SocketAddr) -> io::Result<net::TcpStream> {
+pub(crate) type TcpSocket = SOCKET;
+
+pub(crate) fn new_v4_socket() -> io::Result<TcpSocket> {
     init();
-    new_socket(addr, SOCK_STREAM)
-        .and_then(|socket| {
-            // Required for a future `connect_overlapped` operation to be
-            // executed successfully.
-            let any_addr = inaddr_any(addr);
-            let (raw_addr, raw_addr_length) = socket_addr(&any_addr);
-            syscall!(
-                win_bind(socket, raw_addr, raw_addr_length),
-                PartialEq::eq,
-                SOCKET_ERROR
-            )
-            .and_then(|_| {
-                let (raw_addr, raw_addr_length) = socket_addr(&addr);
-                syscall!(
-                    win_connect(socket, raw_addr, raw_addr_length),
-                    PartialEq::eq,
-                    SOCKET_ERROR
-                )
-                .or_else(|err| match err {
-                    ref err if err.kind() == io::ErrorKind::WouldBlock => Ok(0),
-                    err => Err(err),
-                })
-            })
-            .map(|_| socket)
-            .map_err(|err| {
-                // Close the socket if we hit an error, ignoring the error
-                // from closing since we can't pass back two errors.
-                let _ = unsafe { closesocket(socket) };
-                err
-            })
-        })
-        .map(|socket| unsafe { net::TcpStream::from_raw_socket(socket as StdSocket) })
+    new_socket(PF_INET, SOCK_STREAM)
 }
 
-pub fn bind(addr: SocketAddr) -> io::Result<net::TcpListener> {
+pub(crate) fn new_v6_socket() -> io::Result<TcpSocket> {
     init();
-    new_socket(addr, SOCK_STREAM).and_then(|socket| {
-        let (raw_addr, raw_addr_length) = socket_addr(&addr);
-        syscall!(
-            win_bind(socket, raw_addr, raw_addr_length,),
-            PartialEq::eq,
-            SOCKET_ERROR
-        )
-        .and_then(|_| syscall!(listen(socket, 1024), PartialEq::eq, SOCKET_ERROR))
-        .map_err(|err| {
-            // Close the socket if we hit an error, ignoring the error
-            // from closing since we can't pass back two errors.
-            let _ = unsafe { closesocket(socket) };
-            err
-        })
-        .map(|_| unsafe { net::TcpListener::from_raw_socket(socket as StdSocket) })
-    })
+    new_socket(PF_INET6, SOCK_STREAM)
+}
+
+pub(crate) fn bind(socket: TcpSocket, addr: SocketAddr) -> io::Result<()> {
+    use winsock2::bind;
+
+    let (raw_addr, raw_addr_length) = socket_addr(&addr);
+    syscall!(
+        bind(socket, raw_addr, raw_addr_length),
+        PartialEq::eq,
+        SOCKET_ERROR
+    )?;
+    Ok(())
+}
+
+pub(crate) fn connect(socket: TcpSocket, addr: SocketAddr) -> io::Result<net::TcpStream> {
+    use winsock2::connect;
+
+    let (raw_addr, raw_addr_length) = socket_addr(&addr);
+
+    let res = syscall!(
+        connect(socket, raw_addr, raw_addr_length),
+        PartialEq::eq,
+        SOCKET_ERROR
+    );
+
+    match res {
+        Err(err) if err.kind() != io::ErrorKind::WouldBlock => {
+            Err(err)
+        }
+        _ => {
+            Ok(unsafe { net::TcpStream::from_raw_socket(socket as StdSocket) })
+        }
+    }
+}
+
+pub(crate) fn listen(socket: TcpSocket, backlog: u32) -> io::Result<net::TcpListener> {
+    use winsock2::listen;
+    use std::convert::TryInto;
+
+    let backlog = backlog.try_into().unwrap_or(i32::max_value());
+    syscall!(listen(socket, backlog), PartialEq::eq, SOCKET_ERROR)?;
+    Ok(unsafe { net::TcpListener::from_raw_socket(socket as StdSocket) })
+}
+
+pub(crate) fn close(socket: TcpSocket) {
+    let _ = unsafe { closesocket(socket) };
+}
+
+pub(crate) fn set_reuseaddr(socket: TcpSocket, reuseaddr: bool) -> io::Result<()> {
+    let val: BOOL = if reuseaddr { TRUE } else { FALSE };
+
+    match unsafe { setsockopt(
+        socket,
+        SOL_SOCKET,
+        SO_REUSEADDR,
+        &val as *const _ as *const c_char,
+        size_of::<BOOL>() as c_int,
+    ) } {
+        SOCKET_ERROR => Err(io::Error::last_os_error()),
+        _ => Ok(()),
+    }
+}
+
+pub(crate) fn set_linger(socket: TcpSocket, dur: Option<Duration>) -> io::Result<()> {
+    let val: linger = linger {
+        l_onoff: if dur.is_some() { 1 } else { 0 },
+        l_linger: dur.map(|dur| dur.as_secs() as c_ushort).unwrap_or_default(),
+    };
+
+    match unsafe { setsockopt(
+        socket,
+        SOL_SOCKET,
+        SO_LINGER,
+        &val as *const _ as *const c_char,
+        size_of::<linger>() as c_int,
+    ) } {
+        SOCKET_ERROR => Err(io::Error::last_os_error()),
+        _ => Ok(()),
+    }
 }
 
-pub fn accept(listener: &net::TcpListener) -> io::Result<(net::TcpStream, SocketAddr)> {
+pub(crate) fn accept(listener: &net::TcpListener) -> io::Result<(net::TcpStream, SocketAddr)> {
     // The non-blocking state of `listener` is inherited. See
     // https://docs.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-accept#remarks.
     listener.accept()
diff --git a/src/sys/windows/udp.rs b/src/sys/windows/udp.rs
index 0f83fd0..667c775 100644
--- a/src/sys/windows/udp.rs
+++ b/src/sys/windows/udp.rs
@@ -5,11 +5,11 @@ use std::os::windows::raw::SOCKET as StdSocket; // winapi uses usize, stdlib use
 
 use winapi::um::winsock2::{bind as win_bind, closesocket, SOCKET_ERROR, SOCK_DGRAM};
 
-use crate::sys::windows::net::{init, new_socket, socket_addr};
+use crate::sys::windows::net::{init, new_ip_socket, socket_addr};
 
 pub fn bind(addr: SocketAddr) -> io::Result<net::UdpSocket> {
     init();
-    new_socket(addr, SOCK_DGRAM).and_then(|socket| {
+    new_ip_socket(addr, SOCK_DGRAM).and_then(|socket| {
         let (raw_addr, raw_addr_length) = socket_addr(&addr);
         syscall!(
             win_bind(socket, raw_addr, raw_addr_length,),
diff --git a/src/sys/windows/waker.rs b/src/sys/windows/waker.rs
index 1b21cf1..ab12c3c 100644
--- a/src/sys/windows/waker.rs
+++ b/src/sys/windows/waker.rs
@@ -1,8 +1,8 @@
-use crate::sys::windows::selector::WAKER_OVERLAPPED;
+use crate::sys::windows::Event;
 use crate::sys::windows::Selector;
 use crate::Token;
 
-use miow::iocp::{CompletionPort, CompletionStatus};
+use miow::iocp::CompletionPort;
 use std::io;
 use std::sync::Arc;
 
@@ -21,8 +21,9 @@ impl Waker {
     }
 
     pub fn wake(&self) -> io::Result<()> {
-        // Keep NULL as Overlapped value to notify waking.
-        let status = CompletionStatus::new(0, self.token.0, WAKER_OVERLAPPED);
-        self.port.post(status)
+        let mut ev = Event::new(self.token);
+        ev.set_readable();
+
+        self.port.post(ev.to_completion_status())
     }
 }


Test patch:
diff --git a/tests/tcp.rs b/tests/tcp.rs
index a83f6d7..9691a1a 100644
--- a/tests/tcp.rs
+++ b/tests/tcp.rs
@@ -1,13 +1,9 @@
 #![cfg(all(feature = "os-poll", feature = "tcp"))]
 
-use mio::net::{TcpListener, TcpStream};
+use mio::net::{TcpListener, TcpSocket, TcpStream};
 use mio::{Events, Interest, Poll, Token};
-#[cfg(unix)]
-use net2::TcpStreamExt;
 use std::io::{self, Read, Write};
 use std::net::{self, Shutdown};
-#[cfg(unix)]
-use std::os::unix::io::{FromRawFd, IntoRawFd};
 use std::sync::mpsc::channel;
 use std::thread::{self, sleep};
 use std::time::Duration;
@@ -459,7 +455,6 @@ fn multiple_writes_immediate_success() {
     handle.join().unwrap();
 }
 
-#[cfg(unix)]
 #[test]
 fn connection_reset_by_peer() {
     init();
@@ -473,14 +468,9 @@ fn connection_reset_by_peer() {
     let addr = listener.local_addr().unwrap();
 
     // Connect client
-    let client = net2::TcpBuilder::new_v4().unwrap().to_tcp_stream().unwrap();
-
+    let client = TcpSocket::new_v4().unwrap();
     client.set_linger(Some(Duration::from_millis(0))).unwrap();
-    client.connect(&addr).unwrap();
-
-    // Convert to Mio stream
-    // FIXME: how to convert the stream on Windows?
-    let mut client = unsafe { TcpStream::from_raw_fd(client.into_raw_fd()) };
+    let mut client = client.connect(addr).unwrap();
 
     // Register server
     poll.registry()
diff --git a/tests/tcp_socket.rs b/tests/tcp_socket.rs
new file mode 100644
index 0000000..e741db9
--- /dev/null
+++ b/tests/tcp_socket.rs
@@ -0,0 +1,23 @@
+#![cfg(all(feature = "os-poll", feature = "tcp"))]
+
+use mio::net::TcpSocket;
+
+#[test]
+fn is_send_and_sync() {
+    fn is_send<T: Send>() {}
+    fn is_sync<T: Sync>() {}
+
+    is_send::<TcpSocket>();
+    is_sync::<TcpSocket>();
+}
+
+#[test]
+fn set_reuseaddr() {
+    let addr = "127.0.0.1:0".parse().unwrap();
+
+    let socket = TcpSocket::new_v4().unwrap();
+    socket.set_reuseaddr(true).unwrap();
+    socket.bind(addr).unwrap();
+
+    let _ = socket.listen(128).unwrap();
+}
diff --git a/tests/tcp_stream.rs b/tests/tcp_stream.rs
index b359089..b9ed863 100644
--- a/tests/tcp_stream.rs
+++ b/tests/tcp_stream.rs
@@ -6,6 +6,8 @@ use std::io::{self, IoSlice, IoSliceMut, Read, Write};
 use std::net::{self, Shutdown, SocketAddr};
 #[cfg(unix)]
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd};
+#[cfg(windows)]
+use std::os::windows::io::{AsRawSocket, FromRawSocket};
 use std::sync::{mpsc::channel, Arc, Barrier};
 use std::thread;
 use std::time::Duration;
@@ -749,3 +751,58 @@ fn start_listener(
     });
     (thread_handle, receiver.recv().unwrap())
 }
+
+#[test]
+fn hup_event_on_disconnect() {
+    use mio::net::TcpListener;
+
+    let (mut poll, mut events) = init_with_poll();
+    let addr = "127.0.0.1:0".parse().unwrap();
+
+    let mut listener = TcpListener::bind(addr).unwrap();
+    let addr = listener.local_addr().unwrap();
+    poll.registry()
+        .register(&mut listener, Token(0), Interest::READABLE)
+        .unwrap();
+
+    let mut stream = TcpStream::connect(addr).unwrap();
+    poll.registry()
+        .register(
+            &mut stream,
+            Token(1),
+            Interest::READABLE | Interest::WRITABLE,
+        )
+        .unwrap();
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![
+            ExpectEvent::new(Token(0), Interest::READABLE),
+            ExpectEvent::new(Token(1), Interest::WRITABLE),
+        ],
+    );
+
+    let (sock, _) = listener.accept().unwrap();
+    // Prevent the OS from performing a graceful shutdown
+    set_linger_zero(&sock);
+    drop(sock);
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(Token(1), Interest::READABLE)],
+    );
+}
+
+fn set_linger_zero(socket: &TcpStream) {
+    use mio::net::TcpSocket;
+
+    #[cfg(windows)]
+    let s = unsafe { TcpSocket::from_raw_socket(socket.as_raw_socket()) };
+    #[cfg(unix)]
+    let s = unsafe { TcpSocket::from_raw_fd(socket.as_raw_fd()) };
+
+    s.set_linger(Some(Duration::from_millis(0))).unwrap();
+    std::mem::drop(s);
+}
diff --git a/tests/win_named_pipe.rs b/tests/win_named_pipe.rs
new file mode 100644
index 0000000..65ca3ea
--- /dev/null
+++ b/tests/win_named_pipe.rs
@@ -0,0 +1,314 @@
+#![cfg(windows)]
+
+use std::fs::OpenOptions;
+use std::io::{self, Read, Write};
+use std::os::windows::fs::OpenOptionsExt;
+use std::os::windows::io::{FromRawHandle, IntoRawHandle};
+use std::time::Duration;
+
+use mio::windows::NamedPipe;
+use mio::{Events, Interest, Poll, Token};
+use rand::Rng;
+use winapi::um::winbase::FILE_FLAG_OVERLAPPED;
+
+fn _assert_kinds() {
+    fn _assert_send<T: Send>() {}
+    fn _assert_sync<T: Sync>() {}
+    _assert_send::<NamedPipe>();
+    _assert_sync::<NamedPipe>();
+}
+
+macro_rules! t {
+    ($e:expr) => {
+        match $e {
+            Ok(e) => e,
+            Err(e) => panic!("{} failed with {}", stringify!($e), e),
+        }
+    };
+}
+
+fn server() -> (NamedPipe, String) {
+    let num: u64 = rand::thread_rng().gen();
+    let name = format!(r"\\.\pipe\my-pipe-{}", num);
+    let pipe = t!(NamedPipe::new(&name));
+    (pipe, name)
+}
+
+fn client(name: &str) -> NamedPipe {
+    let mut opts = OpenOptions::new();
+    opts.read(true)
+        .write(true)
+        .custom_flags(FILE_FLAG_OVERLAPPED);
+    let file = t!(opts.open(name));
+    unsafe { NamedPipe::from_raw_handle(file.into_raw_handle()) }
+}
+
+fn pipe() -> (NamedPipe, NamedPipe) {
+    let (pipe, name) = server();
+    (pipe, client(&name))
+}
+
+#[test]
+fn writable_after_register() {
+    let (mut server, mut client) = pipe();
+    let mut poll = t!(Poll::new());
+    t!(poll.registry().register(
+        &mut server,
+        Token(0),
+        Interest::WRITABLE | Interest::READABLE,
+    ));
+    t!(poll
+        .registry()
+        .register(&mut client, Token(1), Interest::WRITABLE));
+
+    let mut events = Events::with_capacity(128);
+    t!(poll.poll(&mut events, None));
+
+    let events = events.iter().collect::<Vec<_>>();
+    assert!(events
+        .iter()
+        .any(|e| { e.token() == Token(0) && e.is_writable() }));
+    assert!(events
+        .iter()
+        .any(|e| { e.token() == Token(1) && e.is_writable() }));
+}
+
+#[test]
+fn write_then_read() {
+    let (mut server, mut client) = pipe();
+    let mut poll = t!(Poll::new());
+    t!(poll.registry().register(
+        &mut server,
+        Token(0),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+    t!(poll.registry().register(
+        &mut client,
+        Token(1),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+
+    let mut events = Events::with_capacity(128);
+    t!(poll.poll(&mut events, None));
+
+    assert_eq!(t!(client.write(b"1234")), 4);
+
+    loop {
+        t!(poll.poll(&mut events, None));
+        let events = events.iter().collect::<Vec<_>>();
+        if let Some(event) = events.iter().find(|e| e.token() == Token(0)) {
+            if event.is_readable() {
+                break;
+            }
+        }
+    }
+
+    let mut buf = [0; 10];
+    assert_eq!(t!(server.read(&mut buf)), 4);
+    assert_eq!(&buf[..4], b"1234");
+}
+
+#[test]
+fn connect_before_client() {
+    let (mut server, name) = server();
+    let mut poll = t!(Poll::new());
+    t!(poll.registry().register(
+        &mut server,
+        Token(0),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+
+    let mut events = Events::with_capacity(128);
+    t!(poll.poll(&mut events, Some(Duration::new(0, 0))));
+    let e = events.iter().collect::<Vec<_>>();
+    assert_eq!(e.len(), 0);
+    assert_eq!(
+        server.connect().err().unwrap().kind(),
+        io::ErrorKind::WouldBlock
+    );
+
+    let mut client = client(&name);
+    t!(poll.registry().register(
+        &mut client,
+        Token(1),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+    loop {
+        t!(poll.poll(&mut events, None));
+        let e = events.iter().collect::<Vec<_>>();
+        if let Some(event) = e.iter().find(|e| e.token() == Token(0)) {
+            if event.is_writable() {
+                break;
+            }
+        }
+    }
+}
+
+#[test]
+fn connect_after_client() {
+    let (mut server, name) = server();
+    let mut poll = t!(Poll::new());
+    t!(poll.registry().register(
+        &mut server,
+        Token(0),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+
+    let mut events = Events::with_capacity(128);
+    t!(poll.poll(&mut events, Some(Duration::new(0, 0))));
+    let e = events.iter().collect::<Vec<_>>();
+    assert_eq!(e.len(), 0);
+
+    let mut client = client(&name);
+    t!(poll.registry().register(
+        &mut client,
+        Token(1),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+    t!(server.connect());
+    loop {
+        t!(poll.poll(&mut events, None));
+        let e = events.iter().collect::<Vec<_>>();
+        if let Some(event) = e.iter().find(|e| e.token() == Token(0)) {
+            if event.is_writable() {
+                break;
+            }
+        }
+    }
+}
+
+#[test]
+fn write_then_drop() {
+    let (mut server, mut client) = pipe();
+    let mut poll = t!(Poll::new());
+    t!(poll.registry().register(
+        &mut server,
+        Token(0),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+    t!(poll.registry().register(
+        &mut client,
+        Token(1),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+    assert_eq!(t!(client.write(b"1234")), 4);
+    drop(client);
+
+    let mut events = Events::with_capacity(128);
+
+    'outer: loop {
+        t!(poll.poll(&mut events, None));
+        let events = events.iter().collect::<Vec<_>>();
+
+        for event in &events {
+            if event.is_readable() && event.token() == Token(0) {
+                break 'outer;
+            }
+        }
+    }
+
+    let mut buf = [0; 10];
+    assert_eq!(t!(server.read(&mut buf)), 4);
+    assert_eq!(&buf[..4], b"1234");
+}
+
+#[test]
+fn connect_twice() {
+    let (mut server, name) = server();
+    let mut c1 = client(&name);
+    let mut poll = t!(Poll::new());
+    t!(poll.registry().register(
+        &mut server,
+        Token(0),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+    t!(poll
+        .registry()
+        .register(&mut c1, Token(1), Interest::READABLE | Interest::WRITABLE,));
+    drop(c1);
+
+    let mut events = Events::with_capacity(128);
+
+    loop {
+        t!(poll.poll(&mut events, None));
+        let events = events.iter().collect::<Vec<_>>();
+        if let Some(event) = events.iter().find(|e| e.token() == Token(0)) {
+            if event.is_readable() {
+                let mut buf = [0; 10];
+
+                match server.read(&mut buf) {
+                    Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
+                    Ok(0) => break,
+                    res => panic!("{:?}", res),
+                }
+            }
+        }
+    }
+
+    t!(server.disconnect());
+    assert_eq!(
+        server.connect().err().unwrap().kind(),
+        io::ErrorKind::WouldBlock
+    );
+
+    let mut c2 = client(&name);
+    t!(poll
+        .registry()
+        .register(&mut c2, Token(2), Interest::READABLE | Interest::WRITABLE,));
+
+    'outer: loop {
+        t!(poll.poll(&mut events, None));
+        let events = events.iter().collect::<Vec<_>>();
+
+        for event in &events {
+            if event.is_writable() && event.token() == Token(0) {
+                break 'outer;
+            }
+        }
+    }
+}
+
+#[test]
+fn reregister_deregister_before_register() {
+    let (mut pipe, _) = server();
+    let poll = t!(Poll::new());
+
+    assert_eq!(
+        poll.registry()
+            .reregister(&mut pipe, Token(0), Interest::READABLE,)
+            .unwrap_err()
+            .kind(),
+        io::ErrorKind::NotFound,
+    );
+
+    assert_eq!(
+        poll.registry().deregister(&mut pipe).unwrap_err().kind(),
+        io::ErrorKind::NotFound,
+    );
+}
+
+#[test]
+fn reregister_deregister_different_poll() {
+    let (mut pipe, _) = server();
+    let poll1 = t!(Poll::new());
+    let poll2 = t!(Poll::new());
+
+    // Register with 1
+    t!(poll1
+        .registry()
+        .register(&mut pipe, Token(0), Interest::READABLE));
+
+    assert_eq!(
+        poll2
+            .registry()
+            .reregister(&mut pipe, Token(0), Interest::READABLE,)
+            .unwrap_err()
+            .kind(),
+        io::ErrorKind::AlreadyExists,
+    );
+
+    assert_eq!(
+        poll2.registry().deregister(&mut pipe).unwrap_err().kind(),
+        io::ErrorKind::AlreadyExists,
+    );
+}