I/O

I/O in Tokio operates in much the same way as in std, but asynchronously. There is a trait for reading (AsyncRead) and a trait for writing (AsyncWrite). Specific types implement these traits as appropriate (TcpStream, File, Stdout). AsyncRead and AsyncWrite are also implemented by a number of data structures, such as Vec<u8> and &[u8]. This allows using byte arrays where a reader or writer is expected.

This page will cover basic I/O reading and writing with Tokio and work through a few examples. The next page will get into a more advanced I/O example.

AsyncRead and AsyncWrite

These two traits provide the facilities to asynchronously read from and write to byte streams. The methods on these traits are typically not called directly, similar to how you don’t manually call the poll method from the Future trait. Instead, you will use them through the utility methods provided by AsyncReadExt and AsyncWriteExt.

Let’s briefly look at a few of these methods. All of these functions are async and must be used with .await.

async fn read()

AsyncReadExt::read provides an async method for reading data into a buffer, returning the number of bytes read.

Note: when read() returns Ok(0), this signifies that the stream is closed. Any further calls to read() will complete immediately with Ok(0). With TcpStream instances, this signifies that the read half of the socket is closed.

  1. use tokio::fs::File;
  2. use tokio::io::{self, AsyncReadExt};
  4. #[tokio::main]
  5. async fn main() -> io::Result<()> {
  6.     let mut f = File::open("foo.txt").await?;
  7.     let mut buffer = [0; 10];
  9.     // read up to 10 bytes
  10.     let n = f.read(&mut buffer[..]).await?;
  12.     println!("The bytes: {:?}", &buffer[..n]);
  13.     Ok(())
  14. }

async fn read_to_end()

AsyncReadExt::read_to_end reads all bytes from the stream until EOF.

  1. use tokio::io::{self, AsyncReadExt};
  2. use tokio::fs::File;
  4. #[tokio::main]
  5. async fn main() -> io::Result<()> {
  6.     let mut f = File::open("foo.txt").await?;
  7.     let mut buffer = Vec::new();
  9.     // read the whole file
  10.     f.read_to_end(&mut buffer).await?;
  11.     Ok(())
  12. }

async fn write()

AsyncWriteExt::write writes a buffer into the writer, returning how many bytes were written.

  1. use tokio::io::{self, AsyncWriteExt};
  2. use tokio::fs::File;
  4. #[tokio::main]
  5. async fn main() -> io::Result<()> {
  6.     let mut file = File::create("foo.txt").await?;
  8.     // Writes some prefix of the byte string, but not necessarily all of it.
  9.     let n = file.write(b"some bytes").await?;
  11.     println!("Wrote the first {} bytes of 'some bytes'.", n);
  12.     Ok(())
  13. }

async fn write_all()

AsyncWriteExt::write_all writes the entire buffer into the writer.

  1. use tokio::io::{self, AsyncWriteExt};
  2. use tokio::fs::File;
  4. #[tokio::main]
  5. async fn main() -> io::Result<()> {
  6.     let mut buffer = File::create("foo.txt").await?;
  8.     buffer.write_all(b"some bytes").await?;
  9.     Ok(())
  10. }

Both traits include a number of other helpful methods. See the API docs for a comprehensive list.

Helper functions

Additionally, just like std, the tokio::io module contains a number of helpful utility functions as well as APIs for working with standard input, standard output and standard error. For example, tokio::io::copy asynchronously copies the entire contents of a reader into a writer.

  1. use tokio::fs::File;
  2. use tokio::io;
  4. #[tokio::main]
  5. async fn main() -> io::Result<()> {
  6.     let mut reader: &[u8] = b"hello";
  7.     let mut file = File::create("foo.txt").await?;
  9.     io::copy(&mut reader, &mut file).await?;
  10.     Ok(())
  11. }

Note that this uses the fact that byte arrays also implement AsyncRead.

Echo server

Let’s practice doing some asynchronous I/O. We will be writing an echo server.

The echo server binds a TcpListener and accepts inbound connections in a loop. For each inbound connection, data is read from the socket and written immediately back to the socket. The client sends data to the server and receives the exact same data back.

We will implement the echo server twice, using slightly different strategies.

Using io::copy()

To start, we will implement the echo logic using the io::copy utility.

You can write up this code in a new binary file:

  1. touch src/bin/echo-server-copy.rs

That you can launch (or just check the compilation) with:

  1. cargo run --bin echo-server-copy

You will be able to try the server using a standard command-line tool such as telnet, or by writing a simple client like the one found in the documentation for tokio::net::TcpStream.

This is a TCP server and needs an accept loop. A new task is spawned to process each accepted socket.

  1. use tokio::io;
  2. use tokio::net::TcpListener;
  4. #[tokio::main]
  5. async fn main() -> io::Result<()> {
  6.     let listener = TcpListener::bind("127.0.0.1:6142").await.unwrap();
  8.     loop {
  9.         let (mut socket, _) = listener.accept().await?;
  11.         tokio::spawn(async move {
  12.             // Copy data here
  13.         });
  14.     }
  15. }

As seen earlier, this utility function takes a reader and a writer and copies data from one to the other. However, we only have a single TcpStream. This single value implements both AsyncRead and AsyncWrite. Because io::copy requires &mut for both the reader and the writer, the socket cannot be used for both arguments.

  1. // This fails to compile
  2. io::copy(&mut socket, &mut socket).await

Splitting a reader + writer

To work around this problem, we must split the socket into a reader handle and a writer handle. The best way to split a reader/writer combo depends on the specific type.

