Reading and Writing Data

Non-blocking I/O

In the overview we mentioned briefly that Tokio’s I/O types implementnon-blocking variants of std::Read and std::Write calledAsyncRead and AsyncWrite. These are an integral part of Tokio’sI/O story, and are important to understand when working with I/O code.

Note: in this section, we’ll primarily talk about AsyncRead, butAsyncWrite is pretty much exactly the same, just for writing data toan I/O resource (like a TCP socket) instead of reading from it.

So, let’s take a look at AsyncRead and see what all the fuss isabout:

use std::io::Read;
pub trait AsyncRead: Read {
    // ...
    // various provided methods
    // ...
}

Huh. What’s going on here? Well, AsyncRead is really just Readfrom std::io, along with an additional contract. The documentationfor AsyncRead reads:

This trait inherits from std::io::Read and indicates that an I/Oobject is non-blocking. All non-blocking I/O objects must return anerror when bytes are unavailable instead of blocking the currentthread.

That last part is critical. If you implement AsyncRead for a type, youare promising that calling read on it will never block. Instead,you are expected to return an io::WouldBlock error toindicate that the operation would have blocked (for example becausethere was no more data available) if it wasn’t non-blocking. Theprovided poll_read method relies on this:

fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, std::io::Error> {
    match self.read(buf) {
        Ok(t) => Ok(Async::Ready(t)),
        Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
            Ok(Async::NotReady)
        }
        Err(e) => Err(e),
    }
}

This code should look familiar. If you squint a little, poll_readlooks a lot like Future::poll. And that’s because that’s almostexactly what it is! A type that implements AsyncRead essentiallybehaves like a future that you can try to read data out of, and it willinform you whether it is Ready (and some data was read) or NotReady(and you’ll have to poll_read again later).

Working with I/O futures

Since AsyncRead (and AsyncWrite) are pretty much futures, you caneasily embed them in your own futures and poll_read them just as youwould poll any other embedded Future. You can even use try_ready!to propagate errors and NotReady as appropriate. We’ll talk more aboutdirectly using these traits in the next section. However, to make lifesimpler in a number of situations, Tokio provides a number of usefulcombinators in tokio::io for performing common I/O operations on topof AsyncRead and AsyncWrite. In general, these provide wrappersaround AsyncRead or AsyncWrite types that implement Future andthat complete when a given read or write operation has completed.

The first handy I/O combinator is read_exact. It takes a mutablebuffer (&mut [u8]) and an implementor of AsyncRead as arguments, andreturns a Future that reads exactly enough bytes to fill the buffer.Internally the returned future just keeps track of how many bytes it hasread thus far, and continues to issue poll_ready on the AsyncRead(returning NotReady if necessary) until it has exactly filled thebuffer. At that point, it returns Ready(buf) with the filled buffer.Let’s take a look:

# extern crate tokio;
use tokio::net::tcp::TcpStream;
use tokio::prelude::*;
# fn main() {
let addr = "127.0.0.1:12345".parse().unwrap();
let read_8_fut = TcpStream::connect(&addr)
    .and_then(|stream| {
        // We need to create a buffer for read_exact to write into.
        // A Vec<u8> is a good starting point.
        // read_exact will read buffer.len() bytes, so we need
        // to make sure the Vec isn't empty!
        let mut buf = vec![0; 8];
        // read_exact returns a Future that resolves when
        // buffer.len() bytes have been read from stream.
        tokio::io::read_exact(stream, buf)
    })
    .inspect(|(_stream, buf)| {
        // Notice that we get both the buffer and the stream back
        // here, so that we can now continue using the stream to
        // send a reply for example.
        println!("got eight bytes: {:x?}", buf);
    });
// We can now either chain more futures onto read_8_fut,
// or if all we wanted to do was read and print those 8
// bytes, we can just use tokio::run to run it (taking
// care to map Future::Item and Future::Error to ()).
# }

A second I/O combinator that is often useful is write_all. It takesa buffer (&[u8]) and an implementor of AsyncWrite as arguments, andreturns a Future that writes out all the bytes of the buffer into theAsyncWrite using poll_write. When the Future resolves, the entirebuffer has been written out and flushed. We can combine this withread_exact to echo whatever the server says back to it:

