Struct async_io::Async

pub struct Async<T> { /* private fields */ }

Async adapter for I/O types.

This type puts an I/O handle into non-blocking mode, registers it in epoll/kqueue/event ports/wepoll, and then provides an async interface for it.

Caveats

Async is a low-level primitive, and as such it comes with some caveats.

For higher-level primitives built on top of Async, look into async-net or async-process (on Unix).

Supported types

Async supports all networking types, as well as some OS-specific file descriptors like timerfd and inotify.

However, do not use Async with types like File, Stdin, Stdout, or Stderr because all operating systems have issues with them when put in non-blocking mode.

Concurrent I/O

Note that &Async<T> implements AsyncRead and AsyncWrite if &T implements those traits, which means tasks can concurrently read and write using shared references.

But there is a catch: only one task can read a time, and only one task can write at a time. It is okay to have two tasks where one is reading and the other is writing at the same time, but it is not okay to have two tasks reading at the same time or writing at the same time. If you try to do that, conflicting tasks will just keep waking each other in turn, thus wasting CPU time.

Besides AsyncRead and AsyncWrite, this caveat also applies to poll_readable() and poll_writable().

However, any number of tasks can be concurrently calling other methods like readable() or read_with().

Closing

Closing the write side of Async with close() simply flushes. If you want to shutdown a TCP or Unix socket, use Shutdown.

Examples

Connect to a server and echo incoming messages back to the server:

use async_io::Async;
use futures_lite::io;
use std::net::TcpStream;

// Connect to a local server.
let stream = Async::<TcpStream>::connect(([127, 0, 0, 1], 8000)).await?;

// Echo all messages from the read side of the stream into the write side.
io::copy(&stream, &stream).await?;

You can use either predefined async methods or wrap blocking I/O operations in Async::read_with(), Async::read_with_mut(), Async::write_with(), and Async::write_with_mut():

use async_io::Async;
use std::net::TcpListener;

let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;

// These two lines are equivalent:
let (stream, addr) = listener.accept().await?;
let (stream, addr) = listener.read_with(|inner| inner.accept()).await?;

Implementations

impl<T: AsRawFd> Async<T>

pub fn new(io: T) -> Result<Async<T>>

Creates an async I/O handle.

This method will put the handle in non-blocking mode and register it in epoll/kqueue/event ports/wepoll.

On Unix systems, the handle must implement AsRawFd, while on Windows it must implement AsRawSocket.

Examples

use async_io::Async;
use std::net::{SocketAddr, TcpListener};

let listener = TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0)))?;
let listener = Async::new(listener)?;

impl<T> Async<T>

pub fn get_ref(&self) -> &T

Gets a reference to the inner I/O handle.

Examples

use async_io::Async;
use std::net::TcpListener;

let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;
let inner = listener.get_ref();

pub fn get_mut(&mut self) -> &mut T

Gets a mutable reference to the inner I/O handle.

Examples

use async_io::Async;
use std::net::TcpListener;

let mut listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;
let inner = listener.get_mut();

pub fn into_inner(self) -> Result<T>

Unwraps the inner I/O handle.

This method will not put the I/O handle back into blocking mode.

Examples

use async_io::Async;
use std::net::TcpListener;

let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;
let inner = listener.into_inner()?;

// Put the listener back into blocking mode.
inner.set_nonblocking(false)?;

pub fn readable(&self) -> Readable<'_, T>

Waits until the I/O handle is readable.

This method completes when a read operation on this I/O handle wouldn’t block.

Examples

use async_io::Async;
use std::net::TcpListener;

let mut listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;

// Wait until a client can be accepted.
listener.readable().await?;

pub fn readable_owned(self: Arc<Self>) -> ReadableOwned<T>

Waits until the I/O handle is readable.

This method completes when a read operation on this I/O handle wouldn’t block.

pub fn writable(&self) -> Writable<'_, T>

Waits until the I/O handle is writable.

This method completes when a write operation on this I/O handle wouldn’t block.

Examples

use async_io::Async;
use std::net::{TcpStream, ToSocketAddrs};

let addr = "example.com:80".to_socket_addrs()?.next().unwrap();
let stream = Async::<TcpStream>::connect(addr).await?;

// Wait until the stream is writable.
stream.writable().await?;

pub fn writable_owned(self: Arc<Self>) -> WritableOwned<T>

Waits until the I/O handle is writable.

This method completes when a write operation on this I/O handle wouldn’t block.

