Addresses, windows, and echo servers

This chapter works up from the bottom of the network stack. Section 20.1 covers addresses, the numbers that name a machine. Section 20.2 covers how much data TCP lets a sender have on the wire at once, and why that limits speed on long links. Sections 20.3 and 20.4 build two echo servers. The first gives every client its own thread. The second serves every client from one thread with epoll. It uses no crate beyond libc, so every system call it makes is visible.

20.1 IPv4 addresses

An IPv4 address names one network interface on one machine. It is a 32-bit number. People write it as four bytes in decimal, separated by dots, such as 10.0.0.1. Each of the four parts is called an octet, so each must be between 0 and 255.

An address alone names a machine. A connection also needs to name a program on that machine. That is the job of the port, a 16-bit number. An address and a port together, such as 127.0.0.1:8080, are a socket address.

Some addresses are reserved for private networks: offices, homes, and the inside of a data center. Routers on the public internet do not forward them. Three ranges are private:

The /8 means that the first 8 bits are fixed and the rest can be anything.

When an address travels inside a packet, it is four bytes in a fixed order: the first octet first. That order is called network byte order, and it is big-endian: the first byte is the most significant. Most CPUs store numbers little-endian, with the least significant byte first. Code that puts an address into a packet has to convert it (figure 20.1).

The program net_ipv4.rs does these three jobs: parse the text, check the private ranges, and convert to and from network order.

parse splits the text on dots and parses each part as a u8:

/// Parse `a.b.c.d`, rejecting anything that is not exactly four octets.
fn parse(input: &str) -> Option<[u8; 4]> {
    let mut parts = input
        .split('.')
        .map(|part| part.parse::<u8>().ok());
    let address = [
        parts.next()??,
        parts.next()??,
        parts.next()??,
        parts.next()??,
    ];
    parts
        .next()
        .is_none()
        .then_some(address)
}

parse is four parts.next()?? expressions and a check that nothing is left over. The iterator yields Option<u8> (the result of part.parse::<u8>().ok()), so next() returns Option<Option<u8>>. The first ? returns None if there is no fourth part. The second returns None if a part is not a valid u8, which is how "256.1.1.1" is rejected. parts.next().is_none().then_some(address) rejects a fifth part.

is_private is the table above, written as one pattern:

/// `10.0.0.0/8`, `172.16.0.0/12`, and `192.168.0.0/16` are not routable on the
/// public internet.
fn is_private(address: [u8; 4]) -> bool {
    matches!(address, [10, ..] | [172, 16..=31, ..] | [192, 168, ..])
}

[10, ..] matches any address starting with 10. [172, 16..=31, ..] uses a range pattern for the second octet: exactly the /12 block from 172.16 to 172.31. Rust’s patterns are shorter here than the bit masks they replace.

The conversions are one standard-library call each:

/// Addresses travel big-endian, so the first octet is the most significant byte.
fn to_network_order(address: [u8; 4]) -> u32 {
    u32::from_be_bytes(address)
}

fn from_network_order(value: u32) -> [u8; 4] {
    value.to_be_bytes()
}

u32::from_be_bytes reads the four octets as a big-endian number on any machine, whatever its native order. to_be_bytes goes back. The program prints each input with its result:

input            parsed              private       as u32
10.0.0.1         [10, 0, 0, 1]          true    167772161
172.16.5.4       [172, 16, 5, 4]        true   2886731012
192.168.1.10     [192, 168, 1, 10]      true   3232235786
8.8.8.8          [8, 8, 8, 8]          false    134744072
256.1.1.1        invalid
1.2.3            invalid

all checks passed

str::parse::<u8> accepts a leading + and leading zeros, so "+1.02.3.4" parses. The standard library’s "1.2.3.4".parse::<std::net::Ipv4Addr>() rejects both, and Ipv4Addr::is_private implements the same three ranges. Production code uses Ipv4Addr.

20.2 How much data can be in flight

TCP turns an unreliable network into a reliable byte stream. The sender numbers every byte it sends. The receiver answers with an acknowledgment, or ACK, that says how many bytes have arrived in order. A byte the sender has sent, but not yet seen acknowledged, is in flight. If its ACK never comes, the sender sends it again.

The time from sending a byte to receiving its ACK is the round-trip time, or RTT. Between two machines in one data center it is under a millisecond. Across a continent it is tens of milliseconds.

A sender may not send without limit. It keeps at most a window of bytes in flight, and stops when the window is full. Two limits set the window:

• The receive window is how much the receiver can buffer. The receiver states it in every ACK. This is flow control: it stops a fast sender from overrunning a slow reader.
• The congestion window is the sender’s own estimate of what the network can carry. It grows while ACKs return and shrinks when packets are lost. This is congestion control.

The sender uses the smaller of the two. Once that many bytes are in flight, it waits for an ACK.

The window decides how fast a single connection can go. The bytes a link holds while they travel are its bandwidth times the round-trip time, the bandwidth-delay product. If the window is smaller than that, the sender runs out of window before the first ACK returns. The link then sits idle for the rest of each round trip. A worked example: a 1 GB/s link with a 100 ms round trip holds 100,000,000 bytes in flight. A 64,000-byte window fills 0.064% of it, so the connection moves 64,000 bytes per 100 ms, which is 640 KB/s (figure 20.2).

The program net_window.rs computes both quantities. The effective window is the smaller of the two limits:

/// The sender may have this much unacknowledged data in flight.
fn effective_window(receive: u32, congestion: u32) -> u32 {
    receive.min(congestion)
}

