diff --git a/.config/nextest.toml b/.config/nextest.toml index 65691871bd..1d52e0c2ef 100644 --- a/.config/nextest.toml +++ b/.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] diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 6f8677ac36..f72a4fb646 100644 --- a/doc/bin/relay/config.md +++ b/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] diff --git a/rs/moq-uring/src/error.rs b/rs/moq-uring/src/error.rs index d2791cc1fb..b4a43aa663 100644 --- a/rs/moq-uring/src/error.rs +++ b/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}"); + } +} diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index 9ca5de176a..41d5b4ce94 100644 --- a/rs/moq-uring/src/udp.rs +++ b/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() { diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index 3a85fb00ec..45142d2823 100644 --- a/rs/moq-uring/src/worker.rs +++ b/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 diff --git a/rs/moq-uring/tests/identity.rs b/rs/moq-uring/tests/identity.rs index 7df4aedb87..76662b0396 100644 --- a/rs/moq-uring/tests/identity.rs +++ b/rs/moq-uring/tests/identity.rs @@ -115,10 +115,14 @@ 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"); @@ -126,6 +130,7 @@ fn an_endpoint_runs_on_the_worker_that_adopted_its_socket() { }) .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. diff --git a/rs/moq-uring/tests/teardown.rs b/rs/moq-uring/tests/teardown.rs index b0bde779f5..c04d58ecb8 100644 --- a/rs/moq-uring/tests/teardown.rs +++ b/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,6 +115,7 @@ 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"); @@ -120,6 +123,7 @@ fn a_published_close_has_already_left_the_client() { }) .expect("server worker") }); + started.recv().expect("the server thread failed to start"); let handle = client_worker.handle(); let sock = handle diff --git a/rs/moq-uring/tests/workers.rs b/rs/moq-uring/tests/workers.rs index 7cd4ae95ce..f3329d9dd2 100644 --- a/rs/moq-uring/tests/workers.rs +++ b/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> = (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).