Nagle's Algorithm Ruined My Afternoon: 40 Milliseconds from Nowhere

wwr/src/server.rs — 67 lines

//! Request/response server for the write-write-read demo.
//!
//! Protocol: 4-byte big-endian body length, then the body; the reply is the
//! 8-byte big-endian per-connection request counter. The server never sets
//! any socket options — every stall in this demo is produced (and later
//! cured) purely on the client side.

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

use anyhow::{Context, Result, ensure};

/// Largest body the server agrees to read, so a corrupt header cannot ask
/// for an absurd allocation.
const MAX_BODY: usize = 65536;

fn main() -> Result<()> {
    // Port 0 = let the kernel pick a free high port; print it for clients.
    let listener = TcpListener::bind("127.0.0.1:0").context("bind 127.0.0.1:0")?;
    let addr = listener.local_addr().context("query local addr")?;
    println!("listening on {addr}");

    for stream in listener.incoming() {
        let stream = stream.context("accept")?;
        let peer = stream.peer_addr().context("query peer addr")?;
        match serve(stream) {
            Ok((requests, bytes)) => {
                println!("conn {peer}: {requests} requests, {bytes} body bytes");
            }
            Err(error) => println!("conn {peer}: error: {error:#}"),
        }
    }
    Ok(())
}

/// Serve one connection until the client closes it; count what went through.
fn serve(mut stream: TcpStream) -> Result<(u64, u64)> {
    let mut requests = 0u64;
    let mut bytes = 0u64;
    let mut body = vec![0u8; MAX_BODY];

    loop {
        let mut header = [0u8; 4];
        // A clean EOF between requests is the normal end of a session.
        match stream.read_exact(&mut header) {
            Ok(()) => {}
            Err(error) if error.kind() == ErrorKind::UnexpectedEof => {
                return Ok((requests, bytes));
            }
            Err(error) => return Err(error).context("read header"),
        }

        let len = u32::from_be_bytes(header) as usize;
        ensure!(len <= MAX_BODY, "body length {len} exceeds {MAX_BODY}");

        // The reply is written only after the WHOLE request arrived — this
        // is what keeps the server silent while delayed ACK counts to 40 ms.
        stream.read_exact(&mut body[..len]).context("read body")?;
        requests += 1;
        bytes += len as u64;
        stream
            .write_all(&requests.to_be_bytes())
            .context("write reply")?;
    }
}

wwr/src/client.rs — 64 lines

//! Naive client: header and body leave in two separate write calls, then the
//! client blocks on the reply. No socket options touched — this is the shape
//! of code that walks straight into the Nagle + delayed ACK trap.

use std::{
    io::{Read, Write},
    net::TcpStream,
    time::{Duration, Instant},
};

use anyhow::{Context, Result, ensure};
use clap::Parser;

mod stats;

use stats::Summary;

/// Write-write-read client: header write, body write, blocking read.
#[derive(Parser)]
struct Args {
    /// Server port on 127.0.0.1 (printed by wwr-server on start).
    #[arg(long)]
    port: u16,
    /// Number of request/response round trips.
    #[arg(long, default_value_t = 1000)]
    iters: u64,
    /// Body size in bytes (the header is always 4 bytes).
    #[arg(long, default_value_t = 16)]
    body: u32,
}

fn main() -> Result<()> {
    let args = Args::parse();
    let mut stream = TcpStream::connect(("127.0.0.1", args.port))
        .with_context(|| format!("connect to 127.0.0.1:{}", args.port))?;
    stream
        .set_read_timeout(Some(Duration::from_secs(10)))
        .context("set read timeout")?;

    let header = args.body.to_be_bytes();
    let body = vec![0x42u8; args.body as usize];
    let mut reply = [0u8; 8];
    let mut samples = Vec::with_capacity(args.iters as usize);

    let begin = Instant::now();
    for expected in 1..=args.iters {
        let start = Instant::now();
        stream.write_all(&header).context("write header")?; // write #1
        stream.write_all(&body).context("write body")?; // write #2
        stream.read_exact(&mut reply).context("read reply")?;
        samples.push(start.elapsed());

        let counter = u64::from_be_bytes(reply);
        ensure!(counter == expected, "reply counter {counter} != {expected}");
    }
    let total = begin.elapsed();

    println!(
        "naive write-write-read: {} iterations, header 4 B + body {} B",
        args.iters, args.body
    );
    println!("{}", Summary::from_samples(&mut samples, total)?);
    Ok(())
}

