From 1a74375a7e53e57f59612f71a5bd0198433d6e28 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 13:24:56 +0200 Subject: [PATCH 1/7] Name the behaviour a reuse-port request asks the kernel for SO_REUSEPORT is spelled the same on Linux and the BSDs and does not mean the same thing. On Linux it permits the duplicate bind and distributes incoming connections across the listeners holding the address. On macOS and the other BSDs it permits the duplicate bind and then gives new connections to whichever socket bound last. A bool cannot express that, so ListenOpts::reuse_port and UdpOpts::reuse_port are now a ReusePort enum: No, Share (duplicate binding, delivery unspecified) and Distribute (the kernel spreads connections across listeners). Distribute is SO_REUSEPORT on Linux and Android, SO_REUSEPORT_LB on FreeBSD 12.0+, and Unsupported everywhere else - refused when the listener is created rather than answered with plain SO_REUSEPORT, which is what produced a listener that is never given any work. Measured on macOS 15: two loops sharing one port under SO_REUSEPORT split 32 connections [0, 32]. AF_UNIX listeners now report Unsupported rather than InvalidInput, matching the rule the rest of the option surface follows. Contract tests: - reuse_port_share asserts the duplicate bind works and that every connection is accepted by someone, and deliberately asserts nothing about which listener, because Share promises nothing about it. - reuse_port_distribute asserts there is no third outcome: a platform either distributes or refuses the listener. - reuse_port_refused covers Windows and WASI, which have neither. - handoff_accept_exactly_once and kernel_accept_exactly_once serve 64 connections across 4 loops on 4 threads by each route and assert every connection was served exactly once, on loops whose handle ceiling is low enough that a leaked handle exhausts it. --- crates/turnloop-contract/src/extended.rs | 425 +++++++++++++++++++++- crates/turnloop-contract/src/lib.rs | 14 +- crates/turnloop-contract/tests/wasi.rs | 10 + crates/turnloop-contract/tests/windows.rs | 23 +- crates/turnloop/src/backend/iocp/mod.rs | 6 +- crates/turnloop/src/backend/ipc.rs | 8 +- crates/turnloop/src/backend/socket.rs | 33 +- crates/turnloop/src/backend/unix.rs | 12 +- crates/turnloop/src/backend/wasi_p2.rs | 6 +- crates/turnloop/src/backend/wasi_p3.rs | 6 +- crates/turnloop/src/types.rs | 85 ++++- 11 files changed, 578 insertions(+), 50 deletions(-) diff --git a/crates/turnloop-contract/src/extended.rs b/crates/turnloop-contract/src/extended.rs index 1b5d336..509d2dd 100644 --- a/crates/turnloop-contract/src/extended.rs +++ b/crates/turnloop-contract/src/extended.rs @@ -171,26 +171,41 @@ pub fn pool_and_dns() { assert!(out.is_empty()); } } -pub fn reuse_port() { +/// Whether this target's kernel can distribute accepts across listeners, which +/// is what [`ReusePort::Distribute`] promises. Keep in step with +/// `backend::socket::reuse_port_option`. +pub const DISTRIBUTES: bool = cfg!(any( + target_os = "linux", + target_os = "android", + target_os = "freebsd" +)); + +/// Bind two listeners to one port with `reuse` and count what each one accepts. +/// +/// Returns `None` if the *second* bind was refused, which is how a backend +/// without the option reports itself. +fn two_listeners(reuse: ReusePort, connections: usize) -> Option<[usize; 2]> { let mut a = Driver::::new(Config::default()).expect("loop a"); let mut b = Driver::::new(Config::default()).expect("loop b"); let opts = ListenOpts { - reuse_port: true, + reuse_port: reuse, ..ListenOpts::default() }; - let ah = a.tcp_listen(localhost(), &opts).expect("listen a"); + let ah = a.tcp_listen(localhost(), &opts).ok()?; let addr = a.local_addr(ah).expect("addr"); - let bh = b.tcp_listen(addr, &opts).expect("listen b same port"); + let bh = b + .tcp_listen(addr, &opts) + .expect("second bind of a reused port"); a.accept_start(ah, Token(1)).expect("accept a"); b.accept_start(bh, Token(2)).expect("accept b"); - let clients: Vec<_> = (0..32) + let clients: Vec<_> = (0..connections) .map(|_| std::net::TcpStream::connect(addr).expect("connect")) .collect(); let mut count = [0usize; 2]; let mut out = Completions::default(); - let until = a.now() + Duration::from_secs(3); + let until = a.now() + Duration::from_secs(10); while count.iter().sum::() < clients.len() { - assert!(a.now() < until); + assert!(a.now() < until, "only {count:?} of {connections} accepted"); for l in [&mut a, &mut b] { l.turn(Timeout::Now, &mut out).expect("turn"); for c in out.drain() { @@ -199,15 +214,92 @@ pub fn reuse_port() { } } } - assert_eq!(count.iter().sum::(), 32); - // DESIGN §5a explicitly says macOS does not kernel-balance SO_REUSEPORT. - if cfg!(any(target_os = "linux", target_os = "freebsd")) { + Some(count) +} + +/// [`ReusePort::Share`] permits the duplicate bind and promises nothing else. +/// +/// Every connection must still be accepted by *someone*, because both sockets +/// hold the address. Which one is deliberately not asserted: on Linux the kernel +/// spreads them and on macOS the last binder takes all 32, and `Share` is +/// honest about covering both. +pub fn reuse_port_share() { + let Some(count) = two_listeners::(ReusePort::Share, 32) else { + // No SO_REUSEPORT on this backend at all; reuse_port_refused covers it. + return; + }; + assert_eq!(count.iter().sum::(), 32, "share {count:?}"); + eprintln!("reuse-port Share accepts: {count:?}"); +} + +/// [`ReusePort::Distribute`] either distributes or refuses the listener. +/// +/// This is the gate that makes the option honest. There is no third outcome: a +/// backend may not accept the request and then leave a listener starved. The +/// starved case is real and is what this exists to prevent — two loops sharing +/// one port under plain `SO_REUSEPORT` on macOS 15 split 32 connections +/// `[0, 32]`, so a host that developed against Linux would ship a server whose +/// first loop never accepts anything. +pub fn reuse_port_distribute() { + let mut l = Driver::::new(Config::default()).expect("loop"); + let attempt = l.tcp_listen( + localhost(), + &ListenOpts { + reuse_port: ReusePort::Distribute, + ..ListenOpts::default() + }, + ); + if !DISTRIBUTES { assert!( - count.iter().all(|n| *n > 0), - "kernel distribution {count:?}" + matches!( + attempt, + Err(Error { + kind: ErrorKind::Unsupported, + .. + }) + ), + "a platform that cannot distribute must refuse, got {attempt:?}" ); + assert!(!l.alive(), "a refused listener leaves nothing behind"); + return; + } + assert!(attempt.is_ok(), "this platform distributes: {attempt:?}"); + drop(l); + let count = two_listeners::(ReusePort::Distribute, 64).expect("distribute binds"); + assert_eq!(count.iter().sum::(), 64, "distribute {count:?}"); + assert!( + count.iter().all(|n| *n > 0), + "every listener must be given work: {count:?}" + ); + eprintln!("reuse-port Distribute accepts: {count:?}"); +} + +/// Backends with no address-reuse mechanism refuse both requests outright. +/// +/// A silent no-op is the failure mode this rejects: `Unsupported` at listen time +/// is recoverable, a listener that never accepts is not diagnosable. +pub fn reuse_port_refused(expected: &[ReusePort]) { + for &reuse in expected { + let mut l = Driver::::new(Config::default()).expect("loop"); + let attempt = l.tcp_listen( + localhost(), + &ListenOpts { + reuse_port: reuse, + ..ListenOpts::default() + }, + ); + assert!( + matches!( + attempt, + Err(Error { + kind: ErrorKind::Unsupported, + .. + }) + ), + "{reuse:?} must be refused, got {attempt:?}" + ); + assert!(!l.alive()); } - eprintln!("reuse-port accepts: {count:?}"); } pub fn handoff_distribution() { use std::io::{Read, Write}; @@ -605,3 +697,310 @@ pub fn long_jobs_settle_once_on_cancel_panic_and_shutdown() { pool_stats().long_busy == 0 }); } + +// --------------------------------------------------------------------------- +// DESIGN §5a.6: multi-threaded accept +// --------------------------------------------------------------------------- + +/// Connections served by one multi-threaded-accept scenario. +const MT_CONNECTIONS: usize = 64; +/// Loops, each on its own thread, sharing the port. +const MT_LOOPS: usize = 4; +/// Handle ceiling for every loop that serves connections. +/// +/// Far below `MT_CONNECTIONS`, and that is the point: each loop serves its +/// connections one at a time and returns the handle, so the workload fits +/// comfortably — but a handle leaked per connection, at accept, at `attach`, at +/// `detach` or at `close`, exhausts the ceiling long before the run ends and +/// turns an invisible orphan into a `ResourceLimit` failure. +const MT_CEILING: usize = 8; + +fn mt_config() -> Config { + Config { + max_handles: MT_CEILING, + max_operations: 64, + ..Config::default() + } +} + +/// Read one connection's two-byte id, echo it, close the handle and drain the +/// close. Returns the id, so a caller can prove which connection this was. +fn serve_one(l: &mut Driver, h: Handle) -> u16 { + let mut out = Completions::default(); + let until = l.now() + Duration::from_secs(30); + let mut got: Vec = Vec::new(); + let mut echoed = false; + l.read(h, ReadBuf::Pooled, Token(1)).expect("read"); + while !echoed { + assert!(l.now() < until, "connection stalled with {got:?}"); + l.turn(Timeout::Until(until), &mut out).expect("turn"); + for c in out.drain() { + match c.result { + OpResult::Read { n, lease: Some(b) } => { + assert_eq!(n, b.as_slice().len()); + got.extend_from_slice(b.as_slice()); + if got.len() < 2 { + l.read(h, ReadBuf::Pooled, Token(1)).expect("read more"); + } else { + l.write(h, WriteBuf::Owned(got.clone()), Token(2)) + .expect("echo"); + } + } + OpResult::Wrote(n) => { + assert_eq!(n, got.len()); + echoed = true; + } + other => panic!("unexpected {other:?}"), + } + } + } + l.close(h, Token(3)).expect("close"); + let mut closed = false; + while !closed { + assert!(l.now() < until, "close never completed"); + l.turn(Timeout::Until(until), &mut out).expect("release"); + for c in out.drain() { + assert!(matches!(c.result, OpResult::Closed), "{:?}", c.result); + closed = true; + } + } + assert_eq!(got.len(), 2, "one id per connection"); + u16::from_le_bytes([got[0], got[1]]) +} + +/// Connect, send a distinct id, and require that exact id back. +/// +/// The echo is what makes "served exactly once" checkable from the outside: a +/// connection dropped at handoff never answers and this fails on the read. +fn mt_clients(addr: SocketAddr, count: usize) -> Vec> { + (0..count) + .map(|i| { + thread::spawn(move || { + use std::io::{Read, Write}; + let mut s = std::net::TcpStream::connect(addr).expect("connect"); + s.set_read_timeout(Some(Duration::from_secs(30))) + .expect("timeout"); + s.write_all(&(i as u16).to_le_bytes()).expect("send id"); + let mut b = [0u8; 2]; + s.read_exact(&mut b).expect("echo"); + assert_eq!(u16::from_le_bytes(b), i as u16, "wrong connection answered"); + }) + }) + .collect() +} + +/// Every id was served, once, by someone. +#[track_caller] +fn mt_verify(mut served: Vec, per_loop: &[usize], route: &str) { + let total: usize = per_loop.iter().sum(); + assert_eq!(total, served.len(), "{route}: counts disagree with ids"); + served.sort_unstable(); + let expected: Vec = (0..MT_CONNECTIONS as u16).collect(); + assert_eq!( + served, expected, + "{route}: every connection exactly once, none lost or duplicated" + ); + eprintln!("{route}: per-loop {per_loop:?}"); +} + +/// One accepting loop, `MT_LOOPS` sibling loops on their own threads, each +/// connection handed over with `detach`/`attach` and then driven to completion +/// on the loop that adopted it. +/// +/// This is DESIGN §5a.6's "everywhere else" route, and on Windows it is the only +/// one: a socket joins exactly one completion port permanently, so a second loop +/// can never be given the same listener. +pub fn handoff_accept_exactly_once() { + let mut acceptor = Driver::::new(mt_config()).expect("acceptor"); + let listener = acceptor + .tcp_listen(localhost(), &ListenOpts::default()) + .expect("listen"); + let addr = acceptor.local_addr(listener).expect("addr"); + let mut senders = Vec::new(); + let mut workers = Vec::new(); + for _ in 0..MT_LOOPS { + let (tx, rx) = std::sync::mpsc::channel::(); + senders.push(tx); + workers.push(thread::spawn(move || { + let mut l = Driver::::new(mt_config()).expect("worker loop"); + let mut served = Vec::new(); + // The acceptor drops its senders when the last connection is gone, + // which is this worker's only stop signal. + while let Ok(d) = rx.recv_timeout(Duration::from_secs(30)) { + let h = l.attach(d, Token(0)).expect("attach"); + served.push(serve_one(&mut l, h)); + } + assert!(!l.alive(), "worker loop still alive at shutdown"); + served + })); + } + let clients = mt_clients(addr, MT_CONNECTIONS); + // Single-shot accept, re-armed per connection. turnloop#77 is open: a + // multishot accept can outrun the handle ceiling within one turn, and this + // loop deliberately runs at a low ceiling, so depending on multishot here + // would be testing that open issue rather than the handoff. + acceptor.accept(listener, Token(0)).expect("accept"); + let mut handed = 0; + let mut out = Completions::default(); + let until = acceptor.now() + Duration::from_secs(60); + while handed < MT_CONNECTIONS { + assert!(acceptor.now() < until, "only {handed} accepted"); + acceptor + .turn(Timeout::Until(until), &mut out) + .expect("turn"); + for c in out.drain() { + let OpResult::Accepted { conn, .. } = c.result else { + panic!("unexpected {:?}", c.result); + }; + let d = acceptor.detach(conn).expect("detach accepted connection"); + senders[handed % MT_LOOPS].send(d).expect("worker alive"); + handed += 1; + if handed < MT_CONNECTIONS { + acceptor.accept(listener, Token(0)).expect("re-arm"); + } + } + } + drop(senders); + for c in clients { + c.join().expect("client verified its own id came back"); + } + let mut served = Vec::new(); + let mut per_loop = Vec::new(); + for w in workers { + let ids = w.join().expect("worker"); + per_loop.push(ids.len()); + served.extend(ids); + } + mt_verify(served, &per_loop, "handoff"); + acceptor.close(listener, Token(9)).expect("close listener"); + acceptor.turn(Timeout::Now, &mut out).expect("release"); + assert!(matches!(out[0].result, OpResult::Closed)); + assert!(!acceptor.alive(), "acceptor still alive at shutdown"); +} + +/// `MT_LOOPS` loops on `MT_LOOPS` threads, each with its own listener on one +/// shared port, with the kernel choosing which loop accepts each connection. +/// +/// This is DESIGN §5a.6's kernel-balanced route. It runs only where +/// [`ReusePort::Distribute`] can be honoured; elsewhere the listener is refused +/// and [`reuse_port_distribute`] is the test that proves it. +/// +/// The share each loop receives is deliberately **not** asserted. Distribution +/// is by 4-tuple hash, so shares are uneven by nature and asserting evenness +/// would be asserting something the kernel never promised. What is asserted is +/// the invariant that matters: every connection served exactly once. +pub fn kernel_accept_exactly_once() { + if !DISTRIBUTES { + return; + } + let opts = ListenOpts { + reuse_port: ReusePort::Distribute, + ..ListenOpts::default() + }; + let done = Arc::new(AtomicUsize::new(0)); + let (addr_tx, addr_rx) = std::sync::mpsc::channel::(); + let (ready_tx, ready_rx) = std::sync::mpsc::channel::<()>(); + let spawn = |bind: Option, + addr_tx: std::sync::mpsc::Sender, + ready_tx: std::sync::mpsc::Sender<()>, + done: Arc| { + thread::spawn(move || { + let mut l = Driver::::new(mt_config()).expect("worker loop"); + let listener = l + .tcp_listen(bind.unwrap_or_else(localhost), &opts) + .expect("shared-port listen"); + if bind.is_none() { + addr_tx + .send(l.local_addr(listener).expect("addr")) + .expect("publish addr"); + } + drop(addr_tx); + ready_tx.send(()).expect("announce bound"); + drop(ready_tx); + let mut served = Vec::new(); + let mut out = Completions::default(); + let until = l.now() + Duration::from_secs(60); + // Stop only once every connection has been served by someone: this + // loop cannot know its own share in advance, because the kernel + // decides it. + while done.load(Ordering::Acquire) < MT_CONNECTIONS { + // Single-shot, for the turnloop#77 reason in the handoff route. + let op = l.accept(listener, Token(0)).expect("accept"); + let mut conn = None; + while conn.is_none() { + assert!(l.now() < until, "accept stalled"); + if done.load(Ordering::Acquire) >= MT_CONNECTIONS { + break; + } + l.turn(Timeout::After(Duration::from_millis(2)), &mut out) + .expect("turn"); + for c in out.drain() { + match c.result { + OpResult::Accepted { conn: h, .. } => conn = Some(h), + OpResult::Cancelled => {} + other => panic!("unexpected {other:?}"), + } + } + } + let Some(h) = conn else { + l.cancel(op); + break; + }; + served.push(serve_one(&mut l, h)); + done.fetch_add(1, Ordering::AcqRel); + } + // Drain whatever the final cancelled accept left behind. + let mut out = Completions::default(); + l.close(listener, Token(9)).expect("close listener"); + let until = l.now() + Duration::from_secs(10); + while l.alive() { + assert!(l.now() < until, "listener never released"); + l.turn(Timeout::Until(until), &mut out).expect("release"); + for c in out.drain() { + assert!( + matches!(c.result, OpResult::Closed | OpResult::Cancelled), + "{:?}", + c.result + ); + } + } + assert!(!l.alive(), "worker loop still alive at shutdown"); + served + }) + }; + let mut workers = vec![spawn( + None, + addr_tx.clone(), + ready_tx.clone(), + Arc::clone(&done), + )]; + let addr = addr_rx.recv().expect("first listener published its port"); + for _ in 1..MT_LOOPS { + workers.push(spawn( + Some(addr), + addr_tx.clone(), + ready_tx.clone(), + Arc::clone(&done), + )); + } + drop((addr_tx, ready_tx)); + // No client may connect before every listener holds the port, or the early + // connections could only ever reach the loops that had bound. + for _ in 0..MT_LOOPS { + ready_rx + .recv_timeout(Duration::from_secs(30)) + .expect("all listeners bound"); + } + let clients = mt_clients(addr, MT_CONNECTIONS); + for c in clients { + c.join().expect("client verified its own id came back"); + } + let mut served = Vec::new(); + let mut per_loop = Vec::new(); + for w in workers { + let ids = w.join().expect("worker"); + per_loop.push(ids.len()); + served.extend(ids); + } + mt_verify(served, &per_loop, "reuse-port"); +} diff --git a/crates/turnloop-contract/src/lib.rs b/crates/turnloop-contract/src/lib.rs index dc424f2..ebc2b11 100644 --- a/crates/turnloop-contract/src/lib.rs +++ b/crates/turnloop-contract/src/lib.rs @@ -279,13 +279,25 @@ mod native { long_jobs_settle_once_on_cancel_panic_and_shutdown::(); } #[test] + fn reuse_port_share_binds_twice() { + reuse_port_share::(); + } + #[test] fn reuse_port_distribution() { - reuse_port::(); + reuse_port_distribute::(); } #[test] fn accept_handoff_distribution() { handoff_distribution::(); } + #[test] + fn multi_threaded_accept_by_handoff() { + handoff_accept_exactly_once::(); + } + #[test] + fn multi_threaded_accept_by_reuse_port() { + kernel_accept_exactly_once::(); + } #[test] fn bounded() { diff --git a/crates/turnloop-contract/tests/wasi.rs b/crates/turnloop-contract/tests/wasi.rs index 577eaae..851b0de 100644 --- a/crates/turnloop-contract/tests/wasi.rs +++ b/crates/turnloop-contract/tests/wasi.rs @@ -164,6 +164,16 @@ fn socket_option_accept_defaults() { fn socket_option_handle_validation() { contract::sockopts::option_handle_validation::(); } +/// `wasi:sockets` has no address-reuse interface at all, so neither reuse-port +/// request can be honoured and both are refused at listen time. That leaves WASI +/// with no multi-core accept route: it is single-threaded in 0.2 and 0.3, and +/// `detach` is `Unsupported` there too (a socket is a component-model resource +/// handle in the component's own table, not a descriptor an embedder can adopt). +#[test] +fn reuse_port_has_no_wasi_interface() { + use turnloop::ReusePort; + contract::reuse_port_refused::(&[ReusePort::Share, ReusePort::Distribute]); +} #[test] fn socket_options_without_a_wasi_interface_are_unsupported() { use std::time::Duration; diff --git a/crates/turnloop-contract/tests/windows.rs b/crates/turnloop-contract/tests/windows.rs index 67952df..7529ab3 100644 --- a/crates/turnloop-contract/tests/windows.rs +++ b/crates/turnloop-contract/tests/windows.rs @@ -25,6 +25,8 @@ contract!( occupancy_classes_do_not_starve_each_other, long_jobs_settle_once_on_cancel_panic_and_shutdown, handoff_distribution, + handoff_accept_exactly_once, + kernel_accept_exactly_once, writev_and_shutdown, capacity_and_stale_ids, paged_growth_preserves_handles, @@ -88,24 +90,13 @@ fn sockets_pass_to_child_and_back() { fn four_loops_cross_post() { turnloop_contract::cross_post::(4, 1000); } +/// Windows has no SO_REUSEPORT, and SO_REUSEADDR there permits *hijacking* an +/// address rather than sharing it, so it must not stand in for either request. +/// Both are refused, which leaves `detach`/`attach` as the only multi-core +/// accept route on this platform — `handoff_accept_exactly_once` above. #[test] fn reuse_port_is_explicitly_unsupported() { - let mut driver = Loop::new(Config::default()).expect("loop"); - let result = driver.tcp_listen( - ([127, 0, 0, 1], 0).into(), - &ListenOpts { - reuse_port: true, - ..ListenOpts::default() - }, - ); - assert!(matches!( - result, - Err(Error { - kind: ErrorKind::Unsupported, - .. - }) - )); - assert!(!driver.alive()); + turnloop_contract::reuse_port_refused::(&[ReusePort::Share, ReusePort::Distribute]); } #[test] fn gui_event_receives_cross_thread_posts() { diff --git a/crates/turnloop/src/backend/iocp/mod.rs b/crates/turnloop/src/backend/iocp/mod.rs index 288e9fd..72203fc 100644 --- a/crates/turnloop/src/backend/iocp/mod.rs +++ b/crates/turnloop/src/backend/iocp/mod.rs @@ -1345,7 +1345,7 @@ unsafe impl Backend for Iocp { ) } Open::Listener { addr, opts } => { - if opts.reuse_port { + if opts.reuse_port.is_enabled() { return Err(unsupported()); } sockopt::validate_accept_defaults(opts.accept_defaults, Kind::Listener)?; @@ -1360,7 +1360,7 @@ unsafe impl Backend for Iocp { (transport, None, None) } Open::Udp { addr, opts } => { - if opts.reuse_port { + if opts.reuse_port.is_enabled() { return Err(unsupported()); } let socket = socket::create(addr.is_ipv6(), true)?; @@ -1372,7 +1372,7 @@ unsafe impl Backend for Iocp { ) } Open::PipeListener { name, opts } => { - if opts.reuse_port { + if opts.reuse_port.is_enabled() { return Err(unsupported()); } sockopt::validate_accept_defaults(opts.accept_defaults, Kind::PipeListener)?; diff --git a/crates/turnloop/src/backend/ipc.rs b/crates/turnloop/src/backend/ipc.rs index 7bda208..1580dff 100644 --- a/crates/turnloop/src/backend/ipc.rs +++ b/crates/turnloop/src/backend/ipc.rs @@ -45,9 +45,15 @@ pub(super) fn open(name: &PipeName, listen: Option) -> Result<(Detac let fd = unsafe { OwnedFd::from_raw_fd(raw) }; socket::configure(raw)?; if let Some(opts) = listen { - if opts.backlog > i32::MAX as u32 || opts.reuse_port { + if opts.backlog > i32::MAX as u32 { return Err(Error::new(ErrorKind::InvalidInput)); } + // A filesystem socket has no port for a second bind to contend for, so + // there is no SO_REUSEPORT for AF_UNIX on any platform. Report that + // rather than binding a listener the request was never applied to. + if opts.reuse_port.is_enabled() { + return Err(Error::new(ErrorKind::Unsupported)); + } // SAFETY: initialized sockaddr of the advertised length, live socket. if unsafe { libc::bind(raw, addr.ptr(), len) } < 0 { return Err(last_error()); diff --git a/crates/turnloop/src/backend/socket.rs b/crates/turnloop/src/backend/socket.rs index e8c44e8..7fd40db 100644 --- a/crates/turnloop/src/backend/socket.rs +++ b/crates/turnloop/src/backend/socket.rs @@ -1,5 +1,5 @@ use super::poller::last_error; -use crate::{Error, ErrorKind, Result}; +use crate::{Error, ErrorKind, Result, ReusePort}; use std::{ mem::{size_of, zeroed}, net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6}, @@ -96,6 +96,37 @@ impl Addr { } } } +/// The `SOL_SOCKET` option that honours a [`ReusePort`] request on this target, +/// or `None` when nothing needs to be set. +/// +/// This is the single place the platform split lives. `SO_REUSEPORT` is spelled +/// identically on Linux and the BSDs and means different things: +/// +/// * **Linux and Android** distribute incoming connections across every listener +/// bound with `SO_REUSEPORT`, selecting one by hashing the connection's +/// 4-tuple. One option satisfies both requests. +/// * **FreeBSD** keeps `SO_REUSEPORT` at its original BSD meaning — duplicate +/// binding, with new connections going to the socket that bound last — and +/// added a separate `SO_REUSEPORT_LB` in 12.0 that distributes. So the two +/// requests are two different options here. +/// * **macOS and the other Apple platforms, NetBSD, OpenBSD and DragonFly** have +/// only the original option. A [`ReusePort::Distribute`] request is refused +/// rather than silently answered with `SO_REUSEPORT`: measured on macOS 15, +/// two loops binding one port with `SO_REUSEPORT` split 32 connections +/// `[0, 32]` — the first listener is not merely under-served, it is never +/// given anything at all. +pub(crate) fn reuse_port_option(reuse: ReusePort) -> Result> { + match reuse { + ReusePort::No => Ok(None), + ReusePort::Share => Ok(Some(libc::SO_REUSEPORT)), + #[cfg(any(target_os = "linux", target_os = "android"))] + ReusePort::Distribute => Ok(Some(libc::SO_REUSEPORT)), + #[cfg(target_os = "freebsd")] + ReusePort::Distribute => Ok(Some(libc::SO_REUSEPORT_LB)), + #[cfg(not(any(target_os = "linux", target_os = "android", target_os = "freebsd")))] + ReusePort::Distribute => Err(Error::new(ErrorKind::Unsupported)), + } +} pub(crate) fn option(fd: RawFd, level: i32, name: i32, value: i32) -> Result<()> { // SAFETY: value is an initialized integer of the size passed to setsockopt. if unsafe { diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs index 1d63448..89c9467 100644 --- a/crates/turnloop/src/backend/unix.rs +++ b/crates/turnloop/src/backend/unix.rs @@ -581,7 +581,7 @@ unsafe impl Backend for Unix { Open::Tcp { addr, opts } => ( addr, Kind::Tcp, - false, + ReusePort::No, 0, opts.nodelay, AcceptDefaults::EMPTY, @@ -613,11 +613,11 @@ unsafe impl Backend for Unix { // TCP listeners need address reuse for TIME_WAIT. Default UDP binds // must stay exclusive: on Linux SO_REUSEADDR also permits two live // bind(:0) sockets to receive the same ephemeral endpoint. - if kind == Kind::Listener || reuse { + if kind == Kind::Listener || reuse.is_enabled() { socket::option(fd.as_raw_fd(), libc::SOL_SOCKET, libc::SO_REUSEADDR, 1)?; } - if reuse { - socket::option(fd.as_raw_fd(), libc::SOL_SOCKET, libc::SO_REUSEPORT, 1)?; + if let Some(option) = super::socket::reuse_port_option(reuse)? { + socket::option(fd.as_raw_fd(), libc::SOL_SOCKET, option, 1)?; } let a = Addr::new(addr); // SAFETY: sockaddr pointer and length refer to initialized storage. @@ -1248,7 +1248,9 @@ mod udp_tests { let mut checked = 0; for addr in ["127.0.0.1:0", "[::1]:0"] { let mut l = Loop::new(Config::default()).expect("loop"); - let opts = UdpOpts { reuse_port: true }; + let opts = UdpOpts { + reuse_port: ReusePort::Share, + }; let first = l .udp_bind(addr.parse().expect("address"), &opts) .expect("first bind"); diff --git a/crates/turnloop/src/backend/wasi_p2.rs b/crates/turnloop/src/backend/wasi_p2.rs index 2770c03..cf0e356 100644 --- a/crates/turnloop/src/backend/wasi_p2.rs +++ b/crates/turnloop/src/backend/wasi_p2.rs @@ -459,7 +459,7 @@ unsafe impl Backend for WasiP2 { if opts.nodelay { return Err(Error::new(ErrorKind::Unsupported)); } - (addr, Kind::Tcp, false, 0, AcceptDefaults::EMPTY) + (addr, Kind::Tcp, ReusePort::No, 0, AcceptDefaults::EMPTY) } Open::Listener { addr, opts } => ( addr, @@ -472,7 +472,9 @@ unsafe impl Backend for WasiP2 { (addr, Kind::Udp, opts.reuse_port, 0, AcceptDefaults::EMPTY) } }; - if reuse { + // wasi:sockets has no address-reuse interface at all, so neither + // Share nor Distribute can be honoured. Both are refused. + if reuse.is_enabled() { return Err(Error::new(ErrorKind::Unsupported)); } sockopt::validate_accept_defaults(accept_defaults)?; diff --git a/crates/turnloop/src/backend/wasi_p3.rs b/crates/turnloop/src/backend/wasi_p3.rs index b329533..2b6baea 100644 --- a/crates/turnloop/src/backend/wasi_p3.rs +++ b/crates/turnloop/src/backend/wasi_p3.rs @@ -426,7 +426,7 @@ unsafe impl Backend for WasiP3 { if opts.nodelay { return Err(Error::new(ErrorKind::Unsupported)); } - (addr, Kind::Tcp, false, 0, AcceptDefaults::EMPTY) + (addr, Kind::Tcp, ReusePort::No, 0, AcceptDefaults::EMPTY) } Open::Listener { addr, opts } => ( addr, @@ -439,7 +439,9 @@ unsafe impl Backend for WasiP3 { (addr, Kind::Udp, opts.reuse_port, 0, AcceptDefaults::EMPTY) } }; - if reuse { + // wasi:sockets has no address-reuse interface at all, so neither + // Share nor Distribute can be honoured. Both are refused. + if reuse.is_enabled() { return Err(Error::new(ErrorKind::Unsupported)); } sockopt::validate_accept_defaults(accept_defaults)?; diff --git a/crates/turnloop/src/types.rs b/crates/turnloop/src/types.rs index 696e40c..26aa30b 100644 --- a/crates/turnloop/src/types.rs +++ b/crates/turnloop/src/types.rs @@ -212,6 +212,75 @@ pub struct TcpOpts { /// Disable the TCP Nagle algorithm for latency-sensitive small writes. pub nodelay: bool, } +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +/// What a second bind of the same address is allowed to do. +/// +/// `SO_REUSEPORT` is spelled the same on Linux and on the BSDs and **does not +/// mean the same thing**. On Linux it both permits the duplicate bind and +/// distributes incoming connections across every listener holding the address. +/// On macOS and the other BSDs the same option permits the duplicate bind and +/// then hands new connections to the socket that bound *last*, so a second +/// listener silently takes the whole port and the first one accepts nothing. +/// +/// A `bool` cannot express that difference, so it is not one. Each variant here +/// names the behaviour the caller is asking the kernel for, and a backend that +/// cannot provide it refuses the listener with `Unsupported` at creation rather +/// than binding a socket that will never be given any work. +/// +/// The variants are ordered by strength, and a platform may satisfy a request +/// with something stronger: [`Share`](Self::Share) on Linux is the same +/// `setsockopt` as [`Distribute`](Self::Distribute) and does distribute. What a +/// variant guarantees is a floor, never a ceiling. +pub enum ReusePort { + /// Exclusive bind: no other socket may hold this address (the default). + #[default] + No, + /// Permit the duplicate bind, and promise nothing about delivery. + /// + /// Several sockets may hold the address at once; which one receives a given + /// connection or datagram is the platform's business. This is the variant + /// for the traditional BSD uses — receiving multicast or broadcast datagrams + /// in several processes, and handing a port to a replacement process during + /// a zero-downtime restart — where last-binder-wins is the desired effect + /// rather than a defect. + /// + /// Supported wherever `SO_REUSEPORT` exists: Linux, Android, macOS and the + /// BSDs. `Unsupported` on Windows, WASI and the web, and on Unix local + /// (`AF_UNIX`) listeners. + Share, + /// Permit the duplicate bind **and** spread incoming connections across + /// every listener holding the address. + /// + /// This is the kernel-balanced route of DESIGN §5a: N loops on N threads + /// each bind the same port, and the kernel decides which loop accepts each + /// connection, with no shared accept lock and no handoff. + /// + /// **It distributes by hash, not by load.** Linux selects the listener by + /// hashing the connection's address 4-tuple; FreeBSD's `SO_REUSEPORT_LB` + /// does the same. Neither asks how busy a listener is, so a loop whose agent + /// is blocked in a long turn keeps being given its share of new connections + /// and they wait in its queue. Even distribution of *connections* is not + /// even distribution of *work*, and a host that needs the latter wants the + /// handoff route ([`Loop::detach`]/[`Loop::attach`]), where the policy is + /// the host's to write. + /// + /// Supported on Linux and Android (`SO_REUSEPORT`) and on FreeBSD + /// (`SO_REUSEPORT_LB`, FreeBSD 12.0+). **`Unsupported` on macOS and other + /// Apple platforms, on NetBSD, OpenBSD and DragonFly, on Windows, on WASI + /// and on the web** — none of them has an option that distributes, and + /// accepting the request by setting plain `SO_REUSEPORT` would produce + /// exactly the silently-starved listener this variant exists to prevent. + /// + /// [`Loop::detach`]: crate::Driver::detach + /// [`Loop::attach`]: crate::Driver::attach + Distribute, +} +impl ReusePort { + /// Whether this request needs a duplicate bind at all. + pub const fn is_enabled(self) -> bool { + !matches!(self, Self::No) + } +} #[derive(Clone, Copy, Debug)] /// Listener backlog, kernel reuse-port configuration and accepted-socket defaults. /// @@ -220,8 +289,8 @@ pub struct TcpOpts { /// by the backend to every TCP listener it binds (TIME_WAIT rebinding). Neither /// can be changed on a socket that is already bound, so neither is an option. pub struct ListenOpts { - /// Enable SO_REUSEPORT when supported; macOS does not promise balanced accepts. - pub reuse_port: bool, + /// What a second bind of this address may do; see [`ReusePort`]. + pub reuse_port: ReusePort, /// Maximum pending connection backlog requested from the OS. pub backlog: u32, /// Options applied to every connection this listener accepts. @@ -230,7 +299,7 @@ pub struct ListenOpts { impl Default for ListenOpts { fn default() -> Self { Self { - reuse_port: false, + reuse_port: ReusePort::No, backlog: 128, accept_defaults: AcceptDefaults::EMPTY, } @@ -393,9 +462,13 @@ pub enum SocketOptionKind { /// UDP binding options. Reuse is bind-time only and stays here, not in /// [`SocketOption`]; everything changeable on a live socket is an option. pub struct UdpOpts { - /// Enable SO_REUSEPORT when supported; macOS does not promise balanced accepts. - /// Defaults to false: a live UDP endpoint cannot be shared by another bind. - pub reuse_port: bool, + /// What a second bind of this address may do; see [`ReusePort`]. + /// + /// Defaults to [`ReusePort::No`]: a live UDP endpoint cannot be shared by + /// another bind. [`ReusePort::Share`] is the variant multicast and broadcast + /// receivers want; [`ReusePort::Distribute`] spreads *datagrams* across the + /// bound sockets on the platforms that can, and is refused elsewhere. + pub reuse_port: ReusePort, } /// The fd/event is borrowed from the driver; it must never be closed by the host. From 4b50d2a571b8e2ce34230f4e125f911665f72ab7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 13:31:54 +0200 Subject: [PATCH 2/7] A scaling harness for both multi-threaded accept routes turnloop-bench --accept-scaling drives N loops on N threads serving one port by either route and reports connections per second with the per-loop service counts beside it, so the distribution is visible next to the total rather than inferred from it. It asserts its own subject ran before it prints anything: every loop must have served at least one connection, the per-loop counts must sum to the budget, and --connections must be at least 100 per loop so a loop cannot be given nothing by chance. A scaling number from a run where three of four loops sat idle would read as a flat curve and mean nothing. --route reuse-port is refused up front, with a status and one line naming the alternative, on the platforms that cannot distribute in the kernel. --- crates/turnloop-bench/src/main.rs | 19 ++ crates/turnloop-bench/src/scaling.rs | 458 +++++++++++++++++++++++++++ 2 files changed, 477 insertions(+) create mode 100644 crates/turnloop-bench/src/scaling.rs diff --git a/crates/turnloop-bench/src/main.rs b/crates/turnloop-bench/src/main.rs index 796a78f..aa8c818 100644 --- a/crates/turnloop-bench/src/main.rs +++ b/crates/turnloop-bench/src/main.rs @@ -7,6 +7,14 @@ use std::{ }; #[cfg(feature = "timer-btree")] mod btree; +#[cfg(any( + target_vendor = "apple", + target_os = "linux", + target_os = "android", + target_os = "freebsd", + windows +))] +mod scaling; #[cfg(feature = "timer-btree")] use btree::Tree as TimerQueue; #[cfg(not(feature = "timer-btree"))] @@ -271,6 +279,17 @@ fn main() { } else { Counter::new() }; + #[cfg(any( + target_vendor = "apple", + target_os = "linux", + target_os = "android", + target_os = "freebsd", + windows + ))] + if std::env::args().any(|a| a == "--accept-scaling") { + scaling::run(&counter, &scaling::Args::parse(std::env::args())); + return; + } #[cfg(any( target_vendor = "apple", target_os = "linux", diff --git a/crates/turnloop-bench/src/scaling.rs b/crates/turnloop-bench/src/scaling.rs new file mode 100644 index 0000000..fef634b --- /dev/null +++ b/crates/turnloop-bench/src/scaling.rs @@ -0,0 +1,458 @@ +//! Multi-threaded accept scaling harness (DESIGN §5a.6). +//! +//! The question this exists to answer is whether **one** turnloop server can use +//! more than one core, and by which of the two routes §5a.6 offers: +//! +//! * `--route reuse-port`: N loops on N threads, each with its own listener on +//! one port, and the kernel deciding which loop accepts each connection. Only +//! where [`ReusePort::Distribute`] can be honoured — Linux, Android, FreeBSD. +//! * `--route handoff`: one accepting loop and N sibling loops on N threads, +//! each connection moved with `detach`/`attach`. Available everywhere native, +//! and on Windows it is the only route, because a socket joins exactly one +//! completion port permanently. +//! +//! The workload is connection-oriented on purpose — accept, one request, one +//! response, close — because that is the shape where the accept path is the +//! thing under test rather than a rounding error. +//! +//! **This harness reports, it does not judge.** It prints one JSON line per run +//! plus the per-loop service counts, and it refuses to print anything at all +//! unless its own subject demonstrably ran: every loop must have bound a +//! listener (reuse-port) or adopted connections (handoff), and every loop must +//! have served at least one connection. A scaling number from a run where three +//! of four loops sat idle is worse than no number. +use crate::counter::Counter; +use std::{ + io::{Read, Write}, + net::{SocketAddr, TcpStream}, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + mpsc, + }, + thread, + time::{Duration, Instant}, +}; +use turnloop::*; + +/// Which of DESIGN §5a.6's two routes to exercise. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Route { + /// Every loop binds the port; the kernel picks the accepting loop. + ReusePort, + /// One loop accepts and hands each connection to a sibling loop. + Handoff, +} + +/// One scaling run's configuration. +pub struct Args { + /// Which route to exercise. + pub route: Route, + /// Serving loops, each on its own thread. + pub loops: usize, + /// Connections to drive through the server in total. + pub connections: usize, + /// Client threads generating the load. + pub clients: usize, + /// Request and response size in bytes. + pub payload: usize, + /// Handle ceiling per serving loop. + /// + /// Generous by default: turnloop#77 is open, so a multishot accept can + /// outrun a tight ceiling within one turn, and a harness that hit that + /// would be measuring the open issue. + pub max_handles: usize, +} + +impl Default for Args { + fn default() -> Self { + Self { + route: Route::Handoff, + loops: 4, + connections: 20_000, + clients: 8, + payload: 64, + max_handles: 4096, + } + } +} + +impl Args { + /// Parse the harness flags out of a command line. + /// + /// Unrecognised arguments are ignored: this shares a binary with the timer + /// and instruction harnesses. + pub fn parse>(args: I) -> Self { + let argv: Vec = args.collect(); + let mut out = Self::default(); + let value = |name: &str| -> Option { + let i = argv.iter().position(|a| a == name)?; + argv.get(i + 1).cloned() + }; + let number = |name: &str, current: usize| -> usize { + value(name).map_or(current, |v| { + v.parse() + .unwrap_or_else(|_| panic!("{name} takes a number")) + }) + }; + if let Some(r) = value("--route") { + out.route = match r.as_str() { + "reuse-port" => Route::ReusePort, + "handoff" => Route::Handoff, + other => panic!("--route is reuse-port or handoff, not {other}"), + }; + } + out.loops = number("--loops", out.loops).max(1); + out.connections = number("--connections", out.connections); + out.clients = number("--clients", out.clients).max(1); + out.payload = number("--payload", out.payload).max(1); + out.max_handles = number("--max-handles", out.max_handles); + out + } + + fn config(&self) -> Config { + Config { + max_handles: self.max_handles, + max_operations: self.max_handles * 4, + ..Config::default() + } + } +} + +/// A serving loop: accept or adopt connections, echo one payload each, close. +struct Server { + slots: Vec>, + payload: usize, + served: usize, +} + +impl Server { + fn new(capacity: usize, payload: usize) -> Self { + Self { + slots: vec![None; capacity], + payload, + served: 0, + } + } + /// Begin serving a connection this loop now owns. + fn adopt(&mut self, l: &mut Loop, h: Handle) { + let i = h.index(); + assert!(self.slots[i].is_none(), "slot {i} reused while live"); + self.slots[i] = Some(h); + l.read(h, ReadBuf::Pooled, Token(i as u64)).expect("read"); + } + /// Advance one completion. Returns true when a connection finished. + fn step(&mut self, l: &mut Loop, c: Completion) -> bool { + let i = c.token.0 as usize; + match c.result { + OpResult::Read { n, lease: Some(b) } => { + let h = self.slots[i].expect("live slot"); + assert!(n > 0); + // Echo exactly what arrived; a short read re-arms rather than + // answering a partial request. + if n < self.payload { + l.read(h, ReadBuf::Pooled, Token(i as u64)).expect("more"); + } else { + l.write(h, WriteBuf::Owned(b.as_slice().to_vec()), Token(i as u64)) + .expect("echo"); + } + false + } + OpResult::Wrote(_) => { + let h = self.slots[i].expect("live slot"); + l.close(h, Token(i as u64)).expect("close"); + false + } + OpResult::Closed => { + self.slots[i] = None; + self.served += 1; + true + } + OpResult::Eof => { + let h = self.slots[i].expect("live slot"); + l.close(h, Token(i as u64)).expect("close"); + false + } + other => panic!("unexpected {other:?}"), + } + } +} + +/// Token reserved for the listener's own multishot accept. +const ACCEPT: Token = Token(u64::MAX); + +/// Connect, send one payload, read it back, close. Repeated by each client +/// thread until the run's connection budget is spent. +fn client(addr: SocketAddr, payload: usize, budget: &AtomicUsize) -> usize { + let request = vec![0x5a_u8; payload]; + let mut response = vec![0_u8; payload]; + let mut done = 0; + while budget + .try_update(Ordering::AcqRel, Ordering::Acquire, |n| n.checked_sub(1)) + .is_ok() + { + let mut s = TcpStream::connect(addr).expect("connect"); + s.set_nodelay(true).expect("nodelay"); + s.set_read_timeout(Some(Duration::from_secs(30))) + .expect("timeout"); + s.write_all(&request).expect("request"); + s.read_exact(&mut response).expect("response"); + assert_eq!(response, request, "wrong bytes came back"); + done += 1; + } + done +} + +/// Run one scaling configuration and print its result. +pub fn run(counter: &Counter, args: &Args) { + assert!( + args.connections >= 100 * args.loops, + "--connections must be at least 100 per loop, or a loop can be given \ + nothing by chance and the run proves nothing" + ); + // Refuse a route this platform cannot take, before any thread is started, + // so the operator gets one line and a status rather than a panic from + // inside a worker and a confusing rendezvous failure behind it. + if args.route == Route::ReusePort { + let mut probe = Loop::new(Config::default()).expect("probe loop"); + if let Err(e) = probe.tcp_listen( + ([127, 0, 0, 1], 0).into(), + &ListenOpts { + reuse_port: ReusePort::Distribute, + ..ListenOpts::default() + }, + ) { + eprintln!( + "--route reuse-port is unavailable here: ReusePort::Distribute is {:?}. Only Linux, Android and FreeBSD distribute accepts in the kernel; on this platform use --route handoff, which is the supported multi-core route (DESIGN 5a.6).", + e.kind + ); + std::process::exit(2); + } + } + let served = Arc::new(AtomicUsize::new(0)); + let budget = Arc::new(AtomicUsize::new(args.connections)); + let (addr, threads) = match args.route { + Route::ReusePort => reuse_port_servers(args, &served), + Route::Handoff => handoff_servers(args, &served), + }; + let before = counter.read().expect("counter"); + let start = Instant::now(); + let payload = args.payload; + let clients: Vec<_> = (0..args.clients) + .map(|_| { + let budget = Arc::clone(&budget); + thread::spawn(move || client(addr, payload, &budget)) + }) + .collect(); + let driven: usize = clients.into_iter().map(|c| c.join().expect("client")).sum(); + let elapsed = start.elapsed(); + let total = counter + .read() + .expect("counter") + .checked_sub(before) + .expect("monotonic counter"); + let per_loop: Vec = threads + .into_iter() + .map(|t| t.join().expect("loop")) + .collect(); + + // Liveness before reporting. A number from a run whose subject never + // executed is the failure mode this project has paid for repeatedly. + assert_eq!(driven, args.connections, "client budget not spent"); + assert_eq!( + per_loop.iter().sum::(), + args.connections, + "connections served {per_loop:?} does not match the budget" + ); + assert_eq!(per_loop.len(), args.loops, "not every loop reported"); + assert!( + per_loop.iter().all(|n| *n > 0), + "a loop served nothing, so this configuration did not run on {} cores: {per_loop:?}", + args.loops + ); + let route = match args.route { + Route::ReusePort => "reuse_port", + Route::Handoff => "handoff", + }; + println!( + "{{\"name\":\"accept_scaling_{route}_{}\",\"operations\":{},\"total\":{total},\ + \"per_operation\":{:.2},\"unit\":\"{}\",\"loops\":{},\"elapsed_ns\":{},\ + \"connections_per_second\":{:.0},\"per_loop\":{per_loop:?}}}", + args.loops, + args.connections, + total as f64 / args.connections as f64, + counter.unit(), + args.loops, + elapsed.as_nanos(), + args.connections as f64 / elapsed.as_secs_f64(), + ); +} + +/// N loops, N threads, one listener each on a shared port. +fn reuse_port_servers( + args: &Args, + served: &Arc, +) -> (SocketAddr, Vec>) { + let opts = ListenOpts { + reuse_port: ReusePort::Distribute, + accept_defaults: AcceptDefaults { + nodelay: true, + keep_alive: None, + }, + ..ListenOpts::default() + }; + let config = args.config(); + let (payload, budget, loops) = (args.payload, args.connections, args.loops); + let (addr_tx, addr_rx) = mpsc::channel::(); + let (ready_tx, ready_rx) = mpsc::channel::<()>(); + let spawn = |bind: Option, + addr_tx: mpsc::Sender, + ready_tx: mpsc::Sender<()>, + served: Arc| { + thread::spawn(move || { + let mut l = Loop::new(config).expect("loop"); + let listener = l + .tcp_listen( + bind.unwrap_or_else(|| ([127, 0, 0, 1], 0).into()), + &opts, + ) + .unwrap_or_else(|e| { + panic!("this platform cannot honour ReusePort::Distribute ({e:?}); use --route handoff") + }); + if bind.is_none() { + addr_tx + .send(l.local_addr(listener).expect("addr")) + .expect("addr"); + } + drop((addr_tx, ready_tx)); + l.accept_start(listener, ACCEPT).expect("accept_start"); + let mut server = Server::new(config.max_handles, payload); + let mut out = Completions::default(); + while served.load(Ordering::Acquire) < budget { + l.turn(Timeout::After(Duration::from_millis(1)), &mut out) + .expect("turn"); + for c in out.drain() { + if c.token == ACCEPT { + let OpResult::Accepted { conn, .. } = c.result else { + panic!("unexpected {:?}", c.result); + }; + server.adopt(&mut l, conn); + } else if server.step(&mut l, c) { + served.fetch_add(1, Ordering::AcqRel); + } + } + } + server.served + }) + }; + let mut threads = vec![spawn( + None, + addr_tx.clone(), + ready_tx.clone(), + Arc::clone(served), + )]; + let addr = addr_rx.recv().expect("first listener published its port"); + for _ in 1..loops { + threads.push(spawn( + Some(addr), + addr_tx.clone(), + ready_tx.clone(), + Arc::clone(served), + )); + } + drop((addr_tx, ready_tx)); + // Every listener must hold the port before the first client connects, or + // the early connections could only ever reach the loops that had bound. + while ready_rx.recv().is_ok() {} + (addr, threads) +} + +/// One accepting loop plus N sibling loops adopting detached connections. +fn handoff_servers( + args: &Args, + served: &Arc, +) -> (SocketAddr, Vec>) { + let config = args.config(); + let (payload, budget, loops) = (args.payload, args.connections, args.loops); + let mut senders = Vec::new(); + let mut threads = Vec::new(); + let mut notifiers = Vec::new(); + let (ready_tx, ready_rx) = mpsc::channel::(); + for _ in 0..loops { + let (tx, rx) = mpsc::channel::(); + senders.push(tx); + let served = Arc::clone(served); + let ready_tx = ready_tx.clone(); + threads.push(thread::spawn(move || { + let mut l = Loop::new(config).expect("worker loop"); + ready_tx.send(l.notifier()).expect("publish notifier"); + drop(ready_tx); + let mut server = Server::new(config.max_handles, payload); + let mut out = Completions::default(); + while served.load(Ordering::Acquire) < budget { + // The acceptor notifies after sending, so a parked worker wakes + // on a handoff rather than polling for one. + while let Ok(d) = rx.try_recv() { + let h = l.attach(d, Token(0)).expect("attach"); + server.adopt(&mut l, h); + } + l.turn(Timeout::After(Duration::from_millis(1)), &mut out) + .expect("turn"); + for c in out.drain() { + if server.step(&mut l, c) { + served.fetch_add(1, Ordering::AcqRel); + } + } + } + server.served + })); + } + drop(ready_tx); + for _ in 0..loops { + notifiers.push(ready_rx.recv().expect("worker notifier")); + } + let (addr_tx, addr_rx) = mpsc::channel::(); + let acceptor_served = Arc::clone(served); + thread::spawn(move || { + let mut l = Loop::new(config).expect("acceptor"); + let listener = l + .tcp_listen( + ([127, 0, 0, 1], 0).into(), + &ListenOpts { + accept_defaults: AcceptDefaults { + nodelay: true, + keep_alive: None, + }, + ..ListenOpts::default() + }, + ) + .expect("listen"); + addr_tx + .send(l.local_addr(listener).expect("addr")) + .expect("addr"); + drop(addr_tx); + l.accept_start(listener, ACCEPT).expect("accept_start"); + let mut out = Completions::default(); + let mut next = 0; + while acceptor_served.load(Ordering::Acquire) < budget { + l.turn(Timeout::After(Duration::from_millis(1)), &mut out) + .expect("turn"); + for c in out.drain() { + let OpResult::Accepted { conn, .. } = c.result else { + panic!("unexpected {:?}", c.result); + }; + // Round-robin. DESIGN §5a.6 puts the policy in the host, and + // this is the simplest one a host could write; least-loaded + // would be the other obvious choice and would need the workers + // to publish their depth. + let d = l.detach(conn).expect("detach"); + if senders[next % loops].send(d).is_ok() { + let _ = notifiers[next % loops].notify(); + } + next += 1; + } + } + }); + let addr = addr_rx.recv().expect("acceptor published its port"); + (addr, threads) +} From 7aeaf24574b24f962c3b702c8d56b5716e45206d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 13:35:21 +0200 Subject: [PATCH 3/7] Write down which multi-core accept route each platform actually has DESIGN 5a.6 said the kernel balances on Linux/FreeBSD. That is right for Linux and wrong for FreeBSD, where plain SO_REUSEPORT keeps its original BSD meaning and SO_REUSEPORT_LB is the option that distributes. It also said macOS does not balance without saying what it does instead, which is give the whole port to the last binder. docs/multi-threaded-accept.md is the page: the two routes, the per-platform table, why Windows has only one of them, what the contract tests assert, and the runbook for the scaling harness - including the step that has to come first, which is raising --clients until the number stops moving, because the client side is synchronous and a server-scaling curve measured against a saturated client is a picture of the client. It also says what a result would mean in each direction, so the sweep can be read rather than interpreted. No timing numbers are recorded: the harness has only been run for correctness, on loaded machines, where a figure would be worse than none. --- DESIGN.md | 17 ++- README.md | 1 + docs/lanes/sockopts.md | 1 + docs/multi-threaded-accept.md | 216 ++++++++++++++++++++++++++++++++++ 4 files changed, 230 insertions(+), 5 deletions(-) create mode 100644 docs/multi-threaded-accept.md diff --git a/DESIGN.md b/DESIGN.md index 1385547..6b81def 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -268,16 +268,19 @@ There are two ways a host drives a loop: - **Windows:** a handle's IOCP association is permanent and inescapable. Windows cannot dissociate one, rejects a second `CreateIoCompletionPort` with `ERROR_INVALID_PARAMETER`, and neither `WSADuplicateSocketW` nor `DuplicateHandle` escapes it — both produce another descriptor for the *same* socket or file object, which is where the association lives. Quiescence makes it inert (no packet can ever arrive for it), and the receiving host drives the transport with synchronous/non-blocking Winsock calls or with overlapped calls whose `OVERLAPPED.hEvent` has its low-order bit set, which suppresses the completion packet. A named-pipe instance keeps `FILE_FLAG_OVERLAPPED`, so for a pipe the tagged-`hEvent` form is the only one. The route back to completion-port-driven I/O is `attach`: an imported association is exactly what the IOCP backend's overlapped-event routing exists for. - **WASI 0.2/0.3 and web:** `Unsupported`. A WASI socket is a component-model resource handle in the component's own table, not a descriptor, and there is no interface that hands one to the embedder; a browser resource is a host JS object. Neither has an identity a host could act on. - **Reporting only:** `Loop::raw_transport(h) -> RawTransport` reports a live transport's native identity for Node's `socket._handle.fd`. The loop keeps ownership; the value is valid until the handle is closed or detached, and is for reporting and read-only queries, never for I/O, closing, mode changes or registration elsewhere. -6. **Multi-threaded accept:** - - **Kernel-balanced:** where the kernel balances load (`SO_REUSEPORT` on Linux/FreeBSD), each loop gets its own listener with `ListenOpts::reuse_port`. - - **Everywhere else** (macOS doesn't balance, and on Windows a socket can join only one completion port): one accepting loop hands connections to other loops with `detach`/`attach`. The policy (round-robin, least-loaded) belongs to the host. +6. **Multi-threaded accept.** One server, more than one core, without a work-stealing loop: a shared loop would have to route every completion back to its owning agent, which rebuilds the per-agent queues and wakeups this crate exists to remove. Two routes, and `ListenOpts::reuse_port` is a `ReusePort` request rather than a `bool` because the two are not the same promise. See [multi-threaded-accept.md](docs/multi-threaded-accept.md). + - **Kernel-distributed** (`ReusePort::Distribute`): each loop gets its own listener on the same port and the kernel picks which loop accepts each connection. `SO_REUSEPORT` on **Linux and Android**; `SO_REUSEPORT_LB` on **FreeBSD** (12.0+) — plain `SO_REUSEPORT` there keeps its original BSD meaning and does *not* distribute, so it is not the same option. **`Unsupported` on macOS and the other Apple platforms, NetBSD, OpenBSD, DragonFly, Windows, WASI and the web**, where the listener is refused at creation instead of being bound and starved. + - **It distributes by hash, not by load.** The listener is chosen by hashing the connection's 4-tuple; nothing asks how busy a loop is, so a loop whose agent is in a long turn keeps being given its share and those connections wait. Measured on Linux 6.8, 64 connections over 4 loops: `[7, 12, 21, 24]`. Even distribution of connections is not even distribution of work. + - **`ReusePort::Share`** is the weaker request — duplicate binding, delivery unspecified — for multicast/broadcast receivers and zero-downtime restarts. It is honest about the BSD behaviour rather than hiding it: measured on macOS 15, two listeners sharing a port with `SO_REUSEPORT` split 32 connections `[0, 32]`, all to the last binder. + - **Handoff, everywhere else:** one accepting loop hands each connection to a sibling loop with `detach`/`attach`. The policy (round-robin, least-loaded) belongs to the host. This works today on every native backend and is the **only** route on Windows, where a socket joins exactly one completion port permanently. 7. **Per-agent timers.** Each loop's timer heap belongs to its agent, which replaces Perry's owner-tagged global timer queues. 8. **Web:** a loop per Web Worker instance. Cross-worker posting goes through host `postMessage`. With cross-origin isolation, a `SharedArrayBuffer` ring plus `Atomics.notify` can back a `Poster` without message copies. 9. **Contract tests:** - N loops on N threads cross-posting under load - completions only on the owning thread (debug assertion) - detach/attach under in-flight I/O - - reuse-port and accept-and-hand-off distribution + - reuse-port share/distribute behaviour, and refusal where a platform cannot honour the request + - accept-and-hand-off distribution, and N loops on N threads serving one port by each route with every connection served exactly once - signal fan-out - `waitAsync` service fairness - a loop on a non-main thread (iOS/Android shape) @@ -460,6 +463,7 @@ Two backends, because both versions matter now: | Timer wait precision | ns (epoll_pwait2 / timerfd) | ns (kevent timespec) | sub-ms via high-res waitable timer | host-dependent (Wasmtime ≈1 ms) | host-dependent (Wasmtime ≈1 ms) | host `setTimeout` (clamped by browser) | | TCP / UDP | non-blocking + readiness | non-blocking + readiness | overlapped Winsock | `wasi:sockets` | `wasi:sockets` | unsupported | | Socket options (§7.7) | full setsockopt set | full setsockopt set, no IPv4 membership by interface index | full Winsock set, no IPv4 membership by interface index | keep-alive, buffer sizes, hop limit | keep-alive, buffer sizes, hop limit | unsupported | +| Multi-threaded accept (§5a.6) | `ReusePort::Distribute` or handoff | handoff (`Distribute` refused; macOS gives the port to the last binder) | handoff only (one IOCP per socket, permanently) | unsupported (single-threaded; no reuse interface, no `detach`) | unsupported | unsupported (no listening sockets) | | Outbound HTTP | protocol crate | protocol crate | protocol crate | protocol crate or `wasi:http` | protocol crate or `wasi:http` | host `fetch` | | WebSocket | protocol crate | protocol crate | protocol crate | protocol crate | protocol crate | host `WebSocket` | | Local IPC | AF_UNIX | AF_UNIX | named pipes (overlapped) | unsupported | unsupported | `postMessage` | @@ -500,7 +504,10 @@ and work on any live socket handle, including one produced by `accept`. them, and the web backend refuses all of them. - **Bind-time options stay in the opts structs.** `SO_REUSEADDR`/`SO_REUSEPORT` cannot be changed on a bound socket, so they belong to `ListenOpts`/`UdpOpts`, - not to `SocketOption`. `IPV6_V6ONLY` is readable on a live socket and kept in + not to `SocketOption`. `reuse_port` is a `ReusePort` request + (`No`/`Share`/`Distribute`) rather than a `bool`, because `SO_REUSEPORT` means + different things on Linux and on the BSDs and a backend that cannot honour the + requested one refuses the listener (§5a.6). `IPV6_V6ONLY` is readable on a live socket and kept in the enum for that, but setting it after bind is refused by every OS. - **`ListenOpts::accept_defaults`** carries the per-connection defaults a server would otherwise apply by hand: `nodelay` and a keep-alive schedule, each at most diff --git a/README.md b/README.md index d6b2306..73e95ce 100644 --- a/README.md +++ b/README.md @@ -46,5 +46,6 @@ soak. The workspace also builds with stable Rust 1.97.1. - [Design and host boundary](DESIGN.md) - [Contributing, checks and private test servers](CONTRIBUTING.md) - [Release and first-publication procedure](RELEASING.md) +- [Multi-threaded accept](docs/multi-threaded-accept.md) - [Integration status and verification](docs/INTEGRATION_REPORT.md) - [MIT license](LICENSE) diff --git a/docs/lanes/sockopts.md b/docs/lanes/sockopts.md index 0bf8f2b..ce17e91 100644 --- a/docs/lanes/sockopts.md +++ b/docs/lanes/sockopts.md @@ -41,6 +41,7 @@ pub enum SocketOptionKind { NoDelay, KeepAlive, Linger, RecvBufferSize, MulticastTtl, MulticastLoop } pub struct ListenOpts { reuse_port: bool, backlog: u32, accept_defaults: AcceptDefaults } +// superseded: reuse_port is now a ReusePort request, see docs/multi-threaded-accept.md pub struct AcceptDefaults { nodelay: bool, keep_alive: Option } ``` diff --git a/docs/multi-threaded-accept.md b/docs/multi-threaded-accept.md new file mode 100644 index 0000000..ea801bd --- /dev/null +++ b/docs/multi-threaded-accept.md @@ -0,0 +1,216 @@ +# Multi-threaded accept + +DESIGN [§5a.6](../DESIGN.md#5a-multithreading). One server on more than one core, +without a work-stealing loop. + +A work-stealing loop is the wrong answer here and is deliberately not offered. +turnloop gives each JS agent its own loop on its own thread because JavaScript +values are thread-local: a completion has to be delivered on the thread whose +heap owns the promise it settles. A loop shared across cores would have to route +every completion back to its owning agent, which rebuilds per-agent queues and +wakeups with worse locality — the second scheduler this crate exists to remove. + +So the parallelism is in the *accept*, not in the loop. Two routes. + +| | Kernel-distributed | Handoff | +|---|---|---| +| Shape | N loops, N threads, N listeners, one port | 1 accepting loop, N sibling loops, `detach`/`attach` | +| Who decides | the kernel, by 4-tuple hash | the host, by whatever policy it writes | +| Linux, Android | `ReusePort::Distribute` (`SO_REUSEPORT`) | yes | +| FreeBSD | `ReusePort::Distribute` (`SO_REUSEPORT_LB`) | yes | +| macOS, other Apple, NetBSD, OpenBSD, DragonFly | **refused** (`Unsupported`) | yes — **the** route | +| Windows | **refused** (`Unsupported`) | yes — **the only** route | +| WASI 0.2/0.3, web | **refused** (`Unsupported`) | not available | + +## `ReusePort`: which behaviour you are asking for + +`SO_REUSEPORT` is spelled the same on Linux and on the BSDs and does not mean the +same thing, so `reuse_port` is not a `bool`. + +```rust +pub enum ReusePort { No, Share, Distribute } +``` + +* **`Share`** — permit the duplicate bind, promise nothing about delivery. This + is the traditional BSD use: multicast and broadcast receivers, and handing a + port to a replacement process during a zero-downtime restart, where + last-binder-wins is the effect you want. +* **`Distribute`** — permit the duplicate bind *and* spread incoming connections + across the listeners. Refused where the platform cannot do it. + +There is deliberately **no third outcome**. A platform either distributes or the +listener is refused when it is created. The failure this prevents is not +hypothetical, and it is silent: two loops sharing one port under plain +`SO_REUSEPORT` on macOS 15 split 32 connections **`[0, 32]`** — the first +listener is not merely under-served, it never accepts anything at all, for the +life of the process, with no error anywhere. A host that developed the Linux path +and shipped it would have a server that looks fine and uses one core. + +`Share` and `Distribute` are the same `setsockopt` on Linux, and that is fine: a +variant states a floor, not a ceiling. On FreeBSD they are genuinely different +options — plain `SO_REUSEPORT` keeps its original BSD meaning there, and +`SO_REUSEPORT_LB` (12.0+) is the one that distributes. + +### `Distribute` distributes by hash, not by load + +The kernel selects a listener by hashing the connection's address 4-tuple. +Nothing asks how busy a loop is. A loop whose agent is inside a long turn keeps +being handed its share, and those connections wait in its queue while another +loop is idle. + +Measured (Linux 6.8, contract test `multi_threaded_accept_by_reuse_port`), 64 +connections over 4 loops: + +``` +reuse-port: per-loop [7, 12, 21, 24] +``` + +A 3.4x spread. At larger connection counts the hash evens out — 4000 connections +over 4 loops gave `[970, 1016, 1019, 995]` — but the *work* behind each +connection is still not what was balanced. If a host needs load-sensitive +placement, that is the handoff route, where the policy is the host's to write. + +## The handoff route + +`Loop::detach(h) -> Detached` (which is `Send`) and `Loop::attach(d, tok)`. +`detach` cancels in-flight operations with their usual exactly-once completions +and refuses with `WouldBlock` until the transport is quiescent, so the sibling +loop adopts something with no operation, buffer, registration or completion +outstanding. + +**This works today on every native backend**, including Windows, and is covered +by contract tests that accept on one loop and then drive reads and writes on a +sibling loop on another thread. It is not a planned capability. + +On **Windows it is the only route**, for a reason that is structural rather than +an omission: a handle's IOCP association is permanent. Windows will not +dissociate one, rejects a second `CreateIoCompletionPort` with +`ERROR_INVALID_PARAMETER`, and neither `WSADuplicateSocketW` nor +`DuplicateHandle` escapes it — both produce another descriptor for the *same* +socket object, which is where the association lives. A second loop therefore +cannot share a listener, and `SO_REUSEADDR` is not a substitute: on Windows it +permits *hijacking* an address rather than sharing it, which is why it is not +mapped onto either `ReusePort` request. + +The handoff's cost, stated plainly: one extra `detach`/`attach` pair and one +cross-thread wake per connection, and an accepting loop that is a single point +of serialization for the accept itself. The kernel route has neither. That is +what macOS and Windows pay for not having `SO_REUSEPORT_LB`. + +## What the contract tests assert + +| test | asserts | +|---|---| +| `reuse_port_share_binds_twice` | the duplicate bind works and every connection is accepted by *someone*; deliberately asserts nothing about which listener, because `Share` promises nothing about it | +| `reuse_port_distribution` | no third outcome: the platform distributes, or the listener is refused with `Unsupported` and leaves nothing behind | +| `reuse_port_is_explicitly_unsupported` (Windows), `reuse_port_has_no_wasi_interface` (WASI) | both requests refused where there is no mechanism | +| `accept_handoff_distribution` | accept on one loop, drive I/O on a sibling loop, round-robin over 4 workers | +| `multi_threaded_accept_by_handoff` | 64 connections, 4 loops on 4 threads, **every connection served exactly once** — each client sends a distinct id and requires that exact id back — and every loop `!alive()` at shutdown | +| `multi_threaded_accept_by_reuse_port` | the same, by the kernel route, where it is available | + +The two `multi_threaded_accept_*` tests run on loops whose handle ceiling +(`max_handles: 8`) is far below the 64 connections they serve. That is the +orphan gate: the workload fits comfortably, but a handle leaked per connection — +at accept, at `attach`, at `detach` or at `close` — exhausts the ceiling long +before the run ends and turns an invisible orphan into a `ResourceLimit` failure. + +Both use **single-shot** accept, re-armed per connection. turnloop[#77] is open — +a multishot accept can outrun the handle ceiling within one turn — and a test +that ran multishot at a low ceiling would be exercising that open issue rather +than the accept route. + +[#77]: https://github.com/PerryTS/turnloop/issues/77 + +## Scaling harness + +`turnloop-bench --accept-scaling` drives N loops on N threads serving one port by +either route: accept, one request, one response, close. Connection-oriented on +purpose, because that is the shape where the accept path is the subject rather +than a rounding error. + +``` +cargo run --release -p turnloop-bench --locked -- --accept-scaling \ + --route handoff|reuse-port \ + --loops N --connections C --clients K --payload B [--portable] +``` + +`--portable` reports elapsed nanoseconds; without it the harness uses the +platform instruction counter (`perf` on Linux, `ri_instructions` on macOS, +`QueryProcessCycleTime` on Windows). Output is one JSON line with +`connections_per_second` and, beside it, `per_loop` — the service count for each +loop, so the distribution is visible next to the total instead of inferred from +it. + +The harness refuses to print anything unless its subject demonstrably ran: the +per-loop counts must sum to the budget, every loop must have served at least one +connection, and `--connections` must be at least 100 per loop so that a loop +cannot be given nothing by chance. A flat scaling curve produced by three of four +loops sitting idle is the failure mode this exists to make impossible. + +### Before believing any curve: saturate the clients first + +The client side is synchronous — `K` threads each looping connect, write, read, +close — so the achievable rate is bounded by `K / RTT` no matter how many server +loops are running. **A server-scaling curve measured against a saturated client +is a picture of the client.** + +So the first run is not a scaling run. Hold `--loops` at its largest value and +raise `--clients` until the number stops moving: + +``` +for k in 8 16 32 64 128; do + cargo run --release -p turnloop-bench --locked -- --accept-scaling \ + --route handoff --loops 8 --connections 40000 --clients $k --portable +done +``` + +Take the smallest `K` at the plateau, double it, and use that for every point in +the scaling sweep. If the plateau is below the single-loop rate, the client is +the bottleneck everywhere and no scaling conclusion is available from this host. + +### The sweep + +``` +for route in handoff reuse-port; do + for n in 1 2 4 8 16; do + cargo run --release -p turnloop-bench --locked -- --accept-scaling \ + --route "$route" --loops $n --connections 100000 --clients "$K" --portable + done +done +``` + +`--route reuse-port` exits 2 with one line on a platform that cannot distribute, +so a macOS or Windows sweep is the handoff row only. + +Run it on a quiet host. Interleave the arms rather than running all of one then +all of the other, so a drift in machine state does not land entirely on one +route. + +### What the results would mean + +Let `T(n)` be `connections_per_second` at `n` loops. + +* **The design works** if `T(n)/T(1)` grows close to linearly to the core count + and then flattens — for the kernel route with `per_loop` counts within roughly + ±10% of each other, and for the handoff route with them near-exactly equal + (round-robin is exact by construction, so anything else is a bug, not + imbalance). Expect the handoff route to start lower than the kernel route at + `n = 1` and to fall behind it as `n` grows: it pays a `detach`/`attach` pair + and a cross-thread wake per connection, and its accepting loop serializes the + accept. That gap is the price of macOS and Windows, and quantifying it is the + point of running both arms. +* **The design does not work** if `T(n)` is flat from `n = 1` — one server cannot + use more than one core by this route — or if it *falls* as `n` grows. + Before concluding either, rule out the client (above) and check `per_loop`: a + flat curve with lopsided `per_loop` is a distribution problem, not a scaling + ceiling. +* **The handoff route's acceptor is the ceiling** if `T(n)` for handoff plateaus + at a value the kernel route passes, while `per_loop` stays even. The accepting + loop is then saturated, and the next move is more than one accepting loop + (several acceptors each with their own listener under `Share`, each feeding a + subset of workers) rather than more workers. +* **Nothing at all** if a run fails its liveness assertions. The harness panics + rather than printing in that case, on purpose. + +No numbers are recorded here. The harness has been run only for correctness, on +loaded machines, where a timing figure would be worse than none. From 396eb77963000b59b35f78bc05d72f92228bdefc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 13:39:01 +0200 Subject: [PATCH 4/7] Resolve the reuse-port request before the socket exists A request the platform cannot honour now fails before socket() is called rather than after SO_REUSEADDR has been set on a descriptor that is about to be dropped. Nothing reaches the kernel for a listener that is refused. --- crates/turnloop/src/backend/unix.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs index 89c9467..125ecf7 100644 --- a/crates/turnloop/src/backend/unix.rs +++ b/crates/turnloop/src/backend/unix.rs @@ -608,6 +608,9 @@ unsafe impl Backend for Unix { if backlog > i32::MAX as u32 { return Err(Error::new(ErrorKind::InvalidInput)); } + // Resolved before the socket exists, so a request this platform cannot + // honour never reaches the kernel at all. + let reuse_option = super::socket::reuse_port_option(reuse)?; let fd = socket::create(addr, kind == Kind::Udp)?; if kind != Kind::Tcp { // TCP listeners need address reuse for TIME_WAIT. Default UDP binds @@ -616,7 +619,7 @@ unsafe impl Backend for Unix { if kind == Kind::Listener || reuse.is_enabled() { socket::option(fd.as_raw_fd(), libc::SOL_SOCKET, libc::SO_REUSEADDR, 1)?; } - if let Some(option) = super::socket::reuse_port_option(reuse)? { + if let Some(option) = reuse_option { socket::option(fd.as_raw_fd(), libc::SOL_SOCKET, option, 1)?; } let a = Addr::new(addr); From 1e1e279db4b589e0da3f98cb8bc76a073a1124d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 13:41:06 +0200 Subject: [PATCH 5/7] Correct the measurement hosts and the counter the harness actually has The figures were cited against macOS 15 and Linux 6.8; they were taken on macOS 26.5 arm64 and Linux 6.17 x86-64. A measurement whose host is wrong in the record cannot be reproduced or contradicted later. The runbook also claimed QueryProcessCycleTime on Windows. turnloop-bench's counter has a Linux arm and a macOS arm and nothing else, so Windows falls back to elapsed nanoseconds - which is the right unit for a scaling sweep in any case, because connections per second is the quantity of interest and an instruction count attributes only the measuring thread's work. --- DESIGN.md | 4 ++-- crates/turnloop-contract/src/extended.rs | 2 +- crates/turnloop/src/backend/socket.rs | 2 +- docs/multi-threaded-accept.md | 13 ++++++++----- 4 files changed, 12 insertions(+), 9 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index 6b81def..2454b19 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -270,8 +270,8 @@ There are two ways a host drives a loop: - **Reporting only:** `Loop::raw_transport(h) -> RawTransport` reports a live transport's native identity for Node's `socket._handle.fd`. The loop keeps ownership; the value is valid until the handle is closed or detached, and is for reporting and read-only queries, never for I/O, closing, mode changes or registration elsewhere. 6. **Multi-threaded accept.** One server, more than one core, without a work-stealing loop: a shared loop would have to route every completion back to its owning agent, which rebuilds the per-agent queues and wakeups this crate exists to remove. Two routes, and `ListenOpts::reuse_port` is a `ReusePort` request rather than a `bool` because the two are not the same promise. See [multi-threaded-accept.md](docs/multi-threaded-accept.md). - **Kernel-distributed** (`ReusePort::Distribute`): each loop gets its own listener on the same port and the kernel picks which loop accepts each connection. `SO_REUSEPORT` on **Linux and Android**; `SO_REUSEPORT_LB` on **FreeBSD** (12.0+) — plain `SO_REUSEPORT` there keeps its original BSD meaning and does *not* distribute, so it is not the same option. **`Unsupported` on macOS and the other Apple platforms, NetBSD, OpenBSD, DragonFly, Windows, WASI and the web**, where the listener is refused at creation instead of being bound and starved. - - **It distributes by hash, not by load.** The listener is chosen by hashing the connection's 4-tuple; nothing asks how busy a loop is, so a loop whose agent is in a long turn keeps being given its share and those connections wait. Measured on Linux 6.8, 64 connections over 4 loops: `[7, 12, 21, 24]`. Even distribution of connections is not even distribution of work. - - **`ReusePort::Share`** is the weaker request — duplicate binding, delivery unspecified — for multicast/broadcast receivers and zero-downtime restarts. It is honest about the BSD behaviour rather than hiding it: measured on macOS 15, two listeners sharing a port with `SO_REUSEPORT` split 32 connections `[0, 32]`, all to the last binder. + - **It distributes by hash, not by load.** The listener is chosen by hashing the connection's 4-tuple; nothing asks how busy a loop is, so a loop whose agent is in a long turn keeps being given its share and those connections wait. Measured on Linux 6.17 x86-64, 64 connections over 4 loops: `[7, 12, 21, 24]`. Even distribution of connections is not even distribution of work. + - **`ReusePort::Share`** is the weaker request — duplicate binding, delivery unspecified — for multicast/broadcast receivers and zero-downtime restarts. It is honest about the BSD behaviour rather than hiding it: measured on macOS 26.5 arm64, two listeners sharing a port with `SO_REUSEPORT` split 32 connections `[0, 32]`, all to the last binder. - **Handoff, everywhere else:** one accepting loop hands each connection to a sibling loop with `detach`/`attach`. The policy (round-robin, least-loaded) belongs to the host. This works today on every native backend and is the **only** route on Windows, where a socket joins exactly one completion port permanently. 7. **Per-agent timers.** Each loop's timer heap belongs to its agent, which replaces Perry's owner-tagged global timer queues. 8. **Web:** a loop per Web Worker instance. Cross-worker posting goes through host `postMessage`. With cross-origin isolation, a `SharedArrayBuffer` ring plus `Atomics.notify` can back a `Poster` without message copies. diff --git a/crates/turnloop-contract/src/extended.rs b/crates/turnloop-contract/src/extended.rs index 509d2dd..11274fd 100644 --- a/crates/turnloop-contract/src/extended.rs +++ b/crates/turnloop-contract/src/extended.rs @@ -237,7 +237,7 @@ pub fn reuse_port_share() { /// This is the gate that makes the option honest. There is no third outcome: a /// backend may not accept the request and then leave a listener starved. The /// starved case is real and is what this exists to prevent — two loops sharing -/// one port under plain `SO_REUSEPORT` on macOS 15 split 32 connections +/// one port under plain `SO_REUSEPORT` on macOS 26.5 split 32 connections /// `[0, 32]`, so a host that developed against Linux would ship a server whose /// first loop never accepts anything. pub fn reuse_port_distribute() { diff --git a/crates/turnloop/src/backend/socket.rs b/crates/turnloop/src/backend/socket.rs index 7fd40db..29e344a 100644 --- a/crates/turnloop/src/backend/socket.rs +++ b/crates/turnloop/src/backend/socket.rs @@ -111,7 +111,7 @@ impl Addr { /// requests are two different options here. /// * **macOS and the other Apple platforms, NetBSD, OpenBSD and DragonFly** have /// only the original option. A [`ReusePort::Distribute`] request is refused -/// rather than silently answered with `SO_REUSEPORT`: measured on macOS 15, +/// rather than silently answered with `SO_REUSEPORT`: measured on macOS 26.5, /// two loops binding one port with `SO_REUSEPORT` split 32 connections /// `[0, 32]` — the first listener is not merely under-served, it is never /// given anything at all. diff --git a/docs/multi-threaded-accept.md b/docs/multi-threaded-accept.md index ea801bd..1c544a1 100644 --- a/docs/multi-threaded-accept.md +++ b/docs/multi-threaded-accept.md @@ -41,7 +41,7 @@ pub enum ReusePort { No, Share, Distribute } There is deliberately **no third outcome**. A platform either distributes or the listener is refused when it is created. The failure this prevents is not hypothetical, and it is silent: two loops sharing one port under plain -`SO_REUSEPORT` on macOS 15 split 32 connections **`[0, 32]`** — the first +`SO_REUSEPORT` on macOS 26.5 arm64 split 32 connections **`[0, 32]`** — the first listener is not merely under-served, it never accepts anything at all, for the life of the process, with no error anywhere. A host that developed the Linux path and shipped it would have a server that looks fine and uses one core. @@ -58,7 +58,7 @@ Nothing asks how busy a loop is. A loop whose agent is inside a long turn keeps being handed its share, and those connections wait in its queue while another loop is idle. -Measured (Linux 6.8, contract test `multi_threaded_accept_by_reuse_port`), 64 +Measured (Linux 6.17 x86-64, contract test `multi_threaded_accept_by_reuse_port`), 64 connections over 4 loops: ``` @@ -134,9 +134,12 @@ cargo run --release -p turnloop-bench --locked -- --accept-scaling \ --loops N --connections C --clients K --payload B [--portable] ``` -`--portable` reports elapsed nanoseconds; without it the harness uses the -platform instruction counter (`perf` on Linux, `ri_instructions` on macOS, -`QueryProcessCycleTime` on Windows). Output is one JSON line with +`--portable` reports elapsed nanoseconds. Without it the harness uses the +platform instruction counter where `turnloop-bench` has one — `perf` on Linux, +`ri_instructions` on macOS — and falls back to elapsed nanoseconds everywhere +else, Windows included. For a scaling sweep `--portable` is the right choice +anyway: `connections_per_second` is the quantity of interest, and an instruction +count attributes only the measuring thread's work. Output is one JSON line with `connections_per_second` and, beside it, `per_loop` — the service count for each loop, so the distribution is visible next to the total instead of inferred from it. From e685b553d5d18d27bdf95ac510be21b89b2b73a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 13:42:40 +0200 Subject: [PATCH 6/7] Say why neither accept route has a thundering herd It is the usual objection to multi-threaded accept and it applies to a shape turnloop does not have: a herd comes from several threads waiting on one listening socket. Under Distribute each loop has its own listener and its own accept queue and exactly one loop wakes; under the handoff route exactly one loop is accepting at all. Each route's real cost is named instead. --- docs/multi-threaded-accept.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/docs/multi-threaded-accept.md b/docs/multi-threaded-accept.md index 1c544a1..cf0cd50 100644 --- a/docs/multi-threaded-accept.md +++ b/docs/multi-threaded-accept.md @@ -70,6 +70,18 @@ over 4 loops gave `[970, 1016, 1019, 995]` — but the *work* behind each connection is still not what was balanced. If a host needs load-sensitive placement, that is the handoff route, where the policy is the host's to write. +### Neither route has a thundering herd + +Worth saying because it is the usual objection to multi-threaded accept, and it +applies to a shape turnloop does not have. A herd comes from several threads +waiting on **one** listening socket and all being woken for one connection. + +Under `Distribute` each loop has its **own** listener with its own accept queue, +and the kernel delivers to exactly one of them, so exactly one loop wakes. Under +the handoff route exactly one loop is accepting at all. The cost each route pays +is elsewhere: uneven shares for the first, and a per-connection `detach`/`attach` +plus a cross-thread wake for the second. + ## The handoff route `Loop::detach(h) -> Detached` (which is `Send`) and `Loop::attach(d, tok)`. From 93460209f188bc5f45d0cceb0b072227bbcf717d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 13:54:07 +0200 Subject: [PATCH 7/7] Claim the client budget with a ticket, not a compare-exchange loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The client threads shared the connection budget through a countdown: `try_update(|n| n.checked_sub(1))`, which is a compare-exchange loop that retries whenever two client threads reach for the same connection at the same moment. That contention sits on the load-generating side of a harness whose whole job is to not be the bottleneck while the server scales, and it grows with `--clients` — exactly the knob the runbook says to raise first, until the client rate stops moving, before any server curve means anything. A ticket counter claims the same work with one wait-free `fetch_add`: each thread takes the next number and stops once the numbers run past the budget. Exactly `--connections` tickets are below the budget, so exactly that many connections are driven, which `driven == args.connections` still asserts. Verified on macOS 26.5 arm64 at 1, 2 and 4 loops: the budget is spent exactly and the per-loop counts sum to it. --- crates/turnloop-bench/src/scaling.rs | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/crates/turnloop-bench/src/scaling.rs b/crates/turnloop-bench/src/scaling.rs index fef634b..82f7fda 100644 --- a/crates/turnloop-bench/src/scaling.rs +++ b/crates/turnloop-bench/src/scaling.rs @@ -183,14 +183,15 @@ const ACCEPT: Token = Token(u64::MAX); /// Connect, send one payload, read it back, close. Repeated by each client /// thread until the run's connection budget is spent. -fn client(addr: SocketAddr, payload: usize, budget: &AtomicUsize) -> usize { +/// +/// `issued` is a ticket counter rather than a countdown: every thread takes the +/// next number and stops once the numbers run past the budget, so the work is +/// claimed exactly once with a single `fetch_add` and no compare-exchange loop. +fn client(addr: SocketAddr, payload: usize, issued: &AtomicUsize, budget: usize) -> usize { let request = vec![0x5a_u8; payload]; let mut response = vec![0_u8; payload]; let mut done = 0; - while budget - .try_update(Ordering::AcqRel, Ordering::Acquire, |n| n.checked_sub(1)) - .is_ok() - { + while issued.fetch_add(1, Ordering::AcqRel) < budget { let mut s = TcpStream::connect(addr).expect("connect"); s.set_nodelay(true).expect("nodelay"); s.set_read_timeout(Some(Duration::from_secs(30))) @@ -230,18 +231,18 @@ pub fn run(counter: &Counter, args: &Args) { } } let served = Arc::new(AtomicUsize::new(0)); - let budget = Arc::new(AtomicUsize::new(args.connections)); + let issued = Arc::new(AtomicUsize::new(0)); let (addr, threads) = match args.route { Route::ReusePort => reuse_port_servers(args, &served), Route::Handoff => handoff_servers(args, &served), }; let before = counter.read().expect("counter"); let start = Instant::now(); - let payload = args.payload; + let (payload, budget) = (args.payload, args.connections); let clients: Vec<_> = (0..args.clients) .map(|_| { - let budget = Arc::clone(&budget); - thread::spawn(move || client(addr, payload, &budget)) + let issued = Arc::clone(&issued); + thread::spawn(move || client(addr, payload, &issued, budget)) }) .collect(); let driven: usize = clients.into_iter().map(|c| c.join().expect("client")).sum();