net_io/server/
mod.rs

1//! The server: listeners, workers, the door, drain, and shutdown.
2
3mod engine;
4mod source;
5
6use std::fmt;
7use std::net::{SocketAddr, ToSocketAddrs};
8use std::sync::Arc;
9use std::thread::JoinHandle;
10use std::time::{Duration, Instant};
11
12use crate::control::ControlLine;
13use crate::driver::DriverKind;
14use crate::error::{CloseReason, ConfigReason, KillReason, NetError, RefusalCause};
15use crate::events::{EventSink, NoopSink};
16use crate::gate::Gate;
17use crate::id::{ConnectionId, WorkerId};
18use crate::limits::Limits;
19use crate::opts::ListenOptions;
20use crate::session::SessionFactory;
21use crate::tcp::{ListenerInfo, TcpConnection};
22use crate::time::Clock;
23use crate::tls::{NoTls, TlsAcceptor};
24
25pub(crate) use engine::Shared;
26
27/// Whether the server is accepting.
28///
29/// ```
30/// assert_ne!(net_io::DoorState::Open, net_io::DoorState::Closed);
31/// ```
32#[derive(Clone, Copy, Debug, PartialEq, Eq)]
33pub enum DoorState {
34    /// Accepting new connections.
35    Open,
36    /// Not accepting; listening sockets stay open.
37    Closed,
38}
39
40/// What a drain did.
41///
42/// ```
43/// let r = net_io::DrainReport { graceful: 3, killed: 1, took: std::time::Duration::from_millis(40) };
44/// assert_eq!(r.graceful + r.killed, 4);
45/// ```
46#[derive(Clone, Copy, Debug, PartialEq, Eq)]
47pub struct DrainReport {
48    /// Connections that finished before the deadline.
49    pub graceful: u32,
50    /// Connections killed at the deadline.
51    pub killed: u32,
52    /// How long the drain took.
53    pub took: Duration,
54}
55
56/// Counters summed over the workers.
57///
58/// ```no_run
59/// let server = net_io::serve("127.0.0.1:0", |_c| Ok(()))?;
60/// let stats = server.stats();
61/// println!("{} open, {} accepted", stats.open, stats.accepted);
62/// # Ok::<(), net_io::NetError>(())
63/// ```
64#[non_exhaustive]
65#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
66pub struct Stats {
67    /// Connections open now.
68    pub open: u64,
69    /// Connections accepted.
70    pub accepted: u64,
71    /// Connections refused.
72    pub refused: u64,
73    /// Connections closed.
74    pub closed: u64,
75    /// Bytes received.
76    pub bytes_in: u64,
77    /// Bytes sent.
78    pub bytes_out: u64,
79    /// Writes the gate denied.
80    pub egress_denied: u64,
81    /// Connections killed.
82    pub killed: u64,
83    /// Connections closed for the minimum rate.
84    pub slow_peers: u64,
85    /// Connections closed by a timeout.
86    pub timeouts: u64,
87    /// Times the process ran out of descriptors.
88    pub descriptor_pressure: u64,
89    /// Completions for connections that had closed.
90    pub stale_completions: u64,
91    /// Sessions that panicked.
92    pub session_panics: u64,
93    refused_by: [u64; RefusalCause::COUNT],
94    closed_by: [u64; CloseReason::COUNT],
95}
96
97impl Stats {
98    /// Refusals with this cause (gate codes summed).
99    #[must_use]
100    pub fn refused_for(&self, cause: RefusalCause) -> u64 {
101        self.refused_by[cause.index()]
102    }
103
104    /// Closes with this reason's variant (inner values ignored).
105    #[must_use]
106    pub fn closed_for(&self, reason: CloseReason) -> u64 {
107        self.closed_by[reason.index()]
108    }
109}
110
111/// One listener of a server.
112///
113/// ```
114/// use net_io::{ListenerSpec, Limits};
115/// let spec = ListenerSpec::tcp("127.0.0.1:0")
116///     .name("admin")
117///     .limits(Limits::builder().max_per_listener(10).build()?);
118/// # let _ = spec;
119/// # Ok::<(), net_io::NetError>(())
120/// ```
121#[must_use]
122#[derive(Debug)]
123pub struct ListenerSpec {
124    pub(crate) addrs: Result<Vec<SocketAddr>, String>,
125    pub(crate) options: Option<ListenOptions>,
126    pub(crate) name: Option<String>,
127    pub(crate) limits: Limits,
128}
129
130impl ListenerSpec {
131    /// A TCP listener on `addr` (resolved now; a failure is reported by `build`).
132    pub fn tcp<A: ToSocketAddrs>(addr: A) -> ListenerSpec {
133        ListenerSpec {
134            addrs: addr
135                .to_socket_addrs()
136                .map(|i| i.collect())
137                .map_err(|e| e.to_string()),
138            options: None,
139            name: None,
140            limits: Limits::default(),
141        }
142    }
143
144    /// A TCP listener with explicit options (their backlog wins over the limits').
145    pub fn tcp_with<A: ToSocketAddrs>(addr: A, options: ListenOptions) -> ListenerSpec {
146        ListenerSpec {
147            options: Some(options),
148            ..ListenerSpec::tcp(addr)
149        }
150    }
151
152    /// A name (up to 64 bytes) for events and hand-off.
153    pub fn name(mut self, name: &str) -> ListenerSpec {
154        self.name = Some(name.to_owned());
155        self
156    }
157
158    /// Per-listener caps and deadlines; values set here override the server's.
159    pub fn limits(mut self, limits: Limits) -> ListenerSpec {
160        self.limits = limits;
161        self
162    }
163}
164
165/// What runs per connection; the unset state (no `build` exists for it).
166#[derive(Clone, Copy, Debug, Default)]
167pub struct NoWork;
168
169/// The unset gate (no `build` exists for it): name `AllowAll` to run without a gate.
170#[derive(Clone, Copy, Debug, Default)]
171pub struct NoGate;
172
173/// What `ServerBuilder::sessions` produces; programs never name it.
174#[derive(Debug)]
175pub struct Sessions<F> {
176    pub(crate) factory: F,
177}
178
179/// What `ServerBuilder::handler` produces; programs never name it.
180pub struct BlockingHandler<H> {
181    pub(crate) handler: H,
182}
183
184impl<H> fmt::Debug for BlockingHandler<H> {
185    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
186        f.write_str("BlockingHandler")
187    }
188}
189
190pub(crate) mod sealed {
191    pub trait Sealed {}
192}
193
194/// What runs per connection: [`Sessions`] or a [`BlockingHandler`]. Sealed.
195pub trait Workload: sealed::Sealed + engine::Serve + Send + Sync + 'static {}
196
197impl<F: SessionFactory> sealed::Sealed for Sessions<F> {}
198impl<F: SessionFactory> Workload for Sessions<F> {}
199
200impl<H> sealed::Sealed for BlockingHandler<H> where
201    H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static
202{
203}
204impl<H> Workload for BlockingHandler<H> where
205    H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static
206{
207}
208
209/// Builds a [`Server`]. A builder without a gate or without `sessions`/`handler`
210/// has no `build` method: forgetting either is a compile error.
211///
212/// # Examples
213///
214/// ```no_run
215/// use net_io::{AllowAll, DriverKind, Limits, ListenerSpec, ServerBuilder};
216/// use std::time::{Duration, Instant};
217///
218/// let server = ServerBuilder::new()
219///     .listener(ListenerSpec::tcp("0.0.0.0:8080").name("public"))
220///     .listener(ListenerSpec::tcp("127.0.0.1:9090").name("admin"))
221///     .driver(DriverKind::Blocking)
222///     .workers(64)
223///     .limits(Limits::builder().idle_timeout(Duration::from_secs(5)).build()?)
224///     .handler(|conn| { conn.send_all(b"hello\n")?; Ok(()) })
225///     .gate(AllowAll)
226///     .build()?;              // listeners bound, workers started, door closed
227/// // ... drop privileges here ...
228/// server.open_door()?;
229/// let report = server.shutdown(Instant::now() + Duration::from_secs(30));
230/// println!("{report:?}");
231/// # Ok::<(), net_io::NetError>(())
232/// ```
233#[must_use]
234pub struct ServerBuilder<W = NoWork, G = NoGate, S = NoopSink, T = NoTls> {
235    listeners: Vec<ListenerSpec>,
236    driver: DriverKind,
237    workers: Option<u16>,
238    limits: Limits,
239    work: W,
240    gate: G,
241    sink: S,
242    tls: T,
243    control: Option<Arc<dyn ControlLine>>,
244    wipe: bool,
245    clock: Option<Arc<dyn Clock>>,
246}
247
248impl ServerBuilder {
249    /// An empty builder.
250    pub fn new() -> ServerBuilder {
251        ServerBuilder {
252            listeners: Vec::new(),
253            driver: DriverKind::Auto,
254            workers: None,
255            limits: Limits::default(),
256            work: NoWork,
257            gate: NoGate,
258            sink: NoopSink,
259            tls: NoTls,
260            control: None,
261            wipe: true,
262            clock: None,
263        }
264    }
265}
266
267impl Default for ServerBuilder {
268    fn default() -> ServerBuilder {
269        ServerBuilder::new()
270    }
271}
272
273impl<W, G, S, T> ServerBuilder<W, G, S, T> {
274    fn map<W2, G2, S2, T2>(
275        self,
276        f: impl FnOnce(W, G, S, T) -> (W2, G2, S2, T2),
277    ) -> ServerBuilder<W2, G2, S2, T2> {
278        let ServerBuilder {
279            listeners,
280            driver,
281            workers,
282            limits,
283            work,
284            gate,
285            sink,
286            tls,
287            control,
288            wipe,
289            clock,
290        } = self;
291        let (work, gate, sink, tls) = f(work, gate, sink, tls);
292        ServerBuilder {
293            listeners,
294            driver,
295            workers,
296            limits,
297            work,
298            gate,
299            sink,
300            tls,
301            control,
302            wipe,
303            clock,
304        }
305    }
306
307    /// Adds a listener (at most 64).
308    pub fn listener(mut self, spec: ListenerSpec) -> Self {
309        self.listeners.push(spec);
310        self
311    }
312
313    /// The driver (default `Auto`).
314    pub fn driver(mut self, kind: DriverKind) -> Self {
315        self.driver = kind;
316        self
317    }
318
319    /// Worker threads (default 128 on the blocking driver; 1 to 4,096).
320    pub fn workers(mut self, n: u16) -> Self {
321        self.workers = Some(n);
322        self
323    }
324
325    /// The server's limits; unset values take the driver's defaults.
326    pub fn limits(mut self, limits: Limits) -> Self {
327        self.limits = limits;
328        self
329    }
330
331    /// Runs a sans-IO session per connection.
332    pub fn sessions<F: SessionFactory>(self, factory: F) -> ServerBuilder<Sessions<F>, G, S, T> {
333        self.map(|_, g, s, t| (Sessions { factory }, g, s, t))
334    }
335
336    /// Runs a blocking handler per connection (on the blocking driver).
337    pub fn handler<H>(self, handler: H) -> ServerBuilder<BlockingHandler<H>, G, S, T>
338    where
339        H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static,
340    {
341        self.map(|_, g, s, t| (BlockingHandler { handler }, g, s, t))
342    }
343
344    /// The connection gate (required; `AllowAll` for none).
345    pub fn gate<G2: Gate>(self, gate: G2) -> ServerBuilder<W, G2, S, T> {
346        self.map(|w, _, s, t| (w, gate, s, t))
347    }
348
349    /// The event sink (default `NoopSink`).
350    pub fn events<S2: EventSink>(self, sink: S2) -> ServerBuilder<W, G, S2, T> {
351        self.map(|w, g, _, t| (w, g, sink, t))
352    }
353
354    /// The control line (default: none).
355    pub fn control<C: ControlLine>(mut self, line: Arc<C>) -> Self {
356        self.control = Some(line);
357        self
358    }
359
360    /// Zero the used part of every buffer when it returns to a pool (default on).
361    pub fn wipe_buffers(mut self, on: bool) -> Self {
362        self.wipe = on;
363        self
364    }
365
366    /// The clock deadlines read (tests inject a `ManualClock`).
367    pub fn clock<K: Clock>(mut self, clock: Arc<K>) -> Self {
368        self.clock = Some(clock);
369        self
370    }
371}
372
373impl<W, G, S, T> ServerBuilder<W, G, S, T>
374where
375    W: Workload,
376    G: Gate,
377    S: EventSink,
378    T: TlsAcceptor,
379{
380    /// Validates everything, binds the listeners, starts the workers, and returns
381    /// the server with the door **closed**.
382    ///
383    /// # Errors
384    ///
385    /// `InvalidConfig` (`field` names the method), bind errors, `Unsupported` for a
386    /// driver this version does not have.
387    pub fn build(self) -> Result<Server, NetError> {
388        let driver = match self.driver.resolve() {
389            DriverKind::Blocking => DriverKind::Blocking,
390            other => {
391                return Err(NetError::Unsupported {
392                    feature: match other {
393                        DriverKind::Uring => "the io_uring driver (after 1.0.0)",
394                        _ => "the reactor driver (0.5.0)",
395                    },
396                    platform: std::env::consts::OS,
397                });
398            }
399        };
400        if self.listeners.is_empty() {
401            return Err(invalid("listener", ConfigReason::Missing));
402        }
403        if self.listeners.len() > 64 {
404            return Err(invalid("listener", ConfigReason::TooLarge { max: 64 }));
405        }
406        let workers = self.workers.unwrap_or(128);
407        if workers == 0 {
408            return Err(invalid("workers", ConfigReason::Zero));
409        }
410        if workers > 4096 {
411            return Err(invalid("workers", ConfigReason::TooLarge { max: 4096 }));
412        }
413        let mut names = Vec::new();
414        for spec in &self.listeners {
415            if let Some(n) = &spec.name {
416                if n.len() > 64 {
417                    return Err(invalid("name", ConfigReason::TooLarge { max: 64 }));
418                }
419                if names.contains(n) {
420                    return Err(invalid("name", ConfigReason::Inconsistent));
421                }
422                names.push(n.clone());
423            }
424        }
425        // On the blocking driver an unset global cap is one connection per worker.
426        let limits = if self.limits.max_connections().is_none() {
427            self.limits
428                .to_builder()
429                .max_connections(u32::from(workers))
430                .build()?
431        } else {
432            self.limits.clone()
433        };
434        limits.validate(workers, crate::sys::descriptor_limit())?;
435        for spec in &self.listeners {
436            limits.overlay(&spec.limits).validate(workers, None)?;
437        }
438        let _ = &self.tls;
439        engine::start(engine::Config {
440            listeners: self.listeners,
441            driver,
442            workers,
443            limits,
444            work: self.work,
445            gate: self.gate,
446            sink: self.sink,
447            control: self.control,
448            wipe: self.wipe,
449            clock: self.clock,
450        })
451    }
452}
453
454fn invalid(field: &'static str, reason: ConfigReason) -> NetError {
455    NetError::InvalidConfig { field, reason }
456}
457
458/// A running server. Dropping it stops it (with a drain), so keep it and call
459/// [`Server::wait_for_stop`] to run until a stop is requested.
460#[must_use = "dropping a Server stops it; call wait_for_stop() to run until a stop is requested"]
461pub struct Server {
462    pub(crate) shared: Arc<Shared>,
463    pub(crate) threads: Vec<JoinHandle<()>>,
464    pub(crate) listeners: Vec<ListenerInfo>,
465    pub(crate) stopped: bool,
466}
467
468impl Server {
469    /// Starts accepting on every listener; a second call is a no-op.
470    ///
471    /// # Errors
472    ///
473    /// `ShuttingDown` while draining or after shutdown; `ForkedChild`.
474    pub fn open_door(&self) -> Result<(), NetError> {
475        self.shared.open_door()
476    }
477
478    /// Stops accepting within one loop iteration; listening sockets stay open. Idempotent.
479    pub fn close_door(&self) {
480        self.shared.close_door();
481    }
482
483    /// Whether the server is accepting.
484    #[must_use]
485    pub fn door(&self) -> DoorState {
486        self.shared.door()
487    }
488
489    /// Closes the door, tells sessions to finish, closes idle connections at once,
490    /// waits for the rest until `deadline`, then kills them with `DrainDeadline`.
491    pub fn drain(&self, deadline: Instant) -> DrainReport {
492        self.shared.drain(deadline)
493    }
494
495    /// Drains, then closes the listening sockets and joins every thread.
496    pub fn shutdown(mut self, deadline: Instant) -> DrainReport {
497        self.stop(deadline)
498    }
499
500    /// Blocks until a stop is requested (`ServerHandle::request_stop`, or the
501    /// control line moves to `Draining` or `Stopping`), then shuts down.
502    pub fn wait_for_stop(mut self) -> DrainReport {
503        let deadline = self.shared.wait_for_stop_request();
504        self.stop(deadline)
505    }
506
507    fn stop(&mut self, deadline: Instant) -> DrainReport {
508        if self.stopped {
509            return DrainReport {
510                graceful: 0,
511                killed: 0,
512                took: Duration::ZERO,
513            };
514        }
515        self.stopped = true;
516        let report = self.shared.drain(deadline);
517        self.shared.stop_threads();
518        for t in self.threads.drain(..) {
519            let _ = t.join();
520        }
521        report
522    }
523
524    /// A cloneable handle for other threads.
525    #[must_use]
526    pub fn handle(&self) -> ServerHandle {
527        ServerHandle {
528            shared: self.shared.clone(),
529        }
530    }
531
532    /// Counters summed over the workers.
533    #[must_use]
534    pub fn stats(&self) -> Stats {
535        self.shared.stats()
536    }
537
538    /// Facts about each listener.
539    #[must_use]
540    pub fn listeners(&self) -> &[ListenerInfo] {
541        &self.listeners
542    }
543}
544
545impl Drop for Server {
546    fn drop(&mut self) {
547        let deadline = Instant::now() + self.shared.drain_timeout;
548        self.stop(deadline);
549    }
550}
551
552impl fmt::Debug for Server {
553    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
554        f.debug_struct("Server")
555            .field("listeners", &self.listeners)
556            .field("door", &self.door())
557            .finish()
558    }
559}
560
561/// A cloneable handle to a running server, safe to use from any thread.
562///
563/// ```no_run
564/// let server = net_io::serve("127.0.0.1:7000", |conn| { conn.send_all(b"hi\n")?; Ok(()) })?;
565/// let handle = server.handle();
566/// std::thread::spawn(move || {
567///     // ... wait for a signal from the application's signal library ...
568///     handle.request_stop();
569/// });
570/// let report = server.wait_for_stop();
571/// println!("drained: {} graceful, {} killed in {:?}", report.graceful, report.killed, report.took);
572/// # Ok::<(), net_io::NetError>(())
573/// ```
574#[derive(Clone)]
575pub struct ServerHandle {
576    pub(crate) shared: Arc<Shared>,
577}
578
579impl ServerHandle {
580    /// Aborts the connection `id`; `false` if it already closed or belongs to another server.
581    pub fn kill(&self, id: ConnectionId, reason: KillReason) -> bool {
582        self.shared.kill(id, reason)
583    }
584
585    /// Stops accepting.
586    pub fn close_door(&self) {
587        self.shared.close_door();
588    }
589
590    /// Makes `Server::wait_for_stop` return after the drain. Does not block.
591    pub fn request_stop(&self) {
592        self.shared.request_stop(None);
593    }
594
595    /// Interrupts every worker's and acceptor's wait so a control-line change is seen at once.
596    pub fn wake_all(&self) {
597        self.shared.wake_all();
598    }
599
600    /// The completion queue of `worker`.
601    #[must_use]
602    pub fn completer(&self, worker: WorkerId) -> Option<crate::completer::Completer> {
603        self.shared.completer(worker)
604    }
605
606    /// Counters summed over the workers.
607    #[must_use]
608    pub fn stats(&self) -> Stats {
609        self.shared.stats()
610    }
611
612    /// Whether the server is accepting.
613    #[must_use]
614    pub fn door(&self) -> DoorState {
615        self.shared.door()
616    }
617}
618
619impl fmt::Debug for ServerHandle {
620    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
621        f.debug_struct("ServerHandle")
622            .field("door", &self.door())
623            .finish()
624    }
625}
626
627impl Stats {
628    pub(crate) fn from_parts(
629        mut base: Stats,
630        refused_by: [u64; RefusalCause::COUNT],
631        closed_by: [u64; CloseReason::COUNT],
632    ) -> Stats {
633        base.refused_by = refused_by;
634        base.closed_by = closed_by;
635        base
636    }
637}
net_io/server/mod.rs

net_io/server/
mod.rs
