use mio::unix::SourceFd; use std::error::Error; use std::fmt; use std::io; use std::os::unix::io::{AsRawFd, RawFd}; use std::{task::Context, task::Poll};
/// Associates an IO object backed by a Unix file descriptor with the tokio /// reactor, allowing for readiness to be polled. The file descriptor must be of /// a type that can be used with the OS polling facilities (ie, `poll`, `epoll`, /// `kqueue`, etc), such as a network socket or pipe, and the file descriptor /// must have the nonblocking mode set to true. /// /// Creating an [`AsyncFd`] registers the file descriptor with the current tokio /// Reactor, allowing you to directly await the file descriptor being readable /// or writable. Once registered, the file descriptor remains registered until /// the [`AsyncFd`] is dropped. /// /// The [`AsyncFd`] takes ownership of an arbitrary object to represent the IO /// object. It is intended that this object will handle closing the file /// descriptor when it is dropped, avoiding resource leaks and ensuring that the /// [`AsyncFd`] can clean up the registration before closing the file descriptor. /// The [`AsyncFd::into_inner`] function can be used to extract the inner object /// to retake control from the tokio IO reactor. /// /// The inner object is required to implement [`AsRawFd`]. This file descriptor /// must not change while [`AsyncFd`] owns the inner object, i.e. the /// [`AsRawFd::as_raw_fd`] method on the inner type must always return the same /// file descriptor when called multiple times. Failure to uphold this results /// in unspecified behavior in the IO driver, which may include breaking /// notifications for other sockets/etc. /// /// Polling for readiness is done by calling the async functions [`readable`] /// and [`writable`]. These functions complete when the associated readiness /// condition is observed. Any number of tasks can query the same `AsyncFd` in /// parallel, on the same or different conditions. /// /// On some platforms, the readiness detecting mechanism relies on /// edge-triggered notifications. This means that the OS will only notify Tokio /// when the file descriptor transitions from not-ready to ready. For this to /// work you should first try to read or write and only poll for readiness /// if that fails with an error of [`std::io::ErrorKind::WouldBlock`]. /// /// Tokio internally tracks when it has received a ready notification, and when /// readiness checking functions like [`readable`] and [`writable`] are called, /// if the readiness flag is set, these async functions will complete /// immediately. This however does mean that it is critical to ensure that this /// ready flag is cleared when (and only when) the file descriptor ceases to be /// ready. The [`AsyncFdReadyGuard`] returned from readiness checking functions /// serves this function; after calling a readiness-checking async function, /// you must use this [`AsyncFdReadyGuard`] to signal to tokio whether the file /// descriptor is no longer in a ready state. /// /// ## Use with to a poll-based API /// /// In some cases it may be desirable to use `AsyncFd` from APIs similar to /// [`TcpStream::poll_read_ready`]. The [`AsyncFd::poll_read_ready`] and /// [`AsyncFd::poll_write_ready`] functions are provided for this purpose. /// Because these functions don't create a future to hold their state, they have /// the limitation that only one task can wait on each direction (read or write) /// at a time. /// /// # Examples /// /// This example shows how to turn [`std::net::TcpStream`] asynchronous using /// `AsyncFd`. It implements the read/write operations both as an `async fn` /// and using the IO traits [`AsyncRead`] and [`AsyncWrite`]. /// /// ```no_run /// use futures::ready; /// use std::io::{self, Read, Write}; /// use std::net::TcpStream; /// use std::pin::Pin; /// use std::task::{Context, Poll}; /// use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; /// use tokio::io::unix::AsyncFd; /// /// pub struct AsyncTcpStream { /// inner: AsyncFd<TcpStream>, /// } /// /// impl AsyncTcpStream { /// pub fn new(tcp: TcpStream) -> io::Result<Self> { /// tcp.set_nonblocking(true)?; /// Ok(Self { /// inner: AsyncFd::new(tcp)?, /// }) /// } /// /// pub async fn read(&self, out: &mut [u8]) -> io::Result<usize> { /// loop { /// let mut guard = self.inner.readable().await?; /// /// match guard.try_io(|inner| inner.get_ref().read(out)) { /// Ok(result) => return result, /// Err(_would_block) => continue, /// } /// } /// } /// /// pub async fn write(&self, buf: &[u8]) -> io::Result<usize> { /// loop { /// let mut guard = self.inner.writable().await?; /// /// match guard.try_io(|inner| inner.get_ref().write(buf)) { /// Ok(result) => return result, /// Err(_would_block) => continue, /// } /// } /// } /// } /// /// impl AsyncRead for AsyncTcpStream { /// fn poll_read( /// self: Pin<&mut Self>, /// cx: &mut Context<'_>, /// buf: &mut ReadBuf<'_> /// ) -> Poll<io::Result<()>> { /// loop { /// let mut guard = ready!(self.inner.poll_read_ready(cx))?; /// /// let unfilled = buf.initialize_unfilled(); /// match guard.try_io(|inner| inner.get_ref().read(unfilled)) { /// Ok(Ok(len)) => { /// buf.advance(len); /// return Poll::Ready(Ok(())); /// }, /// Ok(Err(err)) => return Poll::Ready(Err(err)), /// Err(_would_block) => continue, /// } /// } /// } /// } /// /// impl AsyncWrite for AsyncTcpStream { /// fn poll_write( /// self: Pin<&mut Self>, /// cx: &mut Context<'_>, /// buf: &[u8] /// ) -> Poll<io::Result<usize>> { /// loop { /// let mut guard = ready!(self.inner.poll_write_ready(cx))?; /// /// match guard.try_io(|inner| inner.get_ref().write(buf)) { /// Ok(result) => return Poll::Ready(result), /// Err(_would_block) => continue, /// } /// } /// } /// /// fn poll_flush( /// self: Pin<&mut Self>, /// cx: &mut Context<'_>, /// ) -> Poll<io::Result<()>> { /// // tcp flush is a no-op /// Poll::Ready(Ok(())) /// } /// /// fn poll_shutdown( /// self: Pin<&mut Self>, /// cx: &mut Context<'_>, /// ) -> Poll<io::Result<()>> { /// self.inner.get_ref().shutdown(std::net::Shutdown::Write)?; /// Poll::Ready(Ok(())) /// } /// } /// ``` /// /// [`readable`]: method@Self::readable /// [`writable`]: method@Self::writable /// [`AsyncFdReadyGuard`]: struct@self::AsyncFdReadyGuard /// [`TcpStream::poll_read_ready`]: struct@crate::net::TcpStream /// [`AsyncRead`]: trait@crate::io::AsyncRead /// [`AsyncWrite`]: trait@crate::io::AsyncWrite pubstruct AsyncFd<T: AsRawFd> {
registration: Registration, // The inner value is always present. the Option is required for `drop` and `into_inner`. // In all other methods `unwrap` is valid, and will never panic.
inner: Option<T>,
}
/// Represents an IO-ready event detected on a particular file descriptor that /// has not yet been acknowledged. This is a `must_use` structure to help ensure /// that you do not forget to explicitly clear (or not clear) the event. /// /// This type exposes an immutable reference to the underlying IO object. #[must_use = "You must explicitly choose whether to clear the readiness state by calling a method on ReadyGuard"] pubstruct AsyncFdReadyGuard<'a, T: AsRawFd> {
async_fd: &'a AsyncFd<T>,
event: Option<ReadyEvent>,
}
/// Represents an IO-ready event detected on a particular file descriptor that /// has not yet been acknowledged. This is a `must_use` structure to help ensure /// that you do not forget to explicitly clear (or not clear) the event. /// /// This type exposes a mutable reference to the underlying IO object. #[must_use = "You must explicitly choose whether to clear the readiness state by calling a method on ReadyGuard"] pubstruct AsyncFdReadyMutGuard<'a, T: AsRawFd> {
async_fd: &'a mut AsyncFd<T>,
event: Option<ReadyEvent>,
}
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 /// /// This function panics if there is no current reactor set, or if the `rt` /// feature flag is not enabled. #[inline] #[track_caller] pubfn new(inner: T) -> io::Result<Self> where
T: AsRawFd,
{ Self::with_interest(inner, Interest::READABLE | Interest::WRITABLE)
}
/// Creates an [`AsyncFd`] backed by (and taking ownership of) an object /// implementing [`AsRawFd`], with a specific [`Interest`]. The backing /// file descriptor is cached at the time of creation. /// /// # Panics /// /// This function panics if there is no current reactor set, or if the `rt` /// feature flag is not enabled. #[inline] #[track_caller] pubfn with_interest(inner: T, interest: Interest) -> io::Result<Self> where
T: AsRawFd,
{ Self::new_with_handle_and_interest(inner, scheduler::Handle::current(), interest)
}
/// 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::try_with_interest`]. /// /// This method must be called in the context of a tokio runtime. /// /// In the case of failure, it returns [`AsyncFdTryNewError`] that contains the original object /// passed to this function. /// /// # Panics /// /// This function panics if there is no current reactor set, or if the `rt` /// feature flag is not enabled. #[inline] #[track_caller] pubfn try_new(inner: T) -> Result<Self, AsyncFdTryNewError<T>> where
T: AsRawFd,
{ Self::try_with_interest(inner, Interest::READABLE | Interest::WRITABLE)
}
/// Creates an [`AsyncFd`] backed by (and taking ownership of) an object /// implementing [`AsRawFd`], with a specific [`Interest`]. The backing /// file descriptor is cached at the time of creation. /// /// In the case of failure, it returns [`AsyncFdTryNewError`] that contains the original object /// passed to this function. /// /// # Panics /// /// This function panics if there is no current reactor set, or if the `rt` /// feature flag is not enabled. #[inline] #[track_caller] pubfn try_with_interest(inner: T, interest: Interest) -> Result<Self, AsyncFdTryNewError<T>> where
T: AsRawFd,
{ Self::try_new_with_handle_and_interest(inner, scheduler::Handle::current(), interest)
}
/// Returns a shared reference to the backing object of this [`AsyncFd`]. #[inline] pubfn get_ref(&self) -> &T { self.inner.as_ref().unwrap()
}
/// Returns a mutable reference to the backing object of this [`AsyncFd`]. #[inline] pubfn get_mut(&mutself) -> &mut T { self.inner.as_mut().unwrap()
}
fn take_inner(&mutself) -> Option<T> { let inner = self.inner.take()?; let fd = inner.as_raw_fd();
let _ = self.registration.deregister(&mut SourceFd(&fd));
Some(inner)
}
/// Deregisters this file descriptor and returns ownership of the backing /// object. pubfn into_inner(mutself) -> T { self.take_inner().unwrap()
}
/// Polls for read readiness. /// /// If the file descriptor is not currently ready for reading, this method /// will store a clone of the [`Waker`] from the provided [`Context`]. When the /// file descriptor becomes ready for reading, [`Waker::wake`] will be called. /// /// Note that on multiple calls to [`poll_read_ready`] or /// [`poll_read_ready_mut`], only the `Waker` from the `Context` passed to the /// most recent call is scheduled to receive a wakeup. (However, /// [`poll_write_ready`] retains a second, independent waker). /// /// This method is intended for cases where creating and pinning a future /// via [`readable`] is not feasible. Where possible, using [`readable`] is /// preferred, as this supports polling from multiple tasks at once. /// /// 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`]. /// /// [`poll_read_ready`]: method@Self::poll_read_ready /// [`poll_read_ready_mut`]: method@Self::poll_read_ready_mut /// [`poll_write_ready`]: method@Self::poll_write_ready /// [`readable`]: method@Self::readable /// [`Context`]: struct@std::task::Context /// [`Waker`]: struct@std::task::Waker /// [`Waker::wake`]: method@std::task::Waker::wake pubfn poll_read_ready<'a>(
&'a self,
cx: &mut Context<'_>,
) -> Poll<io::Result<AsyncFdReadyGuard<'a, T>>> { let event = ready!(self.registration.poll_read_ready(cx))?;
/// Polls for read readiness. /// /// If the file descriptor is not currently ready for reading, this method /// will store a clone of the [`Waker`] from the provided [`Context`]. When the /// file descriptor becomes ready for reading, [`Waker::wake`] will be called. /// /// Note that on multiple calls to [`poll_read_ready`] or /// [`poll_read_ready_mut`], only the `Waker` from the `Context` passed to the /// most recent call is scheduled to receive a wakeup. (However, /// [`poll_write_ready`] retains a second, independent waker). /// /// This method is intended for cases where creating and pinning a future /// via [`readable`] is not feasible. Where possible, using [`readable`] is /// preferred, as this supports polling from multiple tasks at once. /// /// This method takes `&mut self`, so it is possible to access the inner IO /// resource mutably when handling the [`AsyncFdReadyMutGuard`]. /// /// [`poll_read_ready`]: method@Self::poll_read_ready /// [`poll_read_ready_mut`]: method@Self::poll_read_ready_mut /// [`poll_write_ready`]: method@Self::poll_write_ready /// [`readable`]: method@Self::readable /// [`Context`]: struct@std::task::Context /// [`Waker`]: struct@std::task::Waker /// [`Waker::wake`]: method@std::task::Waker::wake pubfn poll_read_ready_mut<'a>(
&'a mut self,
cx: &mut Context<'_>,
) -> Poll<io::Result<AsyncFdReadyMutGuard<'a, T>>> { let event = ready!(self.registration.poll_read_ready(cx))?;
/// Polls for write readiness. /// /// If the file descriptor is not currently ready for writing, this method /// will store a clone of the [`Waker`] from the provided [`Context`]. When the /// file descriptor becomes ready for writing, [`Waker::wake`] will be called. /// /// Note that on multiple calls to [`poll_write_ready`] or /// [`poll_write_ready_mut`], only the `Waker` from the `Context` passed to the /// most recent call is scheduled to receive a wakeup. (However, /// [`poll_read_ready`] retains a second, independent waker). /// /// This method is intended for cases where creating and pinning a future /// via [`writable`] is not feasible. Where possible, using [`writable`] is /// preferred, as this supports polling from multiple tasks at once. /// /// 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`]. /// /// [`poll_read_ready`]: method@Self::poll_read_ready /// [`poll_write_ready`]: method@Self::poll_write_ready /// [`poll_write_ready_mut`]: method@Self::poll_write_ready_mut /// [`writable`]: method@Self::readable /// [`Context`]: struct@std::task::Context /// [`Waker`]: struct@std::task::Waker /// [`Waker::wake`]: method@std::task::Waker::wake pubfn poll_write_ready<'a>(
&'a self,
cx: &mut Context<'_>,
) -> Poll<io::Result<AsyncFdReadyGuard<'a, T>>> { let event = ready!(self.registration.poll_write_ready(cx))?;
/// Polls for write readiness. /// /// If the file descriptor is not currently ready for writing, this method /// will store a clone of the [`Waker`] from the provided [`Context`]. When the /// file descriptor becomes ready for writing, [`Waker::wake`] will be called. /// /// Note that on multiple calls to [`poll_write_ready`] or /// [`poll_write_ready_mut`], only the `Waker` from the `Context` passed to the /// most recent call is scheduled to receive a wakeup. (However, /// [`poll_read_ready`] retains a second, independent waker). /// /// This method is intended for cases where creating and pinning a future /// via [`writable`] is not feasible. Where possible, using [`writable`] is /// preferred, as this supports polling from multiple tasks at once. /// /// This method takes `&mut self`, so it is possible to access the inner IO /// resource mutably when handling the [`AsyncFdReadyMutGuard`]. /// /// [`poll_read_ready`]: method@Self::poll_read_ready /// [`poll_write_ready`]: method@Self::poll_write_ready /// [`poll_write_ready_mut`]: method@Self::poll_write_ready_mut /// [`writable`]: method@Self::readable /// [`Context`]: struct@std::task::Context /// [`Waker`]: struct@std::task::Waker /// [`Waker::wake`]: method@std::task::Waker::wake pubfn poll_write_ready_mut<'a>(
&'a mut self,
cx: &mut Context<'_>,
) -> Poll<io::Result<AsyncFdReadyMutGuard<'a, T>>> { let event = ready!(self.registration.poll_write_ready(cx))?;
/// 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()); /// } /// } /// } /// } /// } /// ``` pubasyncfn ready(&self, interest: Interest) -> io::Result<AsyncFdReadyGuard<'_, T>> { let event = self.registration.readiness(interest).await?;
/// 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()); /// } /// } /// } /// } /// } /// ``` pubasyncfn ready_mut(
&mutself,
interest: Interest,
) -> io::Result<AsyncFdReadyMutGuard<'_, T>> { let event = self.registration.readiness(interest).await?;
/// Waits for the file descriptor to become readable, returning a /// [`AsyncFdReadyGuard`] that must be dropped to resume read-readiness /// polling. /// /// 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`]. #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering. pubasyncfn readable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> { self.ready(Interest::READABLE).await
}
/// Waits for the file descriptor to become readable, returning a /// [`AsyncFdReadyMutGuard`] that must be dropped to resume read-readiness /// polling. /// /// This method takes `&mut self`, so it is possible to access the inner IO /// resource mutably when handling the [`AsyncFdReadyMutGuard`]. #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering. pubasyncfn readable_mut<'a>(&'a mutself) -> io::Result<AsyncFdReadyMutGuard<'a, T>> { self.ready_mut(Interest::READABLE).await
}
/// Waits for the file descriptor to become writable, returning a /// [`AsyncFdReadyGuard`] that must be dropped to resume write-readiness /// polling. /// /// 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`]. #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering. pubasyncfn writable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> { self.ready(Interest::WRITABLE).await
}
/// Waits for the file descriptor to become writable, returning a /// [`AsyncFdReadyMutGuard`] that must be dropped to resume write-readiness /// polling. /// /// This method takes `&mut self`, so it is possible to access the inner IO /// resource mutably when handling the [`AsyncFdReadyMutGuard`]. #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering. pubasyncfn writable_mut<'a>(&'a mutself) -> io::Result<AsyncFdReadyMutGuard<'a, T>> { self.ready_mut(Interest::WRITABLE).await
}
/// Reads or writes from the file descriptor using a user-provided IO operation. /// /// The `async_io` method is a convenience utility that waits for the file /// descriptor to become ready, and then executes the provided IO operation. /// Since file descriptors may be marked ready spuriously, the closure will /// be called repeatedly until it returns something other than a /// [`WouldBlock`] error. This is done using the following loop: /// /// ```no_run /// # use std::io::{self, Result}; /// # struct Dox<T> { inner: T } /// # impl<T> Dox<T> { /// # async fn writable(&self) -> Result<&Self> { /// # Ok(self) /// # } /// # fn try_io<R>(&self, _: impl FnMut(&T) -> Result<R>) -> Result<Result<R>> { /// # panic!() /// # } /// async fn async_io<R>(&self, mut f: impl FnMut(&T) -> io::Result<R>) -> io::Result<R> { /// loop { /// // or `readable` if called with the read interest. /// let guard = self.writable().await?; /// /// match guard.try_io(&mut f) { /// Ok(result) => return result, /// Err(_would_block) => continue, /// } /// } /// } /// # } /// ``` /// /// The closure should only return a [`WouldBlock`] error if it has performed /// an IO operation on the file descriptor that failed due to the file descriptor not being /// ready. Returning a [`WouldBlock`] error in any other situation will /// incorrectly clear the readiness flag, which can cause the file descriptor to /// behave incorrectly. /// /// The closure should not perform the IO operation using any of the methods /// defined on the Tokio [`AsyncFd`] type, as this will mess with the /// readiness flag and can cause the file descriptor to behave incorrectly. /// /// This method is not intended to be used with combined interests. /// The closure should perform only one type of IO operation, so it should not /// require more than one ready state. This method may panic or sleep forever /// if it is called with a combined interest. /// /// # Examples /// /// This example sends some bytes on the inner [`std::net::UdpSocket`]. The `async_io` /// method waits for readiness, and retries if the send operation does block. This example /// is equivalent to the one given for [`try_io`]. /// /// ```no_run /// use tokio::io::{Interest, unix::AsyncFd}; /// /// use std::io; /// use std::net::UdpSocket; /// /// #[tokio::main] /// async fn main() -> io::Result<()> { /// let socket = UdpSocket::bind("0.0.0.0:8080")?; /// socket.set_nonblocking(true)?; /// let async_fd = AsyncFd::new(socket)?; /// /// let written = async_fd /// .async_io(Interest::WRITABLE, |inner| inner.send(&[1, 2])) /// .await?; /// /// println!("wrote {written} bytes"); /// /// Ok(()) /// } /// ``` /// /// [`try_io`]: AsyncFdReadyGuard::try_io /// [`WouldBlock`]: std::io::ErrorKind::WouldBlock pubasyncfn async_io<R>(
&self,
interest: Interest, mut f: impl FnMut(&T) -> io::Result<R>,
) -> io::Result<R> { self.registration
.async_io(interest, || f(self.get_ref()))
.await
}
/// Reads or writes from the file descriptor using a user-provided IO operation. /// /// The behavior is the same as [`async_io`], except that the closure can mutate the inner /// value of the [`AsyncFd`]. /// /// [`async_io`]: AsyncFd::async_io pubasyncfn async_io_mut<R>(
&mutself,
interest: Interest, mut f: impl FnMut(&mut T) -> io::Result<R>,
) -> io::Result<R> { self.registration
.async_io(interest, || f(self.inner.as_mut().unwrap()))
.await
}
}
impl<T: AsRawFd> Drop for AsyncFd<T> { fn drop(&mutself) { let _ = self.take_inner();
}
}
impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> { /// Indicates to tokio that the file descriptor is no longer ready. All /// 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. /// /// This method only clears readiness events that happened before the creation of this guard. /// In other words, if the IO resource becomes ready between the creation of the guard and /// this call to `clear_ready`, then the readiness is not actually cleared. pubfn clear_ready(&mutself) { iflet 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`. /// /// This method only clears readiness events that happened before the creation of this guard. /// In other words, if the IO resource becomes ready between the creation of the guard and /// this call to `clear_ready`, then the readiness is not actually cleared. /// /// # 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()); /// } /// } /// } /// } /// } /// ``` pubfn clear_ready_matching(&mutself, ready: Ready) { iflet 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. /// /// While this function is itself a no-op, it satisfies the `#[must_use]` /// constraint on the [`AsyncFdReadyGuard`] type. pubfn retain_ready(&mutself) { // 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 pubfn 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 /// 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 /// clearing the tokio-side state only when a [`WouldBlock`] condition /// occurs. It is the responsibility of the caller to ensure that `f` /// returns [`WouldBlock`] only if the file descriptor that originated this /// `AsyncFdReadyGuard` no longer expresses the readiness state that was queried to /// create this `AsyncFdReadyGuard`. /// /// # Examples /// /// This example sends some bytes to the inner [`std::net::UdpSocket`]. Waiting /// for write-readiness and retrying when the send operation does block are explicit. /// This example can be written more succinctly using [`AsyncFd::async_io`]. /// /// ```no_run /// use tokio::io::unix::AsyncFd; /// /// use std::io; /// use std::net::UdpSocket; /// /// #[tokio::main] /// async fn main() -> io::Result<()> { /// let socket = UdpSocket::bind("0.0.0.0:8080")?; /// socket.set_nonblocking(true)?; /// let async_fd = AsyncFd::new(socket)?; /// /// let written = loop { /// let mut guard = async_fd.writable().await?; /// match guard.try_io(|inner| inner.get_ref().send(&[1, 2])) { /// Ok(result) => { /// break result?; /// } /// Err(_would_block) => { /// // try_io already cleared the file descriptor's readiness state /// continue; /// } /// } /// }; /// /// println!("wrote {written} bytes"); /// /// Ok(()) /// } /// ``` /// /// [`WouldBlock`]: std::io::ErrorKind::WouldBlock // Alias for old name in 0.x #[cfg_attr(docsrs, doc(alias = "with_io"))] pubfn try_io<R>(
&mutself,
f: impl FnOnce(&'a AsyncFd<Inner>) -> io::Result<R>,
) -> Result<io::Result<R>, TryIoError> { let result = f(self.async_fd);
match result {
Err(err) if err.kind() == io::ErrorKind::WouldBlock => { self.clear_ready();
Err(TryIoError(()))
}
result => Ok(result),
}
}
/// Returns a shared reference to the inner [`AsyncFd`]. pubfn get_ref(&self) -> &'a AsyncFd<Inner> { self.async_fd
}
/// Returns a shared reference to the backing object of the inner [`AsyncFd`]. pubfn get_inner(&self) -> &'a Inner { self.get_ref().get_ref()
}
}
impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> { /// Indicates to tokio that the file descriptor is no longer ready. All /// 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. /// /// This method only clears readiness events that happened before the creation of this guard. /// In other words, if the IO resource becomes ready between the creation of the guard and /// this call to `clear_ready`, then the readiness is not actually cleared. pubfn clear_ready(&mutself) { iflet 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`. /// /// This method only clears readiness events that happened before the creation of this guard. /// In other words, if the IO resource becomes ready between the creation of the guard and /// this call to `clear_ready`, then the readiness is not actually cleared. /// /// # 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()); /// } /// } /// } /// } /// } /// ``` pubfn clear_ready_matching(&mutself, ready: Ready) { iflet 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. /// /// While this function is itself a no-op, it satisfies the `#[must_use]` /// constraint on the [`AsyncFdReadyGuard`] type. pubfn retain_ready(&mutself) { // 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 pubfn 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 /// 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 /// clearing the tokio-side state only when a [`WouldBlock`] condition /// occurs. It is the responsibility of the caller to ensure that `f` /// returns [`WouldBlock`] only if the file descriptor that originated this /// `AsyncFdReadyGuard` no longer expresses the readiness state that was queried to /// create this `AsyncFdReadyGuard`. /// /// [`WouldBlock`]: std::io::ErrorKind::WouldBlock pubfn try_io<R>(
&mutself,
f: impl FnOnce(&mut AsyncFd<Inner>) -> io::Result<R>,
) -> Result<io::Result<R>, TryIoError> { let result = f(self.async_fd);
match result {
Err(err) if err.kind() == io::ErrorKind::WouldBlock => { self.clear_ready();
Err(TryIoError(()))
}
result => Ok(result),
}
}
/// Returns a shared reference to the inner [`AsyncFd`]. pubfn get_ref(&self) -> &AsyncFd<Inner> { self.async_fd
}
/// Returns a mutable reference to the inner [`AsyncFd`]. pubfn get_mut(&mutself) -> &mut AsyncFd<Inner> { self.async_fd
}
/// Returns a shared reference to the backing object of the inner [`AsyncFd`]. pubfn get_inner(&self) -> &Inner { self.get_ref().get_ref()
}
/// Returns a mutable reference to the backing object of the inner [`AsyncFd`]. pubfn get_inner_mut(&mutself) -> &mut Inner { self.get_mut().get_mut()
}
}
/// 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)] pubstruct TryIoError(());
/// Error returned by [`try_new`] or [`try_with_interest`]. /// /// [`try_new`]: AsyncFd::try_new /// [`try_with_interest`]: AsyncFd::try_with_interest pubstruct AsyncFdTryNewError<T> {
inner: T,
cause: io::Error,
}
impl<T> AsyncFdTryNewError<T> { /// Returns the original object passed to [`try_new`] or [`try_with_interest`] /// alongside the error that caused these functions to fail. /// /// [`try_new`]: AsyncFd::try_new /// [`try_with_interest`]: AsyncFd::try_with_interest pubfn into_parts(self) -> (T, io::Error) {
(self.inner, self.cause)
}
}
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.