Build a multi-threaded receive pipeline

Give the receive loop one job

Run one dedicated ingest thread per receive socket. It calls recvmmsg or recvmsg, rejects the wrong source, timestamps each datagram, copies it into an owned buffer, and attempts a nonblocking send to a bounded queue. It does not verify Ed25519 signatures, allocate log strings, group FEC sets, decode entries, or call RPC.

At 5,585 packets per second, the mean interval is about 179 microseconds. Bursts make many intervals much shorter. A blocking downstream operation in this loop converts directly into socket queue occupancy.

Use a fixed-capacity object shape so the allocator does not decide receive latency. The following worker split compiles with crossbeam-channel = "0.5" and keeps overload visible:

use crossbeam_channel::{bounded, Receiver, Sender, TrySendError};
use std::{io, net::{Ipv4Addr, SocketAddr, UdpSocket}, sync::atomic::{AtomicU64, Ordering}, thread, time::Instant};

const MAX_PACKET: usize = 1228;

struct Datagram {
    received: Instant,
    len: usize,
    bytes: Box<[u8; MAX_PACKET]>,
}

static QUEUE_DROPS: AtomicU64 = AtomicU64::new(0);

fn ingest(sock: UdpSocket, tx: Sender<Datagram>) -> io::Result<()> {
    let allowed = Ipv4Addr::new(64, 130, 40, 90);
    loop {
        let mut bytes = Box::new([0u8; MAX_PACKET]);
        let (len, peer) = sock.recv_from(bytes.as_mut_slice())?;
        let allowed_source = match peer {
            SocketAddr::V4(v4) => *v4.ip() == allowed,
            SocketAddr::V6(_) => false,
        };
        if allowed_source == false { continue; }
        let item = Datagram { received: Instant::now(), len, bytes };
        if let Err(TrySendError::Full(_)) = tx.try_send(item) {
            QUEUE_DROPS.fetch_add(1, Ordering::Relaxed);
        }
    }
}

fn worker(rx: Receiver<Datagram>) {
    while let Ok(d) = rx.recv() {
        let payload = &d.bytes[..d.len];
        let age = d.received.elapsed();
        std::hint::black_box((payload, age));
    }
}

fn main() -> io::Result<()> {
    let sock = UdpSocket::bind("0.0.0.0:8001")?;
    let (tx, rx) = bounded::<Datagram>(8192);
    for _ in 0..4 { let r = rx.clone(); thread::spawn(move || worker(r)); }
    ingest(sock, tx)
}

This example allocates one box per packet to keep the ownership rule clear. Replace it with a bounded slab or pool after profiling. Return buffers to a pool on every accept, rejection, and error path. A pool exhaustion counter is also an overload counter.

Partition by protocol key

Generic competing workers are suitable for stateless header parsing and signature verification. FEC grouping and ordered reassembly have per-key state. Route every shred with the same (slot, fec_set_index) to the same FEC worker, for example by hashing the tuple modulo worker count. This removes locks inside an FEC set and makes duplicate handling deterministic.

After recovery, route completed data shreds by slot to an ordering worker. A slot can contain multiple completed data ranges, and serialized entry batches may span more than one FEC set. Preserve data-shred index order and use DATA_COMPLETE_SHRED boundaries. Do not assume one FEC set equals one entry batch.

Changing worker count changes the hash partition. Drain the old pipeline before changing it, or use a consistent ownership table keyed by active slot. Moving half-built FEC sets between workers adds synchronization and usually costs more than it saves.

Use bounded stages

Bound every channel and name its full condition. Suggested stages are receive to parse, parse to signature verification, verified shred to FEC owner, recovered data range to deshred, and decoded transaction to consumer. Export depth, capacity, enqueue failures, and oldest-item age for each stage.

Do not block the ingest thread when the first queue is full. Increment a counter and discard the datagram. Blocking would move the same loss into the kernel and hide its exact cause. Later stages can use short blocking sends if preserving a completed unit is more valuable, but cap the wait and observe it.

Use Little's Law as a check. If a stage receives 5,585 items per second and p99 service time is 400 microseconds on one worker, one worker cannot keep up at p99 conditions. Four workers offer theoretical capacity, but measure contention, cache behavior, and the fraction of coding shreds that skip later stages.

Batch expensive work

Ed25519 verification benefits from batching when the library exposes a vetted batch path. Do not create a batch timeout so large that it consumes the latency advantage of raw shreds. Flush on count or a short deadline. Retain individual failure accounting because batch failure may require checking members separately.

Reed-Solomon recovery is per FEC set. Start it only when at least num_data_shreds total unique shards are present and at least one data shard is absent. If all data shreds are present, bypass recovery. Coding shards are redundancy, not payload that must be decoded on every complete set.

Control CPU placement

Name threads and record their CPU time. Put the ingest thread on a quiet core after proving that scheduler movement contributes latency. Keep the NIC receive interrupt placement in mind. Verification workers can occupy a separate CPU set. FEC and decode workers may compete for cache and memory bandwidth, so increasing thread count can reduce throughput.

Use release builds with debug symbols for profiling. Measure packets received per system call, cycles per accepted shred, queue waits, signature verification time, recovery time, entry decode time, and transaction output rate. Tune from the first saturated stage.

Shut down without corrupting state

On shutdown, stop admitting new traffic, close the first sender, and let workers drain in stage order. Set a deadline. Incomplete FEC sets and slots are disposable raw observations, so persist only if restart recovery is an explicit requirement. Report the number discarded at shutdown.

Catch worker panic at the supervisor boundary. A dead verification or FEC worker can leave one hash partition accumulating while aggregate throughput still looks acceptable. Health checks must report live worker count, last progress time by partition, and queue age, not only process liveness.