Drills against the clock

Every chapter so far has shown the long form: the reference implementation with its explanations, its tests, and its history. In a live round you will not have that. You will have a blank editor, a problem stated in two sentences, and about forty minutes. drills.rs is the short form: ten answers written at the length you can type under that pressure. Each group carries one test, so you know it runs.

This chapter is a guide to using them. It maps each drill to its chapter, reads the places where the short form cuts a corner, and ends with a way to rehearse.

30.1 The drills

Listing 30.1 Ten answers sized for a live round. src/problems/drills.rs

//! Compact implementations of the common primitives.
//!
//! Each item is a short standalone implementation. The modules next to this one
//! are the full versions with long explanations and complete test suites.

use std::{
    cell::UnsafeCell,
    collections::{VecDeque, hash_map::DefaultHasher},
    hash::Hasher,
    ops::{Deref, DerefMut},
    sync::{
        Arc,
        Condvar,
        Mutex,
        atomic::{AtomicBool, Ordering},
        mpsc,
    },
    thread::{self, JoinHandle},
    time::Duration,
};

// --- Thread-safe counter (Arc + Mutex) ---

pub fn count_parallel(threads: usize, per_thread: usize) -> usize {
    let counter = Arc::new(Mutex::new(0usize));
    let handles: Vec<_> = (0..threads)
        .map(|_| {
            let counter = Arc::clone(&counter);
            thread::spawn(move || {
                for _ in 0..per_thread {
                    *counter.lock().expect("poisoned") += 1;
                }
            })
        })
        .collect();

    for handle in handles {
        handle.join().expect("worker panicked");
    }
    *counter.lock().expect("poisoned")
}

// --- Scoped threads (borrow instead of Arc) ---

pub fn parallel_sum(values: &[i32]) -> i64 {
    let chunk_size = values.len().div_ceil(4).max(1);
    thread::scope(|scope| {
        values
            .chunks(chunk_size)
            .map(|chunk| {
                scope.spawn(move || {
                    chunk
                        .iter()
                        .map(|value| i64::from(*value))
                        .sum::<i64>()
                })
            })
            .map(|handle| handle.join().expect("worker panicked"))
            .sum()
    })
}

// --- Spin lock from an atomic ---

pub struct SpinLock<T> {
    locked: AtomicBool,
    value: UnsafeCell<T>,
}

unsafe impl<T: Send> Sync for SpinLock<T> {}

impl<T> SpinLock<T> {
    pub const fn new(value: T) -> Self {
        Self {
            locked: AtomicBool::new(false),
            value: UnsafeCell::new(value),
        }
    }

    pub fn lock(&self) -> SpinGuard<'_, T> {
        while self
            .locked
            .swap(true, Ordering::Acquire)
        {
            std::hint::spin_loop();
        }
        SpinGuard { lock: self }
    }
}

pub struct SpinGuard<'a, T> {
    lock: &'a SpinLock<T>,
}

impl<T> Drop for SpinGuard<'_, T> {
    fn drop(&mut self) {
        self.lock
            .locked
            .store(false, Ordering::Release);
    }
}

impl<T> Deref for SpinGuard<'_, T> {
    type Target = T;
    fn deref(&self) -> &T {
        unsafe { &*self.lock.value.get() }
    }
}

impl<T> DerefMut for SpinGuard<'_, T> {
    fn deref_mut(&mut self) -> &mut T {
        unsafe { &mut *self.lock.value.get() }
    }
}

// --- Semaphore (acquire blocks) ---

pub struct Semaphore {
    permits: Mutex<usize>,
    released: Condvar,
}

impl Semaphore {
    pub fn new(permits: usize) -> Self {
        Self {
            permits: Mutex::new(permits),
            released: Condvar::new(),
        }
    }

    pub fn acquire(&self) {
        let mut permits = self.permits.lock().expect("poisoned");
        while *permits == 0 {
            permits = self
                .released
                .wait(permits)
                .expect("poisoned");
        }
        *permits -= 1;
    }

    pub fn try_acquire(&self) -> bool {
        let mut permits = self.permits.lock().expect("poisoned");
        if *permits == 0 {
            false
        } else {
            *permits -= 1;
            true
        }
    }

    pub fn release(&self) {
        *self.permits.lock().expect("poisoned") += 1;
        self.released.notify_one();
    }
}

// --- Bounded blocking queue (backpressure + close) ---

pub struct BlockingQueue<T> {
    inner: Mutex<Inner<T>>,
    not_empty: Condvar,
    not_full: Condvar,
}

struct Inner<T> {
    items: VecDeque<T>,
    capacity: usize,
    closed: bool,
}