wwr/src/stats.rs — 75 lines

//! Latency summary for one client session: percentiles plus wall clock.

use std::{fmt, time::Duration};

use anyhow::{Result, ensure};

/// Percentile snapshot of one run.
pub struct Summary {
    min: Duration,
    p50: Duration,
    p90: Duration,
    p99: Duration,
    max: Duration,
    total: Duration,
    count: usize,
}

impl Summary {
    /// Sort the samples in place and pick the percentile landmarks.
    pub fn from_samples(samples: &mut [Duration], total: Duration) -> Result<Self> {
        ensure!(!samples.is_empty(), "no samples recorded");
        samples.sort_unstable();
        let last = samples.len() - 1;
        let at = |q: f64| samples[(last as f64 * q).round() as usize];
        Ok(Self {
            min: samples[0],
            p50: at(0.50),
            p90: at(0.90),
            p99: at(0.99),
            max: samples[last],
            total,
            count: samples.len(),
        })
    }
}

impl fmt::Display for Summary {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        let rate = self.count as f64 / self.total.as_secs_f64();
        writeln!(
            formatter,
            "min {} | p50 {} | p90 {} | p99 {} | max {}",
            pretty(self.min),
            pretty(self.p50),
            pretty(self.p90),
            pretty(self.p99),
            pretty(self.max),
        )?;
        write!(
            formatter,
            "total {}, {rate:.0} req/s",
            pretty_total(self.total)
        )
    }
}

/// Adaptive unit: microseconds for the fast path, milliseconds for stalls.
fn pretty(duration: Duration) -> String {
    let micros = duration.as_secs_f64() * 1e6;
    if micros >= 1000.0 {
        format!("{:.2} ms", micros / 1000.0)
    } else {
        format!("{micros:.0} \u{b5}s")
    }
}

/// Seconds once the total crosses one second, milliseconds below.
fn pretty_total(duration: Duration) -> String {
    let seconds = duration.as_secs_f64();
    if seconds >= 1.0 {
        format!("{seconds:.1} s")
    } else {
        format!("{:.0} ms", seconds * 1e3)
    }
}

wwr-nodelay/src/client.rs — 66 lines

//! Same write-write-read client as stage 01, one line apart: TCP_NODELAY
//! disables Nagle, so the second write leaves the machine immediately
//! instead of waiting for the ACK of the first.

use std::{
    io::{Read, Write},
    net::TcpStream,
    time::{Duration, Instant},
};

use anyhow::{Context, Result, ensure};
use clap::Parser;

mod stats;

use stats::Summary;

/// Write-write-read client with Nagle's algorithm switched off.
#[derive(Parser)]
struct Args {
    /// Server port on 127.0.0.1 (printed by wwr-server on start).
    #[arg(long)]
    port: u16,
    /// Number of request/response round trips.
    #[arg(long, default_value_t = 5000)]
    iters: u64,
    /// Body size in bytes (the header is always 4 bytes).
    #[arg(long, default_value_t = 16)]
    body: u32,
}

fn main() -> Result<()> {
    let args = Args::parse();
    let mut stream = TcpStream::connect(("127.0.0.1", args.port))
        .with_context(|| format!("connect to 127.0.0.1:{}", args.port))?;
    stream
        .set_read_timeout(Some(Duration::from_secs(10)))
        .context("set read timeout")?;
    // The one-line cure: no more waiting for ACKs before sending small data.
    stream.set_nodelay(true).context("set TCP_NODELAY")?;

    let header = args.body.to_be_bytes();
    let body = vec![0x42u8; args.body as usize];
    let mut reply = [0u8; 8];
    let mut samples = Vec::with_capacity(args.iters as usize);

    let begin = Instant::now();
    for expected in 1..=args.iters {
        let start = Instant::now();
        stream.write_all(&header).context("write header")?; // write #1
        stream.write_all(&body).context("write body")?; // write #2
        stream.read_exact(&mut reply).context("read reply")?;
        samples.push(start.elapsed());

        let counter = u64::from_be_bytes(reply);
        ensure!(counter == expected, "reply counter {counter} != {expected}");
    }
    let total = begin.elapsed();

    println!(
        "TCP_NODELAY write-write-read: {} iterations, header 4 B + body {} B",
        args.iters, args.body
    );
    println!("{}", Summary::from_samples(&mut samples, total)?);
    Ok(())
}

