A client connection pool

Chapter 27 showed that each closed connection leaves a socket in TIME_WAIT, and that a busy client runs out of ports. Chapter 28 added a TLS handshake to every new connection. Both costs come from making connections. A client avoids most of them by keeping a few connections open, and sending request after request over them. A connection pool holds those connections and lends them to the threads that need them.

This chapter builds a pool on the standard library: a Mutex, a Condvar, and a guard type with Drop. The program, conn_pool, runs it against a small line server on loopback, and measures each of the pool’s rules.

29.1 What a new connection costs

Each new TCP connection costs:

• A round trip for the TCP handshake before the first byte, and a second one for TLS.
• A slow start. A new connection begins with a small congestion window and grows it over several round trips (section 20.2). A connection that has been in use already has a large one.
• Work on the server: an accept, a descriptor, and often a thread or a task.
• A socket in TIME_WAIT after the close, on whichever side closed first.

On a network with a 50 ms round trip, the handshakes alone add 100 ms. On an open connection, the same request might take 5 ms.

Keeping connections open has costs too. Each open connection holds a descriptor and buffers on both machines, and a server can hold only so many. A pool therefore keeps a bounded number of connections and lends them out one caller at a time. It follows four rules:

1. A bound. At most max connections are open, counting those lent out and those idle.
2. Automatic return. A borrowed connection comes back when the borrower drops it, unless the borrower reports it broken.
3. Newest first. The most recently returned idle connection is reused first.
4. No stale connections. A connection idle longer than idle_timeout, or one the server has closed, is closed instead of lent.

Figure 29.1 shows the path of a call to get through the four rules.

29.2 The pool’s state

Config holds the limits, and Stats counts what the pool does, so the program can print it:

use std::{
    io::{self, BufRead, BufReader, ErrorKind, Write},
    net::{SocketAddr, TcpListener, TcpStream},
    ops::{Deref, DerefMut},
    sync::{
        Arc,
        Condvar,
        Mutex,
        atomic::{AtomicUsize, Ordering::Relaxed},
    },
    thread,
    time::{Duration, Instant},
};

/// The limits a pool enforces.
#[derive(Clone, Copy)]
struct Config {
    /// The most connections open at once, lent out or idle.
    max: usize,
    /// How long `get` waits for a connection before it gives up.
    wait_timeout: Duration,
    /// How long a connection may sit idle before the pool closes it.
    idle_timeout: Duration,
    /// Whether to check that an idle connection is still open before lending it.
    check_alive: bool,
}

/// What the pool has done, for the demonstration.
#[derive(Default)]
struct Stats {
    connects: AtomicUsize,
    reuses: AtomicUsize,
    evicted: AtomicUsize,
    stale: AtomicUsize,
}

The pool keeps its idle connections in a Vec used as a stack, each with the time it was returned. open counts every connection the pool owns, idle or lent out:

struct Idle {
    stream: TcpStream,
    since: Instant,
}

/// The connections the pool owns: the idle ones, and a count of all open ones.
struct State {
    idle: Vec<Idle>,
    open: usize,
}

struct Pool {
    address: SocketAddr,
    config: Config,
    state: Mutex<State>,
    returned: Condvar,
    stats: Stats,
}

/// A connection lent out by the pool. Dropping it gives it back.
struct Pooled<'a> {
    pool: &'a Pool,
    stream: Option<TcpStream>,
    broken: bool,
}

The invariant is that open equals the number of idle connections plus the number lent out. Every change to open happens under the mutex, so get can trust it when it decides whether to connect.

Pooled is the borrowed connection. It holds a reference to its pool, so the borrow checker keeps every Pooled from outliving the pool. The stream sits in an Option, so Drop can move it out.

29.3 Borrowing a connection

get loops under the lock until it has a connection or runs out of time:

impl Pool {
    fn new(address: SocketAddr, config: Config) -> Self {
        Self {
            address,
            config,
            state: Mutex::new(State {
                idle: Vec::new(),
                open: 0,
            }),
            returned: Condvar::new(),
            stats: Stats::default(),
        }
    }