impl<T> BlockingQueue<T> {
    pub fn new(capacity: usize) -> Self {
        Self {
            inner: Mutex::new(Inner {
                items: VecDeque::new(),
                capacity,
                closed: false,
            }),
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    pub fn push(&self, item: T) -> Result<(), T> {
        let mut inner = self.inner.lock().expect("poisoned");
        loop {
            if inner.closed {
                return Err(item);
            }
            if inner.items.len() < inner.capacity {
                inner.items.push_back(item);
                self.not_empty.notify_one();
                return Ok(());
            }
            inner = self
                .not_full
                .wait(inner)
                .expect("poisoned");
        }
    }

    pub fn pop(&self) -> Option<T> {
        let mut inner = self.inner.lock().expect("poisoned");
        loop {
            if let Some(item) = inner.items.pop_front() {
                self.not_full.notify_one();
                return Some(item);
            }
            if inner.closed {
                return None;
            }
            inner = self
                .not_empty
                .wait(inner)
                .expect("poisoned");
        }
    }

    pub fn close(&self) {
        let mut inner = self.inner.lock().expect("poisoned");
        inner.closed = true;
        drop(inner);
        self.not_empty.notify_all();
        self.not_full.notify_all();
    }
}

// --- Thread pool (fixed workers + channel) ---

pub struct ThreadPool {
    sender: Option<mpsc::Sender<Box<dyn FnOnce() + Send + 'static>>>,
    workers: Vec<JoinHandle<()>>,
}

impl ThreadPool {
    pub fn new(size: usize) -> Self {
        let (sender, receiver) = mpsc::channel::<Box<dyn FnOnce() + Send + 'static>>();
        let receiver = Arc::new(Mutex::new(receiver));
        let workers = (0..size)
            .map(|_| {
                let receiver = Arc::clone(&receiver);
                thread::spawn(move || {
                    loop {
                        let Ok(job) = receiver
                            .lock()
                            .expect("poisoned")
                            .recv()
                        else {
                            return;
                        };
                        job();
                    }
                })
            })
            .collect();
        Self {
            sender: Some(sender),
            workers,
        }
    }

    pub fn execute(&self, job: impl FnOnce() + Send + 'static) {
        self.sender
            .as_ref()
            .expect("pool dropped")
            .send(Box::new(job))
            .expect("worker gone");
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());
        for worker in self.workers.drain(..) {
            let _ = worker.join();
        }
    }
}

// --- Retry with exponential backoff ---

pub fn retry<T, E>(
    max_attempts: u32,
    mut operation: impl FnMut() -> Result<T, E>,
    retryable: impl Fn(&E) -> bool,
) -> Result<T, E> {
    let mut delay = Duration::from_millis(25);
    let mut attempt = 0;
    loop {
        match operation() {
            Ok(value) => return Ok(value),
            Err(error) => {
                attempt += 1;
                if attempt >= max_attempts || !retryable(&error) {
                    return Err(error);
                }
                thread::sleep(delay);
                delay = (delay * 2).min(Duration::from_secs(1));
            }
        }
    }
}

// --- Consistent hashing ring ---

#[derive(Default)]
pub struct HashRing {
    points: Vec<(u64, String)>,
}

impl HashRing {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn add(&mut self, node: &str, replicas: usize) {
        for replica in 0..replicas {
            self.points
                .push((hash(node, replica), node.to_string()));
        }
        self.points
            .sort_unstable_by_key(|(point, _)| *point);
    }

    pub fn get(&self, key: &str) -> Option<&str> {
        if self.points.is_empty() {
            return None;
        }
        let hash = hash(key, 0);
        let index = self
            .points
            .partition_point(|(point, _)| *point <= hash)
            % self.points.len();
        Some(self.points[index].1.as_str())
    }
}

fn hash(input: &str, replica: usize) -> u64 {
    let mut hasher = DefaultHasher::new();
    hasher.write(input.as_bytes());
    hasher.write_usize(replica);
    hasher.finish()
}

// --- HTTP request head ---

pub struct RequestHead {
    pub method: String,
    pub target: String,
    pub headers: Vec<(String, String)>,
}

pub fn parse_request_head(input: &str) -> Option<RequestHead> {
    let mut lines = input.split("\r\n");
    let mut request_line = lines.next()?.split(' ');
    let method = request_line.next()?.to_string();
    let target = request_line.next()?.to_string();
    if request_line.next()? != "HTTP/1.1" {
        return None;
    }

    let mut headers = Vec::new();
    for line in lines {
        if line.is_empty() {
            break;
        }
        let (name, value) = line.split_once(':')?;
        headers.push((name.trim().to_ascii_lowercase(), value.trim().to_string()));
    }
    Some(RequestHead {
        method,
        target,
        headers,
    })
}

// --- Newtype and enum ADT ---

pub struct Port(u16);

impl Port {
    pub fn new(value: u16) -> Result<Self, &'static str> {
        (1..=u16::MAX)
            .contains(&value)
            .then_some(Self(value))
            .ok_or("port must be non-zero")
    }

