diff --git a/DESIGN.md b/DESIGN.md index 1385547..2454b19 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.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. 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/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..82f7fda --- /dev/null +++ b/crates/turnloop-bench/src/scaling.rs @@ -0,0 +1,459 @@ +//! 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. +/// +/// `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 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))) + .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 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, budget) = (args.payload, args.connections); + let clients: Vec<_> = (0..args.clients) + .map(|_| { + 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(); + 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) +} diff --git a/crates/turnloop-contract/src/extended.rs b/crates/turnloop-contract/src/extended.rs index 1b5d336..11274fd 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 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() { + 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..29e344a 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 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. +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..125ecf7 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, @@ -608,16 +608,19 @@ 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 // 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) = reuse_option { + 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 +1251,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. 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..cf0cd50 --- /dev/null +++ b/docs/multi-threaded-accept.md @@ -0,0 +1,231 @@ +# 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 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. + +`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.17 x86-64, 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. + +### 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)`. +`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 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. + +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.