32. Networking with std and Tokio: TCP, UDP, and framing

32.1 A threaded TCP echo server with std

use std::io::{BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream};
use std::time::Duration;

fn handle(stream: TcpStream) -> std::io::Result<()> {
    let peer = stream.peer_addr()?;
    stream.set_read_timeout(Some(Duration::from_secs(30)))?; // idle clients cannot hold threads forever
    stream.set_nodelay(true)?;                               // disable Nagle for small interactive writes
    let mut writer = stream.try_clone()?;
    for line in BufReader::new(stream).lines() {
        let line = line?;
        writeln!(writer, "echo: {line}")?;
    }
    println!("{peer} disconnected");
    Ok(())
}

fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:7878")?;  // socket() + bind() + listen()
    println!("listening on {}", listener.local_addr()?);
    for stream in listener.incoming().take(1) {           // take(1) so the demo exits; remove for a real server
        let stream = stream?;                              // accept()
        std::thread::spawn(move || {
            if let Err(e) = handle(stream) { eprintln!("connection error: {e}"); }
        });
    }
    Ok(())
}

cargo run --bin echo &       # then:
printf 'hello\nworld\n' | nc 127.0.0.1 7878

Every call above maps one-to-one onto the C guide's §21.1 echo server: bind performs socket/setsockopt(SO_REUSEADDR)/bind/listen, and incoming loops over accept. The differences: the socket is closed by Drop, errors are Results instead of -1 and errno, and the buffer cannot overflow.

Across the series: what bind, listen, and accept do on the wire is in Net Ch 19 and Ch 21. Why set_nodelay matters is Ch 26, the 40-millisecond mystery.

32.2 The same server with Tokio, plus length-prefixed framing

TCP is a byte stream, not a message stream. One write of 100 bytes can arrive as two reads of 60 and 40, and two writes can arrive in one read. Every protocol therefore needs framing: delimiters (\n, as in HTTP/1 headers and Redis) or a length prefix (as in HTTP/2, TLS records, gRPC, and Kafka).

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

const MAX_FRAME: u32 = 1 << 20; // 1 MiB: never trust a length field from the network

async fn read_frame(s: &mut TcpStream) -> std::io::Result<Option<Vec<u8>>> {
    let len = match s.read_u32().await {              // 4-byte big-endian length
        Ok(n) => n,
        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None), // clean close
        Err(e) => return Err(e),
    };
    if len > MAX_FRAME {
        return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, format!("frame of {len} bytes")));
    }
    let mut buf = vec![0u8; len as usize];
    s.read_exact(&mut buf).await?;                    // loops until all `len` bytes have arrived
    Ok(Some(buf))
}

async fn write_frame(s: &mut TcpStream, payload: &[u8]) -> std::io::Result<()> {
    let len = u32::try_from(payload.len()).map_err(|_| std::io::Error::other("payload too large"))?;
    let mut out = Vec::with_capacity(4 + payload.len());
    out.extend_from_slice(&len.to_be_bytes());
    out.extend_from_slice(payload);
    s.write_all(&out).await                           // one write: cancel-safe at the frame level
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;

    tokio::spawn(async move {
        loop {
            let (mut sock, _) = listener.accept().await.unwrap();
            tokio::spawn(async move {                 // one TASK per connection, not one thread
                while let Ok(Some(frame)) = read_frame(&mut sock).await {
                    let reply = frame.to_ascii_uppercase();
                    if write_frame(&mut sock, &reply).await.is_err() { break; }
                }
            });
        }
    });

    let mut client = TcpStream::connect(addr).await?;
    for msg in ["hello", "framed", "world"] {
        write_frame(&mut client, msg.as_bytes()).await?;
        let reply = read_frame(&mut client).await?.expect("server replied");
        println!("{}", String::from_utf8_lossy(&reply));
    }
    Ok(())
}

MAX_FRAME is a security control. Without it, a client sending the four bytes FF FF FF FF makes the server allocate 4 GiB. Ch 53 treats every length field from the network this way.

32.3 UDP

use std::net::UdpSocket;
use std::time::{Duration, Instant};

fn main() -> std::io::Result<()> {
    let server = UdpSocket::bind("127.0.0.1:0")?;
    let addr = server.local_addr()?;
    std::thread::spawn(move || {
        let mut buf = [0u8; 1500];                         // one datagram, up to a typical MTU
        while let Ok((n, peer)) = server.recv_from(&mut buf) {
            let _ = server.send_to(&buf[..n], peer);       // echo
        }
    });

    let client = UdpSocket::bind("127.0.0.1:0")?;
    client.set_read_timeout(Some(Duration::from_millis(500)))?; // UDP never tells you about loss
    let mut lost = 0;
    for seq in 0u32..5 {
        let t = Instant::now();
        client.send_to(&seq.to_be_bytes(), addr)?;
        let mut buf = [0u8; 4];
        match client.recv_from(&mut buf) {
            Ok(_) if u32::from_be_bytes(buf) == seq => println!("seq={seq} rtt={:?}", t.elapsed()),
            _ => lost += 1,
        }
    }
    println!("lost {lost}/5");
    Ok(())
}

This is a small version of the networking guide's UDP loss meter lab (Net Ch 20).