prepare service 2.0.0-beta.4 release (#269)

Co-authored-by: lishuo <lishuo.03@bytedance.com>

actix-macros/Cargo.toml

@@ -17,7 +17,7 @@ quote = "1.0.3"
 syn = { version = "^1", features = ["full"] }
 
 [dev-dependencies]
-actix-rt = "2.0.0-beta.3"
+actix-rt = "2.0.0"
 
 futures-util = { version = "0.3", default-features = false }
 trybuild = "1"

actix-rt/README.md

@@ -2,6 +2,4 @@
 
 > Tokio-based single-threaded async runtime for the Actix ecosystem.
 
-See documentation for detailed explanations these components: [https://docs.rs/actix-rt][docs].
+See crate documentation for more: https://docs.rs/actix-rt.
-
-[docs]: https://docs.rs/actix-rt

actix-server/CHANGES.md

@@ -2,8 +2,12 @@
 
 ## Unreleased - 2021-xx-xx
 * Hidden `ServerBuilder::start` method has been removed. Use `ServerBuilder::run`. [#246]
+* Add retry for EINTR(`io::Interrupted`) in `Accept`'s poll loop. [#264]
+* Add `ServerBuilder::worker_max_blocking_threads` for customize blocking thread pool. [#265]
 
 [#246]: https://github.com/actix/actix-net/pull/246
+[#264]: https://github.com/actix/actix-net/pull/264
+[#265]: https://github.com/actix/actix-net/pull/265
 
 
 ## 2.0.0-beta.2 - 2021-01-03

actix-server/Cargo.toml

@@ -25,7 +25,7 @@ default = []
 [dependencies]
 actix-codec = "0.4.0-beta.1"
 actix-rt = { version = "2.0.0", default-features = false }
-actix-service = "2.0.0-beta.3"
+actix-service = "2.0.0-beta.4"
 actix-utils = "3.0.0-beta.1"
 
 futures-core = { version = "0.3.7", default-features = false, features = ["alloc"] }
@@ -36,7 +36,7 @@ slab = "0.4"
 tokio = { version = "1", features = ["sync"] }
 
 [dev-dependencies]
-actix-rt = "2.0.0-beta.2"
+actix-rt = "2.0.0"
 bytes = "1"
 env_logger = "0.8"
 futures-util = { version = "0.3.7", default-features = false, features = ["sink"] }

actix-server/src/accept.rs

@@ -161,9 +161,16 @@ impl Accept {
         let mut events = mio::Events::with_capacity(128);
 
         loop {
-            self.poll
+            if let Err(e) = self.poll.poll(&mut events, None) {
-                .poll(&mut events, None)
+                match e.kind() {
-                .unwrap_or_else(|e| panic!("Poll error: {}", e));
+                    std::io::ErrorKind::Interrupted => {
+                        continue;
+                    }
+                    _ => {
+                        panic!("Poll error: {}", e);
+                    }
+                }
+            }
 
             for event in events.iter() {
                 let token = event.token();

actix-server/src/builder.rs

@@ -19,7 +19,7 @@ use crate::signals::{Signal, Signals};
 use crate::socket::{MioListener, StdSocketAddr, StdTcpListener, ToSocketAddrs};
 use crate::socket::{MioTcpListener, MioTcpSocket};
 use crate::waker_queue::{WakerInterest, WakerQueue};
-use crate::worker::{self, ServerWorker, WorkerAvailability, WorkerHandle};
+use crate::worker::{self, ServerWorker, ServerWorkerConfig, WorkerAvailability, WorkerHandle};
 use crate::{join_all, Token};
 
 /// Server builder
@@ -32,11 +32,11 @@ pub struct ServerBuilder {
     sockets: Vec<(Token, String, MioListener)>,
     accept: AcceptLoop,
     exit: bool,
-    shutdown_timeout: Duration,
     no_signals: bool,
     cmd: UnboundedReceiver<ServerCommand>,
     server: Server,
     notify: Vec<oneshot::Sender<()>>,
+    worker_config: ServerWorkerConfig,
 }
 
 impl Default for ServerBuilder {
@@ -60,11 +60,11 @@ impl ServerBuilder {
             accept: AcceptLoop::new(server.clone()),
             backlog: 2048,
             exit: false,
-            shutdown_timeout: Duration::from_secs(30),
             no_signals: false,
             cmd: rx,
             notify: Vec::new(),
             server,
+            worker_config: ServerWorkerConfig::default(),
         }
     }
 
@@ -78,6 +78,24 @@ impl ServerBuilder {
         self
     }
 
+    /// Set max number of threads for each worker's blocking task thread pool.
+    ///
+    /// One thread pool is set up **per worker**; not shared across workers.
+    ///
+    /// # Examples:
+    /// ```
+    /// # use actix_server::ServerBuilder;
+    /// let builder = ServerBuilder::new()
+    ///     .workers(4) // server has 4 worker thread.
+    ///     .worker_max_blocking_threads(4); // every worker has 4 max blocking threads.
+    /// ```
+    ///
+    /// See [tokio::runtime::Builder::max_blocking_threads] for behavior reference.
+    pub fn worker_max_blocking_threads(mut self, num: usize) -> Self {
+        self.worker_config.max_blocking_threads(num);
+        self
+    }
+
     /// Set the maximum number of pending connections.
     ///
     /// This refers to the number of clients that can be waiting to be served.
@@ -124,7 +142,8 @@ impl ServerBuilder {
     ///
     /// By default shutdown timeout sets to 30 seconds.
     pub fn shutdown_timeout(mut self, sec: u64) -> Self {
-        self.shutdown_timeout = Duration::from_secs(sec);
+        self.worker_config
+            .shutdown_timeout(Duration::from_secs(sec));
         self
     }
 
@@ -297,7 +316,7 @@ impl ServerBuilder {
         let avail = WorkerAvailability::new(waker);
         let services = self.services.iter().map(|v| v.clone_factory()).collect();
 
-        ServerWorker::start(idx, services, avail, self.shutdown_timeout)
+        ServerWorker::start(idx, services, avail, self.worker_config)
     }
 
     fn handle_cmd(&mut self, item: ServerCommand) {

actix-server/src/worker.rs

@@ -133,7 +133,7 @@ pub(crate) struct ServerWorker {
     conns: Counter,
     factories: Vec<Box<dyn InternalServiceFactory>>,
     state: WorkerState,
-    shutdown_timeout: Duration,
+    config: ServerWorkerConfig,
 }
 
 struct WorkerService {
@@ -159,26 +159,62 @@ enum WorkerServiceStatus {
     Stopped,
 }
 
+/// Config for worker behavior passed down from server builder.
+#[derive(Copy, Clone)]
+pub(crate) struct ServerWorkerConfig {
+    shutdown_timeout: Duration,
+    max_blocking_threads: usize,
+}
+
+impl Default for ServerWorkerConfig {
+    fn default() -> Self {
+        // 512 is the default max blocking thread count of tokio runtime.
+        let max_blocking_threads = std::cmp::max(512 / num_cpus::get(), 1);
+        Self {
+            shutdown_timeout: Duration::from_secs(30),
+            max_blocking_threads,
+        }
+    }
+}
+
+impl ServerWorkerConfig {
+    pub(crate) fn max_blocking_threads(&mut self, num: usize) {
+        self.max_blocking_threads = num;
+    }
+
+    pub(crate) fn shutdown_timeout(&mut self, dur: Duration) {
+        self.shutdown_timeout = dur;
+    }
+}
+
 impl ServerWorker {
     pub(crate) fn start(
         idx: usize,
         factories: Vec<Box<dyn InternalServiceFactory>>,
         availability: WorkerAvailability,
-        shutdown_timeout: Duration,
+        config: ServerWorkerConfig,
     ) -> WorkerHandle {
         let (tx1, rx) = unbounded_channel();
         let (tx2, rx2) = unbounded_channel();
         let avail = availability.clone();
 
         // every worker runs in it's own arbiter.
-        Arbiter::new().spawn(Box::pin(async move {
+        // use a custom tokio runtime builder to change the settings of runtime.
+        Arbiter::with_tokio_rt(move || {
+            tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .max_blocking_threads(config.max_blocking_threads)
+                .build()
+                .unwrap()
+        })
+        .spawn(async move {
             availability.set(false);
             let mut wrk = MAX_CONNS_COUNTER.with(move |conns| ServerWorker {
                 rx,
                 rx2,
                 availability,
                 factories,
-                shutdown_timeout,
+                config,
                 services: Vec::new(),
                 conns: conns.clone(),
                 state: WorkerState::Unavailable,
@@ -198,6 +234,8 @@ impl ServerWorker {
                 })
                 .collect::<Vec<_>>();
 
+            // a second spawn to make sure worker future runs as non boxed future.
+            // As Arbiter::spawn would box the future before send it to arbiter.
             spawn(async move {
                 let res: Result<Vec<_>, _> = join_all(fut).await.into_iter().collect();
                 match res {
@@ -220,7 +258,7 @@ impl ServerWorker {
                 }
                 wrk.await
             });
-        }));
+        });
 
         WorkerHandle::new(idx, tx1, tx2, avail)
     }
@@ -324,7 +362,7 @@ impl Future for ServerWorker {
                     info!("Graceful worker shutdown, {} connections", num);
                     self.state = WorkerState::Shutdown(
                         Box::pin(sleep_until(Instant::now() + Duration::from_secs(1))),
-                        Box::pin(sleep_until(Instant::now() + self.shutdown_timeout)),
+                        Box::pin(sleep_until(Instant::now() + self.config.shutdown_timeout)),
                         Some(result),
                     );
                 } else {

actix-service/CHANGES.md

@@ -1,10 +1,13 @@
 # Changes
 
 ## Unreleased - 2021-xx-xx
-* `Service::poll_ready` and `Service::call` take `&self`. [#247]
+
-* `apply_fn` and `apply_fn_factory` would take `Fn(Req, &Service)` function type [#247]
+
-* `apply_cfg` and `apply_cfg_factory` would take `Fn(Req, &Service)` function type [#247]
+## 2.0.0-beta.4 - 2021-02-04
-* `fn_service` module would take `Fn(Req)` function type. [#247]
+* `Service::poll_ready` and `Service::call` receive `&self`. [#247]
+* `apply_fn` and `apply_fn_factory` now receive `Fn(Req, &Service)` function type. [#247]
+* `apply_cfg` and `apply_cfg_factory` now receive `Fn(Req, &Service)` function type. [#247]
+* `fn_service` and friends now receive `Fn(Req)` function type. [#247]
 
 [#247]: https://github.com/actix/actix-net/pull/247
 

actix-service/Cargo.toml

@@ -1,9 +1,10 @@
 [package]
 name = "actix-service"
-version = "2.0.0-beta.3"
+version = "2.0.0-beta.4"
 authors = [
     "Nikolay Kim <fafhrd91@gmail.com>",
     "Rob Ede <robjtede@icloud.com>",
+    "fakeshadow <24548779@qq.com>",
 ]
 description = "Service trait and combinators for representing asynchronous request/response operations."
 keywords = ["network", "framework", "async", "futures", "service"]

actix-service/README.md

@@ -2,6 +2,12 @@
 
 > Service trait and combinators for representing asynchronous request/response operations.
 
-See documentation for detailed explanations these components: [https://docs.rs/actix-service][docs].
+[![crates.io](https://img.shields.io/crates/v/actix-service?label=latest)](https://crates.io/crates/actix-service)
+[![Documentation](https://docs.rs/actix-service/badge.svg?version=2.0.0-beta.4)](https://docs.rs/actix-service/2.0.0-beta.4)
+[![Version](https://img.shields.io/badge/rustc-1.46+-ab6000.svg)](https://blog.rust-lang.org/2020/03/12/Rust-1.46.html)
+![License](https://img.shields.io/crates/l/actix-service.svg)
+[![Dependency Status](https://deps.rs/crate/actix-service/2.0.0-beta.4/status.svg)](https://deps.rs/crate/actix-service/2.0.0-beta.4)
+[![Download](https://img.shields.io/crates/d/actix-service.svg)](https://crates.io/crates/actix-service)
+[![Chat on Discord](https://img.shields.io/discord/771444961383153695?label=chat&logo=discord)](https://discord.gg/NWpN5mmg3x)
 
-[docs]: https://docs.rs/actix-service
+See documentation for detailed explanations of these components: https://docs.rs/actix-service.

actix-tls/Cargo.toml

@@ -31,10 +31,10 @@ connect = []
 openssl = ["tls-openssl", "tokio-openssl"]
 
 # use rustls impls
-rustls = ["tls-rustls", "webpki", "webpki-roots", "tokio-rustls"]
+rustls = ["tokio-rustls", "webpki-roots"]
 
 # use native-tls impls
-native-tls = ["tls-native-tls", "tokio-native-tls"]
+native-tls = ["tokio-native-tls"]
 
 # support http::Uri as connect address
 uri = ["http"]
@@ -42,28 +42,24 @@ uri = ["http"]
 [dependencies]
 actix-codec = "0.4.0-beta.1"
 actix-rt = { version = "2.0.0", default-features = false }
-actix-service = "2.0.0-beta.3"
+actix-service = "2.0.0-beta.4"
 actix-utils = "3.0.0-beta.1"
 
 derive_more = "0.99.5"
 futures-core = { version = "0.3.7", default-features = false, features = ["alloc"] }
 http = { version = "0.2.3", optional = true }
 log = "0.4"
+tokio-util = { version = "0.6.3", default-features = false }
 
 # openssl
 tls-openssl = { package = "openssl", version = "0.10", optional = true }
 tokio-openssl = { version = "0.6", optional = true }
 
-# TODO: Reduce dependencies where tokio wrappers re-export base crate.
-
 # rustls
-tls-rustls = { package = "rustls", version = "0.19", optional = true }
 tokio-rustls = { version = "0.22", optional = true }
-webpki = { version = "0.21", optional = true }
 webpki-roots = { version = "0.21", optional = true }
 
 # native-tls
-tls-native-tls = { package = "native-tls", version = "0.2", optional = true }
 tokio-native-tls = { version = "0.3", optional = true }
 
 [dev-dependencies]

actix-tls/examples/basic.rs

@@ -15,9 +15,9 @@
 //! http --verify=false https://127.0.0.1:8443
 //! ```
 
-// this rename only exists because of how we have organised the crate's feature flags
+// this use only exists because of how we have organised the crate
 // it is not necessary for your actual code
-extern crate tls_rustls as rustls;
+use tokio_rustls::rustls;
 
 use std::{
     env,

actix-tls/src/accept/nativetls.rs

@@ -5,7 +5,7 @@ use actix_service::{Service, ServiceFactory};
 use actix_utils::counter::Counter;
 use futures_core::future::LocalBoxFuture;
 
-pub use native_tls::Error;
+pub use tokio_native_tls::native_tls::Error;
 pub use tokio_native_tls::{TlsAcceptor, TlsStream};
 
 use super::MAX_CONN_COUNTER;

actix-tls/src/accept/rustls.rs

@@ -12,7 +12,7 @@ use actix_utils::counter::{Counter, CounterGuard};
 use futures_core::future::LocalBoxFuture;
 use tokio_rustls::{Accept, TlsAcceptor};
 
-pub use rustls::{ServerConfig, Session};
+pub use tokio_rustls::rustls::{ServerConfig, Session};
 pub use tokio_rustls::server::TlsStream;
 
 use super::MAX_CONN_COUNTER;

actix-tls/src/connect/connector.rs

@@ -11,6 +11,7 @@ use actix_rt::net::TcpStream;
 use actix_service::{Service, ServiceFactory};
 use futures_core::{future::LocalBoxFuture, ready};
 use log::{error, trace};
+use tokio_util::sync::ReusableBoxFuture;
 
 use super::connect::{Address, Connect, ConnectAddrs, Connection};
 use super::error::ConnectError;
@@ -65,7 +66,7 @@ pub enum TcpConnectorResponse<T> {
         req: Option<T>,
         port: u16,
         addrs: Option<VecDeque<SocketAddr>>,
-        stream: Option<LocalBoxFuture<'static, Result<TcpStream, io::Error>>>,
+        stream: Option<ReusableBoxFuture<Result<TcpStream, io::Error>>>,
     },
     Error(Option<ConnectError>),
 }
@@ -90,7 +91,7 @@ impl<T: Address> TcpConnectorResponse<T> {
                 req: Some(req),
                 port,
                 addrs: None,
-                stream: Some(Box::pin(TcpStream::connect(addr))),
+                stream: Some(ReusableBoxFuture::new(TcpStream::connect(addr))),
             },
 
             // when resolver returns multiple socket addr for request they would be popped from
@@ -119,7 +120,7 @@ impl<T: Address> Future for TcpConnectorResponse<T> {
                 stream,
             } => loop {
                 if let Some(new) = stream.as_mut() {
-                    match ready!(new.as_mut().poll(cx)) {
+                    match ready!(new.poll(cx)) {
                         Ok(sock) => {
                             let req = req.take().unwrap();
                             trace!(
@@ -146,7 +147,11 @@ impl<T: Address> Future for TcpConnectorResponse<T> {
 
                 // try to connect
                 let addr = addrs.as_mut().unwrap().pop_front().unwrap();
-                *stream = Some(Box::pin(TcpStream::connect(addr)));
+
+                match stream {
+                    Some(rbf) => rbf.set(TcpStream::connect(addr)),
+                    None => *stream = Some(ReusableBoxFuture::new(TcpStream::connect(addr))),
+                }
             },
         }
     }

actix-tls/src/connect/ssl/rustls.rs

@@ -6,7 +6,7 @@ use std::{
     task::{Context, Poll},
 };
 
-pub use rustls::Session;
+pub use tokio_rustls::rustls::Session;
 pub use tokio_rustls::{client::TlsStream, rustls::ClientConfig};
 pub use webpki_roots::TLS_SERVER_ROOTS;
 
@@ -14,8 +14,8 @@ use actix_codec::{AsyncRead, AsyncWrite};
 use actix_service::{Service, ServiceFactory};
 use futures_core::{future::LocalBoxFuture, ready};
 use log::trace;
+use tokio_rustls::webpki::DNSNameRef;
 use tokio_rustls::{Connect, TlsConnector};
-use webpki::DNSNameRef;
 
 use crate::connect::{Address, Connection};
 

actix-tls/src/lib.rs

@@ -4,12 +4,8 @@
 #![doc(html_logo_url = "https://actix.rs/img/logo.png")]
 #![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
 
-#[cfg(feature = "native-tls")]
-extern crate tls_native_tls as native_tls;
 #[cfg(feature = "openssl")]
 extern crate tls_openssl as openssl;
-#[cfg(feature = "rustls")]
-extern crate tls_rustls as rustls;
 
 #[cfg(feature = "accept")]
 pub mod accept;

actix-tracing/Cargo.toml

@@ -16,7 +16,7 @@ name = "actix_tracing"
 path = "src/lib.rs"
 
 [dependencies]
-actix-service = "2.0.0-beta.3"
+actix-service = "2.0.0-beta.4"
 
 futures-util = { version = "0.3.4", default-features = false }
 tracing = "0.1"

actix-utils/Cargo.toml

@@ -18,7 +18,7 @@ path = "src/lib.rs"
 [dependencies]
 actix-codec = "0.4.0-beta.1"
 actix-rt = { version = "2.0.0", default-features = false }
-actix-service = "2.0.0-beta.3"
+actix-service = "2.0.0-beta.4"
 
 futures-core = { version = "0.3.7", default-features = false }
 futures-sink = { version = "0.3.7", default-features = false }