Asynchronous Servers In Rust

Hello everyone. I’ve wanted to use async i/o in Rust for some time but the verbosity of Mio, the generally accepted Rust async library was holding me back. With the recent release of Tokio I wanted to give it another go. In case you don’t know, Tokio is a library that is built on top of Mio and it aims to make writing clients and servers as easy as possible.

We’re going to be writing a QOTD or Quote of the Day server. In simple terms; it waits for a connection, sends a random quote and closes the connection. You can find more detailed information in my wiki here. This kind of server is especially easy to write since there is no connection state to keep. We will keep our quotes in a text file seperated by %. You can find an example file here.

Reading the quotes

First of all, let’s read our quote file to a vector of String’s.

fn read_quotes(file: &str) -> Vec<String> {
    let mut quotes = Vec::new();
    let mut quote = String::new();

    let quote_file = File::open(file).unwrap();
    for line in BufReader::new(quote_file).lines() {
        if let Ok(line) = line { // If the read is successful
            if line == "%" {
                quotes.push(quote.clone());
                quote.clear();
            }else {
                quote.push_str(&line);
                quote.push('\n');
            }
        }
    }
    quotes.push(quote);
    return quotes;
}

This function reads the file line by line into a String and pushes that String into our vector when it encounters a %. We could accomplish the same thing by reading the whole thing at once and splitting it by % but our method is much more memory friendly.

We can call this function like read_quotes("quotes.txt") and it would return a vector of quotes from quotes.txt.

Writing the server struct

In order to receive events related to sockets, we’re going to make a struct to keep any server state. This struct will need to implement the Task trait from Tokio. Our struct will hold the TcpListener that’s used to accept connections and the vector that holds our quotes.

struct Listener {
    socket: TcpListener,
    quotes: Vec<String>
}

Now let’s implement the Task trait. To do that, we will need to write a function called tick that will be called to handle any events. In that function we’ll accept any connections and send the quotes.

impl Task for Listener {
    fn tick(&mut self) -> io::Result<Tick> {
        // Accept all the sockets until there are no more
        while let Some(mut conn) = try!(self.socket.accept()) {
            // Pick a random quote from the server struct and send it
            // to the client.
            let quote = thread_rng().choose(&self.quotes).unwrap();
            try!(conn.write_all(quote.as_bytes()));
        }

        Ok(Tick::WouldBlock)
    }
}

If you’ve used asynchronous frameworks before, you might notice that we didn’t write any code to subscribe to any events or callbacks. That is because Tokio will handle the event subscriptions automatically for us. When you try to do any i/o in the tick function, Tokio will register to that event and the tick function will be called when any event occurs from that point on.

By returning Ok(Tick::WouldBlock) from the tick function, we’re telling Tokio that we’re waiting for more events and we have more work to do. We can return Ok(Tick::Final) at any point in order to make the reactor drop the task and not call the tick function again.

Plugging into the Tokio event reactor

Now that we have a Task, we can plug it into the event reactor in order have it run and process events. In order to do this, we’ll construct a Reactor and use the reactor::schedule function to connect it to our Task. Here’s the main function that will read the quotes and start the event reactor.

let quotes = read_quotes("quotes.txt");
let reactor = Reactor::default().unwrap();

reactor.handle().oneshot(|| {
    let addr = "0.0.0.0:17".parse().unwrap();
    let listener = TcpListener::bind(&addr).unwrap();

    reactor::schedule(Listener {socket: listener, quotes: quotes});

    Ok(())
});

reactor.run();

After the reactor.run(); call, our server will keep running until we kill it with CTRL-c.

Thanks for reading this article. You can find the whole Rust code here.