The bandwidth-delay product is a multiplication, done with care for integer precision:

/// Bytes in flight needed to keep a link busy: bandwidth times round-trip time.
fn bandwidth_delay_product(bytes_per_second: u64, rtt_millis: u64) -> u64 {
    bytes_per_second * rtt_millis / 1_000
}

It multiplies before it divides, bytes_per_second * rtt_millis / 1_000, which keeps precision in integer arithmetic. The intermediate product overflows u64 only when bytes per second times milliseconds passes about 1.8 × 10¹⁹, far beyond any real link. The program prints:

effective window = min(receiver window, congestion window)
  receiver   64000 B, congestion   16000 B ->   16000 B
  receiver   16000 B, congestion   64000 B ->   16000 B
  receiver   32000 B, congestion   32000 B ->   32000 B

bandwidth-delay product (bytes needed in flight)
       1000000 B/s x   1 ms ->         1000 B
      10000000 B/s x  50 ms ->       500000 B
    1000000000 B/s x 100 ms ->    100000000 B

window 64000 B vs product 100000000 B: link is starved, the window too small

all checks passed

For bulk transfers, latency limits a connection as much as bandwidth does. TCP’s original window field is 16 bits, so it caps at 65,535 bytes. The window-scaling option multiplies it, so a window can reach the size of a fast, long link. On Linux, net.ipv4.tcp_rmem and tcp_wmem bound how large the kernel lets the windows grow.

20.3 A thread-per-connection echo server

An echo server writes back every byte it receives. It is a small server, but it still accepts connections, reads, writes, and notices when the client leaves.

A server reaches the network through a socket, a file descriptor that the kernel connects to a network endpoint. A listening server goes through a fixed sequence of system calls (figure 20.3):

1. socket creates the socket and returns its file descriptor.
2. bind gives it a socket address, so clients know where to connect.
3. listen turns it into a listening socket. The kernel now completes TCP handshakes for it and keeps the finished connections in an accept queue.
4. accept takes one connection from that queue and returns a new file descriptor for it. The listening socket stays open for the next client.
5. read and write on the new descriptor move bytes in both directions.
6. close ends the connection. The kernel sends a FIN, the TCP message that says “no more bytes from me”.

std::net wraps these calls. TcpListener::bind performs socket, bind, and listen. Iterating over incoming() calls accept. The simplest way to serve several clients at once is to give each accepted connection its own thread:

/// Accept connections forever, handling each on its own thread.
pub fn serve(listener: TcpListener) {
    for incoming in listener.incoming() {
        match incoming {
            Ok(stream) => {
                thread::spawn(move || {
                    if let Err(error) = handle_connection(stream) {
                        eprintln!("connection error: {error}");
                    }
                });
            }
            Err(error) => eprintln!("accept error: {error}"),
        }
    }
}

Binding to port 0 asks the kernel for any free port. local_addr() reports which one it chose, which is how both main and the test avoid colliding with other programs. listener.incoming() is an endless iterator over accept, and each accepted stream moves into its own thread.

Each thread runs the whole protocol for one client:

/// Echo every byte back until the peer closes the connection.
fn handle_connection(mut stream: TcpStream) -> std::io::Result<()> {
    stream.set_nodelay(true)?;
    if let Ok(peer) = stream.peer_addr() {
        println!("connection from {peer}");
    }

    let mut buffer = [0u8; 4096];
    loop {
        let read = stream.read(&mut buffer)?;
        if read == 0 {
            // A zero-byte read is the peer's FIN: an orderly close.
            return Ok(());
        }
        stream.write_all(&buffer[..read])?;
    }
}

• set_nodelay(true) disables Nagle’s algorithm. Nagle delays small writes to combine them into fewer packets. That is wrong for an echo server, whose reply is never followed by more data.
• read returns how many bytes arrived, which can be anything from 1 to the buffer size, no matter how the client wrote them. TCP is a byte stream with no message boundaries.
• read returning 0 means the peer sent FIN: an orderly close. The function returns Ok(()) and the stream is closed when it is dropped.
• write_all loops over partial writes until every byte is sent.

The test is a model of how to test a server. It binds an ephemeral port, runs handle_connection for exactly one connection in a thread, writes "hello echo", and then half-closes with shutdown(Shutdown::Write). The half-close sends FIN, so the server’s read returns 0 and the server finishes. The client’s read side stays open, so read_to_end can collect the echo.

Thread per connection is the right default when connections are few or handlers block on other I/O. Each thread costs a kernel stack and scheduling work. At thousands of mostly idle connections, the model spends its memory on stacks that wait. That is the case for the next program.

20.4 An epoll echo server

A thread blocked in read is waiting for one socket. To serve thousands of sockets with one thread, the thread must never wait on any single one. Two kernel features make that possible.

A non-blocking socket never makes its caller wait. When there is nothing to read, read returns at once with the error EAGAIN, which means “try again later”. When the kernel’s send buffer for the socket is full, write returns EAGAIN the same way.

epoll tells the program which sockets have work. The program creates one epoll instance and adds every socket to its interest list. For each socket it says what to watch for: EPOLLIN for “readable” and EPOLLOUT for “writable”. When data arrives on a socket, the kernel puts that socket on the instance’s ready list. epoll_wait sleeps until the ready list is not empty, then returns it (figure 20.4).

The server becomes a loop: wait for ready sockets, do whatever work is possible on each without blocking, and wait again. Figure 20.5 contrasts this with thread per connection.

The program epoll_echo.rs is Linux only. The whole implementation lives in mod linux behind #[cfg(target_os = "linux")], and a second main for every other platform prints a message instead. The program compiles everywhere and runs where epoll exists.

