kumo_server_common/
listener.rs

1//! Binding of TCP listeners with an operator-controllable accept queue.
2//!
3//! The async runtime builds listeners through mio, which hardcodes the
4//! `listen(2)` backlog to 128 with no way to override it. That queue has been
5//! observed to overflow under a connection burst well before CPU saturates,
6//! dropping connections the kernel could otherwise have queued. This module
7//! binds the socket directly so the backlog can be set, defaulting to the
8//! kernel's ceiling.
9
10use anyhow::Context;
11use socket2::{Domain, Protocol, Socket, Type};
12use std::net::{TcpListener, ToSocketAddrs};
13
14/// Returns the kernel's maximum accept-queue length, which is the largest
15/// backlog `listen(2)` will honor. A larger request is silently clamped down
16/// to this value. On Linux this is `net.core.somaxconn`. Elsewhere, and when
17/// it cannot be read, returns a conservative default that the kernel will
18/// clamp down further if needed.
19pub fn kernel_max_backlog() -> u32 {
20    #[cfg(target_os = "linux")]
21    {
22        if let Ok(contents) = std::fs::read_to_string("/proc/sys/net/core/somaxconn") {
23            if let Ok(value) = contents.trim().parse::<u32>() {
24                return value;
25            }
26        }
27    }
28    1024
29}
30
31/// Binds a listening TCP socket at `listen`, requesting an accept queue of
32/// `backlog` entries, or the kernel maximum when `backlog` is `None`. The
33/// returned listener is in non-blocking mode, ready to hand to an async
34/// runtime.
35///
36/// A `backlog` larger than the kernel's ceiling is accepted here but silently
37/// clamped by `listen(2)`.
38pub fn bind_with_backlog(listen: &str, backlog: Option<u32>) -> anyhow::Result<TcpListener> {
39    let backlog = backlog.unwrap_or_else(kernel_max_backlog);
40    let backlog = backlog.min(i32::MAX as u32) as i32;
41
42    let addrs = listen
43        .to_socket_addrs()
44        .with_context(|| format!("resolving {listen}"))?;
45
46    let mut last_err = None;
47    for addr in addrs {
48        match bind_one(addr, backlog) {
49            Ok(listener) => return Ok(listener),
50            Err(err) => last_err = Some(err),
51        }
52    }
53
54    Err(last_err.unwrap_or_else(|| anyhow::anyhow!("{listen} did not resolve to any address")))
55}
56
57fn bind_one(addr: std::net::SocketAddr, backlog: i32) -> anyhow::Result<TcpListener> {
58    let socket = Socket::new(Domain::for_address(addr), Type::STREAM, Some(Protocol::TCP))
59        .context("creating socket")?;
60    // Match the runtime's listeners, which set SO_REUSEADDR: without it, a
61    // restart cannot rebind the address until the previous socket's TIME_WAIT
62    // state expires.
63    socket
64        .set_reuse_address(true)
65        .context("setting SO_REUSEADDR")?;
66    socket
67        .bind(&addr.into())
68        .with_context(|| format!("binding to {addr}"))?;
69    socket
70        .listen(backlog)
71        .with_context(|| format!("listening on {addr}"))?;
72    socket
73        .set_nonblocking(true)
74        .context("setting nonblocking")?;
75    Ok(socket.into())
76}
kumo_server_common/listener.rs

kumo_server_common/
listener.rs