pub fn poll_readable(&self, cx: &mut Context<'_>) -> Poll<Result<()>>

Polls the I/O handle for readability.

When this method returns Poll::Ready, that means the OS has delivered an event indicating readability since the last time this task has called the method and received Poll::Pending.

Caveats

Two different tasks should not call this method concurrently. Otherwise, conflicting tasks will just keep waking each other in turn, thus wasting CPU time.

Note that the AsyncRead implementation for Async also uses this method.

Examples

use async_io::Async;
use futures_lite::future;
use std::net::TcpListener;

let mut listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;

// Wait until a client can be accepted.
future::poll_fn(|cx| listener.poll_readable(cx)).await?;

pub fn poll_writable(&self, cx: &mut Context<'_>) -> Poll<Result<()>>

Polls the I/O handle for writability.

When this method returns Poll::Ready, that means the OS has delivered an event indicating writability since the last time this task has called the method and received Poll::Pending.

Caveats

Two different tasks should not call this method concurrently. Otherwise, conflicting tasks will just keep waking each other in turn, thus wasting CPU time.

Note that the AsyncWrite implementation for Async also uses this method.

Examples

use async_io::Async;
use futures_lite::future;
use std::net::{TcpStream, ToSocketAddrs};

let addr = "example.com:80".to_socket_addrs()?.next().unwrap();
let stream = Async::<TcpStream>::connect(addr).await?;

// Wait until the stream is writable.
future::poll_fn(|cx| stream.poll_writable(cx)).await?;

pub async fn read_with<R>(&self, op: impl FnMut(&T) -> Result<R>) -> Result<R>

Performs a read operation asynchronously.

The I/O handle is registered in the reactor and put in non-blocking mode. This method invokes the op closure in a loop until it succeeds or returns an error other than io::ErrorKind::WouldBlock. In between iterations of the loop, it waits until the OS sends a notification that the I/O handle is readable.

The closure receives a shared reference to the I/O handle.

Examples

use async_io::Async;
use std::net::TcpListener;

let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;

// Accept a new client asynchronously.
let (stream, addr) = listener.read_with(|l| l.accept()).await?;

pub async fn read_with_mut<R>(
    &mut self,
    op: impl FnMut(&mut T) -> Result<R>
) -> Result<R>

Performs a read operation asynchronously.

The I/O handle is registered in the reactor and put in non-blocking mode. This method invokes the op closure in a loop until it succeeds or returns an error other than io::ErrorKind::WouldBlock. In between iterations of the loop, it waits until the OS sends a notification that the I/O handle is readable.

The closure receives a mutable reference to the I/O handle.

Examples

use async_io::Async;
use std::net::TcpListener;

let mut listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;

// Accept a new client asynchronously.
let (stream, addr) = listener.read_with_mut(|l| l.accept()).await?;

pub async fn write_with<R>(&self, op: impl FnMut(&T) -> Result<R>) -> Result<R>

Performs a write operation asynchronously.

The I/O handle is registered in the reactor and put in non-blocking mode. This method invokes the op closure in a loop until it succeeds or returns an error other than io::ErrorKind::WouldBlock. In between iterations of the loop, it waits until the OS sends a notification that the I/O handle is writable.

The closure receives a shared reference to the I/O handle.

Examples

use async_io::Async;
use std::net::UdpSocket;

let socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 8000))?;
socket.get_ref().connect("127.0.0.1:9000")?;

let msg = b"hello";
let len = socket.write_with(|s| s.send(msg)).await?;

pub async fn write_with_mut<R>(
    &mut self,
    op: impl FnMut(&mut T) -> Result<R>
) -> Result<R>

Performs a write operation asynchronously.

The I/O handle is registered in the reactor and put in non-blocking mode. This method invokes the op closure in a loop until it succeeds or returns an error other than io::ErrorKind::WouldBlock. In between iterations of the loop, it waits until the OS sends a notification that the I/O handle is writable.

The closure receives a mutable reference to the I/O handle.

Examples

use async_io::Async;
use std::net::UdpSocket;

let mut socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 8000))?;
socket.get_ref().connect("127.0.0.1:9000")?;

let msg = b"hello";
let len = socket.write_with_mut(|s| s.send(msg)).await?;

impl Async<TcpListener>

pub fn bind<A: Into<SocketAddr>>(addr: A) -> Result<Async<TcpListener>>

Creates a TCP listener bound to the specified address.

Binding with port number 0 will request an available port from the OS.

Examples

use async_io::Async;
use std::net::TcpListener;

let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 0))?;
println!("Listening on {}", listener.get_ref().local_addr()?);

pub async fn accept(&self) -> Result<(Async<TcpStream>, SocketAddr)>

Accepts a new incoming TCP connection.

When a connection is established, it will be returned as a TCP stream together with its remote address.