20.4.1 Setting up the listener by hand

bind_listener is what TcpListener::bind does inside, spelled out. It calls C functions through the libc crate. The crate declares them inside an unsafe extern "C" block, with C’s types. c_int is a C int, and socklen_t is the size of an address in bytes. These are the five it uses, with what each parameter means:

unsafe extern "C" {
    fn socket(domain: c_int, ty: c_int, protocol: c_int) -> c_int; // AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0
    fn setsockopt(
        fd: c_int,
        level: c_int,          // SOL_SOCKET
        name: c_int,           // SO_REUSEADDR
        value: *const c_void,  // points at the option's value
        len: socklen_t,        // the value's size in bytes
    ) -> c_int;
    fn bind(fd: c_int, addr: *const sockaddr, len: socklen_t) -> c_int; // give the socket an address
    fn listen(fd: c_int, backlog: c_int) -> c_int; // backlog: how many connections may wait for accept
    fn getsockname(fd: c_int, addr: *mut sockaddr, len: *mut socklen_t) -> c_int; // read back the address
}

Each returns 0, or a new descriptor for socket, and -1 on failure with the reason in errno. sockaddr is a generic address type. An IPv4 address is a sockaddr_in, passed as a pointer to sockaddr together with its size. The epoll calls the loop uses later are declared the same way:

unsafe extern "C" {
    fn epoll_create1(flags: c_int) -> c_int; // a new epoll instance's fd
    fn epoll_ctl(epfd: c_int, op: c_int, fd: c_int, event: *mut epoll_event) -> c_int; // add, modify, remove
    fn epoll_wait(epfd: c_int, events: *mut epoll_event, maxevents: c_int, timeout: c_int) -> c_int;
}

An epoll_event has two fields: events, the flags such as EPOLLIN, and u64, a number the program chooses. This program stores the fd there, so each event says which socket it is about. epoll_wait fills an array of them and returns how many it filled.

The function:

mod linux {
    // ...
    /// Bind a non-blocking listener on `127.0.0.1:port`; port 0 asks the kernel for
    /// a free port, which is returned alongside the fd.
    pub fn bind_listener(port: u16) -> io::Result<(i32, u16)> {
        // SAFETY: all pointers are to stack values that outlive the call.
        unsafe {
            let fd = libc::socket(libc::AF_INET, libc::SOCK_STREAM | libc::SOCK_NONBLOCK, 0);
            if fd < 0 {
                return Err(io::Error::last_os_error());
            }

            let enable: libc::c_int = 1;
            if libc::setsockopt(
                fd,
                libc::SOL_SOCKET,
                libc::SO_REUSEADDR,
                &enable as *const _ as *const libc::c_void,
                std::mem::size_of::<libc::c_int>() as libc::socklen_t,
            ) < 0
            {
                let error = io::Error::last_os_error();
                libc::close(fd);
                return Err(error);
            }

            let mut address: libc::sockaddr_in = std::mem::zeroed();
            address.sin_family = libc::AF_INET as libc::sa_family_t;
            address.sin_port = port.to_be();
            address.sin_addr.s_addr = libc::INADDR_LOOPBACK.to_be();

            if libc::bind(
                fd,
                &address as *const _ as *const libc::sockaddr,
                std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t,
            ) < 0
            {
                let error = io::Error::last_os_error();
                libc::close(fd);
                return Err(error);
            }

            if libc::listen(fd, 128) < 0 {
                let error = io::Error::last_os_error();
                libc::close(fd);
                return Err(error);
            }

            let mut bound: libc::sockaddr_in = std::mem::zeroed();
            let mut length = std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t;
            if libc::getsockname(fd, &mut bound as *mut _ as *mut libc::sockaddr, &mut length) < 0 {
                let error = io::Error::last_os_error();
                libc::close(fd);
                return Err(error);
            }

            Ok((fd, u16::from_be(bound.sin_port)))
        }
    }
    // ...
}

socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0) creates a non-blocking socket in one call. SO_REUSEADDR lets a restarted server bind while old connections sit in TIME_WAIT. The sockaddr_in is built with std::mem::zeroed(), and its fields are converted to network order: port.to_be() and INADDR_LOOPBACK.to_be(). That is the byte order of section 20.1, used where the kernel reads it. Every failure path closes the fd before returning the error, so nothing leaks, and the SAFETY comment explains why the raw pointers are valid.

20.4.2 The loop

Each client needs a place for bytes that could not be written yet:

mod linux {
    // ...
    /// Per-connection state. `out`/`out_pos` form a write buffer that survives a
    /// partial write.
    #[derive(Default)]
    struct Connection {
        out: Vec<u8>,
        out_pos: usize,
    }

