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

From the sequential baseline to threads

The running example throughout the series is a socket server implementing a simple state machine protocol, described in part 1. The sequential Rust version accepts clients one at a time: the main loop blocks in serve_connection until the client disconnects, then returns to accept the next connection.

use async_socket_server::serve_connection;
use std::net::TcpListener;

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

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

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

With the function serve_connection defined as:

pub enum ProcessingState {
    WaitForMsg,
    InMsg,
}

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

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

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

    Ok(())
}

Making the server concurrent follows the same path as the earlier parts on OS threads. The unbounded one-thread-per-client approach looks like this:

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

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

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

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

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

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

spawn returns a Result<JoinHandle<T>>. Letting the handle drop at the end of each loop iteration detaches the thread rather than joining it — acceptable here only because the loop is infinite and therefore never terminates. Runaway thread growth, one of the pitfalls discussed in part 2, is what motivates a fixed thread pool.

A fixed pool with backpressure

The pool is a fixed set of worker threads that pull jobs from a common channel and run each job to completion; a job is serve_connection for one client.

struct Job {
    stream: TcpStream,
    addr: SocketAddr,
}

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

use crossbeam_channel::{Receiver, bounded};

The Receiver is from the crossbeam_channel crate, used because the built-in std channels are mpsc (multi producer, single consumer), while a job queue shared by worker threads needs multiple consumers. std's mpmc API was experimental and nightly-only at the time of writing, so this sample relies on the well-tested mpmc channels provided by crossbeam_channel.

The main function wires the pool up:

use async_socket_server::serve_connection;
use crossbeam_channel::{Receiver, bounded};
use std::{
    io,
    net::{SocketAddr, TcpListener, TcpStream},
    thread,
};

const NUM_WORKERS: usize = 256;
const JOB_QUEUE_CAPACITY: usize = NUM_WORKERS;

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

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

    let (tx, rx) = bounded::<Job>(JOB_QUEUE_CAPACITY);
    for _ in 0..NUM_WORKERS {
        let receiver = rx.clone();
        thread::Builder::new().spawn(move || worker(receiver))?;
    }
    drop(rx);

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

        // A full queue blocks this loop, preventing it from accepting more
        // connections until a worker becomes available.
        if tx.send(Job { stream, addr }).is_err() {
            return Err(io::Error::other("all connection workers stopped"));
        }
    }
}

Keeping the job channel bounded gives backpressure essentially for free: when clients pile up, the main loop blocks in tx.send and stops accepting new connections on the socket until the queue drains.

Event-driven servers with tokio

The event-driven parts of the series traced a progression from callbacks through promises to async/await; Rust supports all three, and modern code hides the promise layer (Rust calls these futures) under async/await. The language is unusual in that its core includes futures and the async/await keywords, while the async engine — the event loop itself — lives in external crates. tokio is by far the most popular one, and it's what's used here:

use async_socket_server::ProcessingState;
use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

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

    let listener = TcpListener::bind(addr).await?;
    println!("Serving on port {port}");

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("connection received from {addr:?}");

        tokio::spawn(async move {
            if let Err(e) = serve_connection_async(socket).await {
                eprintln!("error serving connection from {addr:?}: {e}");
            } else {
                println!("peer done {addr:?}");
            }
        });
    }
}

For anyone who read the JS code in part 6, this should look familiar apart from the explicit tokio task spawn. Rather than enqueuing a callback on the connection returned by listener.accept, the code spawns a tokio task, which is best understood as a green thread — hence the similar terminology. Such tasks must not make blocking calls, so they use tokio's I/O utilities instead of the blocking std ones, which means the connection handler also needs an async form:

async fn serve_connection_async(mut stream: tokio::net::TcpStream) -> std::io::Result<()> {
    stream.write_all(b"*").await?;

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

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

    Ok(())
}

This is nearly identical to the earlier serve_connection; the real differences are the await calls on reads and writes and the involved types — tokio::net::TcpStream where the synchronous samples had std::net::TcpStream. Underneath, tokio depends on mio for non-blocking I/O APIs across platforms, wrapping OS-specific event loops such as epoll efficiently.

Primality servers, async and with a cache

The series switched its driving example to a primality-testing server that simulates long computations. In tokio it looks like this:

use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

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

    let listener = TcpListener::bind(addr).await?;
    println!("Serving on port {port}");

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("connection received from {addr:?}");

        tokio::spawn(async move {
            if let Err(e) = serve_client(socket).await {
                eprintln!("error serving connection from {addr:?}: {e}");
            } else {
                println!("peer done {addr:?}");
            }
        });
    }
}