Examples

use async_io::Async;
use std::net::TcpListener;

let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 8000))?;
let (stream, addr) = listener.accept().await?;
println!("Accepted client: {}", addr);

pub fn incoming(
    &self
) -> impl Stream<Item = Result<Async<TcpStream>>> + Send + '_

Returns a stream of incoming TCP connections.

The stream is infinite, i.e. it never stops with a None.

Examples

use async_io::Async;
use futures_lite::{pin, stream::StreamExt};
use std::net::TcpListener;

let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 8000))?;
let incoming = listener.incoming();
pin!(incoming);

while let Some(stream) = incoming.next().await {
    let stream = stream?;
    println!("Accepted client: {}", stream.get_ref().peer_addr()?);
}

impl Async<TcpStream>

pub async fn connect<A: Into<SocketAddr>>(addr: A) -> Result<Async<TcpStream>>

Creates a TCP connection to the specified address.

Examples

use async_io::Async;
use std::net::{TcpStream, ToSocketAddrs};

let addr = "example.com:80".to_socket_addrs()?.next().unwrap();
let stream = Async::<TcpStream>::connect(addr).await?;

pub async fn peek(&self, buf: &mut [u8]) -> Result<usize>

Reads data from the stream without removing it from the buffer.

Returns the number of bytes read. Successive calls of this method read the same data.

Examples

use async_io::Async;
use futures_lite::{io::AsyncWriteExt, stream::StreamExt};
use std::net::{TcpStream, ToSocketAddrs};

let addr = "example.com:80".to_socket_addrs()?.next().unwrap();
let mut stream = Async::<TcpStream>::connect(addr).await?;

stream
    .write_all(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
    .await?;

let mut buf = [0u8; 1024];
let len = stream.peek(&mut buf).await?;

impl Async<UdpSocket>

pub fn bind<A: Into<SocketAddr>>(addr: A) -> Result<Async<UdpSocket>>

Creates a UDP socket bound to the specified address.

Binding with port number 0 will request an available port from the OS.

Examples

use async_io::Async;
use std::net::UdpSocket;

let socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 0))?;
println!("Bound to {}", socket.get_ref().local_addr()?);

pub async fn recv_from(&self, buf: &mut [u8]) -> Result<(usize, SocketAddr)>

Receives a single datagram message.

Returns the number of bytes read and the address the message came from.

This method must be called with a valid byte slice of sufficient size to hold the message. If the message is too long to fit, excess bytes may get discarded.

Examples

use async_io::Async;
use std::net::UdpSocket;

let socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 8000))?;

let mut buf = [0u8; 1024];
let (len, addr) = socket.recv_from(&mut buf).await?;

pub async fn peek_from(&self, buf: &mut [u8]) -> Result<(usize, SocketAddr)>

Receives a single datagram message without removing it from the queue.

Returns the number of bytes read and the address the message came from.

This method must be called with a valid byte slice of sufficient size to hold the message. If the message is too long to fit, excess bytes may get discarded.

Examples

use async_io::Async;
use std::net::UdpSocket;

let socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 8000))?;

let mut buf = [0u8; 1024];
let (len, addr) = socket.peek_from(&mut buf).await?;

pub async fn send_to<A: Into<SocketAddr>>(
    &self,
    buf: &[u8],
    addr: A
) -> Result<usize>

Sends data to the specified address.

Returns the number of bytes writen.

Examples

use async_io::Async;
use std::net::UdpSocket;

let socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 0))?;
let addr = socket.get_ref().local_addr()?;

let msg = b"hello";
let len = socket.send_to(msg, addr).await?;

pub async fn recv(&self, buf: &mut [u8]) -> Result<usize>

Receives a single datagram message from the connected peer.

Returns the number of bytes read.

This method must be called with a valid byte slice of sufficient size to hold the message. If the message is too long to fit, excess bytes may get discarded.

The connect method connects this socket to a remote address. This method will fail if the socket is not connected.

Examples

use async_io::Async;
use std::net::UdpSocket;

let socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 8000))?;
socket.get_ref().connect("127.0.0.1:9000")?;

let mut buf = [0u8; 1024];
let len = socket.recv(&mut buf).await?;

pub async fn peek(&self, buf: &mut [u8]) -> Result<usize>

Receives a single datagram message from the connected peer without removing it from the queue.

Returns the number of bytes read and the address the message came from.

This method must be called with a valid byte slice of sufficient size to hold the message. If the message is too long to fit, excess bytes may get discarded.

The connect method connects this socket to a remote address. This method will fail if the socket is not connected.

Examples