    /// Lend a connection: an idle one if there is one, a new one if the pool
    /// is below `max`, or the next one returned, waiting up to `wait_timeout`.
    fn get(&self) -> io::Result<Pooled<'_>> {
        let deadline = Instant::now() + self.config.wait_timeout;
        let mut state = self.state.lock().expect("pool lock");
        loop {
            if let Some(stream) = self.take_idle(&mut state) {
                self.stats.reuses.fetch_add(1, Relaxed);
                return Ok(self.lend(stream));
            }
            if state.open < self.config.max {
                state.open += 1;
                drop(state);
                return self.connect();
            }
            let now = Instant::now();
            if now >= deadline {
                return Err(io::Error::new(ErrorKind::TimedOut, "pool exhausted"));
            }
            state = self
                .returned
                .wait_timeout(state, deadline - now)
                .expect("pool lock")
                .0;
        }
    }
    // ...
}

Each pass tries three things in order:

• An idle connection. take_idle returns the newest usable one, or None.
• A new connection. If fewer than max are open, get counts the new one in open under the lock. Then it drops the lock and connects. A connect can take a whole round trip, or seconds if the server is slow to answer. Holding the lock that long would stop every other caller, including those returning connections. Counting the slot first keeps two callers from both deciding there is room for one more.
• Wait. Otherwise the caller sleeps on the Condvar until a connection is returned or the deadline passes. wait_timeout releases the lock while it sleeps, as in section 16.5. It can also return early, so the loop checks the state again on every wake.

A caller that cannot get a connection by its deadline gets ErrorKind::TimedOut. Waiting forever would turn an overloaded server into callers that never return. The deadline turns overload into an error the caller can handle, as with the bounded queues of chapter 17.

If the connect fails, connect gives the slot back:

impl Pool {
    // ...
    /// Open a new connection. The slot was reserved under the lock, and the
    /// connect runs outside it, so other callers are not held up.
    fn connect(&self) -> io::Result<Pooled<'_>> {
        match TcpStream::connect(self.address) {
            Ok(stream) => {
                self.stats
                    .connects
                    .fetch_add(1, Relaxed);
                Ok(self.lend(stream))
            }
            Err(error) => {
                self.release_slot();
                Err(error)
            }
        }
    }

    fn lend(&self, stream: TcpStream) -> Pooled<'_> {
        Pooled {
            pool: self,
            stream: Some(stream),
            broken: false,
        }
    }
    // ...
}

29.4 Giving it back

Drop returns the connection, so a caller cannot forget to. Early returns and ? return it too:

impl Pool {
    // ...
    /// Take back a connection. A broken one is closed, and its slot is freed.
    fn give_back(&self, stream: TcpStream, broken: bool) {
        if broken {
            return self.release_slot();
        }
        let mut state = self.state.lock().expect("pool lock");
        state.idle.push(Idle {
            stream,
            since: Instant::now(),
        });
        drop(state);
        self.returned.notify_one();
    }

    fn release_slot(&self) {
        self.state
            .lock()
            .expect("pool lock")
            .open -= 1;
        self.returned.notify_one();
    }

    /// (open connections, idle connections)
    fn sizes(&self) -> (usize, usize) {
        let state = self.state.lock().expect("pool lock");
        (state.open, state.idle.len())
    }
}

impl Pooled<'_> {
    /// Close this connection instead of returning it, after an error.
    fn discard(mut self) {
        self.broken = true;
    }
}

impl Deref for Pooled<'_> {
    type Target = TcpStream;

    fn deref(&self) -> &TcpStream {
        self.stream
            .as_ref()
            .expect("stream until drop")
    }
}

impl DerefMut for Pooled<'_> {
    fn deref_mut(&mut self) -> &mut TcpStream {
        self.stream
            .as_mut()
            .expect("stream until drop")
    }
}

impl Drop for Pooled<'_> {
    fn drop(&mut self) {
        if let Some(stream) = self.stream.take() {
            self.pool.give_back(stream, self.broken);
        }
    }
}