    impl Connection {
        fn has_pending_output(&self) -> bool {
            self.out_pos < self.out.len()
        }
    // ...
}

Connection is two fields, out and out_pos, and has_pending_output compares them. That buffer keeps one slow client harmless. Its reply waits in memory, and the loop moves on to other sockets instead of blocking in write.

event_loop is the wait-and-dispatch loop:

mod linux {
    // ...
    /// Run the event loop until `shutdown` becomes true.
    pub fn event_loop(epfd: i32, listen_fd: i32, shutdown: &AtomicBool) -> io::Result<()> {
        epoll_add(epfd, listen_fd, libc::EPOLLIN as u32)?;

        let mut connections: HashMap<i32, Connection> = HashMap::new();
        let mut events = vec![libc::epoll_event { events: 0, u64: 0 }; MAX_EVENTS];

        while !shutdown.load(Ordering::Relaxed) {
            // SAFETY: `events` is a valid, writable buffer of MAX_EVENTS entries.
            let ready = unsafe {
                libc::epoll_wait(
                    epfd,
                    events.as_mut_ptr(),
                    events.len() as i32,
                    EPOLL_TIMEOUT_MS,
                )
            };

            if ready < 0 {
                let error = io::Error::last_os_error();
                if error.raw_os_error() == Some(libc::EINTR) {
                    continue;
                }
                return Err(error);
            }

            for event in &events[..ready as usize] {
                let fd = event.u64 as i32;
                if fd == listen_fd {
                    accept_ready(epfd, listen_fd, &mut connections)?;
                    continue;
                }

                let readable = event.events & (libc::EPOLLIN as u32) != 0;
                let writable = event.events & (libc::EPOLLOUT as u32) != 0;

                if readable && !read_into(fd, &mut connections)? {
                    continue; // connection closed
                }
                if writable && connections.contains_key(&fd) {
                    flush(fd, &mut connections)?;
                }
                if connections.contains_key(&fd) {
                    update_interest(epfd, fd, &mut connections)?;
                }
            }
        }

        Ok(())
    }
    // ...
}

It registers the listening socket for EPOLLIN, then waits, at most 64 events at a time. The wait has a 100 ms timeout, so the loop re-checks the shutdown flag regularly. EINTR (a signal arrived during the wait) is not an error; the loop continues. Each event carries the fd in its u64 field, put there at registration.

When the listener is readable, new connections are waiting in the accept queue:

mod linux {
    // ...
    /// Accept every pending connection, then return; `accept4` with
    /// `SOCK_NONBLOCK` means a client socket can never block the loop.
    fn accept_ready(
        epfd: i32,
        listen_fd: i32,
        connections: &mut HashMap<i32, Connection>,
    ) -> io::Result<()> {
        loop {
            // SAFETY: a null sockaddr is explicitly allowed for accept.
            let client = unsafe {
                libc::accept4(
                    listen_fd,
                    std::ptr::null_mut(),
                    std::ptr::null_mut(),
                    libc::SOCK_NONBLOCK,
                )
            };

            if client < 0 {
                if would_block() {
                    return Ok(());
                }
                return Err(io::Error::last_os_error());
            }

            epoll_add(epfd, client, libc::EPOLLIN as u32)?;
            connections.insert(client, Connection::default());
        }
    }
    // ...
}

accept_ready calls accept4 with SOCK_NONBLOCK in a loop until it returns EAGAIN, registering each new client for EPOLLIN. Accepting until EAGAIN drains a burst of connections in one wake-up.

When a client is readable, its bytes go into its output buffer:

mod linux {
    // ...
    /// Drain readable bytes into the connection's output buffer.
    ///
    /// Returns `false` when the peer closed the connection.
    fn read_into(fd: i32, connections: &mut HashMap<i32, Connection>) -> io::Result<bool> {
        let mut buffer = [0u8; 4096];
        loop {
            // SAFETY: `buffer` is a valid writable region of `buffer.len()` bytes.
            let read =
                unsafe { libc::read(fd, buffer.as_mut_ptr() as *mut libc::c_void, buffer.len()) };

            if read == 0 {
                close_connection(fd, connections);
                return Ok(false);
            }
            if read < 0 {
                if would_block() {
                    break;
                }
                close_connection(fd, connections);
                return Err(io::Error::last_os_error());
            }

            if let Some(connection) = connections.get_mut(&fd) {
                connection
                    .out
                    .extend_from_slice(&buffer[..read as usize]);
            }
        }

        flush(fd, connections)?;
        Ok(true)
    }
    // ...
}

read_into reads until EAGAIN, appending everything to the connection’s out buffer, then tries to flush it. A read of 0 closes the connection.

flush writes as much as the socket accepts:

mod linux {
    // ...
    /// Write as much of the buffered reply as the socket accepts.
    fn flush(fd: i32, connections: &mut HashMap<i32, Connection>) -> io::Result<()> {
        let Some(connection) = connections.get_mut(&fd) else {
            return Ok(());
        };

        while connection.has_pending_output() {
            let remaining = connection.out.len() - connection.out_pos;
            // SAFETY: the slice is valid for `remaining` bytes.
            let written = unsafe {
                libc::write(
                    fd,
                    connection.out[connection.out_pos..].as_ptr() as *const libc::c_void,
                    remaining,
                )
            };

            if written < 0 {
                if would_block() {
                    // Keep the rest buffered; EPOLLOUT will drive the next attempt.
                    return Ok(());
                }
                return Err(io::Error::last_os_error());
            }
            connection.out_pos += written as usize;
        }

        connection.out.clear();
        connection.out_pos = 0;
        Ok(())
    }
    // ...
}

It writes from out_pos onward. When write returns EAGAIN, the socket’s send buffer is full, and the rest stays queued.

After either, update_interest sets what the loop waits for on this socket:

mod linux {
    // ...
    /// Ask for `EPOLLOUT` only while a reply is still buffered.
    fn update_interest(
        epfd: i32,
        fd: i32,
        connections: &mut HashMap<i32, Connection>,
    ) -> io::Result<()> {
        let pending = connections
            .get(&fd)
            .is_some_and(Connection::has_pending_output);
        let events = if pending {
            (libc::EPOLLIN | libc::EPOLLOUT) as u32
        } else {
            libc::EPOLLIN as u32
        };
        epoll_modify(epfd, fd, events)
    }
    // ...
}

It asks for EPOLLOUT only while output is pending. A socket with room in its send buffer is almost always writable, so leaving EPOLLOUT on permanently would wake the loop continuously for nothing.

main ignores SIGPIPE. By default, writing to a socket whose peer has gone away raises SIGPIPE, which kills the process. With the signal ignored, write returns EPIPE instead. Rust’s standard library does this for you at startup on Unix, but a program that makes raw libc calls should not rely on it.

20.4.3 What a production server would change

The program is a complete, tested event loop, and reading it critically is good practice.

• One client’s error stops the server. read_into and flush return Err on any failure other than EAGAIN, and event_loop propagates it with ?. A client that resets its connection (ECONNRESET) makes the whole loop return. A per-connection error should close that connection and continue.
• The output buffer is unbounded. A client that sends continuously and never reads grows out without limit. Backpressure here means turning off EPOLLIN for that client while its buffer is over a threshold.
• epoll_ctl runs after every event. update_interest issues a EPOLL_CTL_MOD system call even when the interest set has not changed. Tracking the current set per connection removes most of them.
• Level-triggered mode. The loop registers without EPOLLET, so epoll keeps reporting a socket while it stays ready. Reading until EAGAIN is required in edge-triggered mode and harmless here.

Frameworks such as mio (under Tokio) wrap exactly this loop, portably over epoll, kqueue, and IOCP. Chapter 22 builds the other half of an async runtime, the part that turns readiness into resumed tasks.

20.5 The complete programs

Listing 20.1 Parse, classify, and convert IPv4 addresses. src/bin/net_ipv4.rs

//! IPv4 addresses: parse a dotted quad, check the private ranges, and show the
//! network byte order used on the wire.
//!
//! Run with: cargo run --bin net_ipv4

/// Parse `a.b.c.d`, rejecting anything that is not exactly four octets.
fn parse(input: &str) -> Option<[u8; 4]> {
    let mut parts = input
        .split('.')
        .map(|part| part.parse::<u8>().ok());
    let address = [
        parts.next()??,
        parts.next()??,
        parts.next()??,
        parts.next()??,
    ];
    parts
        .next()
        .is_none()
        .then_some(address)
}

/// `10.0.0.0/8`, `172.16.0.0/12`, and `192.168.0.0/16` are not routable on the
/// public internet.
fn is_private(address: [u8; 4]) -> bool {
    matches!(address, [10, ..] | [172, 16..=31, ..] | [192, 168, ..])
}

/// Addresses travel big-endian, so the first octet is the most significant byte.
fn to_network_order(address: [u8; 4]) -> u32 {
    u32::from_be_bytes(address)
}

fn from_network_order(value: u32) -> [u8; 4] {
    value.to_be_bytes()
}

fn main() {
    println!(
        "{:<16} {:<18} {:>8} {:>12}",
        "input", "parsed", "private", "as u32"
    );

    for text in [
        "10.0.0.1",
        "172.16.5.4",
        "192.168.1.10",
        "8.8.8.8",
        "256.1.1.1",
        "1.2.3",
    ] {
        match parse(text) {
            Some(address) => println!(
                "{text:<16} {:<18} {:>8} {:>12}",
                format!("{address:?}"),
                is_private(address),
                to_network_order(address)
            ),
            None => println!("{text:<16} invalid"),
        }
    }

    assert_eq!(parse("10.0.0.1"), Some([10, 0, 0, 1]));
    assert_eq!(parse("1.2.3"), None);
    assert_eq!(parse("1.2.3.4.5"), None);
    assert_eq!(parse("256.0.0.1"), None);

    assert!(is_private([10, 1, 2, 3]));
    assert!(is_private([172, 31, 0, 1]));
    assert!(!is_private([172, 32, 0, 1]));
    assert!(!is_private([8, 8, 8, 8]));

    assert_eq!(to_network_order([127, 0, 0, 1]), 0x7f00_0001);
    assert_eq!(from_network_order(0x7f00_0001), [127, 0, 0, 1]);

    println!("\nall checks passed");
}

Listing 20.2 The effective window and the bandwidth-delay product. src/bin/net_window.rs

//! TCP window math: the effective window is the smaller of the receiver and
//! congestion windows, and the bandwidth-delay product is how many bytes must be
//! in flight to fill a link.
//!
//! Run with: cargo run --bin net_window

/// The sender may have this much unacknowledged data in flight.
fn effective_window(receive: u32, congestion: u32) -> u32 {
    receive.min(congestion)
}

/// Bytes in flight needed to keep a link busy: bandwidth times round-trip time.
fn bandwidth_delay_product(bytes_per_second: u64, rtt_millis: u64) -> u64 {
    bytes_per_second * rtt_millis / 1_000
}

fn main() {
    println!("effective window = min(receiver window, congestion window)");
    for (receive, congestion) in [(64_000, 16_000), (16_000, 64_000), (32_000, 32_000)] {
        println!(
            "  receiver {receive:>7} B, congestion {congestion:>7} B -> {:>7} B",
            effective_window(receive, congestion)
        );
    }

    println!("\nbandwidth-delay product (bytes needed in flight)");
    for (rate, rtt) in [(1_000_000, 1), (10_000_000, 50), (1_000_000_000, 100)] {
        println!(
            "  {:>12} B/s x {rtt:>3} ms -> {:>12} B",
            rate,
            bandwidth_delay_product(rate, rtt)
        );
    }

    // A window smaller than the product leaves the link idle, no matter how much
    // bandwidth is available. This is why latency, not only bandwidth, matters.
    let product = bandwidth_delay_product(1_000_000_000, 100);
    let window = u64::from(effective_window(64_000, 1_000_000));
    println!(
        "\nwindow {window} B vs product {product} B: link is {} the window too small",
        if window < product {
            "starved,"
        } else {
            "usable,"
        }
    );

    assert_eq!(effective_window(64_000, 16_000), 16_000);
    assert_eq!(effective_window(16_000, 64_000), 16_000);
    assert_eq!(bandwidth_delay_product(10_000_000, 50), 500_000);
    assert!(window < product);

    println!("\nall checks passed");
}

Listing 20.3 A thread-per-connection echo server, with its test. src/bin/tcp_echo_server.rs

//! TCP echo server: the canonical socket exercise.
//!
//! Demonstrates the server side of the TCP lifecycle: `bind` -> `listen` ->
//! `accept` -> read/write loop -> peer `close` (a `read` of 0). It uses a
//! thread-per-connection model, which is the right default for bounded concurrency
//! and blocking handlers; the event-loop alternative is `src/bin/epoll_echo.rs`.
//!
//! `TCP_NODELAY` is set because an echo server must not sit on Nagle's algorithm
//! waiting to coalesce a reply that will never be followed by more data.
//!
//! ```bash
//! cargo run --bin tcp_echo_server            # binds 127.0.0.1:0, prints the port
//! cargo run --bin tcp_echo_server 0.0.0.0:9000
//! ```

use std::{
    io::{Read, Write},
    net::{TcpListener, TcpStream},
    thread,
};

/// Accept connections forever, handling each on its own thread.
pub fn serve(listener: TcpListener) {
    for incoming in listener.incoming() {
        match incoming {
            Ok(stream) => {
                thread::spawn(move || {
                    if let Err(error) = handle_connection(stream) {
                        eprintln!("connection error: {error}");
                    }
                });
            }
            Err(error) => eprintln!("accept error: {error}"),
        }
    }
}

/// Echo every byte back until the peer closes the connection.
fn handle_connection(mut stream: TcpStream) -> std::io::Result<()> {
    stream.set_nodelay(true)?;
    if let Ok(peer) = stream.peer_addr() {
        println!("connection from {peer}");
    }

    let mut buffer = [0u8; 4096];
    loop {
        let read = stream.read(&mut buffer)?;
        if read == 0 {
            // A zero-byte read is the peer's FIN: an orderly close.
            return Ok(());
        }
        stream.write_all(&buffer[..read])?;
    }
}

fn main() {
    let address = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "127.0.0.1:0".to_string());

