Nuclei

Chapter 6 of 17

Networking

TCP, UDP and Unix sockets, addresses and names, socket options, halves and pipes.

TCP

nuclei::net has the socket types: TcpListener, TcpStream, UdpSocket and, on Unix, UnixListener, UnixStream and UnixDatagram. Each is std's own socket in a Handle: the handle adds the async calls, and std's methods (local_addr, peer_addr, set_ttl, ...) work through it.

src/main.rs
use nuclei::io::{AsyncReadExt, AsyncWriteExt};
use nuclei::net::{TcpListener, TcpStream};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // Port 0: the system picks a free one.
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;

    // The server echoes every byte back until the client ends its side.
    let server = nuclei::spawn(async move {
        let (stream, peer) = listener.accept().await?;
        println!("connection from {}", peer.ip());
        let mut buf = [0u8; 1024];
        loop {
            let n = stream.recv(&mut buf).await?;
            if n == 0 {
                break;
            }
            stream.send(&buf[..n]).await?;
        }
        std::io::Result::Ok(())
    });

    // The client uses the same stream through the read and write helpers.
    let mut client = TcpStream::connect(addr).await?;
    client.write_all(b"hello").await?;
    client.shutdown().await?;
    let mut reply = String::new();
    client.read_to_string(&mut reply).await?;
    assert_eq!(reply, "hello");

    server.await?;
    Ok(())
}
  • bind and connect are async, since a name may need looking up. Each takes the first address that works.
  • recv and send work on a shared reference; the stream is also AsyncRead and AsyncWrite, on the handle and on & the handle, so one task can read while another writes.
  • incoming() gives the connections as a stream. accept_multi() gives a stream that owns a copy of the listener, so it may outlive the handle; on io_uring one accept stays armed for every connection that comes.
  • read_owned and write_owned lend the driver a buffer of yours, and read_vectored_owned and write_vectored_owned lend several in one call (IO utilities).

Addresses and names

Every call that takes an address takes anything ToSocketAddrs accepts: a SocketAddr, an (ip, port) pair, a "host:port" string, a (host, port) pair, or a slice of addresses. Numbers are used at once; a name is looked up on the blocking pool, so no shard waits on the resolver. net::lookup_host looks one up alone.

src/main.rs
use std::net::{Ipv4Addr, SocketAddr};

use nuclei::net::{self, TcpListener};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // A name is looked up on the blocking pool, so no shard waits on the resolver.
    let found: Vec<SocketAddr> = net::lookup_host("localhost:8080").await?.collect();
    assert!(found.iter().all(|addr| addr.port() == 8080));

    // Every call that takes an address takes any of these forms.
    let a = TcpListener::bind("127.0.0.1:0").await?;
    let b = TcpListener::bind(("127.0.0.1", 0)).await?;
    let c = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
    let d = TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0))).await?;
    for listener in [a, b, c, d] {
        assert!(listener.local_addr()?.ip().is_loopback());
    }
    Ok(())
}

UDP

A bound socket sends datagrams to any address and receives them with their sender's; a connected one talks to one peer. A datagram larger than the buffer is cut to fit.

src/main.rs
use nuclei::net::UdpSocket;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let server = UdpSocket::bind("127.0.0.1:0").await?;
    let server_addr = server.local_addr()?;

    // Datagrams to an address, and received with their sender's address.
    let client = UdpSocket::bind("127.0.0.1:0").await?;
    client.send_to(b"ping", server_addr).await?;
    let mut buf = [0u8; 64];
    let (n, from) = server.recv_from(&mut buf).await?;
    assert_eq!((&buf[..n], from), (&b"ping"[..], client.local_addr()?));

    // A connected socket sends to, and receives from, its one peer.
    let peer = UdpSocket::connect(server_addr).await?;
    peer.send(b"hi").await?;
    let (_, from) = server.recv_from(&mut buf).await?;
    server.send_to(b"hello", from).await?;
    let n = peer.recv(&mut buf).await?;
    assert_eq!(&buf[..n], b"hello");
    Ok(())
}

