Don't report RDHUP as HUP (#939)

Currently, EPOLLRDHUP sets UnixReady::hup(). This is incorrect behavior
because higher-level libraries like tokio (correctly) assume that
UnixReady::hup() is unclearable since it signals that both the read and
write halfs are shutdown. In reality, EPOLLRDHUP only means that the TCP
stream has been half-closed and such a half-closed stream can still be
written to.

This will fix a current issue with tokio, which is that tokio infinitely
attempts to write to a half-closed socket that's returning WouldBlock
when it's write buffer is full, an issue which manifests with excessive
CPU usage. I think this may help some of the issues discussed in
tokio-rs/tokio#449

After this change, EOF will still be propagated correctly, because
read-hangups also trigger read-readiness via EPOLLIN. However, if
handling of EPOLLRDHUP is desired to be retained, I believe it should be
implemented as another readiness kind on UnixReady, perhaps
UnixReady::read_hup().

Possible concern of a breaking change: Since it's not currently possible
for a user of mio to differentiate between EPOLLHUP and EPOLLRDHUP, it
must be that no users of mio currently are. There _may_ be applications
that test the "health" of a socket by checking for UnixRead::hup(),
which would previously trigger on EPOLLRDHUP but will no longer with
this change. This will change such applications from considering a
half-closed connection as closed to considering it open. However, I
still beleive this change is a correction of the semantics of HUP and
the desired behavior such applications was already ambiguous.

This also fixes a similar issue with kqueue.

azure-pipelines.yml

@@ -1,5 +1,5 @@
-trigger: ["master"]
-pr: ["master"]
+trigger: ["master", "v0.6.x"]
+pr: ["master", "v0.6.x"]
 
 jobs:
 # Check formatting

src/sys/unix/epoll.rs

@@ -6,7 +6,7 @@ use std::time::Duration;
 use std::{cmp, i32};
 
 use libc::{self, c_int};
-use libc::{EPOLLERR, EPOLLHUP, EPOLLRDHUP, EPOLLONESHOT};
+use libc::{EPOLLERR, EPOLLHUP, EPOLLONESHOT};
 use libc::{EPOLLET, EPOLLOUT, EPOLLIN, EPOLLPRI};
 
 use {io, Ready, PollOpt, Token};
@@ -141,11 +141,6 @@ fn ioevent_to_epoll(interest: Ready, opts: PollOpt) -> u32 {
         kind |= EPOLLOUT;
     }
 
-    if UnixReady::from(interest).is_hup() {
-        kind |= EPOLLRDHUP;
-    }
-
-
     if UnixReady::from(interest).is_priority() {
         kind |= EPOLLPRI;
     }
@@ -228,7 +223,7 @@ impl Events {
                 kind = kind | UnixReady::error();
             }
 
-            if (epoll & EPOLLRDHUP) != 0 || (epoll & EPOLLHUP) != 0 {
+            if (epoll & EPOLLHUP) != 0 {
                 kind = kind | UnixReady::hup();
             }
 

src/sys/unix/kqueue.rs

@@ -291,6 +291,18 @@ impl Events {
                 event::kind_mut(&mut self.events[idx]).insert(Ready::readable());
             } else if e.filter == libc::EVFILT_WRITE as Filter {
                 event::kind_mut(&mut self.events[idx]).insert(Ready::writable());
+
+                // `EV_EOF` set with `EVFILT_WRITE` indicates the connection has been fully
+                // disconnected (read and write), but only sockets...
+                if e.flags & libc::EV_EOF != 0 {
+                    event::kind_mut(&mut self.events[idx]).insert(UnixReady::hup());
+
+                    // When the read end of the socket is closed, EV_EOF is set on
+                    // flags, and fflags contains the error if there is one.
+                    if e.fflags != 0 {
+                        event::kind_mut(&mut self.events[idx]).insert(UnixReady::error());
+                    }
+                }
             }
 #[cfg(any(target_os = "dragonfly",
     target_os = "freebsd", target_os = "ios", target_os = "macos"))]
