actix_server/
builder.rs

1use std::{future::Future, io, num::NonZeroUsize, time::Duration};
2
3use actix_rt::net::TcpStream;
4use futures_core::future::BoxFuture;
5use tokio::sync::{
6    mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
7    watch,
8};
9
10use crate::{
11    availability::MAX_WORKERS,
12    server::ServerCommand,
13    service::{InternalServiceFactory, ServerServiceFactory, StreamNewService},
14    socket::{create_mio_tcp_listener, MioListener, MioTcpListener, StdTcpListener, ToSocketAddrs},
15    worker::ServerWorkerConfig,
16    Server,
17};
18
19/// Multipath TCP (MPTCP) preference.
20///
21/// Currently only useful on Linux.
22///
23#[cfg_attr(target_os = "linux", doc = "Also see [`ServerBuilder::mptcp()`].")]
24#[derive(Debug, Clone)]
25pub enum MpTcp {
26    /// MPTCP will not be used when binding sockets.
27    Disabled,
28
29    /// MPTCP will be attempted when binding sockets. If errors occur, regular TCP will be
30    /// attempted, too.
31    TcpFallback,
32
33    /// MPTCP will be used when binding sockets (with no fallback).
34    NoFallback,
35}
36
37/// [Server] builder.
38pub struct ServerBuilder {
39    pub(crate) threads: usize,
40    pub(crate) token: usize,
41    pub(crate) backlog: u32,
42    pub(crate) factories: Vec<Box<dyn InternalServiceFactory>>,
43    pub(crate) sockets: Vec<(usize, String, MioListener)>,
44    pub(crate) mptcp: MpTcp,
45    pub(crate) exit: bool,
46    pub(crate) listen_os_signals: bool,
47    pub(crate) shutdown_signal: Option<BoxFuture<'static, ()>>,
48    pub(crate) graceful_shutdown_tx: watch::Sender<()>,
49    pub(crate) cmd_tx: UnboundedSender<ServerCommand>,
50    pub(crate) cmd_rx: UnboundedReceiver<ServerCommand>,
51    pub(crate) worker_config: ServerWorkerConfig,
52}
53
54impl Default for ServerBuilder {
55    fn default() -> Self {
56        Self::new()
57    }
58}
59
60impl ServerBuilder {
61    /// Create new Server builder instance
62    pub fn new() -> ServerBuilder {
63        Self::with_default_workers(
64            std::thread::available_parallelism().map_or(2, NonZeroUsize::get),
65        )
66    }
67
68    fn with_default_workers(threads: usize) -> ServerBuilder {
69        let (cmd_tx, cmd_rx) = unbounded_channel();
70        let (graceful_shutdown_tx, _) = watch::channel(());
71
72        ServerBuilder {
73            threads: threads.min(MAX_WORKERS),
74            token: 0,
75            factories: Vec::new(),
76            sockets: Vec::new(),
77            backlog: 2048,
78            mptcp: MpTcp::Disabled,
79            exit: false,
80            listen_os_signals: true,
81            shutdown_signal: None,
82            graceful_shutdown_tx,
83            cmd_tx,
84            cmd_rx,
85            worker_config: ServerWorkerConfig::default(),
86        }
87    }
88
89    /// Returns a signal that resolves when this server starts a graceful shutdown.
90    ///
91    /// The signal cannot stop the server. It only propagates a shutdown that was started by an OS
92    /// signal, [`ServerHandle::stop`](crate::ServerHandle::stop), or [`Self::shutdown_signal`].
93    pub fn graceful_shutdown_signal(&self) -> crate::GracefulShutdownSignal {
94        crate::GracefulShutdownSignal::new(self.graceful_shutdown_tx.subscribe())
95    }
96
97    /// Sets number of workers to start.
98    ///
99    /// See [`bind()`](Self::bind()) for more details on how worker count affects the number of
100    /// server factory instantiations.
101    ///
102    /// The default worker count is determined by [`std::thread::available_parallelism()`], capped
103    /// at 512. See its documentation for how the available parallelism is determined.
104    ///
105    /// `num` must be in the range 1–512.
106    ///
107    /// # Panics
108    ///
109    /// Panics if `num` is 0 or greater than 512.
110    pub fn workers(mut self, num: usize) -> Self {
111        assert_ne!(num, 0, "workers must be greater than 0");
112        assert!(num <= MAX_WORKERS, "workers must not exceed {MAX_WORKERS}");
113        self.threads = num;
114        self
115    }
116
117    /// Set max number of threads for each worker's blocking task thread pool.
118    ///
119    /// One thread pool is set up **per worker**; not shared across workers.
120    ///
121    /// # Examples:
122    /// ```
123    /// # use actix_server::ServerBuilder;
124    /// let builder = ServerBuilder::new()
125    ///     .workers(4) // server has 4 worker thread.
126    ///     .worker_max_blocking_threads(4); // every worker has 4 max blocking threads.
127    /// ```
128    ///
129    /// See [tokio::runtime::Builder::max_blocking_threads] for behavior reference.
130    pub fn worker_max_blocking_threads(mut self, num: usize) -> Self {
131        self.worker_config.max_blocking_threads(num);
132        self
133    }
134
135    /// Set the maximum number of pending connections.
136    ///
137    /// This refers to the number of clients that can be waiting to be served. Exceeding this number
138    /// results in the client getting an error when attempting to connect. It should only affect
139    /// servers under significant load.
140    ///
141    /// Generally set in the 64-2048 range. Default value is 2048.
142    ///
143    /// This method should be called before `bind()` method call.
144    pub fn backlog(mut self, num: u32) -> Self {
145        self.backlog = num;
146        self
147    }
148
149    /// Sets MultiPath TCP (MPTCP) preference on bound sockets.
150    ///
151    /// Multipath TCP (MPTCP) builds on top of TCP to improve connection redundancy and performance
152    /// by sharing a network data stream across multiple underlying TCP sessions. See [mptcp.dev]
153    /// for more info about MPTCP itself.
154    ///
155    /// MPTCP is available on Linux kernel version 5.6 and higher. In addition, you'll also need to
156    /// ensure the kernel option is enabled using `sysctl net.mptcp.enabled=1`.
157    ///
158    /// This method will have no effect if called after a `bind()`.
159    ///
160    /// [mptcp.dev]: https://www.mptcp.dev
161    #[cfg(target_os = "linux")]
162    pub fn mptcp(mut self, mptcp_enabled: MpTcp) -> Self {
163        self.mptcp = mptcp_enabled;
164        self
165    }
166
167    /// Sets the maximum per-worker number of concurrent connections.
168    ///
169    /// All socket listeners will stop accepting connections when this limit is reached for
170    /// each worker.
171    ///
172    /// By default max connections is set to a 25k per worker.
173    pub fn max_concurrent_connections(mut self, num: usize) -> Self {
174        self.worker_config.max_concurrent_connections(num);
175        self
176    }
177
178    #[doc(hidden)]
179    #[deprecated(since = "2.0.0", note = "Renamed to `max_concurrent_connections`.")]
180    pub fn maxconn(self, num: usize) -> Self {
181        self.max_concurrent_connections(num)
182    }
183
184    /// Sets flag to stop Actix `System` after server shutdown.
185    ///
186    /// This has no effect when server is running in a Tokio-only runtime.
187    pub fn system_exit(mut self) -> Self {
188        self.exit = true;
189        self
190    }
191
192    /// Disables OS signal handling.
193    pub fn disable_signals(mut self) -> Self {
194        self.listen_os_signals = false;
195        self
196    }
197
198    /// Specify shutdown signal from a future.
199    ///
200    /// Using this method will prevent OS signal handlers being set up.
201    ///
202    /// Typically, a `CancellationToken` will be used, but any future _can_ be.
203    ///
204    /// # Examples
205    ///
206    /// ```
207    /// # use std::io;
208    /// # use tokio::net::TcpStream;
209    /// # use actix_server::Server;
210    /// # async fn run() -> io::Result<()> {
211    /// use actix_service::fn_service;
212    /// use tokio_util::sync::CancellationToken;
213    ///
214    /// let stop_signal = CancellationToken::new();
215    ///
216    /// Server::build()
217    ///     .bind("shutdown-signal", "127.0.0.1:12345", || {
218    ///         fn_service(|_stream: TcpStream| async { Ok::<_, io::Error>(()) })
219    ///     })?
220    ///     .shutdown_signal(stop_signal.cancelled_owned())
221    ///     .run()
222    ///     .await
223    /// # }
224    /// ```
225    pub fn shutdown_signal<Fut>(mut self, shutdown_signal: Fut) -> Self
226    where
227        Fut: Future<Output = ()> + Send + 'static,
228    {
229        self.shutdown_signal = Some(Box::pin(shutdown_signal));
230        self
231    }
232
233    /// Timeout for graceful workers shutdown in seconds.
234    ///
235    /// After receiving a stop signal, workers have this much time to finish serving requests.
236    /// Workers still alive after the timeout are force dropped.
237    ///
238    /// By default shutdown timeout sets to 30 seconds.
239    pub fn shutdown_timeout(mut self, sec: u64) -> Self {
240        self.worker_config
241            .shutdown_timeout(Duration::from_secs(sec));
242        self
243    }
244
245    /// Adds new service to the server.
246    ///
247    /// Note that, if a DNS lookup is required, resolving hostnames is a blocking operation.
248    ///
249    /// # Worker Count
250    ///
251    /// The `factory` will be instantiated multiple times in most scenarios. The number of
252    /// instantiations is number of [`workers`](Self::workers()) × number of sockets resolved by
253    /// `addrs`.
254    ///
255    /// For example, if you've manually set [`workers`](Self::workers()) to 2, and use `127.0.0.1`
256    /// as the bind `addrs`, then `factory` will be instantiated twice. However, using `localhost`
257    /// as the bind `addrs` can often resolve to both `127.0.0.1` (IPv4) _and_ `::1` (IPv6), causing
258    /// the `factory` to be instantiated 4 times (2 workers × 2 bind addresses).
259    ///
260    /// Using a bind address of `0.0.0.0`, which signals to use all interfaces, may also multiple
261    /// the number of instantiations in a similar way.
262    ///
263    /// # Errors
264    ///
265    /// Returns an `io::Error` if:
266    /// - `addrs` cannot be resolved into one or more socket addresses;
267    /// - all the resolved socket addresses are already bound.
268    pub fn bind<F, U, N>(mut self, name: N, addrs: U, factory: F) -> io::Result<Self>
269    where
270        F: ServerServiceFactory<TcpStream>,
271        U: ToSocketAddrs,
272        N: AsRef<str>,
273    {
274        let sockets = bind_addr(addrs, self.backlog, &self.mptcp)?;
275
276        tracing::trace!("binding server to: {sockets:?}");
277
278        for lst in sockets {
279            let token = self.next_token();
280
281            self.factories.push(StreamNewService::create(
282                name.as_ref().to_string(),
283                token,
284                factory.clone(),
285                lst.local_addr()?,
286            ));
287
288            self.sockets
289                .push((token, name.as_ref().to_string(), MioListener::Tcp(lst)));
290        }
291
292        Ok(self)
293    }
294
295    /// Adds service to the server using a socket listener already bound.
296    ///
297    /// # Worker Count
298    ///
299    /// The `factory` will be instantiated multiple times in most scenarios. The number of
300    /// instantiations is: number of [`workers`](Self::workers()).
301    pub fn listen<F, N: AsRef<str>>(
302        mut self,
303        name: N,
304        lst: StdTcpListener,
305        factory: F,
306    ) -> io::Result<Self>
307    where
308        F: ServerServiceFactory<TcpStream>,
309    {
310        lst.set_nonblocking(true)?;
311        let addr = lst.local_addr()?;
312
313        let token = self.next_token();
314        self.factories.push(StreamNewService::create(
315            name.as_ref().to_string(),
316            token,
317            factory,
318            addr,
319        ));
320
321        self.sockets
322            .push((token, name.as_ref().to_string(), MioListener::from(lst)));
323
324        Ok(self)
325    }
326
327    /// Starts processing incoming connections and return server controller.
328    ///
329    /// # Panics
330    ///
331    /// Panics if no sockets are bound.
332    pub fn run(self) -> Server {
333        if self.sockets.is_empty() {
334            panic!("Server should have at least one bound socket");
335        } else {
336            tracing::info!("starting {} workers", self.threads);
337            Server::new(self)
338        }
339    }
340
341    fn next_token(&mut self) -> usize {
342        let token = self.token;
343        self.token += 1;
344        token
345    }
346}
347
348#[cfg(unix)]
349impl ServerBuilder {
350    /// Adds new service to the server using a UDS (unix domain socket) address.
351    ///
352    /// # Worker Count
353    ///
354    /// The `factory` will be instantiated multiple times in most scenarios. The number of
355    /// instantiations is: number of [`workers`](Self::workers()).
356    pub fn bind_uds<F, U, N>(self, name: N, addr: U, factory: F) -> io::Result<Self>
357    where
358        F: ServerServiceFactory<actix_rt::net::UnixStream>,
359        N: AsRef<str>,
360        U: AsRef<std::path::Path>,
361    {
362        // The path must not exist when we try to bind.
363        // Try to remove it to avoid bind error.
364        if let Err(err) = std::fs::remove_file(addr.as_ref()) {
365            // NotFound is expected and not an issue. Anything else is.
366            if err.kind() != std::io::ErrorKind::NotFound {
367                return Err(err);
368            }
369        }
370
371        let lst = crate::socket::StdUnixListener::bind(addr)?;
372        self.listen_uds(name, lst, factory)
373    }
374
375    /// Adds new service to the server using a UDS (unix domain socket) listener already bound.
376    ///
377    /// Useful when running as a systemd service and a socket FD is acquired externally.
378    ///
379    /// # Worker Count
380    ///
381    /// The `factory` will be instantiated multiple times in most scenarios. The number of
382    /// instantiations is: number of [`workers`](Self::workers()).
383    pub fn listen_uds<F, N: AsRef<str>>(
384        mut self,
385        name: N,
386        lst: crate::socket::StdUnixListener,
387        factory: F,
388    ) -> io::Result<Self>
389    where
390        F: ServerServiceFactory<actix_rt::net::UnixStream>,
391    {
392        use std::net::{IpAddr, Ipv4Addr};
393
394        lst.set_nonblocking(true)?;
395
396        let token = self.next_token();
397        let addr = crate::socket::StdSocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
398
399        self.factories.push(StreamNewService::create(
400            name.as_ref().to_string(),
401            token,
402            factory,
403            addr,
404        ));
405
406        self.sockets
407            .push((token, name.as_ref().to_string(), MioListener::from(lst)));
408
409        Ok(self)
410    }
411}
412
413pub(super) fn bind_addr<S: ToSocketAddrs>(
414    addr: S,
415    backlog: u32,
416    mptcp: &MpTcp,
417) -> io::Result<Vec<MioTcpListener>> {
418    let mut opt_err = None;
419    let mut success = false;
420    let mut sockets = Vec::new();
421
422    for addr in addr.to_socket_addrs()? {
423        match create_mio_tcp_listener(addr, backlog, mptcp) {
424            Ok(lst) => {
425                success = true;
426                sockets.push(lst);
427            }
428            Err(err) => opt_err = Some(err),
429        }
430    }
431
432    if success {
433        Ok(sockets)
434    } else if let Some(err) = opt_err.take() {
435        Err(err)
436    } else {
437        Err(io::Error::other("Can not bind to address."))
438    }
439}
440
441#[cfg(test)]
442mod tests {
443    use super::ServerBuilder;
444
445    #[test]
446    #[should_panic(expected = "workers must not exceed 512")]
447    fn rejects_worker_count_above_limit() {
448        ServerBuilder::new().workers(513);
449    }
450
451    #[test]
452    #[should_panic(expected = "workers must be greater than 0")]
453    fn rejects_zero_workers() {
454        ServerBuilder::new().workers(0);
455    }
456
457    #[test]
458    fn accepts_worker_count_boundaries() {
459        ServerBuilder::new().workers(1);
460        ServerBuilder::new().workers(512);
461    }
462
463    #[test]
464    fn default_worker_count_is_capped() {
465        for detected in [513, 768, 1920, usize::MAX] {
466            assert_eq!(ServerBuilder::with_default_workers(detected).threads, 512);
467        }
468    }
469
470    #[test]
471    fn oversized_default_worker_count_can_be_overridden() {
472        assert_eq!(
473            ServerBuilder::with_default_workers(768).workers(1).threads,
474            1
475        );
476    }
477
478    #[test]
479    fn supported_default_worker_count_is_preserved() {
480        for detected in [1, 2, 64, 512] {
481            assert_eq!(
482                ServerBuilder::with_default_workers(detected).threads,
483                detected
484            );
485        }
486    }
487}
actix_server/builder.rs

actix_server/
builder.rs