peek and peek_from look at the next datagram without taking it, peek_sender names its sender without copying a byte, and recv_buf and recv_buf_from receive into a ReadBuf that need not be initialized.

Unix sockets

Streams and datagrams over a path in the file system, or a connected pair. A stream knows its peer: peer_cred gives the user and group, and the process where the system tells it (Linux, Android, macOS, iOS).

src/main.rs
use nuclei::io::{AsyncReadExt, AsyncWriteExt};
use nuclei::net::{UnixDatagram, UnixListener, UnixStream};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let dir = std::env::temp_dir().join(format!("learn-unix-{}", std::process::id()));
    std::fs::create_dir_all(&dir)?;
    let path = dir.join("server.sock");

    // Streams over a path in the file system.
    let listener = UnixListener::bind(&path)?;
    let mut client = UnixStream::connect(&path).await?;
    let (mut server, _) = listener.accept().await?;
    client.write_all(b"over a path").await?;
    let mut buf = [0u8; 11];
    server.read_exact(&mut buf).await?;
    assert_eq!(&buf, b"over a path");

    // Who is at the other end: user, group and, where the system says, process.
    let peer = server.peer_cred()?;
    assert_eq!(peer.pid(), Some(std::process::id() as i32));

    // A connected pair, and datagrams from a socket bound to no address.
    let (a, b) = UnixStream::pair()?;
    a.send(b"paired").await?;
    assert_eq!(b.recv(&mut buf).await?, 6);
    let inbox = UnixDatagram::bind(dir.join("inbox.sock"))?;
    let sender = UnixDatagram::unbound()?;
    sender.send_to(b"note", dir.join("inbox.sock")).await?;
    assert_eq!(inbox.recv(&mut buf).await?, 4);

    std::fs::remove_dir_all(&dir)
}

A connect to a listener whose backlog is full waits until the listener takes it, as std's blocking connect does. Linux answers such a connect at once, so the connect tries again on its own socket after a pause on a timer (1 ms, doubled up to 64 ms). No thread waits for it, the options set on the socket stay, and dropping the future frees everything.

On Linux a socket may also take a name in the abstract namespace, with no file behind it: bind_addr and connect_addr take std's SocketAddr, made with from_abstract_name.

src/main.rs
use std::os::linux::net::SocketAddrExt;
use std::os::unix::net::SocketAddr;

use nuclei::net::{UnixListener, UnixStream};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // Linux: a name in the abstract namespace, with no file behind it.
    let name = format!("learn-{}", std::process::id());
    let addr = SocketAddr::from_abstract_name(name.as_bytes())?;
    let listener = UnixListener::bind_addr(&addr)?;
    let client = UnixStream::connect_addr(&addr).await?;
    let (server, _) = listener.accept().await?;
    client.send(b"no file").await?;
    let mut buf = [0u8; 7];
    assert_eq!(server.recv(&mut buf).await?, 7);
    Ok(())
}

Socket options

A socket to set up before it listens or connects is a TcpSocket: buffer sizes, address and port reuse, keepalive, TCP_NODELAY, linger, the type of service or traffic class, IPV6_V6ONLY, and on Linux the network interface (bind_device). Then listen or connect turns it into a listener or a stream. A stream that is connected already sets linger and no-delay on the handle. The TCP options below go further.

src/main.rs
use std::time::Duration;

use nuclei::net::TcpSocket;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // Set a socket up before it listens...
    let socket = TcpSocket::new_v4()?;
    socket.set_reuseaddr(true)?;
    socket.set_recv_buffer_size(256 * 1024)?;
    socket.bind("127.0.0.1:0".parse().unwrap())?;
    let listener = socket.listen(1024)?;
    let addr = listener.local_addr()?;

    // ...or before it connects.
    let socket = TcpSocket::new_v4()?;
    socket.set_nodelay(true)?;
    socket.set_keepalive(true)?;
    let stream = socket.connect(addr).await?;
    assert!(stream.nodelay()?);

    // Options of a stream that is connected already: linger on close.
    let (accepted, _) = listener.accept().await?;
    accepted.set_linger(Some(Duration::from_secs(1)))?;
    assert_eq!(accepted.linger()?, Some(Duration::from_secs(1)));
    Ok(())
}

