net_io/server/
mod.rs

1//! The server: listeners, workers, the door, drain, and shutdown.
2
3mod engine;
4#[cfg(feature = "reactor")]
5mod reactor;
6mod source;
7
8use std::any::TypeId;
9use std::fmt;
10use std::net::{SocketAddr, ToSocketAddrs};
11use std::path::PathBuf;
12use std::sync::Arc;
13use std::thread::JoinHandle;
14use std::time::{Duration, Instant};
15
16use crate::control::ControlLine;
17use crate::driver::DriverKind;
18use crate::error::{CloseReason, ConfigReason, KillReason, NetError, RefusalCause};
19use crate::events::{EventSink, NoopSink};
20use crate::gate::Gate;
21use crate::handoff::ListenerHandle;
22use crate::id::{ConnectionId, ListenerId, WorkerId};
23use crate::limits::Limits;
24use crate::opts::{ListenOptions, LocalOptions};
25use crate::proxy::ProxyMode;
26use crate::session::SessionFactory;
27use crate::tcp::{ListenerInfo, TcpConnection};
28use crate::time::Clock;
29use crate::tls::ErasedAcceptor;
30use crate::tls::{NoTls, TlsAcceptor};
31use crate::trusted::TrustedProxies;
32
33pub(crate) use engine::Shared;
34
35/// Whether the server is accepting.
36///
37/// ```
38/// assert_ne!(net_io::DoorState::Open, net_io::DoorState::Closed);
39/// ```
40#[derive(Clone, Copy, Debug, PartialEq, Eq)]
41pub enum DoorState {
42    /// Accepting new connections.
43    Open,
44    /// Not accepting; listening sockets stay open.
45    Closed,
46}
47
48/// What a drain did.
49///
50/// ```
51/// let r = net_io::DrainReport { graceful: 3, killed: 1, took: std::time::Duration::from_millis(40) };
52/// assert_eq!(r.graceful + r.killed, 4);
53/// ```
54#[derive(Clone, Copy, Debug, PartialEq, Eq)]
55pub struct DrainReport {
56    /// Connections that finished before the deadline.
57    pub graceful: u32,
58    /// Connections killed at the deadline.
59    pub killed: u32,
60    /// How long the drain took.
61    pub took: Duration,
62}
63
64/// Counters summed over the workers.
65///
66/// ```no_run
67/// let server = net_io::serve("127.0.0.1:0", |_c| Ok(()))?;
68/// let stats = server.stats();
69/// println!("{} open, {} accepted", stats.open, stats.accepted);
70/// # Ok::<(), net_io::NetError>(())
71/// ```
72#[non_exhaustive]
73#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
74pub struct Stats {
75    /// Connections open now.
76    pub open: u64,
77    /// Connections accepted.
78    pub accepted: u64,
79    /// Connections refused.
80    pub refused: u64,
81    /// Connections closed.
82    pub closed: u64,
83    /// Bytes received.
84    pub bytes_in: u64,
85    /// Bytes sent.
86    pub bytes_out: u64,
87    /// Writes the gate denied.
88    pub egress_denied: u64,
89    /// Connections killed.
90    pub killed: u64,
91    /// Connections closed for the minimum rate.
92    pub slow_peers: u64,
93    /// Connections closed by a timeout.
94    pub timeouts: u64,
95    /// Times the process ran out of descriptors.
96    pub descriptor_pressure: u64,
97    /// Completions for connections that had closed.
98    pub stale_completions: u64,
99    /// Sessions that panicked.
100    pub session_panics: u64,
101    refused_by: [u64; RefusalCause::COUNT],
102    closed_by: [u64; CloseReason::COUNT],
103}
104
105impl Stats {
106    /// Refusals with this cause (gate codes summed).
107    #[must_use]
108    pub fn refused_for(&self, cause: RefusalCause) -> u64 {
109        self.refused_by[cause.index()]
110    }
111
112    /// Closes with this reason's variant (inner values ignored).
113    #[must_use]
114    pub fn closed_for(&self, reason: CloseReason) -> u64 {
115        self.closed_by[reason.index()]
116    }
117
118    /// Every close reason with a non-zero count, one entry per variant (inner
119    /// values are not kept: `Killed` carries reason 0, `Io` class `Other(0)`).
120    ///
121    /// ```no_run
122    /// let server = net_io::serve("127.0.0.1:0", |_c| Ok(()))?;
123    /// for (reason, n) in server.stats().closes() {
124    ///     println!("{}: {n}", reason.as_str());
125    /// }
126    /// # Ok::<(), net_io::NetError>(())
127    /// ```
128    pub fn closes(&self) -> impl Iterator<Item = (CloseReason, u64)> + '_ {
129        CloseReason::BY_INDEX
130            .iter()
131            .zip(self.closed_by.iter())
132            .filter(|(_, n)| **n > 0)
133            .map(|(r, n)| (*r, *n))
134    }
135
136    /// Every refusal cause with a non-zero count (gate codes summed under `Gate(0)`).
137    pub fn refusals(&self) -> impl Iterator<Item = (RefusalCause, u64)> + '_ {
138        RefusalCause::BY_INDEX
139            .iter()
140            .zip(self.refused_by.iter())
141            .filter(|(_, n)| **n > 0)
142            .map(|(r, n)| (*r, *n))
143    }
144}
145
146/// One listener of a server.
147///
148/// ```
149/// use net_io::{ListenerSpec, Limits};
150/// let spec = ListenerSpec::tcp("127.0.0.1:0")
151///     .name("admin")
152///     .limits(Limits::builder().max_per_listener(10).build()?);
153/// # let _ = spec;
154/// # Ok::<(), net_io::NetError>(())
155/// ```
156#[must_use]
157#[derive(Debug)]
158pub struct ListenerSpec {
159    pub(crate) addrs: Result<Vec<SocketAddr>, String>,
160    pub(crate) options: Option<ListenOptions>,
161    pub(crate) name: Option<String>,
162    pub(crate) limits: Limits,
163    pub(crate) proxy: ProxyMode,
164    pub(crate) trusted: Option<Arc<TrustedProxies>>,
165    pub(crate) tls: bool,
166    pub(crate) unix: Option<(PathBuf, LocalOptions)>,
167    #[cfg(all(windows, feature = "local"))]
168    pub(crate) pipe: Option<(String, crate::pipe::PipeOptions)>,
169    pub(crate) adopt: Option<ListenerHandle>,
170}
171
172impl ListenerSpec {
173    /// A TCP listener on `addr` (resolved now; a failure is reported by `build`).
174    pub fn tcp<A: ToSocketAddrs>(addr: A) -> ListenerSpec {
175        ListenerSpec {
176            addrs: addr
177                .to_socket_addrs()
178                .map(|i| i.collect())
179                .map_err(|e| e.to_string()),
180            options: None,
181            name: None,
182            limits: Limits::default(),
183            proxy: ProxyMode::Off,
184            trusted: None,
185            tls: false,
186            unix: None,
187            #[cfg(all(windows, feature = "local"))]
188            pipe: None,
189            adopt: None,
190        }
191    }
192
193    /// A TCP listener with explicit options (their backlog wins over the limits').
194    pub fn tcp_with<A: ToSocketAddrs>(addr: A, options: ListenOptions) -> ListenerSpec {
195        ListenerSpec {
196            options: Some(options),
197            ..ListenerSpec::tcp(addr)
198        }
199    }
200
201    /// A named pipe listener on `\\\\.\\pipe\\<name>` (feature `local`, Windows).
202    /// Remote clients are always rejected, and connections carry `Transport::Pipe`,
203    /// the client's pid, and no IP address: the per-source cap does not apply. With
204    /// `proxy`, every peer is trusted: the pipe's security decides who may connect.
205    ///
206    /// ```no_run
207    /// use net_io::{ListenerSpec, PipeOptions};
208    /// let spec = ListenerSpec::pipe("myapp-control", PipeOptions::new());
209    /// # let _ = spec;
210    /// ```
211    #[cfg(all(windows, feature = "local"))]
212    pub fn pipe(name: &str, options: crate::pipe::PipeOptions) -> ListenerSpec {
213        ListenerSpec {
214            pipe: Some((name.to_owned(), options)),
215            ..ListenerSpec::tcp(std::net::SocketAddr::from(([127, 0, 0, 1], 0)))
216        }
217    }
218
219    /// A Unix domain socket listener at `path` (feature `local`; on Windows, AF_UNIX,
220    /// where NTFS ACLs replace `options`' mode and owner). The socket file gets
221    /// `options`' mode and owner before it listens, and is removed at shutdown if it
222    /// is still the file this server created. Connections carry `Transport::Unix`, the peer's
223    /// credentials, and no IP address: the per-source cap does not apply, and a
224    /// `BlockingHandler` receives them as a `TcpConnection` whose `peer_addr` is
225    /// unspecified. With `proxy`, every peer is trusted: the socket file's
226    /// permissions decide who may connect.
227    ///
228    /// ```
229    /// use net_io::{ListenerSpec, LocalOptions};
230    /// let spec = ListenerSpec::unix("/run/myapp/api.sock", LocalOptions::new().group_access(33));
231    /// # let _ = spec;
232    /// ```
233    pub fn unix(path: impl Into<PathBuf>, options: LocalOptions) -> ListenerSpec {
234        ListenerSpec {
235            addrs: Ok(Vec::new()),
236            options: None,
237            name: None,
238            limits: Limits::default(),
239            proxy: ProxyMode::Off,
240            trusted: None,
241            tls: false,
242            unix: Some((path.into(), options)),
243            #[cfg(all(windows, feature = "local"))]
244            pipe: None,
245            adopt: None,
246        }
247    }
248
249    /// A listener this process was given: inherited from systemd or a FastCGI
250    /// manager, received in a hand-off, or released by another server. Its name
251    /// is the handle's unless [`ListenerSpec::name`] sets one. The socket is not
252    /// bound again; options set at bind time stay as they are, and the backlog is
253    /// reported as unknown. An adopted Unix socket's file is left in place at
254    /// shutdown (the next bind removes it as stale).
255    ///
256    /// ```no_run
257    /// use net_io::ListenerSpec;
258    /// for inherited in net_io::systemd()? {
259    ///     let spec = ListenerSpec::adopt(inherited.handle);
260    ///     # let _ = spec;
261    /// }
262    /// # Ok::<(), net_io::InheritError>(())
263    /// ```
264    pub fn adopt(handle: ListenerHandle) -> ListenerSpec {
265        ListenerSpec {
266            addrs: Ok(Vec::new()),
267            options: None,
268            name: None,
269            limits: Limits::default(),
270            proxy: ProxyMode::Off,
271            trusted: None,
272            tls: false,
273            unix: None,
274            #[cfg(all(windows, feature = "local"))]
275            pipe: None,
276            adopt: Some(handle),
277        }
278    }
279
280    /// A name (up to 64 bytes) for events and hand-off.
281    pub fn name(mut self, name: &str) -> ListenerSpec {
282        self.name = Some(name.to_owned());
283        self
284    }
285
286    /// Per-listener caps and deadlines; values set here override the server's.
287    pub fn limits(mut self, limits: Limits) -> ListenerSpec {
288        self.limits = limits;
289        self
290    }
291
292    /// Requires a PROXY protocol header on every connection (feature `proxy`).
293    /// Peers outside the trusted proxies are closed before any byte is read.
294    ///
295    /// ```
296    /// use net_io::{ListenerSpec, ProxyMode, ProxyVersions};
297    /// let spec = ListenerSpec::tcp("127.0.0.1:0").proxy(ProxyMode::Required { versions: ProxyVersions::Both });
298    /// # let _ = spec;
299    /// ```
300    pub fn proxy(mut self, mode: ProxyMode) -> ListenerSpec {
301        self.proxy = mode;
302        self
303    }
304
305    /// Serves TLS on this listener through the builder's acceptor
306    /// ([`ServerBuilder::tls`], feature `tls`).
307    pub fn tls(mut self, on: bool) -> ListenerSpec {
308        self.tls = on;
309        self
310    }
311
312    /// The proxies trusted on this listener, instead of the server's list.
313    pub fn trusted_proxies(mut self, proxies: TrustedProxies) -> ListenerSpec {
314        self.trusted = Some(Arc::new(proxies));
315        self
316    }
317}
318
319/// What runs per connection; the unset state (no `build` exists for it).
320#[derive(Clone, Copy, Debug, Default)]
321pub struct NoWork;
322
323/// The unset gate (no `build` exists for it): name `AllowAll` to run without a gate.
324#[derive(Clone, Copy, Debug, Default)]
325pub struct NoGate;
326
327/// What `ServerBuilder::sessions` produces; programs never name it.
328#[derive(Debug)]
329pub struct Sessions<F> {
330    pub(crate) factory: F,
331}
332
333#[cfg(feature = "reactor")]
334impl<F> Sessions<F> {
335    pub(crate) fn factory(&self) -> &F {
336        &self.factory
337    }
338}
339
340/// What `ServerBuilder::handler` produces; programs never name it.
341pub struct BlockingHandler<H> {
342    pub(crate) handler: H,
343}
344
345impl<H> fmt::Debug for BlockingHandler<H> {
346    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
347        f.write_str("BlockingHandler")
348    }
349}
350
351pub(crate) mod sealed {
352    pub trait Sealed {}
353}
354
355/// What runs per connection: [`Sessions`] or a [`BlockingHandler`]. Sealed.
356pub trait Workload: sealed::Sealed + engine::Serve + Send + Sync + 'static {}
357
358impl<F: SessionFactory> sealed::Sealed for Sessions<F> {}
359impl<F: SessionFactory> Workload for Sessions<F> {}
360
361impl<H> sealed::Sealed for BlockingHandler<H> where
362    H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static
363{
364}
365impl<H> Workload for BlockingHandler<H> where
366    H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static
367{
368}
369
370/// Builds a [`Server`]. A builder without a gate or without `sessions`/`handler`
371/// has no `build` method: forgetting either is a compile error.
372///
373/// # Examples
374///
375/// ```no_run
376/// use net_io::{AllowAll, DriverKind, Limits, ListenerSpec, ServerBuilder};
377/// use std::time::{Duration, Instant};
378///
379/// let server = ServerBuilder::new()
380///     .listener(ListenerSpec::tcp("0.0.0.0:8080").name("public"))
381///     .listener(ListenerSpec::tcp("127.0.0.1:9090").name("admin"))
382///     .driver(DriverKind::Blocking)
383///     .workers(64)
384///     .limits(Limits::builder().idle_timeout(Duration::from_secs(5)).build()?)
385///     .handler(|conn| { conn.send_all(b"hello\n")?; Ok(()) })
386///     .gate(AllowAll)
387///     .build()?;              // listeners bound, workers started, door closed
388/// // ... drop privileges here ...
389/// server.open_door()?;
390/// let report = server.shutdown(Instant::now() + Duration::from_secs(30));
391/// println!("{report:?}");
392/// # Ok::<(), net_io::NetError>(())
393/// ```
394#[must_use]
395pub struct ServerBuilder<W = NoWork, G = NoGate, S = NoopSink, T = NoTls> {
396    listeners: Vec<ListenerSpec>,
397    driver: DriverKind,
398    workers: Option<u16>,
399    limits: Limits,
400    work: W,
401    gate: G,
402    sink: S,
403    tls: T,
404    control: Option<Arc<dyn ControlLine>>,
405    wipe: bool,
406    clock: Option<Arc<dyn Clock>>,
407    trusted: Option<Arc<TrustedProxies>>,
408}
409
410impl ServerBuilder {
411    /// An empty builder.
412    pub fn new() -> ServerBuilder {
413        ServerBuilder {
414            listeners: Vec::new(),
415            driver: DriverKind::Auto,
416            workers: None,
417            limits: Limits::default(),
418            work: NoWork,
419            gate: NoGate,
420            sink: NoopSink,
421            tls: NoTls,
422            control: None,
423            wipe: true,
424            clock: None,
425            trusted: None,
426        }
427    }
428}
429
430impl Default for ServerBuilder {
431    fn default() -> ServerBuilder {
432        ServerBuilder::new()
433    }
434}
435
436impl<W, G, S, T> ServerBuilder<W, G, S, T> {
437    fn map<W2, G2, S2, T2>(
438        self,
439        f: impl FnOnce(W, G, S, T) -> (W2, G2, S2, T2),
440    ) -> ServerBuilder<W2, G2, S2, T2> {
441        let ServerBuilder {
442            listeners,
443            driver,
444            workers,
445            limits,
446            work,
447            gate,
448            sink,
449            tls,
450            control,
451            wipe,
452            clock,
453            trusted,
454        } = self;
455        let (work, gate, sink, tls) = f(work, gate, sink, tls);
456        ServerBuilder {
457            listeners,
458            driver,
459            workers,
460            limits,
461            work,
462            gate,
463            sink,
464            tls,
465            control,
466            wipe,
467            clock,
468            trusted,
469        }
470    }
471
472    /// The proxies whose PROXY headers are believed, for every listener that
473    /// requires one and has no list of its own.
474    pub fn trusted_proxies(mut self, proxies: TrustedProxies) -> Self {
475        self.trusted = Some(Arc::new(proxies));
476        self
477    }
478
479    /// Adds a listener (at most 64).
480    pub fn listener(mut self, spec: ListenerSpec) -> Self {
481        self.listeners.push(spec);
482        self
483    }
484
485    /// The driver (default `Auto`).
486    pub fn driver(mut self, kind: DriverKind) -> Self {
487        self.driver = kind;
488        self
489    }
490
491    /// Worker threads (default 128 on the blocking driver; 1 to 4,096).
492    pub fn workers(mut self, n: u16) -> Self {
493        self.workers = Some(n);
494        self
495    }
496
497    /// The server's limits; unset values take the driver's defaults.
498    pub fn limits(mut self, limits: Limits) -> Self {
499        self.limits = limits;
500        self
501    }
502
503    /// Runs a sans-IO session per connection.
504    pub fn sessions<F: SessionFactory>(self, factory: F) -> ServerBuilder<Sessions<F>, G, S, T> {
505        self.map(|_, g, s, t| (Sessions { factory }, g, s, t))
506    }
507
508    /// Runs a blocking handler per connection (on the blocking driver).
509    pub fn handler<H>(self, handler: H) -> ServerBuilder<BlockingHandler<H>, G, S, T>
510    where
511        H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static,
512    {
513        self.map(|_, g, s, t| (BlockingHandler { handler }, g, s, t))
514    }
515
516    /// The TLS engine for listeners with `ListenerSpec::tls(true)` (feature `tls`).
517    pub fn tls<T2: TlsAcceptor>(self, acceptor: T2) -> ServerBuilder<W, G, S, T2> {
518        self.map(|w, g, s, _| (w, g, s, acceptor))
519    }
520
521    /// The connection gate (required; `AllowAll` for none).
522    pub fn gate<G2: Gate>(self, gate: G2) -> ServerBuilder<W, G2, S, T> {
523        self.map(|w, _, s, t| (w, gate, s, t))
524    }
525
526    /// The event sink (default `NoopSink`).
527    pub fn events<S2: EventSink>(self, sink: S2) -> ServerBuilder<W, G, S2, T> {
528        self.map(|w, g, _, t| (w, g, sink, t))
529    }
530
531    /// The control line (default: none).
532    pub fn control<C: ControlLine>(mut self, line: Arc<C>) -> Self {
533        self.control = Some(line);
534        self
535    }
536
537    /// Zero the used part of every buffer when it returns to a pool (default on).
538    pub fn wipe_buffers(mut self, on: bool) -> Self {
539        self.wipe = on;
540        self
541    }
542
543    /// The clock deadlines read (tests inject a `ManualClock`).
544    pub fn clock<K: Clock>(mut self, clock: Arc<K>) -> Self {
545        self.clock = Some(clock);
546        self
547    }
548}
549
550impl<W, G, S, T> ServerBuilder<W, G, S, T>
551where
552    W: Workload,
553    G: Gate,
554    S: EventSink,
555    T: TlsAcceptor,
556{
557    /// Validates everything, binds the listeners, starts the workers, and returns
558    /// the server with the door **closed**.
559    ///
560    /// # Errors
561    ///
562    /// `InvalidConfig` (`field` names the method), bind errors, `Unsupported` for a
563    /// driver this version does not have.
564    pub fn build(self) -> Result<Server, NetError> {
565        // A handler holds its worker for the whole connection (blocking pool only).
566        let requested = match self.driver {
567            DriverKind::Auto if <W as engine::Serve>::BLOCKING_ONLY => DriverKind::Blocking,
568            other => other,
569        };
570        let driver = match requested.resolve() {
571            DriverKind::Reactor if <W as engine::Serve>::BLOCKING_ONLY => {
572                return Err(invalid("driver", ConfigReason::Inconsistent));
573            }
574            DriverKind::Reactor if !DriverKind::reactor_built() => {
575                return Err(NetError::Unsupported {
576                    feature: "the reactor driver",
577                    platform: std::env::consts::OS,
578                });
579            }
580            DriverKind::Uring => {
581                return Err(NetError::Unsupported {
582                    feature: "the io_uring driver (after 1.0.0)",
583                    platform: std::env::consts::OS,
584                });
585            }
586            kind => kind,
587        };
588        if self.listeners.is_empty() {
589            return Err(invalid("listener", ConfigReason::Missing));
590        }
591        if self.listeners.len() > 64 {
592            return Err(invalid("listener", ConfigReason::TooLarge { max: 64 }));
593        }
594        let workers = self.workers.unwrap_or_else(|| {
595            if driver == DriverKind::Reactor {
596                std::thread::available_parallelism()
597                    .map_or(4, |n| u16::try_from(n.get()).unwrap_or(4096).min(4096))
598            } else {
599                128
600            }
601        });
602        if workers == 0 {
603            return Err(invalid("workers", ConfigReason::Zero));
604        }
605        if workers > 4096 {
606            return Err(invalid("workers", ConfigReason::TooLarge { max: 4096 }));
607        }
608        let mut listeners = self.listeners;
609        let has_tls = TypeId::of::<T>() != TypeId::of::<NoTls>();
610        let mut names = Vec::new();
611        for spec in &mut listeners {
612            check_spec(spec, self.trusted.as_ref(), has_tls, &names)?;
613            if let Some(n) = &spec.name {
614                names.push(n.clone());
615            }
616        }
617        // On the blocking driver an unset global cap is one connection per worker;
618        // on the reactor the default, but never more than the descriptors allow.
619        let limits = if self.limits.max_connections().is_none() {
620            let cap = if driver == DriverKind::Reactor {
621                reactor_default_cap(workers)
622            } else {
623                u32::from(workers)
624            };
625            self.limits.to_builder().max_connections(cap).build()?
626        } else {
627            self.limits.clone()
628        };
629        limits.validate(workers, crate::sys::descriptor_limit())?;
630        for spec in &listeners {
631            limits.overlay(&spec.limits).validate(workers, None)?;
632        }
633        let tls: Option<Arc<dyn ErasedAcceptor>> = if has_tls {
634            Some(Arc::new(self.tls))
635        } else {
636            None
637        };
638        engine::start(engine::Config {
639            tls,
640            trusted: self.trusted,
641            listeners,
642            driver,
643            workers,
644            limits,
645            work: self.work,
646            gate: self.gate,
647            sink: self.sink,
648            control: self.control,
649            wipe: self.wipe,
650            clock: self.clock,
651        })
652    }
653}
654
655/// The reactor's default global cap: 10,000, less where the process's descriptor
656/// limit would not hold it.
657fn reactor_default_cap(workers: u16) -> u32 {
658    let wanted = 10_000u32;
659    match crate::sys::descriptor_limit() {
660        Some(have) => {
661            let spare = u64::from(workers).saturating_mul(2) + 64;
662            let room = have.saturating_sub(spare).clamp(16, u64::from(wanted));
663            u32::try_from(room).unwrap_or(wanted)
664        }
665        None => wanted,
666    }
667}
668
669fn invalid(field: &'static str, reason: ConfigReason) -> NetError {
670    NetError::InvalidConfig { field, reason }
671}
672
673/// Where `Server::adopt` registers a connection: the listener (by name) whose
674/// limits and caps apply, and an optional starting lane (the gate may change it).
675///
676/// ```
677/// let spec = net_io::AdoptSpec::new("main").lane(net_io::Lane(1));
678/// # let _ = spec;
679/// ```
680#[must_use]
681#[derive(Clone, Debug, PartialEq, Eq)]
682pub struct AdoptSpec {
683    listener: String,
684    lane: Option<crate::info::Lane>,
685}
686
687impl AdoptSpec {
688    /// Registers on the listener named `listener`.
689    pub fn new(listener: &str) -> AdoptSpec {
690        AdoptSpec {
691            listener: listener.to_owned(),
692            lane: None,
693        }
694    }
695
696    /// The lane the connection starts in, before the gate's admission.
697    pub fn lane(mut self, lane: crate::info::Lane) -> AdoptSpec {
698        self.lane = Some(lane);
699        self
700    }
701}
702
703/// The checks `build` and `add_listener` run on one listener: the features it
704/// needs, a TLS engine, trusted proxies (resolved into the spec), and its name.
705fn check_spec(
706    spec: &mut ListenerSpec,
707    server_trusted: Option<&Arc<TrustedProxies>>,
708    has_tls: bool,
709    taken: &[String],
710) -> Result<(), NetError> {
711    if spec.name.is_none() {
712        spec.name = spec
713            .adopt
714            .as_ref()
715            .and_then(|h| h.name())
716            .map(str::to_owned);
717    }
718    let unix = spec.unix.is_some()
719        || spec
720            .adopt
721            .as_ref()
722            .is_some_and(|h| h.kind() == crate::info::Transport::Unix);
723    if unix && !cfg!(feature = "local") {
724        return Err(NetError::Unsupported {
725            feature: "Unix domain sockets (the `local` feature)",
726            platform: std::env::consts::OS,
727        });
728    }
729    if spec.tls {
730        if !cfg!(feature = "tls") {
731            return Err(NetError::Unsupported {
732                feature: "TLS (the `tls` feature)",
733                platform: std::env::consts::OS,
734            });
735        }
736        if !has_tls {
737            return Err(invalid("tls", ConfigReason::Missing));
738        }
739    }
740    if spec.proxy != ProxyMode::Off {
741        if !cfg!(feature = "proxy") {
742            return Err(NetError::Unsupported {
743                feature: "PROXY protocol (the `proxy` feature)",
744                platform: std::env::consts::OS,
745            });
746        }
747        // On a Unix socket the file's permissions decide who may connect.
748        if !unix {
749            let trusted = spec.trusted.clone().or_else(|| server_trusted.cloned());
750            match trusted {
751                Some(t) if !t.is_empty() => spec.trusted = Some(t),
752                // Required without anyone to trust would refuse every connection.
753                _ => return Err(invalid("trusted_proxies", ConfigReason::Missing)),
754            }
755        }
756    }
757    if let Some(n) = &spec.name {
758        if n.len() > 64 {
759            return Err(invalid("name", ConfigReason::TooLarge { max: 64 }));
760        }
761        if taken.contains(n) {
762            return Err(invalid("name", ConfigReason::Inconsistent));
763        }
764    }
765    Ok(())
766}
767
768/// A running server. Dropping it stops it (with a drain), so keep it and call
769/// [`Server::wait_for_stop`] to run until a stop is requested.
770#[must_use = "dropping a Server stops it; call wait_for_stop() to run until a stop is requested"]
771pub struct Server {
772    pub(crate) shared: Arc<Shared>,
773    pub(crate) threads: Vec<JoinHandle<()>>,
774    pub(crate) listeners: Vec<ListenerInfo>,
775    pub(crate) stopped: bool,
776}
777
778impl Server {
779    /// Starts accepting on every listener; a second call is a no-op.
780    ///
781    /// # Errors
782    ///
783    /// `ShuttingDown` while draining or after shutdown; `ForkedChild`.
784    pub fn open_door(&self) -> Result<(), NetError> {
785        self.shared.open_door()
786    }
787
788    /// Stops accepting within one loop iteration; listening sockets stay open. Idempotent.
789    pub fn close_door(&self) {
790        self.shared.close_door();
791    }
792
793    /// Whether the server is accepting.
794    #[must_use]
795    pub fn door(&self) -> DoorState {
796        self.shared.door()
797    }
798
799    /// Closes the door, tells sessions to finish, closes idle connections at once,
800    /// waits for the rest until `deadline`, then kills them with `DrainDeadline`.
801    pub fn drain(&self, deadline: Instant) -> DrainReport {
802        self.shared.drain(deadline)
803    }
804
805    /// Drains, then closes the listening sockets and joins every thread.
806    pub fn shutdown(mut self, deadline: Instant) -> DrainReport {
807        self.stop(deadline)
808    }
809
810    /// Blocks until a stop is requested (`ServerHandle::request_stop`, or the
811    /// control line moves to `Draining` or `Stopping`), then shuts down.
812    pub fn wait_for_stop(mut self) -> DrainReport {
813        let deadline = self.shared.wait_for_stop_request();
814        self.stop(deadline)
815    }
816
817    fn stop(&mut self, deadline: Instant) -> DrainReport {
818        if self.stopped {
819            return DrainReport {
820                graceful: 0,
821                killed: 0,
822                took: Duration::ZERO,
823            };
824        }
825        self.stopped = true;
826        let report = self.shared.drain(deadline);
827        self.shared.stop_threads();
828        for t in self.threads.drain(..) {
829            let _ = t.join();
830        }
831        self.shared.join_listeners();
832        report
833    }
834
835    /// Registers a dialed connection exactly like an accepted one on the
836    /// listener named in `spec`: a new id, the caps, `Gate::admit` with
837    /// `AdmitStage::Adopt`, that listener's limits, the server's handler or
838    /// sessions, its kill handle, and its events. For back-channels and outbound
839    /// connections a server should run and revoke like its own. A gate the
840    /// connection had from `set_gate` is closed (`SessionClosed`) first.
841    ///
842    /// # Errors
843    ///
844    /// `Refused { cause }` (caps, the gate, `Busy`; the connection was reset),
845    /// `InvalidConfig { field: "listener" }` (no listener with that name),
846    /// `InvalidConfig { field: "connection" }` (a TLS connection, or bytes read
847    /// ahead), `ShuttingDown`, `ForkedChild`, `Closed`.
848    ///
849    /// ```no_run
850    /// use net_io::{AdoptSpec, AllowAll, ListenerSpec, ServerBuilder};
851    /// let server = ServerBuilder::new()
852    ///     .listener(ListenerSpec::tcp("127.0.0.1:7000").name("main"))
853    ///     .handler(|conn| { conn.send_all(b"hello, upstream\n")?; Ok(()) })
854    ///     .gate(AllowAll)
855    ///     .build()?;
856    /// let upstream = net_io::connect("10.0.0.9:9000")?;
857    /// let id = server.adopt(upstream, AdoptSpec::new("main"))?;
858    /// # let _ = id;
859    /// # Ok::<(), net_io::NetError>(())
860    /// ```
861    pub fn adopt(&self, conn: TcpConnection, spec: AdoptSpec) -> Result<ConnectionId, NetError> {
862        let mut conn = conn;
863        let link = conn.take_link().ok_or(NetError::Closed)?;
864        let (Some(peer), Some(local)) =
865            (link.info.peer.socket_addr(), link.info.local.socket_addr())
866        else {
867            return Err(invalid("connection", ConfigReason::Inconsistent));
868        };
869        let stream = link
870            .into_stream(CloseReason::SessionClosed)
871            .and_then(crate::sys::Stream::into_tcp)
872            .ok_or(invalid("connection", ConfigReason::Inconsistent))?;
873        self.shared
874            .adopt(stream, peer, local, &spec.listener, spec.lane)
875    }
876
877    /// Owned duplicates of every listening socket, for a hand-off to the next
878    /// process (`send_listeners`, Unix). This server keeps
879    /// accepting meanwhile, so the new process can start before this one stops
880    /// (decision D-042); a later `shutdown` closes only this server's copies, and
881    /// a Unix socket file stays in place for the new process.
882    ///
883    /// # Errors
884    ///
885    /// `Os` when a socket cannot be duplicated, `ForkedChild`.
886    pub fn release_listeners(&self) -> Result<Vec<ListenerHandle>, NetError> {
887        self.shared.release_listeners()
888    }
889
890    /// Binds one more listener on the running server and starts its accept
891    /// thread. Its door is closed until the next [`Server::open_door`] (which
892    /// opens every listener's door); the same checks as `build` apply. The id is
893    /// new: ids of removed listeners are never reused.
894    ///
895    /// # Errors
896    ///
897    /// As `build` for the listener; `InvalidConfig { field: "listener" }` past 64
898    /// listeners, `ShuttingDown`, `ForkedChild`.
899    ///
900    /// ```no_run
901    /// use net_io::{AllowAll, ListenerSpec, ServerBuilder};
902    /// let mut server = ServerBuilder::new()
903    ///     .listener(ListenerSpec::tcp("127.0.0.1:8080"))
904    ///     .handler(|conn| { conn.send_all(b"hi\n")?; Ok(()) })
905    ///     .gate(AllowAll)
906    ///     .build()?;
907    /// server.open_door()?;
908    /// let added = server.add_listener(ListenerSpec::tcp("127.0.0.1:8081").name("extra"))?;
909    /// server.open_door()?; // opens the new listener's door
910    /// # let _ = added;
911    /// # Ok::<(), net_io::NetError>(())
912    /// ```
913    pub fn add_listener(&mut self, spec: ListenerSpec) -> Result<ListenerId, NetError> {
914        let mut spec = spec;
915        if self.listeners.len() >= 64 {
916            return Err(invalid("listener", ConfigReason::TooLarge { max: 64 }));
917        }
918        let names: Vec<String> = self
919            .listeners
920            .iter()
921            .filter_map(|l| l.name.clone())
922            .collect();
923        check_spec(
924            &mut spec,
925            self.shared.trusted.as_ref(),
926            self.shared.tls.is_some(),
927            &names,
928        )?;
929        let workers = u16::try_from(self.shared.worker_count()).unwrap_or(u16::MAX);
930        self.shared
931            .base_limits
932            .overlay(&spec.limits)
933            .validate(workers, None)?;
934        let info = self.shared.add_listener(spec)?;
935        let id = info.id;
936        self.listeners.push(info);
937        Ok(id)
938    }
939
940    /// Closes one listener's socket at once (its connections keep running),
941    /// then drains them: idle ones close now, the rest may finish until
942    /// `deadline` and are then killed (`DrainDeadline`). Other listeners and
943    /// their connections are not touched.
944    ///
945    /// # Errors
946    ///
947    /// `InvalidConfig { field: "listener", reason: Missing }` for an unknown id,
948    /// `ForkedChild`.
949    pub fn remove_listener(
950        &mut self,
951        id: ListenerId,
952        deadline: Instant,
953    ) -> Result<DrainReport, NetError> {
954        let report = self.shared.remove_listener(id, deadline)?;
955        self.listeners.retain(|l| l.id != id);
956        Ok(report)
957    }
958
959    /// A cloneable handle for other threads.
960    #[must_use]
961    pub fn handle(&self) -> ServerHandle {
962        ServerHandle {
963            shared: self.shared.clone(),
964        }
965    }
966
967    /// Counters summed over the workers.
968    #[must_use]
969    pub fn stats(&self) -> Stats {
970        self.shared.stats()
971    }
972
973    /// Facts about each listener.
974    #[must_use]
975    pub fn listeners(&self) -> &[ListenerInfo] {
976        &self.listeners
977    }
978}
979
980impl Drop for Server {
981    fn drop(&mut self) {
982        let deadline = Instant::now() + self.shared.drain_timeout;
983        self.stop(deadline);
984    }
985}
986
987impl fmt::Debug for Server {
988    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
989        f.debug_struct("Server")
990            .field("listeners", &self.listeners)
991            .field("door", &self.door())
992            .finish()
993    }
994}
995
996/// A cloneable handle to a running server, safe to use from any thread.
997///
998/// ```no_run
999/// let server = net_io::serve("127.0.0.1:7000", |conn| { conn.send_all(b"hi\n")?; Ok(()) })?;
1000/// let handle = server.handle();
1001/// std::thread::spawn(move || {
1002///     // ... wait for a signal from the application's signal library ...
1003///     handle.request_stop();
1004/// });
1005/// let report = server.wait_for_stop();
1006/// println!("drained: {} graceful, {} killed in {:?}", report.graceful, report.killed, report.took);
1007/// # Ok::<(), net_io::NetError>(())
1008/// ```
1009#[derive(Clone)]
1010pub struct ServerHandle {
1011    pub(crate) shared: Arc<Shared>,
1012}
1013
1014impl ServerHandle {
1015    /// Aborts the connection `id`; `false` if it already closed or belongs to another server.
1016    pub fn kill(&self, id: ConnectionId, reason: KillReason) -> bool {
1017        self.shared.kill(id, reason)
1018    }
1019
1020    /// Stops accepting.
1021    pub fn close_door(&self) {
1022        self.shared.close_door();
1023    }
1024
1025    /// Makes `Server::wait_for_stop` return after the drain. Does not block.
1026    pub fn request_stop(&self) {
1027        self.shared.request_stop(None);
1028    }
1029
1030    /// Interrupts every worker's and acceptor's wait so a control-line change is seen at once.
1031    pub fn wake_all(&self) {
1032        self.shared.wake_all();
1033    }
1034
1035    /// The completion queue of `worker`.
1036    #[must_use]
1037    pub fn completer(&self, worker: WorkerId) -> Option<crate::completer::Completer> {
1038        self.shared.completer(worker)
1039    }
1040
1041    /// Counters summed over the workers.
1042    #[must_use]
1043    pub fn stats(&self) -> Stats {
1044        self.shared.stats()
1045    }
1046
1047    /// Whether the server is accepting.
1048    #[must_use]
1049    pub fn door(&self) -> DoorState {
1050        self.shared.door()
1051    }
1052}
1053
1054impl fmt::Debug for ServerHandle {
1055    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1056        f.debug_struct("ServerHandle")
1057            .field("door", &self.door())
1058            .finish()
1059    }
1060}
1061
1062impl Stats {
1063    pub(crate) fn from_parts(
1064        mut base: Stats,
1065        refused_by: [u64; RefusalCause::COUNT],
1066        closed_by: [u64; CloseReason::COUNT],
1067    ) -> Stats {
1068        base.refused_by = refused_by;
1069        base.closed_by = closed_by;
1070        base
1071    }
1072}
net_io/server/mod.rs

net_io/server/
mod.rs
