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

ntex_server/net/
builder.rs
