Repository: tokio-rs/mio
Pull request #1360 (2020-10-07 16:23:15)

Add more registering tests
Continuation of pr #1201 by @dtacalau.

In pr #1201 the CI seems stuck.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index afc49bd..5827f0c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,22 @@
+# 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
diff --git a/Cargo.toml b/Cargo.toml
index 74779a3..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.2"
+version       = "0.7.4"
 license       = "MIT"
 authors       = ["Carl Lerche <me@carllerche.com>"]
 description   = "Lightweight non-blocking IO"
-documentation = "https://docs.rs/mio/0.7.2"
+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,6 @@ 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]
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/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 d8342a1..6e98c87 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/mio/0.7.2")]
+#![doc(html_root_url = "https://docs.rs/mio/0.7.4")]
 #![deny(
     missing_docs,
     missing_debug_implementations,
diff --git a/src/net/tcp/socket.rs b/src/net/tcp/socket.rs
index e86dc9a..a91f665 100644
--- a/src/net/tcp/socket.rs
+++ b/src/net/tcp/socket.rs
@@ -4,8 +4,11 @@ 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.
 ///
@@ -78,6 +81,11 @@ impl TcpSocket {
     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 {
@@ -95,7 +103,7 @@ impl AsRawFd for TcpSocket {
 
 #[cfg(unix)]
 impl FromRawFd for TcpSocket {
-    /// Converts a `RawFd` to a `TcpStream`.
+    /// Converts a `RawFd` to a `TcpSocket`.
     ///
     /// # Notes
     ///
@@ -104,4 +112,40 @@ impl FromRawFd for TcpSocket {
     unsafe fn from_raw_fd(fd: RawFd) -> TcpSocket {
         TcpSocket { sys: fd }
     }
-}
\ No newline at end of file
+}
+
+#[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/poll.rs b/src/poll.rs
index 6446281..7ff2038 100644
--- a/src/poll.rs
+++ b/src/poll.rs
@@ -613,6 +613,15 @@ impl Registry {
             .try_clone()
             .map(|selector| Registry { selector })
     }
+
+    /// Internal check to ensure only a single `Waker` is active per [`Poll`]
+    /// instance.
+    #[cfg(debug_assertions)]
+    pub(crate) fn register_waker(&self) {
+        if self.selector.register_waker() {
+            panic!("Only a single `Waker` can be active per `Poll` instance");
+        }
+    }
 }
 
 impl fmt::Debug for Registry {
diff --git a/src/sys/shell/selector.rs b/src/sys/shell/selector.rs
index 0e0c031..5667628 100644
--- a/src/sys/shell/selector.rs
+++ b/src/sys/shell/selector.rs
@@ -18,6 +18,11 @@ impl Selector {
     pub fn select(&self, _: &mut Events, _: Option<Duration>) -> io::Result<()> {
         os_required!();
     }
+
+    #[cfg(debug_assertions)]
+    pub fn register_waker(&self) -> bool {
+        os_required!();
+    }
 }
 
 #[cfg(unix)]
diff --git a/src/sys/shell/tcp.rs b/src/sys/shell/tcp.rs
index 963fccb..de1520b 100644
--- a/src/sys/shell/tcp.rs
+++ b/src/sys/shell/tcp.rs
@@ -1,5 +1,6 @@
 use std::io;
 use std::net::{self, SocketAddr};
+use std::time::Duration;
 
 pub(crate) type TcpSocket = i32;
 
@@ -31,6 +32,10 @@ pub(crate) fn set_reuseaddr(_: TcpSocket, _: bool) -> io::Result<()> {
     os_required!();
 }
 
+pub(crate) fn set_linger(_: TcpSocket, _: Option<Duration>) -> io::Result<()> {
+    os_required!();
+}
+
 pub fn accept(_: &net::TcpListener) -> io::Result<(net::TcpStream, SocketAddr)> {
     os_required!();
 }
diff --git a/src/sys/unix/selector/epoll.rs b/src/sys/unix/selector/epoll.rs
index 13f1617..b8a31a1 100644
--- a/src/sys/unix/selector/epoll.rs
+++ b/src/sys/unix/selector/epoll.rs
@@ -4,7 +4,7 @@ use libc::{EPOLLET, EPOLLIN, EPOLLOUT, EPOLLRDHUP};
 use log::error;
 use std::os::unix::io::{AsRawFd, RawFd};
 #[cfg(debug_assertions)]
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::time::Duration;
 use std::{cmp, i32, io, ptr};
 
@@ -17,6 +17,8 @@ pub struct Selector {
     #[cfg(debug_assertions)]
     id: usize,
     ep: RawFd,
+    #[cfg(debug_assertions)]
+    has_waker: AtomicBool,
 }
 
 impl Selector {
@@ -33,6 +35,8 @@ impl Selector {
             #[cfg(debug_assertions)]
             id: NEXT_ID.fetch_add(1, Ordering::Relaxed),
             ep,
+            #[cfg(debug_assertions)]
+            has_waker: AtomicBool::new(false),
         })
     }
 
@@ -42,6 +46,8 @@ impl Selector {
             #[cfg(debug_assertions)]
             id: self.id,
             ep,
+            #[cfg(debug_assertions)]
+            has_waker: AtomicBool::new(self.has_waker.load(Ordering::Acquire)),
         })
     }
 
