vane_core/
net.rs

1//! Socket setup helpers: listeners with `SO_REUSEPORT` (one per worker for
2//! share-nothing accept), nonblocking tuning, keepalive.
3
4use std::io;
5use std::net::{SocketAddr, TcpListener as StdListener};
6use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
7
8use socket2::{Domain, Protocol, Socket, Type};
9
10/// Creates a nonblocking TCP listener bound to `addr`.
11///
12/// Sets `SO_REUSEADDR`; on Linux also `SO_REUSEPORT` when `reuse_port` is
13/// set so every worker can bind the same address for per-core accept.
14///
15/// # Errors
16/// Bind/listen failure.
17pub fn tcp_listener(addr: SocketAddr, reuse_port: bool, backlog: i32) -> io::Result<StdListener> {
18    let domain = if addr.is_ipv4() {
19        Domain::IPV4
20    } else {
21        Domain::IPV6
22    };
23    let sock = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))?;
24    sock.set_nonblocking(true)?;
25    sock.set_reuse_address(true)?;
26    if reuse_port {
27        // SAFETY: plain setsockopt with a valid option value.
28        set_sock_opt_bool(sock.as_raw_fd(), libc::SOL_SOCKET, libc::SO_REUSEPORT, true)?;
29    }
30    // IPv6 dual-stack explicit (bind v6 only when v6 addr given).
31    if addr.ip().is_ipv6() {
32        set_sock_opt_bool(
33            sock.as_raw_fd(),
34            libc::IPPROTO_IPV6,
35            libc::IPV6_V6ONLY,
36            true,
37        )?;
38    }
39    sock.bind(&addr.into())?;
40    sock.listen(backlog)?;
41    // SAFETY: single ownership transfer of a live listening socket fd.
42    Ok(unsafe { StdListener::from_raw_fd(sock.into_raw_fd()) })
43}
44
45/// Sets `TCP_NODELAY` (request latency over Nagle).
46///
47/// # Errors
48/// setsockopt failure.
49pub fn set_nodelay(fd: RawFd) -> io::Result<()> {
50    set_sock_opt_bool(fd, libc::IPPROTO_TCP, libc::TCP_NODELAY, true)
51}
52
53/// Sets aggressive keepalive so dead upstreams are noticed.
54///
55/// # Errors
56/// setsockopt failure.
57pub fn set_keepalive(fd: RawFd, idle_secs: u32) -> io::Result<()> {
58    set_sock_opt_bool(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE, true)?;
59    // SAFETY: integer options with correct sizes.
60    unsafe {
61        let v: libc::c_int = idle_secs as libc::c_int;
62        libc::setsockopt(
63            fd,
64            libc::IPPROTO_TCP,
65            libc::TCP_KEEPIDLE,
66            std::ptr::addr_of!(v).cast(),
67            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
68        );
69        let i: libc::c_int = 3;
70        libc::setsockopt(
71            fd,
72            libc::IPPROTO_TCP,
73            libc::TCP_KEEPINTVL,
74            std::ptr::addr_of!(i).cast(),
75            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
76        );
77        let c: libc::c_int = 3;
78        libc::setsockopt(
79            fd,
80            libc::IPPROTO_TCP,
81            libc::TCP_KEEPCNT,
82            std::ptr::addr_of!(c).cast(),
83            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
84        );
85    }
86    Ok(())
87}
88
89fn set_sock_opt_bool(fd: RawFd, level: libc::c_int, name: libc::c_int, on: bool) -> io::Result<()> {
90    let v: libc::c_int = i32::from(on);
91    // SAFETY: integer-valued setsockopt with the option's documented size.
92    let rc = unsafe {
93        libc::setsockopt(
94            fd,
95            level,
96            name,
97            std::ptr::addr_of!(v).cast(),
98            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
99        )
100    };
101    if rc == 0 {
102        Ok(())
103    } else {
104        Err(io::Error::last_os_error())
105    }
106}
107
108/// Raw stream descriptor owned by a session. Closed on drop.
109#[derive(Debug)]
110pub struct StreamFd(pub RawFd);
111
112impl StreamFd {
113    /// Underlying descriptor.
114    #[must_use]
115    pub fn fd(&self) -> RawFd {
116        self.0
117    }
118}
119
120impl Drop for StreamFd {
121    fn drop(&mut self) {
122        // SAFETY: single close of an owned descriptor.
123        unsafe { libc::close(self.0) };
124    }
125}
126
127/// `shutdown(SHUT_WR)` — half-close (finishes streaming responses).
128pub fn shutdown_write(fd: RawFd) {
129    // SAFETY: live fd; shutdown failure is ignorable.
130    unsafe {
131        let _ = libc::shutdown(fd, libc::SHUT_WR);
132    }
133}
134
135#[cfg(test)]
136mod net_tests {
137    use super::*;
138
139    #[test]
140    fn tcp_listener_reuseport_binds() {
141        let addr: SocketAddr = "127.0.0.1:0".parse().expect("addr");
142        let l = tcp_listener(addr, true, 64).expect("bind");
143        assert!(l.local_addr().is_ok());
144    }
145
146    #[test]
147    fn set_nodelay_and_keepalive_on_socket() {
148        let a = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
149        let addr = a.local_addr().expect("addr");
150        let client = std::net::TcpStream::connect(addr).expect("connect");
151        let fd = client.as_raw_fd();
152        set_nodelay(fd).expect("nodelay");
153        set_keepalive(fd, 30).expect("keepalive");
154    }
155
156    #[test]
157    fn set_nodelay_rejects_bad_fd() {
158        assert!(set_nodelay(-1).is_err());
159    }
160
161    #[test]
162    fn shutdown_write_sends_fin() {
163        use std::io::Read as _;
164        let a = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
165        let addr = a.local_addr().expect("addr");
166        let client = std::net::TcpStream::connect(addr).expect("connect");
167        let mut server = a.incoming().next().unwrap().expect("accept");
168        shutdown_write(client.as_raw_fd());
169        // Peer sees EOF.
170        let mut buf = [0u8; 1];
171        assert_eq!(server.read(&mut buf).expect("read"), 0);
172    }
173}
vane_core/net.rs

vane_core/
net.rs
