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    /// SIGSEGV and SIGABRT require runtime fatal signal handling, application
203    /// panics require runtime panic handling.
204    ///
205    /// By default, these events stop the server immediately.
206    pub fn graceful_shutdown(mut self) -> Self {
207        self.pool = self.pool.graceful_shutdown();
208        self
209    }
210
211    #[must_use]
212    /// Timeout for graceful worker shutdown.
213    ///
214    /// After receiving a stop signal, workers have this much time to finish
215    /// serving requests. Workers that are still alive after the timeout are
216    /// forcefully dropped.
217    ///
218    /// This bounds the worker as a whole, not an individual connection. Each
219    /// connection is bound separately by `IoConfig::set_shutdown_timeout`, so
220    /// this value should leave room for the connections a worker is still
221    /// draining to shut down themselves.
222    ///
223    /// By default, the timeout is set to 30 seconds.
224    pub fn graceful_shutdown_timeout<T: Into<Millis>>(mut self, timeout: T) -> Self {
225        self.pool = self.pool.graceful_shutdown_timeout(timeout);
226        self
227    }
228
229    /// Runs asynchronous configuration as part of the server building
230    /// process.
231    ///
232    /// Listeners registered on the [`ServiceConfig`] are added to the server.
233    /// Services for them are attached per worker in
234    /// [`ServiceConfig::on_worker_start`]. This is useful for moving parts of
235    /// the configuration to a different module or library.
236    pub async fn configure<F>(mut self, f: F) -> io::Result<Self>
237    where
238        F: AsyncFnOnce(ServiceConfig<Cfg>) -> io::Result<()>,
239    {
240        let cfg = ServiceConfig::new(self.token, self.backlog);
241
242        f(cfg.clone()).await?;
243
244        let (token, sockets, factory) = cfg.into_factory();
245        self.token = token;
246        self.sockets.extend(sockets);
247        self.services.push(factory);
248
249        Ok(self)
250    }
251
252    #[allow(clippy::needless_pass_by_value)]
253    /// Binds TCP listeners and registers a service factory.
254    ///
255    /// A listener is created for every address resolved from `addr`. Binding
256    /// succeeds if at least one of them binds; addresses that fail to bind
257    /// are skipped.
258    ///
259    /// `cfg` is the I/O configuration for accepted connections. `factory` is
260    /// called once per worker with that worker's application state and
261    /// returns the connection service.
262    pub fn bind<F, S, I>(
263        mut self,
264        name: impl AsRef<str>,
265        addr: impl net::ToSocketAddrs,
266        cfg: impl Into<SharedCfg>,
267        factory: F,
268    ) -> io::Result<Self>
269    where
270        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
271        S: Service<Cfg::State, Io> + 'static,
272        I: IntoService<S, Cfg::State, Io> + 'static,
273    {
274        let cfg = cfg.into();
275        let sockets = bind_addr(addr, self.backlog)?;
276
277        let mut tokens = Vec::new();
278        for lst in sockets {
279            let token = self.token.next();
280            self.sockets
281                .push((token, name.as_ref().to_string(), Listener::from_tcp(lst)));
282            tokens.push((token, cfg.clone()));
283        }
284
285        self.services.push(factory::create_factory_service(
286            name.as_ref().to_string(),
287            tokens,
288            factory,
289        ));
290
291        Ok(self)
292    }
293
294    #[cfg(unix)]
295    /// Binds a Unix domain socket and registers a service factory.
296    ///
297    /// Any existing file at `addr` is removed before binding. The socket file
298    /// is removed again when the server stops. See [`bind`](Self::bind) for
299    /// `cfg` and `factory`.
300    pub fn bind_uds<F, I, S>(
301        self,
302        name: impl AsRef<str>,
303        addr: impl AsRef<std::path::Path>,
304        cfg: impl Into<SharedCfg>,
305        factory: F,
306    ) -> io::Result<Self>
307    where
308        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
309        I: IntoService<S, Cfg::State, Io> + 'static,
310        S: Service<Cfg::State, Io> + 'static,
311    {
312        use std::os::unix::net::UnixListener;
313
314        // The path must not exist when we try to bind.
315        // Try to remove it to avoid bind error.
316        if let Err(e) = std::fs::remove_file(addr.as_ref()) {
317            // NotFound is expected and not an issue. Anything else is.
318            if e.kind() != std::io::ErrorKind::NotFound {
319                return Err(e);
320            }
321        }
322
323        let lst = UnixListener::bind(addr)?;
324        self.listen_uds(name, lst, cfg.into(), factory)
325    }
326
327    #[cfg(unix)]
328    /// Registers a service factory for an existing Unix domain listener.
329    ///
330    /// This is useful for socket activation, including listeners acquired
331    /// through systemd. The listener is switched to non-blocking mode. See
332    /// [`bind`](Self::bind) for `cfg` and `factory`.
333    pub fn listen_uds<F, I, S>(
334        mut self,
335        name: impl AsRef<str>,
336        lst: std::os::unix::net::UnixListener,
337        cfg: impl Into<SharedCfg>,
338        factory: F,
339    ) -> io::Result<Self>
340    where
341        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
342        I: IntoService<S, Cfg::State, Io> + 'static,
343        S: Service<Cfg::State, Io> + 'static,
344    {
345        let token = self.token.next();
346        self.services.push(factory::create_factory_service(
347            name.as_ref().to_string(),
348            vec![(token, cfg.into())],
349            factory,
350        ));
351        self.sockets
352            .push((token, name.as_ref().to_string(), Listener::from_uds(lst)));
353        Ok(self)
354    }
355
356    /// Registers a service factory for an existing TCP listener.
357    ///
358    /// The listener is switched to non-blocking mode. See
359    /// [`bind`](Self::bind) for `cfg` and `factory`.
360    pub fn listen<F, S, I>(
361        mut self,
362        name: impl AsRef<str>,
363        lst: net::TcpListener,
364        cfg: impl Into<SharedCfg>,
365        factory: F,
366    ) -> io::Result<Self>
367    where
368        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
369        S: Service<Cfg::State, Io> + 'static,
370        I: IntoService<S, Cfg::State, Io> + 'static,
371    {
372        let token = self.token.next();
373        self.services.push(factory::create_factory_service(
374            name.as_ref().to_string(),
375            vec![(token, cfg.into())],
376            factory,
377        ));
378        self.sockets
379            .push((token, name.as_ref().to_string(), Listener::from_tcp(lst)));
380        Ok(self)
381    }
382
383    /// Starts processing incoming connections and returns a server controller.
384    ///
385    /// # Panics
386    ///
387    /// Panics if no listener has been registered.
388    pub fn run(mut self) -> Server<Connection> {
389        assert!(
390            !self.sockets.is_empty(),
391            "Server should have at least one bound socket"
392        );
393        self.accept.acceptors(self.acceptors);
394        let srv = StreamServer::new(
395            self.accept.notify(),
396            self.state,
397            self.services,
398            self.max_conns,
399        );
400        let svc = self.pool.run(srv);
401
402        svc.metrics_ref().listeners(
403            self.sockets
404                .iter()
405                .map(|sock| (sock.1.clone(), sock.2.local_addr().to_string())),
406        );
407        let sockets = self
408            .sockets
409            .into_iter()
410            .map(|sock| {
411                log::info!("Starting \"{}\" service on {}", sock.1, sock.2);
412                (sock.0, sock.2)
413            })
414            .collect();
415        self.accept.start(sockets, svc.clone());
416
417        svc
418    }
419}
420
421impl<Cfg> fmt::Debug for ServerBuilder<Cfg> {
422    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
423        f.debug_struct("ServerBuilder")
424            .field("name", &self.name)
425            .field("token", &self.token)
426            .field("backlog", &self.backlog)
427            .field("sockets", &self.sockets)
428            .field("accept", &self.accept)
429            .field("acceptors", &self.acceptors)
430            .field("max_conns", &self.max_conns)
431            .field("worker-pool", &self.pool)
432            .finish()
433    }
434}
435
436/// Binds TCP listeners for every address resolved from `addr`.
437///
438/// Succeeds if at least one address binds and returns only the listeners that
439/// bound. Otherwise, returns the last bind error.
440pub fn bind_addr<S: net::ToSocketAddrs>(
441    addr: S,
442    backlog: i32,
443) -> io::Result<Vec<net::TcpListener>> {
444    let mut err = None;
445    let mut succ = false;
446    let mut sockets = Vec::new();
447    for addr in addr.to_socket_addrs()? {
448        match create_tcp_listener(addr, backlog) {
449            Ok(lst) => {
450                succ = true;
451                sockets.push(lst);
452            }
453            Err(e) => err = Some(e),
454        }
455    }
456
457    if succ {
458        Ok(sockets)
459    } else if let Some(e) = err.take() {
460        Err(e)
461    } else {
462        Err(io::Error::new(
463            io::ErrorKind::InvalidInput,
464            "Cannot bind to address.",
465        ))
466    }
467}
468
469/// Creates and binds a TCP listener with the specified listen backlog.
470pub fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::TcpListener> {
471    let builder = match addr {
472        net::SocketAddr::V4(_) => Socket::new(Domain::IPV4, Type::STREAM, None)?,
473        net::SocketAddr::V6(_) => Socket::new(Domain::IPV6, Type::STREAM, None)?,
474    };
475
476    // On Windows, this allows rebinding sockets which are actively in use,
477    // which allows “socket hijacking”, so we explicitly don't set it here.
478    // https://docs.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse
479    #[cfg(not(windows))]
480    builder.set_reuse_address(true)?;
481
482    builder.bind(&SockAddr::from(addr))?;
483    builder.listen(backlog)?;
484    Ok(net::TcpListener::from(builder))
485}
486
487#[cfg(test)]
488mod tests {
489    use super::*;
490
491    #[test]
492    fn test_bind_addr() {
493        let addrs: Vec<net::SocketAddr> = Vec::new();
494        assert!(bind_addr(&addrs[..], 10).is_err());
495    }
496
497    #[ntex::test]
498    async fn test_debug() {
499        let builder = ServerBuilder::default();
500        assert!(format!("{builder:?}").contains("ServerBuilder"));
501    }
502
503    /// STATEMENT: A wakeup of a paused accept loop does not re-register its
504    /// listeners, so connections stay in the listen backlog until resume.
505    ///
506    /// HOW:
507    /// - Start a server, wait until it is ready, and pause it.
508    /// - Wake the paused accept loop with `AcceptorCommand::Timer`.
509    /// - Connect a client and check that the listener metrics count no
510    ///   accepted connection after 200ms.
511    /// - Resume the server and check that the connection is accepted.
512    #[ntex::test]
513    async fn test_paused_wakeup_does_not_accept() {
514        use crate::ServerStatus;
515        use crate::net::AcceptorCommand;
516        use ntex_service::{cfg::SharedCfg, fn_service};
517
518        let lst = net::TcpListener::bind("127.0.0.1:0").unwrap();
519        let addr = lst.local_addr().unwrap();
520        let builder = ServerBuilder::default()
521            .workers(1)
522            .acceptors(1)
523            .disable_signals()
524            .listen("test", lst, SharedCfg::default(), async |&()| {
525                fn_service(async |_: ntex_io::Io| Ok::<_, ()>(()))
526            })
527            .unwrap();
528        let notify = builder.accept.notify();
529        let srv = builder.run();
530        ntex_util::time::timeout(Millis(3000), async {
531            while srv.status() != ServerStatus::Ready {
532                srv.status_changed().await;
533            }
534        })
535        .await
536        .unwrap();
537
538        srv.pause().await;
539        ntex_util::time::sleep(Millis(100)).await;
540        notify.send(AcceptorCommand::Timer);
541        ntex_util::time::sleep(Millis(100)).await;
542
543        let _conn = net::TcpStream::connect(addr).unwrap();
544        ntex_util::time::sleep(Millis(200)).await;
545        assert_eq!(srv.metrics().listeners[0].accepted, 0);
546
547        srv.resume().await;
548        ntex_util::time::timeout(Millis(3000), async {
549            while srv.metrics().listeners[0].accepted == 0 {
550                ntex_util::time::sleep(Millis(10)).await;
551            }
552        })
553        .await
554        .unwrap();
555        srv.stop(false).await;
556    }
557}
ntex_server/net/builder.rs

ntex_server/net/
builder.rs
