Welcome back to Raw Ptr Insights. In the previous installment we ported the relay to async Rust: replaced OS threads with Tokio tasks, traced how #[tokio::main] wraps an epoll-based reactor, and established why into_split() with task::spawn is the correct async duplex pattern where tokio::select! silently drops data. The relay now runs inside Tokio’s work-stealing scheduler with constant OS thread count regardless of session load. What we have not yet done is accept more than one session — listen_mode still calls accept().await once and exits.
The step from single-session to multi-session is not an incremental refinement. It requires two new structural decisions: how to keep the accept loop running indefinitely while relay tasks run concurrently, and how to manage the state those sessions share. The first is mechanically straightforward — loop { accept().await; spawn(relay(session)) } — but it creates an unbounded collection of detached tasks whose lifetimes are invisible to the spawning loop. The second is where the real engineering choices live: whether shared state belongs in a Mutex-guarded HashMap or in an actor task that owns the state exclusively, and whether sessions relay data through a shared broadcast channel or through point-to-point pipes.
Most async Rust tutorials present shared state through the Arc<Mutex<T>> pattern and leave it there. The limitations of that pattern — lock contention on hot paths, the risk of holding a lock across an .await point, and the inability to express backpressure — are not theoretical. In a relay context they are observable: a session that holds the lock while waiting for a slow recipient blocks every other session from registering or deregistering. The actor pattern — a single coordinator task that owns the state and receives commands via mpsc channels — eliminates the lock entirely. We will build both and name which properties of the problem determine the better choice.
The broadcast relay is the structural synthesis of the series. By the end of this installment, the relay will accept connections indefinitely, register each session in a shared directory, and fan out data from any one session to all others via tokio::sync::broadcast. This is the -k (keep-listening) behavior that makes netcat useful as a hub rather than a pipe.
Graceful shutdown closes the loop: abandoned sessions leave ghost entries in the session registry and in /proc/net/tcp, both of which are directly observable. We will use JoinSet to track spawned tasks and clean up on disconnect, and we will show what the difference between a clean and an unclean session lifecycle looks like at the kernel level.
This installment covers four sections:
- The Accept Loop as a Task Factory — builds the indefinite accept loop and establishes task-per-session spawning with ownership detachment via
tokio::spawn. - Sharing State Across Sessions — compares
Arc<Mutex<HashMap>>and the actor pattern for session registry management, with a concrete analysis of which failure modes each creates. - Broadcast and Relay: Connecting Sessions — wires
tokio::sync::broadcastto fan relay data from any session to all others, with an honest account of the lagged receiver failure mode. - Graceful Shutdown and Session Cleanup — tracks task lifetimes with
JoinSet, handles disconnect detection, and correlates abandoned sessions to/proc/net/tcpghost entries.
The Accept Loop as a Task Factory
The Mechanism
The accept loop is the simplest structural change in this installment, but it introduces a lifetime question that deserves precision: what owns a spawned task?
tokio::spawn returns a JoinHandle<T>. If the handle is dropped, the task is detached — it continues running but its return value is discarded. If the handle is held, the task can be awaited to retrieve its return value or abort it. The accept loop spawns one task per session:
use tokio::net::TcpListener;
use tokio::task;
async fn serve(bind_addr: &str) -> io::Result<()> {
let listener = TcpListener::bind(bind_addr).await?;
eprintln!("[*] Listening on {}", bind_addr);
loop {
let (session_stream, peer_addr) = listener.accept().await?;
eprintln!("[+] Session accepted: {}", peer_addr);
// Spawn a task for this session and immediately drop the JoinHandle.
// The task is now detached — it runs to completion independently
// of the accept loop. If the task panics, the error is lost unless
// we track the handle. JoinSet (Section 4) fixes this.
task::spawn(async move {
if let Err(e) = duplex_relay(session_stream).await {
eprintln!("[!] Session {} error: {}", peer_addr, e);
}
});
// The loop continues immediately — accept() is already waiting
// for the next connection while the spawned task handles this one.
}
}Dropping the JoinHandle is intentional here. Holding all handles in a Vec inside the loop would work but requires bookkeeping — tasks that have completed keep their handles live in the vector. JoinSet (Section 4) is the correct tool for tracked task collection.
The accept loop itself is a future that yields at each accept().await. Between yields, the Tokio scheduler runs other tasks — including the relay tasks we spawned. The loop consumes no CPU while waiting for connections; it is parked in the reactor’s epoll interest set.
The Security Correlation
A listener that accepts connections without bound is a target for resource exhaustion. An adversary opening connections but never sending data occupies relay tasks that block in read().await — each costing task state (a few hundred bytes) and a kernel socket buffer (typically 4–128 KB). Unlike the blocking model where each connection costs one OS thread, the task-based model reduces the per-connection cost dramatically. At 10,000 idle connections, the blocking relay would require 80 GB of stack space; the async relay requires roughly 40–100 MB of task and socket buffer overhead. This does not eliminate the exhaustion risk, but it raises the cost threshold by three orders of magnitude. Rate-limiting at the accept loop (listener.set_ttl() does not apply here, but OS-level iptables rate limiting or SO_REUSEPORT with load balancing does) is the correct mitigation — not addressed in this prototype but named as the natural extension.
Sharing State Across Sessions
The Mechanism
Multiple sessions that need to exchange data require a shared registry — a structure both the accept loop and individual relay tasks can read and write. The straightforward approach in async Rust is Arc<Mutex<T>>:
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
type SessionRegistry = Arc<Mutex<HashMap<SocketAddr, SessionHandle>>>;
struct SessionHandle {
peer_addr: SocketAddr,
// Additional per-session state (e.g., a channel sender) goes here.
}Each relay task receives a clone of the Arc, locks the mutex to register itself at startup, does its relay work, then locks again to deregister on shutdown. The borrow checker enforces that the MutexGuard is not held across an .await point — because MutexGuard<T> is not Send in Rust, and spawned tasks require Send bounds. This is a compile-time enforcement of a real correctness requirement: holding a std::sync::Mutex lock across an await point parks the task but keeps the lock acquired, blocking every other task that needs the same lock.
// This does NOT compile — the MutexGuard is held across .await.
// error[E0277]: `MutexGuard<'_, ...>` cannot be sent between threads safely
async fn bad_pattern(registry: SessionRegistry, stream: TcpStream) {
let guard = registry.lock().unwrap();
do_something_async().await; // MutexGuard is not Send; compiler rejects this
drop(guard);
}
// Correct pattern: lock, modify, drop before any await.
async fn good_pattern(registry: SessionRegistry, stream: TcpStream) {
{
let mut guard = registry.lock().unwrap();
guard.insert(stream.peer_addr().unwrap(), SessionHandle { /* ... */ });
} // guard dropped here — lock released before any .await
do_something_async().await;
}For cases where an async-aware mutex is needed — holding state across an await in a context where contention is expected — tokio::sync::Mutex parks the task rather than spinning the thread. The tradeoff: tokio::sync::Mutex is safe to hold across .await but has higher per-acquisition overhead than std::sync::Mutex. For a session registry that is accessed only briefly (insert on connect, remove on disconnect), std::sync::Mutex is the correct choice.
The actor pattern eliminates the mutex entirely. Instead of sharing data behind a lock, a single coordinator task owns the registry exclusively and receives commands via mpsc channels:
use tokio::sync::mpsc;
enum RegistryCommand {
Register { peer_addr: SocketAddr, sender: mpsc::Sender<Vec<u8>> },
Deregister { peer_addr: SocketAddr },
Broadcast { data: Vec<u8>, origin: SocketAddr },
}
async fn registry_actor(mut commands: mpsc::Receiver<RegistryCommand>) {
let mut sessions: HashMap<SocketAddr, mpsc::Sender<Vec<u8>>> = HashMap::new();
while let Some(cmd) = commands.recv().await {
match cmd {
RegistryCommand::Register { peer_addr, sender } => {
sessions.insert(peer_addr, sender);
}
RegistryCommand::Deregister { peer_addr } => {
sessions.remove(&peer_addr);
}
RegistryCommand::Broadcast { data, origin } => {
// Fan out to all sessions except the originator.
sessions.retain(|addr, sender| {
if *addr == origin { return true; }
// A send failure means the session's receiver was dropped
// (it disconnected). Remove the stale entry.
sender.try_send(data.clone()).is_ok()
});
}
}
}
}The actor task is the sole owner of sessions. No lock is needed because no other code can access sessions — all access goes through the mpsc::Sender<RegistryCommand>. Contention becomes task scheduling: commands queue in the channel rather than threads spinning on a mutex.
| Property | Arc<Mutex<HashMap>> | Actor (mpsc) |
|---|---|---|
| Concurrent read access | Only one writer; no concurrent reads | Single owner; all access serialised through channel |
Lock held across .await | Compile error (non-Send guard) | N/A — no lock |
| Backpressure | None — lock acquisition always proceeds | mpsc::Sender::send().await blocks on full channel |
| Failure on receiver disconnect | N/A | send() returns Err; caller knows session is gone |
| Implementation complexity | Low | Moderate — requires command enum and coordinator task |
For a relay registry that is accessed briefly and infrequently, Arc<Mutex<HashMap>> is simpler and correct. For a broadcast fan-out that must propagate data to all sessions with backpressure awareness, the actor pattern is better suited. We use tokio::sync::broadcast in Section 3, which is a purpose-built specialisation of the actor pattern for one-to-many distribution.
The Security Correlation
A shared session registry is an information disclosure surface. Any task that holds a clone of the Arc<Mutex<HashMap>> can enumerate all active sessions — their peer addresses, their start times, their associated channel senders. In an embedded use case where the relay library is consumed by a larger binary with multiple components, the registry’s visibility should be scoped to the minimum required. Rust’s module system enforces this at the type level: making SessionRegistry private to the relay module means no external code can read or modify it without going through the relay’s public API. Encapsulation in Rust is not a best-practice guideline — it is a property the type system can enforce at compile time.
Broadcast and Relay: Connecting Sessions
The Mechanism
tokio::sync::broadcast is a multi-producer, multi-consumer channel where every receiver sees every message sent after it subscribed. A relay that wants to fan data from any one connected session to all others sends to the broadcast channel; each session has a receiver subscribed to it.
use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use tokio::task;
const CHANNEL_CAPACITY: usize = 256;
async fn serve_broadcast(bind_addr: &str) -> io::Result<()> {
let listener = TcpListener::bind(bind_addr).await?;
eprintln!("[*] Broadcast relay on {}", bind_addr);
// One channel shared by all sessions. Each session gets a Sender clone
// and subscribes a Receiver. Data sent by any session is received by all.
let (tx, _) = broadcast::channel::<Vec<u8>>(CHANNEL_CAPACITY);
loop {
let (session_stream, peer_addr) = listener.accept().await?;
eprintln!("[+] Session: {}", peer_addr);
// Clone the sender for this session. All clones share the same channel.
let session_tx = tx.clone();
// Subscribe a new receiver. It will see all messages sent after this
// point — not messages sent before subscription (no replay).
let mut session_rx = tx.subscribe();
task::spawn(async move {
let (mut socket_rx, mut socket_tx) = session_stream.into_split();
let peer = peer_addr;
// Inbound: read from socket, broadcast to all other sessions.
let inbound = task::spawn(async move {
let mut buf = vec![0u8; 8192];
loop {
let n = match socket_rx.read(&mut buf).await {
Ok(0) | Err(_) => break, // EOF or error — session ended
Ok(n) => n,
};
// A send error means no active receivers — not fatal.
let _ = session_tx.send(buf[..n].to_vec());
}
});
// Outbound: receive from broadcast, write to socket.
let outbound = task::spawn(async move {
loop {
match session_rx.recv().await {
Ok(data) => {
if socket_tx.write_all(&data).await.is_err() { break; }
}
// Lagged means this receiver fell behind and missed messages.
// We log and continue — the session gets a gap, not a disconnect.
Err(broadcast::error::RecvError::Lagged(n)) => {
eprintln!("[!] {} lagged — {} messages dropped", peer, n);
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
});
let _ = tokio::join!(inbound, outbound);
eprintln!("[-] Session closed: {}", peer);
});
}
}Technical Reality Check
broadcast::channel has a fixed capacity. When the channel is full — because a slow receiver is not consuming messages — new sends succeed but the lagged receiver’s unread messages are overwritten. The receiver then receives RecvError::Lagged(n), where n is the number of missed messages. This is not an error in the fatal sense; the session continues. But it means a slow client silently loses data. For a relay used as a debugging hub — forwarding shell output to multiple observers — this is acceptable. For a relay used to forward security-critical data — credentials, C2 commands — it is not. The capacity constant is a correctness parameter, not a tuning parameter. Size it based on the maximum burst the slowest receiver can tolerate without lagging, and instrument lagged events as anomalies.
The Security Correlation
Every connected session receives every message broadcast by every other session. This is the intended behaviour for a relay hub, but it is a zero-isolation model: any client that connects receives all data in transit. An attacker who can connect to the relay — either legitimately or through a compromised listener port — immediately receives the full data stream from all active sessions. In a scenario where the relay is used to aggregate output from multiple compromised hosts, a new connection to the relay port gains access to all hosts’ output simultaneously. Authentication and session isolation are not in scope for this prototype and must be addressed before the relay handles sensitive data. Article 5’s CLI will expose the bind address as a configurable flag; restricting the bind address to a loopback or VPN interface is the minimum access control.
Graceful Shutdown and Session Cleanup
The Mechanism
Detached tasks — those whose JoinHandle was dropped — run invisibly to the accept loop. A session that disconnects causes the relay task to return, but the accept loop has no visibility into this. For a short-lived service this is acceptable; for a long-running relay hub, indefinite task accumulation is a resource leak. tokio::task::JoinSet provides a collection of task handles that can be polled for completion without blocking:
use tokio::task::JoinSet;
async fn serve_tracked(bind_addr: &str) -> io::Result<()> {
let listener = TcpListener::bind(bind_addr).await?;
let (tx, _) = broadcast::channel::<Vec<u8>>(256);
let mut sessions: JoinSet<()> = JoinSet::new();
loop {
tokio::select! {
// Accept a new connection and spawn its task into the JoinSet.
accept_result = listener.accept() => {
let (stream, peer_addr) = accept_result?;
let session_tx = tx.clone();
let session_rx = tx.subscribe();
sessions.spawn(async move {
handle_session(stream, peer_addr, session_tx, session_rx).await;
});
}
// Reap completed tasks. join_next() returns None only when the
// JoinSet is empty, which select! handles by never choosing this branch.
Some(result) = sessions.join_next() => {
// A JoinError here means the task panicked — log it.
if let Err(e) = result {
eprintln!("[!] Session task panicked: {}", e);
}
}
}
}
}The select! here is safe — neither branch holds user-space buffered data. The listener.accept() future holds no application data; it is purely a kernel notification. The sessions.join_next() future completes immediately when a task has finished, returning its result. No bytes are at risk of being dropped.
JoinSet::abort_all() cancels all pending tasks — the correct shutdown path when the accept loop itself receives a termination signal. Combined with a tokio::signal::ctrl_c() future in the select! branches, this produces a listener that shuts down cleanly on Ctrl+C rather than being killed by the OS.
The Security Correlation
Abandoned sessions — those that disconnected without completing their relay task’s cleanup — leave observable traces. The kernel’s TCP state machine for a half-closed connection persists in /proc/net/tcp as a CLOSE_WAIT or FIN_WAIT_2 entry until the process closes the fd. If a session task panics before dropping its OwnedWriteHalf, the socket fd stays open and the entry stays in /proc/net/tcp. A process with hundreds of CLOSE_WAIT entries is diagnosable from outside the process by anyone with /proc read access. JoinSet reaps tasks and their associated resources promptly. A relay with clean /proc/net/tcp state is a relay that does not advertise its history to local observers.
Looking Ahead
The relay now handles concurrent sessions indefinitely, fans data through a broadcast channel, and reaps tasks cleanly via JoinSet. The code is a single file — a working multi-session relay server with no separation between the relay engine, the networking layer, and the command-line interface. Article 5 addresses that directly: splitting lib.rs from bin/, replacing the implicit listen/connect branching with a Transport trait that makes the structure explicit, and wiring a full clap-derive CLI that maps canonical nc flags onto a typed configuration type. The goal is a crate that can be both a standalone binary and a library embedded inside other tooling.
Further Reading & Deep Dives:
- For
tokio::sync::broadcastsemantics, channel capacity, and the lagged receiver failure mode, see the Tokio broadcast module documentation.- For the actor pattern in async Rust — owning state in a coordinator task and communicating via channels — see Programming Rust (Blandy et al., Chapter 20) and the Tokio tutorial: Channels.
- For TCP connection state in the kernel and the conditions under which entries persist in
/proc/net/tcp, see The Linux Programming Interface (Kerrisk, Chapter 61).- For
JoinSetand structured task concurrency in Tokio, see the Tokio task module documentation.
Leave a Reply