diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 659f64bfab..0c6000da6a 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -68,6 +68,8 @@ jobs: artifact-path: target/bench/ffi - name: Audio grouping relay benchmark recipe: bench-audio + - name: Relay shard sweep + recipe: bench-shard # Smoke-tests the capture path: exact off-CPU waits, resolved worker stacks and CPU. - name: Relay lock-wait profile recipe: bench-lock fanout-heavy 4 diff --git a/Cargo.lock b/Cargo.lock index a0a7eee062..e742f7cd14 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5024,7 +5024,6 @@ dependencies = [ "moq-sock", "moq-tokio", "nix", - "rand 0.10.3", "rcgen", "rustc-hash 2.1.3", "rustls", diff --git a/bench/README.md b/bench/README.md index 36b665f628..a6424ad7ba 100644 --- a/bench/README.md +++ b/bench/README.md @@ -69,6 +69,21 @@ Runtime comparison requires Linux because io\_uring and relay process metrics come from Linux interfaces. The default worker count is the number of online logical CPUs. +Sweep the relay's QUIC shard count against the connection count, for the +room (`video`: every connection publishes and subscribes) and `fanout` (one +publisher, the rest subscribers) workloads: + +```bash +nix develop --command just bench-shard +nix develop --command just bench-shard "0 1 8 24" "64 512" +``` + +Zero shards is the shared runtime with one Tokio thread per logical CPU; any +other count is that many unpinned Tokio workers. The table reports CPU per +connection, so a steering, dispatch, or per-shard cost that grows with either +axis reads as a slope. Linux only, since nothing else steers a reuseport +group. It runs nightly. + Attribute relay worker lock waits to the stacks that blocked, as a share of each worker's CPU: diff --git a/bench/run.sh b/bench/run.sh index d1fc701c7d..65bbe4b9d0 100755 --- a/bench/run.sh +++ b/bench/run.sh @@ -8,6 +8,7 @@ set -euo pipefail # on the current tree. The current load generator drives every relay so the # workloads stay identical. `--audio` isolates the audio grouping/load-shape matrix. # `--lock` records relay worker off-CPU stacks to attribute std mutex waits. +# `--shard` sweeps the QUIC shard count against the connection count. MODE=${1:-} ROUNDS=${MOQ_BENCH_RUNTIME_ROUNDS:-3} @@ -27,6 +28,17 @@ case $MODE in # Without frame pointers, perf's DWARF unwinder stops at the first relay frame. export RUSTFLAGS="${RUSTFLAGS:+$RUSTFLAGS }-C force-frame-pointers=yes" ;; + --shard) + BASE= + SHARD_COUNTS=${MOQ_BENCH_SHARD_COUNTS:-0 1 4 16} + SHARD_CONNECTIONS=${MOQ_BENCH_SHARD_CONNECTIONS:-16 64 256} + for value in $SHARD_COUNTS $SHARD_CONNECTIONS; do + if [[ ! $value =~ ^[0-9]+$ ]]; then + printf 'shard benchmark counts must be integers, got %q\n' "$value" >&2 + exit 1 + fi + done + ;; --runtime) BASE= RUNTIME_WORKERS=${MOQ_BENCH_RUNTIME_WORKERS:-} @@ -492,6 +504,61 @@ run_runtime_comparison() { print_runtime_comparison } +# Each workload's load axis is its connection count: every room connection +# publishes and subscribes, while fanout adds subscribers to one publisher. A +# cost that grows with the shard count or the load shows up as a slope across +# the shard counts or the connection counts, which one default-versus-single- +# socket run cannot show. +run_shard_sweep() { + if [[ $(uname -s) != Linux ]]; then + printf 'shard sweep needs Linux: only it steers a reuseport group\n' >&2 + return 1 + fi + generate_certificate + + printf 'Building relay and load generator...\n' + CARGO_TARGET_DIR=$CURRENT_TARGET cargo build --locked --release -p moq-relay -p moq-bench + LOAD_BIN=$CURRENT_TARGET/release/moq-bench + HOST_BIN=$CURRENT_TARGET/release/moq-bench-host + + local cpus workload connections shards label + cpus=$(getconf _NPROCESSORS_ONLN) + for workload in video fanout; do + for connections in $SHARD_CONNECTIONS; do + for shards in $SHARD_COUNTS; do + label=shard-$shards-$connections + printf ' %s: %d connections, %d shards\n' "$workload" "$connections" "$shards" + # Zero shards is the shared runtime, given every CPU. + if ((shards == 0)); then + run_workload "$label" "$CURRENT_TARGET/release/moq-relay" "$workload" tokio-shared "$cpus" \ + --connections "$connections" + else + run_workload "$label" "$CURRENT_TARGET/release/moq-relay" "$workload" tokio-workers "$shards" \ + --connections "$connections" + fi + done + done + done + + printf '\nShard sweep (0 shards is the shared runtime on %d threads)\n' "$cpus" + printf '%-8s %11s %7s %10s %10s %8s %7s %8s %9s %9s\n' \ + workload connections shards recv-Mbps recv-fps p99-ms loss-% CPU CPU/conn RSS-MiB + for workload in video fanout; do + for connections in $SHARD_CONNECTIONS; do + for shards in $SHARD_COUNTS; do + jq -r --arg connections "$connections" --arg shards "$shards" '[ + .workload, $connections, $shards, .recv_mbps, .recv_fps, .latency_p99_ms, .loss_pct, + .cpu_cores, (.cpu_cores / ($connections | tonumber)), (.rss_bytes / 1048576) + ] | @tsv' "$RUN/relay/shard-$shards-$connections/$workload/summary.json" | + while IFS=$'\t' read -r name conns count recv fps p99 loss cpu per rss; do + printf '%-8s %11s %7s %10.3f %10.1f %8.1f %7.3f %8.3f %9.5f %9.1f\n' \ + "$name" "$conns" "$count" "$recv" "$fps" "$p99" "$loss" "$cpu" "$per" "$rss" + done + done + done + done +} + run_lock_profile() { if [[ $(uname -s) != Linux ]]; then printf 'lock profiling needs Linux for perf and /proc schedstat\n' >&2 @@ -543,6 +610,9 @@ elif [[ $MODE == --runtime ]]; then elif [[ $MODE == --lock ]]; then run_lock_profile exit 0 +elif [[ $MODE == --shard ]]; then + run_shard_sweep + exit 0 elif [[ -n $BASE ]]; then BASE_COMMIT=$(git rev-parse --verify "$BASE^{commit}") WORKTREE=$RUN/base diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 9bd91a4a5e..5faf7df0a1 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -110,14 +110,15 @@ window, so very small values can prevent other requests from progressing. ## \[runtime] By default one work-stealing runtime serves every connection off one UDP -socket. On Linux, QUIC can instead run on pinned single-threaded workers, each -with its own socket on the listen address (`SO_REUSEPORT`), so a connection is -handled start to finish by one thread and its packets never cross cores. +socket. On Linux, QUIC can instead run on single-threaded workers, each with +its own socket on the listen address (`SO_REUSEPORT`), so a connection is +handled start to finish by one worker thread and its packets never cross +threads. `pin = true` also keeps that thread on one core. ```toml [runtime] workers = 8 # Single-threaded QUIC workers. Omit for the shared runtime. -pin = true # Pin each worker to a core. Default. +pin = false # Pin each worker to a core. Off by default. io_uring = false # Drive them with io_uring instead of tokio. ``` @@ -126,6 +127,10 @@ worker. `workers` needs the `quic` feature. Every worker and the TLS stream listener serve one shared identity, generated or file-backed, so a certificate rotation reaches them together and `/certificate.sha256` stays accurate. +Pinning is off by default because every relay pins its workers to the same +first cores, so two relays on one host, or one limited to fewer CPUs than the +host has, would crowd onto them. Turn it on for a host the relay has to itself. + `io_uring` additionally needs Linux 6.12+ and the `io-uring` cargo feature; it shares the same reloadable TLS identity, serves moq-lite only, and refuses `mtu_discovery`. A setting the build or host cannot deliver refuses to diff --git a/justfile b/justfile index 1b926cd1fc..bc4b8aa04f 100644 --- a/justfile +++ b/justfile @@ -60,6 +60,11 @@ bench-audio: bench-runtime $ROUNDS="3" $WORKERS="": MOQ_BENCH_RUNTIME_ROUNDS="$ROUNDS" MOQ_BENCH_RUNTIME_WORKERS="$WORKERS" bench/run.sh --runtime +# Sweep the relay's QUIC shard count against room and fanout connection counts; +# 0 shards is the shared runtime. Example: `just bench-shard "0 1 8" "64 512"`. +bench-shard $SHARDS="0 1 4 16" $CONNECTIONS="16 64 256": + MOQ_BENCH_SHARD_COUNTS="$SHARDS" MOQ_BENCH_SHARD_CONNECTIONS="$CONNECTIONS" bench/run.sh --shard + # Attribute relay worker std mutex waits by stack, for each worker count. bench-lock $WORKLOAD="fanout" $WORKERS="1 4 16": MOQ_BENCH_LOCK_WORKLOAD="$WORKLOAD" MOQ_BENCH_LOCK_WORKERS="$WORKERS" bench/run.sh --lock diff --git a/quest/m1/quic/README.md b/quest/m1/quic/README.md index f20b9def8b..f6bb5c18e2 100644 --- a/quest/m1/quic/README.md +++ b/quest/m1/quic/README.md @@ -72,8 +72,6 @@ cache limits remain separate; bitrate pacing composes with this memory cap. datagrams join their subscription's send group instead of preempting every stream - [Relay peers get wider limits](/quest/m1/quic/peer-limits.md) - MAX_STREAMS and MAX_DATA are raised after SETUP identifies a cluster peer -- [Shard the endpoint](/quest/m1/quic/shard.md) - the library shards a - server endpoint across cores by default, replacing the `--workers` group ## Related diff --git a/quest/m1/quic/shard.md b/quest/m1/quic/shard.md deleted file mode 100644 index 5646eec724..0000000000 --- a/quest/m1/quic/shard.md +++ /dev/null @@ -1,42 +0,0 @@ -# [M] Shard the QUIC endpoint across cores - -## Goal - -A `moq-quic` server endpoint shards across cores by default where the -platform can steer (Linux): the library -forms the steered `SO_REUSEPORT` group and issues connection IDs that name -their shard, and each runtime (moq-tokio and moq-uring) only drives the -shards it is handed on threads of its own. The relay's `--workers` flag -becomes an optional count override, and the two copies of the group logic -are gone. - -## Plan - -Decided in the 2026-09-30 plan: the library shards and the runtime drives. -The library does not own threads, so embedders (moq-ffi and the bindings over it) keep their -own threading model. - -- `moq_sock::shard` (the group, claims, and steering filter) stays in - `moq-sock` beside `moq_sock::udp` (quinn-udp), the one copy both runtimes - use; `moq-sock` keeps binding and CPU helpers. -- `moq-quic` issues shard-prefixed connection IDs itself, replacing the - custom CID generators in moq-tokio and moq-uring. -- moq-tokio's `worker/group.rs` shrinks to spawning one pinned - `current_thread` runtime per shard; moq-uring's `udp::Bound` group code is - replaced by the library's. -- A server endpoint defaults to one shard per core only where - `SO_REUSEPORT` steering works, which today is Linux: `moq-sock` returns - `Unsupported` for both load balancing and steering elsewhere. macOS and - Windows keep a genuine single-socket endpoint, and an explicit count above - one is refused there rather than silently ignored. -- Before flipping the default, add a benchmark sweeping shard count against - concurrent session load (publishers and subscribers), so steering, - dispatch, and per-shard overhead show up as slopes. A single - default-versus-one-socket run cannot show a cost that grows with either - axis. - -Update `doc/bin/relay/` for the flag change. - -## Related - -- [UDP demux](/quest/m2/p2p/udp-demux.md) - demuxes each shard's socket; keep the demux over whatever the library hands out diff --git a/quest/m2/perf/README.md b/quest/m2/perf/README.md index 78b93c6fbc..f288ab41a2 100644 --- a/quest/m2/perf/README.md +++ b/quest/m2/perf/README.md @@ -77,6 +77,7 @@ optimization. - [Run to quiescence](/quest/m2/perf/uring-quiescence.md) - a received packet's reply is staged in the same turn, under a pass and train budget that keeps the fairness rule - [Ingest batch](/quest/m2/perf/ingest-batch.md) - relay ingest pays one lock, wake, and clock read per burst of whole frames instead of per frame - [Remove moq-uring copies](/quest/m2/perf/uring-copies.md) - egress, stream send and receive, and datagram receive stop copying where the QUIC core already allows it, with before/after measurements on `moq-quic` +- [Shard by default](/quest/m2/perf/shard-default.md) - a saturating shard sweep decides whether an unset `runtime.workers` runs one worker per core on Linux ## Related diff --git a/quest/m2/perf/kio-channel-contention/README.md b/quest/m2/perf/kio-channel-contention/README.md index 45fe90d3d4..c94e961734 100644 --- a/quest/m2/perf/kio-channel-contention/README.md +++ b/quest/m2/perf/kio-channel-contention/README.md @@ -135,7 +135,7 @@ Decided (maintainer, 2026-10-14, per-thread follow-up): - Per-worker sharding is no longer ruled out; two spikes decide. Both run on the thread-per-core workers (`--runtime-workers`, the default once - [Shard](/quest/m1/quic/shard.md) lands) and link workers over an + [Shard by default](/quest/m2/perf/shard-default.md) lands) and link workers over an in-memory transport (`moq_net::transport::memory`), never real QUIC. The [replica spike](/quest/m2/perf/kio-channel-contention/spike-replica.md) diff --git a/quest/m2/perf/shard-default.md b/quest/m2/perf/shard-default.md new file mode 100644 index 0000000000..a765352d56 --- /dev/null +++ b/quest/m2/perf/shard-default.md @@ -0,0 +1,37 @@ +# [M] Shard the relay by default + +## Goal + +A saturating benchmark shows whether thread-per-core QUIC beats the shared +runtime, and the relay's default follows the measurement. If sharding wins, +an unset `runtime.workers` runs one worker per core on Linux, and +`workers = 0` is the explicit way back to the shared runtime. Off Linux, the +shared runtime stays the default. + +## Plan + +Sharding is opt-in: `runtime.workers` is unset by default, and +`runtime.pin` defaults off. The maintainer decided on #5688 to keep it that +way until a measurement justifies the flip. + +- Extend `just bench-shard` to load that saturates the shared runtime, for + example 1024+ connections or more than one load host. At 16 to 256 + connections, #5688's sweep never used one full core, so it showed no + regression from sharding and no win either. The 16-shard groups spent about + 15 to 25% more CPU than the shared runtime at low load, from idle per-core + runtimes. +- Flip the default only if sharding raises the throughput ceiling or cuts + CPU per connection at saturation. A measured no-win deletes this quest. +- Shard only where a group serves the same listener the shared runtime would + open. A stream-only relay, or one with a QUIC-LB `lb_id`, stays shared + unless a count is set. A count sized from `available_parallelism`, capped at + `moq_sock::shard::MAX_SHARDS`, follows cgroup quotas. +- With the default flipped, `runtime.io_uring` no longer needs an explicit + count. + +Update `doc/bin/relay/` for the default change. + +## Related + +- [kio channel contention](/quest/m2/perf/kio-channel-contention/README.md) - its per-thread spikes run on the workers +- [UDP demux](/quest/m2/p2p/udp-demux.md) - demuxes each shard's socket diff --git a/rs/moq-quic/src/cid_generator.rs b/rs/moq-quic/src/cid_generator.rs index 810f5603e7..85215eb070 100644 --- a/rs/moq-quic/src/cid_generator.rs +++ b/rs/moq-quic/src/cid_generator.rs @@ -39,11 +39,12 @@ pub struct InvalidCid; /// Generates purely random connection IDs of a specified length /// /// Random CIDs can be smaller than those produced by [`HashedConnectionIdGenerator`], but cannot be -/// usefully [`validate`](ConnectionIdGenerator::validate)d. +/// usefully [`validate`](ConnectionIdGenerator::validate)d unless they name a shard. #[derive(Debug, Clone, Copy)] pub struct RandomConnectionIdGenerator { cid_len: usize, lifetime: Option, + shard: Option, } impl Default for RandomConnectionIdGenerator { @@ -51,10 +52,41 @@ impl Default for RandomConnectionIdGenerator { Self { cid_len: 8, lifetime: None, + shard: None, } } } +/// One endpoint's slot in a group sharing a port, which its connection IDs name. +#[derive(Debug, Clone, Copy)] +struct Shard { + index: u16, + count: u16, +} + +impl Shard { + /// A first byte congruent to `index` modulo `count`, keeping as much of the byte random as + /// the group size leaves: `256 / count` values. + fn prefix(self) -> u8 { + let count = u32::from(self.count); + let stride = rand::rng().random_range(0..256 / count); + (stride * count + u32::from(self.index)) as u8 + } + + /// Whether `cid` carries this shard's prefix and the marker. + fn owns(self, cid: &[u8]) -> bool { + cid.len() > SHARD_MARKER_BYTE + && u16::from(cid[0]) % self.count == self.index + && cid[SHARD_MARKER_BYTE] & SHARD_MARKER == SHARD_MARKER + } +} + +/// The byte of a sharded CID that carries [`SHARD_MARKER`]. +const SHARD_MARKER_BYTE: usize = 3; +/// Both bits overlay the reserved key-encryption pattern of an SRT data header; see +/// [`RandomConnectionIdGenerator::set_shard`]. +const SHARD_MARKER: u8 = 0x18; + impl RandomConnectionIdGenerator { /// Initialize Random CID generator with a fixed CID length /// @@ -72,16 +104,55 @@ impl RandomConnectionIdGenerator { self.lifetime = Some(d); self } + + /// Name shard `index` of a `count`-endpoint group in every CID + /// + /// The first byte is congruent to `index` modulo `count`, so a `SO_REUSEPORT` steering filter + /// that reduces a packet's first destination CID byte modulo the group size delivers it to the + /// endpoint that issued the CID. The fourth byte also sets both bits of `0x18`, which overlay + /// the reserved key-encryption pattern of an SRT data header, so the filter can tell a short + /// header from another protocol sharing the port. Every other bit stays random: a client + /// chooses the Initial's destination CID, so it can steer every connection to one shard, and + /// that shard alone has to have IDs to spare. [`validate`](ConnectionIdGenerator::validate) + /// then refuses a CID without this shard's prefix and the marker, so a stray packet is dropped + /// rather than answered with a stateless reset for a connection a sibling owns. + /// + /// Panics unless `index < count <= 256`, since one byte cannot name more shards, or if the + /// CIDs are too short to carry the marker. + pub fn set_shard(&mut self, index: u16, count: u16) -> &mut Self { + assert!( + index < count && count <= 256, + "shard {index} of {count} cannot be named by one byte" + ); + assert!( + self.cid_len > SHARD_MARKER_BYTE, + "a {}-byte CID cannot carry the shard marker", + self.cid_len + ); + self.shard = Some(Shard { index, count }); + self + } } impl ConnectionIdGenerator for RandomConnectionIdGenerator { fn generate_cid(&mut self) -> ConnectionId { let mut bytes_arr = [0; MAX_CID_SIZE]; rand::rng().fill_bytes(&mut bytes_arr[..self.cid_len]); + if let Some(shard) = self.shard { + bytes_arr[0] = shard.prefix(); + bytes_arr[SHARD_MARKER_BYTE] |= SHARD_MARKER; + } ConnectionId::new(&bytes_arr[..self.cid_len]) } + fn validate(&self, cid: ConnectionId) -> Result<(), InvalidCid> { + match self.shard { + Some(shard) if !shard.owns(&cid) => Err(InvalidCid), + _ => Ok(()), + } + } + /// Provide the length of dst_cid in short header packet fn cid_len(&self) -> usize { self.cid_len @@ -177,4 +248,94 @@ mod tests { let cid = generator.generate_cid(); generator.validate(cid).unwrap(); } + + fn sharded(index: u16, count: u16) -> RandomConnectionIdGenerator { + let mut generator = RandomConnectionIdGenerator::default(); + generator.set_shard(index, count); + generator + } + + /// Bits that vary across `cids`: a fixed bit is entropy the generator lost. + fn varying_bits(cids: &[ConnectionId]) -> u32 { + let (mut ones, mut zeros) = ([0u8; 8], [0u8; 8]); + for cid in cids { + for (i, byte) in cid.iter().enumerate() { + ones[i] |= byte; + zeros[i] |= !byte; + } + } + (0..8).map(|i| (ones[i] & zeros[i]).count_ones()).sum() + } + + /// The steering filter's only requirement: whatever first byte a shard issues reduces back + /// to that shard, at every group size including the one-value-per-shard ceiling. + #[test] + fn a_shard_cid_names_its_shard() { + for count in [1u16, 2, 3, 7, 8, 64, 255, 256] { + for index in 0..count { + let mut generator = sharded(index, count); + for _ in 0..32 { + let cid = generator.generate_cid(); + assert_eq!(cid.len(), generator.cid_len()); + assert_eq!(u16::from(cid[0]) % count, index); + assert_eq!(cid[3] & 0x18, 0x18, "CID lacks the non-SRT marker"); + generator.validate(cid).unwrap(); + } + } + } + } + + /// A client picks the Initial's destination CID, so it can steer every connection to one + /// shard, and that shard alone has to keep a wide space: only the prefix byte and the two + /// marker bits are spent on the encoding. When the prefix took a byte of a 3-byte nonce, a + /// 256-way shard had 65,536 CIDs, and the endpoint spins looking for a free one once they are + /// taken. + #[test] + fn a_full_size_shard_keeps_its_entropy() { + let mut generator = sharded(0, 256); + let cids: Vec<_> = (0..1024).map(|_| generator.generate_cid()).collect(); + assert_eq!(varying_bits(&cids), 64 - 8 - 2); + } + + /// The unsharded path spends nothing on the encoding. + #[test] + fn an_unsharded_cid_keeps_its_entropy() { + let mut generator = RandomConnectionIdGenerator::default(); + let cids: Vec<_> = (0..1024).map(|_| generator.generate_cid()).collect(); + assert_eq!(varying_bits(&cids), 64); + } + + /// A shard refuses a sibling's CIDs and CIDs without the marker, so a misrouted or foreign + /// packet is dropped rather than reset. + #[test] + fn a_shard_refuses_what_it_did_not_issue() { + let first = sharded(0, 4); + let cid = sharded(1, 4).generate_cid(); + assert!(first.validate(cid).is_err(), "took a sibling's CID"); + + let mut unmarked = sharded(0, 4).generate_cid(); + unmarked[3] &= !0x18; + assert!( + first.validate(unmarked).is_err(), + "took a CID without the marker" + ); + } + + #[test] + #[should_panic] + fn a_shard_past_the_group_is_refused() { + RandomConnectionIdGenerator::default().set_shard(4, 4); + } + + #[test] + #[should_panic] + fn a_group_past_one_byte_is_refused() { + RandomConnectionIdGenerator::default().set_shard(0, 257); + } + + #[test] + #[should_panic] + fn a_cid_too_short_for_the_marker_is_refused() { + RandomConnectionIdGenerator::new(3).set_shard(0, 2); + } } diff --git a/rs/moq-relay/src/config.rs b/rs/moq-relay/src/config.rs index 707ffbc379..e48a066995 100644 --- a/rs/moq-relay/src/config.rs +++ b/rs/moq-relay/src/config.rs @@ -1351,12 +1351,12 @@ uid = [1001] }, Case { name: "declared-default-true-false-from-file", - toml: "[runtime]\npin = false\n", + toml: "[web]\nws = false\n", env: &[], cli: &[], - key: "runtime.pin", - check: |config| assert!(!config.runtime.pin, "file false must beat a declared true default"), - source: "pin", + key: "web.ws", + check: |config| assert!(!config.web.ws, "file false must beat a declared true default"), + source: "ws", }, Case { name: "optional-with-default-from-file", diff --git a/rs/moq-relay/src/runtime.rs b/rs/moq-relay/src/runtime.rs index d9c83cb27d..625e7d89ae 100644 --- a/rs/moq-relay/src/runtime.rs +++ b/rs/moq-relay/src/runtime.rs @@ -1,4 +1,4 @@ -//! Flags for serving QUIC from pinned per-core workers. +//! Flags for serving QUIC from per-core workers. //! //! The workers themselves are [`moq_tokio::worker`]; this is the operator-facing //! half, which is the only part specific to the relay. Unset (the default) keeps @@ -7,13 +7,13 @@ use serde::{Deserialize, Serialize}; /// How the relay lays its QUIC work out over threads. -#[derive(usage::Args, Clone, Debug, Deserialize, Serialize)] +#[derive(usage::Args, Clone, Debug, Default, Deserialize, Serialize)] #[usage(unknown_flags = "error", args_override_self = false)] #[serde(default, deny_unknown_fields)] #[non_exhaustive] pub struct Config { /// Serve QUIC from this many single-threaded workers instead of the shared - /// runtime, each pinned to a core with its own socket on the listen address. + /// runtime, each with its own socket on the listen address. /// /// A connection is handled start to finish by one worker, which trades the /// shared runtime's load balancing for no cross-thread traffic per packet. @@ -22,21 +22,16 @@ pub struct Config { #[usage(long = "runtime-workers", env = "MOQ_RUNTIME_WORKERS", setting = "runtime.workers")] pub workers: Option, - /// Pin each worker to a CPU core, defaulting to on. + /// Pin each worker to a CPU core, defaulting to off. /// /// The mode's measured win comes from each worker owning a socket and a /// runtime, not from pinning: on a single-socket machine, pinned and unpinned /// benchmark inside run-to-run noise of each other. Pinning stops the - /// scheduler migrating a busy worker, which should matter on a multi-socket - /// or NUMA machine, and costs nothing elsewhere. Turn it off when sharing - /// the machine with something that manages CPU placement itself. - #[usage( - long = "runtime-pin", - env = "MOQ_RUNTIME_PIN", - setting = "runtime.pin", - default = "true", - bool_value - )] + /// scheduler migrating a busy worker, which may matter on a multi-socket or + /// NUMA machine with the cores to itself. It is off by default because every + /// process pins its workers to the same first cores: two relays, or a relay + /// limited to fewer CPUs than the host has, would crowd onto those cores. + #[usage(long = "runtime-pin", env = "MOQ_RUNTIME_PIN", setting = "runtime.pin", bool_value)] pub pin: bool, /// Drive the QUIC workers with io_uring instead of tokio. @@ -72,13 +67,3 @@ impl Config { self.io_uring } } - -impl Default for Config { - fn default() -> Self { - Self { - workers: None, - pin: true, - io_uring: false, - } - } -} diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 5df9462027..c0f9b57a60 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -169,23 +169,10 @@ impl Workers { // The group owns the reuseport invariants: it locks the port against a // concurrently-constructing same-UID group (the bind probe covers one // that already holds it), refuses a size the steering filter could not - // address, and hands out members in the order the kernel numbers them - // by. - let mut forming = moq_sock::shard::Group::acquire(requested, config.count) - .with_context(|| format!("failed to take the reuseport group on {requested}"))?; - let count = forming.count(); - - let mut claims = Vec::with_capacity(count as usize); - while let Some(member) = forming.member() { - let shard = member.shard(); - let claim = member - .bind() - .with_context(|| format!("failed to bind worker {} on {}", shard.index(), forming.addr()))?; - claims.push(claim); - } - let mut group = forming - .complete(claims) - .context("failed to complete the reuseport group")?; + // address, and binds members in the order the kernel numbers them by. + let mut group = moq_sock::shard::bind(requested, config.count) + .with_context(|| format!("failed to bind the reuseport group on {requested}"))?; + let count = group.count(); let mut members = Vec::with_capacity(count as usize); while let Some(member) = group.member().context("failed to clone a reuseport member")? { members.push(member); diff --git a/rs/moq-sock/Cargo.toml b/rs/moq-sock/Cargo.toml index 919c9334f2..e1c63232f5 100644 --- a/rs/moq-sock/Cargo.toml +++ b/rs/moq-sock/Cargo.toml @@ -19,7 +19,6 @@ fast-apple-datapath = [] [dependencies] core_affinity = "0.8" libc = { workspace = true } -rand = "0.10" socket2 = { version = "0.6", features = ["all"] } thiserror = "2" tracing = "0.1" @@ -31,4 +30,5 @@ windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_System_ cfg_aliases = "0.2" [dev-dependencies] +rand = "0.10" tracing-test = "0.2" diff --git a/rs/moq-sock/README.md b/rs/moq-sock/README.md index a1659e021d..7cab2a2101 100644 --- a/rs/moq-sock/README.md +++ b/rs/moq-sock/README.md @@ -7,13 +7,14 @@ runtimes (`moq-tokio` and `moq-uring`). `SO_REUSEPORT` for worker-per-core groups (Linux-only, refused loudly elsewhere). - **`shard`**: forming and steering a reuseport group by QUIC connection id. - A `shard::Group` holds the port (probing and locking against a second - same-UID group), fixes the member count, and hands out one `Member` per slot - in index order. Binding a member yields an opaque claim, and completing the - group with every claim attaches a classic-BPF filter before any socket is - released for serving. The completed group retains every member socket, so - dropping one serving handle cannot renumber the survivors. A `Shard` names - the slot a member ended up in, and `cid` makes an issued id that names it. + `shard::bind` holds the port (probing and locking against a second same-UID + group), binds every member in index order, and attaches a classic-BPF filter + before any socket is released for serving. The bound group retains every + member socket, so dropping one serving handle cannot renumber the survivors. + A `Shard` names the slot a member holds, which the endpoint serving it names + in every connection id it issues, alongside a marker no SRT data header + carries (`moq_quic::RandomConnectionIdGenerator::set_shard`). Anything else + on the port, including other protocols, keeps the kernel's 4-tuple hash. - **`cpu`**: pinning worker threads to cores. - **`udp`**: sending and receiving with GSO, GRO, and ECN, a hard fork of [quinn-udp](https://github.com/quinn-rs/quinn). All credit for it goes to diff --git a/rs/moq-sock/src/lib.rs b/rs/moq-sock/src/lib.rs index a0392b4f30..fa3f6641f4 100644 --- a/rs/moq-sock/src/lib.rs +++ b/rs/moq-sock/src/lib.rs @@ -2,7 +2,7 @@ //! runtimes. //! //! [`bind`] opens dual-stack UDP/TCP sockets with sane buffers; -//! [`shard::Group`] forms and steers an `SO_REUSEPORT` group by QUIC connection +//! [`shard::bind`] forms and steers an `SO_REUSEPORT` group by QUIC connection //! id, so a worker-per-core listener keeps each connection on the socket that //! owns it; [`cpu`] pins those workers; [`udp`] (forked from quinn-udp) sends //! and receives with GSO, GRO, and ECN. Both `moq-tokio` and `moq-uring` build diff --git a/rs/moq-sock/src/shard.rs b/rs/moq-sock/src/shard.rs index d772c5303e..af6d349da7 100644 --- a/rs/moq-sock/src/shard.rs +++ b/rs/moq-sock/src/shard.rs @@ -12,37 +12,34 @@ //! //! # The group //! -//! [`Group`] is the whole entry point. It takes the port, fixes the member -//! count, and hands out one [`Member`] per slot in index order. Binding yields -//! an opaque [`Claim`], so no socket can be served while the group is partial. -//! [`Group::complete`] accepts every claim, attaches the steering filter, and -//! returns a [`Bound`] group that hands out sockets while retaining every -//! underlying member for its own lifetime. +//! [`bind`] is the whole entry point. It takes the port, binds every member in +//! index order, attaches the steering filter, and returns a [`Bound`] group +//! that hands out one serving [`Socket`] per member while retaining every +//! underlying socket for its own lifetime. //! //! ```no_run //! # fn example(addr: std::net::SocketAddr) -> Result<(), Box> { -//! let mut group = moq_sock::shard::Group::acquire(addr, 4)?; -//! let mut claims = Vec::new(); -//! while let Some(member) = group.member() { -//! claims.push(member.bind()?); -//! } -//! let mut group = group.complete(claims)?; +//! let mut group = moq_sock::shard::bind(addr, 4)?; //! let mut members = Vec::new(); //! while let Some(member) = group.member()? { //! members.push((member.shard(), member.into_inner())); //! } -//! // Serve each socket, issuing connection IDs with `cid(shard)`. +//! // Serve each socket with an endpoint whose connection IDs name its shard. //! # Ok(()) //! # } //! ``` //! //! # The encoding //! -//! [`cid`] makes the first byte of every connection ID this member issues -//! congruent to its index modulo the group size. It also marks the ID with a -//! bit pattern that no valid SRT data header can carry. The kernel -//! selects `socks[A % count]` for a returned accumulator `A`, so feeding it that -//! byte lands on the owner. +//! The filter reads the first byte of a packet's destination connection ID and +//! reduces it modulo the group size. The endpoint serving a member has to issue +//! 8-byte connection IDs whose first byte is congruent to the member's +//! [`Shard::index`] modulo [`Shard::count`], and whose fourth byte has both +//! bits of `0x18` set, a pattern no valid SRT data header carries, so a short +//! header from another protocol on the port keeps the 4-tuple hash. +//! `moq_quic::RandomConnectionIdGenerator::set_shard` issues exactly those. +//! The kernel selects `socks[A % count]` for a returned accumulator `A`, so +//! feeding it that byte lands on the owner. //! //! A client's first packets carry a connection ID the *client* invented, which //! encodes nothing. Those hash to an arbitrary member, and that is fine: whoever @@ -57,24 +54,23 @@ //! (the kernel pulls the header off first) and uses the return value as the //! index. That rule fits in classic BPF, with no `CAP_BPF`, no //! BPF toolchain in the build, and no map to keep alive across restarts. The -//! cost is that the kernel selects by *position*, so the group has to be built -//! once, in order, and never resized, which is what [`Group`] is for. +//! cost is that the kernel selects by *position*, so the group has to be bound +//! once, in order, and never resized, which is why [`bind`] does all of it. use std::io; use std::net::{SocketAddr, UdpSocket}; -use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; /// One member's slot in a group of sockets sharing a port via `SO_REUSEPORT`. /// /// Every member binds the same address and the kernel spreads inbound datagrams /// across the group, so N servers on N threads can serve one port without a /// shared socket between them. The kernel identifies a member by its *position* -/// in the group, which is why only [`Group::member`] mints one: a shard is -/// meaningful only inside a group that was bound once, in index order, and is -/// never resized. +/// in the group, which is why only [`bind`] mints one: a shard is meaningful +/// only inside a group that was bound once, in index order, and is never +/// resized. /// -/// Carry it wherever the member issues connection IDs, which [`cid`] marks -/// with its index. +/// Carry it wherever the member issues connection IDs, which have to name it +/// (see [the encoding](self#the-encoding)). #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct Shard { index: u16, @@ -82,17 +78,6 @@ pub struct Shard { } impl Shard { - /// Slot `index` of a group of `count` sockets, or `None` if that slot cannot - /// exist: a `count` of zero or past [`MAX_SHARDS`], or an `index` at or past - /// the end. - /// - /// The upper bound is what makes every shard safe to steer: [`cid`] - /// spends `256 / count` values of a byte, so a larger group would leave it - /// none to spend. - fn new(index: u16, count: u16) -> Option { - (count <= MAX_SHARDS && index < count).then_some(Self { index, count }) - } - /// This member's position in the group, from zero. pub fn index(self) -> u16 { self.index @@ -133,200 +118,79 @@ pub enum Error { /// No port could be resolved for an ephemeral request. #[error("failed to resolve an ephemeral port")] Resolve(#[source] io::Error), + + /// A member could not bind, including [`io::ErrorKind::Unsupported`] + /// anywhere but Linux and [`io::ErrorKind::AddrInUse`] when the address is + /// already held. + #[error("failed to bind reuseport member {index}")] + Bind { + /// The member that failed. + index: u16, + /// What the bind returned. + #[source] + source: io::Error, + }, + + /// The steering filter could not be attached. + #[error("failed to steer the reuseport group")] + Steer(#[source] io::Error), } -/// A `SO_REUSEPORT` group being formed. +/// Bind a steered group of `count` sockets sharing `addr`. /// -/// Everything a steered group has to get right lives here rather than in its +/// Everything a steered group has to get right happens here rather than in its /// caller, because none of it is visible in a socket afterwards: /// /// - The port is locked before the first member binds and stays locked until the -/// group and every member it handed out are gone, so a second same-UID group -/// cannot interleave into this one. Hold the group for as long as its sockets -/// are served. +/// returned group and every socket it handed out are gone, so a second +/// same-UID group cannot interleave into this one. Hold the group for as long +/// as its sockets are served. /// - The member count is checked once, against what the steering filter can /// address, and never changes. -/// - [`member`](Self::member) hands out slots in index order, and a [`Member`] -/// is the only thing that can bind into the group. The kernel numbers the -/// group by bind order, so a member that binds out of turn is refused rather -/// than silently taking a sibling's slot. -/// - [`complete`](Self::complete) accepts only every claim from this group and -/// attaches the filter before returning a [`Bound`] group. A claim exposes no -/// socket, so a partial group cannot serve. -/// - [`Bound`] retains every underlying socket. A serving handle may be dropped -/// without removing its member and renumbering the survivors. +/// - Members bind in index order, which is the order the kernel numbers them +/// by, and the filter is attached before any socket is released. /// -/// Members may be bound wherever their sockets are served, a worker's own thread -/// included, as long as each one binds before the next takes its turn. -#[derive(Debug)] -pub struct Group { - count: u16, - - /// The next slot to hand out. Not the same as the next slot to *bind*: a - /// member may be bound on a thread of its own, which is what [`State`] - /// tracks. - next: u16, - - state: Arc, -} - -impl Group { - /// Take the port and fix the group at `count` members. - /// - /// A `count` of zero is a group of one: a lone member is a valid group, and - /// the callers that size a group from a worker count would otherwise each - /// clamp it themselves. - /// - /// An ephemeral port (`0`) is resolved here, with a plain bind that is - /// dropped again, and the group then takes that port like a named one. - /// Never a reuseport bind of port `0`: Linux counts a port held only by - /// same-UID reuseport sockets as free for one, so the kernel could hand - /// back a port another group holds and this one would silently join it. A - /// plain bind conflicts with every holder. Another socket taking the port - /// before the first member binds fails that bind with - /// [`io::ErrorKind::AddrInUse`]. - pub fn acquire(addr: SocketAddr, count: u16) -> Result { - let count = count.max(1); - if count > MAX_SHARDS { - return Err(Error::Count { count, max: MAX_SHARDS }); - } - - let addr = match addr.port() { - 0 => crate::bind::udp(crate::bind::Udp::new(addr)) - .and_then(|socket| socket.local_addr()) - .map_err(Error::Resolve)?, - _ => addr, - }; - - let port = addr.port(); - let lock = Lock::acquire(port).map_err(|_| Error::Overlap { port })?; - - Ok(Self { - count, - next: 0, - state: Arc::new(State::new(addr, lock)), - }) - } - - /// How many sockets share the port. - pub fn count(&self) -> u16 { - self.count - } - - /// The address the group holds, with an ephemeral port already resolved. - pub fn addr(&self) -> SocketAddr { - self.state.addr - } - - /// The next slot to bind, or `None` once every slot has been handed out. - /// - /// Bind each member before taking the next: they join the group in the order - /// they bind, and that order is the only thing the kernel knows them by. - pub fn member(&mut self) -> Option { - let shard = Shard::new(self.next, self.count)?; - self.next += 1; - Some(Member { - shard, - state: self.state.clone(), - }) - } - - /// Complete the group and make its sockets available for serving. - /// - /// Every member must have bound successfully, and the claims must belong to - /// this group in slot order. Failure closes the partial group as a whole. - pub fn complete(self, claims: Vec) -> io::Result { - if claims.len() != usize::from(self.count) { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - format!( - "reuseport group needs {} bound members; {} were provided", - self.count, - claims.len() - ), - )); - } - - for (index, claim) in claims.iter().enumerate() { - if !Arc::ptr_eq(&self.state, &claim.state) || usize::from(claim.shard.index()) != index { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "reuseport claims must belong to this group in slot order", - )); - } - } - - let sockets: Vec<_> = claims.into_iter().map(|claim| (claim.shard, claim.socket)).collect(); - let (_, last) = sockets.last().expect("a group always has at least one member"); - attach(last, self.count)?; - - Ok(Bound { - sockets, - next: 0, - state: self.state, - }) - } -} - -/// One member's claim on a slot in a [`Group`], which binding spends. +/// A `count` of zero is a group of one: a lone member is a valid group. /// -/// Send it wherever the socket is served, a worker's own thread included. It -/// carries the group's address and its share of the group's port lock, so a -/// member still waiting to bind cannot be raced by a second group even if its -/// group is dropped first. -#[derive(Debug)] -pub struct Member { - shard: Shard, - state: Arc, -} - -impl Member { - /// This member's slot, which its connection IDs have to encode - /// ([`cid`]). - pub fn shard(&self) -> Shard { - self.shard - } - - /// Bind this member into the group and return its opaque claim. - /// - /// Fails when a sibling has not bound yet: the kernel numbers a reuseport - /// group by bind order, so a member joining out of turn would take another's - /// slot and steer its traffic, with no error to show for it. - /// - /// The claim exposes no socket. Pass every claim to [`Group::complete`], - /// which attaches the filter before releasing serving handles. - pub fn bind(self) -> io::Result { - let mut bound = self.state.bound(); - if *bound != self.shard.index() { - return Err(io::Error::other(format!( - "reuseport member {} cannot bind while {} of {} are in: the kernel numbers a group by bind order", - self.shard.index(), - *bound, - self.shard.count(), - ))); - } +/// An ephemeral port (`0`) is resolved first, with a plain bind that is +/// dropped again, and the group then takes that port like a named one. +/// Never a reuseport bind of port `0`: Linux counts a port held only by +/// same-UID reuseport sockets as free for one, so the kernel could hand +/// back a port another group holds and this one would silently join it. A +/// plain bind conflicts with every holder. Another socket taking the port +/// before the first member binds fails that bind with +/// [`io::ErrorKind::AddrInUse`]. +pub fn bind(addr: SocketAddr, count: u16) -> Result { + let count = count.max(1); + if count > MAX_SHARDS { + return Err(Error::Count { count, max: MAX_SHARDS }); + } + + let addr = match addr.port() { + 0 => crate::bind::udp(crate::bind::Udp::new(addr)) + .and_then(|socket| socket.local_addr()) + .map_err(Error::Resolve)?, + _ => addr, + }; - let socket = bind(self.state.addr, self.shard)?; - *bound += 1; - drop(bound); + let port = addr.port(); + let lock = Lock::acquire(port).map_err(|_| Error::Overlap { port })?; - Ok(Claim { - shard: self.shard, - socket, - state: self.state, - }) + let mut sockets = Vec::with_capacity(usize::from(count)); + for index in 0..count { + let socket = member(addr, index).map_err(|source| Error::Bind { index, source })?; + sockets.push((Shard { index, count }, socket)); } -} -/// One successfully bound member of a forming [`Group`]. -/// -/// Opaque so a socket cannot be served before every member has joined and the -/// steering filter is attached. Pass all claims to [`Group::complete`]. -#[derive(Debug)] -pub struct Claim { - shard: Shard, - socket: UdpSocket, - state: Arc, + let (_, last) = sockets.last().expect("a group always has at least one member"); + attach(last, count).map_err(Error::Steer)?; + + Ok(Bound { + sockets, + next: 0, + addr, + _lock: lock, + }) } /// A complete steered group that retains every member socket. @@ -334,14 +198,16 @@ pub struct Claim { /// [`member`](Self::member) returns duplicate handles to the retained sockets. /// Dropping one handle stops serving that slot without removing it from the /// kernel group or renumbering any survivor. Dropping this group closes every -/// retained member together. +/// retained member together and releases the port. #[derive(Debug)] pub struct Bound { sockets: Vec<(Shard, UdpSocket)>, next: usize, + addr: SocketAddr, - /// Holds the port lock for exactly as long as the complete socket group. - state: Arc, + /// Held for exactly as long as the complete socket group, and released by + /// the kernel with the process. `None` on a host with no lock directory. + _lock: Option, } impl Bound { @@ -350,9 +216,9 @@ impl Bound { self.sockets.len().try_into().expect("group count fits u16") } - /// The address every member holds. + /// The address every member holds, with an ephemeral port already resolved. pub fn addr(&self) -> SocketAddr { - self.state.addr + self.addr } /// The next socket to serve, or `None` once every member was handed out. @@ -388,50 +254,11 @@ impl Socket { } } -/// What the members of a forming group share: the address and port lock it -/// holds, and how far the group has been bound. -/// -/// Shared rather than owned by the [`Group`] because a member is bound wherever -/// its socket is served, which is usually not where the group lives. The lock -/// rides along for the same reason: a member outstanding after its group is -/// dropped can still bind, so the port has to stay held until the last of them -/// is gone. -#[derive(Debug)] -struct State { - addr: SocketAddr, - - /// How many members have bound, which is also the only index allowed to bind - /// next. - bound: Mutex, - - /// Held until the group and its members are all dropped, and released by the - /// kernel with the process. `None` on a host with no lock directory. - _lock: Option, -} - -impl State { - fn new(addr: SocketAddr, lock: Option) -> Self { - Self { - addr, - bound: Mutex::new(0), - _lock: lock, - } - } - - /// The bound count, whatever a panicking member left behind: a failed bind - /// is reported by the count it did not advance, so there is no torn state a - /// poisoned lock would be protecting. - fn bound(&self) -> MutexGuard<'_, u16> { - self.bound.lock().unwrap_or_else(PoisonError::into_inner) - } -} - -/// Bind one member's socket into the group at `addr`. +/// Bind member `index` of the group at `addr`. /// -/// Private because the order is the whole invariant: reachable only through a -/// [`Member`], which a [`Group`] hands out in index order and only while it -/// holds the port. -fn bind(addr: SocketAddr, shard: Shard) -> io::Result { +/// Private because the order is the whole invariant: only [`bind`] calls this, +/// once per index in order, while it holds the port. +fn member(addr: SocketAddr, index: u16) -> io::Result { // The first member probes with a plain bind before the group forms. // `SO_REUSEPORT` groups by address and UID, so a member would otherwise // *join* any group a same-UID process already has on the address, as two @@ -445,9 +272,8 @@ fn bind(addr: SocketAddr, shard: Shard) -> io::Result { // *constructing* concurrently could each probe while the other holds // nothing yet, which is what [`Lock`] excludes; the probe's job is the // holder the lock cannot see, one that predates it or never took it, or - // one that took an ephemeral group's port after [`Group::acquire`] - // resolved it. - if shard.index() == 0 { + // one that took an ephemeral group's port after [`bind`] resolved it. + if index == 0 { drop(crate::bind::udp(crate::bind::Udp::new(addr))?); } @@ -649,37 +475,9 @@ fn prepare(dir: std::path::PathBuf) -> Option { /// The largest group the steering filter can address. /// /// It selects with one byte of the connection ID, so a member past 255 could -/// never be named: [`cid`] would have no stride to spend and the filter's -/// `byte % count` could never return that index. +/// never be named: the filter's `byte % count` could never return that index. pub const MAX_SHARDS: u16 = 256; -/// The length of a connection ID that the steering filter recognizes. -pub const CID_LEN: usize = 8; - -/// A connection ID that routes to `shard` without matching a valid SRT data header. -/// -/// `count` values of the byte are reserved per member, so this keeps -/// `256 / count` of its randomness: 32 values across 8 members. Two marker bits -/// are fixed in the fourth byte and every other bit stays random. Connection IDs -/// are not secrets, but they are unlinkable only while they look random, which -/// is why the encoding spends only what routing and classification need. -pub fn cid(shard: Shard) -> [u8; CID_LEN] { - use rand::RngExt; - - let mut cid: [u8; CID_LEN] = rand::random(); - let count = u32::from(shard.count()); - // The number of whole strides of `count` that fit in a byte. At least one, - // because a group is never larger than `MAX_SHARDS`. - let strides = 256 / count; - let stride = rand::rng().random_range(0..strides); - - cid[0] = (stride * count + u32::from(shard.index())) as u8; - // On an SRT data packet this byte overlays the PP/O/KK/R flags. KK=0b11 - // is reserved, so the pattern cannot persist on a valid SRT flow. - cid[3] |= 0x18; - cid -} - /// Attach the steering filter to a socket in a reuseport group of `count`. /// /// Applies to the group, so once is enough, and the last member to bind is the @@ -798,154 +596,69 @@ fn program(count: u16) -> [libc::sock_filter; 18] { mod tests { use super::*; - /// A slot has to exist in the group it names, and the group has to be one - /// the steering filter can address: `cid` divides by the count, so - /// an oversized group would panic on a zero-width stride. - #[test] - fn shard_slots_are_bounded() { - assert_eq!(Shard::new(0, 1).map(|shard| shard.count()), Some(1)); - assert_eq!(Shard::new(3, 4).map(|shard| shard.index()), Some(3)); - assert!(Shard::new(4, 4).is_none()); - assert!(Shard::new(0, 0).is_none()); - assert!(Shard::new(0, MAX_SHARDS).is_some()); - assert!(Shard::new(0, MAX_SHARDS + 1).is_none()); - } + /// The length of the connection IDs a sharded endpoint issues. + const CID_LEN: usize = 8; - /// The group is the only source of slots, and it hands out exactly the ones - /// it was sized for, in order. - #[test] - fn a_group_hands_out_every_slot_in_order() { - const COUNT: u16 = 4; - - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), COUNT).unwrap(); - assert_eq!(group.count(), COUNT); - - for index in 0..COUNT { - let member = group.member().expect("a slot per member"); - assert_eq!(member.shard().index(), index); - assert_eq!(member.shard().count(), COUNT); - } - assert!(group.member().is_none(), "the group cannot be resized"); - } - - /// A group of zero is a group of one: every caller sizes a group from a - /// worker count, and a lone member is a valid group. - #[test] - fn an_empty_group_holds_one_member() { - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), 0).unwrap(); - assert_eq!(group.count(), 1); - assert_eq!(group.member().map(|member| member.shard().count()), Some(1)); + /// A connection ID shaped like a sharded endpoint's (see [the + /// encoding](super#the-encoding)): the first byte randomized within the + /// member's stride, so nothing but the reduction can be steering, and the + /// non-SRT marker set. + fn cid(shard: Shard) -> [u8; CID_LEN] { + use rand::RngExt; + let mut id: [u8; CID_LEN] = rand::random(); + let count = u32::from(shard.count()); + let stride = rand::rng().random_range(0..256 / count); + id[0] = (stride * count + u32::from(shard.index())) as u8; + id[3] |= 0x18; + id } /// The steering filter names a member with one byte, so a group past 256 has - /// members it could never reach. Refused where the group is sized, rather - /// than when the first connection ID has no stride left to spend. + /// members it could never reach. Refused before anything binds. #[test] fn an_unaddressable_group_is_refused() { let addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); - assert!(Group::acquire(addr, MAX_SHARDS).is_ok()); assert!(matches!( - Group::acquire(addr, MAX_SHARDS + 1), + bind(addr, MAX_SHARDS + 1), Err(Error::Count { count: 257, max: 256 }) )); } - /// The encoding's only real requirement: whatever byte a member issues has to - /// reduce back to that member. - #[test] - fn a_prefix_reduces_to_its_own_shard() { - // `MAX_SHARDS` is the interesting end: one stride per member, so the prefix - // is the index itself and there is no randomness left to spend. - for count in [1u16, 2, 3, 4, 7, 8, 16, 64, 255, MAX_SHARDS] { - for index in 0..count { - let shard = Shard::new(index, count).unwrap(); - // Sampled, since the prefix is randomized within the member's stride. - for _ in 0..64 { - let id = cid(shard); - let prefix = id[0]; - assert_eq!( - u16::from(prefix) % count, - index, - "prefix {prefix} of shard {index}/{count} steers elsewhere" - ); - assert_eq!(id[3] & 0x18, 0x18, "CID lacks the non-SRT marker"); - } - } - } - } - - /// Every member of a group has to be reachable, or the ones that aren't sit - /// idle while their share of the traffic goes to a sibling. + /// Elsewhere the kernel cannot steer, so a group is refused at its first + /// member rather than served by a socket that silently balances nothing. #[test] - fn prefixes_cover_every_shard() { - const COUNT: u16 = 8; - let mut seen = std::collections::HashSet::new(); - for index in 0..COUNT { - let shard = Shard::new(index, COUNT).unwrap(); - for _ in 0..256 { - seen.insert(u16::from(cid(shard)[0]) % COUNT); - } - } - assert_eq!(seen.len(), usize::from(COUNT)); + #[cfg(not(target_os = "linux"))] + fn a_group_is_refused_off_linux() { + let err = bind("127.0.0.1:0".parse().unwrap(), 2).expect_err("steered a group off Linux"); + assert!(matches!(err, Error::Bind { index: 0, ref source } if source.kind() == io::ErrorKind::Unsupported)); } - /// An ephemeral request gives the first member a port of the kernel's - /// choosing, and the group is only balanced over a port every member holds. + /// The group hands out exactly the slots it was sized for, in order, every + /// one on the port the first member resolved. #[test] #[cfg(target_os = "linux")] - fn a_group_shares_one_ephemeral_port() { + fn a_group_hands_out_every_slot_on_one_port() { const COUNT: u16 = 3; - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), COUNT).unwrap(); - let claims: Vec = (0..COUNT) - .map(|_| group.member().expect("a slot per member").bind().expect("bind member")) - .collect(); - let mut group = group.complete(claims).expect("complete group"); - let sockets: Vec = (0..COUNT) - .map(|_| group.member().unwrap().expect("a socket per member").into_inner()) - .collect(); - + let mut group = bind("127.0.0.1:0".parse().unwrap(), COUNT).unwrap(); + assert_eq!(group.count(), COUNT); let addr = group.addr(); assert_ne!(addr.port(), 0, "the group holds the port its first member bound"); - for socket in &sockets { - assert_eq!( - socket.local_addr().unwrap(), - addr, - "every member holds the group's port" - ); - } - } - - /// The kernel numbers a group by bind order, so a member that binds out of - /// turn would take a sibling's slot and steer its traffic. Refused instead, - /// which is the one part of the ordering rule a caller can still get wrong - /// once its members are bound on threads of their own. - #[test] - #[cfg(target_os = "linux")] - fn binding_out_of_order_is_refused() { - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), 2).unwrap(); - let first = group.member().expect("first slot"); - let second = group.member().expect("second slot"); - - second.bind().expect_err("the second member cannot bind first"); - // Refused, not deferred: nothing joined the group, so the slot the - // kernel would have handed the second member is still the first's. - first.bind().expect("bind the first member"); - assert_ne!(group.addr().port(), 0); + for index in 0..COUNT { + let member = group.member().unwrap().expect("a socket per member"); + assert_eq!(member.shard(), Shard { index, count: COUNT }); + assert_eq!(member.into_inner().local_addr().unwrap(), addr); + } + assert!(group.member().unwrap().is_none(), "the group cannot be resized"); } - /// A forming group exposes claims, not sockets, and refuses completion until - /// every declared member is bound. + /// A group of zero is a group of one: a lone member is a valid group. #[test] #[cfg(target_os = "linux")] - fn a_partial_group_cannot_serve() { - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), 2).unwrap(); - let claim = group.member().unwrap().bind().expect("bind first member"); - - group - .complete(vec![claim]) - .expect_err("a partial group must not complete"); + fn an_empty_group_holds_one_member() { + let group = bind("127.0.0.1:0".parse().unwrap(), 0).unwrap(); + assert_eq!(group.count(), 1); } /// The whole point, end to end against a real kernel: a packet carrying a @@ -959,12 +672,7 @@ mod tests { fn a_connection_id_reaches_its_own_member() { const COUNT: u16 = 4; - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), COUNT).unwrap(); - let mut claims = Vec::new(); - while let Some(member) = group.member() { - claims.push(member.bind().expect("bind group member")); - } - let mut group = group.complete(claims).expect("complete group"); + let mut group = bind("127.0.0.1:0".parse().unwrap(), COUNT).unwrap(); let mut sockets = Vec::new(); let mut shards = Vec::new(); while let Some(member) = group.member().expect("clone retained socket") { @@ -1021,12 +729,7 @@ mod tests { #[cfg(target_os = "linux")] fn non_quic_flows_stay_on_one_member() { const COUNT: u16 = 4; - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), COUNT).unwrap(); - let mut claims = Vec::new(); - while let Some(member) = group.member() { - claims.push(member.bind().expect("bind group member")); - } - let mut group = group.complete(claims).expect("complete group"); + let mut group = bind("127.0.0.1:0".parse().unwrap(), COUNT).unwrap(); let mut sockets = Vec::new(); while let Some(member) = group.member().expect("clone retained socket") { let socket = member.into_inner(); @@ -1082,12 +785,7 @@ mod tests { #[test] #[cfg(target_os = "linux")] fn dropping_a_serving_handle_does_not_resize_the_group() { - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), 2).unwrap(); - let mut claims = Vec::new(); - while let Some(member) = group.member() { - claims.push(member.bind().expect("bind group member")); - } - let mut group = group.complete(claims).expect("complete group"); + let mut group = bind("127.0.0.1:0".parse().unwrap(), 2).unwrap(); let first = group.member().unwrap().expect("first serving socket"); let second = group.member().unwrap().expect("second serving socket"); let shard = second.shard(); @@ -1107,88 +805,47 @@ mod tests { assert_eq!(socket.recv_from(&mut buf).unwrap().0, packet.len()); } - /// A second group on a named port must lose it while the first is alive: two - /// groups constructing at once would each pass the bind probe before either - /// held the port, then interleave into one group whose filter steers each - /// process's traffic to the other's sockets. + /// A second group on a held port must lose it while the first is alive, for + /// a named port and a resolved ephemeral one alike: two groups constructing + /// at once would each pass the bind probe before either held the port, then + /// interleave into one group whose filter steers each process's traffic to + /// the other's sockets. #[test] #[cfg(target_os = "linux")] fn a_second_group_cannot_take_the_port() { // A port the kernel just handed out and nothing holds any more, which is // as close to a reserved port as a test gets. - let addr = { + let named = { let probe = crate::bind::udp(crate::bind::Udp::new("127.0.0.1:0".parse().unwrap())).unwrap(); probe.local_addr().unwrap() }; - let mut first = Group::acquire(addr, 1).unwrap(); - let claim = first.member().unwrap().bind().expect("bind the first group"); - let first = first.complete(vec![claim]).expect("complete the first group"); - - assert!( - matches!(Group::acquire(addr, 1), Err(Error::Overlap { .. })), - "a second group took a held port" - ); - - // The lock dies with the group, so the port is takeable again. - drop(first); - Group::acquire(addr, 1).expect("the released port must be takeable again"); - } - - /// An ephemeral request is resolved to a concrete port and locked before any - /// member binds. A reuseport bind of port `0` could land on a port another - /// same-UID group holds, and join it. - #[test] - #[cfg(target_os = "linux")] - fn an_ephemeral_group_takes_its_port_up_front() { - let group = Group::acquire("127.0.0.1:0".parse().unwrap(), 1).unwrap(); - let addr = group.addr(); - assert_ne!(addr.port(), 0, "the port is resolved before any member binds"); + for addr in [named, "127.0.0.1:0".parse().unwrap()] { + let first = bind(addr, 1).expect("bind the first group"); + let addr = first.addr(); - assert!( - matches!(Group::acquire(addr, 1), Err(Error::Overlap { .. })), - "a second group took an ephemeral group's port" - ); - } - - /// A same-UID reuseport socket that takes an ephemeral group's port between - /// its resolution and the first member's bind must fail that bind, not be - /// joined by it. - #[test] - #[cfg(target_os = "linux")] - fn an_ephemeral_port_taken_before_binding_is_refused() { - let mut group = Group::acquire("127.0.0.1:0".parse().unwrap(), 1).unwrap(); - let _intruder = crate::bind::udp(crate::bind::Udp::new(group.addr()).with_reuse_port(true)).unwrap(); + assert!( + matches!(bind(addr, 1), Err(Error::Overlap { .. })), + "a second group took a held port" + ); - let err = group.member().unwrap().bind().expect_err("joined the intruder's group"); - assert_eq!(err.kind(), io::ErrorKind::AddrInUse); + // The lock dies with the group, so the port is takeable again. + drop(first); + bind(addr, 1).expect("the released port must be takeable again"); + } } - /// A member outlives the group that handed it out and can still bind, so the - /// exclusion has to outlive the group too. Releasing the port at the group's - /// drop would let a second group probe and interleave into this one while its - /// first member had yet to join. + /// A same-UID reuseport socket already on the address, one the lock cannot + /// see, must fail the first member's bind rather than be joined by it. #[test] #[cfg(target_os = "linux")] - fn an_outstanding_member_keeps_the_port() { - let addr = { - let probe = crate::bind::udp(crate::bind::Udp::new("127.0.0.1:0".parse().unwrap())).unwrap(); - probe.local_addr().unwrap() - }; - - let mut group = Group::acquire(addr, 1).unwrap(); - let member = group.member().expect("the only slot"); - drop(group); - - assert!( - matches!(Group::acquire(addr, 1), Err(Error::Overlap { .. })), - "a second group took a port an unbound member still holds" - ); - - let claim = member.bind().expect("bind the outstanding member"); + fn a_held_address_is_refused_not_joined() { + let intruder = + crate::bind::udp(crate::bind::Udp::new("127.0.0.1:0".parse().unwrap()).with_reuse_port(true)).unwrap(); + let addr = intruder.local_addr().unwrap(); - drop(claim); - Group::acquire(addr, 1).expect("the released port must be takeable again"); + let err = bind(addr, 1).expect_err("joined the intruder's group"); + assert!(matches!(err, Error::Bind { index: 0, ref source } if source.kind() == io::ErrorKind::AddrInUse)); } /// The jumps are offsets from the *following* instruction, so an off-by-one diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs index 16959fd75c..0acf266745 100644 --- a/rs/moq-tokio/src/listen.rs +++ b/rs/moq-tokio/src/listen.rs @@ -292,8 +292,6 @@ impl Default for Config { } } -#[cfg(feature = "quic")] -pub(crate) use moq_sock::shard::Shard; /// One server's socket in a complete `SO_REUSEPORT` group. /// /// Crate-private on purpose: [`crate::worker::Workers`] is the only thing that diff --git a/rs/moq-tokio/src/quic.rs b/rs/moq-tokio/src/quic.rs index 4b28ac64ef..f1e9d9fb73 100644 --- a/rs/moq-tokio/src/quic.rs +++ b/rs/moq-tokio/src/quic.rs @@ -19,7 +19,7 @@ mod backend; #[cfg(feature = "quic")] pub use backend::Error; #[cfg(feature = "quic")] -pub(crate) use backend::{Client, Options, Server, accept}; +pub(crate) use backend::{Client, Closer, Options, Server, accept}; // The async QUIC layer driving `moq-quic` on tokio, forked from quinn (see its module docs). // Public only so the workspace's tests can dial a bare endpoint; not a stable API. diff --git a/rs/moq-tokio/src/quic/backend.rs b/rs/moq-tokio/src/quic/backend.rs index e743af6ea7..1dbb82af28 100644 --- a/rs/moq-tokio/src/quic/backend.rs +++ b/rs/moq-tokio/src/quic/backend.rs @@ -597,15 +597,11 @@ impl Socket { }) } - /// Close the socket, unless a [`crate::server::SocketRetainer`] still holds it. + /// Close the socket. A reuseport member stays in its group regardless: the + /// group retains a handle of its own. fn release(&self) { self.io.write().unwrap().take(); } - - /// A handle keeping the socket open after it is released, or `None` once it is. - fn retain(&self) -> Option> { - self.io.read().unwrap().clone() - } } impl driver::AsyncUdpSocket for Socket { @@ -779,7 +775,11 @@ impl Server { count = shard.count(), "encoding the shard in connection IDs" ); - endpoint_config.cid_generator(Arc::new(move || Box::new(ShardIdGenerator::new(shard)))); + endpoint_config.cid_generator(Arc::new(move || { + let mut cids = moq_quic::RandomConnectionIdGenerator::default(); + cids.set_shard(shard.index(), shard.count()); + Box::new(cids) + })); } else if let Some(load_balancer) = load_balancer { let server_id = load_balancer.id; let nonce_len = load_balancer.nonce; @@ -840,16 +840,27 @@ impl Server { /// Close every connection, wait until each has sent its close to the peer, /// then release the socket. pub async fn shutdown(self) { - self.close(); - // Not `wait_idle`, which also sits out each connection's 3 PTO closing - // period: that only repeats the close to a peer that already has it. - self.quic.wait_all_draining().await; + self.closer().close().await; self.socket.release(); } - /// A handle keeping the socket open after this server is gone. - pub fn retain(&self) -> Option> { - self.socket.retain() + /// A handle that closes this server's connections from wherever its owner + /// went, for a worker that has to flush them before its runtime is dropped. + pub fn closer(&self) -> Closer { + Closer(self.quic.clone()) + } +} + +/// Closes a [`Server`]'s connections without owning it; see [`Server::closer`]. +pub(crate) struct Closer(driver::Endpoint); + +impl Closer { + /// Close every connection and wait until each has sent its close to the peer. + pub async fn close(self) { + self.0.close(driver::VarInt::from_u32(0), b"server shutdown"); + // Not `wait_idle`, which also sits out each connection's 3 PTO closing + // period: that only repeats the close to a peer that already has it. + self.0.wait_all_draining().await; } } @@ -953,34 +964,6 @@ pub(crate) async fn accept( } } -// ── ShardIdGenerator ──────────────────────────────────────────────── - -/// Connection IDs whose first byte names the reuseport member that owns them, -/// so the group's steering filter can route later packets back to it. -struct ShardIdGenerator { - shard: crate::listen::Shard, -} - -impl ShardIdGenerator { - fn new(shard: crate::listen::Shard) -> Self { - Self { shard } - } -} - -impl driver::ConnectionIdGenerator for ShardIdGenerator { - fn generate_cid(&mut self) -> driver::ConnectionId { - driver::ConnectionId::new(&moq_sock::shard::cid(self.shard)) - } - - fn cid_len(&self) -> usize { - moq_sock::shard::CID_LEN - } - - fn cid_lifetime(&self) -> Option { - None - } -} - // ── ServerIdGenerator ─────────────────────────────────────────────── struct ServerIdGenerator { diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 370e5d817f..44bcb1893e 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -238,24 +238,6 @@ pub struct Server { websocket: Option, } -/// A worker member's QUIC socket, kept in the reuseport group after the serving -/// [`Server`] is gone. -/// -/// Dropping or closing a serving server closes its socket, which renumbers the -/// survivors. The worker group holds one of these per member until serving has -/// stopped, so a dropped or finished member leaves the steering intact. The -/// fields are never read: holding the sockets is what keeps them open. -/// -/// Only compiled with a QUIC backend, matching the worker group that is its -/// only caller. -#[derive(Clone, Debug)] -#[allow(dead_code)] -#[cfg(feature = "quic")] -pub(crate) struct SocketRetainer { - #[cfg(feature = "quic")] - quic: Option>, -} - impl Server { /// Build a server from its config, binding the QUIC socket up front. /// @@ -518,19 +500,6 @@ impl Server { crate::tls::Certificates::empty() } - /// Clone this server's QUIC endpoint, keeping its socket in the reuseport - /// group after this server is gone. - /// - /// Crate-private: only the worker group retains sockets this way. A worker - /// member serves the same QUIC backend as the tokio listener. - #[cfg(feature = "quic")] - pub(crate) fn retain(&self) -> SocketRetainer { - SocketRetainer { - #[cfg(feature = "quic")] - quic: self.quic.as_ref().and_then(|server| server.retain()), - } - } - #[cfg(not(any( feature = "quic", feature = "iroh", @@ -819,6 +788,12 @@ impl Server { Err(Error::NoBackend("no QUIC listener configured")) } + /// Closes the QUIC connections once this server has been handed away. + #[cfg(feature = "quic")] + pub(crate) fn quic_closer(&self) -> Option { + self.quic.as_ref().map(crate::quic::Server::closer) + } + /// The address the WebSocket listener from [`Config::websocket`] bound to, if one was set. #[cfg(feature = "websocket")] pub fn websocket_local_addr(&self) -> Option { diff --git a/rs/moq-tokio/src/worker/group.rs b/rs/moq-tokio/src/worker/group.rs index c98642d752..5e646aa3a0 100644 --- a/rs/moq-tokio/src/worker/group.rs +++ b/rs/moq-tokio/src/worker/group.rs @@ -6,7 +6,7 @@ use std::sync::{ }; use super::Config; -use crate::{Error, Result, Server, abort::AbortOnDrop, listen::Socket as ShardSocket, server::SocketRetainer}; +use crate::{Error, Result, Server, abort::AbortOnDrop, listen::Socket as ShardSocket}; /// A bound group of QUIC workers sharing one port. /// @@ -91,20 +91,16 @@ impl Workers { // The group owns everything a reuseport group has to get right: it takes // the port before the first member binds and holds it until the group is // dropped, refuses a size the steering filter could not address, and - // hands out one member per slot in the order the kernel numbers them by. - let mut forming = moq_sock::shard::Group::acquire(requested, config.count).map_err(|err| match err { + // binds one member per slot in the order the kernel numbers them by. + let mut group = moq_sock::shard::bind(requested, config.count).map_err(|err| match err { moq_sock::shard::Error::Count { count, max } => Error::WorkerCount { count, max }, - moq_sock::shard::Error::Resolve(err) => crate::quic::Error::BindSocket(err).into(), - // The port lock is the only other way to lose the address, and it is - // held by exactly one thing: another group of this UID. - _ => Error::WorkerOverlap { addr: requested }, + moq_sock::shard::Error::Overlap { .. } => Error::WorkerOverlap { addr: requested }, + moq_sock::shard::Error::Resolve(err) + | moq_sock::shard::Error::Bind { source: err, .. } + | moq_sock::shard::Error::Steer(err) => crate::quic::Error::BindSocket(err).into(), + err => crate::quic::Error::BindSocket(std::io::Error::other(err)).into(), })?; - let count = forming.count(); - let mut claims = Vec::with_capacity(count as usize); - while let Some(member) = forming.member() { - claims.push(member.bind().map_err(crate::quic::Error::BindSocket)?); - } - let mut group = forming.complete(claims).map_err(crate::quic::Error::BindSocket)?; + let count = group.count(); let shared = Arc::new(Shared::default()); @@ -178,21 +174,6 @@ impl Workers { None => Arc::new(Shared::default()), }; - // Retain every socket before handing any server out: dropping a server - // afterwards closes its endpoint, but the retainer keeps the socket in - // the group, so the kernel never renumbers a survivor. - let mut retainers = Vec::with_capacity(self.workers.len()); - for worker in &self.workers { - retainers.push( - worker - .server - .as_ref() - .expect("a bound worker holds its server before the split") - .retain(), - ); - } - *shared.retainers.lock().unwrap() = retainers; - Group { workers: self.workers, certificates: self.certificates, @@ -426,16 +407,14 @@ fn join(threads: Vec<(u16, std::thread::JoinHandle<()>)>) { } } -/// What the workers share: the shutdown signal and the retained sockets. +/// What the workers share: the shutdown signal. /// /// One per group, held by the group and borrowed by every spawner. Cancelling -/// it stops every worker thread; the retainers keep every socket in the -/// reuseport group until the group is gone. +/// it stops every worker thread. #[derive(Debug, Default)] struct Shared { shutdown: AtomicBool, notify: tokio::sync::Notify, - retainers: std::sync::Mutex>, } impl Shared { @@ -774,6 +753,7 @@ fn run( return; } }; + let closer = server.quic_closer(); if ready .send(Ok(Ready { @@ -799,6 +779,13 @@ fn run( Some(spawn) = spawn.recv() => spawn(), } } + + // Dropping the runtime drops the endpoint driver with it, so every + // connection's close has to be sent first. Otherwise a peer only learns + // the session is gone at its idle timeout. + if let Some(closer) = closer { + closer.close().await; + } }); tracing::debug!(index, "QUIC worker stopped"); } diff --git a/rs/moq-tokio/src/worker/mod.rs b/rs/moq-tokio/src/worker/mod.rs index 1cb9991806..b85962f8ae 100644 --- a/rs/moq-tokio/src/worker/mod.rs +++ b/rs/moq-tokio/src/worker/mod.rs @@ -2,9 +2,9 @@ //! //! A `Server` on a work-stealing runtime serves every connection off one UDP //! socket: every packet can cross threads, and every wakeup is a candidate -//! context switch. `Workers` is the opposite shape. Each -//! member is a thread of its own, pinned to a core, running a `current_thread` -//! runtime and owning one socket in a `SO_REUSEPORT` group. A connection lands +//! context switch. `Workers` is the opposite shape. Each member is a thread of +//! its own, optionally pinned to a core, running a `current_thread` runtime and +//! owning one socket in a `SO_REUSEPORT` group. A connection lands //! on one worker and stays there, so the locks its driver and its session take //! are uncontended and nothing is stolen. //! @@ -33,22 +33,22 @@ pub struct Config { /// How many workers to run, each with a thread and a socket of its own. pub count: u16, - /// Pin each worker to a CPU core. + /// Pin each worker to a CPU core, defaulting to off. /// /// The mode's measured win comes from each worker owning a socket and a /// runtime, not from pinning: on a single-socket machine, pinned and unpinned /// benchmark inside run-to-run noise of each other. Pinning stops the - /// scheduler migrating a busy worker, which should matter on a multi-socket - /// or NUMA machine, and costs nothing elsewhere, so it defaults on. Turn it - /// off when sharing the machine with something that manages CPU placement - /// itself. + /// scheduler migrating a busy worker, which may matter on a multi-socket or + /// NUMA machine with the cores to itself. Every group pins worker *i* to the + /// *i*th core, so two groups on a host, or one limited to fewer CPUs than the + /// host has, would crowd onto the first cores. pub pin: bool, } impl Config { - /// `count` workers, pinned. + /// `count` workers, unpinned. pub fn new(count: u16) -> Self { - Self { count, pin: true } + Self { count, pin: false } } /// Whether to pin each worker to a core. diff --git a/rs/moq-tokio/tests/worker.rs b/rs/moq-tokio/tests/worker.rs index 22385c0f89..ba5df60c6f 100644 --- a/rs/moq-tokio/tests/worker.rs +++ b/rs/moq-tokio/tests/worker.rs @@ -139,6 +139,52 @@ async fn dropping_the_workers_releases_the_port() { assert!(owned.is_disjoint(&open_sockets()), "workers left the port bound"); } +/// Stopping a group closes its sessions on the wire. The endpoint driver dies +/// with the worker's runtime, so a close still queued at that point never +/// leaves, and the peer only notices at its idle timeout. +#[tokio::test] +async fn stopping_the_group_closes_its_sessions() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let dir = tempfile::tempdir().unwrap(); + let (cert, key) = certificate(dir.path()); + let workers = bind_workers(listen_config(&cert, &key, 0), Default::default(), config(WORKERS)).unwrap(); + let addr = workers.local_addr(); + let fingerprints = workers.certificates().fingerprints(); + let mut group = workers.split(); + for member in group.members() { + member.serve(|server| async move { + let mut listener = server.listen().await.unwrap(); + while let Some(request) = listener.accept().await { + tokio::spawn(async move { + let (_session, _driver) = request.ok().await.unwrap(); + std::future::pending::<()>().await; + }); + } + }); + } + + let mut connect = moq_tokio::connect::Config::default(); + connect.tls.fingerprint = fingerprints; + connect.once = Some(true); + #[cfg(feature = "websocket")] + { + connect.websocket.enabled = Some(false); + } + // Far past the wait below, so only a close on the wire can end the session. + let mut quic = moq_tokio::quic::Config::default(); + quic.idle_timeout = Some(std::time::Duration::from_secs(600)); + quic.keep_alive = None; + let client = connect.init(quic).unwrap(); + let url: url::Url = format!("https://{addr}/").parse().unwrap(); + let session = client.connect(url).established().await.expect("connect"); + + group.shutdown().await; + tokio::time::timeout(std::time::Duration::from_secs(5), session.closed()) + .await + .expect("the session must see the close") + .expect_err("a stopped server ends the session"); +} + /// The future factory runs on the worker, so the future may hold local state /// across an await without making that state thread-safe, and spawn more of it /// onto the same task set. diff --git a/rs/moq-uring/Cargo.toml b/rs/moq-uring/Cargo.toml index d727a20222..6b470d8204 100644 --- a/rs/moq-uring/Cargo.toml +++ b/rs/moq-uring/Cargo.toml @@ -32,7 +32,6 @@ moq-clock = { workspace = true, optional = true } moq-net = { workspace = true } moq-quic = { workspace = true, features = ["bloom", "rustls-aws-lc-rs"], optional = true } moq-sock = { workspace = true } -rand = "0.10" rustc-hash = { workspace = true } rustls = { version = "0.23", optional = true } # OS-native server-cert verifier for the rustls backends. Version matched to diff --git a/rs/moq-uring/src/quic/driver/mod.rs b/rs/moq-uring/src/quic/driver/mod.rs index 30e53a05a3..c0a2d1bbe5 100644 --- a/rs/moq-uring/src/quic/driver/mod.rs +++ b/rs/moq-uring/src/quic/driver/mod.rs @@ -27,7 +27,7 @@ use rustls::pki_types::{CertificateDer, PrivateKeyDer}; #[cfg(feature = "qlog")] use super::qlog; -use super::{Congestion, Error, Identity, SEGMENT, Transport, client, endpoint::CID_LEN, server}; +use super::{Congestion, Error, Identity, SEGMENT, Transport, client, server}; use crate::udp; /// Per-stream flow control credit. @@ -104,37 +104,22 @@ pub(crate) fn endpoint_config(shard: Option) -> Result, -} - -impl moq_quic::ConnectionIdGenerator for Cids { - fn generate_cid(&mut self) -> moq_quic::ConnectionId { - moq_quic::ConnectionId::new(&super::endpoint::cid(self.shard)) - } - - fn cid_len(&self) -> usize { - CID_LEN - } - - fn cid_lifetime(&self) -> Option { - None - } -} - /// Dial as `config` says. pub(crate) fn client_config(config: &client::Config) -> Result { let provider = provider(); diff --git a/rs/moq-uring/src/quic/endpoint.rs b/rs/moq-uring/src/quic/endpoint.rs index 82daddcae2..6842a87b27 100644 --- a/rs/moq-uring/src/quic/endpoint.rs +++ b/rs/moq-uring/src/quic/endpoint.rs @@ -7,18 +7,18 @@ //! cluster dials at once. A single demux task receives; each connection keeps //! a driver task of its own for timers and egress. //! -//! Every connection id this endpoint issues is [`CID_LEN`] bytes, which is -//! what lets a short header (which does not encode the id's length) be -//! parsed at all. Ids rotate as peers consume them (`NEW_CONNECTION_ID`), so -//! a migrating client stays routable; an Initial for an unsupported version -//! gets a version negotiation packet back. +//! Every connection id this endpoint issues is 8 random bytes, fixed-length +//! because a short header does not encode the id's length. Ids +//! rotate as peers consume them (`NEW_CONNECTION_ID`), so a migrating client +//! stays routable; an Initial for an unsupported version gets a version +//! negotiation packet back. //! //! The socket is the endpoint's identity. Its worker is where the demux and //! every connection driver run, whichever thread built the endpoint. A socket //! adopted as a member of a steered `SO_REUSEPORT` group (one worker per core //! on one port; see [`udp::Bound`](crate::udp::Bound)) brings its slot along, -//! and every id the endpoint issues then carries the -//! [`moq_sock::shard::cid`] steering marker, so the kernel keeps +//! and every id the endpoint issues then names that shard +//! (`moq_quic::RandomConnectionIdGenerator::set_shard`), so the kernel keeps //! delivering a connection's packets to the worker that owns it. Dials //! through the endpoint carry it too, which is what steers a cluster peer's //! responses back to the dialing worker. @@ -27,11 +27,6 @@ use super::server; pub use super::driver::Endpoint; -/// Every connection id this endpoint issues is this long: long enough to be -/// unguessable per socket, and fixed because a short header does not encode -/// the id's length, so parsing one assumes it. -pub const CID_LEN: usize = moq_sock::shard::CID_LEN; - /// What an endpoint serves, beyond dialing. #[derive(Clone, Debug)] #[non_exhaustive] @@ -72,9 +67,3 @@ impl Config { self } } - -/// A fresh [`CID_LEN`]-byte connection id, leading with the steering prefix -/// when the endpoint's socket is a member of a reuseport group. -pub(crate) fn cid(shard: Option) -> [u8; CID_LEN] { - shard.map_or_else(rand::random, moq_sock::shard::cid) -} diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index a090ca00db..6f0fe55a74 100644 --- a/rs/moq-uring/src/udp.rs +++ b/rs/moq-uring/src/udp.rs @@ -388,10 +388,10 @@ impl TxSlot { pub enum Bound { /// A socket on its own, bound by the caller. Lone(UdpSocket), - /// One member of a completed steered `SO_REUSEPORT` group. Every - /// connection id an endpoint on it issues then leads with the member's - /// [`cid`](moq_sock::shard::cid), so the group's filter - /// keeps delivering a connection's packets to this socket. + /// One member of a steered `SO_REUSEPORT` group. Every connection id an + /// endpoint on it issues then names the member's + /// [`Shard`](moq_sock::shard::Shard), so the group's filter keeps + /// delivering a connection's packets to this socket. Member(moq_sock::shard::Socket), } diff --git a/rs/moq-uring/tests/workers.rs b/rs/moq-uring/tests/workers.rs index e427de24b4..8d3ee8f4e5 100644 --- a/rs/moq-uring/tests/workers.rs +++ b/rs/moq-uring/tests/workers.rs @@ -19,7 +19,6 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::task::Poll; -use moq_sock::shard::Group; use moq_uring::{Config, Error, Worker, quic, udp}; fn worker() -> Option { @@ -79,12 +78,7 @@ fn a_steered_group_serves_a_shared_port() { // kernel steers by, and binding before any thread spawns is what // guarantees it. The group is held for as long as the sockets are served, // since it is what holds the port. - let mut group = Group::acquire("127.0.0.1:0".parse().expect("addr"), WORKERS).expect("group"); - let mut claims = Vec::new(); - while let Some(member) = group.member() { - claims.push(member.bind().expect("bind group member")); - } - let mut group = group.complete(claims).expect("complete group"); + let mut group = moq_sock::shard::bind("127.0.0.1:0".parse().expect("addr"), WORKERS).expect("group"); let mut members = Vec::new(); while let Some(member) = group.member().expect("clone retained socket") { members.push(member);