Concurrent Service Capstone

The first design question in any concurrent service decides its failure mode under load: when 10,000 requests arrive, do you spawn 10,000 tasks — or 8 workers pulling from a queue? Unbounded spawn turns traffic spikes into memory exhaustion and uncontrolled latency; a fixed pool of workers consuming one job channel bounds parallelism by construction — at most N jobs are in flight, and the queue absorbs the burst.

The service is a key-value store: clients submit Set(key, value) and Get(key, reply_tx) jobs, workers apply them. Tokio's mpsc channel is the job queue — Sender is cheap to clone (fan-in from many acceptors), and the channel itself is bounded, which adds a second backpressure layer: when the queue is full, the producer's send().await waits instead of allocating. The Sender half also survives across the service's lifetime cheaply, so accept-loop tasks, admin endpoints, and internal timers can all submit jobs without knowing how many workers exist — the pool size becomes a tuning knob you change in one constant.

use tokio::sync::mpsc;

const WORKERS: usize = 8;
const QUEUE: usize = 1024;

pub struct Pool { tx: mpsc::Sender<Job> }

impl Pool {
    pub fn start(store: Store) -> Self {
        let (tx, rx) = mpsc::channel::<Job>(QUEUE);
        let rx = std::sync::Arc::new(tokio::sync::Mutex::new(rx));
        for id in 0..WORKERS {
            let rx = std::sync::Arc::clone(&rx);
            let store = store.clone();
            tokio::spawn(async move {
                loop {
                    let job = { rx.lock().await.recv() }.await;   // release lock between recvs
                    let Some(job) = job else { break };           // all senders dropped
                    job.apply(&store);
                }
                tracing::debug!(id, "worker exited");
            });
        }
        Pool { tx }
    }
}

Read the mechanics, because each is an interview question in disguise. Tokio's Receiver is deliberately not Clone — one queue, many consumers must share it, so it goes inside Arc<tokio::sync::Mutex<>>. The lock guard's scope is one recv(): you lock, call recv().await, and drop the guard before handling the job ({ rx.lock().await.recv() }.await — the braces end the guard's lifetime at the inner expression). Holding the lock across job processing would serialize the whole pool through one mutex — the classic bug that makes a "concurrent" service single-threaded. The else { break } arm is the termination path: recv() returns None exactly when every Sender has been dropped, which later becomes the shutdown mechanism. And the pool size is a constant you tune against the workload: CPU-bound jobs want ≈ cores, I/O-bound jobs want more; either way the count is a decision with a number, not an accident of scheduling.

The std::thread variant deserves one paragraph because interviewers ask for it: spawn WORKERS OS threads sharing Arc<Mutex<Receiver<Job>>> around std::sync::mpsc::channel, block on rx.recv() (no .await), and send() never blocks because the std channel is unbounded (or use sync_channel(N) for bounds). Same architecture, different scheduler — tokio multiplexes workers onto a few OS threads, std pays a kernel thread (≈8 MB stack virtual, real context-switch cost) per worker. Being able to sketch both is the difference between knowing tokio and knowing concurrency.

Two load-testing habits separate the capstone from a toy: drive submit() from a client task that times each round-trip and reports p50/p99 latency, not just throughput — a pool that moves 50k jobs/s with a 2-second tail is a failing service — and watch the queue depth. When the bounded channel is full often, send().await is where your clients feel the overload; that visibility tells you whether the answer is more workers, a bigger queue, or admission control upstream. The channel-plus-shared-state split formalized here is the same one MPSC Channels and Arc Mutex built separately — this capstone welds them together.

The service holds one map shared by all workers, and the question is whether the map should be in the channel or behind a lock. The design that scales is the split: message passing carries the work requests; shared state sits behind Arc<Mutex<>> for the data both jobs must see. Workers read-modify-write the same store, so the map needs exclusion; but a Set arriving as a channel message, not as a lock acquisition from a caller, keeps producers decoupled from the store's guard.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Instant;

#[derive(Clone)]
pub struct Store(Arc<Mutex<HashMap<String, Vec<u8>>>>);

impl Store {
    fn lock(&self) -> MutexGuard<'_, HashMap<String, Vec<u8>>> {
        self.0.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
    }
    pub fn set(&self, k: String, v: Vec<u8>) {
        self.lock().insert(k, v);
    }
    pub fn get(&self, k: &str) -> Option<Vec<u8>> {
        self.lock().get(k).cloned()
    }
    pub fn apply(&self, job: Job) {
        let start = Instant::now();
        match job {
            Job::Set(k, v) => self.set(k, v),
            Job::Get(k, reply) => {
                let hit = self.get(&k);
                let _ = reply.send(hit);          // client may have timed out — ignore
            }
        }
        tracing::debug!(op = job.name(), us = start.elapsed().as_micros() as u64, "job done");
    }
}

Four decisions carry the weight. Store(Arc<Mutex<...>>) wraps rather than derives — the newtype keeps the lock private so no caller can hold the guard across code you don't control, and Clone is cheap (one atomic increment) because every worker gets a handle to the same map. Poison recovery is deliberate: lock().unwrap_or_else(|p| p.into_inner()) continues after a panicking worker rather than crashing every subsequent lock — defensible here because HashMap operations don't panic mid-mutation, but you must be able to say when the opposite choice (propagate the poison) is right: when the guarded data has invariants a panic could have broken. The critical section is one map operation — insert, or get-and-clone — never a network send, a file write, or an .await. Long holds are how a mutex becomes the bottleneck you then spend an afternoon blaming on "Rust's lock being slow." The Get reply goes back through a oneshot channel, so the worker never blocks on the client: if the requester timed out, let _ = reply.send(...) drops the result cleanly instead of panicking on a closed receiver.

