ntex_server/
lib.rs

1//! Worker-based server infrastructure for ntex.
2//!
3//! [`WorkerPool`] runs services across one or more worker threads. The [`net`]
4//! module builds TCP and Unix domain socket servers on top of that pool.
5//! [`Server`] is the controller used to pause, resume, stop, or await a running
6//! server.
7
8#![deny(clippy::pedantic)]
9#![allow(
10    async_fn_in_trait,
11    clippy::clone_on_copy,
12    clippy::must_use_candidate,
13    clippy::missing_fields_in_debug,
14    clippy::missing_errors_doc,
15    clippy::missing_panics_doc,
16    clippy::unused_async
17)]
18
19use ntex_service::Service;
20
21mod affinity;
22mod balance;
23mod manager;
24mod metrics;
25pub mod net;
26mod pool;
27mod server;
28mod state;
29mod status;
30mod wrk;
31
32pub use self::affinity::Affinity;
33pub use self::balance::LoadBalancing;
34pub use self::metrics::{ListenerMetrics, RuntimeMetrics, ServerMetrics, WorkerMetrics};
35pub use self::pool::WorkerPool;
36pub use self::server::Server;
37pub use self::state::{NoConfig, ServerAppConfig};
38pub use self::status::ServerStatus;
39pub use self::wrk::{Worker, WorkerStatus, WorkerStop};
40
41/// Identifier assigned to a server worker.
42#[derive(Default, Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
43pub struct WorkerId(pub(crate) usize);
44
45impl WorkerId {
46    pub(self) fn next(&mut self) -> WorkerId {
47        let id = WorkerId(self.0);
48        self.0 += 1;
49        id
50    }
51}
52
53/// Worker service configuration.
54pub trait ServerConfiguration: Send + Clone + 'static {
55    /// Item dispatched to a worker service.
56    type Item: Send + 'static;
57    /// Service created independently for each worker.
58    type Service: Service<(), Self::Item, Res = (), Error = ()> + 'static;
59
60    /// Creates the service used by one worker.
61    async fn create(&self) -> std::io::Result<Self::Service>;
62
63    /// Called when the server status changes from
64    /// [`Ready`](ServerStatus::Ready) to paused or unavailable.
65    ///
66    /// The server is paused by [`Server::pause`], and it is unavailable
67    /// while no worker can accept items. Not called when the server stops.
68    fn pause(&self) {}
69
70    /// Called when the server status becomes [`Ready`](ServerStatus::Ready).
71    ///
72    /// The server is ready while it is not paused and a worker is available.
73    fn resume(&self) {}
74
75    /// Called when the server's command loop exits.
76    ///
77    /// This happens after a stop requested through [`Server::stop`] or a
78    /// signal, graceful or not, once [`stop`](Self::stop) has completed.
79    fn terminate(&self) {}
80
81    /// Performs asynchronous cleanup when the server stops.
82    ///
83    /// Called once, before the workers are stopped.
84    async fn stop(&self) {}
85
86    /// Returns the index of the logical CPU that receives the network packets
87    /// of the item, if known.
88    ///
89    /// Used by [`LoadBalancing::IncomingCpu`] to send the item to the worker
90    /// placed on that CPU. CPUs are numbered as in [`Worker::start`]. Network
91    /// servers return the receive-side scaling (RSS) CPU of the connection on
92    /// Windows, and its `SO_INCOMING_CPU` on Linux and Android. The default
93    /// returns `None`.
94    ///
95    /// Called on the thread that submits the item with [`Server::process`],
96    /// e.g. the accept thread of a network server.
97    /// Calls are not concurrent. An item submitted while the method runs,
98    /// e.g. by the method itself, is sent without its incoming CPU.
99    fn incoming_cpu(&self, item: &Self::Item) -> Option<usize> {
100        let _ = item;
101        None
102    }
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108
109    #[test]
110    fn worker_id_next() {
111        let mut id = WorkerId::default();
112        assert_eq!(id.next(), WorkerId(0));
113        assert_eq!(id.next(), WorkerId(1));
114        assert_eq!(id, WorkerId(2));
115    }
116}
ntex_server/lib.rs

ntex_server/
lib.rs