@@ -305,16 +317,6 @@ impl Events {
                     event::kind_mut(&mut self.events[idx]).insert(UnixReady::lio());
                 }
             }
-
-            if e.flags & libc::EV_EOF != 0 {
-                event::kind_mut(&mut self.events[idx]).insert(UnixReady::hup());
-
-                // When the read end of the socket is closed, EV_EOF is set on
-                // flags, and fflags contains the error if there is one.
-                if e.fflags != 0 {
-                    event::kind_mut(&mut self.events[idx]).insert(UnixReady::error());
-                }
-            }
         }
 
         ret

src/sys/unix/ready.rs

@@ -198,7 +198,8 @@ impl UnixReady {
     /// **Note that only readable and writable readiness is guaranteed to be
     /// supported on all platforms**. This means that `hup` readiness
     /// should be treated as a hint. For more details, see [readiness] in the
-    /// poll documentation.
+    /// poll documentation. It is also unclear if HUP readiness will remain in 0.7. See
+    /// [here][issue-941].
     ///
     /// See [`Poll`] for more documentation on polling.
     ///
@@ -214,6 +215,7 @@ impl UnixReady {
     ///
     /// [`Poll`]: ../struct.Poll.html
     /// [readiness]: ../struct.Poll.html#readiness-operations
+    /// [issue-941]: https://github.com/tokio-rs/mio/issues/941
     #[inline]
     pub fn hup() -> UnixReady {
         UnixReady(ready_from_usize(HUP))

test/mod.rs

@@ -30,6 +30,7 @@ mod test_reregister_without_poll;
 mod test_smoke;
 mod test_tcp;
 mod test_tcp_level;
+mod test_tcp_shutdown;
 mod test_udp_level;
 mod test_udp_socket;
 mod test_write_then_drop;

test/test_tcp_shutdown.rs

@@ -0,0 +1,73 @@
+use std::net::Shutdown;
+use std::time::Duration;
+
+use mio::{Token, Ready, PollOpt, Poll, Events};
+use mio::net::TcpStream;
+
+macro_rules! wait {
+    ($poll:ident, $ready:ident) => {{
+        use std::time::Instant;
+
+        let now = Instant::now();
+        let mut events = Events::with_capacity(16);
+        let mut found = false;
+
+        while !found {
+            if now.elapsed() > Duration::from_secs(5) {
+                panic!("not ready");
+            }
+
+            $poll.poll(&mut events, Some(Duration::from_secs(1))).unwrap();
+
+            for event in &events {
+                #[cfg(unix)]
+                {
+                    use mio::unix::UnixReady;
+                    assert!(!UnixReady::from(event.readiness()).is_hup());
+                }
+
+                if event.token() == Token(0) && event.readiness().$ready() {
+                    found = true;
+                    break;
+                }
+            }
+        }
+    }};
+}
+
+#[test]
+fn test_write_shutdown() {
+    let poll = Poll::new().unwrap();
+
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let mut ready = Ready::readable() | Ready::writable();
+
+    #[cfg(unix)]
+    {
+        ready |= mio::unix::UnixReady::hup();
+    }
+
+    let client = TcpStream::connect(&addr).unwrap();
+    poll.register(&client,
+                  Token(0),
+                  ready,
+                  PollOpt::edge()).unwrap();
+
+    let (socket, _) = listener.accept().unwrap();
+
+    wait!(poll, is_writable);
+
+    let mut events = Events::with_capacity(16);
+
+    // Polling should not have any events
+    poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
+    assert!(events.iter().next().is_none());
+
+    println!("SHUTTING DOWN");
+    // Now, shutdown the write half of the socket.
+    socket.shutdown(Shutdown::Write).unwrap();
+
+    wait!(poll, is_readable);
+}