io: make EPOLLERR awake AsyncFd::readable

examples/Cargo.toml

@@ -23,6 +23,7 @@ httparse = "1.0"
 httpdate = "1.0"
 once_cell = "1.5.2"
 rand = "0.8.3"
+nix = "0.23"
 
 [target.'cfg(windows)'.dev-dependencies.winapi]
 version = "0.3.8"
@@ -91,3 +92,11 @@ path = "named-pipe-ready.rs"
 [[example]]
 name = "named-pipe-multi-client"
 path = "named-pipe-multi-client.rs"
+
+[[example]]
+name = "fuse"
+path = "fuse.rs"
+
+[[example]]
+name = "fuse2"
+path = "fuse2.rs"

examples/fuse.rs

@@ -0,0 +1,63 @@
+// This "test" is based on the code sample in #3442
+// It is meant to test that `readable()` will surface the
+// polling error to the caller.
+// This program has to be run as `root` to be able to use
+// fuse.
+// The expected output is a polling error along the lines of:
+// ```
+// thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Custom { kind: Other, error: "Polling error" }', examples/fuse.rs:44:14
+// ```
+// This is just to show what testing I'm doing, and I don't intend
+// to get it merged
+use tokio::io::{unix::AsyncFd, Interest};
+
+use nix::{
+    fcntl::{fcntl, open, FcntlArg, OFlag},
+    mount::{mount, umount, MsFlags},
+    sys::stat::Mode,
+    unistd::read,
+};
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+    let fuse = open("/dev/fuse", OFlag::O_RDWR, Mode::empty()).unwrap();
+    let fcntl_arg = FcntlArg::F_SETFL(OFlag::O_NONBLOCK | OFlag::O_LARGEFILE);
+    fcntl(fuse, fcntl_arg).unwrap();
+
+    let options = format!(
+        "fd={},user_id=0,group_id=0,allow_other,rootmode=40000",
+        fuse
+    );
+
+    mount(
+        Some("test"),
+        "/mnt",
+        Some("fuse"),
+        MsFlags::empty(),
+        Some(options.as_str()),
+    )
+    .unwrap();
+
+    let async_fd = AsyncFd::with_interest(fuse, Interest::READABLE).unwrap();
+
+    std::thread::spawn(|| {
+        std::thread::sleep(std::time::Duration::from_secs(2));
+        umount("/mnt").unwrap();
+    });
+
+    let mut buffer = [0; 8192];
+    loop {
+        let result = async_fd
+            .readable()
+            .await
+            .unwrap()
+            .try_io(|_| read(fuse, &mut buffer).map_err(Into::into));
+
+        match result {
+            Err(_) => continue,
+            Ok(result) => {
+                result.unwrap();
+            }
+        }
+    }
+}

examples/fuse2.rs

@@ -0,0 +1,62 @@
+// This "test" is based on the code sample in #3442
+// It is meant to test that `poll_read_ready()` will surface the
+// polling error to the caller.
+// This program has to be run as `root` to be able to use
+// fuse.
+// The expected output is a polling error along the lines of:
+// ```
+// thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Custom { kind: Other, error: "Polling error" }', examples/fuse.rs:44:14
+// ```
+// This is just to show what testing I'm doing, and I don't intend
+// to get it merged
+use tokio::io::{unix::AsyncFd, Interest};
+
+use nix::{
+    fcntl::{fcntl, open, FcntlArg, OFlag},
+    mount::{mount, umount, MsFlags},
+    sys::stat::Mode,
+    unistd::read,
+};
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+    let fuse = open("/dev/fuse", OFlag::O_RDWR, Mode::empty()).unwrap();
+    let fcntl_arg = FcntlArg::F_SETFL(OFlag::O_NONBLOCK | OFlag::O_LARGEFILE);
+    fcntl(fuse, fcntl_arg).unwrap();
+
+    let options = format!(
+        "fd={},user_id=0,group_id=0,allow_other,rootmode=40000",
+        fuse
+    );
+
+    mount(
+        Some("test"),
+        "/mnt",
+        Some("fuse"),
+        MsFlags::empty(),
+        Some(options.as_str()),
+    )
+    .unwrap();
+
+    let async_fd = AsyncFd::with_interest(fuse, Interest::READABLE).unwrap();
+
+    std::thread::spawn(|| {
+        std::thread::sleep(std::time::Duration::from_secs(1));
+        umount("/mnt").unwrap();
+    });
+
+    let mut buffer = [0; 8192];
+    loop {
+        let result = futures::future::poll_fn(|cx| async_fd.poll_read_ready(cx))
+            .await
+            .unwrap()
+            .try_io(|_| read(fuse, &mut buffer).map_err(Into::into));
+
+        match result {
+            Err(_) => continue,
+            Ok(result) => {
+                result.unwrap();
+            }
+        }
+    }
+}

