Let listeners on several reactors share a port (SO_REUSEPORT)

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"

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<Self> {
+        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<Self> {
         let registration = handle.register(listener.raw(), Interest::READABLE)?;

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> {
+        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<Self> {
+        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<Self> {
         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() };

src/signal.rs

@@ -243,7 +243,11 @@ struct Claim(i32);
 
 impl Claim {
     fn acquire(sig: i32) -> Result<Self> {
-        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;
@@ -762,21 +766,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());
         });
     }

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());
+}