ntex_server/net/
builder.rs

1use std::{fmt, io, net, sync::Arc};
2
3use ntex_io::Io;
4use ntex_rt::System;
5use ntex_service::{IntoService, Service, cfg::SharedCfg};
6use ntex_util::time::Millis;
7use socket2::{Domain, SockAddr, Socket, Type};
8
9use crate::{Affinity, LoadBalancing, NoConfig, Server, ServerAppConfig, WorkerPool};
10
11use super::accept::AcceptLoop;
12use super::config::ServiceConfig;
13use super::factory::{self, FactoryServiceType};
14use super::{Connection, StreamServer, Token, socket::Listener};
15
16/// Builder for a network server.
17///
18/// Register listeners and their service factories, configure the worker pool,
19/// and call [`run`](Self::run) to start the server.
20pub struct ServerBuilder<Cfg = NoConfig> {
21    name: String,
22    token: Token,
23    backlog: i32,
24    state: Arc<Cfg>,
25    services: Vec<FactoryServiceType<Cfg>>,
26    sockets: Vec<(Token, String, Listener)>,
27    accept: AcceptLoop,
28    pool: WorkerPool,
29}
30
31impl Default for ServerBuilder {
32    fn default() -> Self {
33        Self::new(NoConfig)
34    }
35}
36
37impl<Cfg> ServerBuilder<Cfg>
38where
39    Cfg: ServerAppConfig,
40{
41    #[must_use]
42    /// Creates a server builder with the specified application configuration.
43    pub fn new(cfg: Cfg) -> ServerBuilder<Cfg> {
44        let sys = System::current();
45        let mut accept = AcceptLoop::default();
46        accept.name(sys.name());
47        if sys.testing() {
48            accept.testing();
49        }
50
51        ServerBuilder {
52            accept,
53            name: sys.name().to_string(),
54            token: Token(0),
55            state: Arc::new(cfg),
56            services: Vec::new(),
57            sockets: Vec::new(),
58            backlog: 2048,
59            pool: WorkerPool::default().name(sys.name()),
60        }
61    }
62
63    #[must_use]
64    /// Sets the server name.
65    ///
66    /// The name is also used for the accept and worker thread names. It
67    /// defaults to the current system name.
68    pub fn name<T: AsRef<str>>(mut self, name: T) -> Self {
69        self.name = name.as_ref().to_string();
70        self.accept.name(self.name.as_str());
71        self.pool = self.pool.name(self.name.as_str());
72        self
73    }
74
75    #[must_use]
76    /// Sets the number of worker threads to start.
77    ///
78    /// By default, the server uses the number of available logical CPUs.
79    pub fn workers(mut self, num: usize) -> Self {
80        self.pool = self.pool.workers(num);
81        self
82    }
83
84    #[must_use]
85    /// Sets the maximum number of pending connections.
86    ///
87    /// This refers to the number of clients that can be waiting to be served.
88    /// Exceeding this number results in the client getting an error when
89    /// attempting to connect. It should only affect servers under significant
90    /// load.
91    ///
92    /// Generally set in the 64-2048 range. Default value is 2048.
93    ///
94    /// It applies to listeners created by later [`bind`](Self::bind) and
95    /// [`configure`](Self::configure) calls. It does not affect listeners
96    /// passed to [`listen`](Self::listen) or `listen_uds`.
97    pub fn backlog(mut self, num: i32) -> Self {
98        self.backlog = num;
99        self
100    }
101
102    #[must_use]
103    /// Sets the maximum per-worker number of concurrent connections.
104    ///
105    /// A worker stops taking new connections while it is at this limit. When
106    /// no worker can take a connection, the listeners stop accepting.
107    ///
108    /// The limit is a process-wide setting shared by every server in the
109    /// process. Set it before the server starts, because each worker reads
110    /// it when its first service is created.
111    ///
112    /// The default is 25,600 connections per worker.
113    pub fn max_connections(self, num: usize) -> Self {
114        super::max_concurrent_connections(num);
115        self
116    }
117
118    #[must_use]
119    /// Stops the current ntex runtime after the server has stopped.
120    ///
121    /// By default, "stop runtime" is disabled.
122    pub fn stop_runtime(mut self) -> Self {
123        self.pool = self.pool.stop_runtime();
124        self
125    }
126
127    #[must_use]
128    /// Stops the server when one of the workers fails.
129    ///
130    /// A worker fails when it panics or its service cannot be created. The
131    /// stop is graceful only if [`graceful_shutdown`](Self::graceful_shutdown)
132    /// is enabled. Without this option, a failed worker is restarted.
133    ///
134    /// By default, "stop on panic" is disabled.
135    pub fn stop_on_panic(mut self) -> Self {
136        self.pool = self.pool.stop_on_panic();
137        self
138    }
139
140    #[must_use]
141    /// Disables signal handling.
142    ///
143    /// By default, the server stops on SIGINT, SIGTERM, and SIGQUIT.
144    pub fn disable_signals(mut self) -> Self {
145        self.pool = self.pool.disable_signals();
146        self
147    }
148
149    #[must_use]
150    /// Sets the CPU affinity of worker threads.
151    ///
152    /// See [`WorkerPool::affinity`].
153    pub fn affinity(mut self, affinity: Affinity) -> Self {
154        self.pool = self.pool.affinity(affinity);
155        self
156    }
157
158    #[must_use]
159    /// Sets how new connections are distributed between workers.
160    ///
161    /// See [`WorkerPool::load_balancing`].
162    pub fn load_balancing(mut self, balancing: LoadBalancing) -> Self {
163        self.pool = self.pool.load_balancing(balancing);
164        self
165    }
166
167    #[must_use]
168    /// Enables graceful shutdown on SIGQUIT, fatal signals, and panics.
169    ///
170    /// When enabled, SIGQUIT, SIGSEGV, SIGABRT, application panics, and
171    /// worker failures with "stop on panic" stop the server gracefully.
172    /// SIGTERM always stops gracefully and SIGINT always stops immediately.
173    ///
174    /// By default, these events stop the server immediately.
175    pub fn graceful_shutdown(mut self) -> Self {
176        self.pool = self.pool.graceful_shutdown();
177        self
178    }
179
180    #[must_use]
181    /// Timeout for graceful worker shutdown.
182    ///
183    /// After receiving a stop signal, workers have this much time to finish
184    /// serving requests. Workers that are still alive after the timeout are
185    /// forcefully dropped.
186    ///
187    /// This bounds the worker as a whole, not an individual connection. Each
188    /// connection is bound separately by `IoConfig::set_shutdown_timeout`, so
189    /// this value should leave room for the connections a worker is still
190    /// draining to shut down themselves.
191    ///
192    /// By default, the timeout is set to 30 seconds.
193    pub fn graceful_shutdown_timeout<T: Into<Millis>>(mut self, timeout: T) -> Self {
194        self.pool = self.pool.graceful_shutdown_timeout(timeout);
195        self
196    }
197
198    /// Runs asynchronous configuration as part of the server building
199    /// process.
200    ///
201    /// Listeners registered on the [`ServiceConfig`] are added to the server.
202    /// Services for them are attached per worker in
203    /// [`ServiceConfig::on_worker_start`]. This is useful for moving parts of
204    /// the configuration to a different module or library.
205    pub async fn configure<F>(mut self, f: F) -> io::Result<Self>
206    where
207        F: AsyncFnOnce(ServiceConfig<Cfg>) -> io::Result<()>,
208    {
209        let cfg = ServiceConfig::new(self.token, self.backlog);
210
211        f(cfg.clone()).await?;
212
213        let (token, sockets, factory) = cfg.into_factory();
214        self.token = token;
215        self.sockets.extend(sockets);
216        self.services.push(factory);
217
218        Ok(self)
219    }
220
221    #[allow(clippy::needless_pass_by_value)]
222    /// Binds TCP listeners and registers a service factory.
223    ///
224    /// A listener is created for every address resolved from `addr`. Binding
225    /// succeeds if at least one of them binds; addresses that fail to bind
226    /// are skipped.
227    ///
228    /// `cfg` is the I/O configuration for accepted connections. `factory` is
229    /// called once per worker with that worker's application state and
230    /// returns the connection service.
231    pub fn bind<F, S, I>(
232        mut self,
233        name: impl AsRef<str>,
234        addr: impl net::ToSocketAddrs,
235        cfg: impl Into<SharedCfg>,
236        factory: F,
237    ) -> io::Result<Self>
238    where
239        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
240        S: Service<Cfg::State, Io> + 'static,
241        I: IntoService<S, Cfg::State, Io> + 'static,
242    {
243        let cfg = cfg.into();
244        let sockets = bind_addr(addr, self.backlog)?;
245
246        let mut tokens = Vec::new();
247        for lst in sockets {
248            let token = self.token.next();
249            self.sockets
250                .push((token, name.as_ref().to_string(), Listener::from_tcp(lst)));
251            tokens.push((token, cfg.clone()));
252        }
253
254        self.services.push(factory::create_factory_service(
255            name.as_ref().to_string(),
256            tokens,
257            factory,
258        ));
259
260        Ok(self)
261    }
262
263    #[cfg(unix)]
264    /// Binds a Unix domain socket and registers a service factory.
265    ///
266    /// Any existing file at `addr` is removed before binding. The socket file
267    /// is removed again when the server stops. See [`bind`](Self::bind) for
268    /// `cfg` and `factory`.
269    pub fn bind_uds<F, I, S>(
270        self,
271        name: impl AsRef<str>,
272        addr: impl AsRef<std::path::Path>,
273        cfg: impl Into<SharedCfg>,
274        factory: F,
275    ) -> io::Result<Self>
276    where
277        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
278        I: IntoService<S, Cfg::State, Io> + 'static,
279        S: Service<Cfg::State, Io> + 'static,
280    {
281        use std::os::unix::net::UnixListener;
282
283        // The path must not exist when we try to bind.
284        // Try to remove it to avoid bind error.
285        if let Err(e) = std::fs::remove_file(addr.as_ref()) {
286            // NotFound is expected and not an issue. Anything else is.
287            if e.kind() != std::io::ErrorKind::NotFound {
288                return Err(e);
289            }
290        }
291
292        let lst = UnixListener::bind(addr)?;
293        self.listen_uds(name, lst, cfg.into(), factory)
294    }
295
296    #[cfg(unix)]
297    /// Registers a service factory for an existing Unix domain listener.
298    ///
299    /// This is useful for socket activation, including listeners acquired
300    /// through systemd. The listener is switched to non-blocking mode. See
301    /// [`bind`](Self::bind) for `cfg` and `factory`.
302    pub fn listen_uds<F, I, S>(
303        mut self,
304        name: impl AsRef<str>,
305        lst: std::os::unix::net::UnixListener,
306        cfg: impl Into<SharedCfg>,
307        factory: F,
308    ) -> io::Result<Self>
309    where
310        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
311        I: IntoService<S, Cfg::State, Io> + 'static,
312        S: Service<Cfg::State, Io> + 'static,
313    {
314        let token = self.token.next();
315        self.services.push(factory::create_factory_service(
316            name.as_ref().to_string(),
317            vec![(token, cfg.into())],
318            factory,
319        ));
320        self.sockets
321            .push((token, name.as_ref().to_string(), Listener::from_uds(lst)));
322        Ok(self)
323    }
324
325    /// Registers a service factory for an existing TCP listener.
326    ///
327    /// The listener is switched to non-blocking mode. See
328    /// [`bind`](Self::bind) for `cfg` and `factory`.
329    pub fn listen<F, S, I>(
330        mut self,
331        name: impl AsRef<str>,
332        lst: net::TcpListener,
333        cfg: impl Into<SharedCfg>,
334        factory: F,
335    ) -> io::Result<Self>
336    where
337        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
338        S: Service<Cfg::State, Io> + 'static,
339        I: IntoService<S, Cfg::State, Io> + 'static,
340    {
341        let token = self.token.next();
342        self.services.push(factory::create_factory_service(
343            name.as_ref().to_string(),
344            vec![(token, cfg.into())],
345            factory,
346        ));
347        self.sockets
348            .push((token, name.as_ref().to_string(), Listener::from_tcp(lst)));
349        Ok(self)
350    }
351
352    /// Starts processing incoming connections and returns a server controller.
353    ///
354    /// # Panics
355    ///
356    /// Panics if no listener has been registered.
357    pub fn run(self) -> Server<Connection> {
358        assert!(
359            !self.sockets.is_empty(),
360            "Server should have at least one bound socket"
361        );
362        let srv = StreamServer::new(self.accept.notify(), self.state, self.services);
363        let svc = self.pool.run(srv);
364
365        svc.metrics_ref().listeners(
366            self.sockets
367                .iter()
368                .map(|sock| (sock.1.clone(), sock.2.local_addr().to_string())),
369        );
370        let sockets = self
371            .sockets
372            .into_iter()
373            .map(|sock| {
374                log::info!("Starting \"{}\" service on {}", sock.1, sock.2);
375                (sock.0, sock.2)
376            })
377            .collect();
378        self.accept.start(sockets, svc.clone());
379
380        svc
381    }
382}
383
384impl<Cfg> fmt::Debug for ServerBuilder<Cfg> {
385    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
386        f.debug_struct("ServerBuilder")
387            .field("name", &self.name)
388            .field("token", &self.token)
389            .field("backlog", &self.backlog)
390            .field("sockets", &self.sockets)
391            .field("accept", &self.accept)
392            .field("worker-pool", &self.pool)
393            .finish()
394    }
395}
396
397/// Binds TCP listeners for every address resolved from `addr`.
398///
399/// Succeeds if at least one address binds and returns only the listeners that
400/// bound. Otherwise, returns the last bind error.
401pub fn bind_addr<S: net::ToSocketAddrs>(
402    addr: S,
403    backlog: i32,
404) -> io::Result<Vec<net::TcpListener>> {
405    let mut err = None;
406    let mut succ = false;
407    let mut sockets = Vec::new();
408    for addr in addr.to_socket_addrs()? {
409        match create_tcp_listener(addr, backlog) {
410            Ok(lst) => {
411                succ = true;
412                sockets.push(lst);
413            }
414            Err(e) => err = Some(e),
415        }
416    }
417
418    if succ {
419        Ok(sockets)
420    } else if let Some(e) = err.take() {
421        Err(e)
422    } else {
423        Err(io::Error::new(
424            io::ErrorKind::InvalidInput,
425            "Cannot bind to address.",
426        ))
427    }
428}
429
430/// Creates and binds a TCP listener with the specified listen backlog.
431pub fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::TcpListener> {
432    let builder = match addr {
433        net::SocketAddr::V4(_) => Socket::new(Domain::IPV4, Type::STREAM, None)?,
434        net::SocketAddr::V6(_) => Socket::new(Domain::IPV6, Type::STREAM, None)?,
435    };
436
437    // On Windows, this allows rebinding sockets which are actively in use,
438    // which allows “socket hijacking”, so we explicitly don't set it here.
439    // https://docs.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse
440    #[cfg(not(windows))]
441    builder.set_reuse_address(true)?;
442
443    builder.bind(&SockAddr::from(addr))?;
444    builder.listen(backlog)?;
445    Ok(net::TcpListener::from(builder))
446}
447
448#[cfg(test)]
449mod tests {
450    use super::*;
451
452    #[test]
453    fn test_bind_addr() {
454        let addrs: Vec<net::SocketAddr> = Vec::new();
455        assert!(bind_addr(&addrs[..], 10).is_err());
456    }
457
458    #[ntex::test]
459    async fn test_debug() {
460        let builder = ServerBuilder::default();
461        assert!(format!("{builder:?}").contains("ServerBuilder"));
462    }
463}
ntex_server/net/builder.rs

ntex_server/net/
builder.rs
