Network Carriers

datum-net is the network satellite crate for Datum. It holds network sources, sinks, and connection utilities built on the core Source / Flow / Sink model from datum-core.

The byte carriers (TLS-over-TCP, UDP, and QUIC) are fully Tokio-native: each network connection runs as a single-owner async task with no blocking I/O seam inside the carrier. Remote StreamRefs add TCP and QUIC carriers on top (the TCP carrier follows the same single-owner async model — see Remote StreamRefs transport). OS threads stay flat (~9 instead of ~1 per connection), and CPU scales down materially at concurrency. A physical-core-sharded Tokio carrier engine is available for extreme concurrency and can be tuned with environment variables. On Linux, an optional io-uring-file feature exposes a tokio-uring file source/sink alongside the default TokioFileIO.

TLS Client And Server

use datum::{Keep, Sink, Source};
use datum_net::{
    TokioTls,
    tls::rustls::{
        ClientConfig, ServerConfig,
        pki_types::ServerName,
    },
};
use std::sync::Arc;

fn run_tls_echo(
    server_config: Arc<ServerConfig>,
    client_config: Arc<ClientConfig>,
) -> datum::StreamResult<()> {
    let (binding, incoming) = TokioTls::bind("127.0.0.1:0", server_config, 8192)
        .to_mat(Sink::head(), Keep::both)
        .run()?;
    let binding = binding.wait()?;

    let server_name = ServerName::try_from("localhost")
        .expect("valid DNS name")
        .to_owned();
    let client = Source::single(b"ping".to_vec())
        .via(TokioTls::outgoing_connection(
            binding.local_addr(),
            server_name,
            client_config,
            8192,
        ))
        .run_with(Sink::head())?;

    let incoming = incoming.wait()?;
    let (source, sink) = incoming.into_parts();
    let request = source.run_with(Sink::head())?.wait()?;
    Source::single(request).run_with(sink)?.wait()?;

    assert_eq!(client.wait()?, b"ping".to_vec());
    Ok(())
}

TokioTls::outgoing_connection_default and TokioTls::bind_default use the default 8 KiB chunk size. TlsIncomingConnection::into_flow() converts an accepted TLS connection into a coupled Flow<Vec<u8>, Vec<u8>, NotUsed> when that is more convenient than handling the byte source and sink separately.

Connection Lifecycle

use datum::{Keep, Sink, Source};
use datum_net::{
    ConnectionLifecycleExt, ConnectionSettings, RetryPolicy, TokioTls,
    tls::rustls::{ClientConfig, pki_types::ServerName},
};
use std::{sync::Arc, time::Duration};

fn run_lifecycle_client(
    addr: std::net::SocketAddr,
    client_config: Arc<ClientConfig>,
) -> datum::StreamResult<Vec<u8>> {
    let settings = ConnectionSettings::default()
        .connect_timeout(Duration::from_millis(250))
        .handshake_timeout(Duration::from_secs(2))
        .retry_policy(
            RetryPolicy::default()
                .max_attempts(4)
                .initial_backoff(Duration::from_millis(50))
                .backoff_multiplier(2.0)
                .max_backoff(Duration::from_millis(250)),
        );

    let server_name = ServerName::try_from("localhost")
        .expect("valid DNS name")
        .to_owned();
    let flow = TokioTls::outgoing_connection_with_lifecycle(
        addr,
        server_name,
        client_config,
        settings,
    )
    .half_close_on_upstream_finish();

    let (connection, response) = Source::single(b"request".to_vec())
        .via_mat(flow, Keep::right)
        .to_mat(Sink::collect(), Keep::both)
        .run()?;

    let _ = connection.wait()?;
    Ok(response.wait()?.concat())
}