UnixSocket is the same for Unix sockets: new_stream or new_datagram, bind, then listen, connect or datagram.

TCP options

A TcpKeepalive times the keepalive probes: time idle before the first, interval between them, and retries unanswered before the connection drops. A field left None keeps the system's value; times are whole seconds. set_keepalive_timing turns the probes on with that timing, on a TcpSocket or on a connected stream, and keepalive_timing reads it back.

src/main.rs
use std::time::Duration;

use nuclei::io::AsyncWriteExt;
use nuclei::net::{TcpKeepalive, TcpSocket};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // A listener's options, set before it listens; incoming_cpu and defer_accept are Linux's.
    let socket = TcpSocket::new_v4()?;
    socket.set_reuseport(true)?;
    socket.set_incoming_cpu(0)?;
    socket.set_defer_accept(Some(Duration::from_secs(1)))?;
    socket.bind("127.0.0.1:0".parse().unwrap())?;
    let listener = socket.listen(128)?;

    // Probes after a minute idle, then every 10 seconds; 5 unanswered drop the connection.
    let socket = TcpSocket::new_v4()?;
    socket.set_keepalive_timing(TcpKeepalive {
        time: Some(Duration::from_secs(60)),
        interval: Some(Duration::from_secs(10)),
        retries: Some(5),
    })?;
    // Sent bytes may go unacknowledged for 30 seconds before the connection drops.
    socket.set_user_timeout(Some(Duration::from_secs(30)))?;
    let mut client = socket.connect(listener.local_addr()?).await?;
    assert_eq!(client.keepalive_timing()?.retries, Some(5));
    assert_eq!(client.user_timeout()?, Some(Duration::from_secs(30)));

    // With defer_accept, the connection is handed to accept once its first bytes came.
    client.write_all(b"hello").await?;
    client.flush().await?;
    let (server, _) = listener.accept().await?;
    server.set_quickack(true)?;
    Ok(())
}
OptionWhat it sets
set_keepalive_timingTurns keepalive probes on (SO_KEEPALIVE), timed as keepalive says (TCP_KEEPIDLE, TCP_KEEPINTVL, TCP_KEEPCNT; macOS calls the first TCP_KEEPALIVE).
set_user_timeoutSets TCP_USER_TIMEOUT (Linux): how long sent bytes may go unacknowledged before the connection is dropped, in whole milliseconds; None or zero leaves it to the system.
set_incoming_cpuSets SO_INCOMING_CPU (Linux): the CPU whose connections this listener takes among listeners sharing a port with SO_REUSEPORT, so each core accepts the connections its own queues received.
set_fastopenSets TCP_FASTOPEN on a socket that will listen: connections may bring data with their handshake. Linux keeps up to queue such handshakes waiting; macOS, FreeBSD and Windows take it as on (above zero) or off.
set_defer_acceptSets TCP_DEFER_ACCEPT (Linux) on a socket that will listen: a connection is handed to accept only once its first data came, or wait passed (whole seconds, which the kernel rounds up to its retransmission steps); None hands each over at once.
set_only_v6Sets IPV6_V6ONLY on an IPv6 socket: with true it takes IPv6 alone, else IPv4 too, through addresses mapped into IPv6.
set_quickackSets TCP_QUICKACK (Linux): with true the stream acknowledges at once rather than waiting to send it with data. The kernel turns it off again as it sees fit, so it is set again where it matters.

A connected TCP stream also has set_keepalive_timing and set_user_timeout. Each option has its getter (user_timeout, incoming_cpu, fastopen, ...). An option the system does not have answers Unsupported, so code for several systems can try it and go on.

A UDP socket set up first

Some options only a socket that is not bound yet takes. UdpSocketBuilder is that socket: new_v4 or new_v6, then address and port reuse, buffer sizes, the type of service (IP_TOS) or the traffic class (IPV6_TCLASS), set_only_v6, and on Linux the network interface (bind_device). bind turns it into a UdpSocket.