async fn serve_client(mut stream: tokio::net::TcpStream) -> std::io::Result<()> {
    let mut buf = [0u8; 1024];

    loop {
        let n = stream.read(&mut buf).await?;
        if n == 0 {
            // Connection closed by the client.
            return Ok(());
        }

        // Parse read buf into u64.
        let num = std::str::from_utf8(&buf[..n])
            .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?
            .trim()
            .parse::<u64>()
            .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;

        let answer = if isprime(num, true) {
            "prime"
        } else {
            "composite"
        };
        stream
            .write_all((answer.to_string() + "\n").as_bytes())
            .await?;
    }
}

Where isprime is:

// Check if n is prime, returning a boolean. The delay parameter is optional;
// if true, the function will block for n milliseconds before computing the
// answer. This is useful for simulating a long-running computation.
fn isprime(n: u64, delay: bool) -> bool {
    if delay {
        std::thread::sleep(std::time::Duration::from_millis(n));
    }

    if n < 2 {
        return false;
    }
    if n % 2 == 0 {
        return n == 2;
    }

    let mut r = 3;
    while r * r <= n {
        if n % r == 0 {
            return false;
        }
        r += 2;
    }
    true
}

This sample makes a blocking job obvious (simulated with a sleep), which is problematic in an async context, as the tokio documentation explains. One remedy is dispatching the blocking work to a separate thread pool and communicating with it over tokio channels, echoing the thread pool design above.

Part 6 also cached results in a local Redis instance, to show how much complexity extra callback layers add in event-driven code and how async/await reduces it. The Rust equivalent uses the redis crate with its tokio component enabled explicitly:

use redis::AsyncTypedCommands;
use redis::aio::MultiplexedConnection;
use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

#[derive(Clone)]
struct AppState {
    redis_connection: MultiplexedConnection,
}

const REDIS_URL: &str = "redis://127.0.0.1";

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

    let app_state = AppState {
        redis_connection: {
            redis::Client::open(REDIS_URL)
                .map_err(io::Error::other)?
                .get_multiplexed_async_connection()
                .await
                .map_err(io::Error::other)?
        },
    };

    let listener = TcpListener::bind(addr).await?;
    println!("Serving on port {port}");

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("connection received from {addr:?}");
        let app_state = app_state.clone();

        tokio::spawn(async move {
            if let Err(e) = serve_client(socket, app_state).await {
                eprintln!("error serving connection from {addr:?}: {e}");
            } else {
                println!("peer done {addr:?}");
            }
        });
    }
}

async fn serve_client(
    mut stream: tokio::net::TcpStream,
    mut app_state: AppState,
) -> std::io::Result<()> {
    let mut buf = [0u8; 1024];

    loop {
        let n = stream.read(&mut buf).await?;
        if n == 0 {
            // Connection closed by the client.
            return Ok(());
        }

        // Parse read buf into u64.
        let num = std::str::from_utf8(&buf[..n])
            .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?
            .trim()
            .parse::<u64>()
            .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;

        // Try the redis cache first. If found in cache, send the cached
        // answer.
        let cachekey = format!("primecache:{num}");
        match app_state.redis_connection.get(cachekey.clone()).await {
            Ok(Some(cached)) => {
                stream.write_all((cached.clone() + "\n").as_bytes()).await?;
                println!("cached num {num} is {cached}");
                continue;
            }
            Ok(None) => {
                // Not found in cache, continue to compute.
            }
            Err(error) => {
                return Err(io::Error::other(error));
            }
        }

        let answer = if isprime(num, true) {
            "prime"
        } else {
            "composite"
        };

        // Save answer in cache and send it to the client.
        app_state
            .redis_connection
            .set(cachekey, answer)
            .await
            .map_err(io::Error::other)?;
        stream
            .write_all((answer.to_string() + "\n").as_bytes())
            .await?;
    }
}

• The function color problem is why the redis crate exposes a dedicated async connection constructor, get_multiplexed_async_connection.
• The Redis connection is shared state between tokio tasks, yet needs no explicit synchronization: MultiplexedConnection is Clone, and cloning it into each new task is safe. Internally it already carries whatever synchronization it needs.
• Thanks to async/await, the body of serve_client stays linear: it awaits the Redis call and continues once it returns. Because the connection is async, any waiting hands control to another task that isn't blocked on I/O.

Rust's async support is strong, but it doesn't dissolve the inherent difficulties — function colors, and the need to keep blocking and non-blocking work carefully separated. Those are usually manageable with extra care, and tokio-based async in Rust remains very popular for its performance.

Code

All the code for this post is available on GitHub.