    let listener = TcpListener::bind(&address).expect("failed to bind");
    let local = listener
        .local_addr()
        .expect("failed to read local address");
    println!("tcp echo server listening on {local}");

    serve(listener);
}

#[cfg(test)]
mod tests {
    use std::net::Shutdown;

    use super::*;

    #[test]
    fn echoes_bytes_until_the_client_closes() {
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
        let address = listener
            .local_addr()
            .expect("local address");

        // Accept exactly one connection, echo it, then return so the test can join.
        let server = thread::spawn(move || {
            let (stream, _) = listener
                .accept()
                .expect("accept one connection");
            handle_connection(stream).expect("echo");
        });

        let mut client = TcpStream::connect(address).expect("connect");
        client
            .write_all(b"hello echo")
            .expect("write");
        client
            .shutdown(Shutdown::Write)
            .expect("half close");

        let mut echoed = Vec::new();
        client
            .read_to_end(&mut echoed)
            .expect("read echo");

        server
            .join()
            .expect("server thread panicked");
        assert_eq!(echoed, b"hello echo");
    }
}

Listing 20.4 One thread, many non-blocking sockets, over libc. src/bin/epoll_echo.rs

//! epoll echo server: one thread, many non-blocking sockets.
//!
//! Reads run until `EAGAIN`, and a reply that does not fit the socket buffer is
//! queued and finished under `EPOLLOUT`, so one slow client cannot stall the loop.
//! Linux only; kqueue and IOCP are the portable equivalents.