use async_io::Async;
use std::net::UdpSocket;

let socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 8000))?;
socket.get_ref().connect("127.0.0.1:9000")?;

let mut buf = [0u8; 1024];
let len = socket.peek(&mut buf).await?;

pub async fn send(&self, buf: &[u8]) -> Result<usize>

Sends data to the connected peer.

Returns the number of bytes written.

The connect method connects this socket to a remote address. This method will fail if the socket is not connected.

Examples

use async_io::Async;
use std::net::UdpSocket;

let socket = Async::<UdpSocket>::bind(([127, 0, 0, 1], 8000))?;
socket.get_ref().connect("127.0.0.1:9000")?;

let msg = b"hello";
let len = socket.send(msg).await?;

impl Async<UnixListener>

pub fn bind<P: AsRef<Path>>(path: P) -> Result<Async<UnixListener>>

Creates a UDS listener bound to the specified path.

Examples

use async_io::Async;
use std::os::unix::net::UnixListener;

let listener = Async::<UnixListener>::bind("/tmp/socket")?;
println!("Listening on {:?}", listener.get_ref().local_addr()?);

pub async fn accept(&self) -> Result<(Async<UnixStream>, UnixSocketAddr)>

Accepts a new incoming UDS stream connection.

When a connection is established, it will be returned as a stream together with its remote address.

Examples

use async_io::Async;
use std::os::unix::net::UnixListener;

let listener = Async::<UnixListener>::bind("/tmp/socket")?;
let (stream, addr) = listener.accept().await?;
println!("Accepted client: {:?}", addr);

pub fn incoming(
    &self
) -> impl Stream<Item = Result<Async<UnixStream>>> + Send + '_

Returns a stream of incoming UDS connections.

The stream is infinite, i.e. it never stops with a None item.

Examples

use async_io::Async;
use futures_lite::{pin, stream::StreamExt};
use std::os::unix::net::UnixListener;

let listener = Async::<UnixListener>::bind("/tmp/socket")?;
let incoming = listener.incoming();
pin!(incoming);

while let Some(stream) = incoming.next().await {
    let stream = stream?;
    println!("Accepted client: {:?}", stream.get_ref().peer_addr()?);
}

impl Async<UnixStream>

pub async fn connect<P: AsRef<Path>>(path: P) -> Result<Async<UnixStream>>

Creates a UDS stream connected to the specified path.

Examples

use async_io::Async;
use std::os::unix::net::UnixStream;

let stream = Async::<UnixStream>::connect("/tmp/socket").await?;

pub fn pair() -> Result<(Async<UnixStream>, Async<UnixStream>)>

Creates an unnamed pair of connected UDS stream sockets.

Examples

use async_io::Async;
use std::os::unix::net::UnixStream;

let (stream1, stream2) = Async::<UnixStream>::pair()?;

impl Async<UnixDatagram>

pub fn bind<P: AsRef<Path>>(path: P) -> Result<Async<UnixDatagram>>

Creates a UDS datagram socket bound to the specified path.

Examples

use async_io::Async;
use std::os::unix::net::UnixDatagram;

let socket = Async::<UnixDatagram>::bind("/tmp/socket")?;

pub fn unbound() -> Result<Async<UnixDatagram>>

Creates a UDS datagram socket not bound to any address.

Examples

use async_io::Async;
use std::os::unix::net::UnixDatagram;

let socket = Async::<UnixDatagram>::unbound()?;

pub fn pair() -> Result<(Async<UnixDatagram>, Async<UnixDatagram>)>

Creates an unnamed pair of connected Unix datagram sockets.

Examples

use async_io::Async;
use std::os::unix::net::UnixDatagram;

let (socket1, socket2) = Async::<UnixDatagram>::pair()?;

pub async fn recv_from(&self, buf: &mut [u8]) -> Result<(usize, UnixSocketAddr)>

Receives data from the socket.

Returns the number of bytes read and the address the message came from.

Examples

use async_io::Async;
use std::os::unix::net::UnixDatagram;

let socket = Async::<UnixDatagram>::bind("/tmp/socket")?;

let mut buf = [0u8; 1024];
let (len, addr) = socket.recv_from(&mut buf).await?;

pub async fn send_to<P: AsRef<Path>>(
    &self,
    buf: &[u8],
    path: P
) -> Result<usize>

Sends data to the specified address.

Returns the number of bytes written.

Examples

use async_io::Async;
use std::os::unix::net::UnixDatagram;

let socket = Async::<UnixDatagram>::unbound()?;

let msg = b"hello";
let addr = "/tmp/socket";
let len = socket.send_to(msg, addr).await?;

