feat(relay): serve QUIC from pinned per-core workers (M1, part 1)

doc/bin/relay/config.md

@@ -45,6 +45,37 @@ Logging configuration.
 level = "info"
 ```
 
+### \[runtime]
+
+How the relay lays its QUIC work out over threads. By default one work-stealing
+runtime serves every connection off one UDP socket.
+
+```toml
+[runtime]
+# Serve QUIC from this many single-threaded workers instead of the shared
+# runtime. Each worker is a thread with its own socket on the listen address
+# (SO_REUSEPORT), and a connection is handled start to finish by one worker, so
+# its packets never cross threads. Everything else (HTTP, WebSocket, tcp/unix,
+# clustering) stays on the shared runtime. Omit to keep QUIC there too.
+#
+# Linux only, and incompatible with listen.tls.generate: each worker would
+# generate a certificate of its own. Point at real certificate files instead.
+#
+# The kernel steers a packet to a worker by hashing its source address and port,
+# so a client that changes address mid-connection (a NAT rebinding, a network
+# change) lands on a worker that has never seen it and loses the connection.
+# Treat this as a benchmarking knob until steering follows the connection ID.
+workers = 8
+
+# Pin each worker to a CPU core. Default: true.
+pin = true
+```
+
+The shared runtime is still there for everything that is not QUIC, and it still
+sizes its thread pool to the machine. Set `TOKIO_WORKER_THREADS` in the
+environment to bound it, so the workers are not competing with a full second
+pool for the same cores.
+
 ### \[listen]
 
 QUIC/WebTransport server settings. Optionally add plaintext qmux stream

rs/moq-relay/Cargo.toml

@@ -47,6 +47,7 @@ axum-server = { version = "0.8", features = ["tls-rustls"] }
 bytes = "1"
 bytesize = "2.4.2"
 clap = { version = "4", features = ["derive"] }
+core_affinity = "0.8"
 futures = "0.3"
 http-body = "1"
 http-cache-reqwest = { version = "1.0.0-alpha.6", features = ["manager-moka", "url-standard"], default-features = false }

rs/moq-relay/src/config.rs

@@ -35,6 +35,12 @@ pub struct Config {
 	#[serde(default)]
 	pub log: moq_tokio::Log,
 
+	/// How QUIC work is laid out over threads. One shared runtime unless
+	/// `runtime.workers` is set.
+	#[command(flatten)]
+	#[serde(default)]
+	pub runtime: crate::RuntimeConfig,
+
 	/// Cluster configuration.
 	#[command(flatten)]
 	#[serde(default)]
@@ -369,6 +375,31 @@ depth = 2
 	/// Regression test for the clap+TOML clobber bug applied to `[cache]`. Both
 	/// fields are `Option<String>` so a TOML-configured cache size survives the
 	/// CLI re-parse when no `--cache-*` flag is passed.
+	#[test]
+	fn cli_does_not_clobber_toml_runtime() {
+		let _env = EnvGuard::clear(&["MOQ_RUNTIME_WORKERS", "MOQ_RUNTIME_PIN"]);
+
+		let toml = r#"
+[runtime]
+workers = 8
+pin = false
+"#;
+		let dir = std::env::temp_dir().join("moq-relay-config-test");
+		std::fs::create_dir_all(&dir).unwrap();
+		let path = dir.join("runtime-toml-wins.toml");
+		std::fs::write(&path, toml).unwrap();
+
+		let args = vec![std::ffi::OsString::from("moq-relay"), std::ffi::OsString::from(&path)];
+		let config = Config::parse_and_merge(args).expect("config load");
+
+		assert_eq!(config.runtime.workers, Some(8));
+		assert_eq!(
+			config.runtime.pin,
+			Some(false),
+			"TOML's runtime.pin=false must survive the CLI re-parse"
+		);
+	}
+
 	#[test]
 	fn cli_does_not_clobber_toml_cache() {
 		let _env = EnvGuard::clear(&["MOQ_CACHE_CAPACITY", "MOQ_CACHE_HEADROOM", "MOQ_CACHE_DURATION"]);

rs/moq-relay/src/lib.rs

@@ -20,6 +20,7 @@ mod internal;
 mod listener;
 mod nodes;
 mod relay;
+mod runtime;
 mod shutdown;
 mod stats;
 #[cfg(test)]
@@ -50,6 +51,7 @@ pub use config::*;
 pub use connection::*;
 pub use internal::*;
 pub use relay::*;
+pub use runtime::*;
 pub use shutdown::*;
 pub use stats::*;
 pub use web::*;

rs/moq-relay/src/relay.rs

@@ -24,7 +24,9 @@
 
 use anyhow::Context;
 
-use crate::{Auth, Cluster, Config, Connection, DEFAULT_DRAIN_TIMEOUT, Internal, Shutdown, ShutdownTrigger, Web};
+use crate::{
+	Auth, Cluster, Config, Connection, DEFAULT_DRAIN_TIMEOUT, Internal, Shutdown, ShutdownTrigger, Web, Workers,
+};
 
 /// A fully assembled relay: the listeners and the shared cluster behind them.
 ///
@@ -35,8 +37,10 @@ use crate::{Auth, Cluster, Config, Connection, DEFAULT_DRAIN_TIMEOUT, Internal,
 ///
 /// `#[non_exhaustive]`, so destructure with a trailing `..` (a pattern naming
 /// every field does not compile outside this crate) or move out the fields you
