Repository: tokio-rs/mio
Pull request #1110 (2019-10-25 04:15:38)

Add read_close and write_close readiness events
## Motivation

The existing `is_hup` method on `Events` was difficult to add additional checks
to that were both cross-platform and guaranteed no false positives.

For example, if the write half of a socket is closed in certain scenarios, this
causes a set of expected events to be triggered. With `is_read_hup` and
`is_hup`, there was no way to check for for this set of events that did not also
result in false positives on other platforms.

## Solution

To solve this, `is_read_closed` and `is_write_closed` are being introduced as
cross-platform methods on `Events` that are best effort implementations--meaning
false negatives are possible, but false positives will not occur.

Finding the expected event behavior for each of these platforms has not been
very easy, so below is best-effort summary of what can be expected. #1091
provides a lot of the visibility into where this data came from, but a lot of
the scenarios checked for change when they occur independently.

### epoll

A helpful thing to keep in mind is that `EPOLLHUP` is a mask for a socket's
`RCV_SHUTDOWN` and `SEND_SHUTDOWN` flags. Once both of these flags are set,
`EPOLLHUP` will trigger.

#### is_read_closed

`EPOLLRDHUP` is now a registered event when there is interest in
`Interest::READABLE`.

`EPOLLRDHUP` occurs after *either* of the following events that sets the
`RCV_SHTUDOWN` flag:
  1. The local socket has called `shutdown(Shutdown::Read)`
  2. The peer stream has called `shutdown(Shutdown::Write)` which sends a `FIN`
     packet; that packet has been acknowledged by the local socket.

`EPOLLHUP` occurs after *both* of the following:
  1. `EPOLLRDHUP` by the previous point
  2. The local socket has called `shutdown(Shutdown::Write)` sending a `FIN` to
     the peer and marking setting its own `SEND_SHUTDOWN` flag.

`is_read_closed` checks for *either* of the following:
  1. `EPOLLHUP`
  2. `EPOLLIN` and `EPOLLRDHUP`

#### is_write_closed

`is_write_closed` is a hard event to check for with `epoll` without returning
false positives. The only special-case set of events that can be checked is both
`EPOLLOUT` and `EPOLLERR` which occurs when a Unix pipe sender closes.

Therefore, `is_write_closed` checks for *either* of the following:
  1. `EPOLLHUP` by the previous point
  2. `EPOLLOUT` and `EPOLLERR`

#### Things to note (or at least I found interesting)

A peer calling `shutdown(Shutdown::Both)` does not result in `EPOLLHUP` on the
local socket. It results in a `EPOLLIN` | `EPOLLOUT` | `EPOLLRDHUP` because the
peer has sent a `FIN` packet that has been acknowledged. Keep in mind however,
the local socket has not *also* sent a `FIN`, therefore it's `SEND_SHUTDOWN`
flag is not set.

Except for Unix pipes, a local socket calling `shutdown(Shutdown::Write)` only
results in a `EPOLLOUT`, which is not very helpful! Because there is no
`EPOLLRDHUP` equivalent event for when the `SEND_SHUTDOWN` flag is set, it is
hard to say whether or not the write half is _actually_ closed.

### kqueue

`kqueue` is has fairly predictable event behavior and therefore read and write
closed events are easier check. Note that some of the event checks in the tests
will only compile for `kqueue` platforms because `is_write_closed` will not
return false negatives.

#### is_read_closed

`is_read_closed` checks for the `EVFILT_READ` filter *and* `EV_EOF` flag

#### is_write_closed

`is_write_closed` checks for the `EVFILT_WRITE` filter *and* `EV_EOF` flag.

#### Things to note (or at least I found interesting)

A peer calling `shutdown(Shutdown::Both)` has the same result as it does in
`epoll`. `is_read_closed` will return `true`, but nothing has happend on the
local socket to indicate the write half will error.

### IOCP

`IOCP` is the least tested selector in this change, and therefore has few
situations where `is_read_closed` and `is_write_closed` do not return false
negatives. It checks for the equivalent behavior as before, mainly that
`is_write_closed` is equivalent to the current `is_hup`.

`is_read_closed` checks for the `AFD_POLL_DISCONNECT` flag.

`is_write_closed` checks for the `AFD_POLL_ABORT` flag.

Signed-off-by: Kevin Leimkuhler <kleimkuhler@icloud.com>