#[cfg(target_os = "linux")]
mod linux {
    use std::{
        collections::HashMap,
        io,
        sync::atomic::{AtomicBool, Ordering},
    };

    const MAX_EVENTS: usize = 64;
    const EPOLL_TIMEOUT_MS: i32 = 100;

    /// Per-connection state. `out`/`out_pos` form a write buffer that survives a
    /// partial write.
    #[derive(Default)]
    struct Connection {
        out: Vec<u8>,
        out_pos: usize,
    }

    impl Connection {
        fn has_pending_output(&self) -> bool {
            self.out_pos < self.out.len()
        }
    }

    /// Bind a non-blocking listener on `127.0.0.1:port`; port 0 asks the kernel for
    /// a free port, which is returned alongside the fd.
    pub fn bind_listener(port: u16) -> io::Result<(i32, u16)> {
        // SAFETY: all pointers are to stack values that outlive the call.
        unsafe {
            let fd = libc::socket(libc::AF_INET, libc::SOCK_STREAM | libc::SOCK_NONBLOCK, 0);
            if fd < 0 {
                return Err(io::Error::last_os_error());
            }

            let enable: libc::c_int = 1;
            if libc::setsockopt(
                fd,
                libc::SOL_SOCKET,
                libc::SO_REUSEADDR,
                &enable as *const _ as *const libc::c_void,
                std::mem::size_of::<libc::c_int>() as libc::socklen_t,
            ) < 0
            {
                let error = io::Error::last_os_error();
                libc::close(fd);
                return Err(error);
            }

            let mut address: libc::sockaddr_in = std::mem::zeroed();
            address.sin_family = libc::AF_INET as libc::sa_family_t;
            address.sin_port = port.to_be();
            address.sin_addr.s_addr = libc::INADDR_LOOPBACK.to_be();

            if libc::bind(
                fd,
                &address as *const _ as *const libc::sockaddr,
                std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t,
            ) < 0
            {
                let error = io::Error::last_os_error();
                libc::close(fd);
                return Err(error);
            }

            if libc::listen(fd, 128) < 0 {
                let error = io::Error::last_os_error();
                libc::close(fd);
                return Err(error);
            }

            let mut bound: libc::sockaddr_in = std::mem::zeroed();
            let mut length = std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t;
            if libc::getsockname(fd, &mut bound as *mut _ as *mut libc::sockaddr, &mut length) < 0 {
                let error = io::Error::last_os_error();
                libc::close(fd);
                return Err(error);
            }

            Ok((fd, u16::from_be(bound.sin_port)))
        }
    }

