I/O with Tokio

The tokio crate comes with TCP and UDP networking types. Unlike the types instd, Tokio’s networking types are based on the poll model and will notify thetask executors when their readiness states change (data is received and writebuffers are flushed). In the tokio::net module you’ll find types likeTcpListener, TcpStream, and UdpSocket.

All of these types provide both a future API as well as a pollAPI.

The Tokio net types are powered by a Mio based reactor that, by default, isstarted up lazily on a background thread. See reactor documentation for moredetails.

Using the Future API

We’ve already seen some of this earlier in the guide with the incomingfunction as well as the helpers found in tokio_io::io.

These helpers include:

  • incoming: A stream of inbound TCP connections.
  • read_exact: Read exactly n bytes into a buffer.
  • read_to_end: Read all bytes into a buffer.
  • write_all: Write the entire contents of a buffer.

  • copy: Copy bytes from one I/O handle to another.
    A lot of these functions / helpers are generic over the AsyncRead andAsyncWrite traits. These traits are similar to Read and Write fromstd, but are only for types that are “future aware”, i.e. follow themandated properties:

  • Calls to read or write are nonblocking, they never block the callingthread.

  • If a call would otherwise block then the function returns a value indicating so.If this happens then the current future’s task is scheduled to receive anotification when the I/O is ready again.
    Note that users of AsyncRead and AsyncWrite types should usepoll_read and poll_write instead of directly calling read and write.

For example, here is how to accept connections, read 5 bytes from them, thenwrite the 5 bytes back to the socket:

  1. # #![deny(deprecated)]
  2. # extern crate tokio;
  3. #
  4. # use tokio::io;
  5. # use tokio::net::TcpListener;
  6. # use tokio::prelude::*;
  7. # fn main() {
  8. #     let addr = "127.0.0.1:6142".parse().unwrap();
  9. #     let listener = TcpListener::bind(&addr).unwrap();
  10. let server = listener.incoming().for_each(|socket| {
  11.     println!("accepted socket; addr={:?}", socket.peer_addr().unwrap());
  13.     let buf = vec![0; 5];
  15.     let connection = io::read_exact(socket, buf)
  16.         .and_then(|(socket, buf)| {
  17.             io::write_all(socket, buf)
  18.         })
  19.         .then(|_| Ok(())); // Just discard the socket and buffer
  21.     // Spawn a new task that processes the socket:
  22.     tokio::spawn(connection);
  24.     Ok(())
  25. })
  26. # ;
  27. # }

Using the Poll API

The Poll based API is to be used when implementing Future by hand and you needto return Async. This is useful when you need to implement your owncombinators that handle custom logic.

For example, this is how the read_exact future could be implemented for aTcpStream.

  1. # #![deny(deprecated)]
  2. # extern crate tokio;
  3. # #[macro_use]
  4. # extern crate futures;
  5. # use tokio::io;
  6. # use tokio::prelude::*;
  7. #
  8. # use tokio::net::TcpStream;
  9. # use std::mem;
  10. pub struct ReadExact {
  11.     state: State,
  12. }
  14. enum State {
  15.     Reading {
  16.         stream: TcpStream,
  17.         buf: Vec<u8>,
  18.         pos: usize,
  19.     },
  20.     Empty,
  21. }
  23. impl Future for ReadExact {
  24.     type Item = (TcpStream, Vec<u8>);
  25.     type Error = io::Error;
  27.     fn poll(&mut self) -> Result<Async<Self::Item>, io::Error> {
  28.         match self.state {
  29.             State::Reading {
  30.                 ref mut stream,
  31.                 ref mut buf,
  32.                 ref mut pos
  33.             } => {
  34.                 while *pos < buf.len() {
  35.                     let n = try_ready!({
  36.                         stream.poll_read(&mut buf[*pos..])
  37.                     });
  38.                     *pos += n;
  39.                     if n == 0 {
  40.                         let err = io::Error::new(
  41.                             io::ErrorKind::UnexpectedEof,
  42.                             "early eof");
  44.                         return Err(err)
  45.                     }
  46.                 }
  47.             }
  48.             State::Empty => panic!("poll a ReadExact after it's done"),
  49.         }
  51.         match mem::replace(&mut self.state, State::Empty) {
  52.             State::Reading { stream, buf, .. } => {
  53.                 Ok(Async::Ready((stream, buf)))
  54.             }
  55.             State::Empty => panic!(),
  56.         }
  57.     }
  58. }
  59. # pub fn main() {}

Datagrams

Note that most of this discussion has been around I/O or byte streams, whichUDP importantly is not! To accommodate this, however, the UdpSocket typealso provides a number of methods for working with it conveniently:

  • send_dgram allows you to express sending a datagram as a future, returningan error if the entire datagram couldn’t be sent at once.
  • recv_dgram expresses reading a datagram into a buffer, yielding both thebuffer and the address it came from.
    Next up: Example: A Chat Server