src/main.rs
use nuclei::net::{UdpSocket, UdpSocketBuilder};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // Options a UDP socket takes before it binds.
    let builder = UdpSocketBuilder::new_v4()?;
    builder.set_reuseport(true)?;
    builder.set_recv_buffer_size(256 * 1024)?;
    builder.set_tos(0x10)?;
    assert!(builder.recv_buffer_size()? >= 256 * 1024);
    let server = builder.bind("127.0.0.1:0".parse().unwrap())?;
    let addr = server.local_addr()?;

    // SO_REUSEPORT on both lets a second socket bind the same port.
    let second = UdpSocketBuilder::new_v4()?;
    second.set_reuseport(true)?;
    let second = second.bind(addr)?;
    assert_eq!(second.local_addr()?, addr);
    drop(second);

    // Bound, it is a UdpSocket like any other.
    let client = UdpSocket::bind("127.0.0.1:0").await?;
    client.send_to(b"ping", addr).await?;
    let mut buf = [0u8; 16];
    let (n, from) = server.recv_from(&mut buf).await?;
    assert_eq!((&buf[..n], from), (&b"ping"[..], client.local_addr()?));
    Ok(())
}

Two halves

A TCP or Unix stream splits into a read half and a write half: split() borrows the stream and allocates nothing; into_split() gives owned halves that move into tasks of their own, and reunite puts them back together.

src/main.rs
use nuclei::io::{AsyncReadExt, AsyncWriteExt};
use nuclei::net::{TcpListener, TcpStream};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let mut client = TcpStream::connect(listener.local_addr()?).await?;
    let (stream, _) = listener.accept().await?;

    // Owned halves: the read half and the write half go to tasks of their own.
    let (mut read, mut write) = stream.into_split();
    let reader = nuclei::spawn(async move {
        let mut got = [0u8; 4];
        read.read_exact(&mut got).await?;
        std::io::Result::Ok((read, got))
    });
    let writer = nuclei::spawn(async move {
        write.write_all(b"pong").await?;
        write.flush().await?;
        std::io::Result::Ok(write)
    });

    client.write_all(b"ping").await?;
    let mut reply = [0u8; 4];
    client.read_exact(&mut reply).await?;
    assert_eq!(&reply, b"pong");

    // The two halves go back together into the stream.
    let (read, got) = reader.await?;
    assert_eq!(&got, b"ping");
    let stream = read.reunite(writer.await?).expect("halves of one stream");
    assert_eq!(stream.peer_addr()?, client.local_addr()?);
    Ok(())
}

Ending a stream

A stream ends in one of three ways, each telling the peer something different:

  • shutdown().await, from AsyncWriteExt, flushes and ends the write side: the peer reads every byte, then the end.
  • shut(how), on a TCP or Unix stream, is std's shutdown made once the bytes written behind reached the kernel. Shutdown::Write and Both wait for those bytes, so the peer reads them all before the end; Read leaves the writes alone. Where the calling thread's shard runs io_uring, the reading half is shut through the ring.
  • abort() closes a TCP stream now: SO_LINGER goes to zero, the bytes written behind are dropped, and the peer's reads fail with ConnectionReset.
src/main.rs
use std::io::ErrorKind;
use std::net::Shutdown;

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

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

    // shut waits until the kernel has the bytes written behind, then ends the write side.
    let mut client = TcpStream::connect(addr).await?;
    let (mut server, _) = listener.accept().await?;
    client.write_all(b"every byte").await?;
    client.shut(Shutdown::Write).await?;
    let mut got = Vec::new();
    server.read_to_end(&mut got).await?;
    assert_eq!(got, b"every byte");

    // abort drops what is behind and resets the connection: the peer reads an error.
    let client = TcpStream::connect(addr).await?;
    let (server, _) = listener.accept().await?;
    client.abort()?;
    let mut buf = [0u8; 16];
    let reset = server.recv(&mut buf).await.unwrap_err();
    assert_eq!(reset.kind(), ErrorKind::ConnectionReset);
    Ok(())
}

After shut(Read) or shut(Both), reads answer as std's do on each system: Linux hands out what came before, then the end; macOS and Windows end at once, so there the handle drops what it read ahead too. The call is not named shutdown, so stream.shutdown(how) still reaches std's own through the handle, which shuts the halves at once: bytes still behind then never reach the peer.