    /// Run the event loop until `shutdown` becomes true.
    pub fn event_loop(epfd: i32, listen_fd: i32, shutdown: &AtomicBool) -> io::Result<()> {
        epoll_add(epfd, listen_fd, libc::EPOLLIN as u32)?;

        let mut connections: HashMap<i32, Connection> = HashMap::new();
        let mut events = vec![libc::epoll_event { events: 0, u64: 0 }; MAX_EVENTS];

        while !shutdown.load(Ordering::Relaxed) {
            // SAFETY: `events` is a valid, writable buffer of MAX_EVENTS entries.
            let ready = unsafe {
                libc::epoll_wait(
                    epfd,
                    events.as_mut_ptr(),
                    events.len() as i32,
                    EPOLL_TIMEOUT_MS,
                )
            };

            if ready < 0 {
                let error = io::Error::last_os_error();
                if error.raw_os_error() == Some(libc::EINTR) {
                    continue;
                }
                return Err(error);
            }

            for event in &events[..ready as usize] {
                let fd = event.u64 as i32;
                if fd == listen_fd {
                    accept_ready(epfd, listen_fd, &mut connections)?;
                    continue;
                }

                let readable = event.events & (libc::EPOLLIN as u32) != 0;
                let writable = event.events & (libc::EPOLLOUT as u32) != 0;

                if readable && !read_into(fd, &mut connections)? {
                    continue; // connection closed
                }
                if writable && connections.contains_key(&fd) {
                    flush(fd, &mut connections)?;
                }
                if connections.contains_key(&fd) {
                    update_interest(epfd, fd, &mut connections)?;
                }
            }
        }

        Ok(())
    }

    /// Accept every pending connection, then return; `accept4` with
    /// `SOCK_NONBLOCK` means a client socket can never block the loop.
    fn accept_ready(
        epfd: i32,
        listen_fd: i32,
        connections: &mut HashMap<i32, Connection>,
    ) -> io::Result<()> {
        loop {
            // SAFETY: a null sockaddr is explicitly allowed for accept.
            let client = unsafe {
                libc::accept4(
                    listen_fd,
                    std::ptr::null_mut(),
                    std::ptr::null_mut(),
                    libc::SOCK_NONBLOCK,
                )
            };

            if client < 0 {
                if would_block() {
                    return Ok(());
                }
                return Err(io::Error::last_os_error());
            }

            epoll_add(epfd, client, libc::EPOLLIN as u32)?;
            connections.insert(client, Connection::default());
        }
    }

    /// Drain readable bytes into the connection's output buffer.
    ///
    /// Returns `false` when the peer closed the connection.
    fn read_into(fd: i32, connections: &mut HashMap<i32, Connection>) -> io::Result<bool> {
        let mut buffer = [0u8; 4096];
        loop {
            // SAFETY: `buffer` is a valid writable region of `buffer.len()` bytes.
            let read =
                unsafe { libc::read(fd, buffer.as_mut_ptr() as *mut libc::c_void, buffer.len()) };

            if read == 0 {
                close_connection(fd, connections);
                return Ok(false);
            }
            if read < 0 {
                if would_block() {
                    break;
                }
                close_connection(fd, connections);
                return Err(io::Error::last_os_error());
            }

            if let Some(connection) = connections.get_mut(&fd) {
                connection
                    .out
                    .extend_from_slice(&buffer[..read as usize]);
            }
        }

        flush(fd, connections)?;
        Ok(true)
    }

    /// Write as much of the buffered reply as the socket accepts.
    fn flush(fd: i32, connections: &mut HashMap<i32, Connection>) -> io::Result<()> {
        let Some(connection) = connections.get_mut(&fd) else {
            return Ok(());
        };

        while connection.has_pending_output() {
            let remaining = connection.out.len() - connection.out_pos;
            // SAFETY: the slice is valid for `remaining` bytes.
            let written = unsafe {
                libc::write(
                    fd,
                    connection.out[connection.out_pos..].as_ptr() as *const libc::c_void,
                    remaining,
                )
            };

            if written < 0 {
                if would_block() {
                    // Keep the rest buffered; EPOLLOUT will drive the next attempt.
                    return Ok(());
                }
                return Err(io::Error::last_os_error());
            }
            connection.out_pos += written as usize;
        }

        connection.out.clear();
        connection.out_pos = 0;
        Ok(())
    }