ConnectionSettings::default() bounds TCP connect and TLS handshake to 30 seconds and performs one attempt. RetryPolicy::max_attempts includes the first attempt. Upstream completion on TCP/TLS connection byte flows shuts down the write direction (FIN / TLS close) while the read direction continues until the peer finishes, matching Akka Tcp half-close behavior where the transport supports it. ConnectionLifecycleExt::half_close_on_upstream_finish() is an explicit marker for that behavior.

UDP Datagrams

use datum::{Keep, Sink, Source};
use datum_net::{Datagram, TokioUdp};

fn run_udp_once() -> datum::StreamResult<()> {
    let (binding, received) = TokioUdp::bind("127.0.0.1:0", 2048, 16)
        .take(1)
        .to_mat(Sink::head(), Keep::both)
        .run()?;
    let binding = binding.wait()?;

    Source::single(Datagram::new(b"ping".to_vec(), binding.local_addr()))
        .run_with(TokioUdp::send_sink("127.0.0.1:0"))?
        .wait()?;

    let datagram = received.wait()?;
    assert_eq!(datagram.payload(), b"ping");
    assert!(datagram.remote().ip().is_loopback());
    Ok(())
}

TokioUdp::bind_default, TokioUdp::bind_flow_default, and TokioUdp::connect_default use a 65,536-byte receive buffer per datagram and a 64-datagram in-process receive buffer. Datagram preserves UDP message boundaries: one socket receive emits one Datagram, and one upstream Datagram is sent with one send_to.

UDP is lossy and has no end-to-end flow control. Datum keeps receive buffering bounded; when the configured in-process buffer is full, additional datagrams are dropped, and the operating system may also drop datagrams before Datum receives them. Use an application-level protocol if ordering, retries, acknowledgements, or reliable delivery are required.

QUIC Bidirectional Streams

use datum::{Keep, Sink, Source};
use datum_net::{TokioQuic, quic::quinn};

fn run_quic_echo(
    server_config: quinn::ServerConfig,
    client_config: quinn::ClientConfig,
) -> datum::StreamResult<()> {
    let (binding, incoming) = TokioQuic::bind("127.0.0.1:0", server_config, 8192)
        .to_mat(Sink::head(), Keep::both)
        .run()?;
    let binding = binding.wait()?;

    let client_connection = TokioQuic::connect(
        binding.local_addr(),
        "localhost",
        client_config,
        8192,
    )
    .run_with(Sink::head())?
    .wait()?;

    let incoming = incoming.wait()?;
    let client_response = Source::single(b"ping".to_vec())
        .via(client_connection.open_bi_default())
        .run_with(Sink::head())?;

    let server_stream = incoming
        .accept_bi_default()
        .run_with(Sink::head())?
        .wait()?;
    let (server_source, server_sink) = server_stream.into_parts();
    let request = server_source.run_with(Sink::head())?.wait()?;
    Source::single(request).run_with(server_sink)?.wait()?;

    assert_eq!(client_response.wait()?, b"ping".to_vec());
    Ok(())
}

Build quinn::ServerConfig / quinn::ClientConfig from rustls configs with quinn::ServerConfig::with_crypto(Arc::new(QuicServerConfig::try_from(server_crypto)?)) and quinn::ClientConfig::new(Arc::new(QuicClientConfig::try_from(client_crypto)?)). The datum_net::quic module re-exports quinn, rustls, and crypto so callers can use the same types without adding duplicate imports.

QUIC streams are reliable, ordered, and flow-controlled. Datum maps a bidirectional QUIC stream to a Flow<Vec<u8>, Vec<u8>, StreamCompletion<QuicStream>>; read-side buffering is bounded and write-side backpressure follows Quinn's stream flow control. Quinn only exposes a newly opened bidirectional stream to the peer after the opener writes data or sends FIN, so call accept_bi_default after the initiating side has started the stream.

Bidirectional streams ship first. Uni-directional QUIC streams and richer connection lifecycle controls for QUIC are deferred to keep this slice small and fully tested.

Plain TCP remains in the core Streaming IO guide.

Async/Tokio-native carriers