    pub fn get(self) -> u16 {
        self.0
    }
}

pub enum Route {
    Health,
    List,
    ById(String),
}

impl Route {
    pub fn parse(path: &str) -> Option<Self> {
        match path {
            "/healthz" => Some(Self::Health),
            "/v1/gpu-workloads" => Some(Self::List),
            _ => path
                .strip_prefix("/v1/gpu-workloads/")
                .map(|id| Self::ById(id.to_string())),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::AtomicUsize;

    use super::*;

    #[test]
    fn counter_scoped_sum_and_spin_lock() {
        assert_eq!(count_parallel(4, 100), 400);
        assert_eq!(parallel_sum(&[1, 2, 3, 4]), 10);

        let lock = Arc::new(SpinLock::new(0usize));
        let handles: Vec<_> = (0..4)
            .map(|_| {
                let lock = Arc::clone(&lock);
                thread::spawn(move || {
                    for _ in 0..100 {
                        *lock.lock() += 1;
                    }
                })
            })
            .collect();
        for handle in handles {
            handle.join().expect("worker panicked");
        }
        assert_eq!(*lock.lock(), 400);
    }

    #[test]
    fn semaphore_queue_and_pool() {
        let semaphore = Semaphore::new(1);
        assert!(semaphore.try_acquire());
        assert!(!semaphore.try_acquire());
        semaphore.release();
        assert!(semaphore.try_acquire());

        let queue = BlockingQueue::new(1);
        assert_eq!(queue.push(1), Ok(()));
        queue.close();
        assert_eq!(queue.pop(), Some(1));
        assert_eq!(queue.pop(), None);
        assert_eq!(queue.push(2), Err(2));

        let pool = ThreadPool::new(3);
        let counter = Arc::new(AtomicUsize::new(0));
        for _ in 0..100 {
            let counter = Arc::clone(&counter);
            pool.execute(move || {
                counter.fetch_add(1, Ordering::SeqCst);
            });
        }
        drop(pool);
        assert_eq!(counter.load(Ordering::SeqCst), 100);
    }

    #[test]
    fn retry_ring_http_and_adt() {
        let mut attempts = 0;
        let ok = retry(
            3,
            || {
                attempts += 1;
                if attempts < 3 {
                    Err("transient")
                } else {
                    Ok(7)
                }
            },
            |_| true,
        );
        assert_eq!(ok, Ok(7));

        let mut attempts = 0;
        let bad: Result<(), &str> = retry(
            3,
            || {
                attempts += 1;
                Err("permanent")
            },
            |_| false,
        );
        assert!(bad.is_err());
        assert_eq!(attempts, 1);

        let mut ring = HashRing::new();
        ring.add("a", 64);
        ring.add("b", 64);
        let before = ring.get("job-1").map(str::to_string);
        ring.add("c", 64);
        let after = ring.get("job-1").map(str::to_string);
        assert!(after == before || after.as_deref() == Some("c"));

        let head = parse_request_head("GET /v1/gpu-workloads HTTP/1.1\r\nHost: api\r\n\r\n")
            .expect("valid");
        assert_eq!(
            (head.method.as_str(), head.target.as_str()),
            ("GET", "/v1/gpu-workloads")
        );
        assert_eq!(head.headers[0], ("host".to_string(), "api".to_string()));
        assert!(parse_request_head("GET / HTTP/2\r\n\r\n").is_none());

        assert!(Port::new(8080).is_ok());
        assert!(Port::new(0).is_err());
        assert!(matches!(Route::parse("/healthz"), Some(Route::Health)));
        assert!(matches!(
            Route::parse("/v1/gpu-workloads/wl-1"),
            Some(Route::ById(id)) if id == "wl-1"
        ));
    }
}

The module comment states the file’s role in one line: the neighbouring modules are the reference versions, and this file is what you type. Table 30.1 pairs each drill with the chapter that explains its full form.

30.2 What the short forms leave out

A drill is allowed to be simpler than its reference. Knowing what it left out is part of the answer, because the follow-up question is usually aimed at exactly that.

The spin lock uses swap. while self.locked.swap(true, Acquire) { spin_loop() } is a test-and-set lock: every waiting thread writes the lock’s cache line on every iteration. The reference version in chapter 16 uses compare_exchange_weak and spins on a load, so waiters share the line read-only until it is released. Both are correct; the reference scales better under contention. The drill also omits the SAFETY comments, which in a real review you would add back.

The semaphore has no guard. acquire and release are separate calls, so a caller that panics between them leaks a permit. The reference in section 16.5 returns a SemaphoreGuard whose Drop gives the permit back.

The thread pool receives in a let-else. let Ok(job) = receiver.lock().expect(..).recv() else { return; }; is a let statement, so the MutexGuard temporary is dropped at its end, before job() runs. The pool is fully parallel, which a timing confirms: four 200 ms jobs on four workers take 205 ms. The same receive written as while let would serialize the pool (chapter 18, figure 18.2), so this is the form to memorize.

The hash ring wraps with %. partition_point(..) % self.points.len() maps an index one past the end back to 0 in one expression. The reference uses an if instead. The drill also hashes in a plainer way: it calls hasher.write(input.as_bytes()) and write_usize(replica) directly. It stores a String per virtual node where the reference shares an Arc<str>.

The request parser checks the version and little else. parse_request_head accepts only HTTP/1.1, splits headers at the first colon, and lowercases the names. It has no size limits, no Host check, and no body framing. That is the right scope for a ten-minute question about a request head. For a question about serving requests, chapter 21’s checks are the answer.

Port::new is a one-expression validator. (1..=u16::MAX).contains(&value).then_some(Self(value)).ok_or(..) reads left to right as the rule. Route::parse matches the two fixed paths and falls back to strip_prefix for the id route, returning None for everything else.

30.3 The drill that is slower than it looks

parallel_sum changed behaviour in the short form. Compare it with the reference in chapter 16:

// drills.rs
values
    .chunks(chunk_size)
    .map(|chunk| scope.spawn(move || ...))
    .map(|handle| handle.join().expect("worker panicked"))
    .sum()

// threads.rs
let handles: Vec<_> = values.chunks(chunk_size).map(|chunk| scope.spawn(...)).collect();
handles.into_iter().map(|handle| handle.join().expect("worker panicked")).sum()

Iterators are lazy. In the drill, sum pulls one item through both maps before it asks for the next. The first chunk’s thread is therefore spawned and then joined before the second chunk’s thread is spawned. The four workers run one after another. The reference collects every JoinHandle into a Vec first, which spawns all four, and only then joins them.

The result is the same number either way, so the drill’s test passes. Timing four workers that each sleep 200 ms shows the difference:

lazy spawn/join chain, four 200 ms workers: 809.402756ms (sum 4)
collect first, then join:              203.482125ms (sum 4)

The fix is one collect() between the two maps. Chapter 18 drew the same lesson from the thread pool. A concurrency test that checks only the answer cannot tell parallel from serial. Put a barrier in it, as thread_pool_v2’s jobs_run_at_the_same_time does, or time it.

30.4 Running a round

A 45-minute coding round has a shape. Planning it keeps the last ten minutes, where most of the signal is, from being spent debugging (figure 30.1).

• Clarify. Ask what the inputs are, what should happen on bad input, and what the limits are (size, concurrency, latency). Each answer becomes a type or a check.
• Types first. Write the enums, newtypes, and signatures before the bodies. A Result<T, ParseError> return type decided in minute six saves a rewrite in minute twenty-five.
• The happy path, then each edge. Get one case working end to end, then add the edges you listed.
• Tests. Test the edges you named, not only the example given.
• Failure modes. Say what breaks at scale and what you would change: the reference versions in this book are that list, written out.

30.5 Practicing with this repository

The repository is set up for the drill loop that the problem index in appendix C describes. Write a minimal correct version in 30 to 45 minutes, then explain its failure modes and how you would test them.

1. Pick a drill. Read its chapter once.
2. Close the book and the file. Write it from an empty buffer, in a scratch binary under src/bin/.
3. Run cargo test on it, then run the reference tests: cargo test thread_pool, cargo test bounded_queue, cargo test --bin http_server.
4. Compare with the reference. Write down what you left out, and whether it was a choice.
5. Say the failure modes out loud, as if to a reviewer.

The chapters’ “Questions that come up” sections are the follow-ups to rehearse. The history this book has followed, eleven merges, nine pools, six buffers, is what that loop looks like when you keep the drafts.

30.6 Questions that come up

“Your parallel sum gives the right answer. Is it parallel?” Check that all threads are spawned before any is joined. A lazy iterator that spawns and joins in one chain is serial.

“What would you add to this spin lock for production?” Spin on a relaxed load before retrying the exchange. Back off or yield under contention. Add SAFETY comments, and prefer std::sync::Mutex unless measurements say otherwise.

“Your semaphore’s caller panicked. What happened to the permit?” It leaked. Return a guard whose Drop releases it.

“How did you decide what to leave out?” By the question’s scope. Name what a full version adds, and where it would change the answer.