io: support PRIORITY epoll events (#5566)

Add support for epoll priority events. The commit adds `Interest::PRIORITY`, `ready`, and `ready_mut` functions to `AsyncFd`.

Closes #4885

tokio/Cargo.toml

@@ -103,7 +103,7 @@ pin-project-lite = "0.2.0"
 
 # Everything else is optional...
 bytes = { version = "1.0.0", optional = true }
-mio = { version = "0.8.4", optional = true, default-features = false }
+mio = { version = "0.8.6", optional = true, default-features = false }
 num_cpus = { version = "1.8.0", optional = true }
 parking_lot = { version = "0.12.0", optional = true }
 

tokio/src/io/async_fd.rs

@@ -1,4 +1,4 @@
-use crate::io::Interest;
+use crate::io::{Interest, Ready};
 use crate::runtime::io::{ReadyEvent, Registration};
 use crate::runtime::scheduler;
 
@@ -201,13 +201,14 @@ pub struct AsyncFdReadyMutGuard<'a, T: AsRawFd> {
     event: Option<ReadyEvent>,
 }
 
-const ALL_INTEREST: Interest = Interest::READABLE.add(Interest::WRITABLE);
-
 impl<T: AsRawFd> AsyncFd<T> {
     /// Creates an AsyncFd backed by (and taking ownership of) an object
     /// implementing [`AsRawFd`]. The backing file descriptor is cached at the
     /// time of creation.
     ///
+    /// Only configures the [`Interest::READABLE`] and [`Interest::WRITABLE`] interests. For more
+    /// control, use [`AsyncFd::with_interest`].
+    ///
     /// This method must be called in the context of a tokio runtime.
     ///
     /// # Panics
@@ -220,11 +221,12 @@ impl<T: AsRawFd> AsyncFd<T> {
     where
         T: AsRawFd,
     {
-        Self::with_interest(inner, ALL_INTEREST)
+        Self::with_interest(inner, Interest::READABLE | Interest::WRITABLE)
     }
 
-    /// Creates new instance as `new` with additional ability to customize interest,
+    /// Creates an AsyncFd backed by (and taking ownership of) an object
-    /// allowing to specify whether file descriptor will be polled for read, write or both.
+    /// implementing [`AsRawFd`], with a specific [`Interest`]. The backing
+    /// file descriptor is cached at the time of creation.
     ///
     /// # Panics
     ///
@@ -440,7 +442,96 @@ impl<T: AsRawFd> AsyncFd<T> {
         .into()
     }
 
-    async fn readiness(&self, interest: Interest) -> io::Result<AsyncFdReadyGuard<'_, T>> {
+    /// Waits for any of the requested ready states, returning a
+    /// [`AsyncFdReadyGuard`] that must be dropped to resume
+    /// polling for the requested ready states.
+    ///
+    /// The function may complete without the file descriptor being ready. This is a
+    /// false-positive and attempting an operation will return with
+    /// `io::ErrorKind::WouldBlock`. The function can also return with an empty
+    /// [`Ready`] set, so you should always check the returned value and possibly
+    /// wait again if the requested states are not set.
+    ///
+    /// When an IO operation does return `io::ErrorKind::WouldBlock`, the readiness must be cleared.
+    /// When a combined interest is used, it is important to clear only the readiness
+    /// that is actually observed to block. For instance when the combined
+    /// interest `Interest::READABLE | Interest::WRITABLE` is used, and a read blocks, only
+    /// read readiness should be cleared using the [`AsyncFdReadyGuard::clear_ready_matching`] method:
+    /// `guard.clear_ready_matching(Ready::READABLE)`.
+    /// Also clearing the write readiness in this case would be incorrect. The [`AsyncFdReadyGuard::clear_ready`]
+    /// method clears all readiness flags.
+    ///
+    /// This method takes `&self`, so it is possible to call this method
+    /// concurrently with other methods on this struct. This method only
+    /// provides shared access to the inner IO resource when handling the
+    /// [`AsyncFdReadyGuard`].
+    ///
+    /// # Examples
+    ///
+    /// Concurrently read and write to a [`std::net::TcpStream`] on the same task without
+    /// splitting.
+    ///
+    /// ```no_run
+    /// use std::error::Error;
+    /// use std::io;
+    /// use std::io::{Read, Write};
+    /// use std::net::TcpStream;
+    /// use tokio::io::unix::AsyncFd;
+    /// use tokio::io::{Interest, Ready};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn Error>> {
+    ///     let stream = TcpStream::connect("127.0.0.1:8080")?;
+    ///     stream.set_nonblocking(true)?;
+    ///     let stream = AsyncFd::new(stream)?;
+    ///
+    ///     loop {
+    ///         let mut guard = stream
+    ///             .ready(Interest::READABLE | Interest::WRITABLE)
+    ///             .await?;
+    ///
+    ///         if guard.ready().is_readable() {
+    ///             let mut data = vec![0; 1024];
+    ///             // Try to read data, this may still fail with `WouldBlock`
+    ///             // if the readiness event is a false positive.
+    ///             match stream.get_ref().read(&mut data) {
+    ///                 Ok(n) => {
+    ///                     println!("read {} bytes", n);
+    ///                 }
+    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                     // a read has blocked, but a write might still succeed.
+    ///                     // clear only the read readiness.
+    ///                     guard.clear_ready_matching(Ready::READABLE);
+    ///                     continue;
+    ///                 }
+    ///                 Err(e) => {
+    ///                     return Err(e.into());
+    ///                 }
+    ///             }
+    ///         }
+    ///
+    ///         if guard.ready().is_writable() {
+    ///             // Try to write data, this may still fail with `WouldBlock`
+    ///             // if the readiness event is a false positive.
+    ///             match stream.get_ref().write(b"hello world") {
+    ///                 Ok(n) => {
+    ///                     println!("write {} bytes", n);
+    ///                 }
+    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                     // a write has blocked, but a read might still succeed.
+    ///                     // clear only the write readiness.
+    ///                     guard.clear_ready_matching(Ready::WRITABLE);
+    ///                     continue;
+    ///                 }
+    ///                 Err(e) => {
+    ///                     return Err(e.into());
+    ///                 }
+    ///             }
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    pub async fn ready(&self, interest: Interest) -> io::Result<AsyncFdReadyGuard<'_, T>> {
         let event = self.registration.readiness(interest).await?;
 
         Ok(AsyncFdReadyGuard {
@@ -449,7 +540,94 @@ impl<T: AsRawFd> AsyncFd<T> {
         })
     }
 
-    async fn readiness_mut(
+    /// Waits for any of the requested ready states, returning a
+    /// [`AsyncFdReadyMutGuard`] that must be dropped to resume
+    /// polling for the requested ready states.
+    ///
+    /// The function may complete without the file descriptor being ready. This is a
+    /// false-positive and attempting an operation will return with
+    /// `io::ErrorKind::WouldBlock`. The function can also return with an empty
+    /// [`Ready`] set, so you should always check the returned value and possibly
+    /// wait again if the requested states are not set.
+    ///
+    /// When an IO operation does return `io::ErrorKind::WouldBlock`, the readiness must be cleared.
+    /// When a combined interest is used, it is important to clear only the readiness
+    /// that is actually observed to block. For instance when the combined
+    /// interest `Interest::READABLE | Interest::WRITABLE` is used, and a read blocks, only
+    /// read readiness should be cleared using the [`AsyncFdReadyMutGuard::clear_ready_matching`] method:
+    /// `guard.clear_ready_matching(Ready::READABLE)`.
+    /// Also clearing the write readiness in this case would be incorrect.
+    /// The [`AsyncFdReadyMutGuard::clear_ready`] method clears all readiness flags.
+    ///
+    /// This method takes `&mut self`, so it is possible to access the inner IO
+    /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
+    ///
+    /// # Examples
+    ///
+    /// Concurrently read and write to a [`std::net::TcpStream`] on the same task without
+    /// splitting.
+    ///
+    /// ```no_run
+    /// use std::error::Error;
+    /// use std::io;
+    /// use std::io::{Read, Write};
+    /// use std::net::TcpStream;
+    /// use tokio::io::unix::AsyncFd;
+    /// use tokio::io::{Interest, Ready};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn Error>> {
+    ///     let stream = TcpStream::connect("127.0.0.1:8080")?;
+    ///     stream.set_nonblocking(true)?;
+    ///     let mut stream = AsyncFd::new(stream)?;
+    ///
+    ///     loop {
+    ///         let mut guard = stream
+    ///             .ready_mut(Interest::READABLE | Interest::WRITABLE)
+    ///             .await?;
+    ///
+    ///         if guard.ready().is_readable() {
+    ///             let mut data = vec![0; 1024];
+    ///             // Try to read data, this may still fail with `WouldBlock`
+    ///             // if the readiness event is a false positive.
+    ///             match guard.get_inner_mut().read(&mut data) {
+    ///                 Ok(n) => {
+    ///                     println!("read {} bytes", n);
+    ///                 }
+    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                     // a read has blocked, but a write might still succeed.
+    ///                     // clear only the read readiness.
+    ///                     guard.clear_ready_matching(Ready::READABLE);
+    ///                     continue;
+    ///                 }
+    ///                 Err(e) => {
+    ///                     return Err(e.into());
+    ///                 }
+    ///             }
+    ///         }
+    ///
+    ///         if guard.ready().is_writable() {
+    ///             // Try to write data, this may still fail with `WouldBlock`
+    ///             // if the readiness event is a false positive.
+    ///             match guard.get_inner_mut().write(b"hello world") {
+    ///                 Ok(n) => {
+    ///                     println!("write {} bytes", n);
+    ///                 }
+    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                     // a write has blocked, but a read might still succeed.
+    ///                     // clear only the write readiness.
+    ///                     guard.clear_ready_matching(Ready::WRITABLE);
+    ///                     continue;
+    ///                 }
+    ///                 Err(e) => {
+    ///                     return Err(e.into());
+    ///                 }
+    ///             }
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    pub async fn ready_mut(
         &mut self,
         interest: Interest,
     ) -> io::Result<AsyncFdReadyMutGuard<'_, T>> {
@@ -471,7 +649,7 @@ impl<T: AsRawFd> AsyncFd<T> {
     /// [`AsyncFdReadyGuard`].
     #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
     pub async fn readable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> {
-        self.readiness(Interest::READABLE).await
+        self.ready(Interest::READABLE).await
     }
 
     /// Waits for the file descriptor to become readable, returning a
@@ -482,7 +660,7 @@ impl<T: AsRawFd> AsyncFd<T> {
     /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
     #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
     pub async fn readable_mut<'a>(&'a mut self) -> io::Result<AsyncFdReadyMutGuard<'a, T>> {
-        self.readiness_mut(Interest::READABLE).await
+        self.ready_mut(Interest::READABLE).await
     }
 
     /// Waits for the file descriptor to become writable, returning a
@@ -495,7 +673,7 @@ impl<T: AsRawFd> AsyncFd<T> {
     /// [`AsyncFdReadyGuard`].
     #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
     pub async fn writable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> {
-        self.readiness(Interest::WRITABLE).await
+        self.ready(Interest::WRITABLE).await
     }
 
     /// Waits for the file descriptor to become writable, returning a
@@ -506,7 +684,7 @@ impl<T: AsRawFd> AsyncFd<T> {
     /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
     #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
     pub async fn writable_mut<'a>(&'a mut self) -> io::Result<AsyncFdReadyMutGuard<'a, T>> {
-        self.readiness_mut(Interest::WRITABLE).await
+        self.ready_mut(Interest::WRITABLE).await
     }
 
     /// Reads or writes from the file descriptor using a user-provided IO operation.
@@ -641,22 +819,117 @@ impl<T: AsRawFd> Drop for AsyncFd<T> {
 }
 
 impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> {
-    /// Indicates to tokio that the file descriptor is no longer ready. The
+    /// Indicates to tokio that the file descriptor is no longer ready. All
-    /// internal readiness flag will be cleared, and tokio will wait for the
+    /// internal readiness flags will be cleared, and tokio will wait for the
     /// next edge-triggered readiness notification from the OS.
     ///
+    /// This function is commonly used with guards returned by [`AsyncFd::readable`] and
+    /// [`AsyncFd::writable`].
+    ///
     /// It is critical that this function not be called unless your code
     /// _actually observes_ that the file descriptor is _not_ ready. Do not call
     /// it simply because, for example, a read succeeded; it should be called
     /// when a read is observed to block.
-    ///
-    /// [`drop`]: method@std::mem::drop
     pub fn clear_ready(&mut self) {
         if let Some(event) = self.event.take() {
             self.async_fd.registration.clear_readiness(event);
         }
     }
 
+    /// Indicates to tokio that the file descriptor no longer has a specific readiness.
+    /// The internal readiness flag will be cleared, and tokio will wait for the
+    /// next edge-triggered readiness notification from the OS.
+    ///
+    /// This function is useful in combination with the [`AsyncFd::ready`] method when a
+    /// combined interest like `Interest::READABLE | Interest::WRITABLE` is used.
+    ///
+    /// It is critical that this function not be called unless your code
+    /// _actually observes_ that the file descriptor is _not_ ready for the provided `Ready`.
+    /// Do not call it simply because, for example, a read succeeded; it should be called
+    /// when a read is observed to block. Only clear the specific readiness that is observed to
+    /// block. For example when a read blocks when using a combined interest,
+    /// only clear `Ready::READABLE`.
+    ///
+    /// # Examples
+    ///
+    /// Concurrently read and write to a [`std::net::TcpStream`] on the same task without
+    /// splitting.
+    ///
+    /// ```no_run
+    /// use std::error::Error;
+    /// use std::io;
+    /// use std::io::{Read, Write};
+    /// use std::net::TcpStream;
+    /// use tokio::io::unix::AsyncFd;
+    /// use tokio::io::{Interest, Ready};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn Error>> {
+    ///     let stream = TcpStream::connect("127.0.0.1:8080")?;
+    ///     stream.set_nonblocking(true)?;
+    ///     let stream = AsyncFd::new(stream)?;
+    ///
+    ///     loop {
+    ///         let mut guard = stream
+    ///             .ready(Interest::READABLE | Interest::WRITABLE)
+    ///             .await?;
+    ///
+    ///         if guard.ready().is_readable() {
+    ///             let mut data = vec![0; 1024];
+    ///             // Try to read data, this may still fail with `WouldBlock`
+    ///             // if the readiness event is a false positive.
+    ///             match stream.get_ref().read(&mut data) {
+    ///                 Ok(n) => {
+    ///                     println!("read {} bytes", n);
+    ///                 }
+    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                     // a read has blocked, but a write might still succeed.
+    ///                     // clear only the read readiness.
+    ///                     guard.clear_ready_matching(Ready::READABLE);
+    ///                     continue;
+    ///                 }
+    ///                 Err(e) => {
+    ///                     return Err(e.into());
+    ///                 }
+    ///             }
+    ///         }
+    ///
+    ///         if guard.ready().is_writable() {
+    ///             // Try to write data, this may still fail with `WouldBlock`
+    ///             // if the readiness event is a false positive.
+    ///             match stream.get_ref().write(b"hello world") {
+    ///                 Ok(n) => {
+    ///                     println!("write {} bytes", n);
+    ///                 }
+    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                     // a write has blocked, but a read might still succeed.
+    ///                     // clear only the write readiness.
+    ///                     guard.clear_ready_matching(Ready::WRITABLE);
+    ///                     continue;
+    ///                 }
+    ///                 Err(e) => {
+    ///                     return Err(e.into());
+    ///                 }
+    ///             }
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    pub fn clear_ready_matching(&mut self, ready: Ready) {
+        if let Some(mut event) = self.event.take() {
+            self.async_fd
+                .registration
+                .clear_readiness(event.with_ready(ready));
+
+            // the event is no longer ready for the readiness that was just cleared
+            event.ready = event.ready - ready;
+
+            if !event.ready.is_empty() {
+                self.event = Some(event);
+            }
+        }
+    }
+
     /// This method should be invoked when you intentionally want to keep the
     /// ready flag asserted.
     ///
@@ -666,6 +939,20 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> {
         // no-op
     }
 
+    /// Get the [`Ready`] value associated with this guard.
+    ///
+    /// This method will return the empty readiness state if
+    /// [`AsyncFdReadyGuard::clear_ready`] has been called on
+    /// the guard.
+    ///
+    /// [`Ready`]: crate::io::Ready
+    pub fn ready(&self) -> Ready {
+        match &self.event {
+            Some(event) => event.ready,
+            None => Ready::EMPTY,
+        }
+    }
+
     /// Performs the provided IO operation.
     ///
     /// If `f` returns a [`WouldBlock`] error, the readiness state associated
@@ -751,22 +1038,117 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> {
 }
 
 impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> {
-    /// Indicates to tokio that the file descriptor is no longer ready. The
+    /// Indicates to tokio that the file descriptor is no longer ready. All
-    /// internal readiness flag will be cleared, and tokio will wait for the
+    /// internal readiness flags will be cleared, and tokio will wait for the
     /// next edge-triggered readiness notification from the OS.
     ///
+    /// This function is commonly used with guards returned by [`AsyncFd::readable_mut`] and
+    /// [`AsyncFd::writable_mut`].
+    ///
     /// It is critical that this function not be called unless your code
     /// _actually observes_ that the file descriptor is _not_ ready. Do not call
     /// it simply because, for example, a read succeeded; it should be called
     /// when a read is observed to block.
-    ///
-    /// [`drop`]: method@std::mem::drop
     pub fn clear_ready(&mut self) {
         if let Some(event) = self.event.take() {
             self.async_fd.registration.clear_readiness(event);
         }
     }
 
+    /// Indicates to tokio that the file descriptor no longer has a specific readiness.
+    /// The internal readiness flag will be cleared, and tokio will wait for the
+    /// next edge-triggered readiness notification from the OS.
+    ///
+    /// This function is useful in combination with the [`AsyncFd::ready_mut`] method when a
+    /// combined interest like `Interest::READABLE | Interest::WRITABLE` is used.
+    ///
+    /// It is critical that this function not be called unless your code
+    /// _actually observes_ that the file descriptor is _not_ ready for the provided `Ready`.
+    /// Do not call it simply because, for example, a read succeeded; it should be called
+    /// when a read is observed to block. Only clear the specific readiness that is observed to
+    /// block. For example when a read blocks when using a combined interest,
+    /// only clear `Ready::READABLE`.
+    ///
+    /// # Examples
+    ///
+    /// Concurrently read and write to a [`std::net::TcpStream`] on the same task without
+    /// splitting.
+    ///
+    /// ```no_run
+    /// use std::error::Error;
+    /// use std::io;
+    /// use std::io::{Read, Write};
+    /// use std::net::TcpStream;
+    /// use tokio::io::unix::AsyncFd;
+    /// use tokio::io::{Interest, Ready};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn Error>> {
+    ///     let stream = TcpStream::connect("127.0.0.1:8080")?;
+    ///     stream.set_nonblocking(true)?;
+    ///     let mut stream = AsyncFd::new(stream)?;
+    ///
+    ///     loop {
+    ///         let mut guard = stream
+    ///             .ready_mut(Interest::READABLE | Interest::WRITABLE)
+    ///             .await?;
+    ///
+    ///         if guard.ready().is_readable() {
+    ///             let mut data = vec![0; 1024];
+    ///             // Try to read data, this may still fail with `WouldBlock`
+    ///             // if the readiness event is a false positive.
+    ///             match guard.get_inner_mut().read(&mut data) {
+    ///                 Ok(n) => {
+    ///                     println!("read {} bytes", n);
+    ///                 }
+    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                     // a read has blocked, but a write might still succeed.
+    ///                     // clear only the read readiness.
+    ///                     guard.clear_ready_matching(Ready::READABLE);
+    ///                     continue;
+    ///                 }
+    ///                 Err(e) => {
+    ///                     return Err(e.into());
+    ///                 }
+    ///             }
+    ///         }
+    ///
+    ///         if guard.ready().is_writable() {
+    ///             // Try to write data, this may still fail with `WouldBlock`
+    ///             // if the readiness event is a false positive.
+    ///             match guard.get_inner_mut().write(b"hello world") {
+    ///                 Ok(n) => {
+    ///                     println!("write {} bytes", n);
+    ///                 }
+    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                     // a write has blocked, but a read might still succeed.
+    ///                     // clear only the write readiness.
+    ///                     guard.clear_ready_matching(Ready::WRITABLE);
+    ///                     continue;
+    ///                 }
+    ///                 Err(e) => {
+    ///                     return Err(e.into());
+    ///                 }
+    ///             }
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    pub fn clear_ready_matching(&mut self, ready: Ready) {
+        if let Some(mut event) = self.event.take() {
+            self.async_fd
+                .registration
+                .clear_readiness(event.with_ready(ready));
+
+            // the event is no longer ready for the readiness that was just cleared
+            event.ready = event.ready - ready;
+
+            if !event.ready.is_empty() {
+                self.event = Some(event);
+            }
+        }
+    }
+
     /// This method should be invoked when you intentionally want to keep the
     /// ready flag asserted.
     ///
@@ -776,6 +1158,20 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> {
         // no-op
     }
 
+    /// Get the [`Ready`] value associated with this guard.
+    ///
+    /// This method will return the empty readiness state if
+    /// [`AsyncFdReadyGuard::clear_ready`] has been called on
+    /// the guard.
+    ///
+    /// [`Ready`]: super::Ready
+    pub fn ready(&self) -> Ready {
+        match &self.event {
+            Some(event) => event.ready,
+            None => Ready::EMPTY,
+        }
+    }
+
     /// Performs the provided IO operation.
     ///
     /// If `f` returns a [`WouldBlock`] error, the readiness state associated

tokio/src/io/interest.rs

@@ -44,6 +44,11 @@ impl Interest {
     /// Writable interest includes write-closed events.
     pub const WRITABLE: Interest = Interest(mio::Interest::WRITABLE);
 
+    /// Returns a `Interest` set representing priority completion interests.
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    #[cfg_attr(docsrs, doc(cfg(any(target_os = "linux", target_os = "android"))))]
+    pub const PRIORITY: Interest = Interest(mio::Interest::PRIORITY);
+
     /// Returns true if the value includes readable interest.
     ///
     /// # Examples
@@ -78,6 +83,25 @@ impl Interest {
         self.0.is_writable()
     }
 
+    /// Returns true if the value includes priority interest.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::Interest;
+    ///
+    /// assert!(!Interest::READABLE.is_priority());
+    /// assert!(Interest::PRIORITY.is_priority());
+    ///
+    /// let both = Interest::READABLE | Interest::PRIORITY;
+    /// assert!(both.is_priority());
+    /// ```
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    #[cfg_attr(docsrs, doc(cfg(any(target_os = "linux", target_os = "android"))))]
+    pub const fn is_priority(self) -> bool {
+        self.0.is_priority()
+    }
+
     /// Add together two `Interest` values.
     ///
     /// This function works from a `const` context.
@@ -104,6 +128,8 @@ impl Interest {
         match self {
             Interest::READABLE => Ready::READABLE | Ready::READ_CLOSED,
             Interest::WRITABLE => Ready::WRITABLE | Ready::WRITE_CLOSED,
+            #[cfg(any(target_os = "linux", target_os = "android"))]
+            Interest::PRIORITY => Ready::PRIORITY | Ready::READ_CLOSED,
             _ => Ready::EMPTY,
         }
     }

tokio/src/io/poll_evented.rs

@@ -73,6 +73,9 @@ cfg_io_driver! {
 impl<E: Source> PollEvented<E> {
     /// Creates a new `PollEvented` associated with the default reactor.
     ///
+    /// The returned `PollEvented` has readable and writable interests. For more control, use
+    /// [`Self::new_with_interest`].
+    ///
     /// # Panics
     ///
     /// This function panics if thread-local runtime is not set.

tokio/src/io/ready.rs

@@ -7,6 +7,8 @@ const READABLE: usize = 0b0_01;
 const WRITABLE: usize = 0b0_10;
 const READ_CLOSED: usize = 0b0_0100;
 const WRITE_CLOSED: usize = 0b0_1000;
+#[cfg(any(target_os = "linux", target_os = "android"))]
+const PRIORITY: usize = 0b1_0000;
 
 /// Describes the readiness state of an I/O resources.
 ///
@@ -31,7 +33,17 @@ impl Ready {
     /// Returns a `Ready` representing write closed readiness.
     pub const WRITE_CLOSED: Ready = Ready(WRITE_CLOSED);
 
+    /// Returns a `Ready` representing priority readiness.
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    #[cfg_attr(docsrs, doc(cfg(any(target_os = "linux", target_os = "android"))))]
+    pub const PRIORITY: Ready = Ready(PRIORITY);
+
     /// Returns a `Ready` representing readiness for all operations.
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    pub const ALL: Ready = Ready(READABLE | WRITABLE | READ_CLOSED | WRITE_CLOSED | PRIORITY);
+
+    /// Returns a `Ready` representing readiness for all operations.
+    #[cfg(not(any(target_os = "linux", target_os = "android")))]
     pub const ALL: Ready = Ready(READABLE | WRITABLE | READ_CLOSED | WRITE_CLOSED);
 
     // Must remain crate-private to avoid adding a public dependency on Mio.
@@ -65,6 +77,13 @@ impl Ready {
             ready |= Ready::WRITE_CLOSED;
         }
 
+        #[cfg(any(target_os = "linux", target_os = "android"))]
+        {
+            if event.is_priority() {
+                ready |= Ready::PRIORITY;
+            }
+        }
+
         ready
     }
 
@@ -144,6 +163,23 @@ impl Ready {
         self.contains(Ready::WRITE_CLOSED)
     }
 
+    /// Returns `true` if the value includes priority `readiness`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::Ready;
+    ///
+    /// assert!(!Ready::EMPTY.is_priority());
+    /// assert!(!Ready::WRITABLE.is_priority());
+    /// assert!(Ready::PRIORITY.is_priority());
+    /// ```
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    #[cfg_attr(docsrs, doc(cfg(any(target_os = "linux", target_os = "android"))))]
+    pub fn is_priority(self) -> bool {
+        self.contains(Ready::PRIORITY)
+    }
+
     /// Returns true if `self` is a superset of `other`.
     ///
     /// `other` may represent more than one readiness operations, in which case
@@ -191,6 +227,12 @@ cfg_io_readiness! {
                 ready |= Ready::WRITE_CLOSED;
             }
 
+            #[cfg(any(target_os = "linux", target_os = "android"))]
+            if interest.is_priority() {
+                ready |= Ready::PRIORITY;
+                ready |= Ready::READ_CLOSED;
+            }
+
             ready
         }
 
@@ -240,11 +282,16 @@ impl ops::Sub<Ready> for Ready {
 
 impl fmt::Debug for Ready {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
-        fmt.debug_struct("Ready")
+        let mut fmt = fmt.debug_struct("Ready");
-            .field("is_readable", &self.is_readable())
+
+        fmt.field("is_readable", &self.is_readable())
             .field("is_writable", &self.is_writable())
             .field("is_read_closed", &self.is_read_closed())
-            .field("is_write_closed", &self.is_write_closed())
+            .field("is_write_closed", &self.is_write_closed());
-            .finish()
+
+        #[cfg(any(target_os = "linux", target_os = "android"))]
+        fmt.field("is_priority", &self.is_priority());
+
+        fmt.finish()
     }
 }

tokio/src/runtime/io/mod.rs

@@ -63,6 +63,18 @@ pub(crate) struct ReadyEvent {
     is_shutdown: bool,
 }
 
+cfg_net_unix!(
+    impl ReadyEvent {
+        pub(crate) fn with_ready(&self, ready: Ready) -> Self {
+            Self {
+                ready,
+                tick: self.tick,
+                is_shutdown: self.is_shutdown,
+            }
+        }
+    }
+);
+
 struct IoDispatcher {
     allocator: slab::Allocator<ScheduledIo>,
     is_shutdown: bool,

tokio/tests/io_async_fd.rs

@@ -599,3 +599,89 @@ fn driver_shutdown_wakes_poll_race() {
         assert_err!(futures::executor::block_on(poll_writable(&afd_a)));
     }
 }
+
+#[tokio::test]
+#[cfg(any(target_os = "linux", target_os = "android"))]
+async fn priority_event_on_oob_data() {
+    use std::net::SocketAddr;
+
+    use tokio::io::Interest;
+
+    let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
+
+    let listener = std::net::TcpListener::bind(addr).unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let client = std::net::TcpStream::connect(addr).unwrap();
+    let client = AsyncFd::with_interest(client, Interest::PRIORITY).unwrap();
+
+    let (stream, _) = listener.accept().unwrap();
+
+    // Sending out of band data should trigger priority event.
+    send_oob_data(&stream, b"hello").unwrap();
+
+    let _ = client.ready(Interest::PRIORITY).await.unwrap();
+}
+
+#[cfg(any(target_os = "linux", target_os = "android"))]
+fn send_oob_data<S: AsRawFd>(stream: &S, data: &[u8]) -> io::Result<usize> {
+    unsafe {
+        let res = libc::send(
+            stream.as_raw_fd(),
+            data.as_ptr().cast(),
+            data.len(),
+            libc::MSG_OOB,
+        );
+        if res == -1 {
+            Err(io::Error::last_os_error())
+        } else {
+            Ok(res as usize)
+        }
+    }
+}
+
+#[tokio::test]
+async fn clear_ready_matching_clears_ready() {
+    use tokio::io::{Interest, Ready};
+
+    let (a, mut b) = socketpair();
+
+    let afd_a = AsyncFd::new(a).unwrap();
+    b.write_all(b"0").unwrap();
+
+    let mut guard = afd_a
+        .ready(Interest::READABLE | Interest::WRITABLE)
+        .await
+        .unwrap();
+
+    assert_eq!(guard.ready(), Ready::READABLE | Ready::WRITABLE);
+
+    guard.clear_ready_matching(Ready::READABLE);
+    assert_eq!(guard.ready(), Ready::WRITABLE);
+
+    guard.clear_ready_matching(Ready::WRITABLE);
+    assert_eq!(guard.ready(), Ready::EMPTY);
+}
+
+#[tokio::test]
+async fn clear_ready_matching_clears_ready_mut() {
+    use tokio::io::{Interest, Ready};
+
+    let (a, mut b) = socketpair();
+
+    let mut afd_a = AsyncFd::new(a).unwrap();
+    b.write_all(b"0").unwrap();
+
+    let mut guard = afd_a
+        .ready_mut(Interest::READABLE | Interest::WRITABLE)
+        .await
+        .unwrap();
+
+    assert_eq!(guard.ready(), Ready::READABLE | Ready::WRITABLE);
+
+    guard.clear_ready_matching(Ready::READABLE);
+    assert_eq!(guard.ready(), Ready::WRITABLE);
+
+    guard.clear_ready_matching(Ready::WRITABLE);
+    assert_eq!(guard.ready(), Ready::EMPTY);
+}