Getting Started with Tokio

If you're interested in writing servers in Rust, Tokio is a project that should definitely be on your radar. It promises to leverage the exciting new work on futures in Rust to make it easy to build extremely efficient networked servers with natural, composable code.

It's also a project that can be hard to wrap your head around when you first encounter it. In this post, I'll attempt to illustrate how the various pieces of Tokio fit together by building a very simple server on top of a very simple protocol.

As a disclaimer before we get any further, I want to emphasize that I am very much a beginner when it comes to Tokio and much of what follows is, at best, woefully incomplete. Also, the project has not reached a stable version at the time of this writing, so it's possible that these APIs will change.

Update: In the time since I originally wrote this, the 0.1 releases of all the tokio crates have been published and the APIs have stabilized a bit. I've update the code here to match the released APIs and added a section on the new tokio-io crate.

Introduction

With that out of the way, what is Tokio? The project's most direct inspiration was Scala's Finagle. Unlike the Scala and the JVM, however, Rust didn't have mature asynchronous I/O libraries on which to build. As a result, Tokio ends up being more like Finagle + Netty than Finagle alone.

Coming from the Ruby world, I've found that Rack is also a useful point of comparison when trying to understand Tokio. Like Rack, Tokio provides two major benefits for application developers looking to write servers:

A simple, standard interface for writing services A design encouraging the creation of generic, reusable middleware

As you might expect from projects in languages as different as Ruby and Rust, the similarities pretty much end there. The list of differences is long, but the two most important are that Tokio is not specific to HTTP and that it derives a great deal of power from the Rust type system in a way that Rack obviously cannot.

Before we dive into any code, I'll try to give a brief overview of how the Tokio project is organized and how the pieces fit together. Layering is a very important concept to Tokio and that's represented in the organization of the project.

To start, there are two projects outside of Tokio itself that help put it in context. The first is mio and the second is futures. I won't cover these in detail, but a short summary is that mio provides a thin wrapper over OS mechanisms for asynchronous I/O (e.g. epoll ), and futures provides an efficient, composable interface for asynchronous tasks (think Result , but async).

The base layer of Tokio, tokio-core, is an integration of mio with futures. This sounds pretty obvious at a high level, but it makes a huge difference in usability when compared with mio itself. It gives you things like your normal TcpListener and UdpSocket , but asynchronous, as well as a reactor (i.e. event loop) on which to run them.

Next, there's tokio-io. This crate was introduced as an extraction from tokio-core and includes some fundamental traits encapsulating async I/O. These traits express the semantics of raw byte streams and the logic of turning them into discrete "frames", or meaningful chunks of bytes. An implementation would allow you start dealing with things like HTTP header frames or Websocket data frames instead of a stream of raw bytes. It also allows implementers of these traits to avoid a hard dependency on a specific runtime like tokio-core.

Going up a layer, we find tokio-proto. The point of this crate is to help you turn those frames from tokio-core into full requests and responses for whatever network protocol you're using (e.g. HTTP). It has all kinds of tools to help protocol implementers handle pipelining, multiplexing, streaming, etc.

Finally, at the top of the stack and of the most interest to application developers, there is tokio-service. This is a very small crate and exists pretty much just to define the Service trait:

trait Service { type Request ; type Response ; type Error ; type Future : Future < Item = Self :: Response , Error = Self :: Error > ; fn call ( & self , req : Self :: Request ) -> Self :: Future ; }

In English, this means you implement a service by defining what you accept, what you return, how you can fail, and a function that contains the logic. If you get rid of the types, this should look a lot like Rack. Just like with Rack, it's easy to imagine a service that wraps another service, transforming the input and output arbitrarily.

If you're an application developer, this is really the only layer you'll need to care about. The idea is that the community will provide high-quality implementations of common protocols and you'll be free to layer services on top while completely forgetting that tokio-core, tokio-io, and tokio-proto even exist.

Building an example server

As a learning exercise, we'll now implement a server from the ground (i.e. tokio-core) up, based on a simple line protocol. When we're done, we should be able to connect with nc and exchange some simple requests and responses.

To get started, we'll mostly be following the example from the tokio-proto docs. The basic steps are:

Implement a codec for our protocol Implement a ServerProto for our codec Implement a Service that can run on top of our server Use the provided TcpServer to run our service

Implementing the codec