wwr-onewrite/src/client.rs — 84 lines

//! The other cure: keep Nagle ON, but hand the kernel the whole request in
//! one syscall — either a single concatenated buffer, or writev with two
//! iovecs. Nagle never sees a lonely unfinished segment, so nothing stalls.

use std::{
    io::{IoSlice, Read, Write},
    net::TcpStream,
    time::{Duration, Instant},
};

use anyhow::{Context, Result, ensure};
use clap::Parser;

mod stats;

use stats::Summary;

/// One-syscall client: whole request per write, Nagle left enabled.
#[derive(Parser)]
struct Args {
    /// Server port on 127.0.0.1 (printed by wwr-server on start).
    #[arg(long)]
    port: u16,
    /// Number of request/response round trips.
    #[arg(long, default_value_t = 5000)]
    iters: u64,
    /// Body size in bytes (the header is always 4 bytes).
    #[arg(long, default_value_t = 16)]
    body: u32,
    /// Use write_vectored (writev) with two iovecs instead of one buffer.
    #[arg(long)]
    vectored: bool,
}

fn main() -> Result<()> {
    let args = Args::parse();
    let mut stream = TcpStream::connect(("127.0.0.1", args.port))
        .with_context(|| format!("connect to 127.0.0.1:{}", args.port))?;
    stream
        .set_read_timeout(Some(Duration::from_secs(10)))
        .context("set read timeout")?;

    let header = args.body.to_be_bytes();
    let body = vec![0x42u8; args.body as usize];

    // The concatenated form: header and body glued once, outside the loop.
    let mut request = Vec::with_capacity(header.len() + body.len());
    request.extend_from_slice(&header);
    request.extend_from_slice(&body);

    let mut reply = [0u8; 8];
    let mut samples = Vec::with_capacity(args.iters as usize);

    let begin = Instant::now();
    for expected in 1..=args.iters {
        let start = Instant::now();
        if args.vectored {
            // Two logical buffers, one syscall, one segment on the wire.
            let slices = [IoSlice::new(&header), IoSlice::new(&body)];
            let sent = stream.write_vectored(&slices).context("writev request")?;
            ensure!(sent == request.len(), "short vectored write: {sent}");
        } else {
            stream.write_all(&request).context("write request")?;
        }
        stream.read_exact(&mut reply).context("read reply")?;
        samples.push(start.elapsed());

        let counter = u64::from_be_bytes(reply);
        ensure!(counter == expected, "reply counter {counter} != {expected}");
    }
    let total = begin.elapsed();

    let how = if args.vectored {
        "writev, two iovecs"
    } else {
        "one buffer"
    };
    println!(
        "single write ({how}), Nagle still on: {} iterations, 4+{} B",
        args.iters, args.body
    );
    println!("{}", Summary::from_samples(&mut samples, total)?);
    Ok(())
}

flood/src/server.rs — 43 lines

//! Sink server for the tiny-write flood: drain everything until EOF, then
//! confirm the byte count back to the client so it knows every segment has
//! been sent and delivered before it reads its own TCP_INFO counters.

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

use anyhow::{Context, Result};

fn main() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").context("bind 127.0.0.1:0")?;
    let addr = listener.local_addr().context("query local addr")?;
    println!("listening on {addr}");

    for stream in listener.incoming() {
        let stream = stream.context("accept")?;
        let peer = stream.peer_addr().context("query peer addr")?;
        match drain(stream) {
            Ok(total) => println!("conn {peer}: drained {total} bytes"),
            Err(error) => println!("conn {peer}: error: {error:#}"),
        }
    }
    Ok(())
}

/// Read until the client half-closes, then echo the total byte count back.
fn drain(mut stream: TcpStream) -> Result<u64> {
    let mut buf = vec![0u8; 65536];
    let mut total = 0u64;
    loop {
        let count = stream.read(&mut buf).context("read")?;
        if count == 0 {
            break;
        }
        total += count as u64;
    }
    stream
        .write_all(&total.to_be_bytes())
        .context("write confirmation")?;
    Ok(total)
}

flood/src/client.rs — 88 lines

//! The flood where Nagle is the hero: a stream of one-byte writes with no
//! read on the critical path. With Nagle the kernel coalesces them into a
//! handful of segments; with TCP_NODELAY every byte pays for its own packet.