kleimkuhler: @Thomasdezeeuw @carllerche This is ready for review now. I updated the PR comment to include a summary of what these methods check for. Most of this was through the data that #1091 exposes, but these events are fairly independent of each other.
Thomasdezeeuw: If you can rebase on master and fix the small issue with the panic message we can merge this.
kleimkuhler: @Thomasdezeeuw I merged master instead of rebasing as to not reset the diff. It should be good to go
Thomasdezeeuw: @kleimkuhler GitHub says "This branch cannot be rebased due to conflicts". Also we don't usually squash commits but rebased them on the master branch.

Can you locally run:
```bash
git fetch upstream # assuming upstream is tokio-rs/mio
git rebase upstream/master
git push
```

Patch:
diff --git a/src/event/event.rs b/src/event/event.rs
index f5e226a..9b42a90 100644
--- a/src/event/event.rs
+++ b/src/event/event.rs
@@ -62,59 +62,65 @@ impl Event {
         sys::event::is_error(&self.inner)
     }
 
-    /// Returns true if the event contains HUP readiness.
+    /// Returns true if the event contains read closed readiness.
     ///
     /// # Notes
     ///
-    /// Method is available on all platforms, but not all platforms trigger the
-    /// HUP event.
+    /// Read closed readiness can be expected after any of the following have
+    /// occurred:
+    /// * The local stream has shutdown the read half of its socket
+    /// * The local stream has shtudown both the read half and the write half
+    ///   of its socket
+    /// * The peer stream has shtudown the write half its socket; this sends a
+    ///   `FIN` packet that has been received by the local stream
     ///
-    /// Because of the above be cautions when using this in cross-platform
-    /// applications, Mio makes no attempt at normalising this indicator and
-    /// only provides a convenience method to read it. Refer to the selector
-    /// documentation (below) when using this indicator.
+    /// Method is a best effort implementation. While some platforms may not
+    /// return readiness when read half is closed, it is guaranteed that
+    /// false-positives will not occur.
     ///
     /// The table below shows what flags are checked on what OS.
     ///
     /// | [OS selector] | Flag(s) checked |
     /// |---------------|-----------------|
-    /// | [epoll]       | `EPOLLHUP`      |
+    /// | [epoll]       | `EPOLLHUP`, or  |
+    /// |               | `EPOLLIN` and EPOLLRDHUP` |
     /// | [kqueue]      | `EV_EOF`        |
     ///
     /// [OS selector]: ../struct.Poll.html#implementation-notes
     /// [epoll]: http://man7.org/linux/man-pages/man7/epoll.7.html
     /// [kqueue]: https://www.freebsd.org/cgi/man.cgi?query=kqueue&sektion=2
     #[inline]
