Concurrent Servers: Part 7 - Rust - Eli Bendersky's website

Toggle navigation

Eli Bendersky's website

About

Projects

Archives

This is part 7 in a series of posts on writing concurrent network servers.<br>In this part, we discuss how the challenges described in earlier parts are<br>tackled in the Rust programming language.

All posts in the series:

Part 1 - Introduction

Part 2 - Threads

Part 3 - Event-driven

Part 4 - libuv

Part 5 - Redis case study

Part 6 - Callbacks, Promises and async/await

Part 7 - Rust (this part)

Several years have passed since the previous parts were published. I've recently<br>went over them to make sure the information presented is still relevant and<br>all the code samples build and run using modern toolchains. I strongly<br>recommend reviewing the previous parts before reading this one.

This post assumes a basic familiarity with the Rust programming language. It<br>will only explain Rust constructs when we encounter code that wouldn't appear in<br>an introductory book or tutorial.

Setting the baseline - a sequential state machine server

The first few parts in the series focused on a socket server that implements<br>a simple state machine protocol. See part 1<br>for a complete description of the protocol. Let's start by showing how this<br>protocol is implemented in a basic sequential Rust server:

use async_socket_server::serve_connection;<br>use std::net::TcpListener;

fn main() -> std::io::Result()> {<br>let port = match std::env::args().nth(1) {<br>Some(s) => s,<br>None => "9090".to_string(),<br>};<br>let addr = format!("127.0.0.1:{port}");

let listener = TcpListener::bind(addr)?;<br>println!("Serving on port {port}");

loop {<br>let (stream, addr) = listener.accept()?;<br>println!("connection received from {}", addr);<br>if let Err(e) = serve_connection(stream) {<br>eprintln!("error serving connection: {}", e);<br>} else {<br>println!("peer done {addr}");

With the function serve_connection defined as:

pub enum ProcessingState {<br>WaitForMsg,<br>InMsg,

pub fn serve_connection(mut stream: TcpStream) -> std::io::Result()> {<br>stream.write_all(b"*")?;

let mut state = ProcessingState::WaitForMsg;<br>let mut buf = [0u8; 1024];<br>loop {<br>let n = stream.read(&mut buf)?;<br>if n == 0 {<br>// Connection closed by the client.<br>break;

for byte in &buf[..n] {<br>match state {<br>ProcessingState::WaitForMsg => {<br>if *byte == b'^' {<br>state = ProcessingState::InMsg;<br>ProcessingState::InMsg => {<br>if *byte == b'$' {<br>state = ProcessingState::WaitForMsg;<br>} else {<br>let newbyte = byte.wrapping_add(1);<br>stream.write_all(&[newbyte])?;

Ok(())

As a reminder, this server version is sequential because it accepts clients<br>one by one; the main loop blocks on serve_connection until it's done (the<br>client closes the connection), and only then goes back to accept the next<br>client.

One thread per client

Clearly, handling clients one by one won't do. In<br>part 2,<br>we've discussed approaches that use OS threads to handle clients concurrently.<br>Let's start with the unbounded one-thread-per-client solution in Rust:

use async_socket_server::serve_connection;<br>use std::{net::TcpListener, thread};

fn main() -> std::io::Result()> {<br>let port = match std::env::args().nth(1) {<br>Some(s) => s,<br>None => "9090".to_string(),<br>};<br>let addr = format!("127.0.0.1:{port}");

let listener = TcpListener::bind(addr)?;<br>println!("Serving on port {port}");

loop {<br>let (stream, addr) = listener.accept()?;<br>println!("connection received from {}", addr);

let res = thread::Builder::new().spawn(move || {<br>if let Err(e) = serve_connection(stream) {<br>eprintln!("error serving connection: {}", e);<br>} else {<br>println!("peer done {addr}");<br>});

if let Err(e) = res {<br>eprintln!("error spawning thread: {}", e);

The spawn method returns a Result>; on success, we allow<br>the handle to be dropped at the end of the loop iteration. In Rust, this<br>detaches the thread; we don't actually wait for it to complete. This<br>is reasonable for our code sample, because the loop is infinite; it never<br>terminates anyway. The potential for runaway threads is just one of the issues<br>with the unbounded threads approach discussed in part 2. The solution is to use<br>a fixed thread pool.

Thread pool

Before diving into the code, a quick note on the design: the thread pool is<br>a fixed set of threads that await "jobs" and handle them to completion. In our<br>case a "job" is serve_connection for a specific client. There are many ways<br>to implement a thread pool; for our use case, I went with a set of threads that<br>all get a shared channel to which the main thread sends jobs. A worker thread<br>picks up the next job from the channel, serves it to completion, and goes back<br>to waiting for the next job. Here's how this looks in code:

struct Job {<br>stream: TcpStream,<br>addr: SocketAddr,

fn worker(receiver: ReceiverJob>) {<br>while let Ok(job) = receiver.recv() {<br>if let Err(e) = serve_connection(job.stream) {<br>eprintln!("error serving connection from {}: {}", job.addr, e);<br>} else {<br>println!("peer done {}", job.addr);

What is Receiver? It's a type from the crossbeam_channel crate:

use...