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

ntex_server/net/
builder.rs