give_back pushes the connection onto the idle stack and wakes one waiting caller. discard marks the guard broken, so Drop closes the connection and frees its slot instead. A caller discards a connection after any error on it. A failed request may have left half a response in the socket. The next caller would read it as the start of its own.

ask puts the two together. It is how the program sends every pooled request:

/// Send one line, and read one line back.
fn request(stream: &mut TcpStream, line: &str) -> io::Result<String> {
    writeln!(stream, "{line}")?;
    let mut reply = String::new();
    if BufReader::new(&*stream).read_line(&mut reply)? == 0 {
        return Err(io::Error::new(
            ErrorKind::UnexpectedEof,
            "server closed the connection",
        ));
    }
    Ok(reply.trim_end().to_string())
}

/// One request on a pooled connection. A connection that failed is discarded,
/// so the pool does not lend it again.
fn ask(pool: &Pool, line: &str) -> io::Result<String> {
    let mut connection = pool.get()?;
    let reply = request(&mut connection, line);
    if reply.is_err() {
        connection.discard();
    }
    reply
}

29.5 Stale connections

The pool takes the newest idle connection first. Under light load, a few connections then do all the work. The rest sink to the bottom of the stack and age past idle_timeout. Taking the oldest first would keep every connection in light use, and none would ever be closed.

The server also closes connections it considers idle. Most HTTP servers do it after somewhere between 5 seconds and a few minutes. The server’s FIN arrives while the connection sits in the pool, and nothing reads it. The next caller to borrow that connection writes its request into a socket the server has closed. Its read then returns 0, or the write draws a reset.

Figure 29.2 shows the timing with this chapter’s numbers. The server closes idle connections after 50 ms, and the pool reuses the connection after 100 ms.

take_idle handles both kinds of stale connection:

/// Whether the server has closed `stream`, checked without blocking: a
/// readable socket with nothing to read has seen the server's `FIN`.
fn is_alive(stream: &TcpStream) -> bool {
    if stream.set_nonblocking(true).is_err() {
        return false;
    }
    let alive = match stream.peek(&mut [0]) {
        Err(error) => error.kind() == ErrorKind::WouldBlock,
        Ok(_) => false,
    };
    stream.set_nonblocking(false).is_ok() && alive
}

impl Pool {
    // ...
    /// The most recently returned idle connection that is fresh and open.
    /// Connections idle too long, or closed by the server, are dropped.
    fn take_idle(&self, state: &mut State) -> Option<TcpStream> {
        while let Some(idle) = state.idle.pop() {
            if idle.since.elapsed() > self.config.idle_timeout {
                state.open -= 1;
                self.stats.evicted.fetch_add(1, Relaxed);
            } else if self.config.check_alive && !is_alive(&idle.stream) {
                state.open -= 1;
                self.stats.stale.fetch_add(1, Relaxed);
            } else {
                return Some(idle.stream);
            }
        }
        None
    }
    // ...
}

is_alive peeks at the socket without blocking. On an idle connection nothing should be waiting to be read, so WouldBlock means the connection is open. A peek that returns 0 bytes has found the server’s FIN. A peek that returns data has found bytes nobody asked for. Either way, the connection is closed.

The check narrows the window, and does not close it. The server can close the connection between the peek and the request. Two more measures cover the rest:

• Set the pool’s idle_timeout shorter than the server’s idle timeout, so the pool closes connections before the server does.
• Retry an idempotent request once on a new connection when a reused one fails, with the retry policy of chapter 19.

Animation 29.1 follows the pool through borrowing, waiting, eviction, and a stale connection.

29.6 Measured

main runs four scenes against the line server. The first sends 2,000 requests from eight threads, first with a new connection for each request, then through a pool of four:

1. 2000 requests from 8 threads
  a new connection each:   2000 connections,  169.8 ms
  a pool of 4:                4 connections,   80.4 ms, 1996 reuses

On loopback a connect costs microseconds, and the pool still halved the time. Each request without the pool also left a socket in TIME_WAIT: 2,000 in 0.17 seconds. At that rate, the ephemeral ports of section 27.2.1 run out in under 3 seconds. On a real network with TLS, each of those connections would also cost two round trips.