use std::{
    io::{Read, Write},
    net::{Shutdown, TcpStream},
    time::{Duration, Instant},
};

use anyhow::{Context, Result, ensure};
use clap::Parser;

mod tcpinfo;

use tcpinfo::segment_counts;

/// Tiny-write flood client with per-socket segment accounting.
#[derive(Parser)]
struct Args {
    /// Server port on 127.0.0.1 (printed by flood-server on start).
    #[arg(long)]
    port: u16,
    /// Number of write calls to issue back to back.
    #[arg(long, default_value_t = 200_000)]
    writes: u64,
    /// Bytes per write call.
    #[arg(long, default_value_t = 1)]
    size: u32,
    /// Disable Nagle's algorithm for the flood.
    #[arg(long)]
    nodelay: bool,
}

fn main() -> Result<()> {
    let args = Args::parse();
    let mut stream = TcpStream::connect(("127.0.0.1", args.port))
        .with_context(|| format!("connect to 127.0.0.1:{}", args.port))?;
    stream
        .set_read_timeout(Some(Duration::from_secs(10)))
        .context("set read timeout")?;
    stream
        .set_nodelay(args.nodelay)
        .context("set TCP_NODELAY")?;

    let chunk = vec![0x2au8; args.size as usize];
    let sent = args.writes * u64::from(args.size);

    let begin = Instant::now();
    for _ in 0..args.writes {
        stream.write_all(&chunk).context("write chunk")?;
    }
    let write_loop = begin.elapsed();

    // Half-close, then wait for the server's byte-count confirmation: once
    // it arrives, every data segment has been sent, delivered and counted.
    stream.shutdown(Shutdown::Write).context("shutdown write")?;
    let mut reply = [0u8; 8];
    stream.read_exact(&mut reply).context("read confirmation")?;
    let total = begin.elapsed();

    let drained = u64::from_be_bytes(reply);
    ensure!(drained == sent, "server drained {drained} of {sent} bytes");

    let counts = segment_counts(&stream)?;
    ensure!(counts.data_segs_out > 0, "no data segments counted");
    let per_segment = sent as f64 / f64::from(counts.data_segs_out);

    let mode = if args.nodelay {
        "TCP_NODELAY"
    } else {
        "Nagle on"
    };
    println!(
        "{mode}: {} writes x {} B = {sent} bytes",
        args.writes, args.size
    );
    println!(
        "write loop {:.0} ms, all bytes at the server after {:.0} ms",
        write_loop.as_secs_f64() * 1e3,
        total.as_secs_f64() * 1e3,
    );
    println!(
        "data segments out: {} ({:.0} payload bytes per segment); segs_out {}",
        counts.data_segs_out, per_segment, counts.segs_out
    );
    Ok(())
}

flood/src/tcpinfo.rs — 42 lines

//! TCP_INFO peek: how many segments this socket actually put on the wire.

use std::{io, mem, net::TcpStream, os::fd::AsRawFd};

use anyhow::{Result, ensure};
use libc::{IPPROTO_TCP, TCP_INFO, c_void, getsockopt, socklen_t, tcp_info};

/// Outgoing segment counters of one socket, as the kernel counted them.
pub struct SegmentCounts {
    /// Every segment sent, including handshake, pure ACKs and FIN.
    pub segs_out: u32,
    /// Only segments that carried payload bytes.
    pub data_segs_out: u32,
}

/// Read the socket's `struct tcp_info` via getsockopt and pick the counters.
pub fn segment_counts(stream: &TcpStream) -> Result<SegmentCounts> {
    let mut info: tcp_info = unsafe { mem::zeroed() };
    let mut len = mem::size_of::<tcp_info>() as socklen_t;
    let rc = unsafe {
        getsockopt(
            stream.as_raw_fd(),
            IPPROTO_TCP,
            TCP_INFO,
            (&raw mut info).cast::<c_void>(),
            &mut len,
        )
    };
    ensure!(
        rc == 0,
        "getsockopt(TCP_INFO): {}",
        io::Error::last_os_error()
    );
    ensure!(
        len as usize <= mem::size_of::<tcp_info>(),
        "kernel returned an oversized tcp_info: {len} bytes"
    );
    Ok(SegmentCounts {
        segs_out: info.tcpi_segs_out,
        data_segs_out: info.tcpi_data_segs_out,
    })
}