fix(uring): bound io_uring tests by the shared memlock budget

.config/nextest.toml

@@ -24,6 +24,18 @@ retries = 0
 # workload once, and the large origin fan-outs take ~20s each for no signal.
 default-filter = "not kind(bench)"
 
+# Every io_uring ring is charged to the user's `RLIMIT_MEMLOCK` (8 MiB by
+# default), a budget shared with every other process that user runs, and
+# desktop apps built on libuv already spend most of it. A worker's ring is
+# ~84 KiB, so a parallel run of tests holding a few workers each exhausts it
+# and fails with `ENOMEM`. Four at a time stays near 1 MiB.
+[[profile.default.overrides]]
+filter = "package(moq-uring) | binary_id(moq-relay::runtime_uring) | (binary_id(moq-relay::embed) & test(/uring/))"
+test-group = "io-uring"
+
+[test-groups]
+io-uring = { max-threads = 4 }
+
 # CI has noisier neighbours and cold caches, so give a test longer before
 # calling it wedged.
 [profile.ci]

doc/bin/relay/config.md

@@ -90,7 +90,10 @@ and refuses to start anywhere it cannot deliver. `[quic]` applies either way,
 except that `mtu_discovery` (its datagram path sends a fixed payload) and the
 three flow-control windows (these workers run fixed ones) are refused under
 `io_uring` rather than quietly ignored. Each worker reports its own counters at