The other scenes check one rule each:

2. The bound: 2 connections, both lent out
  third get after 103 ms: Err(Custom { kind: TimedOut, error: "pool exhausted" })
  after one is returned: Ok(()), sizes (open, idle) = (2, 1)

3. Idle eviction: idle_timeout 100 ms
  connects 2, evicted 1

4. A server that closes idle connections after 50 ms
  check_alive false: Err(Custom { kind: UnexpectedEof, error: "server closed the connection" }), stale discarded 0
  check_alive true : Ok("ok second"), stale discarded 1

With both connections lent out, the third get waited its full 100 ms deadline and failed. Once a connection came back, the next get reused it. A connection idle for 150 ms against a 100 ms limit was closed, and the second request connected again. Against a server that closes idle connections after 50 ms, the pool without the check lent a closed connection, and the request failed. With the check, the pool found the server’s FIN, discarded the connection, and connected again.

29.7 The complete program

Listing 29.1 A bounded connection pool with automatic return, idle eviction, and a liveness check. src/bin/conn_pool.rs

//! A client connection pool: at most `max` connections to one server, each
//! lent out behind a guard that gives it back when dropped. Idle connections
//! are reused, closed after `idle_timeout`, and checked before reuse in case
//! the server closed them first.
//!
//! Run with: cargo run --release --bin conn_pool

use std::{
    io::{self, BufRead, BufReader, ErrorKind, Write},
    net::{SocketAddr, TcpListener, TcpStream},
    ops::{Deref, DerefMut},
    sync::{
        Arc,
        Condvar,
        Mutex,
        atomic::{AtomicUsize, Ordering::Relaxed},
    },
    thread,
    time::{Duration, Instant},
};

/// The limits a pool enforces.
#[derive(Clone, Copy)]
struct Config {
    /// The most connections open at once, lent out or idle.
    max: usize,
    /// How long `get` waits for a connection before it gives up.
    wait_timeout: Duration,
    /// How long a connection may sit idle before the pool closes it.
    idle_timeout: Duration,
    /// Whether to check that an idle connection is still open before lending it.
    check_alive: bool,
}

/// What the pool has done, for the demonstration.
#[derive(Default)]
struct Stats {
    connects: AtomicUsize,
    reuses: AtomicUsize,
    evicted: AtomicUsize,
    stale: AtomicUsize,
}

struct Idle {
    stream: TcpStream,
    since: Instant,
}

/// The connections the pool owns: the idle ones, and a count of all open ones.
struct State {
    idle: Vec<Idle>,
    open: usize,
}

struct Pool {
    address: SocketAddr,
    config: Config,
    state: Mutex<State>,
    returned: Condvar,
    stats: Stats,
}

/// A connection lent out by the pool. Dropping it gives it back.
struct Pooled<'a> {
    pool: &'a Pool,
    stream: Option<TcpStream>,
    broken: bool,
}

/// Whether the server has closed `stream`, checked without blocking: a
/// readable socket with nothing to read has seen the server's `FIN`.
fn is_alive(stream: &TcpStream) -> bool {
    if stream.set_nonblocking(true).is_err() {
        return false;
    }
    let alive = match stream.peek(&mut [0]) {
        Err(error) => error.kind() == ErrorKind::WouldBlock,
        Ok(_) => false,
    };
    stream.set_nonblocking(false).is_ok() && alive
}

impl Pool {
    fn new(address: SocketAddr, config: Config) -> Self {
        Self {
            address,
            config,
            state: Mutex::new(State {
                idle: Vec::new(),
                open: 0,
            }),
            returned: Condvar::new(),
            stats: Stats::default(),
        }
    }

