diff --git a/tokio/src/io/async_fd.rs b/tokio/src/io/async_fd.rs index 901f663ac..08d7b91f5 100644 --- a/tokio/src/io/async_fd.rs +++ b/tokio/src/io/async_fd.rs @@ -64,7 +64,7 @@ use std::{task::Context, task::Poll}; /// /// ```no_run /// use futures::ready; -/// use std::io::{self, Read, Write, ErrorKind}; +/// use std::io::{self, Read, Write}; /// use std::net::TcpStream; /// use std::pin::Pin; /// use std::task::{Context, Poll}; @@ -86,9 +86,9 @@ use std::{task::Context, task::Poll}; /// loop { /// let mut guard = self.inner.readable().await?; /// -/// match guard.with_io(|inner| inner.get_ref().read(out)) { -/// Err(err) if err.kind() == ErrorKind::WouldBlock => continue, -/// result => return result, +/// match guard.try_io(|inner| inner.get_ref().read(out)) { +/// Ok(result) => return result, +/// Err(_would_block) => continue, /// } /// } /// } @@ -103,9 +103,9 @@ use std::{task::Context, task::Poll}; /// loop { /// let mut guard = ready!(self.inner.poll_write_ready(cx))?; /// -/// match guard.with_io(|inner| inner.get_ref().write(buf)) { -/// Err(err) if err.kind() == ErrorKind::WouldBlock => continue, -/// result => return Poll::Ready(result), +/// match guard.try_io(|inner| inner.get_ref().write(buf)) { +/// Ok(result) => return Poll::Ready(result), +/// Err(_would_block) => continue, /// } /// } /// } @@ -114,24 +114,15 @@ use std::{task::Context, task::Poll}; /// self: Pin<&mut Self>, /// cx: &mut Context<'_>, /// ) -> Poll> { -/// loop { -/// let mut guard = ready!(self.inner.poll_write_ready(cx))?; -/// -/// match guard.with_io(|inner| inner.get_ref().flush()) { -/// Err(err) if err.kind() == ErrorKind::WouldBlock => continue, -/// result => return Poll::Ready(result), -/// } -/// } +/// // tcp flush is a no-op +/// Poll::Ready(Ok(())) /// } /// /// fn poll_shutdown( -/// mut self: Pin<&mut Self>, +/// self: Pin<&mut Self>, /// cx: &mut Context<'_>, /// ) -> Poll> { -/// ready!(self.as_mut().poll_flush(cx))?; -/// /// self.inner.get_ref().shutdown(std::net::Shutdown::Write)?; -/// /// Poll::Ready(Ok(())) /// } /// } @@ -510,8 +501,12 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> { // no-op } - /// Performs the IO operation `f`; if `f` returns a [`WouldBlock`] error, - /// the readiness state associated with this file descriptor is cleared. + /// Performs the provided IO operation. + /// + /// If `f` returns a [`WouldBlock`] error, the readiness state associated + /// with this file descriptor is cleared, and the method returns + /// `Err(TryIoError::WouldBlock)`. You will typically need to poll the + /// `AsyncFd` again when this happens. /// /// This method helps ensure that the readiness state of the underlying file /// descriptor remains in sync with the tokio-side readiness state, by @@ -522,10 +517,10 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> { /// create this `AsyncFdReadyGuard`. /// /// [`WouldBlock`]: std::io::ErrorKind::WouldBlock - pub fn with_io( + pub fn try_io( &mut self, f: impl FnOnce(&AsyncFd) -> io::Result, - ) -> io::Result { + ) -> Result, TryIoError> { let result = f(self.async_fd); if let Err(e) = result.as_ref() { @@ -534,32 +529,10 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> { } } - result - } - - /// Performs the IO operation `f`; if `f` returns [`Pending`], the readiness - /// state associated with this file descriptor is cleared. - /// - /// This method helps ensure that the readiness state of the underlying file - /// descriptor remains in sync with the tokio-side readiness state, by - /// clearing the tokio-side state only when a [`Pending`] condition occurs. - /// It is the responsibility of the caller to ensure that `f` returns - /// [`Pending`] only if the file descriptor that originated this - /// `AsyncFdReadyGuard` no longer expresses the readiness state that was queried to - /// create this `AsyncFdReadyGuard`. - /// - /// [`Pending`]: std::task::Poll::Pending - pub fn with_poll( - &mut self, - f: impl FnOnce(&AsyncFd) -> std::task::Poll, - ) -> std::task::Poll { - let result = f(&self.async_fd); - - if result.is_pending() { - self.clear_ready(); + match result { + Err(err) if err.kind() == io::ErrorKind::WouldBlock => Err(TryIoError(())), + result => Ok(result), } - - result } } @@ -589,8 +562,12 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> { // no-op } - /// Performs the IO operation `f`; if `f` returns a [`WouldBlock`] error, - /// the readiness state associated with this file descriptor is cleared. + /// Performs the provided IO operation. + /// + /// If `f` returns a [`WouldBlock`] error, the readiness state associated + /// with this file descriptor is cleared, and the method returns + /// `Err(TryIoError::WouldBlock)`. You will typically need to poll the + /// `AsyncFd` again when this happens. /// /// This method helps ensure that the readiness state of the underlying file /// descriptor remains in sync with the tokio-side readiness state, by @@ -601,10 +578,10 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> { /// create this `AsyncFdReadyGuard`. /// /// [`WouldBlock`]: std::io::ErrorKind::WouldBlock - pub fn with_io( + pub fn try_io( &mut self, f: impl FnOnce(&mut AsyncFd) -> io::Result, - ) -> io::Result { + ) -> Result, TryIoError> { let result = f(&mut self.async_fd); if let Err(e) = result.as_ref() { @@ -613,32 +590,10 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> { } } - result - } - - /// Performs the IO operation `f`; if `f` returns [`Pending`], the readiness - /// state associated with this file descriptor is cleared. - /// - /// This method helps ensure that the readiness state of the underlying file - /// descriptor remains in sync with the tokio-side readiness state, by - /// clearing the tokio-side state only when a [`Pending`] condition occurs. - /// It is the responsibility of the caller to ensure that `f` returns - /// [`Pending`] only if the file descriptor that originated this - /// `AsyncFdReadyGuard` no longer expresses the readiness state that was queried to - /// create this `AsyncFdReadyGuard`. - /// - /// [`Pending`]: std::task::Poll::Pending - pub fn with_poll( - &mut self, - f: impl FnOnce(&mut AsyncFd) -> std::task::Poll, - ) -> std::task::Poll { - let result = f(&mut self.async_fd); - - if result.is_pending() { - self.clear_ready(); + match result { + Err(err) if err.kind() == io::ErrorKind::WouldBlock => Err(TryIoError(())), + result => Ok(result), } - - result } } @@ -657,3 +612,12 @@ impl<'a, T: std::fmt::Debug + AsRawFd> std::fmt::Debug for AsyncFdReadyMutGuard< .finish() } } + +/// The error type returned by [`try_io`]. +/// +/// This error indicates that the IO resource returned a [`WouldBlock`] error. +/// +/// [`WouldBlock`]: std::io::ErrorKind::WouldBlock +/// [`try_io`]: method@AsyncFdReadyGuard::try_io +#[derive(Debug)] +pub struct TryIoError(()); diff --git a/tokio/src/io/mod.rs b/tokio/src/io/mod.rs index 9b97ec392..6942a7105 100644 --- a/tokio/src/io/mod.rs +++ b/tokio/src/io/mod.rs @@ -222,7 +222,7 @@ cfg_net_unix! { pub mod unix { //! Asynchronous IO structures specific to Unix-like operating systems. - pub use super::async_fd::{AsyncFd, AsyncFdReadyGuard, AsyncFdReadyMutGuard}; + pub use super::async_fd::{AsyncFd, AsyncFdReadyGuard, AsyncFdReadyMutGuard, TryIoError}; } } diff --git a/tokio/tests/io_async_fd.rs b/tokio/tests/io_async_fd.rs index fc1002f03..d1586bb36 100644 --- a/tokio/tests/io_async_fd.rs +++ b/tokio/tests/io_async_fd.rs @@ -201,7 +201,10 @@ async fn reset_readable() { let mut guard = readable.await.unwrap(); - guard.with_io(|_| afd_a.get_ref().read(&mut [0])).unwrap(); + guard + .try_io(|_| afd_a.get_ref().read(&mut [0])) + .unwrap() + .unwrap(); // `a` is not readable, but the reactor still thinks it is // (because we have not observed a not-ready error yet) @@ -233,12 +236,10 @@ async fn reset_writable() { let mut guard = afd_a.writable().await.unwrap(); // Write until we get a WouldBlock. This also clears the ready state. - loop { - if let Err(e) = guard.with_io(|_| afd_a.get_ref().write(&[0; 512][..])) { - assert_eq!(ErrorKind::WouldBlock, e.kind()); - break; - } - } + while guard + .try_io(|_| afd_a.get_ref().write(&[0; 512][..])) + .is_ok() + {} // Writable state should be cleared now. let writable = afd_a.writable(); @@ -313,9 +314,7 @@ async fn reregister() { } #[tokio::test] -async fn with_poll() { - use std::task::Poll; - +async fn try_io() { let (a, mut b) = socketpair(); b.write_all(b"0").unwrap(); @@ -327,13 +326,13 @@ async fn with_poll() { afd_a.get_ref().read_exact(&mut [0]).unwrap(); // Should not clear the readable state - let _ = guard.with_poll(|_| Poll::Ready(())); + let _ = guard.try_io(|_| Ok(())); // Still readable... let _ = afd_a.readable().await.unwrap(); // Should clear the readable state - let _ = guard.with_poll(|_| Poll::Pending::<()>); + let _ = guard.try_io(|_| io::Result::<()>::Err(ErrorKind::WouldBlock.into())); // Assert not readable let readable = afd_a.readable();