A stream that is only dropped still delivers its bytes behind and ends cleanly. One whose time for that runs out resets the connection, as abort does (How it works).

Files and large writes

send_file(&file, offset, len) sends a range of a file to a TCP or Unix stream. Where the system has sendfile(2) or TransmitFile, the bytes go from the page cache to the socket inside the kernel; elsewhere they are read and written, which works the same. Each call runs on the blocking pool and sends at most send_file_chunk bytes (1 MiB by default), so a cold page holds a pool thread for one chunk at most.

write_zero_copy writes a lent buffer as write_owned does, and asks the kernel not to copy it either. The write ends once the kernel took the last byte; the buffer comes back through a Released once the peer acknowledged every byte. Only io_uring sends zero-copy, on the runtime's workers, and only while the write has at least zero_copy_threshold bytes left (1 MiB by default). Below it, and on the other backends, the kernel copies and the Released is ready at once.

src/main.rs
use std::net::Shutdown;

use nuclei::fs::File;
use nuclei::io::AsyncReadExt;
use nuclei::net::{TcpListener, TcpStream};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    nuclei::fs::write("page.html", vec![b'x'; 64 * 1024]).await?;
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let reader = nuclei::spawn(async move {
        let (mut stream, _) = listener.accept().await?;
        let mut all = Vec::new();
        stream.read_to_end(&mut all).await?;
        std::io::Result::Ok(all.len())
    });

    let stream = TcpStream::connect(addr).await?;
    // send_file: the kernel sends from the page cache, the bytes never pass through the program.
    let file = File::open("page.html").await?;
    assert_eq!(stream.send_file(&file, 0, 64 * 1024).await?, 64 * 1024);

    // write_zero_copy: the kernel sends from the buffer, which comes back once it let go.
    let (written, released) = stream.write_zero_copy(vec![b'y'; 2 * 1024 * 1024]).await;
    assert_eq!(written?, 2 * 1024 * 1024);
    let buf = released.await;
    assert_eq!(buf.len(), 2 * 1024 * 1024);

    stream.shut(Shutdown::Write).await?;
    assert_eq!(reader.await?, 64 * 1024 + 2 * 1024 * 1024);
    nuclei::fs::remove_file("page.html").await
}

Pipes and FIFOs

net::unix::pipe::pipe() makes an anonymous pipe: a Sender and a Receiver, async streams like any other. A named pipe in the file system opens with pipe::OpenOptions, as a receiver or a sender.

src/main.rs
use nuclei::io::{AsyncReadExt, AsyncWriteExt};
use nuclei::net::unix::pipe;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // An anonymous pipe: what goes into the sender comes out of the receiver.
    let (mut sender, mut receiver) = pipe::pipe()?;
    let writer = nuclei::spawn(async move {
        sender.write_all(b"through a pipe").await?;
        // Dropping the sender ends the stream for the reader.
        drop(sender);
        std::io::Result::Ok(())
    });
    let mut text = String::new();
    receiver.read_to_string(&mut text).await?;
    assert_eq!(text, "through a pipe");
    writer.await
}

std's calls through the handle

std's methods reach the socket through the handle, and the handle's own methods of the same name take their place where it has them (connect, send, linger, ...). A few of std's go around the bytes the handle read ahead and wrote behind, or block its shard:

  • shutdown(how) shuts the halves at once, ahead of the bytes still behind; shut waits for them.
  • set_nonblocking(false) makes the driver's own calls on the socket block the shard's thread, and every task of the shard with them.
  • take_error takes the socket's pending error, which the handle's connect in progress or its next operation would have answered.
  • std's Read and Write on a shared reference to the socket read past the bytes read ahead, write ahead of the bytes written behind, and answer WouldBlock where the handle's own calls wait.
  • UdpSocket::connect(addr) makes a new connected socket. socket.connect(addr) on a socket you have is std's: it connects that socket in place and looks a name up with a blocking call on the shard's thread, so give it a numeric address, or look the name up first with net::lookup_host.

The Handle reference has the details.