    /// Lend a connection: an idle one if there is one, a new one if the pool
    /// is below `max`, or the next one returned, waiting up to `wait_timeout`.
    fn get(&self) -> io::Result<Pooled<'_>> {
        let deadline = Instant::now() + self.config.wait_timeout;
        let mut state = self.state.lock().expect("pool lock");
        loop {
            if let Some(stream) = self.take_idle(&mut state) {
                self.stats.reuses.fetch_add(1, Relaxed);
                return Ok(self.lend(stream));
            }
            if state.open < self.config.max {
                state.open += 1;
                drop(state);
                return self.connect();
            }
            let now = Instant::now();
            if now >= deadline {
                return Err(io::Error::new(ErrorKind::TimedOut, "pool exhausted"));
            }
            state = self
                .returned
                .wait_timeout(state, deadline - now)
                .expect("pool lock")
                .0;
        }
    }

    /// The most recently returned idle connection that is fresh and open.
    /// Connections idle too long, or closed by the server, are dropped.
    fn take_idle(&self, state: &mut State) -> Option<TcpStream> {
        while let Some(idle) = state.idle.pop() {
            if idle.since.elapsed() > self.config.idle_timeout {
                state.open -= 1;
                self.stats.evicted.fetch_add(1, Relaxed);
            } else if self.config.check_alive && !is_alive(&idle.stream) {
                state.open -= 1;
                self.stats.stale.fetch_add(1, Relaxed);
            } else {
                return Some(idle.stream);
            }
        }
        None
    }

    /// Open a new connection. The slot was reserved under the lock, and the
    /// connect runs outside it, so other callers are not held up.
    fn connect(&self) -> io::Result<Pooled<'_>> {
        match TcpStream::connect(self.address) {
            Ok(stream) => {
                self.stats
                    .connects
                    .fetch_add(1, Relaxed);
                Ok(self.lend(stream))
            }
            Err(error) => {
                self.release_slot();
                Err(error)
            }
        }
    }

    fn lend(&self, stream: TcpStream) -> Pooled<'_> {
        Pooled {
            pool: self,
            stream: Some(stream),
            broken: false,
        }
    }

    /// Take back a connection. A broken one is closed, and its slot is freed.
    fn give_back(&self, stream: TcpStream, broken: bool) {
        if broken {
            return self.release_slot();
        }
        let mut state = self.state.lock().expect("pool lock");
        state.idle.push(Idle {
            stream,
            since: Instant::now(),
        });
        drop(state);
        self.returned.notify_one();
    }

    fn release_slot(&self) {
        self.state
            .lock()
            .expect("pool lock")
            .open -= 1;
        self.returned.notify_one();
    }

    /// (open connections, idle connections)
    fn sizes(&self) -> (usize, usize) {
        let state = self.state.lock().expect("pool lock");
        (state.open, state.idle.len())
    }
}

impl Pooled<'_> {
    /// Close this connection instead of returning it, after an error.
    fn discard(mut self) {
        self.broken = true;
    }
}

impl Deref for Pooled<'_> {
    type Target = TcpStream;

    fn deref(&self) -> &TcpStream {
        self.stream
            .as_ref()
            .expect("stream until drop")
    }
}

impl DerefMut for Pooled<'_> {
    fn deref_mut(&mut self) -> &mut TcpStream {
        self.stream
            .as_mut()
            .expect("stream until drop")
    }
}

impl Drop for Pooled<'_> {
    fn drop(&mut self) {
        if let Some(stream) = self.stream.take() {
            self.pool.give_back(stream, self.broken);
        }
    }
}

/// Send one line, and read one line back.
fn request(stream: &mut TcpStream, line: &str) -> io::Result<String> {
    writeln!(stream, "{line}")?;
    let mut reply = String::new();
    if BufReader::new(&*stream).read_line(&mut reply)? == 0 {
        return Err(io::Error::new(
            ErrorKind::UnexpectedEof,
            "server closed the connection",
        ));
    }
    Ok(reply.trim_end().to_string())
}

/// One request on a pooled connection. A connection that failed is discarded,
/// so the pool does not lend it again.
fn ask(pool: &Pool, line: &str) -> io::Result<String> {
    let mut connection = pool.get()?;
    let reply = request(&mut connection, line);
    if reply.is_err() {
        connection.discard();
    }
    reply
}