# extern crate tokio;
use tokio::net::tcp::TcpStream;
use tokio::prelude::*;
# fn main() {
# let addr = "127.0.0.1:12345".parse().unwrap();
let echo_fut = TcpStream::connect(&addr)
    .and_then(|stream| {
        // We're going to read the first 32 bytes the server sends us
        // and then just echo them back:
        let mut buf = vec![0; 32];
        // First, we need to read the server's message
        tokio::io::read_exact(stream, buf)
    })
    .and_then(|(stream, buf)| {
        // Then, we use write_all to write the entire buffer back:
        tokio::io::write_all(stream, buf)
    })
    .inspect(|(_stream, buf)| {
        println!("echoed back {} bytes: {:x?}", buf.len(), buf);
    });
// As before, we can chain more futures onto echo_fut,
// or declare ourselves finished and run it with tokio::run.
# }

Tokio also comes with an I/O combinator to implement this kind ofcopying. It is (perhaps unsurprisingly) called copy. copy takesan AsyncRead and an AsyncWrite, and continuously writes all thebytes read out from the AsyncRead into the AsyncWrite untilpollread indicates that the input has been closed and all the byteshave been written out and flushed to the output. This is the combinatorwe used in our echo server! It greatly simplifies our example fromabove, and also makes it work for _any amount of server data!

# extern crate tokio;
use tokio::net::tcp::TcpStream;
use tokio::prelude::*;
# fn main() {
# let addr = "127.0.0.1:12345".parse().unwrap();
let echo_fut = TcpStream::connect(&addr)
    .and_then(|stream| {
        // First, we need to get a separate read and write handle for
        // the connection so that we can forward one to the other.
        // See "Split I/O resources" below for more details.
        let (reader, writer) = stream.split();
        // Then, we can use copy to send all the read bytes to the
        // writer, and return how many bytes it read/wrote.
        tokio::io::copy(reader, writer)
    })
    .inspect(|(bytes_copied, r, w)| {
        println!("echoed back {} bytes", bytes_copied);
    });
# }

Pretty neat!

The combinators we’ve talked about so far are all for pretty low-leveloperations: read these bytes, write these bytes, copy these bytes. Oftentimes though, you want to operate on higher-level representations, like“lines”. Tokio has you covered there too! lines takes anAsyncRead, and returns a Stream that yields each line from the inputuntil there are no more lines to read:

# extern crate tokio;
use tokio::net::tcp::TcpStream;
use tokio::prelude::*;
# fn main() {
# let addr = "127.0.0.1:12345".parse().unwrap();
let lines_fut = TcpStream::connect(&addr).and_then(|stream| {
    // We want to parse out each line we receive on stream.
    // To do that, we may need to buffer input for a little while
    // (if the server sends two lines in one packet for example).
    // Because of that, lines requires that the AsyncRead it is
    // given *also* implements BufRead. This may be familiar if
    // you've ever used the lines() method from std::io::BufRead.
    // Luckily, BufReader from the standard library gives us that!
    let stream = std::io::BufReader::new(stream);
    tokio::io::lines(stream).for_each(|line| {
        println!("server sent us the line: {}", line);
        // This closure is called for each line we receive,
        // and returns a Future that represents the work we
        // want to do before accepting the next line.
        // In this case, we just wanted to print, so we
        // don't need to do anything more.
        Ok(())
    })
});
# }

There are also plenty more I/O combinators in tokio::io that you maywant to take a look at before you decide to write your own!

Split I/O resources

Both the copy example above and the echo server contained thismysterious-looking snippet:

let (reader, writer) = socket.split();
let bytes_copied = tokio::io::copy(reader, writer);

As the comment above it explains, we split the TcpStream (socket)into a read “half” and a write “half”, and use the copy combinatorwe discussed above to produce a Future that asynchronously copies allthe data from the read half to the write half. But why is this “split”required in the first place? After all, AsyncRead::poll_ready andAsyncWrite::poll_write just take &mut self.

To answer that, we need to think back to Rust’s ownership system alittle. Recall that Rust only allows you to have a single mutablereference to a given variable at a time. But we have to pass two_arguments to copy, one for where to read from, and one for where towrite to. However, once we pass a mutable reference to the TcpStreamas one of the arguments, we cannot also construct a second mutablereference to it to pass as the second argument! _We know that copywon’t read and write at the same time to those, but that’s notexpressed in copy’s types.

Enter split, a provided method on the AsyncRead trait when thetype also implements AsyncWrite. If we look at the signature, we see