@@ -93,6 +99,11 @@ impl Selector {
     pub fn deregister(&self, fd: RawFd) -> io::Result<()> {
         syscall!(epoll_ctl(self.ep, libc::EPOLL_CTL_DEL, fd, ptr::null_mut())).map(|_| ())
     }
+
+    #[cfg(debug_assertions)]
+    pub fn register_waker(&self) -> bool {
+        self.has_waker.swap(true, Ordering::AcqRel)
+    }
 }
 
 cfg_net! {
diff --git a/src/sys/unix/selector/kqueue.rs b/src/sys/unix/selector/kqueue.rs
index 2ebac9a..fc5b9ed 100644
--- a/src/sys/unix/selector/kqueue.rs
+++ b/src/sys/unix/selector/kqueue.rs
@@ -4,7 +4,7 @@ use std::mem::MaybeUninit;
 use std::ops::{Deref, DerefMut};
 use std::os::unix::io::{AsRawFd, RawFd};
 #[cfg(debug_assertions)]
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::time::Duration;
 use std::{cmp, io, ptr, slice};
 
@@ -69,6 +69,8 @@ pub struct Selector {
     #[cfg(debug_assertions)]
     id: usize,
     kq: RawFd,
+    #[cfg(debug_assertions)]
+    has_waker: AtomicBool,
 }
 
 impl Selector {
@@ -79,6 +81,8 @@ impl Selector {
                 #[cfg(debug_assertions)]
                 id: NEXT_ID.fetch_add(1, Ordering::Relaxed),
                 kq,
+                #[cfg(debug_assertions)]
+                has_waker: AtomicBool::new(false),
             })
     }
 
@@ -88,6 +92,8 @@ impl Selector {
             #[cfg(debug_assertions)]
             id: self.id,
             kq,
+            #[cfg(debug_assertions)]
+            has_waker: AtomicBool::new(self.has_waker.load(Ordering::Acquire)),
         })
     }
 
@@ -208,6 +214,11 @@ impl Selector {
         kevent_register(self.kq, &mut changes, &[libc::ENOENT as Data])
     }
 
+    #[cfg(debug_assertions)]
+    pub fn register_waker(&self) -> bool {
+        self.has_waker.swap(true, Ordering::AcqRel)
+    }
+
     // Used by `Waker`.
     #[cfg(any(target_os = "freebsd", target_os = "ios", target_os = "macos"))]
     pub fn setup_waker(&self, token: Token) -> io::Result<()> {
diff --git a/src/sys/unix/tcp.rs b/src/sys/unix/tcp.rs
index 9d59632..81b371b 100644
--- a/src/sys/unix/tcp.rs
+++ b/src/sys/unix/tcp.rs
@@ -1,6 +1,7 @@
 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_socket, socket_addr, to_socket_addr};
@@ -54,8 +55,21 @@ pub(crate) fn set_reuseaddr(socket: TcpSocket, reuseaddr: bool) -> io::Result<()
         libc::SO_REUSEADDR,
         &val as *const libc::c_int as *const libc::c_void,
         size_of::<libc::c_int>() as libc::socklen_t,
-    ))?;
-    Ok(())
+    )).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/selector.rs b/src/sys/windows/selector.rs
index 792a5c5..9715582 100644
--- a/src/sys/windows/selector.rs
+++ b/src/sys/windows/selector.rs
@@ -1,14 +1,18 @@
 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 std::collections::VecDeque;