As of v0.7.0, every datum-net byte carrier (TLS/UDP/QUIC) is fully Tokio-native. The previous architecture called Tokio blocking_recv on the read side and Handle::block_on on the write side from synchronous Datum stream operators. That seam scaled to ~1 OS thread per active connection — 1,033 threads at 1,024 concurrent streams. The single-owner async-task model eliminates that scaling entirely.

The new architecture, measured for each carrier:

Carrier	Single-stream wall	1,024-stream threads	1,024-stream CPU
TLS-over-TCP	+32% faster	9 (was 1,033)	−27%
UDP datagrams	tied (±0.3%)	9 (was 1,033)	−39%
QUIC bidir	+39% faster	9 (was 1,033)	−30%

The shared async_carrier helper (AsyncCommandSender + DemandBatcher) provides a bounded demand-and-command channel between the synchronous Datum stream core and each single-owner Tokio task. The TLS, UDP, and QUIC byte carriers all use the same helper. The remote-StreamRefs carriers are described separately below — the TCP carrier is a single-owner async task following the same pattern, while the QUIC carrier drives the protocol with a dedicated per-direction loop.

The public TokioTls, TokioUdp, and TokioQuic APIs are unchanged. No test had to be modified. The only user-visible change is that threads no longer scale with connection count.

High-concurrency sharding

