net_io/server/
mod.rs

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

net_io/server/
mod.rs
