net_io/
opts.rs

1//! Socket options: before bind, per accepted or dialed connection, and live.
2
3use std::time::Duration;
4
5use crate::error::NetError;
6use crate::sys::{self, AsSock};
7
8/// Keepalive probes: idle time before the first, interval between them, and count.
9///
10/// ```
11/// let k = net_io::Keepalive::DEFAULT;
12/// assert_eq!(k.count, 6);
13/// ```
14#[derive(Clone, Copy, Debug, PartialEq, Eq)]
15pub struct Keepalive {
16    /// Idle time before the first probe.
17    pub idle: Duration,
18    /// Time between probes.
19    pub interval: Duration,
20    /// Probes before the connection is dropped.
21    pub count: u32,
22}
23
24impl Keepalive {
25    /// 60 s, 10 s, 6.
26    pub const DEFAULT: Keepalive = Keepalive {
27        idle: Duration::from_secs(60),
28        interval: Duration::from_secs(10),
29        count: 6,
30    };
31}
32
33/// What happens to unsent data at close.
34///
35/// ```
36/// assert_ne!(net_io::Linger::Off, net_io::Linger::Abortive);
37/// ```
38#[derive(Clone, Copy, Debug, PartialEq, Eq)]
39pub enum Linger {
40    /// The OS default: close returns at once, the kernel sends what it can.
41    Off,
42    /// RST on close; kernel resources are freed at once.
43    Abortive,
44    /// Close waits up to this long for unsent data.
45    Timeout(Duration),
46}
47
48impl Linger {
49    fn to_sys(self) -> Option<Duration> {
50        match self {
51            Linger::Off => None,
52            Linger::Abortive => Some(Duration::ZERO),
53            Linger::Timeout(d) => Some(d),
54        }
55    }
56}
57
58/// Options applied to each accepted or dialed TCP connection. `None` leaves the OS default.
59///
60/// ```
61/// use net_io::{Keepalive, TcpOptions};
62/// let opts = TcpOptions::new().nodelay(true).keepalive(Keepalive::DEFAULT);
63/// assert_eq!(opts.nodelay, Some(true));
64/// ```
65#[must_use]
66#[non_exhaustive]
67#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
68pub struct TcpOptions {
69    /// `TCP_NODELAY`.
70    pub nodelay: Option<bool>,
71    /// Keepalive probes.
72    pub keepalive: Option<Keepalive>,
73    /// Linger at close.
74    pub linger: Option<Linger>,
75    /// `SO_SNDBUF`.
76    pub send_buffer: Option<u32>,
77    /// `SO_RCVBUF`.
78    pub receive_buffer: Option<u32>,
79    /// How long sent bytes may stay unacknowledged before the OS drops the connection
80    /// (`TCP_USER_TIMEOUT`; `TCP_MAXRT` on Windows, in whole seconds).
81    pub user_timeout: Option<Duration>,
82    /// Hop limit (`IP_TTL` / `IPV6_UNICAST_HOPS`).
83    pub ttl: Option<u8>,
84}
85
86impl TcpOptions {
87    /// Every option at the OS default.
88    pub fn new() -> TcpOptions {
89        TcpOptions::default()
90    }
91
92    /// `TCP_NODELAY`.
93    pub fn nodelay(mut self, on: bool) -> TcpOptions {
94        self.nodelay = Some(on);
95        self
96    }
97
98    /// Keepalive probes.
99    pub fn keepalive(mut self, k: Keepalive) -> TcpOptions {
100        self.keepalive = Some(k);
101        self
102    }
103
104    /// Linger at close.
105    pub fn linger(mut self, l: Linger) -> TcpOptions {
106        self.linger = Some(l);
107        self
108    }
109
110    /// `SO_SNDBUF`.
111    pub fn send_buffer(mut self, bytes: u32) -> TcpOptions {
112        self.send_buffer = Some(bytes);
113        self
114    }
115
116    /// `SO_RCVBUF`.
117    pub fn receive_buffer(mut self, bytes: u32) -> TcpOptions {
118        self.receive_buffer = Some(bytes);
119        self
120    }
121
122    /// How long sent bytes may stay unacknowledged before the OS drops the connection.
123    /// `Unsupported` on macOS and FreeBSD. The write timeout covers the same case
124    /// inside net-io; this also covers the bytes already in the kernel at a close.
125    pub fn user_timeout(mut self, timeout: Duration) -> TcpOptions {
126        self.user_timeout = Some(timeout);
127        self
128    }
129
130    /// Hop limit.
131    pub fn ttl(mut self, hops: u8) -> TcpOptions {
132        self.ttl = Some(hops);
133        self
134    }
135
136    /// Applies the options that are set to `stream`.
137    pub(crate) fn apply<S: AsSock>(&self, socket: &S, v6: bool) -> Result<(), NetError> {
138        let sock = socket.sock();
139        let os = |op: &'static str| move |e| NetError::from_os(op, e);
140        if let Some(on) = self.nodelay {
141            sys::set_nodelay(sock, on).map_err(os("TCP_NODELAY"))?;
142        }
143        if let Some(k) = self.keepalive {
144            check_keepalive(&k)?;
145            sys::set_keepalive(sock, Some((k.idle, k.interval, k.count)))
146                .map_err(os("keepalive"))?;
147        }
148        if let Some(l) = self.linger {
149            sys::set_linger(sock, l.to_sys()).map_err(os("SO_LINGER"))?;
150        }
151        if let Some(n) = self.send_buffer {
152            sys::set_send_buffer(sock, n).map_err(os("SO_SNDBUF"))?;
153        }
154        if let Some(n) = self.receive_buffer {
155            sys::set_receive_buffer(sock, n).map_err(os("SO_RCVBUF"))?;
156        }
157        if let Some(t) = self.user_timeout {
158            sys::set_user_timeout(sock, t).map_err(|e| {
159                if e.kind() == std::io::ErrorKind::Unsupported {
160                    NetError::Unsupported {
161                        feature: "user_timeout",
162                        platform: std::env::consts::OS,
163                    }
164                } else {
165                    NetError::from_os("TCP_USER_TIMEOUT", e)
166                }
167            })?;
168        }
169        if let Some(h) = self.ttl {
170            sys::set_ttl(sock, v6, h).map_err(os("IP_TTL"))?;
171        }
172        Ok(())
173    }
174
175    /// Whether any option is set.
176    pub(crate) fn is_empty(&self) -> bool {
177        *self == TcpOptions::default()
178    }
179}
180
181fn check_keepalive(k: &Keepalive) -> Result<(), NetError> {
182    if cfg!(windows) && k.count > 255 {
183        return Err(NetError::InvalidConfig {
184            field: "keepalive",
185            reason: crate::error::ConfigReason::TooLarge { max: 255 },
186        });
187    }
188    if k.idle.is_zero() || k.interval.is_zero() || k.count == 0 {
189        return Err(NetError::InvalidConfig {
190            field: "keepalive",
191            reason: crate::error::ConfigReason::Zero,
192        });
193    }
194    Ok(())
195}
196
197/// Options set on a listening socket before `bind` and `listen`.
198///
199/// ```
200/// use net_io::{ListenOptions, TcpListener};
201/// let options = ListenOptions::new().backlog(4096).reuse_port(cfg!(target_os = "linux"));
202/// let listener = TcpListener::bind_with("127.0.0.1:0", options)?;
203/// println!("effective backlog {:?}", listener.info().backlog);
204/// # Ok::<(), net_io::NetError>(())
205/// ```
206#[must_use]
207#[derive(Clone, Copy, Debug, PartialEq, Eq)]
208pub struct ListenOptions {
209    pub(crate) backlog: u32,
210    pub(crate) reuse_address: bool,
211    pub(crate) reuse_port: bool,
212    pub(crate) ipv6_only: bool,
213    pub(crate) send_buffer: Option<u32>,
214    pub(crate) receive_buffer: Option<u32>,
215    pub(crate) defer_accept: Option<Duration>,
216    pub(crate) fast_open: Option<u32>,
217    pub(crate) incoming_cpu: Option<u32>,
218    pub(crate) per_worker: bool,
219    pub(crate) tcp: TcpOptions,
220}
221
222impl Default for ListenOptions {
223    fn default() -> ListenOptions {
224        ListenOptions {
225            backlog: 1024,
226            reuse_address: cfg!(unix),
227            reuse_port: false,
228            ipv6_only: true,
229            send_buffer: None,
230            receive_buffer: None,
231            defer_accept: None,
232            fast_open: None,
233            incoming_cpu: None,
234            per_worker: false,
235            tcp: TcpOptions::new().nodelay(true),
236        }
237    }
238}
239
240impl ListenOptions {
241    /// Backlog 1,024; reuse address on Unix, exclusive address use on Windows;
242    /// IPv6-only; `TCP_NODELAY` on accepted sockets.
243    pub fn new() -> ListenOptions {
244        ListenOptions::default()
245    }
246
247    /// The listen backlog (capped by the OS; the effective value is in `ListenerInfo`).
248    pub fn backlog(mut self, n: u32) -> ListenOptions {
249        self.backlog = n;
250        self
251    }
252
253    /// Unix: `SO_REUSEADDR`. Windows: `false` = `SO_EXCLUSIVEADDRUSE`, `true` = `SO_REUSEADDR`.
254    pub fn reuse_address(mut self, on: bool) -> ListenOptions {
255        self.reuse_address = on;
256        self
257    }
258
259    /// `SO_REUSEPORT` (`Unsupported` on Windows).
260    pub fn reuse_port(mut self, on: bool) -> ListenOptions {
261        self.reuse_port = on;
262        self
263    }
264
265    /// `IPV6_V6ONLY` for IPv6 addresses (on by default; decision D-026).
266    pub fn ipv6_only(mut self, on: bool) -> ListenOptions {
267        self.ipv6_only = on;
268        self
269    }
270
271    /// `SO_SNDBUF` on the listener (before `listen`).
272    pub fn send_buffer(mut self, bytes: u32) -> ListenOptions {
273        self.send_buffer = Some(bytes);
274        self
275    }
276
277    /// `SO_RCVBUF` on the listener (before `listen`, so window scaling is negotiated).
278    pub fn receive_buffer(mut self, bytes: u32) -> ListenOptions {
279        self.receive_buffer = Some(bytes);
280        self
281    }
282
283    /// `accept` returns once the client has sent bytes, or after `wait`
284    /// (`TCP_DEFER_ACCEPT`, whole seconds; `Unsupported` off Linux). A client that
285    /// connects and stays silent costs no accept for that long. Cost: the gate sees
286    /// the connection later, and a server that speaks first must not use this.
287    pub fn defer_accept(mut self, wait: Option<Duration>) -> ListenOptions {
288        self.defer_accept = wait;
289        self
290    }
291
292    /// Server TCP Fast Open with this many pending requests. Saves a round trip for a
293    /// client that returns. Security: the first flight of data can be replayed by the
294    /// network, so turn this on only for requests that are safe to run twice; the gate
295    /// and the PROXY check run as usual. Linux also needs `net.ipv4.tcp_fastopen`.
296    pub fn fast_open(mut self, queue: Option<u32>) -> ListenOptions {
297        self.fast_open = queue;
298        self
299    }
300
301    /// Accepts on this listener the connections the given CPU received
302    /// (`SO_INCOMING_CPU`, Linux; `Unsupported` elsewhere). For one reuse-port
303    /// listener per CPU.
304    pub fn incoming_cpu(mut self, cpu: Option<u32>) -> ListenOptions {
305        self.incoming_cpu = cpu;
306        self
307    }
308
309    /// One socket and one accept thread per reactor worker, in a `SO_REUSEPORT` group
310    /// (Linux; `Unsupported` elsewhere, `InvalidConfig` on the blocking driver). Each
311    /// feeds its own worker, so no connection crosses to another worker, and no accept
312    /// thread is a bottleneck; with `ServerBuilder::placement` each socket also takes
313    /// the connections its worker's CPU received. Costs: a socket and a thread per
314    /// worker, the kernel's hash decides the balance, and such a listener cannot be
315    /// handed off: closing one socket of a group resets what is queued on it.
316    pub fn per_worker(mut self, on: bool) -> ListenOptions {
317        self.per_worker = on;
318        self.reuse_port = self.reuse_port || on;
319        self
320    }
321
322    /// Options applied to every accepted socket.
323    pub fn tcp_options(mut self, options: TcpOptions) -> ListenOptions {
324        self.tcp = options;
325        self
326    }
327
328    pub(crate) fn params(&self) -> sys::ListenParams {
329        sys::ListenParams {
330            reuse_address: self.reuse_address,
331            reuse_port: self.reuse_port,
332            ipv6_only: self.ipv6_only,
333            send_buffer: self.send_buffer,
334            receive_buffer: self.receive_buffer,
335            backlog: self.backlog,
336            defer_accept: self.defer_accept,
337            fast_open: self.fast_open,
338            incoming_cpu: self.incoming_cpu,
339        }
340    }
341}
342
343/// Live get and set on an open socket.
344///
345/// ```no_run
346/// use net_io::{Keepalive, Linger};
347/// let conn = net_io::connect("127.0.0.1:7000")?;
348/// conn.options().set_keepalive(Some(Keepalive::DEFAULT))?;
349/// conn.options().set_linger(Linger::Abortive)?;
350/// let effective = conn.options().set_receive_buffer(256 * 1024)?;
351/// println!("receive buffer is {effective} bytes");
352/// # Ok::<(), net_io::NetError>(())
353/// ```
354#[derive(Debug)]
355pub struct SocketOptions<'a> {
356    pub(crate) sock: sys::Sock<'a>,
357    pub(crate) v6: bool,
358}
359
360fn os(op: &'static str) -> impl Fn(std::io::Error) -> NetError {
361    move |e| NetError::from_os(op, e)
362}
363
364impl SocketOptions<'_> {
365    /// `TCP_NODELAY`.
366    ///
367    /// # Errors
368    ///
369    /// `Os` when the OS call fails.
370    pub fn nodelay(&self) -> Result<bool, NetError> {
371        sys::nodelay(self.sock).map_err(os("TCP_NODELAY"))
372    }
373
374    /// Sets `TCP_NODELAY`.
375    ///
376    /// # Errors
377    ///
378    /// `Os` when the OS call fails.
379    pub fn set_nodelay(&self, on: bool) -> Result<(), NetError> {
380        sys::set_nodelay(self.sock, on).map_err(os("TCP_NODELAY"))
381    }
382
383    /// Keepalive probes (`None` = off).
384    ///
385    /// # Errors
386    ///
387    /// `Os` when the OS call fails.
388    pub fn keepalive(&self) -> Result<Option<Keepalive>, NetError> {
389        Ok(sys::keepalive(self.sock)
390            .map_err(os("keepalive"))?
391            .map(|(idle, interval, count)| Keepalive {
392                idle,
393                interval,
394                count,
395            }))
396    }
397
398    /// Sets keepalive probes (`None` = off).
399    ///
400    /// # Errors
401    ///
402    /// `InvalidConfig` for a zero value or a count above 255 on Windows; `Os`.
403    pub fn set_keepalive(&self, k: Option<Keepalive>) -> Result<(), NetError> {
404        if let Some(k) = &k {
405            check_keepalive(k)?;
406        }
407        sys::set_keepalive(self.sock, k.map(|k| (k.idle, k.interval, k.count)))
408            .map_err(os("keepalive"))
409    }
410
411    /// Linger at close.
412    ///
413    /// # Errors
414    ///
415    /// `Os` when the OS call fails.
416    pub fn linger(&self) -> Result<Linger, NetError> {
417        Ok(match sys::linger(self.sock).map_err(os("SO_LINGER"))? {
418            None => Linger::Off,
419            Some(d) if d.is_zero() => Linger::Abortive,
420            Some(d) => Linger::Timeout(d),
421        })
422    }
423
424    /// Sets linger at close.
425    ///
426    /// # Errors
427    ///
428    /// `Os` when the OS call fails.
429    pub fn set_linger(&self, l: Linger) -> Result<(), NetError> {
430        sys::set_linger(self.sock, l.to_sys()).map_err(os("SO_LINGER"))
431    }
432
433    /// The effective `SO_SNDBUF`.
434    ///
435    /// # Errors
436    ///
437    /// `Os` when the OS call fails.
438    pub fn send_buffer(&self) -> Result<u32, NetError> {
439        sys::send_buffer(self.sock).map_err(os("SO_SNDBUF"))
440    }
441
442    /// Sets `SO_SNDBUF` and returns the effective value (Linux doubles it).
443    ///
444    /// # Errors
445    ///
446    /// `Os` when the OS call fails.
447    pub fn set_send_buffer(&self, bytes: u32) -> Result<u32, NetError> {
448        sys::set_send_buffer(self.sock, bytes).map_err(os("SO_SNDBUF"))
449    }
450
451    /// The effective `SO_RCVBUF`.
452    ///
453    /// # Errors
454    ///
455    /// `Os` when the OS call fails.
456    pub fn receive_buffer(&self) -> Result<u32, NetError> {
457        sys::receive_buffer(self.sock).map_err(os("SO_RCVBUF"))
458    }
459
460    /// Sets `SO_RCVBUF` and returns the effective value (Linux doubles it).
461    ///
462    /// # Errors
463    ///
464    /// `Os` when the OS call fails.
465    pub fn set_receive_buffer(&self, bytes: u32) -> Result<u32, NetError> {
466        sys::set_receive_buffer(self.sock, bytes).map_err(os("SO_RCVBUF"))
467    }
468
469    /// The hop limit.
470    ///
471    /// # Errors
472    ///
473    /// `Os` when the OS call fails.
474    pub fn ttl(&self) -> Result<u8, NetError> {
475        sys::ttl(self.sock, self.v6).map_err(os("IP_TTL"))
476    }
477
478    /// Sets the hop limit.
479    ///
480    /// # Errors
481    ///
482    /// `Os` when the OS call fails.
483    pub fn set_ttl(&self, hops: u8) -> Result<(), NetError> {
484        sys::set_ttl(self.sock, self.v6, hops).map_err(os("IP_TTL"))
485    }
486}
487
488/// Options for a Unix socket listener. The defaults: mode `0600` (only the
489/// owner may connect, decision D-024), stale socket files removed, backlog 1,024.
490///
491/// ```
492/// let options = net_io::LocalOptions::new().group_access(33).require_uid(1000);
493/// # let _ = options;
494/// ```
495#[must_use]
496#[derive(Clone, Debug, PartialEq, Eq)]
497pub struct LocalOptions {
498    pub(crate) mode: u32,
499    pub(crate) owner: Option<u32>,
500    pub(crate) group: Option<u32>,
501    pub(crate) remove_stale: bool,
502    pub(crate) require_uid: Option<u32>,
503    pub(crate) backlog: u32,
504}
505
506impl Default for LocalOptions {
507    fn default() -> LocalOptions {
508        LocalOptions::new()
509    }
510}
511
512impl LocalOptions {
513    /// Mode `0600`, stale socket files removed, backlog 1,024.
514    pub fn new() -> LocalOptions {
515        LocalOptions {
516            mode: 0o600,
517            owner: None,
518            group: None,
519            remove_stale: true,
520            require_uid: None,
521            backlog: 1024,
522        }
523    }
524
525    /// The socket file's permission bits (only the low 9 bits are used).
526    pub fn mode(mut self, mode: u32) -> LocalOptions {
527        self.mode = mode & 0o777;
528        self
529    }
530
531    /// The socket file's owner (needs the privilege to change owners).
532    pub fn owner(mut self, uid: u32) -> LocalOptions {
533        self.owner = Some(uid);
534        self
535    }
536
537    /// The socket file's group; the mode is unchanged.
538    pub fn group(mut self, gid: u32) -> LocalOptions {
539        self.group = Some(gid);
540        self
541    }
542
543    /// The socket file's group and mode `0660`: members of `gid` may connect.
544    pub fn group_access(mut self, gid: u32) -> LocalOptions {
545        self.group = Some(gid);
546        self.mode = 0o660;
547        self
548    }
549
550    /// Whether a socket file nobody listens on is removed before binding
551    /// (default on). A live listener's file, or anything that is not a socket,
552    /// is never removed.
553    pub fn remove_stale(mut self, on: bool) -> LocalOptions {
554        self.remove_stale = on;
555        self
556    }
557
558    /// Refuses peers whose user id is not `uid`, at accept, before any read.
559    pub fn require_uid(mut self, uid: u32) -> LocalOptions {
560        self.require_uid = Some(uid);
561        self
562    }
563
564    /// The listen backlog (capped by the OS).
565    pub fn backlog(mut self, backlog: u32) -> LocalOptions {
566        self.backlog = backlog;
567        self
568    }
569}
570
571#[cfg(test)]
572mod tests {
573    use super::*;
574
575    #[test]
576    fn test_local_options_builders() {
577        let o = LocalOptions::new();
578        assert_eq!((o.mode, o.remove_stale, o.backlog), (0o600, true, 1024));
579        let o = o
580            .group_access(7)
581            .owner(3)
582            .require_uid(9)
583            .mode(0o1777)
584            .backlog(5);
585        assert_eq!(o.mode, 0o777);
586        assert_eq!(
587            (o.group, o.owner, o.require_uid),
588            (Some(7), Some(3), Some(9))
589        );
590        assert_eq!(LocalOptions::default(), LocalOptions::new());
591    }
592
593    #[test]
594    fn test_keepalive_zero_rejected() {
595        let k = Keepalive {
596            idle: Duration::ZERO,
597            interval: Duration::from_secs(1),
598            count: 1,
599        };
600        assert!(check_keepalive(&k).is_err());
601        assert!(check_keepalive(&Keepalive::DEFAULT).is_ok());
602    }
603
604    #[test]
605    fn test_listen_defaults() {
606        let o = ListenOptions::new();
607        assert_eq!(o.backlog, 1024);
608        assert!(o.ipv6_only);
609        assert_eq!(o.reuse_address, cfg!(unix));
610        assert_eq!(o.tcp.nodelay, Some(true));
611    }
612
613    #[test]
614    fn test_tcp_options_is_empty() {
615        assert!(TcpOptions::new().is_empty());
616        assert!(!TcpOptions::new().ttl(5).is_empty());
617    }
618}
net_io/opts.rs

net_io/
opts.rs