pub async fn recv(&self, buf: &mut [u8]) -> Result<usize>

Receives data from the connected peer.

Returns the number of bytes read and the address the message came from.

The connect method connects this socket to a remote address. This method will fail if the socket is not connected.

Examples

use async_io::Async;
use std::os::unix::net::UnixDatagram;

let socket = Async::<UnixDatagram>::bind("/tmp/socket1")?;
socket.get_ref().connect("/tmp/socket2")?;

let mut buf = [0u8; 1024];
let len = socket.recv(&mut buf).await?;

pub async fn send(&self, buf: &[u8]) -> Result<usize>

Sends data to the connected peer.

Returns the number of bytes written.

The connect method connects this socket to a remote address. This method will fail if the socket is not connected.

Examples

use async_io::Async;
use std::os::unix::net::UnixDatagram;

let socket = Async::<UnixDatagram>::bind("/tmp/socket1")?;
socket.get_ref().connect("/tmp/socket2")?;

let msg = b"hello";
let len = socket.send(msg).await?;

Trait Implementations

impl<T> AsMut<T> for Async<T>

fn as_mut(&mut self) -> &mut T

Performs the conversion.

impl<T: AsRawFd> AsRawFd for Async<T>

fn as_raw_fd(&self) -> RawFd

Extracts the raw file descriptor.

impl<T> AsRef<T> for Async<T>

fn as_ref(&self) -> &T

Performs the conversion.

impl<T: Read> AsyncRead for Async<T>

fn poll_read(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
    buf: &mut [u8]
) -> Poll<Result<usize>>

Attempt to read from the AsyncRead into buf.

fn poll_read_vectored(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
    bufs: &mut [IoSliceMut<'_>]
) -> Poll<Result<usize>>

Attempt to read from the AsyncRead into bufs using vectored IO operations.

impl<T> AsyncRead for &Async<T> where
    for<'a> &'a T: Read, 

fn poll_read(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
    buf: &mut [u8]
) -> Poll<Result<usize>>

Attempt to read from the AsyncRead into buf.

fn poll_read_vectored(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
    bufs: &mut [IoSliceMut<'_>]
) -> Poll<Result<usize>>

Attempt to read from the AsyncRead into bufs using vectored IO operations.

impl<T: Write> AsyncWrite for Async<T>

fn poll_write(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
    buf: &[u8]
) -> Poll<Result<usize>>

Attempt to write bytes from buf into the object.

fn poll_write_vectored(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
    bufs: &[IoSlice<'_>]
) -> Poll<Result<usize>>

Attempt to write bytes from bufs into the object using vectored IO operations.

fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Attempt to flush the object, ensuring that any buffered data reach their destination.

fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Attempt to close the object.

impl<T> AsyncWrite for &Async<T> where
    for<'a> &'a T: Write, 

fn poll_write(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
    buf: &[u8]
) -> Poll<Result<usize>>

Attempt to write bytes from buf into the object.

fn poll_write_vectored(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
    bufs: &[IoSlice<'_>]
) -> Poll<Result<usize>>

Attempt to write bytes from bufs into the object using vectored IO operations.

fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Attempt to flush the object, ensuring that any buffered data reach their destination.

fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Attempt to close the object.

impl<T: Debug> Debug for Async<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> Drop for Async<T>

fn drop(&mut self)

Executes the destructor for this type.

impl TryFrom<TcpListener> for Async<TcpListener>

type Error = Error

The type returned in the event of a conversion error.

fn try_from(listener: TcpListener) -> Result<Self>

Performs the conversion.

impl TryFrom<TcpStream> for Async<TcpStream>

type Error = Error

The type returned in the event of a conversion error.

fn try_from(stream: TcpStream) -> Result<Self>

Performs the conversion.

impl TryFrom<UdpSocket> for Async<UdpSocket>

type Error = Error

The type returned in the event of a conversion error.

fn try_from(socket: UdpSocket) -> Result<Self>

Performs the conversion.

impl TryFrom<UnixDatagram> for Async<UnixDatagram>

type Error = Error

The type returned in the event of a conversion error.

fn try_from(socket: UnixDatagram) -> Result<Self>

Performs the conversion.

impl TryFrom<UnixListener> for Async<UnixListener>

type Error = Error

The type returned in the event of a conversion error.

fn try_from(listener: UnixListener) -> Result<Self>

Performs the conversion.

impl TryFrom<UnixStream> for Async<UnixStream>

type Error = Error

The type returned in the event of a conversion error.

fn try_from(stream: UnixStream) -> Result<Self>

Performs the conversion.

impl<T> Unpin for Async<T>