+use std::io;
 use std::marker::PhantomPinned;
 use std::os::windows::io::RawSocket;
 use std::pin::Pin;
@@ -17,7 +21,6 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
-use std::io;
 use winapi::shared::ntdef::NT_SUCCESS;
 use winapi::shared::ntdef::{HANDLE, PVOID};
 use winapi::shared::ntstatus::STATUS_CANCELLED;
@@ -226,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,
@@ -332,8 +327,9 @@ static NEXT_ID: AtomicUsize = AtomicUsize::new(0);
 pub struct Selector {
     #[cfg(debug_assertions)]
     id: usize,
-
     pub(super) inner: Arc<SelectorInner>,
+    #[cfg(debug_assertions)]
+    has_waker: AtomicBool,
 }
 
 impl Selector {
@@ -345,6 +341,8 @@ impl Selector {
                 #[cfg(debug_assertions)]
                 id,
                 inner: Arc::new(inner),
+                #[cfg(debug_assertions)]
+                has_waker: AtomicBool::new(false),
             }
         })
     }
@@ -354,6 +352,8 @@ impl Selector {
             #[cfg(debug_assertions)]
             id: self.id,
             inner: Arc::clone(&self.inner),
+            #[cfg(debug_assertions)]
+            has_waker: AtomicBool::new(self.has_waker.load(Ordering::Acquire)),
         })
     }
 
@@ -365,6 +365,11 @@ impl Selector {
         self.inner.select(events, timeout)
     }
 
+    #[cfg(debug_assertions)]
+    pub fn register_waker(&self) -> bool {
+        self.has_waker.swap(true, Ordering::AcqRel)
+    }
+
     pub(super) fn clone_port(&self) -> Arc<CompletionPort> {
         self.inner.cp.clone()
     }
@@ -504,7 +509,7 @@ impl SelectorInner {
             } 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;
