Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Rename with_io to try_io and drop with_poll #3306

Merged
merged 6 commits into from
Dec 21, 2020
Merged

Rename with_io to try_io and drop with_poll #3306

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8f4c16e
Rename with_io to try_io and drop with_poll
Darksonn Dec 21, 2020
43369c7
rustfmt
Darksonn Dec 21, 2020
dfdabf1
Update WouldBlock to opaque struct
Darksonn Dec 21, 2020
63766b2
Fix test
Darksonn Dec 21, 2020
bf5d9e1
tcp flush is a no-op
Darksonn Dec 21, 2020
ed7d5ab
fix test
Darksonn Dec 21, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 49 additions & 69 deletions tokio/src/io/async_fd.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,12 +64,12 @@ 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};
/// use tokio::io::AsyncWrite;
/// use tokio::io::unix::AsyncFd;
/// use tokio::io::unix::{AsyncFd, TryIoError};
///
/// pub struct AsyncTcpStream {
/// inner: AsyncFd<TcpStream>,
Expand All @@ -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(TryIoError::WouldBlock) => continue,
/// }
/// }
/// }
Expand All @@ -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(TryIoError::WouldBlock) => continue,
/// }
/// }
/// }
Expand All @@ -117,9 +117,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().flush()) {
/// Err(err) if err.kind() == ErrorKind::WouldBlock => continue,
/// result => return Poll::Ready(result),
/// match guard.try_io(|inner| inner.get_ref().flush()) {
/// Ok(result) => return Poll::Ready(result),
/// Err(TryIoError::WouldBlock) => continue,
/// }
/// }
/// }
Expand Down Expand Up @@ -510,8 +510,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
Expand All @@ -522,10 +526,10 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> {
/// create this `AsyncFdReadyGuard`.
///
/// [`WouldBlock`]: std::io::ErrorKind::WouldBlock
pub fn with_io<R>(
pub fn try_io<R>(
&mut self,
f: impl FnOnce(&AsyncFd<Inner>) -> io::Result<R>,
) -> io::Result<R> {
) -> Result<io::Result<R>, TryIoError> {
let result = f(self.async_fd);

if let Err(e) = result.as_ref() {
Expand All @@ -534,32 +538,12 @@ 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<R>(
&mut self,
f: impl FnOnce(&AsyncFd<Inner>) -> std::task::Poll<R>,
) -> std::task::Poll<R> {
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::WouldBlock)
},
result => Ok(result),
}

result
}
}

Expand Down Expand Up @@ -589,8 +573,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
Expand All @@ -601,10 +589,10 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> {
/// create this `AsyncFdReadyGuard`.
///
/// [`WouldBlock`]: std::io::ErrorKind::WouldBlock
pub fn with_io<R>(
pub fn try_io<R>(
&mut self,
f: impl FnOnce(&mut AsyncFd<Inner>) -> io::Result<R>,
) -> io::Result<R> {
) -> Result<io::Result<R>, TryIoError> {
let result = f(&mut self.async_fd);

if let Err(e) = result.as_ref() {
Expand All @@ -613,32 +601,12 @@ 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<R>(
&mut self,
f: impl FnOnce(&mut AsyncFd<Inner>) -> std::task::Poll<R>,
) -> std::task::Poll<R> {
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::WouldBlock)
},
result => Ok(result),
}

result
}
}

Expand All @@ -657,3 +625,15 @@ impl<'a, T: std::fmt::Debug + AsRawFd> std::fmt::Debug for AsyncFdReadyMutGuard<
.finish()
}
}

/// The error type returned by [`try_io`].
///
/// [`try_io`]: method@AsyncFdReadyGuard::try_io
#[derive(Debug)]
pub enum TryIoError {
Darksonn marked this conversation as resolved.
Show resolved Hide resolved
/// This error indicates that the IO resource returned a [`WouldBlock`] error.
///
/// [`WouldBlock`]: std::io::ErrorKind::WouldBlock
WouldBlock,
}

2 changes: 1 addition & 1 deletion tokio/src/io/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
}
}

Expand Down
17 changes: 5 additions & 12 deletions tokio/tests/io_async_fd.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ 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)
Expand Down Expand Up @@ -233,12 +233,7 @@ 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();
Expand Down Expand Up @@ -313,9 +308,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();
Expand All @@ -327,13 +320,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();
Expand Down