From 4b7587e9d4c8929cd9d812b64a4338be010f5512 Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 24 Sep 2026 03:23:52 +0700 Subject: [PATCH 1/3] Let listeners on several reactors share a port with SO_REUSEPORT A socket accepted by a listener is registered with that listener's reactor, so a server that wants more than one reactor needs a listening socket per reactor. There was no way to make one: bind set only SO_REUSEADDR, so a second bind on the port failed. endpoint-libs serves on one thread because of it and warns that WS_SHARDS is ignored. Listener::bind_shared and TcpListener::bind_shared set SO_REUSEPORT too. Linux balances connections across the sockets; macOS accepts the binds without balancing. Refused for Unix-domain addresses. Release as 0.1.13. --- Cargo.toml | 2 +- src/net.rs | 7 +++++++ src/reactor/socket.rs | 25 +++++++++++++++++++++++++ tests/reactor_net.rs | 18 ++++++++++++++++++ 4 files changed, 51 insertions(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index bb5abe5..d24838a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nagoya" -version = "0.1.12" +version = "0.1.13" edition = "2021" rust-version = "1.85" license = "MIT OR Apache-2.0" diff --git a/src/net.rs b/src/net.rs index 002958e..e32212a 100644 --- a/src/net.rs +++ b/src/net.rs @@ -516,6 +516,13 @@ impl TcpListener { Self::from_listener(listener, handle) } + /// Bind with `SO_REUSEPORT` and register, so that listeners on several + /// reactors can share `addr`. See [`Listener::bind_shared`]. + pub fn bind_shared(addr: Addr, handle: &Handle) -> Result { + let listener = Listener::bind_shared(addr, Self::BACKLOG)?; + Self::from_listener(listener, handle) + } + /// Adopt an already bound listener. pub fn from_listener(listener: Listener, handle: &Handle) -> Result { let registration = handle.register(listener.raw(), Interest::READABLE)?; diff --git a/src/reactor/socket.rs b/src/reactor/socket.rs index e94eea5..0dfc7c6 100644 --- a/src/reactor/socket.rs +++ b/src/reactor/socket.rs @@ -834,6 +834,28 @@ impl TcpListener { /// it, because a bind that quietly removes whatever it finds would remove a /// live server's socket. pub fn bind(addr: Addr, backlog: i32) -> Result { + Self::bind_with(addr, backlog, false) + } + + /// Bind with `SO_REUSEPORT`, so several listeners can share one port. + /// + /// This is how one server listens on many reactors: each binds its own + /// socket to the same address, and the kernel hands every connection to + /// exactly one of them. An accepted socket is then registered with the + /// reactor that accepted it and never has to cross a thread. Every socket + /// on the port must ask for this, the first included. + /// + /// Linux balances connections across the sockets. macOS and the BSDs + /// accept the binds but do not balance, so there it is correct but not + /// faster. Not meaningful for Unix-domain sockets, and refused for them. + pub fn bind_shared(addr: Addr, backlog: i32) -> Result { + if addr.family() == libc::AF_UNIX { + return Err(Errno(libc::EINVAL)); + } + Self::bind_with(addr, backlog, true) + } + + fn bind_with(addr: Addr, backlog: i32, reuse_port: bool) -> Result { let family = addr.family(); let fd = stream_socket(&addr)?; // Without this, a restart fails to bind while the previous socket's @@ -842,6 +864,9 @@ impl TcpListener { if family != libc::AF_UNIX { set_flag(fd.raw(), libc::SOL_SOCKET, libc::SO_REUSEADDR, true)?; } + if reuse_port { + set_flag(fd.raw(), libc::SOL_SOCKET, libc::SO_REUSEPORT, true)?; + } // SAFETY: zeroed storage, then written by `write_to`. let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() }; diff --git a/tests/reactor_net.rs b/tests/reactor_net.rs index 4346258..d794e22 100644 --- a/tests/reactor_net.rs +++ b/tests/reactor_net.rs @@ -296,3 +296,21 @@ fn a_write_of_the_whole_send_buffer_fills_it() { Err(_) => panic!("follow-up write failed for an unrelated reason"), } } + +#[test] +fn shared_listeners_on_two_reactors_hold_one_port() { + let first_reactor = Reactor::local().expect("reactor"); + let first = + TcpListener::bind_shared(Addr::localhost(0), &first_reactor.handle()).expect("bind"); + let addr = first.local_addr().expect("addr"); + + // A second reactor, as a second shard would have, binds the same port. + let second_reactor = Reactor::local().expect("reactor"); + let second = TcpListener::bind_shared(addr, &second_reactor.handle()) + .expect("second bind on the same port"); + assert_eq!(second.local_addr().expect("addr"), addr); + + // A plain bind does not share: the port is taken. + let third_reactor = Reactor::local().expect("reactor"); + assert!(TcpListener::bind(addr, &third_reactor.handle()).is_err()); +} From d278dbab1a4676092378979de32b01de9b6b3181 Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 24 Sep 2026 05:15:07 +0700 Subject: [PATCH 2/3] Aim the cross-thread signal test at the waiting thread, and bound it The test called raise(SIGHUP) on a helper thread, and raise targets the calling thread. On Linux that helper inherited the block Signal::new takes, so SIGHUP stayed pending on the helper, and signalfd, which reports only its reader's and process-directed signals, never became readable. The waiter blocked forever, holding the test lock, and the job ran into CI's two-minute limit with a_parked_waiter queued behind it. pthread_kill the waiter instead, which is the case the test is named for, and wrap the wait in a five-second timeout so a regression fails in seconds. --- src/signal.rs | 25 +++++++++++++++++++------ 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/src/signal.rs b/src/signal.rs index 959996e..d4f8593 100644 --- a/src/signal.rs +++ b/src/signal.rs @@ -762,21 +762,34 @@ mod tests { let _guard = lock(); let reactor = Reactor::local().expect("reactor"); let mut signal = Signal::new(SignalKind::hangup(), &reactor.handle()).expect("signal"); + // Aimed at this thread, the waiter. Not `raise` on the other thread: + // that targets the raising thread, which on Linux inherited the block + // `Signal::new` took, so the signal stayed pending there and + // `signalfd`, which reports only its reader's and process-wide + // signals, never saw it. The test hung until CI cancelled the job. + let waiter = unsafe { libc::pthread_self() } as usize; let started = AtomicBool::new(false); std::thread::scope(|scope| { scope.spawn(|| { while !started.load(Ordering::Acquire) { std::thread::yield_now(); } - // The waiter needs to reach the kernel wait. A raise that - // lands before that is still delivered, by the byte already - // being in the pipe; the sleep is what makes this the parked - // case rather than that one. + // The waiter needs to reach the kernel wait. A signal that + // lands before that is still delivered, by being pending + // already; the sleep is what makes this the parked case. std::thread::sleep(Duration::from_millis(50)); - unsafe { libc::raise(libc::SIGHUP) }; + let status = unsafe { libc::pthread_kill(waiter as libc::pthread_t, libc::SIGHUP) }; + assert_eq!(status, 0, "pthread_kill"); }); started.store(true, Ordering::Release); - let got = block_on_with(&reactor, signal.recv()).expect("recv"); + // Bounded, so a regression fails here in seconds rather than + // holding the test lock until CI gives up on the job. + let got = block_on_with( + &reactor, + crate::timeout(Duration::from_secs(5), signal.recv()), + ) + .expect("no SIGHUP within 5s") + .expect("recv"); assert_eq!(got, SignalKind::hangup()); }); } From 5a4b0f6927d9d92f01d27d5d6d4b663eb8b3ae2d Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 24 Sep 2026 05:19:10 +0700 Subject: [PATCH 3/3] Refuse SIGKILL and SIGSTOP before asking the platform The BSDs refuse them at sigaction. Linux does not: signalfd drops an uncatchable signal from its mask without an error, so Signal::new(SIGKILL) succeeded there and returned a waiter that could never fire. The test that says so had never run on Linux, because the cross-thread test hung first. Checked in Claim::acquire now, so both platforms answer EINVAL. --- src/signal.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/signal.rs b/src/signal.rs index d4f8593..d98d283 100644 --- a/src/signal.rs +++ b/src/signal.rs @@ -243,7 +243,11 @@ struct Claim(i32); impl Claim { fn acquire(sig: i32) -> Result { - if !(1..32).contains(&sig) { + // SIGKILL and SIGSTOP cannot be caught, and `sigaction` says so on + // the BSDs. On Linux `signalfd` silently drops them from its mask and + // succeeds, which would hand back a waiter that can never fire. Refused + // here, before either platform is asked, so both answer the same. + if !(1..32).contains(&sig) || sig == libc::SIGKILL || sig == libc::SIGSTOP { return Err(Errno(libc::EINVAL)); } let bit = 1u64 << sig;