tokio/Cargo.toml

@@ -142,6 +142,9 @@ wasm-bindgen-test = "0.3.0"
 [target.'cfg(target_os = "freebsd")'.dev-dependencies]
 mio-aio = { version = "0.6.0", features = ["tokio"] }
 
+[target.'cfg(target_os = "linux")'.dev-dependencies]
+userfaultfd = "0.4"
+
 [target.'cfg(loom)'.dev-dependencies]
 loom = { version = "0.5", features = ["futures", "checkpoint"] }
 

tokio/src/io/driver/ready.rs

@@ -7,6 +7,7 @@ const READABLE: usize = 0b0_01;
 const WRITABLE: usize = 0b0_10;
 const READ_CLOSED: usize = 0b0_0100;
 const WRITE_CLOSED: usize = 0b0_1000;
+const ERROR: usize = 0b1_0000;
 
 /// Describes the readiness state of an I/O resources.
 ///
@@ -31,8 +32,11 @@ impl Ready {
     /// Returns a `Ready` representing write closed readiness.
     pub const WRITE_CLOSED: Ready = Ready(WRITE_CLOSED);
 
+    /// Returns a `Ready` representing error readiness
+    pub const ERROR: Ready = Ready(ERROR);
+
     /// Returns a `Ready` representing readiness for all operations.
-    pub const ALL: Ready = Ready(READABLE | WRITABLE | READ_CLOSED | WRITE_CLOSED);
+    pub const ALL: Ready = Ready(READABLE | WRITABLE | READ_CLOSED | WRITE_CLOSED | ERROR);
 
     // Must remain crate-private to avoid adding a public dependency on Mio.
     pub(crate) fn from_mio(event: &mio::event::Event) -> Ready {
@@ -65,6 +69,10 @@ impl Ready {
             ready |= Ready::WRITE_CLOSED;
         }
 
+        if event.is_error() {
+            ready |= Ready::ERROR;
+        }
+
         ready
     }
 
@@ -144,6 +152,21 @@ impl Ready {
         self.contains(Ready::WRITE_CLOSED)
     }
 
+    /// Returns `true` if the value includes write-closed `readiness`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::Ready;
+    ///
+    /// assert!(!Ready::EMPTY.is_write_closed());
+    /// assert!(!Ready::WRITABLE.is_write_closed());
+    /// assert!(Ready::WRITE_CLOSED.is_write_closed());
+    /// ```
+    pub fn is_error(self) -> bool {
+        self.contains(Ready::ERROR)
+    }
+
     /// Returns true if `self` is a superset of `other`.
     ///
     /// `other` may represent more than one readiness operations, in which case
@@ -245,6 +268,7 @@ impl fmt::Debug for Ready {
             .field("is_writable", &self.is_writable())
             .field("is_read_closed", &self.is_read_closed())
             .field("is_write_closed", &self.is_write_closed())
+            .field("is_error", &self.is_error())
             .finish()
     }
 }

tokio/src/io/driver/registration.rs

@@ -161,8 +161,10 @@ impl Registration {
             return Poll::Ready(Err(gone()));
         }
 
+        // Regardless of whether we got a polling error or an actual
+        // `ReadyEvent`, count it as work having been done
         coop.made_progress();
-        Poll::Ready(Ok(ev))
+        Poll::Ready(Ok(ev?))
     }
 
     fn poll_io<R>(
@@ -242,7 +244,7 @@ cfg_io_readiness! {
                     )));
                 }
 
-                Pin::new(&mut fut).poll(cx).map(Ok)
+                Pin::new(&mut fut).poll(cx)
             }).await
         }