@@ -706,7 +711,7 @@ impl Drop for SelectorInner {
                             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
@@ -730,16 +735,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 c5b0410..46ac1ac 100644
--- a/src/sys/windows/tcp.rs
+++ b/src/sys/windows/tcp.rs
@@ -1,11 +1,15 @@
 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::{
-    self,
-    closesocket, PF_INET, PF_INET6, SOCKET, 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::{init, new_socket, socket_addr};
@@ -69,11 +73,6 @@ pub(crate) fn close(socket: TcpSocket) {
 }
 
 pub(crate) fn set_reuseaddr(socket: TcpSocket, reuseaddr: bool) -> io::Result<()> {
-    use winapi::ctypes::{c_char, c_int};
-    use winapi::shared::minwindef::{BOOL, TRUE, FALSE};
-    use winapi::um::winsock2::{setsockopt, SOCKET_ERROR, SO_REUSEADDR, SOL_SOCKET};
-    use std::mem::size_of;
-
     let val: BOOL = if reuseaddr { TRUE } else { FALSE };
 
     match unsafe { setsockopt(
@@ -83,9 +82,26 @@ pub(crate) fn set_reuseaddr(socket: TcpSocket, reuseaddr: bool) -> io::Result<()
         &val as *const _ as *const c_char,
         size_of::<BOOL>() as c_int,
     ) } {
-        0 => Ok(()),
         SOCKET_ERROR => Err(io::Error::last_os_error()),
-        _ => panic!("unexpected return value"),
+        _ => 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(()),
     }
 }
 
diff --git a/src/waker.rs b/src/waker.rs
index 44766ce..b8e4496 100644
--- a/src/waker.rs
+++ b/src/waker.rs
@@ -16,7 +16,7 @@ use std::io;
 /// `Waker` events are only guaranteed to be delivered while the `Waker` value
 /// is alive.
 ///
-/// Only a single `Waker` should active per [`Poll`], if multiple threads need
+/// Only a single `Waker` can be active per [`Poll`], if multiple threads need
 /// access to the `Waker` it can be shared via for example an `Arc`. What
 /// happens if multiple `Waker`s are registered with the same `Poll` is
 /// undefined.
@@ -81,6 +81,8 @@ pub struct Waker {
 impl Waker {
     /// Create a new `Waker`.
     pub fn new(registry: &Registry, token: Token) -> io::Result<Waker> {
+        #[cfg(debug_assertions)]
+        registry.register_waker();
         sys::Waker::new(poll::selector(&registry), token).map(|inner| Waker { inner })
     }
 


Test patch:
diff --git a/tests/poll.rs b/tests/poll.rs
index b14bf88..3ec4fea 100644
--- a/tests/poll.rs
+++ b/tests/poll.rs
@@ -401,7 +401,7 @@ fn reregister_interest_token_usage() {
 // On kqueue platforms registering twice (not *re*registering) works, but that
 // is not a test goal, so it is not tested.
 #[test]
-#[cfg(any(target_os = "linux", target_os = "windows"))]
+#[cfg(debug_assertions)] // Check is only present when debug assertions are enabled.
 pub fn double_register_different_token() {
     init();
     let poll = Poll::new().unwrap();
@@ -412,10 +412,11 @@ pub fn double_register_different_token() {
         .register(&mut listener, Token(0), Interest::READABLE)
         .unwrap();
 
-    assert!(poll
-        .registry()
-        .register(&mut listener, Token(1), Interest::READABLE)
-        .is_err());
+    assert_error(
+        poll.registry()
+            .register(&mut listener, Token(1), Interest::READABLE),
+        "already registered",
+    );
 }
 
 #[test]
@@ -481,6 +482,50 @@ fn poll_ok_after_cancelling_pending_ops() {
     handle.join().expect("unable to join thread");
 }
 
+// This test checks the following reregister constraint:
+// The event source must have previously been registered with this instance
+// of `Poll`, otherwise the behavior is undefined.
+//
+// This test is done on Windows and epoll platforms where reregistering a
+// source without a previous register is defined behavior that fail with an
+// error code.
+//
+// On kqueue platforms reregistering w/o registering works but that's not a
+// test goal, so it is not tested.
+#[test]
+#[cfg(debug_assertions)] // Check is only present when debug assertions are enabled.
+fn reregister_without_register() {
+    let poll = Poll::new().expect("unable to create Poll instance");
+
+    let mut listener = TcpListener::bind(any_local_address()).unwrap();
+
+    assert_error(
+        poll.registry()
+            .reregister(&mut listener, ID1, Interest::READABLE),
+        "not registered",
+    );
+}
+
+// This test checks the following register/deregister constraint:
+// The event source must have previously been registered with this instance
+// of `Poll`, otherwise the behavior is undefined.
+//
+// This test is done on Windows and epoll platforms where deregistering a
+// source without a previous register is defined behavior that fail with an
+// error code.
+//
+// On kqueue platforms deregistering w/o registering works but that's not a
+// test goal, so it is not tested.
+#[test]
+#[cfg(debug_assertions)] // Check is only present when debug assertions are enabled.
+fn deregister_without_register() {
+    let poll = Poll::new().expect("unable to create Poll instance");
+
+    let mut listener = TcpListener::bind(any_local_address()).unwrap();
+
+    assert_error(poll.registry().deregister(&mut listener), "not registered");
+}
+
 struct TestEventSource {
     registrations: Vec<(Token, Interest)>,
     reregistrations: Vec<(Token, Interest)>,
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_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/waker.rs b/tests/waker.rs
index c691e77..02678ff 100644
--- a/tests/waker.rs
+++ b/tests/waker.rs
@@ -106,6 +106,27 @@ fn waker_multiple_wakeups_different_thread() {
     handle2.join().unwrap();
 }
 
+#[test]
+#[cfg_attr(
+    not(debug_assertions),
+    ignore = "only works with debug_assertions enabled"
+)]
+#[should_panic = "Only a single `Waker` can be active per `Poll` instance"]
+fn using_multiple_wakers_panics() {
+    init();
+
+    let poll = Poll::new().expect("unable to create new Poll instance");
+    let token1 = Token(10);
+    let token2 = Token(11);
+
+    let waker1 = Waker::new(poll.registry(), token1).expect("unable to first waker");
+    // This should panic.
+    let waker2 = Waker::new(poll.registry(), token2).unwrap();
+
+    drop(waker1);
+    drop(waker2);
+}
+
 fn expect_waker_event(poll: &mut Poll, events: &mut Events, token: Token) {
     poll.poll(events, Some(Duration::from_millis(100))).unwrap();
     assert!(!events.is_empty());