Chapter 12
Concurrency - Handling Multiple Connections
Our server handles one connection at a time. While it's processing a request, every other client waits. Time to fix that with threads, shared state, and a thread pool.
Threads with std::thread::spawn
The simplest concurrency model: spawn a new OS thread for each connection. Rust's standard library makes this straightforward:
use std::thread;
fn main() {
let handle = thread::spawn(|| {
println!("Hello from a new thread!");
42 // thread can return a value
});
println!("Hello from the main thread!");
// Wait for the thread to finish and get its return value
let result = handle.join().unwrap();
println!("Thread returned: {}", result);
}thread::spawn takes a closure and runs it on a new OS thread. It returns a JoinHandle - calling .join() blocks the current thread until the spawned thread finishes.
Let's apply this to our server. The change is minimal - move handle_connection into a spawned thread:
use std::net::TcpListener;
use std::thread;
fn main() {
let router = handler::build_router();
let addr = "127.0.0.1:8080";
let listener = TcpListener::bind(addr).expect("failed to bind");
println!("Listening on http://{}", addr);
for stream in listener.incoming() {
match stream {
Ok(stream) => {
// Problem: how do we share 'router' across threads?
thread::spawn(move || {
handle_connection(stream, &router); // ✗ won't compile
});
}
Err(e) => eprintln!("Accept failed: {}", e),
}
}
}This won't compile. The closure needs move to take ownership of stream (threads need to own their data), but it also tries to borrow router. We can't moverouter because we need it for every connection, and we can't borrow it because the borrow might outlive the main thread's loop iteration.
This is exactly the kind of bug that Rust catches at compile time. In C or Go, you'd have a data race waiting to happen. In Rust, you need Arc.
Shared State with Arc<Mutex<T>>
Two types from the standard library solve the shared-state problem:
Arc<T>(Atomic Reference Counted) - shared ownership across threads. Multiple threads can hold a clone of the sameArc, and the data is freed when the last clone is dropped.Mutex<T>(Mutual Exclusion) - ensures only one thread accesses the data at a time. You.lock()to get a guard that dereferences to the inner value.
Arc - Shared Ownership Across Threads
Our Router is read-only after construction - multiple threads just need to read from it. Arc alone is enough for this:
use std::sync::Arc;
use std::net::TcpListener;
use std::thread;
fn main() {
let router = Arc::new(handler::build_router());
let addr = "127.0.0.1:8080";
let listener = TcpListener::bind(addr).expect("failed to bind");
println!("Listening on http://{}", addr);
for stream in listener.incoming() {
match stream {
Ok(stream) => {
let router = Arc::clone(&router); // cheap clone - just increments refcount
thread::spawn(move || {
handle_connection(stream, &router);
});
}
Err(e) => eprintln!("Accept failed: {}", e),
}
}
}Arc::clone(&router) doesn't copy the router - it creates a new pointer to the same data and increments an atomic reference count. Each thread owns its own Arc, and the router is freed when the last Arc is dropped.
Rc<T> is the single-threaded version of Arc - it's faster because it uses non-atomic reference counting. The compiler won't let you send an Rc to another thread (it doesn't implement Send). When you need shared ownership across threads, use Arc.Mutex - Mutable Shared State
What if threads need to write to shared data? For example, tracking request counts across all connections. That's where Mutex comes in:
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
// Shared mutable state: request counts per path
let stats: Arc<Mutex<HashMap<String, u64>>> =
Arc::new(Mutex::new(HashMap::new()));
for stream in listener.incoming() {
if let Ok(stream) = stream {
let router = Arc::clone(&router);
let stats = Arc::clone(&stats);
thread::spawn(move || {
let request = read_request(&stream);
// Lock the mutex to access the data
{
let mut counts = stats.lock().unwrap();
*counts.entry(request.path.clone()).or_insert(0) += 1;
} // MutexGuard is dropped here - lock is released
let response = router.route(&request);
send_response(&stream, &response);
});
}
}Key points about Mutex:
.lock()blocks until no other thread holds the lock. It returns aMutexGuard- a smart pointer that dereferences to the inner value.- The lock is released when the
MutexGuardis dropped. Use a block{ }to limit how long you hold the lock. .lock().unwrap()panics if another thread panicked while holding the lock (a "poisoned" mutex). In production, handle this with.lock().unwrap_or_else(|e| e.into_inner()).
Message Passing with Channels
An alternative to shared state is message passing - threads communicate by sending values through a channel instead of sharing memory. Rust's mpsc module (multiple producer, single consumer) provides this:
use std::sync::mpsc;
use std::thread;
// Create a channel
let (sender, receiver) = mpsc::channel();
// Spawn workers that send results back
for i in 0..4 {
let tx = sender.clone();
thread::spawn(move || {
let result = format!("Worker {} done", i);
tx.send(result).unwrap();
});
}
drop(sender); // Drop the original sender so the receiver knows when all senders are gone
// Receive all messages
for message in receiver {
println!("{}", message);
}mpsc::channel() returns a (Sender, Receiver) pair. Senders can be cloned and given to multiple threads. send() is non-blocking. The receiver iterates over incoming messages, blocking until a message arrives or all senders are dropped.
For our webserver, channels are useful for decoupling request acceptance from processing - and they're the foundation of our thread pool:
use std::sync::mpsc;
use std::net::TcpStream;
// Log stats from worker threads without sharing a mutex
enum LogMessage {
Request { method: String, path: String, status: u16 },
Error(String),
}
let (log_tx, log_rx) = mpsc::channel::<LogMessage>();
// Logger thread - single consumer
thread::spawn(move || {
for msg in log_rx {
match msg {
LogMessage::Request { method, path, status } => {
println!("[{}] {} {} → {}", chrono::Local::now().format("%H:%M:%S"),
method, path, status);
}
LogMessage::Error(e) => eprintln!("[ERROR] {}", e),
}
}
});
// Workers send log messages - no shared mutex needed
// log_tx.send(LogMessage::Request { ... }).unwrap();Arc<Mutex>) for small, fast updates like counters. Use channels when work flows in one direction (producer → consumer) or when you want to decouple components. Both patterns are valid - pick whichever models your problem better.Building a Thread Pool
Spawning a thread per connection is simple but dangerous - if 10,000 clients connect at once, you spawn 10,000 OS threads and probably crash. A thread pool pre-creates a fixed number of threads and distributes work among them.
The design: a pool of worker threads sharing a channel. The main thread sends jobs (closures) through the channel. Workers pick up jobs and execute them:
┌─── Worker 0 ◀── execute job
│
main ──▶ channel ├─── Worker 1 ◀── execute job
(send │
jobs) ├─── Worker 2 ◀── (idle, waiting)
│
└─── Worker 3 ◀── execute jobuse std::sync::{mpsc, Arc, Mutex};
use std::thread;
type Job = Box<dyn FnOnce() + Send + 'static>;
pub struct ThreadPool {
workers: Vec<Worker>,
sender: Option<mpsc::Sender<Job>>,
}
struct Worker {
id: usize,
handle: Option<thread::JoinHandle<()>>,
}
impl ThreadPool {
/// Create a pool with the given number of threads.
pub fn new(size: usize) -> Self {
assert!(size > 0, "ThreadPool size must be > 0");
let (sender, receiver) = mpsc::channel::<Job>();
let receiver = Arc::new(Mutex::new(receiver));
let workers: Vec<Worker> = (0..size)
.map(|id| Worker::new(id, Arc::clone(&receiver)))
.collect();
ThreadPool {
workers,
sender: Some(sender),
}
}
/// Submit a job to the pool.
pub fn execute<F>(&self, job: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(job);
self.sender.as_ref().unwrap().send(job).unwrap();
}
}
impl Drop for ThreadPool {
fn drop(&mut self) {
// Drop the sender to signal workers to stop
drop(self.sender.take());
// Wait for all workers to finish
for worker in &mut self.workers {
if let Some(handle) = worker.handle.take() {
handle.join().unwrap();
}
}
}
}
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Self {
let handle = thread::spawn(move || loop {
// Lock the receiver, get the next job
let message = receiver.lock().unwrap().recv();
match message {
Ok(job) => {
job(); // Execute the job
}
Err(_) => {
// Channel closed - sender was dropped, time to exit
break;
}
}
});
Worker {
id,
handle: Some(handle),
}
}
}Let's trace the key design decisions:
type Job = Box<dyn FnOnce() + Send + 'static>- a job is a boxed closure that can be sent to another thread (Send), runs once (FnOnce), and owns its data ('static).Arc<Mutex<Receiver>>- the receiver is shared among all workers. TheMutexensures only one worker picks up each job.Arcgives each worker thread shared ownership.receiver.lock().unwrap().recv()- lock the mutex, then callrecv()which blocks until a job arrives. The lock is released immediately afterrecv()returns because theMutexGuardis a temporary.Dropimplementation - when the pool is dropped, it drops the sender (closing the channel), which causes all workers'recv()calls to returnErr. Then it joins each thread. Graceful shutdown.
Now wire it into the server:
mod pool;
use std::sync::Arc;
use std::net::TcpListener;
use pool::ThreadPool;
fn main() {
let router = Arc::new(handler::build_router());
let pool = ThreadPool::new(4);
let addr = "127.0.0.1:8080";
let listener = TcpListener::bind(addr).expect("failed to bind");
println!("Listening on http://{} (4 threads)", addr);
for stream in listener.incoming() {
match stream {
Ok(stream) => {
let router = Arc::clone(&router);
pool.execute(move || {
handle_connection(stream, &router);
});
}
Err(e) => eprintln!("Accept failed: {}", e),
}
}
} // pool is dropped here - waits for all workers to finishThat's it. Four threads handle all connections. If all four are busy, new connections queue in the channel until a worker is free. No thread explosion, predictable resource usage.
rayon add work-stealing, dynamic sizing, and panic handling - but the core idea is identical: workers pull jobs from a shared channel.Understanding Send and Sync
You've seen Send and 'static in the thread pool code. These are two marker traits that the compiler uses to enforce thread safety:
Send - can be transferred to another thread
A type is Send if it's safe to move it to another thread. Most types are Send. Notable exceptions: Rc<T> (non-atomic reference counting) and raw pointers.
Sync - can be referenced from multiple threads
A type is Sync if &T is Send - meaning multiple threads can hold immutable references to it simultaneously. Mutex<T> is Sync even when T isn't, because the mutex controls access.
use std::rc::Rc;
use std::sync::Arc;
// Rc is NOT Send - this won't compile
// thread::spawn(move || {
// let x = Rc::new(5); // Rc can't cross thread boundaries
// });
// Arc IS Send - this is fine
thread::spawn(move || {
let x = Arc::new(5); // Arc uses atomic reference counting
});
// &Mutex<T> is Send (via Sync), so threads can share a mutex
// &Vec<T> is Sync when T is Sync, so threads can read from shared vectorsYou almost never implement Send or Sync manually. The compiler derives them automatically for any type whose fields are all Send/Sync. They're checked at compile time - if you try to send a non-Send type to another thread, you get a compile error, not a runtime bug.
Exercise
- Add
Arc<Mutex<HashMap>>request counting to the thread-pool server. Add aGET /statsroute that locks the mutex and returns the counts as HTML. Test with multiple concurrent requests:for i in $(seq 1 20); do curl -s http://localhost:8080/ & done; wait - Add a configurable pool size: read the thread count from an environment variable with
std::env::var("THREADS").ok().and_then(|s| s.parse().ok()).unwrap_or(4). What happens with 1 thread? With 100? - Implement a
shutdownmechanism: add aGET /shutdownroute that signals the main loop to stop accepting connections. Hint: use anArc<AtomicBool>flag. - Try using
Rcinstead ofArcfor the router. Read the compiler error carefully - it tells you exactly whyRccan't cross thread boundaries. - Add a logging channel: spawn a dedicated logger thread that receives
LogMessagevalues viampsc::channel. Worker threads send log messages instead of printing directly. Why is this better than each thread printing to stdout?
What's Next
Our server now handles multiple connections concurrently with a thread pool. The borrow checker and Send/Sync traits ensure we can't introduce data races - the compiler enforces thread safety at compile time.
But threads have limits. Each OS thread uses 1–8 MB of stack memory. A pool of 4 threads means only 4 connections can be processed at the same time. For thousands of concurrent connections, we need a different model. In the next chapter, we introduce async Rust and Tokio - lightweight tasks that share threads, handling thousands of connections without thousands of OS threads.