    /// Ask for `EPOLLOUT` only while a reply is still buffered.
    fn update_interest(
        epfd: i32,
        fd: i32,
        connections: &mut HashMap<i32, Connection>,
    ) -> io::Result<()> {
        let pending = connections
            .get(&fd)
            .is_some_and(Connection::has_pending_output);
        let events = if pending {
            (libc::EPOLLIN | libc::EPOLLOUT) as u32
        } else {
            libc::EPOLLIN as u32
        };
        epoll_modify(epfd, fd, events)
    }

    fn close_connection(fd: i32, connections: &mut HashMap<i32, Connection>) {
        connections.remove(&fd);
        // SAFETY: `fd` is owned by us and open at this point.
        unsafe {
            libc::close(fd);
        }
    }

    fn epoll_add(epfd: i32, fd: i32, events: u32) -> io::Result<()> {
        let mut event = libc::epoll_event {
            events,
            u64: fd as u64,
        };
        // SAFETY: `event` outlives the call.
        if unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, fd, &mut event) } < 0 {
            Err(io::Error::last_os_error())
        } else {
            Ok(())
        }
    }

    fn epoll_modify(epfd: i32, fd: i32, events: u32) -> io::Result<()> {
        let mut event = libc::epoll_event {
            events,
            u64: fd as u64,
        };
        // SAFETY: `event` outlives the call.
        if unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_MOD, fd, &mut event) } < 0 {
            Err(io::Error::last_os_error())
        } else {
            Ok(())
        }
    }

    fn would_block() -> bool {
        matches!(
            io::Error::last_os_error().raw_os_error(),
            Some(libc::EAGAIN)
        )
    }
}

#[cfg(target_os = "linux")]
fn main() {
    use std::sync::atomic::AtomicBool;

    let port: u16 = std::env::args()
        .nth(1)
        .and_then(|argument| argument.parse().ok())
        .unwrap_or(9000);

    let (listen_fd, bound_port) = linux::bind_listener(port).expect("failed to bind");
    // SAFETY: epoll_create1 takes flags and returns an fd or -1.
    let epfd = unsafe { libc::epoll_create1(0) };
    assert!(epfd >= 0, "epoll_create1 failed");

    // A write to a peer that vanished raises SIGPIPE, which by default kills the
    // process. Servers ignore it and handle EPIPE from write instead.
    // SAFETY: installing SIG_IGN is a single-threaded startup action.
    unsafe {
        libc::signal(libc::SIGPIPE, libc::SIG_IGN);
    }

    println!("epoll echo server listening on 127.0.0.1:{bound_port} (Ctrl-C to stop)");

    let shutdown = AtomicBool::new(false);
    let result = linux::event_loop(epfd, listen_fd, &shutdown);
    // SAFETY: both fds are owned by this process.
    unsafe {
        libc::close(epfd);
        libc::close(listen_fd);
    }

    if let Err(error) = result {
        eprintln!("event loop error: {error}");
        std::process::exit(1);
    }
}

#[cfg(not(target_os = "linux"))]
fn main() {
    eprintln!("epoll_echo requires Linux; use kqueue or IOCP on this platform.");
}

#[cfg(all(test, target_os = "linux"))]
mod tests {
    use std::{
        io::{Read, Write},
        net::TcpStream,
        sync::{
            Arc,
            atomic::{AtomicBool, Ordering},
        },
        thread,
    };

    use super::linux;

    #[test]
    fn echoes_a_message_through_the_event_loop() {
        let (listen_fd, port) = linux::bind_listener(0).expect("bind");
        // SAFETY: epoll_create1 takes flags and returns an fd or -1.
        let epfd = unsafe { libc::epoll_create1(0) };
        assert!(epfd >= 0, "epoll_create1 failed");

        let shutdown = Arc::new(AtomicBool::new(false));
        let server_shutdown = Arc::clone(&shutdown);
        let server = thread::spawn(move || {
            let result = linux::event_loop(epfd, listen_fd, &server_shutdown);
            // SAFETY: the test owns both fds.
            unsafe {
                libc::close(epfd);
                libc::close(listen_fd);
            }
            result
        });

        let mut client = TcpStream::connect(("127.0.0.1", port)).expect("connect");
        client
            .write_all(b"epoll echo")
            .expect("write");
        let mut echoed = [0u8; 10];
        client
            .read_exact(&mut echoed)
            .expect("read echo");
        assert_eq!(&echoed, b"epoll echo");

        shutdown.store(true, Ordering::Relaxed);
        drop(client);
        let result = server
            .join()
            .expect("server thread panicked");
        assert!(result.is_ok(), "event loop returned {result:?}");
    }
}

20.6 Questions that come up

“What does a read of 0 bytes mean on a TCP socket?” The peer closed its sending side (FIN). No more data will arrive. It is not an error.

“Why set TCP_NODELAY?” To stop Nagle’s algorithm from holding back small writes. Request-response protocols that write a small reply and wait want it off.

“Thread per connection or an event loop?” Threads for a modest number of connections and simple blocking code. An event loop, or an async runtime on top of one, for many mostly idle connections, where per-thread stacks and context switches dominate.

“Level-triggered or edge-triggered epoll?” Level-triggered reports a ready fd until it is no longer ready. Edge-triggered reports only transitions, so the handler must drain the fd until EAGAIN. Otherwise it may never hear about the remaining data.

“A 10 Gb/s link between regions delivers 50 MB/s. Why?” Check the bandwidth-delay product against the window: at 80 ms RTT, 10 Gb/s needs about 100 MB in flight. A small socket buffer, or window scaling disabled somewhere on the path, caps throughput at window / RTT.