Where does contention actually land? Every lock() is an atomic compare-exchange plus possible futex sleep; with 8 workers on short critical sections, the map rarely fights, and throughput scales near-linearly. The moment you report real numbers, the next lever appears: a single global Mutex<HashMap> serializes all writes, so a measured plateau under write-heavy load is the signal to shard — sixteen maps keyed by k.hash() % 16, each its own Arc<Mutex<>>, sixteen-way write parallelism. Sharding is the honest answer to "your lock is the bottleneck": not a lock-free crate from crates.io, but measured stripes.

The observability piece belongs here too, not bolted on later: the tracing::debug! span per job becomes tracing::info_span!("job", op, key) recorded with .enter() around the apply, and with the tracing-subscriber JSON formatter on stderr you can attribute latency to a worker and an operation without a debugger. Contention you cannot see is contention you will blame on the wrong layer — a per-op histogram turns "the lock feels slow" into "99th percentile lock wait is 4 ms at 8 workers." Both moves — the short-section discipline and the shard-when-measured rule — are exactly what the interviewer is probing when they ask "how do you protect shared state in a server?": exclusion where data is shared, ownership where work travels, and numbers before rewrites. The measurement half is the final chapter; first, the service has to stop cleanly.

A load balancer sends SIGTERM and your service must drain: stop accepting, let in-flight jobs finish, drop workers in an orderly way, exit 0. Killing threads is not shutdown — a worker mid-set might hold the lock when the process dies. The mechanism that composes with the architecture you already built: a shutdown signal every worker can receive, delivered by a tokio::sync::broadcast channel (the multi-consumer sibling of mpsc — one sender, every listener gets the message).

use tokio::sync::broadcast;

pub struct Pool {
    tx: mpsc::Sender<Job>,
    stop: broadcast::Sender<()>,
    handles: Vec<tokio::task::JoinHandle<()>>,
}

impl Pool {
    pub fn shutdown(self) {
        let _ = self.stop.send(());              // wake every worker
        drop(self.tx);                           // queue drains, recv() returns None
        for h in self.handles {
            let _ = tokio::time::timeout(std::time::Duration::from_secs(5), h).await;
        }
    }
}

// inside each worker, replacing the bare loop:
let mut stop_rx = stop.subscribe();
loop {
    tokio::select! {
        job = rx.lock().await.recv() => {
            let Some(job) = job else { break };
            job.apply(&store);
        }
        _ = stop_rx.recv() => break,             // exit AFTER the current job
    }
}

Walk the sequence, because its correctness is subtle in three places. stop.send(()) reaches every worker — broadcast fans one message out to all current and future subscribers, unlike mpsc where each message goes to exactly one receiver; the alternative sketched in interviews, an AtomicBool polled per iteration, works too but a blocked recv() never checks the flag, so you'd still need a wakeup — which is precisely what the broadcast (or a sync_channel sentinel job) provides. Then drop(self.tx) closes the queue: the current job finishes first — each worker completes its critical section before noticing closure — and the next recv() returns None, taking the else break. That ordering is what "graceful" means here: no job is abandoned mid-flight. The per-worker timeout on JoinHandle bounds the drain; a worker still stuck after 5 seconds is logged and, since the function returns anyway, the process proceeds to exit. select! returning from the job branch first also naturally keeps serving until the signal arrives — new work during drain is handled, and shutdown cannot interrupt a job, because the future doing the work is the one being awaited in the same branch. tracing::info!("shutdown complete", workers = handles.len()) makes the drain visible in the logs.

Last, the claim every optimization needs: numbers. Criterion drives the hot path in a controlled harness — warm-up, many samples, statistics, and a regression detector that fails a CI comparison when a change loses more than the noise threshold:

use criterion::{criterion_group, criterion_main, Criterion, Throughput};

fn bench_set_get(c: &mut Criterion) {
    let rt = tokio::runtime::Builder::new_current_thread().enable_time().build().unwrap();
    let pool = rt.block_on(async { Pool::start(Store::default()) });
    let mut group = c.benchmark_group("kv");
    group.throughput(Throughput::Elements(1));
    group.bench_function("set-then-get", |b| {
        b.iter(|| rt.block_on(async {
            let (reply, rx) = tokio::sync::oneshot::channel();
            pool.submit(Job::Set("k".into(), vec![0u8; 64])).await.unwrap();
            pool.submit(Job::Get("k".into(), reply)).await.unwrap();
            criterion::black_box(rx.await.unwrap());
        }));
    });
    group.finish();
}
criterion_group!(benches, bench_set_get);
criterion_main!(benches);

Run it before each change and after: baseline with a 4-worker pool, then 8, then the sharded store from the previous chapter, and write the three throughput numbers into the README with the hardware they were measured on. The discipline is the deliverable — "I sharded the map because criterion showed 1.3× → 3.9× ops/s at 8 workers" is a sentence that ends optimization debates. That before/after documentation, plus the production gate from CLI Tool Capstone, is what makes this capstone a service you could actually run: bounded queue, short locks, observable shutdown, and a benchmark trail of every decision.