/// A line server on loopback that closes a connection idle for `idle_close`.
/// Returns its address and a count of the connections it has accepted.
fn start_server(idle_close: Duration) -> io::Result<(SocketAddr, Arc<AtomicUsize>)> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?;
    let accepted = Arc::new(AtomicUsize::new(0));
    let counter = Arc::clone(&accepted);
    thread::spawn(move || {
        for stream in listener.incoming().flatten() {
            counter.fetch_add(1, Relaxed);
            thread::spawn(move || serve(stream, idle_close));
        }
    });
    Ok((address, accepted))
}

/// Answer each line with "ok", until the client closes or stays quiet too long.
fn serve(stream: TcpStream, idle_close: Duration) -> io::Result<()> {
    stream.set_read_timeout(Some(idle_close))?;
    let mut writer = stream.try_clone()?;
    for line in BufReader::new(stream).lines() {
        let line = line?;
        writeln!(writer, "ok {line}")?;
    }
    Ok(())
}

fn main() -> io::Result<()> {
    const REQUESTS: usize = 2_000;
    const THREADS: usize = 8;
    let config = Config {
        max: 4,
        wait_timeout: Duration::from_secs(1),
        idle_timeout: Duration::from_secs(30),
        check_alive: true,
    };

    println!("1. {REQUESTS} requests from {THREADS} threads");
    let (address, accepted) = start_server(Duration::from_secs(30))?;
    let start = Instant::now();
    thread::scope(|scope| {
        for _ in 0..THREADS {
            scope.spawn(|| -> io::Result<()> {
                for _ in 0..REQUESTS / THREADS {
                    request(&mut TcpStream::connect(address)?, "ping")?;
                }
                Ok(())
            });
        }
    });
    println!(
        "  a new connection each:  {:>5} connections, {:>6.1} ms",
        accepted.load(Relaxed),
        start.elapsed().as_secs_f64() * 1000.0
    );

    let (address, accepted) = start_server(Duration::from_secs(30))?;
    let pool = Pool::new(address, config);
    let start = Instant::now();
    thread::scope(|scope| {
        for _ in 0..THREADS {
            scope.spawn(|| -> io::Result<()> {
                for _ in 0..REQUESTS / THREADS {
                    ask(&pool, "ping")?;
                }
                Ok(())
            });
        }
    });
    println!(
        "  a pool of {}:            {:>5} connections, {:>6.1} ms, {} reuses",
        config.max,
        accepted.load(Relaxed),
        start.elapsed().as_secs_f64() * 1000.0,
        pool.stats.reuses.load(Relaxed)
    );

    println!("\n2. The bound: 2 connections, both lent out");
    let pool = Pool::new(
        address,
        Config {
            max: 2,
            wait_timeout: Duration::from_millis(100),
            ..config
        },
    );
    let (first, second) = (pool.get()?, pool.get()?);
    let start = Instant::now();
    let third = pool.get().map(|_| ());
    println!(
        "  third get after {:.0} ms: {third:?}",
        start.elapsed().as_secs_f64() * 1000.0
    );
    drop(first);
    let fourth = pool.get().map(|_| ());
    println!(
        "  after one is returned: {fourth:?}, sizes (open, idle) = {:?}",
        pool.sizes()
    );
    drop(second);

    println!("\n3. Idle eviction: idle_timeout 100 ms");
    let pool = Pool::new(
        address,
        Config {
            idle_timeout: Duration::from_millis(100),
            ..config
        },
    );
    ask(&pool, "first")?;
    thread::sleep(Duration::from_millis(150));
    ask(&pool, "second")?;
    println!(
        "  connects {}, evicted {}",
        pool.stats.connects.load(Relaxed),
        pool.stats.evicted.load(Relaxed)
    );

    println!("\n4. A server that closes idle connections after 50 ms");
    let (address, _) = start_server(Duration::from_millis(50))?;
    for check_alive in [false, true] {
        let pool = Pool::new(
            address,
            Config {
                check_alive,
                ..config
            },
        );
        ask(&pool, "first")?;
        thread::sleep(Duration::from_millis(100));
        let outcome = ask(&pool, "second");
        println!(
            "  check_alive {check_alive:<5}: {outcome:?}, stale discarded {}",
            pool.stats.stale.load(Relaxed)
        );
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    fn config(max: usize) -> Config {
        Config {
            max,
            wait_timeout: Duration::from_millis(50),
            idle_timeout: Duration::from_secs(30),
            check_alive: true,
        }
    }

    #[test]
    fn a_returned_connection_is_reused() {
        let (address, accepted) = start_server(Duration::from_secs(5)).unwrap();
        let pool = Pool::new(address, config(2));
        for _ in 0..10 {
            assert_eq!(request(&mut pool.get().unwrap(), "x").unwrap(), "ok x");
        }
        assert_eq!(accepted.load(Relaxed), 1);
        assert_eq!(pool.stats.reuses.load(Relaxed), 9);
    }

    #[test]
    fn the_pool_never_exceeds_max() {
        let (address, _) = start_server(Duration::from_secs(5)).unwrap();
        let pool = Pool::new(address, config(1));
        let held = pool.get().unwrap();
        let error = pool.get().map(|_| ()).unwrap_err();
        assert_eq!(error.kind(), ErrorKind::TimedOut);
        drop(held);
        assert!(pool.get().is_ok());
    }

    #[test]
    fn a_discarded_connection_frees_its_slot() {
        let (address, _) = start_server(Duration::from_secs(5)).unwrap();
        let pool = Pool::new(address, config(1));
        pool.get().unwrap().discard();
        assert_eq!(pool.sizes(), (0, 0));
        drop(pool.get().unwrap());
        assert_eq!(pool.stats.connects.load(Relaxed), 2);
    }
}

$ cargo run --release --bin conn_pool
1. 2000 requests from 8 threads
  a new connection each:   2000 connections,  169.8 ms
  a pool of 4:                4 connections,   80.4 ms, 1996 reuses

2. The bound: 2 connections, both lent out
  third get after 103 ms: Err(Custom { kind: TimedOut, error: "pool exhausted" })
  after one is returned: Ok(()), sizes (open, idle) = (2, 1)

3. Idle eviction: idle_timeout 100 ms
  connects 2, evicted 1

4. A server that closes idle connections after 50 ms
  check_alive false: Err(Custom { kind: UnexpectedEof, error: "server closed the connection" }), stale discarded 0
  check_alive true : Ok("ok second"), stale discarded 1

29.8 Questions that come up

“How large should the pool be?” Large enough for the requests in flight at peak. That is the request rate times the time each request holds a connection. At 2,000 requests per second and 5 ms each, that is 10. Each client process multiplies it on the server, so the server’s connection limit divided by the number of clients is the upper bound.

“Why not one pool shared by every server address?” A connection is to one address. Real clients keep one pool per host and port, often with a limit per host and one overall.

“What does HTTP/2 change?” HTTP/2 sends many requests at once over one connection, each in its own stream. A client then needs one connection per server, or a few, and the pool limits streams rather than connections.

“What happens when the server restarts?” Every idle connection in the pool is closed by the server at once. The liveness check catches most of them. Requests already in flight fail, and the idempotent ones can be retried.

“Is a Mutex too slow for a pool?” The lock is held for a few pushes and pops. It is never held across a connect or a request. At hundreds of thousands of requests per second, a pool can split into shards with one lock each, as chapter 14 split a map.

Chapter 30 collects compact versions of the structures from the earlier parts. Each is the shortest program that still shows the structure working.

Exercises

1. Add a max_lifetime to Config, and close a connection older than it even if it is in use often. Why do production pools have this limit, when the server’s address can change behind a DNS name?
2. Give get a fairness guarantee: callers that started waiting first get connections first. What does Condvar::notify_one promise about order, and what structure would you add?
3. Make ask retry once on a new connection when a reused connection fails, but only if the request is marked idempotent.
4. Run scene 1 with max set to 1, 2, 8, and 16, and plot the time. Where does the curve flatten, and why?
5. Wrap the pool’s connections in TLS from chapter 28, and measure scene 1 again with and without the pool.