-[`/metrics`](/bin/relay/http#get-metrics).
+[`/metrics`](/bin/relay/http#get-metrics). The kernel charges each worker's
+ring (~100 KiB, plus a page per socket) to `RLIMIT_MEMLOCK`, a budget shared by
+every io\_uring the user runs; raise it (`LimitMEMLOCK=` under systemd) if
+workers fail to start with a message naming that limit.
 
 ## \[web]
 

rs/moq-uring/src/error.rs

@@ -12,3 +12,47 @@ pub enum Error {
 	#[error("io error: {0}")]
 	Io(#[from] std::io::Error),
 }
+
+impl Error {
+	/// A failed ring setup or registration, naming the locked-memory limit when
+	/// that is what ran out.
+	///
+	/// The kernel charges every ring and provided-buffer ring to the user's
+	/// `RLIMIT_MEMLOCK`, shared with every other io_uring that user runs, so a
+	/// bare `ENOMEM` points at the wrong culprit.
+	pub(crate) fn ring(err: std::io::Error) -> Self {
+		if err.raw_os_error() != Some(libc::ENOMEM) {
+			return Self::Io(err);
+		}
+		let mut limit = libc::rlimit {
+			rlim_cur: 0,
+			rlim_max: 0,
+		};
+		// SAFETY: valid out-pointer.
+		let limit = match unsafe { libc::getrlimit(libc::RLIMIT_MEMLOCK, &mut limit) } {
+			0 => format!("{} KiB", limit.rlim_cur / 1024),
+			_ => "unknown".into(),
+		};
+		Self::Io(std::io::Error::new(
+			err.kind(),
+			format!(
+				"{err}: io_uring memory counts against RLIMIT_MEMLOCK ({limit}), shared by every process of this user; \
+				 raise it with `ulimit -l` or systemd's `LimitMEMLOCK=`"
+			),
+		))
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn a_ring_enomem_names_the_memlock_limit() {
+		let err = Error::ring(std::io::Error::from_raw_os_error(libc::ENOMEM));
+		assert!(err.to_string().contains("RLIMIT_MEMLOCK"), "{err}");
+
+		let err = Error::ring(std::io::Error::from_raw_os_error(libc::EBADF));
+		assert!(!err.to_string().contains("RLIMIT_MEMLOCK"), "{err}");
+	}
+}

rs/moq-uring/src/udp.rs

@@ -576,7 +576,8 @@ impl Socket {
 				unsafe {
 					io_ring
 						.submitter()
-						.register_buf_ring_with_flags(ring.ptr.as_ptr() as u64, rx_cap, bgid, 0)?;
+						.register_buf_ring_with_flags(ring.ptr.as_ptr() as u64, rx_cap, bgid, 0)
+						.map_err(Error::ring)?;
 				}
 			}
 			for (bid, buf) in bufs.iter_mut().enumerate() {

rs/moq-uring/src/worker.rs

@@ -118,7 +118,7 @@ impl Worker {
 						kernel_release()
 					))
 				}
-				_ => Error::Io(err),
+				_ => Error::ring(err),
 			})?;
 
 		// One feature bit gates the whole floor: MIN_TIMEOUT landed in 6.12

rs/moq-uring/tests/identity.rs

@@ -115,17 +115,22 @@ fn an_endpoint_runs_on_the_worker_that_adopted_its_socket() {
 	// A client on its own thread and worker, dialing the endpoint. Its
 	// Initial reaches the socket immediately; only the owner can read it.
 	// The client is driven until the server closes on it: its side of the
-	// handshake completes before the server's, so it cannot stop earlier.
+	// handshake completes before the server's, so it cannot stop earlier. It
+	// reports in before dialing, so a setup failure fails here instead of as
+	// an accept that never arrives.
+	let (ready, started) = std::sync::mpsc::channel();
 	let client = std::thread::spawn(move || {
 		let mut worker = Worker::new(Config::default()).expect("client worker");
 		let sock = socket(&worker.handle());
+		ready.send(()).expect("test alive");
 		worker
 			.block_on(async move {
 				let mut conn = quic::client::connect(sock, &dial_config(addr)).await.expect("dial");
 				std::future::poll_fn(|cx| web_transport_trait::poll::Session::poll_closed(&mut conn, cx)).await
 			})
 			.expect("client loop")
 	});
+	started.recv().expect("the client thread failed to start");
 
 	// Driving the bystander polls the accept from its loop, but the demux
 	// that would feed it is a task on the owner, which is not running.

rs/moq-uring/tests/teardown.rs

@@ -103,7 +103,9 @@ fn a_published_close_has_already_left_the_client() {
 	let server_addr = server_sock.local_addr().expect("server addr");
 
 	// The server outlives the client's teardown, so its verdict is only about
-	// what the client managed to send.
+	// what the client managed to send. It reports in once serving, so a setup
+	// failure fails here instead of as a dial that idles out.
+	let (ready, started) = std::sync::mpsc::channel();
 	let server = std::thread::spawn(move || -> quic::Error {
 		let mut worker = Worker::new(Config::default()).expect("server worker");
 		let handle = worker.handle();
@@ -113,13 +115,15 @@ fn a_published_close_has_already_left_the_client() {
 			quic::endpoint::Config::default().with_server(server_config(&certs)),
 		)
 		.expect("endpoint");
+		ready.send(()).expect("test alive");
 		worker
 			.block_on(async move {
 				let mut conn = endpoint.accept().await.expect("accept");
 				std::future::poll_fn(|cx| conn.poll_closed(cx)).await
 			})
 			.expect("server worker")
 	});
+	started.recv().expect("the server thread failed to start");
 
 	let handle = client_worker.handle();
 	let sock = handle

rs/moq-uring/tests/workers.rs

@@ -96,6 +96,9 @@ fn a_steered_group_serves_a_shared_port() {
 	// One stop each: the wakers are per-thread, so a shared slot would let one
 	// worker's registration clobber the other's.
 	let stops: Vec<Arc<Stop>> = (0..WORKERS).map(|_| Arc::new(Stop::default())).collect();
+	// Each worker reports in once serving, so a setup failure fails here
+	// instead of as the dials hashed to it idling out.
+	let (ready, started) = std::sync::mpsc::channel();
 
 	let threads: Vec<_> = members
 		.into_iter()
@@ -105,6 +108,7 @@ fn a_steered_group_serves_a_shared_port() {
 			let stop = stop.clone();
 			let cert = certs.cert.clone();
 			let key = certs.key.clone();
+			let ready = ready.clone();
 			std::thread::spawn(move || {
 				let shard = member.shard();
 				let mut worker = Worker::new(Config::default()).expect("worker");
@@ -125,10 +129,15 @@ fn a_steered_group_serves_a_shared_port() {
 						accepted[usize::from(shard.index())].fetch_add(1, Ordering::AcqRel);
 					}
 				});
+				ready.send(()).expect("test alive");
 				worker.block_on(stop.wait()).expect("worker loop");
 			})
 		})
 		.collect();
+	drop(ready);
+	for _ in 0..WORKERS {
+		started.recv().expect("a worker thread failed to start");
+	}
 
 	// Dial the shared port repeatedly from one client worker. Every handshake
 	// completing is the steering assertion (see the module docs).