-    pub fn is_hup(&self) -> bool {
-        sys::event::is_hup(&self.inner)
+    pub fn is_read_closed(&self) -> bool {
+        sys::event::is_read_closed(&self.inner)
     }
 
-    /// Returns true if the event contains read HUP readiness.
+    /// Returns true if the event contains write closed readiness.
     ///
     /// # Notes
     ///
-    /// Method is available on all platforms, but not all platforms trigger the
-    /// read HUP event.
+    /// On [epoll] this is essentially a check for `EPOLLHUP` flag as the
+    /// local stream shutting down its write half does not trigger this event.
     ///
-    /// Because of the above be cautions when using this in cross-platform
-    /// applications, Mio makes no attempt at normalising this indicator and
-    /// only provides a convenience method to read it. We advice looking at the
-    /// documentation provided for the selectors (see below) when using this
-    /// indicator.
+    /// On [kqueue] the local stream shutting down the write half of its
+    /// socket will trigger this event.
+    ///
+    /// Method is a best effort implementation. While some platforms may not
+    /// return readiness when write half is closed, it is guaranteed that
+    /// false-positives will not occur.
     ///
     /// The table below shows what flags are checked on what OS.
     ///
     /// | [OS selector] | Flag(s) checked |
     /// |---------------|-----------------|
-    /// | [epoll]       | `EPOLLRDHUP`    |
+    /// | [epoll]       | `EPOLLHUP`, or  |
+    /// |               | `EPOLLOUT` and `EPOLLERR` |
     /// | [kqueue]      | `EV_EOF`        |
     ///
     /// [OS selector]: ../struct.Poll.html#implementation-notes
     /// [epoll]: http://man7.org/linux/man-pages/man7/epoll.7.html
     /// [kqueue]: https://www.freebsd.org/cgi/man.cgi?query=kqueue&sektion=2
-    #[inline]
-    pub fn is_read_hup(&self) -> bool {
-        sys::event::is_read_hup(&self.inner)
+    pub fn is_write_closed(&self) -> bool {
+        sys::event::is_write_closed(&self.inner)
     }
 
     /// Returns true if the event contains priority readiness.
@@ -190,8 +196,8 @@ impl fmt::Debug for Event {
             .field("readable", &self.is_readable())
             .field("writable", &self.is_writable())
             .field("error", &self.is_error())
-            .field("hup", &self.is_hup())
-            .field("read_hup", &self.is_read_hup())
+            .field("read_closed", &self.is_read_closed())
+            .field("write_closed", &self.is_write_closed())
             .field("priority", &self.is_priority())
             .field("aio", &self.is_aio())
             .field("lio", &self.is_lio())
diff --git a/src/poll.rs b/src/poll.rs
index 52e20e6..9495c82 100644
--- a/src/poll.rs
+++ b/src/poll.rs
@@ -102,23 +102,24 @@ use std::{fmt, io};
 /// The only readiness operations that are guaranteed to be present on all
 /// supported platforms are [`readable`] and [`writable`]. All other readiness
 /// operations may have false negatives and as such should be considered
-/// **hints**. This means that if a socket is registered with [`readable`],
-/// [`error`], and [`hup`] interest, and either an error or hup is received, a
-/// readiness event will be generated for the socket, but it **may** only
-/// include `readable` readiness. Also note that, given the potential for
-/// spurious events, receiving a readiness event with `hup` or `error` doesn't
-/// actually mean that a `read` on the socket will return a result matching the
-/// readiness event.
-///
-/// In other words, portable programs that explicitly check for [`hup`] or
-/// [`error`] readiness should be doing so as an **optimization** and always be
-/// able to handle an error or HUP situation when performing the actual read
-/// operation.
+/// **hints**. This means that if a socket is registered with [`readable`]
+/// interest and either an error or close is received, a readiness event will
+/// be generated for the socket, but it **may** only include `readable`
+/// readiness. Also note that, given the potential for spurious events,
+/// receiving a readiness event with `read_closed`, `write_closed`, or `error`
+/// doesn't actually mean that a `read` on the socket will return a result
+/// matching the readiness event.
+///
+/// In other words, portable programs that explicitly check for [`read_closed`],
+/// [`write_closed`], or [`error`] readiness should be doing so as an
+/// **optimization** and always be able to handle an error or close situation
+/// when performing the actual read operation.
 ///
 /// [`readable`]: crate::event::Event::is_readable
 /// [`writable`]: crate::event::Event::is_writable
 /// [`error`]: crate::event::Event::is_error
-/// [`hup`]: crate::event::Event::is_hup
+/// [`read_closed`]: crate::event::Event::is_read_closed
+/// [`write_closed`]: crate::event::Event::is_write_closed
 ///
 /// ### Registering handles
 ///
diff --git a/src/sys/unix/epoll.rs b/src/sys/unix/epoll.rs
index ba695ff..20bfe7d 100644
--- a/src/sys/unix/epoll.rs
+++ b/src/sys/unix/epoll.rs
@@ -1,7 +1,7 @@
 use crate::sys::Events;
 use crate::{Interests, Token};
 
-use libc::{EPOLLET, EPOLLIN, EPOLLOUT};
+use libc::{EPOLLET, EPOLLIN, EPOLLOUT, EPOLLRDHUP};
 use log::error;
 use std::os::unix::io::{AsRawFd, RawFd};
 #[cfg(debug_assertions)]
@@ -92,7 +92,7 @@ fn interests_to_epoll(interests: Interests) -> u32 {
     let mut kind = EPOLLET;
 
     if interests.is_readable() {
-        kind |= EPOLLIN;
+        kind = kind | EPOLLIN | EPOLLRDHUP;
     }
 
     if interests.is_writable() {
@@ -139,12 +139,20 @@ pub mod event {
         (event.events as libc::c_int & libc::EPOLLERR) != 0
     }
 
-    pub fn is_hup(event: &Event) -> bool {
-        (event.events as libc::c_int & libc::EPOLLHUP) != 0
+    pub fn is_read_closed(event: &Event) -> bool {
+        // Both halves of the socket have closed
+        event.events as libc::c_int & libc::EPOLLHUP != 0
+            // Socket has received FIN or called shutdown(SHUT_RD)
+            || (event.events as libc::c_int & libc::EPOLLIN != 0
+                && event.events as libc::c_int & libc::EPOLLRDHUP != 0)
     }
 
-    pub fn is_read_hup(event: &Event) -> bool {
-        (event.events as libc::c_int & libc::EPOLLRDHUP) != 0
+    pub fn is_write_closed(event: &Event) -> bool {
+        // Both halves of the socket have closed
+        event.events as libc::c_int & libc::EPOLLHUP != 0
+            // Unix pipe write end has closed
+            || (event.events as libc::c_int & libc::EPOLLOUT != 0
+                && event.events as libc::c_int & libc::EPOLLERR != 0)
     }
 
     pub fn is_priority(event: &Event) -> bool {
diff --git a/src/sys/unix/kqueue.rs b/src/sys/unix/kqueue.rs
index 41753a6..a870ff3 100644
--- a/src/sys/unix/kqueue.rs
+++ b/src/sys/unix/kqueue.rs
@@ -341,12 +341,12 @@ pub mod event {
             (event.flags & libc::EV_EOF) != 0 && event.fflags != 0
     }
 
-    pub fn is_hup(event: &Event) -> bool {
-        event.filter == libc::EVFILT_WRITE && (event.flags & libc::EV_EOF) != 0
+    pub fn is_read_closed(event: &Event) -> bool {
+        event.filter == libc::EVFILT_READ && event.flags & libc::EV_EOF != 0
     }
 
-    pub fn is_read_hup(event: &Event) -> bool {
-        event.filter == libc::EVFILT_READ && (event.flags & libc::EV_EOF) != 0
+    pub fn is_write_closed(event: &Event) -> bool {
+        event.filter == libc::EVFILT_WRITE && event.flags & libc::EV_EOF != 0
     }
 
     pub fn is_priority(_: &Event) -> bool {
diff --git a/src/sys/windows/event.rs b/src/sys/windows/event.rs
index b75e351..0477699 100644
--- a/src/sys/windows/event.rs
+++ b/src/sys/windows/event.rs
@@ -14,7 +14,7 @@ pub fn token(event: &Event) -> Token {
 }
 
 pub fn is_readable(event: &Event) -> bool {
-    if is_error(event) || is_read_hup(event) {
+    if is_error(event) || is_read_closed(event) {
         return true;
     }
     event.flags & (afd::AFD_POLL_RECEIVE | afd::AFD_POLL_ACCEPT) != 0
@@ -31,12 +31,12 @@ pub fn is_error(event: &Event) -> bool {
     event.flags & afd::AFD_POLL_CONNECT_FAIL != 0
 }
 
-pub fn is_hup(event: &Event) -> bool {
-    event.flags & afd::AFD_POLL_ABORT != 0
+pub fn is_read_closed(event: &Event) -> bool {
+    event.flags & afd::AFD_POLL_DISCONNECT != 0
 }
 
-pub fn is_read_hup(event: &Event) -> bool {
-    event.flags & afd::AFD_POLL_DISCONNECT != 0
+pub fn is_write_closed(event: &Event) -> bool {
+    event.flags & afd::AFD_POLL_ABORT != 0
 }
 
 pub fn is_priority(event: &Event) -> bool {


Test patch:
diff --git a/tests/tcp.rs b/tests/tcp.rs
index faa3b4a..b33710f 100644
--- a/tests/tcp.rs
+++ b/tests/tcp.rs
@@ -622,7 +622,7 @@ fn write_error() {
 }
 
 macro_rules! wait {
-    ($poll:ident, $ready:ident, $expect_hup: expr) => {{
+    ($poll:ident, $ready:ident, $expect_read_closed: expr) => {{
         use std::time::Instant;
 
         let now = Instant::now();
@@ -639,23 +639,10 @@ macro_rules! wait {
                 .unwrap();
 
             for event in &events {
-                // Hup is only generated on kqueue platforms.
-                #[cfg(any(
-                    target_os = "dragonfly",
-                    target_os = "freebsd",
-                    target_os = "ios",
-                    target_os = "macos",
-                    target_os = "netbsd",
-                    target_os = "openbsd"
-                ))]
-                {
-                    if $expect_hup {
-                        assert!(event.is_read_hup());
-                    }
-                }
-
-                if !$expect_hup {
-                    assert!(!event.is_hup());
+                if $expect_read_closed {
+                    assert!(event.is_read_closed());
+                } else {
+                    assert!(!event.is_read_closed() && !event.is_write_closed());
                 }
 
                 if event.token() == Token(0) && event.$ready() {
@@ -676,11 +663,13 @@ fn write_shutdown() {
     let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
     let addr = listener.local_addr().unwrap();
 
-    let interests = Interests::READABLE | Interests::WRITABLE;
-
     let client = TcpStream::connect(addr).unwrap();
     poll.registry()
-        .register(&client, Token(0), interests)
+        .register(
+            &client,
+            Token(0),
+            Interests::READABLE.add(Interests::WRITABLE),
+        )
         .unwrap();
 
     let (socket, _) = listener.accept().unwrap();
diff --git a/tests/tcp_stream.rs b/tests/tcp_stream.rs
index f583fd2..a9f15d1 100644
--- a/tests/tcp_stream.rs
+++ b/tests/tcp_stream.rs
@@ -1,14 +1,16 @@
+use log::warn;
 use std::io::{self, IoSlice, IoSliceMut, Read, Write};
 use std::net::{self, Shutdown, SocketAddr};
 #[cfg(unix)]
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd};
-use std::sync::mpsc::channel;
-use std::sync::{Arc, Barrier};
+use std::sync::{mpsc::channel, Arc, Barrier};
 use std::thread;
+use std::time::Duration;
 
 use mio::net::TcpStream;
 use mio::{Interests, Token};
 
+#[macro_use]
 mod util;
 
 use util::{
@@ -169,7 +171,7 @@ fn ttl() {
     init();
 
     let barrier = Arc::new(Barrier::new(2));
-    let (thread_handle, address) = start_listener(1, Some(barrier.clone()));
+    let (thread_handle, address) = start_listener(1, Some(barrier.clone()), false);
 
     let stream = TcpStream::connect(address).unwrap();
 
@@ -187,7 +189,7 @@ fn nodelay() {
     let (mut poll, mut events) = init_with_poll();
 
     let barrier = Arc::new(Barrier::new(2));
-    let (thread_handle, address) = start_listener(1, Some(barrier.clone()));
+    let (thread_handle, address) = start_listener(1, Some(barrier.clone()), false);
 
     let stream = TcpStream::connect(address).unwrap();
 
@@ -363,7 +365,7 @@ fn shutdown_both() {
 fn raw_fd() {
     init();
 
-    let (thread_handle, address) = start_listener(1, None);
+    let (thread_handle, address) = start_listener(1, None, false);
 
     let stream = TcpStream::connect(address).unwrap();
     let address = stream.local_addr().unwrap();
@@ -452,6 +454,124 @@ fn deregistering() {
     thread_handle.join().expect("unable to join thread");
 }
 
+#[test]
+#[cfg_attr(
+    windows,
+    ignore = "fails on Windows; client read closed events are not triggered"
+)]
+fn tcp_shutdown_client_read_close_event() {
+    let (mut poll, mut events) = init_with_poll();
+    let barrier = Arc::new(Barrier::new(2));
+
+    let (handle, sockaddr) = start_listener(1, Some(barrier.clone()), false);
+    let stream = assert_ok!(TcpStream::connect(sockaddr));
+
+    let interests = Interests::READABLE | Interests::WRITABLE;
+
+    assert_ok!(poll.registry().register(&stream, ID1, interests));
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(ID1, Interests::WRITABLE)],
+    );
+
+    assert_ok!(stream.shutdown(Shutdown::Read));
+    expect_readiness!(poll, events, is_read_closed);
+
+    barrier.wait();
+    handle.join().expect("failed to join thread");
+}
+
+#[test]
+#[cfg_attr(windows, ignore = "fails; client write_closed events are not found")]
+#[cfg_attr(
+    any(target_os = "linux", target_os = "android", target_os = "solaris"),
+    ignore = "fails; client write_closed events are not found"
+)]
+fn tcp_shutdown_client_write_close_event() {
+    let (mut poll, mut events) = init_with_poll();
+    let barrier = Arc::new(Barrier::new(2));
+
+    let (handle, sockaddr) = start_listener(1, Some(barrier.clone()), false);
+    let stream = assert_ok!(TcpStream::connect(sockaddr));
+
+    let interests = Interests::READABLE | Interests::WRITABLE;
+
+    assert_ok!(poll.registry().register(&stream, ID1, interests));
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(ID1, Interests::WRITABLE)],
+    );
+
+    assert_ok!(stream.shutdown(Shutdown::Write));
+    expect_readiness!(poll, events, is_write_closed);
+
+    barrier.wait();
+    handle.join().expect("failed to join thread");
+}
+
+#[test]
+fn tcp_shutdown_server_write_close_event() {
+    let (mut poll, mut events) = init_with_poll();
+    let barrier = Arc::new(Barrier::new(2));
+
+    let (handle, sockaddr) = start_listener(1, Some(barrier.clone()), true);
+    let stream = assert_ok!(TcpStream::connect(sockaddr));
+
+    assert_ok!(poll.registry().register(
+        &stream,
+        ID1,
+        Interests::READABLE.add(Interests::WRITABLE)
+    ));
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(ID1, Interests::WRITABLE)],
+    );
+
+    barrier.wait();
+
+    expect_readiness!(poll, events, is_read_closed);
+
+    barrier.wait();
+    handle.join().expect("failed to join thread");
+}
+
+#[test]
+#[cfg_attr(
+    windows,
+    ignore = "fails on Windows; client close events are not found"
+)]
+fn tcp_shutdown_client_both_close_event() {
+    let (mut poll, mut events) = init_with_poll();
+    let barrier = Arc::new(Barrier::new(2));
+
+    let (handle, sockaddr) = start_listener(1, Some(barrier.clone()), false);
+    let stream = assert_ok!(TcpStream::connect(sockaddr));
+
+    assert_ok!(poll.registry().register(
+        &stream,
+        ID1,
+        Interests::READABLE.add(Interests::WRITABLE)
+    ));
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(ID1, Interests::WRITABLE)],
+    );
+
+    assert_ok!(stream.shutdown(Shutdown::Both));
+    expect_readiness!(poll, events, is_write_closed);
+
+    barrier.wait();
+    handle.join().expect("failed to join thread");
+}
+
 /// Start a listener that accepts `n_connections` connections on the returned
 /// address. It echos back any data it reads from the connection before
 /// accepting another one.
@@ -494,6 +614,7 @@ fn echo_listener(addr: SocketAddr, n_connections: usize) -> (thread::JoinHandle<
 fn start_listener(
     n_connections: usize,
     barrier: Option<Arc<Barrier>>,
+    shutdown_write: bool,
 ) -> (thread::JoinHandle<()>, SocketAddr) {
     let (sender, receiver) = channel();
     let thread_handle = thread::spawn(move || {
@@ -505,6 +626,11 @@ fn start_listener(
             let (stream, _) = listener.accept().unwrap();
             if let Some(ref barrier) = barrier {
                 barrier.wait();
+
+                if shutdown_write {
+                    assert_ok!(stream.shutdown(Shutdown::Write));
+                    barrier.wait();
+                }
             }
             drop(stream);
         }
diff --git a/tests/uds.rs b/tests/uds.rs
index 0323228..07a1624 100644
--- a/tests/uds.rs
+++ b/tests/uds.rs
@@ -2,13 +2,17 @@
 #[macro_use]
 mod util;
 
+use log::warn;
 use mio::net::{UnixDatagram, UnixListener, UnixStream};
 use mio::unix::SocketAddr;
 use mio::{Interests, Token};
 use std::io::{self, IoSlice, IoSliceMut, Read, Write};
 use std::net::Shutdown;
+use std::os::unix::net;
 use std::sync::mpsc::{channel, Receiver};
+use std::sync::{Arc, Barrier};
 use std::thread;
+use std::time::Duration;
 use tempdir::TempDir;
 use util::{
     assert_send, assert_sync, assert_would_block, expect_events, expect_no_events, init_with_poll,
@@ -280,7 +284,7 @@ fn shutdown_read() {
     assert_ok!(poll.registry().register(
         &local,
         LOCAL,
-        Interests::WRITABLE.add(Interests::READABLE)
+        Interests::READABLE.add(Interests::WRITABLE)
     ));
 
     expect_events(
@@ -299,6 +303,7 @@ fn shutdown_read() {
     );
 
     assert_ok!(local.shutdown(Shutdown::Read));
+    expect_readiness!(poll, events, is_read_closed);
 
     // Shutting down the reading side is different on each platform. For example
     // on Linux based systems we can still read.
@@ -355,6 +360,16 @@ fn shutdown_write() {
 
     assert_ok!(local.shutdown(Shutdown::Write));
 
+    #[cfg(any(
+        target_os = "dragonfly",
+        target_os = "freebsd",
+        target_os = "ios",
+        target_os = "macos",
+        target_os = "netbsd",
+        target_os = "openbsd"
+    ))]
+    expect_readiness!(poll, events, is_write_closed);
+
     let err = assert_err!(local.write(DATA2));
     assert_eq!(err.kind(), io::ErrorKind::BrokenPipe);
 
@@ -402,6 +417,7 @@ fn shutdown_both() {
     );
 
     assert_ok!(local.shutdown(Shutdown::Both));
+    expect_readiness!(poll, events, is_write_closed);
 
     // Shutting down the reading side is different on each platform. For example
     // on Linux based systems we can still read.
@@ -430,6 +446,36 @@ fn shutdown_both() {
     assert_ok!(handle.join());
 }
 
+#[test]
+fn uds_shutdown_listener_write() {
+    let barrier = Arc::new(Barrier::new(2));
+    let (mut poll, mut events) = init_with_poll();
+
+    let (handle, remote_addr) = noop_listener(1, barrier.clone());
+
+    let path = remote_addr.as_pathname().expect("not a pathname");
+    let stream = assert_ok!(UnixStream::connect(path));
+
+    assert_ok!(poll.registry().register(
+        &stream,
+        LOCAL,
+        Interests::READABLE.add(Interests::WRITABLE)
+    ));
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(LOCAL, Interests::WRITABLE)],
+    );
+
+    barrier.wait();
+
+    expect_readiness!(poll, events, is_read_closed);
+
+    barrier.wait();
+    handle.join().expect("failed to join thread");
+}
+
 fn echo_remote(
     connections: usize,
     sync_receiver: Receiver<()>,
@@ -476,3 +522,26 @@ fn echo_remote(
     });
     (handle, assert_ok!(addr_receiver.recv()))
 }
+
+fn noop_listener(
+    connections: usize,
+    barrier: Arc<Barrier>,
+) -> (thread::JoinHandle<()>, net::SocketAddr) {
+    let (sender, receiver) = channel();
+    let handle = thread::spawn(move || {
+        let dir = assert_ok!(TempDir::new("uds"));
+        let path = dir.path().join("foo");
+        let listener = assert_ok!(net::UnixListener::bind(path.clone()));
+        let local_address = assert_ok!(listener.local_addr());
+        assert_ok!(sender.send(local_address));
+
+        for _ in 0..connections {
+            let (stream, _) = listener.accept().unwrap();
+            barrier.wait();
+            assert_ok!(stream.shutdown(Shutdown::Write));
+            barrier.wait();
+            drop(stream);
+        }
+    });
+    (handle, assert_ok!(receiver.recv()))
+}
diff --git a/tests/util/mod.rs b/tests/util/mod.rs
index b3a1f5d..395750c 100644
--- a/tests/util/mod.rs
+++ b/tests/util/mod.rs
@@ -56,6 +56,31 @@ macro_rules! assert_err {
     }};
 }
 
+/// Expect specific readiness on an event.
+#[macro_export]
+macro_rules! expect_readiness {
+    ($poll:ident, $events:ident, $readiness:ident) => {
+        let mut found = false;
+        // Poll a couple of times in case the event does not immediately
+        // happen
+        'outer: for _ in 0..3 {
+            assert_ok!($poll.poll(&mut $events, Some(Duration::from_millis(500))));
+            for event in $events.iter() {
+                if event.$readiness() {
+                    found = true;
+                    break 'outer;
+                } else {
+                    // Accept sporadic events.
+                    warn!("got unexpected event: {:?}", event);
+                }
+            }
+        }
+        if !found {
+            panic!("failed to find event readiness")
+        }
+    };
+}
+
 pub fn init() {
     static INIT: Once = Once::new();