tokio/io/async_fd.rs
1use crate::io::{Interest, Ready};
2use crate::runtime::io::{ReadyEvent, Registration};
3use crate::runtime::scheduler;
4
5use mio::unix::SourceFd;
6use std::error::Error;
7use std::fmt;
8use std::io;
9use std::os::unix::io::{AsRawFd, RawFd};
10use std::task::{ready, Context, Poll};
11
12/// Associates an IO object backed by a Unix file descriptor with the tokio
13/// reactor, allowing for readiness to be polled. The file descriptor must be of
14/// a type that can be used with the OS polling facilities (ie, `poll`, `epoll`,
15/// `kqueue`, etc), such as a network socket or pipe, and the file descriptor
16/// must have the nonblocking mode set to true.
17///
18/// Creating an [`AsyncFd`] registers the file descriptor with the current tokio
19/// Reactor, allowing you to directly await the file descriptor being readable
20/// or writable. Once registered, the file descriptor remains registered until
21/// the [`AsyncFd`] is dropped.
22///
23/// The [`AsyncFd`] takes ownership of an arbitrary object to represent the IO
24/// object. It is intended that the inner object will handle closing the file
25/// descriptor when it is dropped, avoiding resource leaks and ensuring that the
26/// [`AsyncFd`] can clean up the registration before closing the file descriptor.
27/// The [`AsyncFd::into_inner`] function can be used to extract the inner object
28/// to retake control from the tokio IO reactor. The [`OwnedFd`] type is often
29/// used as the inner object, as it is the simplest type that closes the fd on
30/// drop.
31///
32/// The inner object is required to implement [`AsRawFd`]. This file descriptor
33/// must remain open and must not change while [`AsyncFd`] owns the inner
34/// object, i.e. the [`AsRawFd::as_raw_fd`] method on the inner type must always
35/// return the same valid file descriptor referring to the same underlying file
36/// description when called multiple times. Violating this requirement leads to
37/// undefined behavior.
38///
39/// Polling for readiness is done by calling the async functions [`readable`]
40/// and [`writable`]. These functions complete when the associated readiness
41/// condition is observed. Any number of tasks can query the same `AsyncFd` in
42/// parallel, on the same or different conditions.
43///
44/// On some platforms, the readiness detecting mechanism relies on
45/// edge-triggered notifications. This means that the OS will only notify Tokio
46/// when the file descriptor transitions from not-ready to ready. For this to
47/// work you should first try to read or write and only poll for readiness
48/// if that fails with an error of [`std::io::ErrorKind::WouldBlock`].
49///
50/// Tokio internally tracks when it has received a ready notification, and when
51/// readiness checking functions like [`readable`] and [`writable`] are called,
52/// if the readiness flag is set, these async functions will complete
53/// immediately. This however does mean that it is critical to ensure that this
54/// ready flag is cleared when (and only when) the file descriptor ceases to be
55/// ready. The [`AsyncFdReadyGuard`] returned from readiness checking functions
56/// serves this function; after calling a readiness-checking async function,
57/// you must use this [`AsyncFdReadyGuard`] to signal to tokio whether the file
58/// descriptor is no longer in a ready state.
59///
60/// ## Use with to a poll-based API
61///
62/// In some cases it may be desirable to use `AsyncFd` from APIs similar to
63/// [`TcpStream::poll_read_ready`]. The [`AsyncFd::poll_read_ready`] and
64/// [`AsyncFd::poll_write_ready`] functions are provided for this purpose.
65/// Because these functions don't create a future to hold their state, they have
66/// the limitation that only one task can wait on each direction (read or write)
67/// at a time.
68///
69/// # Examples
70///
71/// This example shows how to turn [`std::net::TcpStream`] asynchronous using
72/// `AsyncFd`. It implements the read/write operations both as an `async fn`
73/// and using the IO traits [`AsyncRead`] and [`AsyncWrite`].
74///
75/// ```no_run
76/// use std::io::{self, Read, Write};
77/// use std::net::TcpStream;
78/// use std::pin::Pin;
79/// use std::task::{ready, Context, Poll};
80/// use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
81/// use tokio::io::unix::AsyncFd;
82///
83/// pub struct AsyncTcpStream {
84/// inner: AsyncFd<TcpStream>,
85/// }
86///
87/// impl AsyncTcpStream {
88/// pub fn new(tcp: TcpStream) -> io::Result<Self> {
89/// tcp.set_nonblocking(true)?;
90/// // SAFETY: `TcpStream` owns a valid file descriptor and we do not
91/// // replace or close it while owned by `AsyncFd`.
92/// Ok(Self {
93/// inner: unsafe { AsyncFd::register(tcp)? },
94/// })
95/// }
96///
97/// pub async fn read(&self, out: &mut [u8]) -> io::Result<usize> {
98/// loop {
99/// let mut guard = self.inner.readable().await?;
100///
101/// match guard.try_io(|inner| inner.get_ref().read(out)) {
102/// Ok(result) => return result,
103/// Err(_would_block) => continue,
104/// }
105/// }
106/// }
107///
108/// pub async fn write(&self, buf: &[u8]) -> io::Result<usize> {
109/// loop {
110/// let mut guard = self.inner.writable().await?;
111///
112/// match guard.try_io(|inner| inner.get_ref().write(buf)) {
113/// Ok(result) => return result,
114/// Err(_would_block) => continue,
115/// }
116/// }
117/// }
118/// }
119///
120/// impl AsyncRead for AsyncTcpStream {
121/// fn poll_read(
122/// self: Pin<&mut Self>,
123/// cx: &mut Context<'_>,
124/// buf: &mut ReadBuf<'_>
125/// ) -> Poll<io::Result<()>> {
126/// loop {
127/// let mut guard = ready!(self.inner.poll_read_ready(cx))?;
128///
129/// let unfilled = buf.initialize_unfilled();
130/// match guard.try_io(|inner| inner.get_ref().read(unfilled)) {
131/// Ok(Ok(len)) => {
132/// buf.advance(len);
133/// return Poll::Ready(Ok(()));
134/// },
135/// Ok(Err(err)) => return Poll::Ready(Err(err)),
136/// Err(_would_block) => continue,
137/// }
138/// }
139/// }
140/// }
141///
142/// impl AsyncWrite for AsyncTcpStream {
143/// fn poll_write(
144/// self: Pin<&mut Self>,
145/// cx: &mut Context<'_>,
146/// buf: &[u8]
147/// ) -> Poll<io::Result<usize>> {
148/// loop {
149/// let mut guard = ready!(self.inner.poll_write_ready(cx))?;
150///
151/// match guard.try_io(|inner| inner.get_ref().write(buf)) {
152/// Ok(result) => return Poll::Ready(result),
153/// Err(_would_block) => continue,
154/// }
155/// }
156/// }
157///
158/// fn poll_flush(
159/// self: Pin<&mut Self>,
160/// cx: &mut Context<'_>,
161/// ) -> Poll<io::Result<()>> {
162/// // tcp flush is a no-op
163/// Poll::Ready(Ok(()))
164/// }
165///
166/// fn poll_shutdown(
167/// self: Pin<&mut Self>,
168/// cx: &mut Context<'_>,
169/// ) -> Poll<io::Result<()>> {
170/// self.inner.get_ref().shutdown(std::net::Shutdown::Write)?;
171/// Poll::Ready(Ok(()))
172/// }
173/// }
174/// ```
175///
176/// [`readable`]: method@Self::readable
177/// [`writable`]: method@Self::writable
178/// [`AsyncFdReadyGuard`]: struct@self::AsyncFdReadyGuard
179/// [`TcpStream::poll_read_ready`]: struct@crate::net::TcpStream
180/// [`AsyncRead`]: trait@crate::io::AsyncRead
181/// [`AsyncWrite`]: trait@crate::io::AsyncWrite
182/// [`OwnedFd`]: struct@std::os::fd::OwnedFd
183pub struct AsyncFd<T: AsRawFd> {
184 registration: Registration,
185 // The inner value is always present. the Option is required for `drop` and `into_inner`.
186 // In all other methods `unwrap` is valid, and will never panic.
187 inner: Option<T>,
188}
189
190/// Represents an IO-ready event detected on a particular file descriptor that
191/// has not yet been acknowledged. This is a `must_use` structure to help ensure
192/// that you do not forget to explicitly clear (or not clear) the event.
193///
194/// This type exposes an immutable reference to the underlying IO object.
195#[must_use = "You must explicitly choose whether to clear the readiness state by calling a method on ReadyGuard"]
196pub struct AsyncFdReadyGuard<'a, T: AsRawFd> {
197 async_fd: &'a AsyncFd<T>,
198 event: Option<ReadyEvent>,
199}
200
201/// Represents an IO-ready event detected on a particular file descriptor that
202/// has not yet been acknowledged. This is a `must_use` structure to help ensure
203/// that you do not forget to explicitly clear (or not clear) the event.
204///
205/// This type exposes a mutable reference to the underlying IO object.
206#[must_use = "You must explicitly choose whether to clear the readiness state by calling a method on ReadyGuard"]
207pub struct AsyncFdReadyMutGuard<'a, T: AsRawFd> {
208 async_fd: &'a mut AsyncFd<T>,
209 event: Option<ReadyEvent>,
210}
211
212impl<T: AsRawFd> AsyncFd<T> {
213 /// Creates an `AsyncFd` backed by (and taking ownership of) an object
214 /// implementing [`AsRawFd`].
215 ///
216 /// Only configures the [`Interest::READABLE`] and [`Interest::WRITABLE`] interests. For more
217 /// control, use [`AsyncFd::register_with_interest`].
218 ///
219 /// This method must be called in the context of a tokio runtime.
220 ///
221 /// In the case of failure, it returns an [`AsyncFdRegisterError`] that contains the original
222 /// object passed to this function.
223 ///
224 /// # Safety
225 ///
226 /// The file descriptor returned by [`inner.as_raw_fd()`](AsRawFd::as_raw_fd) must be a valid,
227 /// open file descriptor, and it must remain open and refer to the same underlying file
228 /// description for the entire lifetime of the `AsyncFd` (i.e., until the `AsyncFd` is
229 /// dropped, or [`AsyncFd::into_inner`] is called, or the `AsyncFd` is
230 /// [forgotten](std::mem::forget)). While owned by this `AsyncFd`, every call to
231 /// [`inner.as_raw_fd()`](AsRawFd::as_raw_fd) must return that same file descriptor.
232 ///
233 /// # Panics
234 ///
235 /// This function panics if there is no current reactor set, or if the `rt`
236 /// feature flag is not enabled.
237 #[inline]
238 #[track_caller]
239 pub unsafe fn register(inner: T) -> Result<Self, AsyncFdRegisterError<T>> {
240 // SAFETY: Upheld by the caller.
241 unsafe { Self::register_with_interest(inner, Interest::READABLE | Interest::WRITABLE) }
242 }
243
244 /// Creates an `AsyncFd` backed by (and taking ownership of) an object
245 /// implementing [`AsRawFd`], with a specific [`Interest`].
246 ///
247 /// This method must be called in the context of a tokio runtime.
248 ///
249 /// In the case of failure, it returns an [`AsyncFdRegisterError`] that contains the original
250 /// object passed to this function.
251 ///
252 /// # Safety
253 ///
254 /// The file descriptor returned by [`inner.as_raw_fd()`](AsRawFd::as_raw_fd) must be a valid,
255 /// open file descriptor, and it must remain open and refer to the same underlying file
256 /// description for the entire lifetime of the `AsyncFd` (i.e., until the `AsyncFd` is
257 /// dropped, or [`AsyncFd::into_inner`] is called, or the `AsyncFd` is
258 /// [forgotten](std::mem::forget)). While owned by this `AsyncFd`, every call to
259 /// [`inner.as_raw_fd()`](AsRawFd::as_raw_fd) must return that same file descriptor.
260 ///
261 /// # Panics
262 ///
263 /// This function panics if there is no current reactor set, or if the `rt`
264 /// feature flag is not enabled.
265 #[inline]
266 #[track_caller]
267 pub unsafe fn register_with_interest(
268 inner: T,
269 interest: Interest,
270 ) -> Result<Self, AsyncFdRegisterError<T>> {
271 let fd = inner.as_raw_fd();
272
273 match Registration::new_with_interest_and_handle(
274 &mut SourceFd(&fd),
275 interest,
276 scheduler::Handle::current(),
277 ) {
278 Ok(registration) => Ok(AsyncFd {
279 registration,
280 inner: Some(inner),
281 }),
282 Err(cause) => Err(AsyncFdRegisterError { inner, cause }),
283 }
284 }
285
286 /// Returns a shared reference to the backing object of this [`AsyncFd`].
287 #[inline]
288 pub fn get_ref(&self) -> &T {
289 self.inner.as_ref().unwrap()
290 }
291
292 /// Returns a mutable reference to the backing object of this `AsyncFd`.
293 ///
294 /// This may be used to bypass the readiness of the IO resource, so that you
295 /// can read or write to it even if the fd is not ready. Be aware that this
296 /// method must not be used to exchange the underlying IO resource for a
297 /// different one. Doing so is a violation of the safety requirements on
298 /// [`AsyncFd::register`].
299 #[inline]
300 pub fn get_mut(&mut self) -> &mut T {
301 self.inner.as_mut().unwrap()
302 }
303
304 fn take_inner(&mut self) -> Option<T> {
305 let inner = self.inner.take()?;
306 let fd = inner.as_raw_fd();
307
308 let _ = self.registration.deregister(&mut SourceFd(&fd));
309
310 Some(inner)
311 }
312
313 /// Deregisters this file descriptor and returns ownership of the backing
314 /// object.
315 pub fn into_inner(mut self) -> T {
316 self.take_inner().unwrap()
317 }
318
319 /// Polls for read readiness.
320 ///
321 /// If the file descriptor is not currently ready for reading, this method
322 /// will store a clone of the [`Waker`] from the provided [`Context`]. When the
323 /// file descriptor becomes ready for reading, [`Waker::wake`] will be called.
324 ///
325 /// Note that on multiple calls to [`poll_read_ready`] or
326 /// [`poll_read_ready_mut`], only the `Waker` from the `Context` passed to the
327 /// most recent call is scheduled to receive a wakeup. (However,
328 /// [`poll_write_ready`] retains a second, independent waker).
329 ///
330 /// This method is intended for cases where creating and pinning a future
331 /// via [`readable`] is not feasible. Where possible, using [`readable`] is
332 /// preferred, as this supports polling from multiple tasks at once.
333 ///
334 /// This method takes `&self`, so it is possible to call this method
335 /// concurrently with other methods on this struct. This method only
336 /// provides shared access to the inner IO resource when handling the
337 /// [`AsyncFdReadyGuard`].
338 ///
339 /// [`poll_read_ready`]: method@Self::poll_read_ready
340 /// [`poll_read_ready_mut`]: method@Self::poll_read_ready_mut
341 /// [`poll_write_ready`]: method@Self::poll_write_ready
342 /// [`readable`]: method@Self::readable
343 /// [`Context`]: struct@std::task::Context
344 /// [`Waker`]: struct@std::task::Waker
345 /// [`Waker::wake`]: method@std::task::Waker::wake
346 pub fn poll_read_ready<'a>(
347 &'a self,
348 cx: &mut Context<'_>,
349 ) -> Poll<io::Result<AsyncFdReadyGuard<'a, T>>> {
350 let event = ready!(self.registration.poll_read_ready(cx))?;
351
352 Poll::Ready(Ok(AsyncFdReadyGuard {
353 async_fd: self,
354 event: Some(event),
355 }))
356 }
357
358 /// Polls for read readiness.
359 ///
360 /// If the file descriptor is not currently ready for reading, this method
361 /// will store a clone of the [`Waker`] from the provided [`Context`]. When the
362 /// file descriptor becomes ready for reading, [`Waker::wake`] will be called.
363 ///
364 /// Note that on multiple calls to [`poll_read_ready`] or
365 /// [`poll_read_ready_mut`], only the `Waker` from the `Context` passed to the
366 /// most recent call is scheduled to receive a wakeup. (However,
367 /// [`poll_write_ready`] retains a second, independent waker).
368 ///
369 /// This method is intended for cases where creating and pinning a future
370 /// via [`readable`] is not feasible. Where possible, using [`readable`] is
371 /// preferred, as this supports polling from multiple tasks at once.
372 ///
373 /// This method takes `&mut self`, so it is possible to access the inner IO
374 /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
375 ///
376 /// [`poll_read_ready`]: method@Self::poll_read_ready
377 /// [`poll_read_ready_mut`]: method@Self::poll_read_ready_mut
378 /// [`poll_write_ready`]: method@Self::poll_write_ready
379 /// [`readable`]: method@Self::readable
380 /// [`Context`]: struct@std::task::Context
381 /// [`Waker`]: struct@std::task::Waker
382 /// [`Waker::wake`]: method@std::task::Waker::wake
383 pub fn poll_read_ready_mut<'a>(
384 &'a mut self,
385 cx: &mut Context<'_>,
386 ) -> Poll<io::Result<AsyncFdReadyMutGuard<'a, T>>> {
387 let event = ready!(self.registration.poll_read_ready(cx))?;
388
389 Poll::Ready(Ok(AsyncFdReadyMutGuard {
390 async_fd: self,
391 event: Some(event),
392 }))
393 }
394
395 /// Polls for write readiness.
396 ///
397 /// If the file descriptor is not currently ready for writing, this method
398 /// will store a clone of the [`Waker`] from the provided [`Context`]. When the
399 /// file descriptor becomes ready for writing, [`Waker::wake`] will be called.
400 ///
401 /// Note that on multiple calls to [`poll_write_ready`] or
402 /// [`poll_write_ready_mut`], only the `Waker` from the `Context` passed to the
403 /// most recent call is scheduled to receive a wakeup. (However,
404 /// [`poll_read_ready`] retains a second, independent waker).
405 ///
406 /// This method is intended for cases where creating and pinning a future
407 /// via [`writable`] is not feasible. Where possible, using [`writable`] is
408 /// preferred, as this supports polling from multiple tasks at once.
409 ///
410 /// This method takes `&self`, so it is possible to call this method
411 /// concurrently with other methods on this struct. This method only
412 /// provides shared access to the inner IO resource when handling the
413 /// [`AsyncFdReadyGuard`].
414 ///
415 /// [`poll_read_ready`]: method@Self::poll_read_ready
416 /// [`poll_write_ready`]: method@Self::poll_write_ready
417 /// [`poll_write_ready_mut`]: method@Self::poll_write_ready_mut
418 /// [`writable`]: method@Self::readable
419 /// [`Context`]: struct@std::task::Context
420 /// [`Waker`]: struct@std::task::Waker
421 /// [`Waker::wake`]: method@std::task::Waker::wake
422 pub fn poll_write_ready<'a>(
423 &'a self,
424 cx: &mut Context<'_>,
425 ) -> Poll<io::Result<AsyncFdReadyGuard<'a, T>>> {
426 let event = ready!(self.registration.poll_write_ready(cx))?;
427
428 Poll::Ready(Ok(AsyncFdReadyGuard {
429 async_fd: self,
430 event: Some(event),
431 }))
432 }
433
434 /// Polls for write readiness.
435 ///
436 /// If the file descriptor is not currently ready for writing, this method
437 /// will store a clone of the [`Waker`] from the provided [`Context`]. When the
438 /// file descriptor becomes ready for writing, [`Waker::wake`] will be called.
439 ///
440 /// Note that on multiple calls to [`poll_write_ready`] or
441 /// [`poll_write_ready_mut`], only the `Waker` from the `Context` passed to the
442 /// most recent call is scheduled to receive a wakeup. (However,
443 /// [`poll_read_ready`] retains a second, independent waker).
444 ///
445 /// This method is intended for cases where creating and pinning a future
446 /// via [`writable`] is not feasible. Where possible, using [`writable`] is
447 /// preferred, as this supports polling from multiple tasks at once.
448 ///
449 /// This method takes `&mut self`, so it is possible to access the inner IO
450 /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
451 ///
452 /// [`poll_read_ready`]: method@Self::poll_read_ready
453 /// [`poll_write_ready`]: method@Self::poll_write_ready
454 /// [`poll_write_ready_mut`]: method@Self::poll_write_ready_mut
455 /// [`writable`]: method@Self::readable
456 /// [`Context`]: struct@std::task::Context
457 /// [`Waker`]: struct@std::task::Waker
458 /// [`Waker::wake`]: method@std::task::Waker::wake
459 pub fn poll_write_ready_mut<'a>(
460 &'a mut self,
461 cx: &mut Context<'_>,
462 ) -> Poll<io::Result<AsyncFdReadyMutGuard<'a, T>>> {
463 let event = ready!(self.registration.poll_write_ready(cx))?;
464
465 Poll::Ready(Ok(AsyncFdReadyMutGuard {
466 async_fd: self,
467 event: Some(event),
468 }))
469 }
470
471 /// Waits for any of the requested ready states, returning a
472 /// [`AsyncFdReadyGuard`] that must be dropped to resume
473 /// polling for the requested ready states.
474 ///
475 /// The function may complete without the file descriptor being ready. This is a
476 /// false-positive and attempting an operation will return with
477 /// `io::ErrorKind::WouldBlock`. The function can also return with an empty
478 /// [`Ready`] set, so you should always check the returned value and possibly
479 /// wait again if the requested states are not set.
480 ///
481 /// When an IO operation does return `io::ErrorKind::WouldBlock`, the readiness must be cleared.
482 /// When a combined interest is used, it is important to clear only the readiness
483 /// that is actually observed to block. For instance when the combined
484 /// interest `Interest::READABLE | Interest::WRITABLE` is used, and a read blocks, only
485 /// read readiness should be cleared using the [`AsyncFdReadyGuard::clear_ready_matching`] method:
486 /// `guard.clear_ready_matching(Ready::READABLE)`.
487 /// Also clearing the write readiness in this case would be incorrect. The [`AsyncFdReadyGuard::clear_ready`]
488 /// method clears all readiness flags.
489 ///
490 /// This method takes `&self`, so it is possible to call this method
491 /// concurrently with other methods on this struct. This method only
492 /// provides shared access to the inner IO resource when handling the
493 /// [`AsyncFdReadyGuard`].
494 ///
495 /// # Examples
496 ///
497 /// Concurrently read and write to a [`std::net::TcpStream`] on the same task without
498 /// splitting.
499 ///
500 /// ```no_run
501 /// use std::error::Error;
502 /// use std::io;
503 /// use std::io::{Read, Write};
504 /// use std::net::TcpStream;
505 /// use tokio::io::unix::AsyncFd;
506 /// use tokio::io::{Interest, Ready};
507 ///
508 /// #[tokio::main]
509 /// async fn main() -> Result<(), Box<dyn Error>> {
510 /// let stream = TcpStream::connect("127.0.0.1:8080")?;
511 /// stream.set_nonblocking(true)?;
512 /// // SAFETY: `TcpStream` owns a valid file descriptor and we do not
513 /// // replace or close it while owned by `AsyncFd`.
514 /// let stream = unsafe { AsyncFd::register(stream)? };
515 ///
516 /// loop {
517 /// let mut guard = stream
518 /// .ready(Interest::READABLE | Interest::WRITABLE)
519 /// .await?;
520 ///
521 /// if guard.ready().is_readable() {
522 /// let mut data = vec![0; 1024];
523 /// // Try to read data, this may still fail with `WouldBlock`
524 /// // if the readiness event is a false positive.
525 /// match stream.get_ref().read(&mut data) {
526 /// Ok(n) => {
527 /// println!("read {} bytes", n);
528 /// }
529 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
530 /// // a read has blocked, but a write might still succeed.
531 /// // clear only the read readiness.
532 /// guard.clear_ready_matching(Ready::READABLE);
533 /// continue;
534 /// }
535 /// Err(e) => {
536 /// return Err(e.into());
537 /// }
538 /// }
539 /// }
540 ///
541 /// if guard.ready().is_writable() {
542 /// // Try to write data, this may still fail with `WouldBlock`
543 /// // if the readiness event is a false positive.
544 /// match stream.get_ref().write(b"hello world") {
545 /// Ok(n) => {
546 /// println!("write {} bytes", n);
547 /// }
548 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
549 /// // a write has blocked, but a read might still succeed.
550 /// // clear only the write readiness.
551 /// guard.clear_ready_matching(Ready::WRITABLE);
552 /// continue;
553 /// }
554 /// Err(e) => {
555 /// return Err(e.into());
556 /// }
557 /// }
558 /// }
559 /// }
560 /// }
561 /// ```
562 pub async fn ready(&self, interest: Interest) -> io::Result<AsyncFdReadyGuard<'_, T>> {
563 let event = self.registration.readiness(interest).await?;
564
565 Ok(AsyncFdReadyGuard {
566 async_fd: self,
567 event: Some(event),
568 })
569 }
570
571 /// Waits for any of the requested ready states, returning a
572 /// [`AsyncFdReadyMutGuard`] that must be dropped to resume
573 /// polling for the requested ready states.
574 ///
575 /// The function may complete without the file descriptor being ready. This is a
576 /// false-positive and attempting an operation will return with
577 /// `io::ErrorKind::WouldBlock`. The function can also return with an empty
578 /// [`Ready`] set, so you should always check the returned value and possibly
579 /// wait again if the requested states are not set.
580 ///
581 /// When an IO operation does return `io::ErrorKind::WouldBlock`, the readiness must be cleared.
582 /// When a combined interest is used, it is important to clear only the readiness
583 /// that is actually observed to block. For instance when the combined
584 /// interest `Interest::READABLE | Interest::WRITABLE` is used, and a read blocks, only
585 /// read readiness should be cleared using the [`AsyncFdReadyMutGuard::clear_ready_matching`] method:
586 /// `guard.clear_ready_matching(Ready::READABLE)`.
587 /// Also clearing the write readiness in this case would be incorrect.
588 /// The [`AsyncFdReadyMutGuard::clear_ready`] method clears all readiness flags.
589 ///
590 /// This method takes `&mut self`, so it is possible to access the inner IO
591 /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
592 ///
593 /// # Examples
594 ///
595 /// Concurrently read and write to a [`std::net::TcpStream`] on the same task without
596 /// splitting.
597 ///
598 /// ```no_run
599 /// use std::error::Error;
600 /// use std::io;
601 /// use std::io::{Read, Write};
602 /// use std::net::TcpStream;
603 /// use tokio::io::unix::AsyncFd;
604 /// use tokio::io::{Interest, Ready};
605 ///
606 /// #[tokio::main]
607 /// async fn main() -> Result<(), Box<dyn Error>> {
608 /// let stream = TcpStream::connect("127.0.0.1:8080")?;
609 /// stream.set_nonblocking(true)?;
610 /// // SAFETY: `TcpStream` owns a valid file descriptor and we do not
611 /// // replace or close it while owned by `AsyncFd`.
612 /// let mut stream = unsafe { AsyncFd::register(stream)? };
613 ///
614 /// loop {
615 /// let mut guard = stream
616 /// .ready_mut(Interest::READABLE | Interest::WRITABLE)
617 /// .await?;
618 ///
619 /// if guard.ready().is_readable() {
620 /// let mut data = vec![0; 1024];
621 /// // Try to read data, this may still fail with `WouldBlock`
622 /// // if the readiness event is a false positive.
623 /// match guard.get_inner_mut().read(&mut data) {
624 /// Ok(n) => {
625 /// println!("read {} bytes", n);
626 /// }
627 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
628 /// // a read has blocked, but a write might still succeed.
629 /// // clear only the read readiness.
630 /// guard.clear_ready_matching(Ready::READABLE);
631 /// continue;
632 /// }
633 /// Err(e) => {
634 /// return Err(e.into());
635 /// }
636 /// }
637 /// }
638 ///
639 /// if guard.ready().is_writable() {
640 /// // Try to write data, this may still fail with `WouldBlock`
641 /// // if the readiness event is a false positive.
642 /// match guard.get_inner_mut().write(b"hello world") {
643 /// Ok(n) => {
644 /// println!("write {} bytes", n);
645 /// }
646 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
647 /// // a write has blocked, but a read might still succeed.
648 /// // clear only the write readiness.
649 /// guard.clear_ready_matching(Ready::WRITABLE);
650 /// continue;
651 /// }
652 /// Err(e) => {
653 /// return Err(e.into());
654 /// }
655 /// }
656 /// }
657 /// }
658 /// }
659 /// ```
660 pub async fn ready_mut(
661 &mut self,
662 interest: Interest,
663 ) -> io::Result<AsyncFdReadyMutGuard<'_, T>> {
664 let event = self.registration.readiness(interest).await?;
665
666 Ok(AsyncFdReadyMutGuard {
667 async_fd: self,
668 event: Some(event),
669 })
670 }
671
672 /// Waits for the file descriptor to become readable, returning a
673 /// [`AsyncFdReadyGuard`] that must be dropped to resume read-readiness
674 /// polling.
675 ///
676 /// This method takes `&self`, so it is possible to call this method
677 /// concurrently with other methods on this struct. This method only
678 /// provides shared access to the inner IO resource when handling the
679 /// [`AsyncFdReadyGuard`].
680 ///
681 /// # Cancel safety
682 ///
683 /// This method is cancel safe. Once a readiness event occurs, the method
684 /// will continue to return immediately until the readiness event is
685 /// consumed by an attempt to read or write that fails with `WouldBlock` or
686 /// `Poll::Pending`.
687 #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
688 pub async fn readable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> {
689 self.ready(Interest::READABLE).await
690 }
691
692 /// Waits for the file descriptor to become readable, returning a
693 /// [`AsyncFdReadyMutGuard`] that must be dropped to resume read-readiness
694 /// polling.
695 ///
696 /// This method takes `&mut self`, so it is possible to access the inner IO
697 /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
698 ///
699 /// # Cancel safety
700 ///
701 /// This method is cancel safe. Once a readiness event occurs, the method
702 /// will continue to return immediately until the readiness event is
703 /// consumed by an attempt to read or write that fails with `WouldBlock` or
704 /// `Poll::Pending`.
705 #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
706 pub async fn readable_mut<'a>(&'a mut self) -> io::Result<AsyncFdReadyMutGuard<'a, T>> {
707 self.ready_mut(Interest::READABLE).await
708 }
709
710 /// Waits for the file descriptor to become writable, returning a
711 /// [`AsyncFdReadyGuard`] that must be dropped to resume write-readiness
712 /// polling.
713 ///
714 /// This method takes `&self`, so it is possible to call this method
715 /// concurrently with other methods on this struct. This method only
716 /// provides shared access to the inner IO resource when handling the
717 /// [`AsyncFdReadyGuard`].
718 ///
719 /// # Cancel safety
720 ///
721 /// This method is cancel safe. Once a readiness event occurs, the method
722 /// will continue to return immediately until the readiness event is
723 /// consumed by an attempt to read or write that fails with `WouldBlock` or
724 /// `Poll::Pending`.
725 #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
726 pub async fn writable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> {
727 self.ready(Interest::WRITABLE).await
728 }
729
730 /// Waits for the file descriptor to become writable, returning a
731 /// [`AsyncFdReadyMutGuard`] that must be dropped to resume write-readiness
732 /// polling.
733 ///
734 /// This method takes `&mut self`, so it is possible to access the inner IO
735 /// resource mutably when handling the [`AsyncFdReadyMutGuard`].
736 ///
737 /// # Cancel safety
738 ///
739 /// This method is cancel safe. Once a readiness event occurs, the method
740 /// will continue to return immediately until the readiness event is
741 /// consumed by an attempt to read or write that fails with `WouldBlock` or
742 /// `Poll::Pending`.
743 #[allow(clippy::needless_lifetimes)] // The lifetime improves rustdoc rendering.
744 pub async fn writable_mut<'a>(&'a mut self) -> io::Result<AsyncFdReadyMutGuard<'a, T>> {
745 self.ready_mut(Interest::WRITABLE).await
746 }
747
748 /// Reads or writes from the file descriptor using a user-provided IO operation.
749 ///
750 /// The `async_io` method is a convenience utility that waits for the file
751 /// descriptor to become ready, and then executes the provided IO operation.
752 /// Since file descriptors may be marked ready spuriously, the closure will
753 /// be called repeatedly until it returns something other than a
754 /// [`WouldBlock`] error. This is done using the following loop:
755 ///
756 /// ```no_run
757 /// # use std::io::{self, Result};
758 /// # struct Dox<T> { inner: T }
759 /// # impl<T> Dox<T> {
760 /// # async fn writable(&self) -> Result<&Self> {
761 /// # Ok(self)
762 /// # }
763 /// # fn try_io<R>(&self, _: impl FnMut(&T) -> Result<R>) -> Result<Result<R>> {
764 /// # panic!()
765 /// # }
766 /// async fn async_io<R>(&self, mut f: impl FnMut(&T) -> io::Result<R>) -> io::Result<R> {
767 /// loop {
768 /// // or `readable` if called with the read interest.
769 /// let guard = self.writable().await?;
770 ///
771 /// match guard.try_io(&mut f) {
772 /// Ok(result) => return result,
773 /// Err(_would_block) => continue,
774 /// }
775 /// }
776 /// }
777 /// # }
778 /// ```
779 ///
780 /// The closure should only return a [`WouldBlock`] error if it has performed
781 /// an IO operation on the file descriptor that failed due to the file descriptor not being
782 /// ready. Returning a [`WouldBlock`] error in any other situation will
783 /// incorrectly clear the readiness flag, which can cause the file descriptor to
784 /// behave incorrectly.
785 ///
786 /// The closure should not perform the IO operation using any of the methods
787 /// defined on the Tokio [`AsyncFd`] type, as this will mess with the
788 /// readiness flag and can cause the file descriptor to behave incorrectly.
789 ///
790 /// This method is not intended to be used with combined interests.
791 /// The closure should perform only one type of IO operation, so it should not
792 /// require more than one ready state. This method may panic or sleep forever
793 /// if it is called with a combined interest.
794 ///
795 /// # Examples
796 ///
797 /// This example sends some bytes on the inner [`std::net::UdpSocket`]. The `async_io`
798 /// method waits for readiness, and retries if the send operation does block. This example
799 /// is equivalent to the one given for [`try_io`].
800 ///
801 /// ```no_run
802 /// use tokio::io::{Interest, unix::AsyncFd};
803 ///
804 /// use std::io;
805 /// use std::net::UdpSocket;
806 ///
807 /// #[tokio::main]
808 /// async fn main() -> io::Result<()> {
809 /// let socket = UdpSocket::bind("0.0.0.0:8080")?;
810 /// socket.set_nonblocking(true)?;
811 /// // SAFETY: `UdpSocket` owns a valid file descriptor and we do not
812 /// // replace or close it while owned by `AsyncFd`.
813 /// let async_fd = unsafe { AsyncFd::register(socket)? };
814 ///
815 /// let written = async_fd
816 /// .async_io(Interest::WRITABLE, |inner| inner.send(&[1, 2]))
817 /// .await?;
818 ///
819 /// println!("wrote {written} bytes");
820 ///
821 /// Ok(())
822 /// }
823 /// ```
824 ///
825 /// [`try_io`]: AsyncFdReadyGuard::try_io
826 /// [`WouldBlock`]: std::io::ErrorKind::WouldBlock
827 pub async fn async_io<R>(
828 &self,
829 interest: Interest,
830 mut f: impl FnMut(&T) -> io::Result<R>,
831 ) -> io::Result<R> {
832 self.registration
833 .async_io(interest, || f(self.get_ref()))
834 .await
835 }
836
837 /// Reads or writes from the file descriptor using a user-provided IO operation.
838 ///
839 /// The behavior is the same as [`async_io`], except that the closure can mutate the inner
840 /// value of the [`AsyncFd`].
841 ///
842 /// [`async_io`]: AsyncFd::async_io
843 pub async fn async_io_mut<R>(
844 &mut self,
845 interest: Interest,
846 mut f: impl FnMut(&mut T) -> io::Result<R>,
847 ) -> io::Result<R> {
848 self.registration
849 .async_io(interest, || f(self.inner.as_mut().unwrap()))
850 .await
851 }
852
853 /// Tries to read or write from the file descriptor using a user-provided IO operation.
854 ///
855 /// If the file descriptor is ready, the provided closure is called. The closure
856 /// should attempt to perform IO operation on the file descriptor by manually
857 /// calling the appropriate syscall. If the operation fails because the
858 /// file descriptor is not actually ready, then the closure should return a
859 /// `WouldBlock` error and the readiness flag is cleared. The return value
860 /// of the closure is then returned by `try_io`.
861 ///
862 /// If the file descriptor is not ready, then the closure is not called
863 /// and a `WouldBlock` error is returned.
864 ///
865 /// The closure should only return a `WouldBlock` error if it has performed
866 /// an IO operation on the file descriptor that failed due to the file descriptor not being
867 /// ready. Returning a `WouldBlock` error in any other situation will
868 /// incorrectly clear the readiness flag, which can cause the file descriptor to
869 /// behave incorrectly.
870 ///
871 /// The closure should not perform the IO operation using any of the methods
872 /// defined on the Tokio `AsyncFd` type, as this will mess with the
873 /// readiness flag and can cause the file descriptor to behave incorrectly.
874 ///
875 /// This method is not intended to be used with combined interests.
876 /// The closure should perform only one type of IO operation, so it should not
877 /// require more than one ready state. This method may panic or sleep forever
878 /// if it is called with a combined interest.
879 pub fn try_io<R>(
880 &self,
881 interest: Interest,
882 f: impl FnOnce(&T) -> io::Result<R>,
883 ) -> io::Result<R> {
884 self.registration
885 .try_io(interest, || f(self.inner.as_ref().unwrap()))
886 }
887
888 /// Tries to read or write from the file descriptor using a user-provided IO operation.
889 ///
890 /// The behavior is the same as [`try_io`], except that the closure can mutate the inner
891 /// value of the [`AsyncFd`].
892 ///
893 /// [`try_io`]: AsyncFd::try_io
894 pub fn try_io_mut<R>(
895 &mut self,
896 interest: Interest,
897 f: impl FnOnce(&mut T) -> io::Result<R>,
898 ) -> io::Result<R> {
899 self.registration
900 .try_io(interest, || f(self.inner.as_mut().unwrap()))
901 }
902
903 /// Deprecated. Use [`AsyncFd::register`] instead. See [#8587] for more information.
904 ///
905 /// [#8587]: https://github.com/tokio-rs/tokio/pull/8587
906 #[inline]
907 #[track_caller]
908 #[deprecated(
909 note = "unsound due to lack of I/O safety; use `AsyncFd::register` instead (see https://github.com/tokio-rs/tokio/pull/8587)"
910 )]
911 pub fn new(inner: T) -> io::Result<Self> {
912 unsafe { Ok(Self::register(inner)?) }
913 }
914
915 /// Deprecated. Use [`AsyncFd::register_with_interest`] instead. See [#8587] for more information.
916 ///
917 /// [#8587]: https://github.com/tokio-rs/tokio/pull/8587
918 #[inline]
919 #[track_caller]
920 #[deprecated(
921 note = "unsound due to lack of I/O safety; use `AsyncFd::register_with_interest` instead (see https://github.com/tokio-rs/tokio/pull/8587)"
922 )]
923 pub fn with_interest(inner: T, interest: Interest) -> io::Result<Self> {
924 unsafe { Ok(Self::register_with_interest(inner, interest)?) }
925 }
926
927 /// Deprecated. Use [`AsyncFd::register`] instead. See [#8587] for more information.
928 ///
929 /// [#8587]: https://github.com/tokio-rs/tokio/pull/8587
930 #[inline]
931 #[track_caller]
932 #[deprecated(
933 note = "unsound due to lack of I/O safety; use `AsyncFd::register` instead (see https://github.com/tokio-rs/tokio/pull/8587)"
934 )]
935 pub fn try_new(inner: T) -> Result<Self, AsyncFdRegisterError<T>> {
936 unsafe { Self::register(inner) }
937 }
938
939 /// Deprecated. Use [`AsyncFd::register_with_interest`] instead. See [#8587] for more information.
940 ///
941 /// [#8587]: https://github.com/tokio-rs/tokio/pull/8587
942 #[inline]
943 #[track_caller]
944 #[deprecated(
945 note = "unsound due to lack of I/O safety; use `AsyncFd::register_with_interest` instead (see https://github.com/tokio-rs/tokio/pull/8587)"
946 )]
947 pub fn try_with_interest(
948 inner: T,
949 interest: Interest,
950 ) -> Result<Self, AsyncFdRegisterError<T>> {
951 unsafe { Self::register_with_interest(inner, interest) }
952 }
953}
954
955impl<T: AsRawFd> AsRawFd for AsyncFd<T> {
956 fn as_raw_fd(&self) -> RawFd {
957 self.inner.as_ref().unwrap().as_raw_fd()
958 }
959}
960
961impl<T: AsRawFd> std::os::unix::io::AsFd for AsyncFd<T> {
962 fn as_fd(&self) -> std::os::unix::io::BorrowedFd<'_> {
963 // SAFETY: By the safety requirements of `AsyncFd::register` and
964 // `AsyncFd::register_with_interest`, the file descriptor returned by
965 // `as_raw_fd()` is valid and remains open for the lifetime of `self`.
966 unsafe { std::os::unix::io::BorrowedFd::borrow_raw(self.as_raw_fd()) }
967 }
968}
969
970impl<T: std::fmt::Debug + AsRawFd> std::fmt::Debug for AsyncFd<T> {
971 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
972 f.debug_struct("AsyncFd")
973 .field("inner", &self.inner)
974 .finish()
975 }
976}
977
978impl<T: AsRawFd> Drop for AsyncFd<T> {
979 fn drop(&mut self) {
980 let _ = self.take_inner();
981 }
982}
983
984impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> {
985 /// Indicates to tokio that the file descriptor is no longer ready. All
986 /// internal readiness flags will be cleared, and tokio will wait for the
987 /// next edge-triggered readiness notification from the OS.
988 ///
989 /// This function is commonly used with guards returned by [`AsyncFd::readable`] and
990 /// [`AsyncFd::writable`].
991 ///
992 /// It is critical that this function not be called unless your code
993 /// _actually observes_ that the file descriptor is _not_ ready. Do not call
994 /// it simply because, for example, a read succeeded; it should be called
995 /// when a read is observed to block.
996 ///
997 /// This method only clears readiness events that happened before the creation of this guard.
998 /// In other words, if the IO resource becomes ready between the creation of the guard and
999 /// this call to `clear_ready`, then the readiness is not actually cleared.
1000 pub fn clear_ready(&mut self) {
1001 if let Some(event) = self.event.take() {
1002 self.async_fd.registration.clear_readiness(event);
1003 }
1004 }
1005
1006 /// Indicates to tokio that the file descriptor no longer has a specific readiness.
1007 /// The internal readiness flag will be cleared, and tokio will wait for the
1008 /// next edge-triggered readiness notification from the OS.
1009 ///
1010 /// This function is useful in combination with the [`AsyncFd::ready`] method when a
1011 /// combined interest like `Interest::READABLE | Interest::WRITABLE` is used.
1012 ///
1013 /// It is critical that this function not be called unless your code
1014 /// _actually observes_ that the file descriptor is _not_ ready for the provided `Ready`.
1015 /// Do not call it simply because, for example, a read succeeded; it should be called
1016 /// when a read is observed to block. Only clear the specific readiness that is observed to
1017 /// block. For example when a read blocks when using a combined interest,
1018 /// only clear `Ready::READABLE`.
1019 ///
1020 /// This method only clears readiness events that happened before the creation of this guard.
1021 /// In other words, if the IO resource becomes ready between the creation of the guard and
1022 /// this call to `clear_ready_matching`, then the readiness is not actually cleared.
1023 ///
1024 /// # Examples
1025 ///
1026 /// Concurrently read and write to a [`std::net::TcpStream`] on the same task without
1027 /// splitting.
1028 ///
1029 /// ```no_run
1030 /// use std::error::Error;
1031 /// use std::io;
1032 /// use std::io::{Read, Write};
1033 /// use std::net::TcpStream;
1034 /// use tokio::io::unix::AsyncFd;
1035 /// use tokio::io::{Interest, Ready};
1036 ///
1037 /// #[tokio::main]
1038 /// async fn main() -> Result<(), Box<dyn Error>> {
1039 /// let stream = TcpStream::connect("127.0.0.1:8080")?;
1040 /// stream.set_nonblocking(true)?;
1041 /// // SAFETY: `TcpStream` owns a valid file descriptor and we do not
1042 /// // replace or close it while owned by `AsyncFd`.
1043 /// let stream = unsafe { AsyncFd::register(stream)? };
1044 ///
1045 /// loop {
1046 /// let mut guard = stream
1047 /// .ready(Interest::READABLE | Interest::WRITABLE)
1048 /// .await?;
1049 ///
1050 /// if guard.ready().is_readable() {
1051 /// let mut data = vec![0; 1024];
1052 /// // Try to read data, this may still fail with `WouldBlock`
1053 /// // if the readiness event is a false positive.
1054 /// match stream.get_ref().read(&mut data) {
1055 /// Ok(n) => {
1056 /// println!("read {} bytes", n);
1057 /// }
1058 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
1059 /// // a read has blocked, but a write might still succeed.
1060 /// // clear only the read readiness.
1061 /// guard.clear_ready_matching(Ready::READABLE);
1062 /// continue;
1063 /// }
1064 /// Err(e) => {
1065 /// return Err(e.into());
1066 /// }
1067 /// }
1068 /// }
1069 ///
1070 /// if guard.ready().is_writable() {
1071 /// // Try to write data, this may still fail with `WouldBlock`
1072 /// // if the readiness event is a false positive.
1073 /// match stream.get_ref().write(b"hello world") {
1074 /// Ok(n) => {
1075 /// println!("write {} bytes", n);
1076 /// }
1077 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
1078 /// // a write has blocked, but a read might still succeed.
1079 /// // clear only the write readiness.
1080 /// guard.clear_ready_matching(Ready::WRITABLE);
1081 /// continue;
1082 /// }
1083 /// Err(e) => {
1084 /// return Err(e.into());
1085 /// }
1086 /// }
1087 /// }
1088 /// }
1089 /// }
1090 /// ```
1091 pub fn clear_ready_matching(&mut self, ready: Ready) {
1092 if let Some(mut event) = self.event.take() {
1093 self.async_fd
1094 .registration
1095 .clear_readiness(event.with_ready(ready));
1096
1097 // the event is no longer ready for the readiness that was just cleared
1098 event.ready = event.ready - ready;
1099
1100 if !event.ready.is_empty() {
1101 self.event = Some(event);
1102 }
1103 }
1104 }
1105
1106 /// This method should be invoked when you intentionally want to keep the
1107 /// ready flag asserted.
1108 ///
1109 /// While this function is itself a no-op, it satisfies the `#[must_use]`
1110 /// constraint on the [`AsyncFdReadyGuard`] type.
1111 pub fn retain_ready(&mut self) {
1112 // no-op
1113 }
1114
1115 /// Get the [`Ready`] value associated with this guard.
1116 ///
1117 /// This method will return the empty readiness state if
1118 /// [`AsyncFdReadyGuard::clear_ready`] has been called on
1119 /// the guard.
1120 ///
1121 /// [`Ready`]: crate::io::Ready
1122 pub fn ready(&self) -> Ready {
1123 match &self.event {
1124 Some(event) => event.ready,
1125 None => Ready::EMPTY,
1126 }
1127 }
1128
1129 /// Performs the provided IO operation.
1130 ///
1131 /// If `f` returns a [`WouldBlock`] error, the readiness state associated
1132 /// with this file descriptor is cleared, and the method returns
1133 /// `Err(TryIoError::WouldBlock)`. You will typically need to poll the
1134 /// `AsyncFd` again when this happens.
1135 ///
1136 /// This method helps ensure that the readiness state of the underlying file
1137 /// descriptor remains in sync with the tokio-side readiness state, by
1138 /// clearing the tokio-side state only when a [`WouldBlock`] condition
1139 /// occurs. It is the responsibility of the caller to ensure that `f`
1140 /// returns [`WouldBlock`] only if the file descriptor that originated this
1141 /// `AsyncFdReadyGuard` no longer expresses the readiness state that was queried to
1142 /// create this `AsyncFdReadyGuard`.
1143 ///
1144 /// # Examples
1145 ///
1146 /// This example sends some bytes to the inner [`std::net::UdpSocket`]. Waiting
1147 /// for write-readiness and retrying when the send operation does block are explicit.
1148 /// This example can be written more succinctly using [`AsyncFd::async_io`].
1149 ///
1150 /// ```no_run
1151 /// use tokio::io::unix::AsyncFd;
1152 ///
1153 /// use std::io;
1154 /// use std::net::UdpSocket;
1155 ///
1156 /// #[tokio::main]
1157 /// async fn main() -> io::Result<()> {
1158 /// let socket = UdpSocket::bind("0.0.0.0:8080")?;
1159 /// socket.set_nonblocking(true)?;
1160 /// // SAFETY: `UdpSocket` owns a valid file descriptor and we do not
1161 /// // replace or close it while owned by `AsyncFd`.
1162 /// let async_fd = unsafe { AsyncFd::register(socket)? };
1163 ///
1164 /// let written = loop {
1165 /// let mut guard = async_fd.writable().await?;
1166 /// match guard.try_io(|inner| inner.get_ref().send(&[1, 2])) {
1167 /// Ok(result) => {
1168 /// break result?;
1169 /// }
1170 /// Err(_would_block) => {
1171 /// // try_io already cleared the file descriptor's readiness state
1172 /// continue;
1173 /// }
1174 /// }
1175 /// };
1176 ///
1177 /// println!("wrote {written} bytes");
1178 ///
1179 /// Ok(())
1180 /// }
1181 /// ```
1182 ///
1183 /// [`WouldBlock`]: std::io::ErrorKind::WouldBlock
1184 // Alias for old name in 0.x
1185 #[cfg_attr(docsrs, doc(alias = "with_io"))]
1186 pub fn try_io<R>(
1187 &mut self,
1188 f: impl FnOnce(&'a AsyncFd<Inner>) -> io::Result<R>,
1189 ) -> Result<io::Result<R>, TryIoError> {
1190 let result = f(self.async_fd);
1191
1192 match result {
1193 Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
1194 self.clear_ready();
1195 Err(TryIoError(()))
1196 }
1197 result => Ok(result),
1198 }
1199 }
1200
1201 /// Returns a shared reference to the inner [`AsyncFd`].
1202 pub fn get_ref(&self) -> &'a AsyncFd<Inner> {
1203 self.async_fd
1204 }
1205
1206 /// Returns a shared reference to the backing object of the inner [`AsyncFd`].
1207 pub fn get_inner(&self) -> &'a Inner {
1208 self.get_ref().get_ref()
1209 }
1210}
1211
1212impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> {
1213 /// Indicates to tokio that the file descriptor is no longer ready. All
1214 /// internal readiness flags will be cleared, and tokio will wait for the
1215 /// next edge-triggered readiness notification from the OS.
1216 ///
1217 /// This function is commonly used with guards returned by [`AsyncFd::readable_mut`] and
1218 /// [`AsyncFd::writable_mut`].
1219 ///
1220 /// It is critical that this function not be called unless your code
1221 /// _actually observes_ that the file descriptor is _not_ ready. Do not call
1222 /// it simply because, for example, a read succeeded; it should be called
1223 /// when a read is observed to block.
1224 ///
1225 /// This method only clears readiness events that happened before the creation of this guard.
1226 /// In other words, if the IO resource becomes ready between the creation of the guard and
1227 /// this call to `clear_ready`, then the readiness is not actually cleared.
1228 pub fn clear_ready(&mut self) {
1229 if let Some(event) = self.event.take() {
1230 self.async_fd.registration.clear_readiness(event);
1231 }
1232 }
1233
1234 /// Indicates to tokio that the file descriptor no longer has a specific readiness.
1235 /// The internal readiness flag will be cleared, and tokio will wait for the
1236 /// next edge-triggered readiness notification from the OS.
1237 ///
1238 /// This function is useful in combination with the [`AsyncFd::ready_mut`] method when a
1239 /// combined interest like `Interest::READABLE | Interest::WRITABLE` is used.
1240 ///
1241 /// It is critical that this function not be called unless your code
1242 /// _actually observes_ that the file descriptor is _not_ ready for the provided `Ready`.
1243 /// Do not call it simply because, for example, a read succeeded; it should be called
1244 /// when a read is observed to block. Only clear the specific readiness that is observed to
1245 /// block. For example when a read blocks when using a combined interest,
1246 /// only clear `Ready::READABLE`.
1247 ///
1248 /// This method only clears readiness events that happened before the creation of this guard.
1249 /// In other words, if the IO resource becomes ready between the creation of the guard and
1250 /// this call to `clear_ready_matching`, then the readiness is not actually cleared.
1251 ///
1252 /// # Examples
1253 ///
1254 /// Concurrently read and write to a [`std::net::TcpStream`] on the same task without
1255 /// splitting.
1256 ///
1257 /// ```no_run
1258 /// use std::error::Error;
1259 /// use std::io;
1260 /// use std::io::{Read, Write};
1261 /// use std::net::TcpStream;
1262 /// use tokio::io::unix::AsyncFd;
1263 /// use tokio::io::{Interest, Ready};
1264 ///
1265 /// #[tokio::main]
1266 /// async fn main() -> Result<(), Box<dyn Error>> {
1267 /// let stream = TcpStream::connect("127.0.0.1:8080")?;
1268 /// stream.set_nonblocking(true)?;
1269 /// // SAFETY: `TcpStream` owns a valid file descriptor and we do not
1270 /// // replace or close it while owned by `AsyncFd`.
1271 /// let mut stream = unsafe { AsyncFd::register(stream)? };
1272 ///
1273 /// loop {
1274 /// let mut guard = stream
1275 /// .ready_mut(Interest::READABLE | Interest::WRITABLE)
1276 /// .await?;
1277 ///
1278 /// if guard.ready().is_readable() {
1279 /// let mut data = vec![0; 1024];
1280 /// // Try to read data, this may still fail with `WouldBlock`
1281 /// // if the readiness event is a false positive.
1282 /// match guard.get_inner_mut().read(&mut data) {
1283 /// Ok(n) => {
1284 /// println!("read {} bytes", n);
1285 /// }
1286 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
1287 /// // a read has blocked, but a write might still succeed.
1288 /// // clear only the read readiness.
1289 /// guard.clear_ready_matching(Ready::READABLE);
1290 /// continue;
1291 /// }
1292 /// Err(e) => {
1293 /// return Err(e.into());
1294 /// }
1295 /// }
1296 /// }
1297 ///
1298 /// if guard.ready().is_writable() {
1299 /// // Try to write data, this may still fail with `WouldBlock`
1300 /// // if the readiness event is a false positive.
1301 /// match guard.get_inner_mut().write(b"hello world") {
1302 /// Ok(n) => {
1303 /// println!("write {} bytes", n);
1304 /// }
1305 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
1306 /// // a write has blocked, but a read might still succeed.
1307 /// // clear only the write readiness.
1308 /// guard.clear_ready_matching(Ready::WRITABLE);
1309 /// continue;
1310 /// }
1311 /// Err(e) => {
1312 /// return Err(e.into());
1313 /// }
1314 /// }
1315 /// }
1316 /// }
1317 /// }
1318 /// ```
1319 pub fn clear_ready_matching(&mut self, ready: Ready) {
1320 if let Some(mut event) = self.event.take() {
1321 self.async_fd
1322 .registration
1323 .clear_readiness(event.with_ready(ready));
1324
1325 // the event is no longer ready for the readiness that was just cleared
1326 event.ready = event.ready - ready;
1327
1328 if !event.ready.is_empty() {
1329 self.event = Some(event);
1330 }
1331 }
1332 }
1333
1334 /// This method should be invoked when you intentionally want to keep the
1335 /// ready flag asserted.
1336 ///
1337 /// While this function is itself a no-op, it satisfies the `#[must_use]`
1338 /// constraint on the [`AsyncFdReadyGuard`] type.
1339 pub fn retain_ready(&mut self) {
1340 // no-op
1341 }
1342
1343 /// Get the [`Ready`] value associated with this guard.
1344 ///
1345 /// This method will return the empty readiness state if
1346 /// [`AsyncFdReadyGuard::clear_ready`] has been called on
1347 /// the guard.
1348 ///
1349 /// [`Ready`]: super::Ready
1350 pub fn ready(&self) -> Ready {
1351 match &self.event {
1352 Some(event) => event.ready,
1353 None => Ready::EMPTY,
1354 }
1355 }
1356
1357 /// Performs the provided IO operation.
1358 ///
1359 /// If `f` returns a [`WouldBlock`] error, the readiness state associated
1360 /// with this file descriptor is cleared, and the method returns
1361 /// `Err(TryIoError::WouldBlock)`. You will typically need to poll the
1362 /// `AsyncFd` again when this happens.
1363 ///
1364 /// This method helps ensure that the readiness state of the underlying file
1365 /// descriptor remains in sync with the tokio-side readiness state, by
1366 /// clearing the tokio-side state only when a [`WouldBlock`] condition
1367 /// occurs. It is the responsibility of the caller to ensure that `f`
1368 /// returns [`WouldBlock`] only if the file descriptor that originated this
1369 /// `AsyncFdReadyGuard` no longer expresses the readiness state that was queried to
1370 /// create this `AsyncFdReadyGuard`.
1371 ///
1372 /// [`WouldBlock`]: std::io::ErrorKind::WouldBlock
1373 pub fn try_io<R>(
1374 &mut self,
1375 f: impl FnOnce(&mut AsyncFd<Inner>) -> io::Result<R>,
1376 ) -> Result<io::Result<R>, TryIoError> {
1377 let result = f(self.async_fd);
1378
1379 match result {
1380 Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
1381 self.clear_ready();
1382 Err(TryIoError(()))
1383 }
1384 result => Ok(result),
1385 }
1386 }
1387
1388 /// Returns a shared reference to the inner [`AsyncFd`].
1389 pub fn get_ref(&self) -> &AsyncFd<Inner> {
1390 self.async_fd
1391 }
1392
1393 /// Returns a mutable reference to the inner [`AsyncFd`].
1394 pub fn get_mut(&mut self) -> &mut AsyncFd<Inner> {
1395 self.async_fd
1396 }
1397
1398 /// Returns a shared reference to the backing object of the inner [`AsyncFd`].
1399 pub fn get_inner(&self) -> &Inner {
1400 self.get_ref().get_ref()
1401 }
1402
1403 /// Returns a mutable reference to the backing object of the inner [`AsyncFd`].
1404 ///
1405 /// This may be used to interact with the underlying file. Be aware that this
1406 /// method must not be used to exchange the underlying IO resource for a
1407 /// different one. Doing so is a violation of the safety requirements on
1408 /// [`AsyncFd::register`].
1409 pub fn get_inner_mut(&mut self) -> &mut Inner {
1410 self.get_mut().get_mut()
1411 }
1412}
1413
1414impl<'a, T: std::fmt::Debug + AsRawFd> std::fmt::Debug for AsyncFdReadyGuard<'a, T> {
1415 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1416 f.debug_struct("ReadyGuard")
1417 .field("async_fd", &self.async_fd)
1418 .finish()
1419 }
1420}
1421
1422impl<'a, T: std::fmt::Debug + AsRawFd> std::fmt::Debug for AsyncFdReadyMutGuard<'a, T> {
1423 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1424 f.debug_struct("MutReadyGuard")
1425 .field("async_fd", &self.async_fd)
1426 .finish()
1427 }
1428}
1429
1430/// The error type returned by [`try_io`].
1431///
1432/// This error indicates that the IO resource returned a [`WouldBlock`] error.
1433///
1434/// [`WouldBlock`]: std::io::ErrorKind::WouldBlock
1435/// [`try_io`]: method@AsyncFdReadyGuard::try_io
1436#[derive(Debug)]
1437pub struct TryIoError(());
1438
1439/// Error returned by [`register`] or [`register_with_interest`].
1440///
1441/// [`register`]: AsyncFd::register
1442/// [`register_with_interest`]: AsyncFd::register_with_interest
1443pub struct AsyncFdRegisterError<T> {
1444 inner: T,
1445 cause: io::Error,
1446}
1447
1448/// Deprecated. Use [`AsyncFdRegisterError`] instead.
1449#[doc(hidden)]
1450#[deprecated(note = "use `AsyncFdRegisterError` instead")]
1451pub type AsyncFdTryNewError<T> = AsyncFdRegisterError<T>;
1452
1453impl<T> AsyncFdRegisterError<T> {
1454 /// Returns the original object passed to [`register`] or [`register_with_interest`]
1455 /// alongside the error that caused these functions to fail.
1456 ///
1457 /// [`register`]: AsyncFd::register
1458 /// [`register_with_interest`]: AsyncFd::register_with_interest
1459 pub fn into_parts(self) -> (T, io::Error) {
1460 (self.inner, self.cause)
1461 }
1462}
1463
1464impl<T> fmt::Display for AsyncFdRegisterError<T> {
1465 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1466 fmt::Display::fmt(&self.cause, f)
1467 }
1468}
1469
1470impl<T> fmt::Debug for AsyncFdRegisterError<T> {
1471 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1472 fmt::Debug::fmt(&self.cause, f)
1473 }
1474}
1475
1476impl<T> Error for AsyncFdRegisterError<T> {
1477 fn source(&self) -> Option<&(dyn Error + 'static)> {
1478 Some(&self.cause)
1479 }
1480}
1481
1482impl<T> From<AsyncFdRegisterError<T>> for io::Error {
1483 fn from(value: AsyncFdRegisterError<T>) -> Self {
1484 value.cause
1485 }
1486}