A single multi-thread Tokio runtime with one listener hits a wall cliff at extreme concurrency (4,668 ms/op at 1,024 TLS streams versus the bridge's 3,528 ms/op — a +32% regression caused by single-runtime scheduling contention).

The sharded-Tokio carrier (ShardedTokioCarrierExecution) closes that cliff. Each shard is a dedicated OS thread running its own current_thread Tokio runtime owning a disjoint subset of connections. Sharding is internal (public APIs unchanged) and activates automatically when the active connection count crosses the configured minimum.

Physical-core shard-count heuristic

physical_cores = num_cpus::get_physical()
cgroup_cap     = std::thread::available_parallelism()
cores          = clamp(physical_cores, 1, cgroup_cap)
shards         = clamp(cores, 1, min(cores, active_connections))

• num_cpus::get_physical() returns the physical core count (not SMT threads).
• The result is capped at available_parallelism() for cgroup/container safety.
• Shard count is further clamped to min(cores, active_connections) so low-connection workloads don't over-shard.
• No sharding below ~2 physical cores or below the configurable connection threshold (default 64).

Environment variables

Variable	Default	Description
DATUM_NET_SHARDED_TOKIO_SHARDS	heuristic	Override the shard count.
DATUM_NET_SHARDED_TOKIO_MIN_CONNECTIONS	64	Minimum active connections before sharding activates.
DATUM_NET_SHARDED_TOKIO_DISABLE	(unset)	Set to 1 to disable sharding entirely.

Suggested shard counts by physical cores

Physical cores	Shards (default)	Rationale
2	0 (no sharding)	Below the 2-core threshold.
4	4	Worth it at ≥64 connections.
8 (4c/8t SMT)	4	Physical cores, not SMT siblings.
16	16	Common server SKU.
32	32	High-core server.
64 / 96	min(cores, active)	Active-connection clamp dominates.

TLS sharding results (96-physical-core AMD EPYC 7R13)

Concurrent streams	Non-sharded p50 ms	Sharded p50 ms	Wall speedup	CPU improvement
1	20.4	22.0 (fallback)	—	—
64	348.8	36.4	9.6×	1.32×
256	1,163.9	117.6	9.9×	1.45×
1,024	4,401.3	415.4	10.6×	1.49×

At 64+ connections, the sharded path provides ~9–10× wall improvement with no low-concurrency regression. The sharded carrier engine is carrier-agnostic by design, but is currently wired only into the TLS carrier; the UDP and QUIC byte carriers use the shared single-owner Tokio task model without sharding.

Remote StreamRefs transport

datum-core owns the protocol seam (protobuf handshake, cumulative demand, sequenced elements, completion, failure, and terminal ack — mirroring Akka StreamRefs). datum-net carries the frames over two transports:

Carrier	Transport	Encrypted	Use case
Plaintext TCP	serve_source_ref_over_tcp / source_ref_over_tcp / serve_sink_ref_over_tcp / sink_ref_over_tcp	No	Trusted/loopback; Akka Artery-TCP-equivalent baseline.
Encrypted QUIC	serve_source_ref_over_quic / source_ref_over_quic / serve_sink_ref_over_quic / sink_ref_over_quic	Yes (TLS 1.3)	Untrusted/WAN.

Both carriers share the same protobuf protocol and StreamRefSettings (32-element buffer, 30s subscription timeout). The QUIC carrier routes StreamRefs frames through a length-delimited frame decoder over a single QUIC bidi stream. The TCP carrier uses the same frame decoder over tokio::net::TcpStream.

The TCP carrier runs as a single-owner async Tokio task (WP-F3), the same pattern as the TLS/UDP/QUIC byte carriers. The QUIC StreamRefs carrier instead drives the protocol with a dedicated per-direction loop. Because remote StreamRefs are long-lived and low-cardinality (one bidi stream per remote), this difference does not affect the connection-count thread scaling that motivated the byte-carrier work.

The same-process direct splice remains the fast local path (~6.5× faster than Akka) and is untouched by the remote transport layer.

For full API details and examples, see the StreamRefs guide.

io-uring-file feature (Linux only)

On Linux kernels ≥5.10, an optional io-uring-file feature enables a UringFileIO source/sink backed by tokio-uring. It sits alongside the default TokioFileIO (unchanged, always available) and is opt-in at compile time:

[dependencies]
datum-net = { version = "0.7", features = ["io-uring-file"] }

The feature is Linux-only; non-Linux targets compile without the feature and fall back to TokioFileIO. All internal unsafe is in the tokio-uring dependency; datum-net itself is forbid(unsafe_code).

Integrated benchmark results (full stream machinery, not raw syscalls):

Operation	io-uring-file vs TokioFileIO
256 KiB read	1.70× (median)
64 MiB read	1.33×
256 KiB write	noisy/slower — not a defaulting signal
64 MiB write	1.38×

Performance

Datum-net benchmarks are measured the same way as the core library: 5× warmup, 5× measurement, whole-process CPU via /proc/self/stat, plus peak RSS and thread count. The full benchmark record lives in roadmap/benchmarks/net.md.

Akka comparison (forced-remote, equal honest measurement)

After WP-RT5 corrected the Akka harness to force real Artery-TCP remoting (the prior comparison table measured an in-JVM ActorRef/SourceRef handoff, not a real network boundary), the honest standings are:

Area	Datum wall	Akka wall	Verdict
TLS echo (64 B roundtrip)	553 µs/op	1,899 µs/op	3.43× faster
UDP send/receive (128 × 64 B)	651 µs/op	1,069 µs/op	1.64× faster
Remote StreamRefs TCP (1,024 elements)	1,951 µs/op	18,612 µs/op	9.54× faster
Remote StreamRefs TCP reuse	2,072 µs/op	18,895 µs/op	9.12× faster
Remote StreamRefs QUIC (1,024 elements)	6,611 µs/op	17,530 µs/op	2.65× faster
Remote StreamRefs QUIC reuse	5,169 µs/op	18,848 µs/op	3.65× faster

On the plaintext TCP StreamRefs path (the fair Artery-TCP-equivalent baseline), Datum is ∼10× faster wall, ∼15× lower CPU, and ∼35× lower allocation than warmed forced-remote Akka. The N-sweep gives a per-element wall slope of 1.27 µs/elem for Datum vs 17.0 µs/elem for Akka.

All performance claims are accompanied by the benchmark record; the comparison rows, methodology, caveats, and per-axis verdicts are in roadmap/benchmarks/net.md.