-/// need one at a time. Dropping any field you do not name is safe: the pieces
-/// that must outlive setup are owned by [`Self::cluster`].
+/// need one at a time. Dropping a field you do not name is safe for all but
+/// [`Self::workers`]: the rest of what must outlive setup is owned by
+/// [`Self::cluster`], while the workers own their threads and sockets, so
+/// dropping them un-binds QUIC.
 #[non_exhaustive]
 pub struct Relay {
 	/// The QUIC/WebTransport server, already bound. Feed it to [`serve`].
@@ -73,6 +77,11 @@ pub struct Relay {
 
 	/// Starts graceful shutdown for [`Self::shutdown`].
 	pub shutdown_trigger: ShutdownTrigger,
+
+	/// The thread-per-core QUIC workers, already bound. Drive them with
+	/// [`Workers::serve`]; inert unless `runtime.workers` is configured, in which
+	/// case [`Self::server`] carries no QUIC listener of its own.
+	pub workers: Workers,
 }
 
 impl Relay {
@@ -89,15 +98,27 @@ impl Relay {
 		let mtls_enabled = !config.listen.tls.root.is_empty();
 		let server_versions = config.listen.versions();
 
+		// Bind the QUIC workers first: they own the listen address when configured,
+		// so the server below must not also try to bind it.
+		let workers = Workers::bind(&config).context("failed to start the QUIC workers")?;
+
+		let mut listen = config.listen.clone();
+		if workers.enabled() {
+			listen.listeners = moq_tokio::listen::Listeners::Stream;
+		}
+
 		#[allow(unused_mut)]
-		let mut server = config.listen.init(config.quic.clone())?;
+		let mut server = listen.init(config.quic.clone())?;
 		let client = config.connect.clone().init(config.quic.clone())?;
 
 		// `None` for a stream-only server (no QUIC); any other error is real.
-		let addr = match server.local_addr() {
-			Ok(addr) => Some(addr),
-			Err(moq_tokio::Error::NoBackend(_)) => None,
-			Err(err) => return Err(err).context("failed to resolve the QUIC bind address"),
+		let addr = match workers.local_addr() {
+			Some(addr) => Some(addr),
+			None => match server.local_addr() {
+				Ok(addr) => Some(addr),
+				Err(moq_tokio::Error::NoBackend(_)) => None,
+				Err(err) => return Err(err).context("failed to resolve the QUIC bind address"),
+			},
 		};
 
 		#[cfg(feature = "iroh")]
@@ -140,7 +161,9 @@ impl Relay {
 		let drain_timeout = config.drain_timeout.unwrap_or(DEFAULT_DRAIN_TIMEOUT);
 		let (shutdown_trigger, shutdown) = Shutdown::new(drain_timeout);
 		// Create a web server too. mTLS for HTTPS is opt-in via `--web-https-root`.
-		let web = Web::new(auth.clone(), cluster.clone(), server.certificates(), config.web)
+		// The workers hold the certificates when they own QUIC; the server has none.
+		let certificates = workers.certificates().unwrap_or_else(|| server.certificates());
+		let web = Web::new(auth.clone(), cluster.clone(), certificates, config.web)
 			.with_shutdown(shutdown.clone())
 			.with_versions(server_versions);
 
@@ -170,6 +193,7 @@ impl Relay {
 			addr,
 			shutdown,
 			shutdown_trigger,
+			workers,
 		})
 	}
 
@@ -187,6 +211,7 @@ impl Relay {
 			web,
 			shutdown,
 			shutdown_trigger,
+			workers,
 			..
 		} = self;
 
@@ -209,7 +234,8 @@ impl Relay {
 			Err(err) = started.run() => Err(err).context("cluster failed"),
 			Err(err) = web.run() => Err(err).context("web server failed"),
 			Err(err) = internal.run() => Err(err).context("internal server failed"),
-			Err(err) = serve(server, cluster, auth, shutdown.clone()) => Err(err).context("server failed"),
+			Err(err) = serve(server, cluster.clone(), auth.clone(), shutdown.clone()) => Err(err).context("server failed"),
+			Err(err) = workers.serve(cluster, auth, shutdown.clone()) => Err(err).context("QUIC workers failed"),
 			Err(err) = jemalloc => Err(err).context("jemalloc profiler failed"),
 			res = drain_on_signal(shutdown_trigger, shutdown.drain_timeout) => res,
 			else => Ok(()),