fn split(self) -> (ReadHalf<Self>, WriteHalf<Self>)
  where Self: AsyncWrite { ... }

The returned ReadHalf implements AsyncRead, and the WriteHalfimplements AsyncWrite. And crucially, we now have two _separate_pointers into our type, which we can pass around separately. This comesin handy for copy, but it also means that we can pass each half toa different future, and handle the reads and writes completelyindependently! Behind the scenes, split ensures that if we both tryto read and write at the same time, only one of them happen at a time.

Transports

Turning an AsyncRead into a Stream (like lines does) or anAsyncWrite into a Sink is pretty common in applications that need todo I/O. They often want to abstract away the way bytes are retrievedfrom or put on the wire, and let most of their application code dealwith more convenient “requests” and “response” types. This is oftenknown as “framing”: instead of viewing your connections as consisting ofjust bytes in/bytes out, you view them as “frames” of application datathat are received and sent. A framed stream of bytes is often referredto as a “transport”.

Transports are typically implemented using a codec. For example,lines represents a very simple codec that separates a byte string bythe newline character, \n, and parses each frame as a string beforepassing it to the application. Tokio provides helpers for implementingnew codecs in tokio::codec; you implement the Encoder andDecoder traits for your transport, and use Framed::new to makea Sink + Stream from your byte stream (like a TcpStream). It’salmost like magic! There are versions for doing just the read or writeside of a codec too (like lines). Let’s take a look at writing asimple implementation of a line-based codec (even though LinesCodecexists):

# extern crate tokio;
extern crate bytes;
use bytes::{BufMut, BytesMut};
use tokio::codec::{Decoder, Encoder};
use tokio::prelude::*;
// This is where we'd keep track of any extra book-keeping information
// our transport needs to operate.
struct LinesCodec;
// Turns string errors into std::io::Error
fn bad_utf8<E>(_: E) -> std::io::Error {
    std::io::Error::new(std::io::ErrorKind::InvalidData, "Unable to decode input as UTF8")
}
// First, we implement encoding, because it's so straightforward.
// Just write out the bytes of the string followed by a newline!
// Easy-peasy.
impl Encoder for LinesCodec {
    type Item = String;
    type Error = std::io::Error;
    fn encode(&mut self, line: Self::Item, buf: &mut BytesMut) -> Result<(), Self::Error> {
        // Note that we're given a BytesMut here to write into.
        // BytesMut comes from the bytes crate, and aims to give
        // efficient read/write access to a buffer. To use it,
        // we have to reserve memory before we try to write to it.
        buf.reserve(line.len() + 1);
        // And now, we write out our stuff!
        buf.put(line);
        buf.put_u8(b'\n');
        Ok(())
    }
}
// The decoding is a little trickier, because we need to look for
// newline characters. We also need to handle *two* cases: the "normal"
// case where we're just asked to find the next string in a bunch of
// bytes, and the "end" case where the input has ended, and we need
// to find any remaining strings (the last of which may not end with a
// newline!
impl Decoder for LinesCodec {
    type Item = String;
    type Error = std::io::Error;
    // Find the next line in buf!
    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        Ok(if let Some(offset) = buf.iter().position(|b| *b == b'\n') {
            // We found a newline character in this buffer!
            // Cut out the line from the buffer so we don't return it again.
            let mut line = buf.split_to(offset + 1);
            // And then parse it as UTF-8
            Some(
                std::str::from_utf8(&line[..line.len() - 1])
                    .map_err(bad_utf8)?
                    .to_string(),
            )
        } else {
            // There are no newlines in this buffer, so no lines to speak of.
            // Tokio will make sure to call this again when we have more bytes.
            None
        })
    }
    // Find the next line in buf when there will be no more data coming.
    fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        Ok(match self.decode(buf)? {
            Some(frame) => {
                // There's a regular line here, so we may as well just return that.
                Some(frame)
            },
            None => {
                // There are no more lines in buf!
                // We know there are no more bytes coming though,
                // so we just return the remainder, if any.
                if buf.is_empty() {
                    None
                } else {
                    Some(
                        std::str::from_utf8(&buf.take()[..])
                            .map_err(bad_utf8)?
                            .to_string(),
                    )
                }
            }
        })
    }
}

Next up: Using AsyncRead and AsyncWrite directly