Since we don't need to keep track of any state, we'll start with a unit struct:

pub struct LineCodec ;

Then we'll implement the Encoder and Decoder traits for our struct, which require us to define the types of our input and output frames as well as functions for encoding and decoding them. In our case, we'll be dealing with newline-delimited strings going in each direction. Luckily, the tokio-line project was recently updated and has some good example code we can borrow:

// Borrowed from tokio-line with slight modifications impl Decoder for LineCodec { type Item = String ; type Error = io :: Error ; fn decode ( & mut self , buf : & mut BytesMut ) -> Result < Option < Self :: Item > , io :: Error > { // If our buffer contains a newline... if let Some ( n ) = buf . as_ref (). iter (). position ( | b | * b == b'

' ) { // remove this line and the newline from the buffer. let line = buf . split_to ( n ); buf . split_to ( 1 ); // Also remove the '

'. // Turn this data into a UTF-8 string and return it return match str :: from_utf8 ( line . as_ref ()) { Ok ( s ) => Ok ( Some ( s . to_string ())), Err ( _ ) => Err ( io :: Error :: new ( io :: ErrorKind :: Other , "invalid string" )), } } // Otherwise, we don't have enough data for a full message yet Ok ( None ) } } impl Encoder for LineCodec { type Item = String ; type Error = io :: Error ; fn encode ( & mut self , msg : Self :: Item , buf : & mut BytesMut ) -> io :: Result < () > { for byte in msg . as_bytes () { buf . put_u8 ( * byte ); } buf . put_u8 ( b'

' ); Ok (()) } }

That's a lot to digest, so don't worry if the details aren't perfectly clear. The important part is that we declare that we're dealing with strings going in and out, and we provide functions to read and write those strings with newline delimiters.

Implementing the server protocol

Next, we define another unit struct for our ServerProto . There's no real logic here; we're just declaring how different pieces fit together:

struct LineServerProto ; impl < T : AsyncRead + AsyncWrite + 'static > ServerProto < T > for LineServerProto { type Request = String ; type Response = String ; type Transport = Framed < T , LineCodec > ; type BindTransport = Result < Self :: Transport , io :: Error > ; fn bind_transport ( & self , io : T ) -> Self :: BindTransport { Ok ( io . framed ( LineCodec )) } }

The first thing to notice is that we're generic over any type T that implements the AsyncRead and AsyncWrite traits from tokio-io. This just means that our server can run on top of any I/O-like object (i.e. anything that lets it read and write bytes). This object is then combined with our codec into a Framed object, also from tokio-io.

The idea here is that the decode function of our codec knows how to "chunk" the bytes from our I/O object into a stream of discrete messages (i.e. frames). In the other direction, encode takes outgoing messages and writes them out as a stream of bytes. This allows our higher level services to deal only with full messages.

Implementing the service

At this point, we have almost everything we need to instantiate a tokio-proto TcpServer and begin serving requests. The only thing missing is an actual Service to run on top of our protocol:

struct HelloWorldService ; impl Service for HelloWorldService { type Request = String ; type Response = String ; type Error = io :: Error ; type Future = FutureResult < Self :: Response , Self :: Error > ; fn call ( & self , req : String ) -> Self :: Future { if req . contains ( '

' ) { futures :: failed ( io :: Error :: new ( io :: ErrorKind :: InvalidInput , "message contained new line" )) } else { let resp = match req . as_str () { "hello" => "world" . to_string (), _ => "idk" . to_string (), }; futures :: finished ( resp ) } } }

The payoff here is that we get to write a relatively vanilla Rust function that operates on arbitrary Rust types, using futures but not caring at all about things like TCP or serialization formats.

Running the server

To actually run this thing, all that's left is to feed it to a TcpServer along with our protocol and an address to bind to:

fn main () { let addr = "0.0.0.0:12345" . parse (). unwrap (); TcpServer :: new ( LineServerProto , addr ) . serve ( || Ok ( HelloWorldService )); }

Then we can connect and take it for a spin:

$ nc localhost 12345 hello world what? idk ^C

And it seems to work! You can get the full code here if you want to try it out yourself.

Hopefully that was an informative high-level tour of Tokio, and hopefully I didn't get anything too terribly wrong. You could take this example a lot further, layering middleware for timeouts or request parsing, exploring the client side of what tokio-proto provides, etc, etc, but I think this post is long enough as it is.