Any reader + writer type can be split using the io::split utility. This function takes a single value and returns separate reader and writer handles. These two handles can be used independently, including from separate tasks.

For example, the echo client could handle concurrent reads and writes like this:

  1. use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
  2. use tokio::net::TcpStream;
  4. #[tokio::main]
  5. async fn main() -> io::Result<()> {
  6.     let socket = TcpStream::connect("127.0.0.1:6142").await?;
  7.     let (mut rd, mut wr) = io::split(socket);
  9.     // Write data in the background
  10.     let write_task = tokio::spawn(async move {
  11.         wr.write_all(b"hello\r\n").await?;
  12.         wr.write_all(b"world\r\n").await?;
  14.         // Sometimes, the rust type inferencer needs
  15.         // a little help
  16.         Ok::<_, io::Error>(())
  17.     });
  19.     let mut buf = vec![0; 128];
  21.     loop {
  22.         let n = rd.read(&mut buf).await?;
  24.         if n == 0 {
  25.             break;
  26.         }
  28.         println!("GOT {:?}", &buf[..n]);
  29.     }
  31.     Ok(())
  32. }

Because io::split supports any value that implements AsyncRead + AsyncWrite and returns independent handles, internally io::split uses an Arc and a Mutex. This overhead can be avoided with TcpStream. TcpStream offers two specialized split functions.

TcpStream::split takes a reference to the stream and returns a reader and writer handle. Because a reference is used, both handles must stay on the same task that split() was called from. This specialized split is zero-cost. There is no Arc or Mutex needed. TcpStream also provides into_split which supports handles that can move across tasks at the cost of only an Arc.

Because io::copy() is called on the same task that owns the TcpStream, we can use TcpStream::split. The task that processes the echo logic becomes:

  1. tokio::spawn(async move {
  2.     let (mut rd, mut wr) = socket.split();
  4.     if io::copy(&mut rd, &mut wr).await.is_err() {
  5.         eprintln!("failed to copy");
  6.     }
  7. });

You can find the entire code here.

Manual copying

Now let’s look at how we would write the echo server by copying the data manually. To do this, we use AsyncReadExt::read and AsyncWriteExt::write_all.

The full echo server is as follows:

  1. use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
  2. use tokio::net::TcpListener;
  4. #[tokio::main]
  5. async fn main() -> io::Result<()> {
  6.     let listener = TcpListener::bind("127.0.0.1:6142").await.unwrap();
  8.     loop {
  9.         let (mut socket, _) = listener.accept().await?;
  11.         tokio::spawn(async move {
  12.             let mut buf = vec![0; 1024];
  14.             loop {
  15.                 match socket.read(&mut buf).await {
  16.                     // Return value of `Ok(0)` signifies that the remote has
  17.                     // closed
  18.                     Ok(0) => return,
  19.                     Ok(n) => {
  20.                         // Copy the data back to socket
  21.                         if socket.write_all(&buf[..n]).await.is_err() {
  22.                             // Unexpected socket error. There isn't much we can
  23.                             // do here so just stop processing.
  24.                             return;
  25.                         }
  26.                     }
  27.                     Err(_) => {
  28.                         // Unexpected socket error. There isn't much we can do
  29.                         // here so just stop processing.
  30.                         return;
  31.                     }
  32.                 }
  33.             }
  34.         });
  35.     }
  36. }

(You can put this code into src/bin/echo-server.rs and launch it with cargo run --bin echo-server).

Let’s break it down. First, since the AsyncRead and AsyncWrite utilities are used, the extension traits must be brought into scope.

  1. use tokio::io::{self, AsyncReadExt, AsyncWriteExt};

Allocating a buffer

The strategy is to read some data from the socket into a buffer then write the contents of the buffer back to the socket.

  1. let mut buf = vec![0; 1024];

A stack buffer is explicitly avoided. Recall from earlier, we noted that all task data that lives across calls to .await must be stored by the task. In this case, buf is used across .await calls. All task data is stored in a single allocation. You can think of it as an enum where each variant is the data that needs to be stored for a specific call to .await.

If the buffer is represented by a stack array, the internal structure for tasks spawned per accepted socket might look something like:

  1. struct Task {
  2.     // internal task fields here
  3.     task: enum {
  4.         AwaitingRead {
  5.             socket: TcpStream,
  6.             buf: [BufferType],
  7.         },
  8.         AwaitingWriteAll {
  9.             socket: TcpStream,
  10.             buf: [BufferType],
  11.         }
  13.     }
  14. }

If a stack array is used as the buffer type, it will be stored inline in the task structure. This will make the task structure very big. Additionally, buffer sizes are often page sized. This will, in turn, make Task an awkward size: $page-size + a-few-bytes.

The compiler optimizes the layout of async blocks further than a basic enum. In practice, variables are not moved around between variants as would be required with an enum. However, the task struct size is at least as big as the largest variable.

Because of this, it is usually more efficient to use a dedicated allocation for the buffer.

Handling EOF

When the read half of the TCP stream is shut down, a call to read() returns Ok(0). It is important to exit the read loop at this point. Forgetting to break from the read loop on EOF is a common source of bugs.

  1. loop {
  2.     match socket.read(&mut buf).await {
  3.         // Return value of `Ok(0)` signifies that the remote has
  4.         // closed
  5.         Ok(0) => return,
  6.         // ... other cases handled here
  7.     }
  8. }

Forgetting to break from the read loop usually results in a 100% CPU infinite loop situation. As the socket is closed, socket.read() returns immediately. The loop then repeats forever.

Full code can be found here.