process: Wake up read and write on EPOLLERR (#2218)

## Motivation

#2174

On epoll platforms, the read end of a pipe closing is signaled to the write end
through the `EPOLLERR` event [[1](http://man7.org/linux/man-pages/man2/epoll_ctl.2.html)]. If readiness is not registered for this
event, it will silently pass through `epoll_wait` calls.

Additionally, this specific case that `EPOLLERR` is triggered leaves the write
end of the pipe (parent process) waiting for a wakeup that never occurs.

## Solution

Similar to the `HUP` event on Unix platforms, errors are now always masked
through registrations so that both read and write ends of a connection are made
aware of errors.

In cases where pipes are used and the read end closes, write ends that are
waiting for a wakeup are properly notified and try to write again. This allows
a client to observe `BrokenPipe` and go through the proper cleanup and/or
restablishment of connection.

Closes #2174

Signed-off-by: Kevin Leimkuhler <[email protected]>

tokio/src/io/driver/mod.rs

@@ -146,7 +146,7 @@ impl Driver {
             return;
         }
 
-        if ready.is_writable() || platform::is_hup(ready) {
+        if ready.is_writable() || platform::is_hup(ready) || platform::is_error(ready) {
             wr = io.writer.take_waker();
         }
 
@@ -293,7 +293,7 @@ impl Direction {
                 // Everything except writable is signaled through read.
                 mio::Ready::all() - mio::Ready::writable()
             }
-            Direction::Write => mio::Ready::writable() | platform::hup(),
+            Direction::Write => mio::Ready::writable() | platform::hup() | platform::error(),
         }
     }
 }

tokio/src/io/driver/platform.rs

@@ -12,6 +12,14 @@ mod sys {
     pub(crate) fn is_hup(ready: Ready) -> bool {
         UnixReady::from(ready).is_hup()
     }
+
+    pub(crate) fn error() -> Ready {
+        UnixReady::error().into()
+    }
+
+    pub(crate) fn is_error(ready: Ready) -> bool {
+        UnixReady::from(ready).is_error()
+    }
 }
 
 #[cfg(windows)]
@@ -25,4 +33,12 @@ mod sys {
     pub(crate) fn is_hup(_: Ready) -> bool {
         false
     }
+
+    pub(crate) fn error() -> Ready {
+        Ready::empty()
+    }
+
+    pub(crate) fn is_error(_: Ready) -> bool {
+        false
+    }
 }

tokio/src/io/poll_evented.rs

@@ -123,7 +123,7 @@ macro_rules! poll_ready {
     ($me:expr, $mask:expr, $cache:ident, $take:ident, $poll:expr) => {{
         // Load cached & encoded readiness.
         let mut cached = $me.inner.$cache.load(Relaxed);
-        let mask = $mask | platform::hup();
+        let mask = $mask | platform::hup() | platform::error();
 
         // See if the current readiness matches any bits.
         let mut ret = mio::Ready::from_usize(cached) & $mask;

tokio/src/io/registration.rs

@@ -229,18 +229,26 @@ impl Registration {
         }
 
         let mask = direction.mask();
-        let mask_no_hup = (mask - platform::hup()).as_usize();
+        let mask_no_hup = (mask - platform::hup() - platform::error()).as_usize();
 
         let sched = inner.io_dispatch.get(self.address).unwrap();
 
-        // This consumes the current readiness state **except** for HUP. HUP is
+        // This consumes the current readiness state **except** for HUP and
-        // excluded because a) it is a final state and never transitions out of
+        // error. HUP and error are excluded because a) they are final states
-        // HUP and b) both the read AND the write directions need to be able to
+        // and never transitition out and b) both the read AND the write
-        // observe this state.
+        // directions need to be able to obvserve these states.
         //
-        // If HUP were to be cleared when `direction` is `Read`, then when
+        // # Platform-specific behavior
-        // `poll_ready` is called again with a _`direction` of `Write`, the HUP
+        //
-        // state would not be visible.
+        // HUP and error readiness are platform-specific. On epoll platforms,
+        // HUP has specific conditions that must be met by both peers of a
+        // connection in order to be triggered.
+        //
+        // On epoll platforms, `EPOLLERR` is signaled through
+        // `UnixReady::error()` and is important to be observable by both read
+        // AND write. A specific case that `EPOLLERR` occurs is when the read
+        // end of a pipe is closed. When this occurs, a peer blocked by
+        // writing to the pipe should be notified.
         let curr_ready = sched
             .set_readiness(self.address, |curr| curr & (!mask_no_hup))
             .unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address));

tokio/tests/process_issue_2174.rs

@@ -0,0 +1,46 @@
+#![cfg(feature = "process")]
+#![warn(rust_2018_idioms)]
+// This test reveals a difference in behavior of kqueue on FreeBSD. When the
+// reader disconnects, there does not seem to be an `EVFILT_WRITE` filter that
+// is returned.
+//
+// It is expected that `EVFILT_WRITE` would be returned with either the
+// `EV_EOF` or `EV_ERROR` flag set. If either flag is set a write would be
+// attempted, but that does not seem to occur.
+#![cfg(all(unix, not(target_os = "freebsd")))]
+
+use std::process::Stdio;
+use std::time::Duration;
+use tokio::prelude::*;
+use tokio::process::Command;
+use tokio::time;
+use tokio_test::assert_err;
+
+#[tokio::test]
+async fn issue_2174() {
+    let mut child = Command::new("sleep")
+        .arg("2")
+        .stdin(Stdio::piped())
+        .stdout(Stdio::null())
+        .spawn()
+        .unwrap();
+    let mut input = child.stdin.take().unwrap();
+
+    // Writes will buffer up to 65_636. This *should* loop at least 8 times
+    // and then register interest.
+    let handle = tokio::spawn(async move {
+        let data = [0u8; 8192];
+        loop {
+            input.write_all(&data).await.unwrap();
+        }
+    });
+
+    // Sleep enough time so that the child process's stdin's buffer fills.
+    time::delay_for(Duration::from_secs(1)).await;
+
+    // Kill the child process.
+    child.kill().unwrap();
+    let _ = child.await;
+
+    assert_err!(handle.await);
+}