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/src/signal.rs b/src/signal.rs index 959996e..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; @@ -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()); }); } 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()); +}