change actix-router version to 0.4.0

* Remove restart_worker test

* Remove Slab

* Rework counter

* Make counter limit switch accurate

* Remove backpressure. Add pause state

* make changes for review

* fix doc comment for counter

LICENSE-APACHE

@@ -186,7 +186,7 @@
       same "printed page" as the copyright notice for easier
       identification within third-party archives.
 
-   Copyright 2017-NOW Nikolay Kim
+   Copyright 2017-NOW Actix Team
 
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.

LICENSE-MIT

@@ -1,4 +1,4 @@
-Copyright (c) 2017 Nikolay Kim
+Copyright (c) 2017-NOW Actix Team
 
 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated

actix-codec/CHANGES.md

@@ -3,6 +3,10 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 0.4.0 - 2021-04-20
+* No significant changes since v0.4.0-beta.1.
+
+
 ## 0.4.0-beta.1 - 2020-12-28
 * Replace `pin-project` with `pin-project-lite`. [#237]
 * Upgrade `tokio` dependency to `1`. [#237]
@@ -23,28 +27,28 @@
 ## 0.3.0-beta.1 - 2020-08-19
 * Use `.advance()` instead of `.split_to()`.
 * Upgrade `tokio-util` to `0.3`.
-* Improve `BytesCodec` `.encode()` performance
-* Simplify `BytesCodec` `.decode()` 
+* Improve `BytesCodec::encode()` performance.
+* Simplify `BytesCodec::decode()`.
 * Rename methods on `Framed` to better describe their use.
 * Add method on `Framed` to get a pinned reference to the underlying I/O.
 * Add method on `Framed` check emptiness of read buffer.
 
 
 ## 0.2.0 - 2019-12-10
-* Use specific futures dependencies
+* Use specific futures dependencies.
 
 
 ## 0.2.0-alpha.4
-* Fix buffer remaining capacity calculation
+* Fix buffer remaining capacity calculation.
 
 
 ## 0.2.0-alpha.3
-* Use tokio 0.2
-* Fix low/high watermark for write/read buffers
+* Use tokio 0.2.
+* Fix low/high watermark for write/read buffers.
 
 
 ## 0.2.0-alpha.2
-* Migrated to `std::future`
+* Migrated to `std::future`.
 
 
 ## 0.1.2 - 2019-03-27
@@ -56,4 +60,4 @@
 
 
 ## 0.1.0 - 2018-12-09
-* Move codec to separate crate
+* Move codec to separate crate.

actix-codec/Cargo.toml

@@ -1,12 +1,10 @@
 [package]
 name = "actix-codec"
-version = "0.4.0-beta.1"
+version = "0.4.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Codec utilities for working with framed protocols"
 keywords = ["network", "framework", "async", "futures"]
-homepage = "https://actix.rs"
-repository = "https://github.com/actix/actix-net.git"
-documentation = "https://docs.rs/actix-codec"
+repository = "https://github.com/actix/actix-net"
 categories = ["network-programming", "asynchronous"]
 license = "MIT OR Apache-2.0"
 edition = "2018"

actix-codec/src/lib.rs

@@ -7,7 +7,7 @@
 //! [`Sink`]: futures_sink::Sink
 //! [`Stream`]: futures_core::Stream
 
-#![deny(rust_2018_idioms, nonstandard_style)]
+#![deny(rust_2018_idioms, nonstandard_style, future_incompatible)]
 #![warn(missing_docs)]
 #![doc(html_logo_url = "https://actix.rs/img/logo.png")]
 #![doc(html_favicon_url = "https://actix.rs/favicon.ico")]

actix-router/CHANGES.md

@@ -3,6 +3,21 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 0.4.0 - 2021-06-06
+* When matching path parameters, `%25` is now kept in the percent-encoded form; no longer decoded to `%`. [#357]
+* Path tail patterns now match new lines (`\n`) in request URL. [#360]
+* Fixed a safety bug where `Path` could return a malformed string after percent decoding. [#359]
+* Methods `Path::{add, add_static}` now take `impl Into<Cow<'static, str>>`. [#345]
+
+[#345]: https://github.com/actix/actix-net/pull/345
+[#357]: https://github.com/actix/actix-net/pull/357
+[#359]: https://github.com/actix/actix-net/pull/359
+[#360]: https://github.com/actix/actix-net/pull/360
+
+## 0.3.0 - 2019-12-31
+* Version was yanked previously. See https://crates.io/crates/actix-router/0.3.0
+
+
 ## 0.2.7 - 2021-02-06
 * Add `Router::recognize_checked` [#247]
 

actix-router/Cargo.toml

@@ -1,12 +1,10 @@
 [package]
 name = "actix-router"
-version = "0.2.7"
+version = "0.4.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Resource path matching library"
 keywords = ["actix", "router", "routing"]
-homepage = "https://actix.rs"
-repository = "https://github.com/actix/actix-net.git"
-documentation = "https://docs.rs/actix-router"
+repository = "https://github.com/actix/actix-net"
 license = "MIT OR Apache-2.0"
 edition = "2018"
 

actix-router/src/path.rs

@@ -1,3 +1,4 @@
+use std::borrow::Cow;
 use std::ops::Index;
 
 use serde::de;
@@ -5,12 +6,18 @@ use serde::de;
 use crate::de::PathDeserializer;
 use crate::{Resource, ResourcePath};
 
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone)]
 pub(crate) enum PathItem {
-    Static(&'static str),
+    Static(Cow<'static, str>),
     Segment(u16, u16),
 }
 
+impl Default for PathItem {
+    fn default() -> Self {
+        Self::Static(Cow::Borrowed(""))
+    }
+}
+
 /// Resource path match information
 ///
 /// If resource path contains variable patterns, `Path` stores them.
@@ -18,7 +25,7 @@ pub(crate) enum PathItem {
 pub struct Path<T> {
     path: T,
     pub(crate) skip: u16,
-    pub(crate) segments: Vec<(&'static str, PathItem)>,
+    pub(crate) segments: Vec<(Cow<'static, str>, PathItem)>,
 }
 
 impl<T: Default> Default for Path<T> {
@@ -95,18 +102,24 @@ impl<T: ResourcePath> Path<T> {
         self.skip += n;
     }
 
-    pub(crate) fn add(&mut self, name: &'static str, value: PathItem) {
+    pub(crate) fn add(&mut self, name: impl Into<Cow<'static, str>>, value: PathItem) {
         match value {
-            PathItem::Static(s) => self.segments.push((name, PathItem::Static(s))),
-            PathItem::Segment(begin, end) => self
-                .segments
-                .push((name, PathItem::Segment(self.skip + begin, self.skip + end))),
+            PathItem::Static(s) => self.segments.push((name.into(), PathItem::Static(s))),
+            PathItem::Segment(begin, end) => self.segments.push((
+                name.into(),
+                PathItem::Segment(self.skip + begin, self.skip + end),
+            )),
         }
     }
 
     #[doc(hidden)]
-    pub fn add_static(&mut self, name: &'static str, value: &'static str) {
-        self.segments.push((name, PathItem::Static(value)));
+    pub fn add_static(
+        &mut self,
+        name: impl Into<Cow<'static, str>>,
+        value: impl Into<Cow<'static, str>>,
+    ) {
+        self.segments
+            .push((name.into(), PathItem::Static(value.into())));
     }
 
     /// Check if there are any matched patterns

actix-router/src/resource.rs

@@ -1,6 +1,7 @@
 use std::cmp::min;
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
+use std::mem;
 
 use regex::{escape, Regex, RegexSet};
 
@@ -9,6 +10,11 @@ use crate::{IntoPattern, Resource, ResourcePath};
 
 const MAX_DYNAMIC_SEGMENTS: usize = 16;
 
+/// Regex flags to allow '.' in regex to match '\n'
+///
+/// See the docs under: https://docs.rs/regex/1.5.4/regex/#grouping-and-flags
+const REGEX_FLAGS: &str = "(?s-m)";
+
 /// ResourceDef describes an entry in resources table
 ///
 /// Resource definition can contain only 16 dynamic segments
@@ -272,15 +278,12 @@ impl ResourceDef {
                 true
             }
             PatternType::Dynamic(ref re, ref names, len) => {
-                let mut idx = 0;
                 let mut pos = 0;
-                let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] =
-                    [PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
+                let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] = Default::default();
 
                 if let Some(captures) = re.captures(path.path()) {
                     for (no, name) in names.iter().enumerate() {
                         if let Some(m) = captures.name(&name) {
-                            idx += 1;
                             pos = m.end();
                             segments[no] = PathItem::Segment(m.start() as u16, m.end() as u16);
                         } else {
@@ -294,8 +297,8 @@ impl ResourceDef {
                 } else {
                     return false;
                 }
-                for idx in 0..idx {
-                    path.add(names[idx], segments[idx]);
+                for i in 0..names.len() {
+                    path.add(names[i], mem::take(&mut segments[i]));
                 }
                 path.skip((pos + len) as u16);
                 true
@@ -303,15 +306,12 @@ impl ResourceDef {
             PatternType::DynamicSet(ref re, ref params) => {
                 if let Some(idx) = re.matches(path.path()).into_iter().next() {
                     let (ref pattern, ref names, len) = params[idx];
-                    let mut idx = 0;
                     let mut pos = 0;
-                    let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] =
-                        [PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
+                    let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] = Default::default();
 
                     if let Some(captures) = pattern.captures(path.path()) {
                         for (no, name) in names.iter().enumerate() {
                             if let Some(m) = captures.name(&name) {
-                                idx += 1;
                                 pos = m.end();
                                 segments[no] =
                                     PathItem::Segment(m.start() as u16, m.end() as u16);
@@ -326,8 +326,8 @@ impl ResourceDef {
                     } else {
                         return false;
                     }
-                    for idx in 0..idx {
-                        path.add(names[idx], segments[idx]);
+                    for i in 0..names.len() {
+                        path.add(names[i], mem::take(&mut segments[i]));
                     }
                     path.skip((pos + len) as u16);
                     true
@@ -385,15 +385,12 @@ impl ResourceDef {
                 true
             }
             PatternType::Dynamic(ref re, ref names, len) => {
-                let mut idx = 0;
                 let mut pos = 0;
-                let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] =
-                    [PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
+                let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] = Default::default();
 
                 if let Some(captures) = re.captures(res.resource_path().path()) {
                     for (no, name) in names.iter().enumerate() {
                         if let Some(m) = captures.name(&name) {
-                            idx += 1;
                             pos = m.end();
                             segments[no] = PathItem::Segment(m.start() as u16, m.end() as u16);
                         } else {
@@ -413,8 +410,8 @@ impl ResourceDef {
                 }
 
                 let path = res.resource_path();
-                for idx in 0..idx {
-                    path.add(names[idx], segments[idx]);
+                for i in 0..names.len() {
+                    path.add(names[i], mem::take(&mut segments[i]));
                 }
                 path.skip((pos + len) as u16);
                 true
@@ -423,15 +420,12 @@ impl ResourceDef {
                 let path = res.resource_path().path();
                 if let Some(idx) = re.matches(path).into_iter().next() {
                     let (ref pattern, ref names, len) = params[idx];
-                    let mut idx = 0;
                     let mut pos = 0;
-                    let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] =
-                        [PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
+                    let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] = Default::default();
 
                     if let Some(captures) = pattern.captures(path) {
                         for (no, name) in names.iter().enumerate() {
                             if let Some(m) = captures.name(&name) {
-                                idx += 1;
                                 pos = m.end();
                                 segments[no] =
                                     PathItem::Segment(m.start() as u16, m.end() as u16);
@@ -452,8 +446,8 @@ impl ResourceDef {
                     }
 
                     let path = res.resource_path();
-                    for idx in 0..idx {
-                        path.add(names[idx], segments[idx]);
+                    for i in 0..names.len() {
+                        path.add(names[i], mem::take(&mut segments[i]));
                     }
                     path.skip((pos + len) as u16);
                     true
@@ -581,11 +575,8 @@ impl ResourceDef {
         mut for_prefix: bool,
     ) -> (String, Vec<PatternElement>, bool, usize) {
         if pattern.find('{').is_none() {
-            // TODO: MSRV: 1.45
-            #[allow(clippy::manual_strip)]
-            return if pattern.ends_with('*') {
-                let path = &pattern[..pattern.len() - 1];
-                let re = String::from("^") + path + "(.*)";
+            return if let Some(path) = pattern.strip_suffix('*') {
+                let re = format!("{}^{}(.*)", REGEX_FLAGS, path);
                 (re, vec![PatternElement::Str(String::from(path))], true, 0)
             } else {
                 (
@@ -598,7 +589,7 @@ impl ResourceDef {
         }
 
         let mut elements = Vec::new();
-        let mut re = String::from("^");
+        let mut re = format!("{}^", REGEX_FLAGS);
         let mut dyn_elements = 0;
 
         while let Some(idx) = pattern.find('{') {
@@ -831,6 +822,32 @@ mod tests {
         assert!(re.is_match("/user/2345/sdg"));
     }
 
+    #[test]
+    fn test_newline() {
+        let re = ResourceDef::new("/user/a\nb");
+        assert!(re.is_match("/user/a\nb"));
+        assert!(!re.is_match("/user/a\nb/profile"));
+
+        let re = ResourceDef::new("/a{x}b/test/a{y}b");
+        let mut path = Path::new("/a\nb/test/a\nb");
+        assert!(re.match_path(&mut path));
+        assert_eq!(path.get("x").unwrap(), "\n");
+        assert_eq!(path.get("y").unwrap(), "\n");
+
+        let re = ResourceDef::new("/user/*");
+        assert!(re.is_match("/user/a\nb/"));
+
+        let re = ResourceDef::new("/user/{id}*");
+        let mut path = Path::new("/user/a\nb/a\nb");
+        assert!(re.match_path(&mut path));
+        assert_eq!(path.get("id").unwrap(), "a\nb/a\nb");
+
+        let re = ResourceDef::new("/user/{id:.*}");
+        let mut path = Path::new("/user/a\nb/a\nb");
+        assert!(re.match_path(&mut path));
+        assert_eq!(path.get("id").unwrap(), "a\nb/a\nb");
+    }
+
     #[cfg(feature = "http")]
     #[test]
     fn test_parse_urlencoded_param() {

actix-router/src/url.rs

@@ -31,7 +31,7 @@ fn set_bit(array: &mut [u8], ch: u8) {
 }
 
 thread_local! {
-    static DEFAULT_QUOTER: Quoter = Quoter::new(b"@:", b"/+");
+    static DEFAULT_QUOTER: Quoter = Quoter::new(b"@:", b"%/+");
 }
 
 #[derive(Default, Clone, Debug)]
@@ -170,11 +170,7 @@ impl Quoter {
             idx += 1;
         }
 
-        cloned.map(|data| {
-            // SAFETY: we get data from http::Uri, which does UTF-8 checks already
-            // this code only decodes valid pct encoded values
-            unsafe { String::from_utf8_unchecked(data) }
-        })
+        cloned.map(|data| String::from_utf8_lossy(&data).into_owned())
     }
 }
 
@@ -204,24 +200,69 @@ mod tests {
     use super::*;
     use crate::{Path, ResourceDef};
 
+    const PROTECTED: &[u8] = b"%/+";
+
+    fn match_url(pattern: &'static str, url: impl AsRef<str>) -> Path<Url> {
+        let re = ResourceDef::new(pattern);
+        let uri = Uri::try_from(url.as_ref()).unwrap();
+        let mut path = Path::new(Url::new(uri));
+        assert!(re.match_path(&mut path));
+        path
+    }
+
+    fn percent_encode(data: &[u8]) -> String {
+        data.into_iter().map(|c| format!("%{:02X}", c)).collect()
+    }
+
     #[test]
     fn test_parse_url() {
-        let re = ResourceDef::new("/user/{id}/test");
+        let re = "/user/{id}/test";
 
-        let url = Uri::try_from("/user/2345/test").unwrap();
-        let mut path = Path::new(Url::new(url));
-        assert!(re.match_path(&mut path));
+        let path = match_url(re, "/user/2345/test");
         assert_eq!(path.get("id").unwrap(), "2345");
 
-        let url = Uri::try_from("/user/qwe%25/test").unwrap();
-        let mut path = Path::new(Url::new(url));
-        assert!(re.match_path(&mut path));
-        assert_eq!(path.get("id").unwrap(), "qwe%");
+        // "%25" should never be decoded into '%' to gurantee the output is a valid
+        // percent-encoded format
+        let path = match_url(re, "/user/qwe%25/test");
+        assert_eq!(path.get("id").unwrap(), "qwe%25");
 
-        let url = Uri::try_from("/user/qwe%25rty/test").unwrap();
-        let mut path = Path::new(Url::new(url));
-        assert!(re.match_path(&mut path));
-        assert_eq!(path.get("id").unwrap(), "qwe%rty");
+        let path = match_url(re, "/user/qwe%25rty/test");
+        assert_eq!(path.get("id").unwrap(), "qwe%25rty");
+    }
+
+    #[test]
+    fn test_protected_chars() {
+        let encoded = percent_encode(PROTECTED);
+        let path = match_url("/user/{id}/test", format!("/user/{}/test", encoded));
+        assert_eq!(path.get("id").unwrap(), &encoded);
+    }
+
+    #[test]
+    fn test_non_protecteed_ascii() {
+        let nonprotected_ascii = ('\u{0}'..='\u{7F}')
+            .filter(|&c| c.is_ascii() && !PROTECTED.contains(&(c as u8)))
+            .collect::<String>();
+        let encoded = percent_encode(nonprotected_ascii.as_bytes());
+        let path = match_url("/user/{id}/test", format!("/user/{}/test", encoded));
+        assert_eq!(path.get("id").unwrap(), &nonprotected_ascii);
+    }
+
+    #[test]
+    fn test_valid_utf8_multibyte() {
+        let test = ('\u{FF00}'..='\u{FFFF}').collect::<String>();
+        let encoded = percent_encode(test.as_bytes());
+        let path = match_url("/a/{id}/b", format!("/a/{}/b", &encoded));
+        assert_eq!(path.get("id").unwrap(), &test);
+    }
+
+    #[test]
+    fn test_invalid_utf8() {
+        let invalid_utf8 = percent_encode((0x80..=0xff).collect::<Vec<_>>().as_slice());
+        let uri = Uri::try_from(format!("/{}", invalid_utf8)).unwrap();
+        let path = Path::new(Url::new(uri));
+
+        // We should always get a valid utf8 string
+        assert!(String::from_utf8(path.path().as_bytes().to_owned()).is_ok());
     }
 
     #[test]

actix-server/CHANGES.md

@@ -1,6 +1,17 @@
 # Changes
 
 ## Unreleased - 2021-xx-xx
+* Remove `config` module. `ServiceConfig`, `ServiceRuntime` public types are removed due to this change. [#349]
+* Remove `ServerBuilder::configure` [#349]
+
+[#349]: https://github.com/actix/actix-net/pull/349
+
+
+## 2.0.0-beta.5 - 2021-04-20
+* Server shutdown would notify all workers to exit regardless if shutdown is graceful.
+  This would make all worker shutdown immediately in force shutdown case. [#333]
+  
+[#333]: https://github.com/actix/actix-net/pull/333
 
 
 ## 2.0.0-beta.4 - 2021-04-01

actix-server/Cargo.toml

@@ -1,14 +1,13 @@
 [package]
 name = "actix-server"
-version = "2.0.0-beta.4"
+version = "2.0.0-beta.5"
 authors = [
     "Nikolay Kim <fafhrd91@gmail.com>",
     "fakeshadow <24548779@qq.com>",
 ]
 description = "General purpose TCP server built for the Actix ecosystem"
 keywords = ["network", "framework", "async", "futures"]
-homepage = "https://actix.rs"
-repository = "https://github.com/actix/actix-net.git"
+repository = "https://github.com/actix/actix-net"
 categories = ["network-programming", "asynchronous"]
 license = "MIT OR Apache-2.0"
 edition = "2018"
@@ -22,14 +21,13 @@ default = []
 
 [dependencies]
 actix-rt = { version = "2.0.0", default-features = false }
-actix-service = "2.0.0-beta.5"
-actix-utils = "3.0.0-beta.2"
+actix-service = "2.0.0"
+actix-utils = "3.0.0"
 
 futures-core = { version = "0.3.7", default-features = false, features = ["alloc"] }
 log = "0.4"
 mio = { version = "0.7.6", features = ["os-poll", "net"] }
 num_cpus = "1.13"
-slab = "0.4"
 tokio = { version = "1.2", features = ["sync"] }
 
 [dev-dependencies]

actix-server/examples/tcp-echo.rs

@@ -9,15 +9,17 @@
 //! Start typing. When you press enter the typed line will be echoed back. The server will log
 //! the length of each line it echos and the total size of data sent when the connection is closed.
 
-use std::sync::{
-    atomic::{AtomicUsize, Ordering},
-    Arc,
+use std::{
+    env, io,
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    },
 };
-use std::{env, io};
 
 use actix_rt::net::TcpStream;
 use actix_server::Server;
-use actix_service::pipeline_factory;
+use actix_service::{fn_service, ServiceFactoryExt as _};
 use bytes::BytesMut;
 use futures_util::future::ok;
 use log::{error, info};
@@ -25,7 +27,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 #[actix_rt::main]
 async fn main() -> io::Result<()> {
-    env::set_var("RUST_LOG", "actix=trace,basic=trace");
+    env::set_var("RUST_LOG", "info");
     env_logger::init();
 
     let count = Arc::new(AtomicUsize::new(0));
@@ -41,7 +43,7 @@ async fn main() -> io::Result<()> {
             let count = Arc::clone(&count);
             let num2 = Arc::clone(&count);
 
-            pipeline_factory(move |mut stream: TcpStream| {
+            fn_service(move |mut stream: TcpStream| {
                 let count = Arc::clone(&count);
 
                 async move {

actix-server/src/accept.rs

@@ -7,21 +7,14 @@ use actix_rt::{
 };
 use log::{error, info};
 use mio::{Interest, Poll, Token as MioToken};
-use slab::Slab;
 
 use crate::server::Server;
-use crate::socket::{MioListener, SocketAddr};
+use crate::socket::MioListener;
 use crate::waker_queue::{WakerInterest, WakerQueue, WAKER_TOKEN};
-use crate::worker::{Conn, WorkerHandle};
-use crate::Token;
+use crate::worker::{Conn, WorkerHandleAccept};
 
 struct ServerSocketInfo {
-    /// Address of socket. Mainly used for logging.
-    addr: SocketAddr,
-
-    /// Beware this is the crate token for identify socket and should not be confused
-    /// with `mio::Token`.
-    token: Token,
+    token: usize,
 
     lst: MioListener,
 
@@ -65,8 +58,8 @@ impl AcceptLoop {
 
     pub(crate) fn start(
         &mut self,
-        socks: Vec<(Token, MioListener)>,
-        handles: Vec<WorkerHandle>,
+        socks: Vec<(usize, MioListener)>,
+        handles: Vec<WorkerHandleAccept>,
     ) {
         let srv = self.srv.take().expect("Can not re-use AcceptInfo");
         let poll = self.poll.take().unwrap();
@@ -80,10 +73,71 @@ impl AcceptLoop {
 struct Accept {
     poll: Poll,
     waker: WakerQueue,
-    handles: Vec<WorkerHandle>,
+    handles: Vec<WorkerHandleAccept>,
     srv: Server,
     next: usize,
-    backpressure: bool,
+    avail: Availability,
+    paused: bool,
+}
+
+/// Array of u128 with every bit as marker for a worker handle's availability.
+struct Availability([u128; 4]);
+
+impl Default for Availability {
+    fn default() -> Self {
+        Self([0; 4])
+    }
+}
+
+impl Availability {
+    /// Check if any worker handle is available
+    #[inline(always)]
+    fn available(&self) -> bool {
+        self.0.iter().any(|a| *a != 0)
+    }
+
+    /// Check if worker handle is available by index
+    #[inline(always)]
+    fn get_available(&self, idx: usize) -> bool {
+        let (offset, idx) = Self::offset(idx);
+
+        self.0[offset] & (1 << idx as u128) != 0
+    }
+
+    /// Set worker handle available state by index.
+    fn set_available(&mut self, idx: usize, avail: bool) {
+        let (offset, idx) = Self::offset(idx);
+
+        let off = 1 << idx as u128;
+        if avail {
+            self.0[offset] |= off;
+        } else {
+            self.0[offset] &= !off
+        }
+    }
+
+    /// Set all worker handle to available state.
+    /// This would result in a re-check on all workers' availability.
+    fn set_available_all(&mut self, handles: &[WorkerHandleAccept]) {
+        handles.iter().for_each(|handle| {
+            self.set_available(handle.idx(), true);
+        })
+    }
+
+    /// Get offset and adjusted index of given worker handle index.
+    fn offset(idx: usize) -> (usize, usize) {
+        if idx < 128 {
+            (0, idx)
+        } else if idx < 128 * 2 {
+            (1, idx - 128)
+        } else if idx < 128 * 3 {
+            (2, idx - 128 * 2)
+        } else if idx < 128 * 4 {
+            (3, idx - 128 * 3)
+        } else {
+            panic!("Max WorkerHandle count is 512")
+        }
+    }
 }
 
 /// This function defines errors that are per-connection. Which basically
@@ -103,9 +157,9 @@ impl Accept {
     pub(crate) fn start(
         poll: Poll,
         waker: WakerQueue,
-        socks: Vec<(Token, MioListener)>,
+        socks: Vec<(usize, MioListener)>,
         srv: Server,
-        handles: Vec<WorkerHandle>,
+        handles: Vec<WorkerHandleAccept>,
     ) {
         // Accept runs in its own thread and would want to spawn additional futures to current
         // actix system.
@@ -114,9 +168,10 @@ impl Accept {
             .name("actix-server accept loop".to_owned())
             .spawn(move || {
                 System::set_current(sys);
-                let (mut accept, sockets) =
+                let (mut accept, mut sockets) =
                     Accept::new_with_sockets(poll, waker, socks, handles, srv);
-                accept.poll_with(sockets);
+
+                accept.poll_with(&mut sockets);
             })
             .unwrap();
     }
@@ -124,29 +179,30 @@ impl Accept {
     fn new_with_sockets(
         poll: Poll,
         waker: WakerQueue,
-        socks: Vec<(Token, MioListener)>,
-        handles: Vec<WorkerHandle>,
+        socks: Vec<(usize, MioListener)>,
+        handles: Vec<WorkerHandleAccept>,
         srv: Server,
-    ) -> (Accept, Slab<ServerSocketInfo>) {
-        let mut sockets = Slab::new();
-        for (hnd_token, mut lst) in socks.into_iter() {
-            let addr = lst.local_addr();
+    ) -> (Accept, Vec<ServerSocketInfo>) {
+        let sockets = socks
+            .into_iter()
+            .map(|(token, mut lst)| {
+                // Start listening for incoming connections
+                poll.registry()
+                    .register(&mut lst, MioToken(token), Interest::READABLE)
+                    .unwrap_or_else(|e| panic!("Can not register io: {}", e));
 
-            let entry = sockets.vacant_entry();
-            let token = entry.key();
+                ServerSocketInfo {
+                    token,
+                    lst,
+                    timeout: None,
+                }
+            })
+            .collect();
 
-            // Start listening for incoming connections
-            poll.registry()
-                .register(&mut lst, MioToken(token), Interest::READABLE)
-                .unwrap_or_else(|e| panic!("Can not register io: {}", e));
+        let mut avail = Availability::default();
 
-            entry.insert(ServerSocketInfo {
-                addr,
-                token: hnd_token,
-                lst,
-                timeout: None,
-            });
-        }
+        // Assume all handles are avail at construct time.
+        avail.set_available_all(&handles);
 
         let accept = Accept {
             poll,
@@ -154,257 +210,264 @@ impl Accept {
             handles,
             srv,
             next: 0,
-            backpressure: false,
+            avail,
+            paused: false,
         };
 
         (accept, sockets)
     }
 
-    fn poll_with(&mut self, mut sockets: Slab<ServerSocketInfo>) {
+    fn poll_with(&mut self, sockets: &mut [ServerSocketInfo]) {
         let mut events = mio::Events::with_capacity(128);
 
         loop {
             if let Err(e) = self.poll.poll(&mut events, None) {
                 match e.kind() {
-                    std::io::ErrorKind::Interrupted => {
-                        continue;
-                    }
-                    _ => {
-                        panic!("Poll error: {}", e);
-                    }
+                    io::ErrorKind::Interrupted => {}
+                    _ => panic!("Poll error: {}", e),
                 }
             }
 
             for event in events.iter() {
                 let token = event.token();
                 match token {
-                    // This is a loop because interests for command from previous version was
-                    // a loop that would try to drain the command channel. It's yet unknown
-                    // if it's necessary/good practice to actively drain the waker queue.
-                    WAKER_TOKEN => 'waker: loop {
-                        // take guard with every iteration so no new interest can be added
-                        // until the current task is done.
-                        let mut guard = self.waker.guard();
-                        match guard.pop_front() {
-                            // worker notify it becomes available. we may want to recover
-                            // from backpressure.
-                            Some(WakerInterest::WorkerAvailable) => {
-                                drop(guard);
-                                self.maybe_backpressure(&mut sockets, false);
-                            }
-                            // a new worker thread is made and it's handle would be added to Accept
-                            Some(WakerInterest::Worker(handle)) => {
-                                drop(guard);
-                                // maybe we want to recover from a backpressure.
-                                self.maybe_backpressure(&mut sockets, false);
-                                self.handles.push(handle);
-                            }
-                            // got timer interest and it's time to try register socket(s) again
-                            Some(WakerInterest::Timer) => {
-                                drop(guard);
-                                self.process_timer(&mut sockets)
-                            }
-                            Some(WakerInterest::Pause) => {
-                                drop(guard);
-                                self.deregister_all(&mut sockets);
-                            }
-                            Some(WakerInterest::Resume) => {
-                                drop(guard);
-                                sockets.iter_mut().for_each(|(token, info)| {
-                                    self.register_logged(token, info);
-                                });
-                            }
-                            Some(WakerInterest::Stop) => {
-                                return self.deregister_all(&mut sockets);
-                            }
-                            // waker queue is drained
-                            None => {
-                                // Reset the WakerQueue before break so it does not grow infinitely
-                                WakerQueue::reset(&mut guard);
-                                break 'waker;
-                            }
+                    WAKER_TOKEN => {
+                        let exit = self.handle_waker(sockets);
+                        if exit {
+                            info!("Accept is stopped.");
+                            return;
                         }
-                    },
+                    }
                     _ => {
                         let token = usize::from(token);
-                        self.accept(&mut sockets, token);
+                        self.accept(sockets, token);
                     }
                 }
             }
         }
     }
 
-    fn process_timer(&self, sockets: &mut Slab<ServerSocketInfo>) {
+    fn handle_waker(&mut self, sockets: &mut [ServerSocketInfo]) -> bool {
+        // This is a loop because interests for command from previous version was
+        // a loop that would try to drain the command channel. It's yet unknown
+        // if it's necessary/good practice to actively drain the waker queue.
+        loop {
+            // take guard with every iteration so no new interest can be added
+            // until the current task is done.
+            let mut guard = self.waker.guard();
+            match guard.pop_front() {
+                // worker notify it becomes available.
+                Some(WakerInterest::WorkerAvailable(idx)) => {
+                    drop(guard);
+
+                    self.avail.set_available(idx, true);
+
+                    if !self.paused {
+                        self.accept_all(sockets);
+                    }
+                }
+                // a new worker thread is made and it's handle would be added to Accept
+                Some(WakerInterest::Worker(handle)) => {
+                    drop(guard);
+
+                    self.avail.set_available(handle.idx(), true);
+                    self.handles.push(handle);
+
+                    if !self.paused {
+                        self.accept_all(sockets);
+                    }
+                }
+                // got timer interest and it's time to try register socket(s) again
+                Some(WakerInterest::Timer) => {
+                    drop(guard);
+
+                    self.process_timer(sockets)
+                }
+                Some(WakerInterest::Pause) => {
+                    drop(guard);
+
+                    if !self.paused {
+                        self.paused = true;
+
+                        self.deregister_all(sockets);
+                    }
+                }
+                Some(WakerInterest::Resume) => {
+                    drop(guard);
+
+                    if self.paused {
+                        self.paused = false;
+
+                        sockets.iter_mut().for_each(|info| {
+                            self.register_logged(info);
+                        });
+
+                        self.accept_all(sockets);
+                    }
+                }
+                Some(WakerInterest::Stop) => {
+                    if !self.paused {
+                        self.deregister_all(sockets);
+                    }
+
+                    return true;
+                }
+                // waker queue is drained
+                None => {
+                    // Reset the WakerQueue before break so it does not grow infinitely
+                    WakerQueue::reset(&mut guard);
+
+                    return false;
+                }
+            }
+        }
+    }
+
+    fn process_timer(&self, sockets: &mut [ServerSocketInfo]) {
         let now = Instant::now();
         sockets
             .iter_mut()
             // Only sockets that had an associated timeout were deregistered.
-            .filter(|(_, info)| info.timeout.is_some())
-            .for_each(|(token, info)| {
+            .filter(|info| info.timeout.is_some())
+            .for_each(|info| {
                 let inst = info.timeout.take().unwrap();
 
                 if now < inst {
                     info.timeout = Some(inst);
-                } else if !self.backpressure {
-                    self.register_logged(token, info);
+                } else if !self.paused {
+                    self.register_logged(info);
                 }
 
-                // Drop the timeout if server is in backpressure and socket timeout is expired.
-                // When server recovers from backpressure it will register all sockets without
+                // Drop the timeout if server is paused and socket timeout is expired.
+                // When server recovers from pause it will register all sockets without
                 // a timeout value so this socket register will be delayed till then.
             });
     }
 
     #[cfg(not(target_os = "windows"))]
-    fn register(&self, token: usize, info: &mut ServerSocketInfo) -> io::Result<()> {
+    fn register(&self, info: &mut ServerSocketInfo) -> io::Result<()> {
+        let token = MioToken(info.token);
         self.poll
             .registry()
-            .register(&mut info.lst, MioToken(token), Interest::READABLE)
+            .register(&mut info.lst, token, Interest::READABLE)
     }
 
     #[cfg(target_os = "windows")]
-    fn register(&self, token: usize, info: &mut ServerSocketInfo) -> io::Result<()> {
+    fn register(&self, info: &mut ServerSocketInfo) -> io::Result<()> {
         // On windows, calling register without deregister cause an error.
         // See https://github.com/actix/actix-web/issues/905
         // Calling reregister seems to fix the issue.
+        let token = MioToken(info.token);
         self.poll
             .registry()
-            .register(&mut info.lst, mio::Token(token), Interest::READABLE)
+            .register(&mut info.lst, token, Interest::READABLE)
             .or_else(|_| {
-                self.poll.registry().reregister(
-                    &mut info.lst,
-                    mio::Token(token),
-                    Interest::READABLE,
-                )
+                self.poll
+                    .registry()
+                    .reregister(&mut info.lst, token, Interest::READABLE)
             })
     }
 
-    fn register_logged(&self, token: usize, info: &mut ServerSocketInfo) {
-        match self.register(token, info) {
-            Ok(_) => info!("Resume accepting connections on {}", info.addr),
+    fn register_logged(&self, info: &mut ServerSocketInfo) {
+        match self.register(info) {
+            Ok(_) => info!("Resume accepting connections on {}", info.lst.local_addr()),
             Err(e) => error!("Can not register server socket {}", e),
         }
     }
 
-    fn deregister(&self, info: &mut ServerSocketInfo) -> io::Result<()> {
-        self.poll.registry().deregister(&mut info.lst)
-    }
-
     fn deregister_logged(&self, info: &mut ServerSocketInfo) {
-        match self.deregister(info) {
-            Ok(_) => info!("Paused accepting connections on {}", info.addr),
+        match self.poll.registry().deregister(&mut info.lst) {
+            Ok(_) => info!("Paused accepting connections on {}", info.lst.local_addr()),
             Err(e) => {
                 error!("Can not deregister server socket {}", e)
             }
         }
     }
 
-    fn deregister_all(&self, sockets: &mut Slab<ServerSocketInfo>) {
-        sockets.iter_mut().for_each(|(_, info)| {
-            self.deregister_logged(info);
-        });
-    }
-
-    fn maybe_backpressure(&mut self, sockets: &mut Slab<ServerSocketInfo>, on: bool) {
-        // Only operate when server is in a different backpressure than the given flag.
-        if self.backpressure != on {
-            if on {
-                self.backpressure = true;
-                // TODO: figure out if timing out sockets can be safely de-registered twice.
-                self.deregister_all(sockets);
-            } else {
-                self.backpressure = false;
-                sockets
-                    .iter_mut()
-                    // Only operate on sockets without associated timeout.
-                    // Sockets with it will attempt to re-register when their timeout expires.
-                    .filter(|(_, info)| info.timeout.is_none())
-                    .for_each(|(token, info)| self.register_logged(token, info));
-            }
-        }
-    }
-
-    fn accept_one(&mut self, sockets: &mut Slab<ServerSocketInfo>, mut conn: Conn) {
-        if self.backpressure {
-            // send_connection would remove fault worker from handles.
-            // worst case here is conn get dropped after all handles are gone.
-            while !self.handles.is_empty() {
-                match self.send_connection(sockets, conn) {
-                    Ok(_) => return,
-                    Err(c) => conn = c,
-                }
-            }
-        } else {
-            // Do one round and try to send conn to all workers until it succeed.
-            // Start from self.next.
-            let mut idx = 0;
-            while idx < self.handles.len() {
-                idx += 1;
-                if self.handles[self.next].available() {
-                    match self.send_connection(sockets, conn) {
-                        Ok(_) => return,
-                        Err(c) => conn = c,
-                    }
-                } else {
-                    self.set_next();
-                }
-            }
-            // Sending Conn failed due to either all workers are in error or not available.
-            // Enter backpressure state and try again.
-            self.maybe_backpressure(sockets, true);
-            self.accept_one(sockets, conn);
-        }
-    }
-
-    // Set next worker handle that would accept work.
-    fn set_next(&mut self) {
-        self.next = (self.next + 1) % self.handles.len();
+    fn deregister_all(&self, sockets: &mut [ServerSocketInfo]) {
+        // This is a best effort implementation with following limitation:
+        //
+        // Every ServerSocketInfo with associate timeout will be skipped and it's timeout
+        // is removed in the process.
+        //
+        // Therefore WakerInterest::Pause followed by WakerInterest::Resume in a very short
+        // gap (less than 500ms) would cause all timing out ServerSocketInfos be reregistered
+        // before expected timing.
+        sockets
+            .iter_mut()
+            // Take all timeout.
+            // This is to prevent Accept::process_timer method re-register a socket afterwards.
+            .map(|info| (info.timeout.take(), info))
+            // Socket info with a timeout is already deregistered so skip them.
+            .filter(|(timeout, _)| timeout.is_none())
+            .for_each(|(_, info)| self.deregister_logged(info));
     }
 
     // Send connection to worker and handle error.
-    fn send_connection(
-        &mut self,
-        sockets: &mut Slab<ServerSocketInfo>,
-        conn: Conn,
-    ) -> Result<(), Conn> {
-        match self.handles[self.next].send(conn) {
+    fn send_connection(&mut self, conn: Conn) -> Result<(), Conn> {
+        let next = self.next();
+        match next.send(conn) {
             Ok(_) => {
+                // Increment counter of WorkerHandle.
+                // Set worker to unavailable with it hit max (Return false).
+                if !next.inc_counter() {
+                    let idx = next.idx();
+                    self.avail.set_available(idx, false);
+                }
                 self.set_next();
                 Ok(())
             }
             Err(conn) => {
-                // worker lost contact and could be gone. a message is sent to
-                // `ServerBuilder` future to notify it a new worker should be made.
-                // after that remove the fault worker and enter backpressure if necessary.
-                self.srv.worker_faulted(self.handles[self.next].idx);
-                self.handles.swap_remove(self.next);
+                // Worker thread is error and could be gone.
+                // Remove worker handle and notify `ServerBuilder`.
+                self.remove_next();
+
                 if self.handles.is_empty() {
                     error!("No workers");
-                    self.maybe_backpressure(sockets, true);
                     // All workers are gone and Conn is nowhere to be sent.
                     // Treat this situation as Ok and drop Conn.
                     return Ok(());
                 } else if self.handles.len() <= self.next {
                     self.next = 0;
                 }
+
                 Err(conn)
             }
         }
     }
 
-    fn accept(&mut self, sockets: &mut Slab<ServerSocketInfo>, token: usize) {
+    fn accept_one(&mut self, mut conn: Conn) {
         loop {
-            let info = sockets
-                .get_mut(token)
-                .expect("ServerSocketInfo is removed from Slab");
+            let next = self.next();
+            let idx = next.idx();
+
+            if self.avail.get_available(idx) {
+                match self.send_connection(conn) {
+                    Ok(_) => return,
+                    Err(c) => conn = c,
+                }
+            } else {
+                self.avail.set_available(idx, false);
+                self.set_next();
+
+                if !self.avail.available() {
+                    while let Err(c) = self.send_connection(conn) {
+                        conn = c;
+                    }
+                    return;
+                }
+            }
+        }
+    }
+
+    fn accept(&mut self, sockets: &mut [ServerSocketInfo], token: usize) {
+        while self.avail.available() {
+            let info = &mut sockets[token];
 
             match info.lst.accept() {
                 Ok(io) => {
-                    let msg = Conn {
-                        io,
-                        token: info.token,
-                    };
-                    self.accept_one(sockets, msg);
+                    let conn = Conn { io, token };
+                    self.accept_one(conn);
                 }
                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return,
                 Err(ref e) if connection_error(e) => continue,
@@ -431,4 +494,99 @@ impl Accept {
             };
         }
     }
+
+    fn accept_all(&mut self, sockets: &mut [ServerSocketInfo]) {
+        sockets
+            .iter_mut()
+            .map(|info| info.token)
+            .collect::<Vec<_>>()
+            .into_iter()
+            .for_each(|idx| self.accept(sockets, idx))
+    }
+
+    #[inline(always)]
+    fn next(&self) -> &WorkerHandleAccept {
+        &self.handles[self.next]
+    }
+
+    /// Set next worker handle that would accept connection.
+    #[inline(always)]
+    fn set_next(&mut self) {
+        self.next = (self.next + 1) % self.handles.len();
+    }
+
+    /// Remove next worker handle that fail to accept connection.
+    fn remove_next(&mut self) {
+        let handle = self.handles.swap_remove(self.next);
+        let idx = handle.idx();
+        // A message is sent to `ServerBuilder` future to notify it a new worker
+        // should be made.
+        self.srv.worker_faulted(idx);
+        self.avail.set_available(idx, false);
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::Availability;
+
+    fn single(aval: &mut Availability, idx: usize) {
+        aval.set_available(idx, true);
+        assert!(aval.available());
+
+        aval.set_available(idx, true);
+
+        aval.set_available(idx, false);
+        assert!(!aval.available());
+
+        aval.set_available(idx, false);
+        assert!(!aval.available());
+    }
+
+    fn multi(aval: &mut Availability, mut idx: Vec<usize>) {
+        idx.iter().for_each(|idx| aval.set_available(*idx, true));
+
+        assert!(aval.available());
+
+        while let Some(idx) = idx.pop() {
+            assert!(aval.available());
+            aval.set_available(idx, false);
+        }
+
+        assert!(!aval.available());
+    }
+
+    #[test]
+    fn availability() {
+        let mut aval = Availability::default();
+
+        single(&mut aval, 1);
+        single(&mut aval, 128);
+        single(&mut aval, 256);
+        single(&mut aval, 511);
+
+        let idx = (0..511).filter(|i| i % 3 == 0 && i % 5 == 0).collect();
+
+        multi(&mut aval, idx);
+
+        multi(&mut aval, (0..511).collect())
+    }
+
+    #[test]
+    #[should_panic]
+    fn overflow() {
+        let mut aval = Availability::default();
+        single(&mut aval, 512);
+    }
+
+    #[test]
+    fn pin_point() {
+        let mut aval = Availability::default();
+
+        aval.set_available(438, true);
+
+        aval.set_available(479, true);
+
+        assert_eq!(aval.0[3], 1 << (438 - 384) | 1 << (479 - 384));
+    }
 }

actix-server/src/builder.rs

@@ -8,28 +8,29 @@ use std::{
 
 use actix_rt::{self as rt, net::TcpStream, time::sleep, System};
 use log::{error, info};
-use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
-use tokio::sync::oneshot;
+use tokio::sync::{
+    mpsc::{unbounded_channel, UnboundedReceiver},
+    oneshot,
+};
 
 use crate::accept::AcceptLoop;
-use crate::config::{ConfiguredService, ServiceConfig};
+use crate::join_all;
 use crate::server::{Server, ServerCommand};
 use crate::service::{InternalServiceFactory, ServiceFactory, StreamNewService};
 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, ServerWorkerConfig, WorkerAvailability, WorkerHandle};
-use crate::{join_all, Token};
+use crate::worker::{ServerWorker, ServerWorkerConfig, WorkerHandleAccept, WorkerHandleServer};
 
 /// Server builder
 pub struct ServerBuilder {
     threads: usize,
-    token: Token,
+    token: usize,
     backlog: u32,
-    handles: Vec<(usize, WorkerHandle)>,
+    handles: Vec<(usize, WorkerHandleServer)>,
     services: Vec<Box<dyn InternalServiceFactory>>,
-    sockets: Vec<(Token, String, MioListener)>,
+    sockets: Vec<(usize, String, MioListener)>,
     accept: AcceptLoop,
     exit: bool,
     no_signals: bool,
@@ -53,7 +54,7 @@ impl ServerBuilder {
 
         ServerBuilder {
             threads: num_cpus::get(),
-            token: Token::default(),
+            token: 0,
             handles: Vec::new(),
             services: Vec::new(),
             sockets: Vec::new(),
@@ -117,8 +118,8 @@ impl ServerBuilder {
     /// reached for each worker.
     ///
     /// By default max connections is set to a 25k per worker.
-    pub fn maxconn(self, num: usize) -> Self {
-        worker::max_concurrent_connections(num);
+    pub fn maxconn(mut self, num: usize) -> Self {
+        self.worker_config.max_concurrent_connections(num);
         self
     }
 
@@ -146,32 +147,6 @@ impl ServerBuilder {
         self
     }
 
-    /// Execute external configuration as part of the server building process.
-    ///
-    /// This function is useful for moving parts of configuration to a different module or
-    /// even library.
-    pub fn configure<F>(mut self, f: F) -> io::Result<ServerBuilder>
-    where
-        F: Fn(&mut ServiceConfig) -> io::Result<()>,
-    {
-        let mut cfg = ServiceConfig::new(self.threads, self.backlog);
-
-        f(&mut cfg)?;
-
-        if let Some(apply) = cfg.apply {
-            let mut srv = ConfiguredService::new(apply);
-            for (name, lst) in cfg.services {
-                let token = self.token.next();
-                srv.stream(token, name.clone(), lst.local_addr()?);
-                self.sockets.push((token, name, MioListener::Tcp(lst)));
-            }
-            self.services.push(Box::new(srv));
-        }
-        self.threads = cfg.threads;
-
-        Ok(self)
-    }
-
     /// Add new service to the server.
     pub fn bind<F, U, N: AsRef<str>>(mut self, name: N, addr: U, factory: F) -> io::Result<Self>
     where
@@ -181,7 +156,7 @@ impl ServerBuilder {
         let sockets = bind_addr(addr, self.backlog)?;
 
         for lst in sockets {
-            let token = self.token.next();
+            let token = self.next_token();
             self.services.push(StreamNewService::create(
                 name.as_ref().to_string(),
                 token,
@@ -230,7 +205,7 @@ impl ServerBuilder {
     {
         use std::net::{IpAddr, Ipv4Addr};
         lst.set_nonblocking(true)?;
-        let token = self.token.next();
+        let token = self.next_token();
         let addr = StdSocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
         self.services.push(StreamNewService::create(
             name.as_ref().to_string(),
@@ -256,7 +231,7 @@ impl ServerBuilder {
         lst.set_nonblocking(true)?;
         let addr = lst.local_addr()?;
 
-        let token = self.token.next();
+        let token = self.next_token();
         self.services.push(StreamNewService::create(
             name.as_ref().to_string(),
             token,
@@ -280,10 +255,11 @@ impl ServerBuilder {
             // start workers
             let handles = (0..self.threads)
                 .map(|idx| {
-                    let handle = self.start_worker(idx, self.accept.waker_owned());
-                    self.handles.push((idx, handle.clone()));
+                    let (handle_accept, handle_server) =
+                        self.start_worker(idx, self.accept.waker_owned());
+                    self.handles.push((idx, handle_server));
 
-                    handle
+                    handle_accept
                 })
                 .collect();
 
@@ -311,11 +287,14 @@ impl ServerBuilder {
         }
     }
 
-    fn start_worker(&self, idx: usize, waker: WakerQueue) -> WorkerHandle {
-        let avail = WorkerAvailability::new(waker);
+    fn start_worker(
+        &self,
+        idx: usize,
+        waker_queue: WakerQueue,
+    ) -> (WorkerHandleAccept, WorkerHandleServer) {
         let services = self.services.iter().map(|v| v.clone_factory()).collect();
 
-        ServerWorker::start(idx, services, avail, self.worker_config)
+        ServerWorker::start(idx, services, waker_queue, self.worker_config)
     }
 
     fn handle_cmd(&mut self, item: ServerCommand) {
@@ -373,45 +352,29 @@ impl ServerBuilder {
                 let notify = std::mem::take(&mut self.notify);
 
                 // stop workers
-                if !self.handles.is_empty() && graceful {
-                    let iter = self
-                        .handles
-                        .iter()
-                        .map(move |worker| worker.1.stop(graceful))
-                        .collect();
+                let stop = self
+                    .handles
+                    .iter()
+                    .map(move |worker| worker.1.stop(graceful))
+                    .collect();
 
-                    let fut = join_all(iter);
-
-                    rt::spawn(async move {
-                        let _ = fut.await;
-                        if let Some(tx) = completion {
-                            let _ = tx.send(());
-                        }
-                        for tx in notify {
-                            let _ = tx.send(());
-                        }
-                        if exit {
-                            rt::spawn(async {
-                                sleep(Duration::from_millis(300)).await;
-                                System::current().stop();
-                            });
-                        }
-                    });
-                } else {
-                    // we need to stop system if server was spawned
-                    if self.exit {
-                        rt::spawn(async {
-                            sleep(Duration::from_millis(300)).await;
-                            System::current().stop();
-                        });
+                rt::spawn(async move {
+                    if graceful {
+                        let _ = join_all(stop).await;
                     }
+
                     if let Some(tx) = completion {
                         let _ = tx.send(());
                     }
                     for tx in notify {
                         let _ = tx.send(());
                     }
-                }
+
+                    if exit {
+                        sleep(Duration::from_millis(300)).await;
+                        System::current().stop();
+                    }
+                });
             }
             ServerCommand::WorkerFaulted(idx) => {
                 let mut found = false;
@@ -437,13 +400,20 @@ impl ServerBuilder {
                         break;
                     }
 
-                    let handle = self.start_worker(new_idx, self.accept.waker_owned());
-                    self.handles.push((new_idx, handle.clone()));
-                    self.accept.wake(WakerInterest::Worker(handle));
+                    let (handle_accept, handle_server) =
+                        self.start_worker(new_idx, self.accept.waker_owned());
+                    self.handles.push((new_idx, handle_server));
+                    self.accept.wake(WakerInterest::Worker(handle_accept));
                 }
             }
         }
     }
+
+    fn next_token(&mut self) -> usize {
+        let token = self.token;
+        self.token += 1;
+        token
+    }
 }
 
 impl Future for ServerBuilder {

actix-server/src/config.rs

@@ -1,287 +0,0 @@
-use std::collections::HashMap;
-use std::future::Future;
-use std::{fmt, io};
-
-use actix_rt::net::TcpStream;
-use actix_service::{
-    fn_service, IntoServiceFactory as IntoBaseServiceFactory,
-    ServiceFactory as BaseServiceFactory,
-};
-use actix_utils::counter::CounterGuard;
-use futures_core::future::LocalBoxFuture;
-use log::error;
-
-use crate::builder::bind_addr;
-use crate::service::{BoxedServerService, InternalServiceFactory, StreamService};
-use crate::socket::{MioStream, MioTcpListener, StdSocketAddr, StdTcpListener, ToSocketAddrs};
-use crate::{ready, Token};
-
-pub struct ServiceConfig {
-    pub(crate) services: Vec<(String, MioTcpListener)>,
-    pub(crate) apply: Option<Box<dyn ServiceRuntimeConfiguration>>,
-    pub(crate) threads: usize,
-    pub(crate) backlog: u32,
-}
-
-impl ServiceConfig {
-    pub(super) fn new(threads: usize, backlog: u32) -> ServiceConfig {
-        ServiceConfig {
-            threads,
-            backlog,
-            services: Vec::new(),
-            apply: None,
-        }
-    }
-
-    /// Set number of workers to start.
-    ///
-    /// By default server uses number of available logical cpu as workers
-    /// count.
-    pub fn workers(&mut self, num: usize) {
-        self.threads = num;
-    }
-
-    /// Add new service to server
-    pub fn bind<U, N: AsRef<str>>(&mut self, name: N, addr: U) -> io::Result<&mut Self>
-    where
-        U: ToSocketAddrs,
-    {
-        let sockets = bind_addr(addr, self.backlog)?;
-
-        for lst in sockets {
-            self._listen(name.as_ref(), lst);
-        }
-
-        Ok(self)
-    }
-
-    /// Add new service to server
-    pub fn listen<N: AsRef<str>>(&mut self, name: N, lst: StdTcpListener) -> &mut Self {
-        self._listen(name, MioTcpListener::from_std(lst))
-    }
-
-    /// Register service configuration function. This function get called
-    /// during worker runtime configuration. It get executed in worker thread.
-    pub fn apply<F>(&mut self, f: F) -> io::Result<()>
-    where
-        F: Fn(&mut ServiceRuntime) + Send + Clone + 'static,
-    {
-        self.apply = Some(Box::new(f));
-        Ok(())
-    }
-
-    fn _listen<N: AsRef<str>>(&mut self, name: N, lst: MioTcpListener) -> &mut Self {
-        if self.apply.is_none() {
-            self.apply = Some(Box::new(not_configured));
-        }
-        self.services.push((name.as_ref().to_string(), lst));
-        self
-    }
-}
-
-pub(super) struct ConfiguredService {
-    rt: Box<dyn ServiceRuntimeConfiguration>,
-    names: HashMap<Token, (String, StdSocketAddr)>,
-    topics: HashMap<String, Token>,
-    services: Vec<Token>,
-}
-
-impl ConfiguredService {
-    pub(super) fn new(rt: Box<dyn ServiceRuntimeConfiguration>) -> Self {
-        ConfiguredService {
-            rt,
-            names: HashMap::new(),
-            topics: HashMap::new(),
-            services: Vec::new(),
-        }
-    }
-
-    pub(super) fn stream(&mut self, token: Token, name: String, addr: StdSocketAddr) {
-        self.names.insert(token, (name.clone(), addr));
-        self.topics.insert(name, token);
-        self.services.push(token);
-    }
-}
-
-impl InternalServiceFactory for ConfiguredService {
-    fn name(&self, token: Token) -> &str {
-        &self.names[&token].0
-    }
-
-    fn clone_factory(&self) -> Box<dyn InternalServiceFactory> {
-        Box::new(Self {
-            rt: self.rt.clone(),
-            names: self.names.clone(),
-            topics: self.topics.clone(),
-            services: self.services.clone(),
-        })
-    }
-
-    fn create(&self) -> LocalBoxFuture<'static, Result<Vec<(Token, BoxedServerService)>, ()>> {
-        // configure services
-        let mut rt = ServiceRuntime::new(self.topics.clone());
-        self.rt.configure(&mut rt);
-        rt.validate();
-        let mut names = self.names.clone();
-        let tokens = self.services.clone();
-
-        // construct services
-        Box::pin(async move {
-            let mut services = rt.services;
-            // TODO: Proper error handling here
-            for f in rt.onstart.into_iter() {
-                f.await;
-            }
-            let mut res = vec![];
-            for token in tokens {
-                if let Some(srv) = services.remove(&token) {
-                    let newserv = srv.new_service(());
-                    match newserv.await {
-                        Ok(serv) => {
-                            res.push((token, serv));
-                        }
-                        Err(_) => {
-                            error!("Can not construct service");
-                            return Err(());
-                        }
-                    }
-                } else {
-                    let name = names.remove(&token).unwrap().0;
-                    res.push((
-                        token,
-                        Box::new(StreamService::new(fn_service(move |_: TcpStream| {
-                            error!("Service {:?} is not configured", name);
-                            ready::<Result<_, ()>>(Ok(()))
-                        }))),
-                    ));
-                };
-            }
-            Ok(res)
-        })
-    }
-}
-
-pub(super) trait ServiceRuntimeConfiguration: Send {
-    fn clone(&self) -> Box<dyn ServiceRuntimeConfiguration>;
-
-    fn configure(&self, rt: &mut ServiceRuntime);
-}
-
-impl<F> ServiceRuntimeConfiguration for F
-where
-    F: Fn(&mut ServiceRuntime) + Send + Clone + 'static,
-{
-    fn clone(&self) -> Box<dyn ServiceRuntimeConfiguration> {
-        Box::new(self.clone())
-    }
-
-    fn configure(&self, rt: &mut ServiceRuntime) {
-        (self)(rt)
-    }
-}
-
-fn not_configured(_: &mut ServiceRuntime) {
-    error!("Service is not configured");
-}
-
-pub struct ServiceRuntime {
-    names: HashMap<String, Token>,
-    services: HashMap<Token, BoxedNewService>,
-    onstart: Vec<LocalBoxFuture<'static, ()>>,
-}
-
-impl ServiceRuntime {
-    fn new(names: HashMap<String, Token>) -> Self {
-        ServiceRuntime {
-            names,
-            services: HashMap::new(),
-            onstart: Vec::new(),
-        }
-    }
-
-    fn validate(&self) {
-        for (name, token) in &self.names {
-            if !self.services.contains_key(&token) {
-                error!("Service {:?} is not configured", name);
-            }
-        }
-    }
-
-    /// Register service.
-    ///
-    /// Name of the service must be registered during configuration stage with
-    /// *ServiceConfig::bind()* or *ServiceConfig::listen()* methods.
-    pub fn service<T, F>(&mut self, name: &str, service: F)
-    where
-        F: IntoBaseServiceFactory<T, TcpStream>,
-        T: BaseServiceFactory<TcpStream, Config = ()> + 'static,
-        T::Future: 'static,
-        T::Service: 'static,
-        T::InitError: fmt::Debug,
-    {
-        // let name = name.to_owned();
-        if let Some(token) = self.names.get(name) {
-            self.services.insert(
-                *token,
-                Box::new(ServiceFactory {
-                    inner: service.into_factory(),
-                }),
-            );
-        } else {
-            panic!("Unknown service: {:?}", name);
-        }
-    }
-
-    /// Execute future before services initialization.
-    pub fn on_start<F>(&mut self, fut: F)
-    where
-        F: Future<Output = ()> + 'static,
-    {
-        self.onstart.push(Box::pin(fut))
-    }
-}
-
-type BoxedNewService = Box<
-    dyn BaseServiceFactory<
-        (Option<CounterGuard>, MioStream),
-        Response = (),
-        Error = (),
-        InitError = (),
-        Config = (),
-        Service = BoxedServerService,
-        Future = LocalBoxFuture<'static, Result<BoxedServerService, ()>>,
-    >,
->;
-
-struct ServiceFactory<T> {
-    inner: T,
-}
-
-impl<T> BaseServiceFactory<(Option<CounterGuard>, MioStream)> for ServiceFactory<T>
-where
-    T: BaseServiceFactory<TcpStream, Config = ()>,
-    T::Future: 'static,
-    T::Service: 'static,
-    T::Error: 'static,
-    T::InitError: fmt::Debug + 'static,
-{
-    type Response = ();
-    type Error = ();
-    type Config = ();
-    type Service = BoxedServerService;
-    type InitError = ();
-    type Future = LocalBoxFuture<'static, Result<BoxedServerService, ()>>;
-
-    fn new_service(&self, _: ()) -> Self::Future {
-        let fut = self.inner.new_service(());
-        Box::pin(async move {
-            match fut.await {
-                Ok(s) => Ok(Box::new(StreamService::new(s)) as BoxedServerService),
-                Err(e) => {
-                    error!("Can not construct service: {:?}", e);
-                    Err(())
-                }
-            }
-        })
-    }
-}

actix-server/src/lib.rs

@@ -6,7 +6,6 @@
 
 mod accept;
 mod builder;
-mod config;
 mod server;
 mod service;
 mod signals;
@@ -16,7 +15,6 @@ mod waker_queue;
 mod worker;
 
 pub use self::builder::ServerBuilder;
-pub use self::config::{ServiceConfig, ServiceRuntime};
 pub use self::server::Server;
 pub use self::service::ServiceFactory;
 pub use self::test_server::TestServer;
@@ -28,51 +26,11 @@ use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-/// Socket ID token
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub(crate) struct Token(usize);
-
-impl Default for Token {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl Token {
-    fn new() -> Self {
-        Self(0)
-    }
-
-    pub(crate) fn next(&mut self) -> Token {
-        let token = Token(self.0);
-        self.0 += 1;
-        token
-    }
-}
-
 /// Start server building process
 pub fn new() -> ServerBuilder {
     ServerBuilder::default()
 }
 
-// temporary Ready type for std::future::{ready, Ready}; Can be removed when MSRV surpass 1.48
-#[doc(hidden)]
-pub struct Ready<T>(Option<T>);
-
-pub(crate) fn ready<T>(t: T) -> Ready<T> {
-    Ready(Some(t))
-}
-
-impl<T> Unpin for Ready<T> {}
-
-impl<T> Future for Ready<T> {
-    type Output = T;
-
-    fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
-        Poll::Ready(self.get_mut().0.take().unwrap())
-    }
-}
-
 // a poor man's join future. joined future is only used when starting/stopping the server.
 // pin_project and pinned futures are overkill for this task.
 pub(crate) struct JoinAll<T> {
@@ -132,6 +90,8 @@ impl<T> Future for JoinAll<T> {
 mod test {
     use super::*;
 
+    use actix_utils::future::ready;
+
     #[actix_rt::test]
     async fn test_join_all() {
         let futs = vec![ready(Ok(1)), ready(Err(3)), ready(Ok(9))];

actix-server/src/service.rs

@@ -3,12 +3,12 @@ use std::net::SocketAddr;
 use std::task::{Context, Poll};
 
 use actix_service::{Service, ServiceFactory as BaseServiceFactory};
-use actix_utils::counter::CounterGuard;
+use actix_utils::future::{ready, Ready};
 use futures_core::future::LocalBoxFuture;
 use log::error;
 
 use crate::socket::{FromStream, MioStream};
-use crate::{ready, Ready, Token};
+use crate::worker::WorkerCounterGuard;
 
 pub trait ServiceFactory<Stream: FromStream>: Send + Clone + 'static {
     type Factory: BaseServiceFactory<Stream, Config = ()>;
@@ -17,16 +17,16 @@ pub trait ServiceFactory<Stream: FromStream>: Send + Clone + 'static {
 }
 
 pub(crate) trait InternalServiceFactory: Send {
-    fn name(&self, token: Token) -> &str;
+    fn name(&self, token: usize) -> &str;
 
     fn clone_factory(&self) -> Box<dyn InternalServiceFactory>;
 
-    fn create(&self) -> LocalBoxFuture<'static, Result<Vec<(Token, BoxedServerService)>, ()>>;
+    fn create(&self) -> LocalBoxFuture<'static, Result<(usize, BoxedServerService), ()>>;
 }
 
 pub(crate) type BoxedServerService = Box<
     dyn Service<
-        (Option<CounterGuard>, MioStream),
+        (WorkerCounterGuard, MioStream),
         Response = (),
         Error = (),
         Future = Ready<Result<(), ()>>,
@@ -47,7 +47,7 @@ impl<S, I> StreamService<S, I> {
     }
 }
 
-impl<S, I> Service<(Option<CounterGuard>, MioStream)> for StreamService<S, I>
+impl<S, I> Service<(WorkerCounterGuard, MioStream)> for StreamService<S, I>
 where
     S: Service<I>,
     S::Future: 'static,
@@ -62,7 +62,7 @@ where
         self.service.poll_ready(ctx).map_err(|_| ())
     }
 
-    fn call(&self, (guard, req): (Option<CounterGuard>, MioStream)) -> Self::Future {
+    fn call(&self, (guard, req): (WorkerCounterGuard, MioStream)) -> Self::Future {
         ready(match FromStream::from_mio(req) {
             Ok(stream) => {
                 let f = self.service.call(stream);
@@ -83,7 +83,7 @@ where
 pub(crate) struct StreamNewService<F: ServiceFactory<Io>, Io: FromStream> {
     name: String,
     inner: F,
-    token: Token,
+    token: usize,
     addr: SocketAddr,
     _t: PhantomData<Io>,
 }
@@ -95,7 +95,7 @@ where
 {
     pub(crate) fn create(
         name: String,
-        token: Token,
+        token: usize,
         inner: F,
         addr: SocketAddr,
     ) -> Box<dyn InternalServiceFactory> {
@@ -114,7 +114,7 @@ where
     F: ServiceFactory<Io>,
     Io: FromStream + Send + 'static,
 {
-    fn name(&self, _: Token) -> &str {
+    fn name(&self, _: usize) -> &str {
         &self.name
     }
 
@@ -128,14 +128,14 @@ where
         })
     }
 
-    fn create(&self) -> LocalBoxFuture<'static, Result<Vec<(Token, BoxedServerService)>, ()>> {
+    fn create(&self) -> LocalBoxFuture<'static, Result<(usize, BoxedServerService), ()>> {
         let token = self.token;
         let fut = self.inner.create().new_service(());
         Box::pin(async move {
             match fut.await {
                 Ok(inner) => {
                     let service = Box::new(StreamService::new(inner)) as _;
-                    Ok(vec![(token, service)])
+                    Ok((token, service))
                 }
                 Err(_) => Err(()),
             }

actix-server/src/socket.rs

@@ -12,18 +12,7 @@ pub(crate) use {
 use std::{fmt, io};
 
 use actix_rt::net::TcpStream;
-use mio::event::Source;
-use mio::net::TcpStream as MioTcpStream;
-use mio::{Interest, Registry, Token};
-
-#[cfg(windows)]
-use std::os::windows::io::{FromRawSocket, IntoRawSocket};
-#[cfg(unix)]
-use {
-    actix_rt::net::UnixStream,
-    mio::net::{SocketAddr as MioSocketAddr, UnixStream as MioUnixStream},
-    std::os::unix::io::{FromRawFd, IntoRawFd},
-};
+use mio::{event::Source, Interest, Registry, Token};
 
 pub(crate) enum MioListener {
     Tcp(MioTcpListener),
@@ -34,9 +23,15 @@ pub(crate) enum MioListener {
 impl MioListener {
     pub(crate) fn local_addr(&self) -> SocketAddr {
         match *self {
-            MioListener::Tcp(ref lst) => SocketAddr::Tcp(lst.local_addr().unwrap()),
+            MioListener::Tcp(ref lst) => lst
+                .local_addr()
+                .map(SocketAddr::Tcp)
+                .unwrap_or(SocketAddr::Unknown),
             #[cfg(unix)]
-            MioListener::Uds(ref lst) => SocketAddr::Uds(lst.local_addr().unwrap()),
+            MioListener::Uds(ref lst) => lst
+                .local_addr()
+                .map(SocketAddr::Uds)
+                .unwrap_or(SocketAddr::Unknown),
         }
     }
 
@@ -121,25 +116,27 @@ impl fmt::Debug for MioListener {
 impl fmt::Display for MioListener {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match *self {
-            MioListener::Tcp(ref lst) => write!(f, "{}", lst.local_addr().ok().unwrap()),
+            MioListener::Tcp(ref lst) => write!(f, "{:?}", lst),
             #[cfg(unix)]
-            MioListener::Uds(ref lst) => write!(f, "{:?}", lst.local_addr().ok().unwrap()),
+            MioListener::Uds(ref lst) => write!(f, "{:?}", lst),
         }
     }
 }
 
 pub(crate) enum SocketAddr {
+    Unknown,
     Tcp(StdSocketAddr),
     #[cfg(unix)]
-    Uds(MioSocketAddr),
+    Uds(mio::net::SocketAddr),
 }
 
 impl fmt::Display for SocketAddr {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match *self {
-            SocketAddr::Tcp(ref addr) => write!(f, "{}", addr),
+            Self::Unknown => write!(f, "Unknown SocketAddr"),
+            Self::Tcp(ref addr) => write!(f, "{}", addr),
             #[cfg(unix)]
-            SocketAddr::Uds(ref addr) => write!(f, "{:?}", addr),
+            Self::Uds(ref addr) => write!(f, "{:?}", addr),
         }
     }
 }
@@ -147,18 +144,19 @@ impl fmt::Display for SocketAddr {
 impl fmt::Debug for SocketAddr {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match *self {
-            SocketAddr::Tcp(ref addr) => write!(f, "{:?}", addr),
+            Self::Unknown => write!(f, "Unknown SocketAddr"),
+            Self::Tcp(ref addr) => write!(f, "{:?}", addr),
             #[cfg(unix)]
-            SocketAddr::Uds(ref addr) => write!(f, "{:?}", addr),
+            Self::Uds(ref addr) => write!(f, "{:?}", addr),
         }
     }
 }
 
 #[derive(Debug)]
 pub enum MioStream {
-    Tcp(MioTcpStream),
+    Tcp(mio::net::TcpStream),
     #[cfg(unix)]
-    Uds(MioUnixStream),
+    Uds(mio::net::UnixStream),
 }
 
 /// helper trait for converting mio stream to tokio stream.
@@ -166,47 +164,60 @@ pub trait FromStream: Sized {
     fn from_mio(sock: MioStream) -> io::Result<Self>;
 }
 
-// FIXME: This is a workaround and we need an efficient way to convert between mio and tokio stream
-#[cfg(unix)]
-impl FromStream for TcpStream {
-    fn from_mio(sock: MioStream) -> io::Result<Self> {
-        match sock {
-            MioStream::Tcp(mio) => {
-                let raw = IntoRawFd::into_raw_fd(mio);
-                // SAFETY: This is a in place conversion from mio stream to tokio stream.
-                TcpStream::from_std(unsafe { FromRawFd::from_raw_fd(raw) })
-            }
-            MioStream::Uds(_) => {
-                panic!("Should not happen, bug in server impl");
-            }
-        }
-    }
-}
-
-// FIXME: This is a workaround and we need an efficient way to convert between mio and tokio stream
 #[cfg(windows)]
-impl FromStream for TcpStream {
-    fn from_mio(sock: MioStream) -> io::Result<Self> {
-        match sock {
-            MioStream::Tcp(mio) => {
-                let raw = IntoRawSocket::into_raw_socket(mio);
-                // SAFETY: This is a in place conversion from mio stream to tokio stream.
-                TcpStream::from_std(unsafe { FromRawSocket::from_raw_socket(raw) })
+mod win_impl {
+    use super::*;
+
+    use std::os::windows::io::{FromRawSocket, IntoRawSocket};
+
+    // FIXME: This is a workaround and we need an efficient way to convert between mio and tokio stream
+    impl FromStream for TcpStream {
+        fn from_mio(sock: MioStream) -> io::Result<Self> {
+            match sock {
+                MioStream::Tcp(mio) => {
+                    let raw = IntoRawSocket::into_raw_socket(mio);
+                    // SAFETY: This is a in place conversion from mio stream to tokio stream.
+                    TcpStream::from_std(unsafe { FromRawSocket::from_raw_socket(raw) })
+                }
             }
         }
     }
 }
 
-// FIXME: This is a workaround and we need an efficient way to convert between mio and tokio stream
 #[cfg(unix)]
-impl FromStream for UnixStream {
-    fn from_mio(sock: MioStream) -> io::Result<Self> {
-        match sock {
-            MioStream::Tcp(_) => panic!("Should not happen, bug in server impl"),
-            MioStream::Uds(mio) => {
-                let raw = IntoRawFd::into_raw_fd(mio);
-                // SAFETY: This is a in place conversion from mio stream to tokio stream.
-                UnixStream::from_std(unsafe { FromRawFd::from_raw_fd(raw) })
+mod unix_impl {
+    use super::*;
+
+    use std::os::unix::io::{FromRawFd, IntoRawFd};
+
+    use actix_rt::net::UnixStream;
+
+    // FIXME: This is a workaround and we need an efficient way to convert between mio and tokio stream
+    impl FromStream for TcpStream {
+        fn from_mio(sock: MioStream) -> io::Result<Self> {
+            match sock {
+                MioStream::Tcp(mio) => {
+                    let raw = IntoRawFd::into_raw_fd(mio);
+                    // SAFETY: This is a in place conversion from mio stream to tokio stream.
+                    TcpStream::from_std(unsafe { FromRawFd::from_raw_fd(raw) })
+                }
+                MioStream::Uds(_) => {
+                    panic!("Should not happen, bug in server impl");
+                }
+            }
+        }
+    }
+
+    // FIXME: This is a workaround and we need an efficient way to convert between mio and tokio stream
+    impl FromStream for UnixStream {
+        fn from_mio(sock: MioStream) -> io::Result<Self> {
+            match sock {
+                MioStream::Tcp(_) => panic!("Should not happen, bug in server impl"),
+                MioStream::Uds(mio) => {
+                    let raw = IntoRawFd::into_raw_fd(mio);
+                    // SAFETY: This is a in place conversion from mio stream to tokio stream.
+                    UnixStream::from_std(unsafe { FromRawFd::from_raw_fd(raw) })
+                }
             }
         }
     }

actix-server/src/waker_queue.rs

@@ -6,7 +6,7 @@ use std::{
 
 use mio::{Registry, Token as MioToken, Waker};
 
-use crate::worker::WorkerHandle;
+use crate::worker::WorkerHandleAccept;
 
 /// Waker token for `mio::Poll` instance.
 pub(crate) const WAKER_TOKEN: MioToken = MioToken(usize::MAX);
@@ -72,7 +72,7 @@ impl WakerQueue {
 pub(crate) enum WakerInterest {
     /// `WorkerAvailable` is an interest from `Worker` notifying `Accept` there is a worker
     /// available and can accept new tasks.
-    WorkerAvailable,
+    WorkerAvailable(usize),
     /// `Pause`, `Resume`, `Stop` Interest are from `ServerBuilder` future. It listens to
     /// `ServerCommand` and notify `Accept` to do exactly these tasks.
     Pause,
@@ -84,6 +84,6 @@ pub(crate) enum WakerInterest {
     Timer,
     /// `Worker` is an interest happen after a worker runs into faulted state(This is determined
     /// by if work can be sent to it successfully).`Accept` would be waked up and add the new
-    /// `WorkerHandle`.
-    Worker(WorkerHandle),
+    /// `WorkerHandleAccept`.
+    Worker(WorkerHandleAccept),
 }

actix-server/src/worker.rs

@@ -1,138 +1,210 @@
-use std::future::Future;
-use std::pin::Pin;
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::sync::Arc;
-use std::task::{Context, Poll};
-use std::time::Duration;
+use std::{
+    future::Future,
+    mem,
+    pin::Pin,
+    rc::Rc,
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    },
+    task::{Context, Poll},
+    time::Duration,
+};
 
-use actix_rt::time::{sleep, Sleep};
-use actix_rt::{spawn, Arbiter};
-use actix_utils::counter::Counter;
-use futures_core::future::LocalBoxFuture;
+use actix_rt::{
+    spawn,
+    time::{sleep, Instant, Sleep},
+    Arbiter,
+};
+use futures_core::{future::LocalBoxFuture, ready};
 use log::{error, info, trace};
-use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
-use tokio::sync::oneshot;
+use tokio::sync::{
+    mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
+    oneshot,
+};
 
+use crate::join_all;
 use crate::service::{BoxedServerService, InternalServiceFactory};
 use crate::socket::MioStream;
 use crate::waker_queue::{WakerInterest, WakerQueue};
-use crate::{join_all, Token};
 
-pub(crate) struct WorkerCommand(Conn);
-
-/// Stop worker message. Returns `true` on successful shutdown
-/// and `false` if some connections still alive.
-pub(crate) struct StopCommand {
+/// Stop worker message. Returns `true` on successful graceful shutdown.
+/// and `false` if some connections still alive when shutdown execute.
+pub(crate) struct Stop {
     graceful: bool,
-    result: oneshot::Sender<bool>,
+    tx: oneshot::Sender<bool>,
 }
 
 #[derive(Debug)]
 pub(crate) struct Conn {
     pub io: MioStream,
-    pub token: Token,
+    pub token: usize,
 }
 
-static MAX_CONNS: AtomicUsize = AtomicUsize::new(25600);
+fn handle_pair(
+    idx: usize,
+    tx1: UnboundedSender<Conn>,
+    tx2: UnboundedSender<Stop>,
+    counter: Counter,
+) -> (WorkerHandleAccept, WorkerHandleServer) {
+    let accept = WorkerHandleAccept {
+        idx,
+        tx: tx1,
+        counter,
+    };
 
-/// Sets the maximum per-worker number of concurrent connections.
+    let server = WorkerHandleServer { idx, tx: tx2 };
+
+    (accept, server)
+}
+
+/// counter: Arc<AtomicUsize> field is owned by `Accept` thread and `ServerWorker` thread.
 ///
-/// All socket listeners will stop accepting connections when this limit is
-/// reached for each worker.
+/// `Accept` would increment the counter and `ServerWorker` would decrement it.
 ///
-/// By default max connections is set to a 25k per worker.
-pub fn max_concurrent_connections(num: usize) {
-    MAX_CONNS.store(num, Ordering::Relaxed);
-}
-
-thread_local! {
-    static MAX_CONNS_COUNTER: Counter =
-        Counter::new(MAX_CONNS.load(Ordering::Relaxed));
-}
-
-pub(crate) fn num_connections() -> usize {
-    MAX_CONNS_COUNTER.with(|conns| conns.total())
-}
-
-// a handle to worker that can send message to worker and share the availability of worker to other
-// thread.
+/// # Atomic Ordering:
+///
+/// `Accept` always look into it's cached `Availability` field for `ServerWorker` state.
+/// It lazily increment counter after successful dispatching new work to `ServerWorker`.
+/// On reaching counter limit `Accept` update it's cached `Availability` and mark worker as
+/// unable to accept any work.
+///
+/// `ServerWorker` always decrement the counter when every work received from `Accept` is done.
+/// On reaching counter limit worker would use `mio::Waker` and `WakerQueue` to wake up `Accept`
+/// and notify it to update cached `Availability` again to mark worker as able to accept work again.
+///
+/// Hense a wake up would only happen after `Accept` increment it to limit.
+/// And a decrement to limit always wake up `Accept`.
 #[derive(Clone)]
-pub(crate) struct WorkerHandle {
-    pub idx: usize,
-    tx1: UnboundedSender<WorkerCommand>,
-    tx2: UnboundedSender<StopCommand>,
-    avail: WorkerAvailability,
+pub(crate) struct Counter {
+    counter: Arc<AtomicUsize>,
+    limit: usize,
 }
 
-impl WorkerHandle {
-    pub fn new(
-        idx: usize,
-        tx1: UnboundedSender<WorkerCommand>,
-        tx2: UnboundedSender<StopCommand>,
-        avail: WorkerAvailability,
-    ) -> Self {
-        WorkerHandle {
+impl Counter {
+    pub(crate) fn new(limit: usize) -> Self {
+        Self {
+            counter: Arc::new(AtomicUsize::new(1)),
+            limit,
+        }
+    }
+
+    /// Increment counter by 1 and return true when hitting limit
+    #[inline(always)]
+    pub(crate) fn inc(&self) -> bool {
+        self.counter.fetch_add(1, Ordering::Relaxed) != self.limit
+    }
+
+    /// Decrement counter by 1 and return true if crossing limit.
+    #[inline(always)]
+    pub(crate) fn dec(&self) -> bool {
+        self.counter.fetch_sub(1, Ordering::Relaxed) == self.limit
+    }
+
+    pub(crate) fn total(&self) -> usize {
+        self.counter.load(Ordering::SeqCst) - 1
+    }
+}
+
+pub(crate) struct WorkerCounter {
+    idx: usize,
+    inner: Rc<(WakerQueue, Counter)>,
+}
+
+impl Clone for WorkerCounter {
+    fn clone(&self) -> Self {
+        Self {
+            idx: self.idx,
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl WorkerCounter {
+    pub(crate) fn new(idx: usize, waker_queue: WakerQueue, counter: Counter) -> Self {
+        Self {
             idx,
-            tx1,
-            tx2,
-            avail,
+            inner: Rc::new((waker_queue, counter)),
         }
     }
 
-    pub fn send(&self, msg: Conn) -> Result<(), Conn> {
-        self.tx1.send(WorkerCommand(msg)).map_err(|msg| msg.0 .0)
+    #[inline(always)]
+    pub(crate) fn guard(&self) -> WorkerCounterGuard {
+        WorkerCounterGuard(self.clone())
     }
 
-    pub fn available(&self) -> bool {
-        self.avail.available()
+    fn total(&self) -> usize {
+        self.inner.1.total()
+    }
+}
+
+pub(crate) struct WorkerCounterGuard(WorkerCounter);
+
+impl Drop for WorkerCounterGuard {
+    fn drop(&mut self) {
+        let (waker_queue, counter) = &*self.0.inner;
+        if counter.dec() {
+            waker_queue.wake(WakerInterest::WorkerAvailable(self.0.idx));
+        }
+    }
+}
+
+/// Handle to worker that can send connection message to worker and share the
+/// availability of worker to other thread.
+///
+/// Held by [Accept](crate::accept::Accept).
+pub(crate) struct WorkerHandleAccept {
+    idx: usize,
+    tx: UnboundedSender<Conn>,
+    counter: Counter,
+}
+
+impl WorkerHandleAccept {
+    #[inline(always)]
+    pub(crate) fn idx(&self) -> usize {
+        self.idx
     }
 
-    pub fn stop(&self, graceful: bool) -> oneshot::Receiver<bool> {
-        let (result, rx) = oneshot::channel();
-        let _ = self.tx2.send(StopCommand { graceful, result });
+    #[inline(always)]
+    pub(crate) fn send(&self, msg: Conn) -> Result<(), Conn> {
+        self.tx.send(msg).map_err(|msg| msg.0)
+    }
+
+    #[inline(always)]
+    pub(crate) fn inc_counter(&self) -> bool {
+        self.counter.inc()
+    }
+}
+
+/// Handle to worker than can send stop message to worker.
+///
+/// Held by [ServerBuilder](crate::builder::ServerBuilder).
+pub(crate) struct WorkerHandleServer {
+    pub idx: usize,
+    tx: UnboundedSender<Stop>,
+}
+
+impl WorkerHandleServer {
+    pub(crate) fn stop(&self, graceful: bool) -> oneshot::Receiver<bool> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self.tx.send(Stop { graceful, tx });
         rx
     }
 }
 
-#[derive(Clone)]
-pub(crate) struct WorkerAvailability {
-    waker: WakerQueue,
-    available: Arc<AtomicBool>,
-}
-
-impl WorkerAvailability {
-    pub fn new(waker: WakerQueue) -> Self {
-        WorkerAvailability {
-            waker,
-            available: Arc::new(AtomicBool::new(false)),
-        }
-    }
-
-    pub fn available(&self) -> bool {
-        self.available.load(Ordering::Acquire)
-    }
-
-    pub fn set(&self, val: bool) {
-        let old = self.available.swap(val, Ordering::Release);
-        // notify the accept on switched to available.
-        if !old && val {
-            self.waker.wake(WakerInterest::WorkerAvailable);
-        }
-    }
-}
-
 /// Service worker.
 ///
 /// Worker accepts Socket objects via unbounded channel and starts stream processing.
 pub(crate) struct ServerWorker {
-    rx: UnboundedReceiver<WorkerCommand>,
-    rx2: UnboundedReceiver<StopCommand>,
-    services: Vec<WorkerService>,
-    availability: WorkerAvailability,
-    conns: Counter,
-    factories: Vec<Box<dyn InternalServiceFactory>>,
+    // UnboundedReceiver<Conn> should always be the first field.
+    // It must be dropped as soon as ServerWorker dropping.
+    rx: UnboundedReceiver<Conn>,
+    rx2: UnboundedReceiver<Stop>,
+    counter: WorkerCounter,
+    services: Box<[WorkerService]>,
+    factories: Box<[Box<dyn InternalServiceFactory>]>,
     state: WorkerState,
-    config: ServerWorkerConfig,
+    shutdown_timeout: Duration,
 }
 
 struct WorkerService {
@@ -163,6 +235,7 @@ enum WorkerServiceStatus {
 pub(crate) struct ServerWorkerConfig {
     shutdown_timeout: Duration,
     max_blocking_threads: usize,
+    max_concurrent_connections: usize,
 }
 
 impl Default for ServerWorkerConfig {
@@ -172,6 +245,7 @@ impl Default for ServerWorkerConfig {
         Self {
             shutdown_timeout: Duration::from_secs(30),
             max_blocking_threads,
+            max_concurrent_connections: 25600,
         }
     }
 }
@@ -181,6 +255,10 @@ impl ServerWorkerConfig {
         self.max_blocking_threads = num;
     }
 
+    pub(crate) fn max_concurrent_connections(&mut self, num: usize) {
+        self.max_concurrent_connections = num;
+    }
+
     pub(crate) fn shutdown_timeout(&mut self, dur: Duration) {
         self.shutdown_timeout = dur;
     }
@@ -190,13 +268,15 @@ impl ServerWorker {
     pub(crate) fn start(
         idx: usize,
         factories: Vec<Box<dyn InternalServiceFactory>>,
-        availability: WorkerAvailability,
+        waker_queue: WakerQueue,
         config: ServerWorkerConfig,
-    ) -> WorkerHandle {
+    ) -> (WorkerHandleAccept, WorkerHandleServer) {
         let (tx1, rx) = unbounded_channel();
         let (tx2, rx2) = unbounded_channel();
-        let avail = availability.clone();
 
+        let counter = Counter::new(config.max_concurrent_connections);
+
+        let counter_clone = counter.clone();
         // every worker runs in it's own arbiter.
         // use a custom tokio runtime builder to change the settings of runtime.
         Arbiter::with_tokio_rt(move || {
@@ -207,80 +287,83 @@ impl ServerWorker {
                 .unwrap()
         })
         .spawn(async move {
-            availability.set(false);
-            let mut wrk = MAX_CONNS_COUNTER.with(move |conns| ServerWorker {
-                rx,
-                rx2,
-                availability,
-                factories,
-                config,
-                services: Vec::new(),
-                conns: conns.clone(),
-                state: WorkerState::Unavailable,
-            });
-
-            let fut = wrk
-                .factories
+            let fut = factories
                 .iter()
                 .enumerate()
                 .map(|(idx, factory)| {
                     let fut = factory.create();
-                    async move {
-                        fut.await.map(|r| {
-                            r.into_iter().map(|(t, s)| (idx, t, s)).collect::<Vec<_>>()
-                        })
-                    }
+                    async move { fut.await.map(|(t, s)| (idx, t, s)) }
                 })
                 .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.
+            // a second spawn to run !Send future tasks.
             spawn(async move {
-                let res: Result<Vec<_>, _> = join_all(fut).await.into_iter().collect();
-                match res {
-                    Ok(services) => {
-                        for item in services {
-                            for (factory, token, service) in item {
-                                assert_eq!(token.0, wrk.services.len());
-                                wrk.services.push(WorkerService {
-                                    factory,
-                                    service,
-                                    status: WorkerServiceStatus::Unavailable,
-                                });
-                            }
-                        }
-                    }
+                let res = join_all(fut)
+                    .await
+                    .into_iter()
+                    .collect::<Result<Vec<_>, _>>();
+                let services = match res {
+                    Ok(res) => res
+                        .into_iter()
+                        .fold(Vec::new(), |mut services, (factory, token, service)| {
+                            assert_eq!(token, services.len());
+                            services.push(WorkerService {
+                                factory,
+                                service,
+                                status: WorkerServiceStatus::Unavailable,
+                            });
+                            services
+                        })
+                        .into_boxed_slice(),
                     Err(e) => {
                         error!("Can not start worker: {:?}", e);
                         Arbiter::current().stop();
+                        return;
                     }
-                }
-                wrk.await
+                };
+
+                // a third spawn to make sure ServerWorker runs as non boxed future.
+                spawn(ServerWorker {
+                    rx,
+                    rx2,
+                    services,
+                    counter: WorkerCounter::new(idx, waker_queue, counter_clone),
+                    factories: factories.into_boxed_slice(),
+                    state: Default::default(),
+                    shutdown_timeout: config.shutdown_timeout,
+                });
             });
         });
 
-        WorkerHandle::new(idx, tx1, tx2, avail)
+        handle_pair(idx, tx1, tx2, counter)
+    }
+
+    fn restart_service(&mut self, idx: usize, factory_id: usize) {
+        let factory = &self.factories[factory_id];
+        trace!("Service {:?} failed, restarting", factory.name(idx));
+        self.services[idx].status = WorkerServiceStatus::Restarting;
+        self.state = WorkerState::Restarting(Restart {
+            factory_id,
+            token: idx,
+            fut: factory.create(),
+        });
     }
 
     fn shutdown(&mut self, force: bool) {
-        if force {
-            self.services.iter_mut().for_each(|srv| {
-                if srv.status == WorkerServiceStatus::Available {
-                    srv.status = WorkerServiceStatus::Stopped;
-                }
+        self.services
+            .iter_mut()
+            .filter(|srv| srv.status == WorkerServiceStatus::Available)
+            .for_each(|srv| {
+                srv.status = if force {
+                    WorkerServiceStatus::Stopped
+                } else {
+                    WorkerServiceStatus::Stopping
+                };
             });
-        } else {
-            self.services.iter_mut().for_each(move |srv| {
-                if srv.status == WorkerServiceStatus::Available {
-                    srv.status = WorkerServiceStatus::Stopping;
-                }
-            });
-        }
     }
 
-    fn check_readiness(&mut self, cx: &mut Context<'_>) -> Result<bool, (Token, usize)> {
-        let mut ready = self.conns.available(cx);
-        let mut failed = None;
+    fn check_readiness(&mut self, cx: &mut Context<'_>) -> Result<bool, (usize, usize)> {
+        let mut ready = true;
         for (idx, srv) in self.services.iter_mut().enumerate() {
             if srv.status == WorkerServiceStatus::Available
                 || srv.status == WorkerServiceStatus::Unavailable
@@ -290,7 +373,7 @@ impl ServerWorker {
                         if srv.status == WorkerServiceStatus::Unavailable {
                             trace!(
                                 "Service {:?} is available",
-                                self.factories[srv.factory].name(Token(idx))
+                                self.factories[srv.factory].name(idx)
                             );
                             srv.status = WorkerServiceStatus::Available;
                         }
@@ -301,7 +384,7 @@ impl ServerWorker {
                         if srv.status == WorkerServiceStatus::Available {
                             trace!(
                                 "Service {:?} is unavailable",
-                                self.factories[srv.factory].name(Token(idx))
+                                self.factories[srv.factory].name(idx)
                             );
                             srv.status = WorkerServiceStatus::Unavailable;
                         }
@@ -309,173 +392,168 @@ impl ServerWorker {
                     Poll::Ready(Err(_)) => {
                         error!(
                             "Service {:?} readiness check returned error, restarting",
-                            self.factories[srv.factory].name(Token(idx))
+                            self.factories[srv.factory].name(idx)
                         );
-                        failed = Some((Token(idx), srv.factory));
                         srv.status = WorkerServiceStatus::Failed;
+                        return Err((idx, srv.factory));
                     }
                 }
             }
         }
-        if let Some(idx) = failed {
-            Err(idx)
-        } else {
-            Ok(ready)
-        }
+
+        Ok(ready)
     }
 }
 
 enum WorkerState {
     Available,
     Unavailable,
-    Restarting(
-        usize,
-        Token,
-        LocalBoxFuture<'static, Result<Vec<(Token, BoxedServerService)>, ()>>,
-    ),
-    Shutdown(
-        Pin<Box<Sleep>>,
-        Pin<Box<Sleep>>,
-        Option<oneshot::Sender<bool>>,
-    ),
+    Restarting(Restart),
+    Shutdown(Shutdown),
+}
+
+struct Restart {
+    factory_id: usize,
+    token: usize,
+    fut: LocalBoxFuture<'static, Result<(usize, BoxedServerService), ()>>,
+}
+
+// Shutdown keep states necessary for server shutdown:
+// Sleep for interval check the shutdown progress.
+// Instant for the start time of shutdown.
+// Sender for send back the shutdown outcome(force/grace) to StopCommand caller.
+struct Shutdown {
+    timer: Pin<Box<Sleep>>,
+    start_from: Instant,
+    tx: oneshot::Sender<bool>,
+}
+
+impl Default for WorkerState {
+    fn default() -> Self {
+        Self::Unavailable
+    }
+}
+
+impl Drop for ServerWorker {
+    fn drop(&mut self) {
+        // Stop the Arbiter ServerWorker runs on on drop.
+        Arbiter::current().stop();
+    }
 }
 
 impl Future for ServerWorker {
     type Output = ();
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.as_mut().get_mut();
+
         // `StopWorker` message handler
-        if let Poll::Ready(Some(StopCommand { graceful, result })) =
-            Pin::new(&mut self.rx2).poll_recv(cx)
+        if let Poll::Ready(Some(Stop { graceful, tx })) = Pin::new(&mut this.rx2).poll_recv(cx)
         {
-            self.availability.set(false);
-            let num = num_connections();
+            let num = this.counter.total();
             if num == 0 {
                 info!("Shutting down worker, 0 connections");
-                let _ = result.send(true);
+                let _ = tx.send(true);
                 return Poll::Ready(());
             } else if graceful {
-                self.shutdown(false);
-                let num = num_connections();
-                if num != 0 {
-                    info!("Graceful worker shutdown, {} connections", num);
-                    self.state = WorkerState::Shutdown(
-                        Box::pin(sleep(Duration::from_secs(1))),
-                        Box::pin(sleep(self.config.shutdown_timeout)),
-                        Some(result),
-                    );
-                } else {
-                    let _ = result.send(true);
-                    return Poll::Ready(());
-                }
+                info!("Graceful worker shutdown, {} connections", num);
+                this.shutdown(false);
+
+                this.state = WorkerState::Shutdown(Shutdown {
+                    timer: Box::pin(sleep(Duration::from_secs(1))),
+                    start_from: Instant::now(),
+                    tx,
+                });
             } else {
                 info!("Force shutdown worker, {} connections", num);
-                self.shutdown(true);
-                let _ = result.send(false);
+                this.shutdown(true);
+
+                let _ = tx.send(false);
                 return Poll::Ready(());
             }
         }
 
-        match self.state {
-            WorkerState::Unavailable => match self.check_readiness(cx) {
+        match this.state {
+            WorkerState::Unavailable => match this.check_readiness(cx) {
                 Ok(true) => {
-                    self.state = WorkerState::Available;
-                    self.availability.set(true);
+                    this.state = WorkerState::Available;
                     self.poll(cx)
                 }
                 Ok(false) => Poll::Pending,
                 Err((token, idx)) => {
-                    trace!(
-                        "Service {:?} failed, restarting",
-                        self.factories[idx].name(token)
-                    );
-                    self.services[token.0].status = WorkerServiceStatus::Restarting;
-                    self.state =
-                        WorkerState::Restarting(idx, token, self.factories[idx].create());
+                    this.restart_service(token, idx);
                     self.poll(cx)
                 }
             },
-            WorkerState::Restarting(idx, token, ref mut fut) => {
-                match fut.as_mut().poll(cx) {
-                    Poll::Ready(Ok(item)) => {
-                        // only interest in the first item?
-                        if let Some((token, service)) = item.into_iter().next() {
-                            trace!(
-                                "Service {:?} has been restarted",
-                                self.factories[idx].name(token)
-                            );
-                            self.services[token.0].created(service);
-                            self.state = WorkerState::Unavailable;
-                            return self.poll(cx);
-                        }
-                    }
-                    Poll::Ready(Err(_)) => {
+            WorkerState::Restarting(ref mut restart) => {
+                let factory_id = restart.factory_id;
+                let token = restart.token;
+
+                let (token_new, service) = ready!(restart.fut.as_mut().poll(cx))
+                    .unwrap_or_else(|_| {
                         panic!(
                             "Can not restart {:?} service",
-                            self.factories[idx].name(token)
-                        );
-                    }
-                    Poll::Pending => return Poll::Pending,
-                }
+                            this.factories[factory_id].name(token)
+                        )
+                    });
+
+                assert_eq!(token, token_new);
+
+                trace!(
+                    "Service {:?} has been restarted",
+                    this.factories[factory_id].name(token)
+                );
+
+                this.services[token].created(service);
+                this.state = WorkerState::Unavailable;
+
                 self.poll(cx)
             }
-            WorkerState::Shutdown(ref mut t1, ref mut t2, ref mut tx) => {
-                let num = num_connections();
-                if num == 0 {
-                    let _ = tx.take().unwrap().send(true);
-                    Arbiter::current().stop();
-                    return Poll::Ready(());
-                }
+            WorkerState::Shutdown(ref mut shutdown) => {
+                // Wait for 1 second.
+                ready!(shutdown.timer.as_mut().poll(cx));
 
-                // check graceful timeout
-                if Pin::new(t2).poll(cx).is_ready() {
-                    let _ = tx.take().unwrap().send(false);
-                    self.shutdown(true);
-                    Arbiter::current().stop();
-                    return Poll::Ready(());
+                if this.counter.total() == 0 {
+                    // Graceful shutdown.
+                    if let WorkerState::Shutdown(shutdown) = mem::take(&mut this.state) {
+                        let _ = shutdown.tx.send(true);
+                    }
+                    Poll::Ready(())
+                } else if shutdown.start_from.elapsed() >= this.shutdown_timeout {
+                    // Timeout forceful shutdown.
+                    if let WorkerState::Shutdown(shutdown) = mem::take(&mut this.state) {
+                        let _ = shutdown.tx.send(false);
+                    }
+                    Poll::Ready(())
+                } else {
+                    // Reset timer and wait for 1 second.
+                    let time = Instant::now() + Duration::from_secs(1);
+                    shutdown.timer.as_mut().reset(time);
+                    shutdown.timer.as_mut().poll(cx)
                 }
-
-                // sleep for 1 second and then check again
-                if t1.as_mut().poll(cx).is_ready() {
-                    *t1 = Box::pin(sleep(Duration::from_secs(1)));
-                    let _ = t1.as_mut().poll(cx);
-                }
-
-                Poll::Pending
             }
             // actively poll stream and handle worker command
             WorkerState::Available => loop {
-                match self.check_readiness(cx) {
-                    Ok(true) => (),
+                match this.check_readiness(cx) {
+                    Ok(true) => {}
                     Ok(false) => {
                         trace!("Worker is unavailable");
-                        self.availability.set(false);
-                        self.state = WorkerState::Unavailable;
+                        this.state = WorkerState::Unavailable;
                         return self.poll(cx);
                     }
                     Err((token, idx)) => {
-                        trace!(
-                            "Service {:?} failed, restarting",
-                            self.factories[idx].name(token)
-                        );
-                        self.availability.set(false);
-                        self.services[token.0].status = WorkerServiceStatus::Restarting;
-                        self.state =
-                            WorkerState::Restarting(idx, token, self.factories[idx].create());
+                        this.restart_service(token, idx);
                         return self.poll(cx);
                     }
                 }
 
-                match Pin::new(&mut self.rx).poll_recv(cx) {
-                    // handle incoming io stream
-                    Poll::Ready(Some(WorkerCommand(msg))) => {
-                        let guard = self.conns.get();
-                        let _ = self.services[msg.token.0]
-                            .service
-                            .call((Some(guard), msg.io));
+                // handle incoming io stream
+                match ready!(Pin::new(&mut this.rx).poll_recv(cx)) {
+                    Some(msg) => {
+                        let guard = this.counter.guard();
+                        let _ = this.services[msg.token].service.call((guard, msg.io));
                     }
-                    Poll::Pending => return Poll::Pending,
-                    Poll::Ready(None) => return Poll::Ready(()),
+                    None => return Poll::Ready(()),
                 };
             },
         }

actix-server/tests/test_server.rs

@@ -1,7 +1,8 @@
-use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{mpsc, Arc};
-use std::{net, thread, time};
+use std::{net, thread, time::Duration};
 
+use actix_rt::{net::TcpStream, time::sleep};
 use actix_server::Server;
 use actix_service::fn_service;
 use actix_utils::future::ok;
@@ -37,7 +38,7 @@ fn test_bind() {
     });
     let (_, sys) = rx.recv().unwrap();
 
-    thread::sleep(time::Duration::from_millis(500));
+    thread::sleep(Duration::from_millis(500));
     assert!(net::TcpStream::connect(addr).is_ok());
     sys.stop();
     let _ = h.join();
@@ -64,7 +65,7 @@ fn test_listen() {
     });
     let sys = rx.recv().unwrap();
 
-    thread::sleep(time::Duration::from_millis(500));
+    thread::sleep(Duration::from_millis(500));
     assert!(net::TcpStream::connect(addr).is_ok());
     sys.stop();
     let _ = h.join();
@@ -73,11 +74,11 @@ fn test_listen() {
 #[test]
 #[cfg(unix)]
 fn test_start() {
+    use std::io::Read;
+
     use actix_codec::{BytesCodec, Framed};
-    use actix_rt::net::TcpStream;
     use bytes::Bytes;
     use futures_util::sink::SinkExt;
-    use std::io::Read;
 
     let addr = unused_addr();
     let (tx, rx) = mpsc::channel();
@@ -112,16 +113,16 @@ fn test_start() {
 
     // pause
     let _ = srv.pause();
-    thread::sleep(time::Duration::from_millis(200));
+    thread::sleep(Duration::from_millis(200));
     let mut conn = net::TcpStream::connect(addr).unwrap();
-    conn.set_read_timeout(Some(time::Duration::from_millis(100)))
+    conn.set_read_timeout(Some(Duration::from_millis(100)))
         .unwrap();
     let res = conn.read_exact(&mut buf);
     assert!(res.is_err());
 
     // resume
     let _ = srv.resume();
-    thread::sleep(time::Duration::from_millis(100));
+    thread::sleep(Duration::from_millis(100));
     assert!(net::TcpStream::connect(addr).is_ok());
     assert!(net::TcpStream::connect(addr).is_ok());
     assert!(net::TcpStream::connect(addr).is_ok());
@@ -133,61 +134,320 @@ fn test_start() {
 
     // stop
     let _ = srv.stop(false);
-    thread::sleep(time::Duration::from_millis(100));
+    thread::sleep(Duration::from_millis(100));
     assert!(net::TcpStream::connect(addr).is_err());
 
-    thread::sleep(time::Duration::from_millis(100));
+    thread::sleep(Duration::from_millis(100));
     sys.stop();
     let _ = h.join();
 }
 
-#[test]
-fn test_configure() {
-    let addr1 = unused_addr();
-    let addr2 = unused_addr();
-    let addr3 = unused_addr();
+#[actix_rt::test]
+async fn test_max_concurrent_connections() {
+    // Note:
+    // A tcp listener would accept connects based on it's backlog setting.
+    //
+    // The limit test on the other hand is only for concurrent tcp stream limiting a work
+    // thread accept.
+
+    use tokio::io::AsyncWriteExt;
+
+    let addr = unused_addr();
     let (tx, rx) = mpsc::channel();
-    let num = Arc::new(AtomicUsize::new(0));
-    let num2 = num.clone();
+
+    let counter = Arc::new(AtomicUsize::new(0));
+    let counter_clone = counter.clone();
+
+    let max_conn = 3;
 
     let h = thread::spawn(move || {
-        let num = num2.clone();
-        let sys = actix_rt::System::new();
-        let srv = sys.block_on(lazy(|_| {
-            Server::build()
+        actix_rt::System::new().block_on(async {
+            let server = Server::build()
+                // Set a relative higher backlog.
+                .backlog(12)
+                // max connection for a worker is 3.
+                .maxconn(max_conn)
+                .workers(1)
                 .disable_signals()
-                .configure(move |cfg| {
+                .bind("test", addr, move || {
+                    let counter = counter.clone();
+                    fn_service(move |_io: TcpStream| {
+                        let counter = counter.clone();
+                        async move {
+                            counter.fetch_add(1, Ordering::SeqCst);
+                            sleep(Duration::from_secs(20)).await;
+                            counter.fetch_sub(1, Ordering::SeqCst);
+                            Ok::<(), ()>(())
+                        }
+                    })
+                })?
+                .run();
+
+            let _ = tx.send((server.clone(), actix_rt::System::current()));
+
+            server.await
+        })
+    });
+
+    let (srv, sys) = rx.recv().unwrap();
+
+    let mut conns = vec![];
+
+    for _ in 0..12 {
+        let conn = tokio::net::TcpStream::connect(addr).await.unwrap();
+        conns.push(conn);
+    }
+
+    sleep(Duration::from_secs(5)).await;
+
+    // counter would remain at 3 even with 12 successful connection.
+    // and 9 of them remain in backlog.
+    assert_eq!(max_conn, counter_clone.load(Ordering::SeqCst));
+
+    for mut conn in conns {
+        conn.shutdown().await.unwrap();
+    }
+
+    srv.stop(false).await;
+
+    sys.stop();
+    let _ = h.join().unwrap();
+}
+
+#[actix_rt::test]
+async fn test_service_restart() {
+    use std::task::{Context, Poll};
+
+    use actix_service::{fn_factory, Service};
+    use futures_core::future::LocalBoxFuture;
+    use tokio::io::AsyncWriteExt;
+
+    struct TestService(Arc<AtomicUsize>);
+
+    impl Service<TcpStream> for TestService {
+        type Response = ();
+        type Error = ();
+        type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+        fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            let TestService(ref counter) = self;
+            let c = counter.fetch_add(1, Ordering::SeqCst);
+            // Force the service to restart on first readiness check.
+            if c > 0 {
+                Poll::Ready(Ok(()))
+            } else {
+                Poll::Ready(Err(()))
+            }
+        }
+
+        fn call(&self, _: TcpStream) -> Self::Future {
+            Box::pin(async { Ok(()) })
+        }
+    }
+
+    let addr1 = unused_addr();
+    let addr2 = unused_addr();
+    let (tx, rx) = mpsc::channel();
+    let num = Arc::new(AtomicUsize::new(0));
+    let num2 = Arc::new(AtomicUsize::new(0));
+
+    let num_clone = num.clone();
+    let num2_clone = num2.clone();
+
+    let h = thread::spawn(move || {
+        let num = num.clone();
+        actix_rt::System::new().block_on(async {
+            let server = Server::build()
+                .backlog(1)
+                .disable_signals()
+                .bind("addr1", addr1, move || {
                     let num = num.clone();
-                    let lst = net::TcpListener::bind(addr3).unwrap();
-                    cfg.bind("addr1", addr1)
-                        .unwrap()
-                        .bind("addr2", addr2)
-                        .unwrap()
-                        .listen("addr3", lst)
-                        .apply(move |rt| {
-                            let num = num.clone();
-                            rt.service("addr1", fn_service(|_| ok::<_, ()>(())));
-                            rt.service("addr3", fn_service(|_| ok::<_, ()>(())));
-                            rt.on_start(lazy(move |_| {
-                                let _ = num.fetch_add(1, Relaxed);
-                            }))
-                        })
+                    fn_factory(move || {
+                        let num = num.clone();
+                        async move { Ok::<_, ()>(TestService(num)) }
+                    })
+                })
+                .unwrap()
+                .bind("addr2", addr2, move || {
+                    let num2 = num2.clone();
+                    fn_factory(move || {
+                        let num2 = num2.clone();
+                        async move { Ok::<_, ()>(TestService(num2)) }
+                    })
                 })
                 .unwrap()
                 .workers(1)
-                .run()
-        }));
+                .run();
 
-        let _ = tx.send((srv, actix_rt::System::current()));
-        let _ = sys.run();
+            let _ = tx.send((server.clone(), actix_rt::System::current()));
+            server.await
+        })
     });
-    let (_, sys) = rx.recv().unwrap();
-    thread::sleep(time::Duration::from_millis(500));
 
-    assert!(net::TcpStream::connect(addr1).is_ok());
-    assert!(net::TcpStream::connect(addr2).is_ok());
-    assert!(net::TcpStream::connect(addr3).is_ok());
-    assert_eq!(num.load(Relaxed), 1);
+    let (server, sys) = rx.recv().unwrap();
+
+    for _ in 0..5 {
+        TcpStream::connect(addr1)
+            .await
+            .unwrap()
+            .shutdown()
+            .await
+            .unwrap();
+        TcpStream::connect(addr2)
+            .await
+            .unwrap()
+            .shutdown()
+            .await
+            .unwrap();
+    }
+
+    sleep(Duration::from_secs(3)).await;
+
+    assert!(num_clone.load(Ordering::SeqCst) > 5);
+    assert!(num2_clone.load(Ordering::SeqCst) > 5);
+
     sys.stop();
-    let _ = h.join();
+    let _ = server.stop(false);
+    let _ = h.join().unwrap();
+}
+
+#[ignore]
+#[actix_rt::test]
+async fn worker_restart() {
+    use actix_service::{Service, ServiceFactory};
+    use futures_core::future::LocalBoxFuture;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    struct TestServiceFactory(Arc<AtomicUsize>);
+
+    impl ServiceFactory<TcpStream> for TestServiceFactory {
+        type Response = ();
+        type Error = ();
+        type Config = ();
+        type Service = TestService;
+        type InitError = ();
+        type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
+
+        fn new_service(&self, _: Self::Config) -> Self::Future {
+            let counter = self.0.fetch_add(1, Ordering::Relaxed);
+
+            Box::pin(async move { Ok(TestService(counter)) })
+        }
+    }
+
+    struct TestService(usize);
+
+    impl Service<TcpStream> for TestService {
+        type Response = ();
+        type Error = ();
+        type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+        actix_service::always_ready!();
+
+        fn call(&self, stream: TcpStream) -> Self::Future {
+            let counter = self.0;
+
+            let mut stream = stream.into_std().unwrap();
+            use std::io::Write;
+            let str = counter.to_string();
+            let buf = str.as_bytes();
+
+            let mut written = 0;
+
+            while written < buf.len() {
+                if let Ok(n) = stream.write(&buf[written..]) {
+                    written += n;
+                }
+            }
+            stream.flush().unwrap();
+            stream.shutdown(net::Shutdown::Write).unwrap();
+
+            // force worker 2 to restart service once.
+            if counter == 2 {
+                panic!("panic on purpose")
+            } else {
+                Box::pin(async { Ok(()) })
+            }
+        }
+    }
+
+    let addr = unused_addr();
+    let (tx, rx) = mpsc::channel();
+
+    let counter = Arc::new(AtomicUsize::new(1));
+    let h = thread::spawn(move || {
+        let counter = counter.clone();
+        actix_rt::System::new().block_on(async {
+            let server = Server::build()
+                .disable_signals()
+                .bind("addr", addr, move || TestServiceFactory(counter.clone()))
+                .unwrap()
+                .workers(2)
+                .run();
+
+            let _ = tx.send((server.clone(), actix_rt::System::current()));
+            server.await
+        })
+    });
+
+    let (server, sys) = rx.recv().unwrap();
+
+    sleep(Duration::from_secs(3)).await;
+
+    let mut buf = [0; 8];
+
+    // worker 1 would not restart and return it's id consistently.
+    let mut stream = TcpStream::connect(addr).await.unwrap();
+    let n = stream.read(&mut buf).await.unwrap();
+    let id = String::from_utf8_lossy(&buf[0..n]);
+    assert_eq!("1", id);
+    stream.shutdown().await.unwrap();
+
+    // worker 2 dead after return response.
+    let mut stream = TcpStream::connect(addr).await.unwrap();
+    let n = stream.read(&mut buf).await.unwrap();
+    let id = String::from_utf8_lossy(&buf[0..n]);
+    assert_eq!("2", id);
+    stream.shutdown().await.unwrap();
+
+    // request to worker 1
+    let mut stream = TcpStream::connect(addr).await.unwrap();
+    let n = stream.read(&mut buf).await.unwrap();
+    let id = String::from_utf8_lossy(&buf[0..n]);
+    assert_eq!("1", id);
+    stream.shutdown().await.unwrap();
+
+    // TODO: Remove sleep if it can pass CI.
+    sleep(Duration::from_secs(3)).await;
+
+    // worker 2 restarting and work goes to worker 1.
+    let mut stream = TcpStream::connect(addr).await.unwrap();
+    let n = stream.read(&mut buf).await.unwrap();
+    let id = String::from_utf8_lossy(&buf[0..n]);
+    assert_eq!("1", id);
+    stream.shutdown().await.unwrap();
+
+    // TODO: Remove sleep if it can pass CI.
+    sleep(Duration::from_secs(3)).await;
+
+    // worker 2 restarted but worker 1 was still the next to accept connection.
+    let mut stream = TcpStream::connect(addr).await.unwrap();
+    let n = stream.read(&mut buf).await.unwrap();
+    let id = String::from_utf8_lossy(&buf[0..n]);
+    assert_eq!("1", id);
+    stream.shutdown().await.unwrap();
+
+    // TODO: Remove sleep if it can pass CI.
+    sleep(Duration::from_secs(3)).await;
+
+    // worker 2 accept connection again but it's id is 3.
+    let mut stream = TcpStream::connect(addr).await.unwrap();
+    let n = stream.read(&mut buf).await.unwrap();
+    let id = String::from_utf8_lossy(&buf[0..n]);
+    assert_eq!("3", id);
+    stream.shutdown().await.unwrap();
+
+    sys.stop();
+    let _ = server.stop(false);
+    let _ = h.join().unwrap();
 }

actix-service/CHANGES.md

@@ -3,6 +3,12 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 2.0.0 - 2021-04-16
+* Removed pipeline and related structs/functions. [#335]
+
+[#335]: https://github.com/actix/actix-net/pull/335
+
+
 ## 2.0.0-beta.5 - 2021-03-15
 * Add default `Service` trait impl for `Rc<S: Service>` and `&S: Service`. [#288]
 * Add `boxed::rc_service` function for constructing `boxed::RcService` type [#290]

actix-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-service"
-version = "2.0.0-beta.5"
+version = "2.0.0"
 authors = [
     "Nikolay Kim <fafhrd91@gmail.com>",
     "Rob Ede <robjtede@icloud.com>",
@@ -8,11 +8,8 @@ authors = [
 ]
 description = "Service trait and combinators for representing asynchronous request/response operations."
 keywords = ["network", "framework", "async", "futures", "service"]
-homepage = "https://actix.rs"
-repository = "https://github.com/actix/actix-net.git"
-documentation = "https://docs.rs/actix-service"
-readme = "README.md"
 categories = ["network-programming", "asynchronous"]
+repository = "https://github.com/actix/actix-net"
 license = "MIT OR Apache-2.0"
 edition = "2018"
 
@@ -22,8 +19,10 @@ path = "src/lib.rs"
 
 [dependencies]
 futures-core = { version = "0.3.7", default-features = false }
+paste = "1"
 pin-project-lite = "0.2"
 
 [dev-dependencies]
 actix-rt = "2.0.0"
+actix-utils = "3.0.0"
 futures-util = { version = "0.3.7", default-features = false }

actix-service/README.md

@@ -3,10 +3,10 @@
 > Service trait and combinators for representing asynchronous request/response operations.
 
 [![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.5)](https://docs.rs/actix-service/2.0.0-beta.5)
+[![Documentation](https://docs.rs/actix-service/badge.svg?version=2.0.0)](https://docs.rs/actix-service/2.0.0)
 [![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.5/status.svg)](https://deps.rs/crate/actix-service/2.0.0-beta.5)
+[![Dependency Status](https://deps.rs/crate/actix-service/2.0.0/status.svg)](https://deps.rs/crate/actix-service/2.0.0)
 ![Download](https://img.shields.io/crates/d/actix-service.svg)
 [![Chat on Discord](https://img.shields.io/discord/771444961383153695?label=chat&logo=discord)](https://discord.gg/NWpN5mmg3x)
 

actix-service/src/and_then.rs

@@ -11,11 +11,11 @@ use pin_project_lite::pin_project;
 
 use super::{Service, ServiceFactory};
 
-/// Service for the `and_then` combinator, chaining a computation onto the end
-/// of another service which completes successfully.
+/// Service for the `and_then` combinator, chaining a computation onto the end of another service
+/// which completes successfully.
 ///
 /// This is created by the `Pipeline::and_then` method.
-pub(crate) struct AndThenService<A, B, Req>(Rc<(A, B)>, PhantomData<Req>);
+pub struct AndThenService<A, B, Req>(Rc<(A, B)>, PhantomData<Req>);
 
 impl<A, B, Req> AndThenService<A, B, Req> {
     /// Create new `AndThen` combinator
@@ -64,7 +64,7 @@ where
 }
 
 pin_project! {
-    pub(crate) struct AndThenServiceResponse<A, B, Req>
+    pub struct AndThenServiceResponse<A, B, Req>
     where
         A: Service<Req>,
         B: Service<A::Response, Error = A::Error>,
@@ -117,7 +117,7 @@ where
 }
 
 /// `.and_then()` service factory combinator
-pub(crate) struct AndThenServiceFactory<A, B, Req>
+pub struct AndThenServiceFactory<A, B, Req>
 where
     A: ServiceFactory<Req>,
     A::Config: Clone,
@@ -200,7 +200,7 @@ where
 }
 
 pin_project! {
-    pub(crate) struct AndThenServiceFactoryResponse<A, B, Req>
+    pub struct AndThenServiceFactoryResponse<A, B, Req>
     where
         A: ServiceFactory<Req>,
         B: ServiceFactory<A::Response>,
@@ -272,7 +272,9 @@ mod tests {
     use futures_util::future::lazy;
 
     use crate::{
-        fn_factory, ok, pipeline, pipeline_factory, ready, Ready, Service, ServiceFactory,
+        fn_factory, ok,
+        pipeline::{pipeline, pipeline_factory},
+        ready, Ready, Service, ServiceFactory,
     };
 
     struct Srv1(Rc<Cell<usize>>);

actix-service/src/apply.rs

@@ -214,7 +214,11 @@ mod tests {
     use futures_util::future::lazy;
 
     use super::*;
-    use crate::{ok, pipeline, pipeline_factory, Ready, Service, ServiceFactory};
+    use crate::{
+        ok,
+        pipeline::{pipeline, pipeline_factory},
+        Ready, Service, ServiceFactory,
+    };
 
     #[derive(Clone)]
     struct Srv;

actix-service/src/boxed.rs

@@ -3,26 +3,30 @@
 use alloc::{boxed::Box, rc::Rc};
 use core::{future::Future, pin::Pin};
 
+use paste::paste;
+
 use crate::{Service, ServiceFactory};
 
-/// A boxed future without a Send bound or lifetime parameters.
+/// A boxed future with no send bound or lifetime parameters.
 pub type BoxFuture<T> = Pin<Box<dyn Future<Output = T>>>;
 
 macro_rules! service_object {
     ($name: ident, $type: tt, $fn_name: ident) => {
-        /// Type alias for service trait object.
-        pub type $name<Req, Res, Err> = $type<
-            dyn Service<Req, Response = Res, Error = Err, Future = BoxFuture<Result<Res, Err>>>,
-        >;
+        paste! {
+            #[doc = "Type alias for service trait object using `" $type "`."]
+            pub type $name<Req, Res, Err> = $type<
+                dyn Service<Req, Response = Res, Error = Err, Future = BoxFuture<Result<Res, Err>>>,
+            >;
 
-        /// Create service trait object.
-        pub fn $fn_name<S, Req>(service: S) -> $name<Req, S::Response, S::Error>
-        where
-            S: Service<Req> + 'static,
-            Req: 'static,
-            S::Future: 'static,
-        {
-            $type::new(ServiceWrapper::new(service))
+            #[doc = "Wraps service as a trait object using [`" $name "`]."]
+            pub fn $fn_name<S, Req>(service: S) -> $name<Req, S::Response, S::Error>
+            where
+                S: Service<Req> + 'static,
+                Req: 'static,
+                S::Future: 'static,
+            {
+                $type::new(ServiceWrapper::new(service))
+            }
         }
     };
 }
@@ -56,10 +60,10 @@ where
     }
 }
 
-/// Wrapper for a service factory trait object that will produce a boxed trait object service.
+/// Wrapper for a service factory that will map it's services to boxed trait object services.
 pub struct BoxServiceFactory<Cfg, Req, Res, Err, InitErr>(Inner<Cfg, Req, Res, Err, InitErr>);
 
-/// Create service factory trait object.
+/// Wraps a service factory that returns service trait objects.
 pub fn factory<SF, Req>(
     factory: SF,
 ) -> BoxServiceFactory<SF::Config, Req, SF::Response, SF::Error, SF::InitError>

actix-service/src/ext.rs

@@ -1,8 +1,12 @@
 use crate::{
-    map::Map, map_err::MapErr, transform_err::TransformMapInitErr, Service, ServiceFactory,
-    Transform,
+    and_then::{AndThenService, AndThenServiceFactory},
+    map::Map,
+    map_err::MapErr,
+    transform_err::TransformMapInitErr,
+    IntoService, IntoServiceFactory, Service, ServiceFactory, Transform,
 };
 
+/// An extension trait for [`Service`]s that provides a variety of convenient adapters.
 pub trait ServiceExt<Req>: Service<Req> {
     /// Map this service's output to a different type, returning a new service
     /// of the resulting type.
@@ -36,10 +40,27 @@ pub trait ServiceExt<Req>: Service<Req> {
     {
         MapErr::new(self, f)
     }
+
+    /// Call another service after call to this one has resolved successfully.
+    ///
+    /// This function can be used to chain two services together and ensure that the second service
+    /// isn't called until call to the fist service have finished. Result of the call to the first
+    /// service is used as an input parameter for the second service's call.
+    ///
+    /// Note that this function consumes the receiving service and returns a wrapped version of it.
+    fn and_then<I, S1>(self, service: I) -> AndThenService<Self, S1, Req>
+    where
+        Self: Sized,
+        I: IntoService<S1, Self::Response>,
+        S1: Service<Self::Response, Error = Self::Error>,
+    {
+        AndThenService::new(self, service.into_service())
+    }
 }
 
 impl<S, Req> ServiceExt<Req> for S where S: Service<Req> {}
 
+/// An extension trait for [`ServiceFactory`]s that provides a variety of convenient adapters.
 pub trait ServiceFactoryExt<Req>: ServiceFactory<Req> {
     /// Map this service's output to a different type, returning a new service
     /// of the resulting type.
@@ -68,10 +89,27 @@ pub trait ServiceFactoryExt<Req>: ServiceFactory<Req> {
     {
         crate::map_init_err::MapInitErr::new(self, f)
     }
+
+    /// Call another service after call to this one has resolved successfully.
+    fn and_then<I, SF1>(self, factory: I) -> AndThenServiceFactory<Self, SF1, Req>
+    where
+        Self: Sized,
+        Self::Config: Clone,
+        I: IntoServiceFactory<SF1, Self::Response>,
+        SF1: ServiceFactory<
+            Self::Response,
+            Config = Self::Config,
+            Error = Self::Error,
+            InitError = Self::InitError,
+        >,
+    {
+        AndThenServiceFactory::new(self, factory.into_factory())
+    }
 }
 
 impl<SF, Req> ServiceFactoryExt<Req> for SF where SF: ServiceFactory<Req> {}
 
+/// An extension trait for [`Transform`]s that provides a variety of convenient adapters.
 pub trait TransformExt<S, Req>: Transform<S, Req> {
     /// Return a new `Transform` whose init error is mapped to to a different type.
     fn map_init_err<F, E>(self, f: F) -> TransformMapInitErr<Self, S, Req, F, E>

actix-service/src/lib.rs

@@ -1,7 +1,8 @@
 //! See [`Service`] docs for information on this crate's foundational trait.
 
 #![no_std]
-#![deny(rust_2018_idioms, nonstandard_style)]
+#![deny(rust_2018_idioms, nonstandard_style, future_incompatible)]
+#![warn(missing_docs)]
 #![allow(clippy::type_complexity)]
 #![doc(html_logo_url = "https://actix.rs/img/logo.png")]
 #![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
@@ -37,7 +38,6 @@ pub use self::apply_cfg::{apply_cfg, apply_cfg_factory};
 pub use self::ext::{ServiceExt, ServiceFactoryExt, TransformExt};
 pub use self::fn_service::{fn_factory, fn_factory_with_config, fn_service};
 pub use self::map_config::{map_config, unit_config};
-pub use self::pipeline::{pipeline, pipeline_factory, Pipeline, PipelineFactory};
 pub use self::transform::{apply, ApplyTransform, Transform};
 
 #[allow(unused_imports)]
@@ -53,8 +53,14 @@ use self::ready::{err, ok, ready, Ready};
 /// async fn(Request) -> Result<Response, Err>
 /// ```
 ///
-/// The `Service` trait just generalizes this form where each parameter is described as an
-/// associated type on the trait. Services can also have mutable state that influence computation.
+/// The `Service` trait just generalizes this form. Requests are defined as a generic type parameter
+/// and responses and other details are defined as associated types on the trait impl. Notice that
+/// this design means that services can receive many request types and converge them to a single
+/// response type.
+///
+/// Services can also have mutable state that influence computation by using a `Cell`, `RefCell`
+/// or `Mutex`. Services intentionally do not take `&mut self` to reduce overhead in the
+/// common cases.
 ///
 /// `Service` provides a symmetric and uniform API; the same abstractions can be used to represent
 /// both clients and servers. Services describe only _transformation_ operations which encourage
@@ -64,11 +70,10 @@ use self::ready::{err, ok, ready, Ready};
 /// ```ignore
 /// struct MyService;
 ///
-/// impl Service for MyService {
-///      type Request = u8;
+/// impl Service<u8> for MyService {
 ///      type Response = u64;
 ///      type Error = MyError;
-///      type Future = Pin<Box<Future<Output=Result<Self::Response, Self::Error>>>>;
+///      type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
 ///
 ///      fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { ... }
 ///
@@ -77,10 +82,13 @@ use self::ready::{err, ok, ready, Ready};
 /// ```
 ///
 /// Sometimes it is not necessary to implement the Service trait. For example, the above service
-/// could be rewritten as a simple function and passed to [fn_service](fn_service()).
+/// could be rewritten as a simple function and passed to [`fn_service`](fn_service()).
 ///
 /// ```ignore
 /// async fn my_service(req: u8) -> Result<u64, MyError>;
+///
+/// let svc = fn_service(my_service)
+/// svc.call(123)
 /// ```
 pub trait Service<Req> {
     /// Responses given by the service.
@@ -94,13 +102,12 @@ pub trait Service<Req> {
 
     /// Returns `Ready` when the service is able to process requests.
     ///
-    /// If the service is at capacity, then `Pending` is returned and the task
-    /// is notified when the service becomes ready again. This function is
-    /// expected to be called while on a task.
+    /// If the service is at capacity, then `Pending` is returned and the task is notified when the
+    /// service becomes ready again. This function is expected to be called while on a task.
     ///
-    /// This is a **best effort** implementation. False positives are permitted.
-    /// It is permitted for the service to return `Ready` from a `poll_ready`
-    /// call and the next invocation of `call` results in an error.
+    /// This is a best effort implementation. False positives are permitted. It is permitted for
+    /// the service to return `Ready` from a `poll_ready` call and the next invocation of `call`
+    /// results in an error.
     ///
     /// # Notes
     /// 1. `poll_ready` might be called on a different task to `call`.
@@ -109,25 +116,26 @@ pub trait Service<Req> {
 
     /// Process the request and return the response asynchronously.
     ///
-    /// This function is expected to be callable off task. As such,
-    /// implementations should take care to not call `poll_ready`. If the
-    /// service is at capacity and the request is unable to be handled, the
-    /// returned `Future` should resolve to an error.
+    /// This function is expected to be callable off-task. As such, implementations of `call` should
+    /// take care to not call `poll_ready`. If the service is at capacity and the request is unable
+    /// to be handled, the returned `Future` should resolve to an error.
     ///
-    /// Calling `call` without calling `poll_ready` is permitted. The
-    /// implementation must be resilient to this fact.
+    /// Invoking `call` without first invoking `poll_ready` is permitted. Implementations must be
+    /// resilient to this fact.
     fn call(&self, req: Req) -> Self::Future;
 }
 
 /// Factory for creating `Service`s.
 ///
-/// Acts as a service factory. This is useful for cases where new `Service`s
-/// must be produced. One case is a TCP server listener. The listener
-/// accepts new TCP streams, obtains a new `Service` using the
-/// `ServiceFactory` trait, and uses the new `Service` to process inbound
-/// requests on that new TCP stream.
+/// This is useful for cases where new `Service`s must be produced. One case is a TCP
+/// server listener: a listener accepts new connections, constructs a new `Service` for each using
+/// the `ServiceFactory` trait, and uses the new `Service` to process inbound requests on that new
+/// connection.
 ///
 /// `Config` is a service factory configuration type.
+///
+/// Simple factories may be able to use [`fn_factory`] or [`fn_factory_with_config`] to
+/// reduce boilerplate.
 pub trait ServiceFactory<Req> {
     /// Responses given by the created services.
     type Response;
@@ -144,7 +152,7 @@ pub trait ServiceFactory<Req> {
     /// Errors potentially raised while building a service.
     type InitError;
 
-    /// The future of the `Service` instance.
+    /// The future of the `Service` instance.g
     type Future: Future<Output = Result<Self::Service, Self::InitError>>;
 
     /// Create and return a new service asynchronously.

actix-service/src/macros.rs

@@ -1,6 +1,6 @@
-/// A boilerplate implementation of [`Service::poll_ready`] that always signals readiness.
+/// An implementation of [`poll_ready`]() that always signals readiness.
 ///
-/// [`Service::poll_ready`]: crate::Service::poll_ready
+/// [`poll_ready`]: crate::Service::poll_ready
 ///
 /// # Examples
 /// ```no_run
@@ -34,12 +34,12 @@ macro_rules! always_ready {
     };
 }
 
-/// A boilerplate implementation of [`Service::poll_ready`] that forwards readiness checks to a
+/// An implementation of [`poll_ready`] that forwards readiness checks to a
 /// named struct field.
 ///
 /// Tuple structs are not supported.
 ///
-/// [`Service::poll_ready`]: crate::Service::poll_ready
+/// [`poll_ready`]: crate::Service::poll_ready
 ///
 /// # Examples
 /// ```no_run

actix-service/src/pipeline.rs

@@ -1,3 +1,6 @@
+// TODO: see if pipeline is necessary
+#![allow(dead_code)]
+
 use core::{
     marker::PhantomData,
     task::{Context, Poll},
@@ -11,7 +14,7 @@ use crate::then::{ThenService, ThenServiceFactory};
 use crate::{IntoService, IntoServiceFactory, Service, ServiceFactory};
 
 /// Construct new pipeline with one service in pipeline chain.
-pub fn pipeline<I, S, Req>(service: I) -> Pipeline<S, Req>
+pub(crate) fn pipeline<I, S, Req>(service: I) -> Pipeline<S, Req>
 where
     I: IntoService<S, Req>,
     S: Service<Req>,
@@ -23,7 +26,7 @@ where
 }
 
 /// Construct new pipeline factory with one service factory.
-pub fn pipeline_factory<I, SF, Req>(factory: I) -> PipelineFactory<SF, Req>
+pub(crate) fn pipeline_factory<I, SF, Req>(factory: I) -> PipelineFactory<SF, Req>
 where
     I: IntoServiceFactory<SF, Req>,
     SF: ServiceFactory<Req>,
@@ -35,7 +38,7 @@ where
 }
 
 /// Pipeline service - pipeline allows to compose multiple service into one service.
-pub struct Pipeline<S, Req> {
+pub(crate) struct Pipeline<S, Req> {
     service: S,
     _phantom: PhantomData<Req>,
 }
@@ -157,7 +160,7 @@ impl<S: Service<Req>, Req> Service<Req> for Pipeline<S, Req> {
 }
 
 /// Pipeline factory
-pub struct PipelineFactory<SF, Req> {
+pub(crate) struct PipelineFactory<SF, Req> {
     factory: SF,
     _phantom: PhantomData<Req>,
 }

actix-service/src/then.rs

@@ -246,7 +246,11 @@ mod tests {
 
     use futures_util::future::lazy;
 
-    use crate::{err, ok, pipeline, pipeline_factory, ready, Ready, Service, ServiceFactory};
+    use crate::{
+        err, ok,
+        pipeline::{pipeline, pipeline_factory},
+        ready, Ready, Service, ServiceFactory,
+    };
 
     #[derive(Clone)]
     struct Srv1(Rc<Cell<usize>>);

actix-service/src/transform.rs

@@ -21,13 +21,12 @@ where
     ApplyTransform::new(t, factory.into_factory())
 }
 
-/// The `Transform` trait defines the interface of a service factory that wraps inner service
-/// during construction.
+/// Defines the interface of a service factory that wraps inner service during construction.
 ///
-/// Transform(middleware) wraps inner service and runs during inbound and/or outbound processing in
-/// the request/response lifecycle. It may modify request and/or response.
+/// Transformers wrap an inner service and runs during inbound and/or outbound processing in the
+/// service lifecycle. It may modify request and/or response.
 ///
-/// For example, timeout transform:
+/// For example, a timeout service wrapper:
 ///
 /// ```ignore
 /// pub struct Timeout<S> {
@@ -35,18 +34,14 @@ where
 ///     timeout: Duration,
 /// }
 ///
-/// impl<S> Service for Timeout<S>
-/// where
-///     S: Service,
-/// {
-///     type Request = S::Request;
+/// impl<S: Service<Req>, Req> Service<Req> for Timeout<S> {
 ///     type Response = S::Response;
 ///     type Error = TimeoutError<S::Error>;
 ///     type Future = TimeoutServiceResponse<S>;
 ///
 ///     actix_service::forward_ready!(service);
 ///
-///     fn call(&self, req: S::Request) -> Self::Future {
+///     fn call(&self, req: Req) -> Self::Future {
 ///         TimeoutServiceResponse {
 ///             fut: self.service.call(req),
 ///             sleep: Sleep::new(clock::now() + self.timeout),
@@ -55,26 +50,22 @@ where
 /// }
 /// ```
 ///
-/// Timeout service in above example is decoupled from underlying service implementation and could
-/// be applied to any service.
+/// This wrapper service is decoupled from the underlying service implementation and could be
+/// applied to any service.
 ///
-/// The `Transform` trait defines the interface of a Service factory. `Transform` is often
+/// The `Transform` trait defines the interface of a service wrapper. `Transform` is often
 /// implemented for middleware, defining how to construct a middleware Service. A Service that is
 /// constructed by the factory takes the Service that follows it during execution as a parameter,
 /// assuming ownership of the next Service.
 ///
-/// Factory for `Timeout` middleware from the above example could look like this:
+/// A transform for the `Timeout` middleware could look like this:
 ///
 /// ```ignore
 /// pub struct TimeoutTransform {
 ///     timeout: Duration,
 /// }
 ///
-/// impl<S> Transform<S> for TimeoutTransform
-/// where
-///     S: Service,
-/// {
-///     type Request = S::Request;
+/// impl<S: Service<Req>, Req> Transform<S, Req> for TimeoutTransform {
 ///     type Response = S::Response;
 ///     type Error = TimeoutError<S::Error>;
 ///     type InitError = S::Error;
@@ -82,7 +73,7 @@ where
 ///     type Future = Ready<Result<Self::Transform, Self::InitError>>;
 ///
 ///     fn new_transform(&self, service: S) -> Self::Future {
-///         ready(Ok(TimeoutService {
+///         ready(Ok(Timeout {
 ///             service,
 ///             timeout: self.timeout,
 ///         }))
@@ -227,3 +218,53 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use core::time::Duration;
+
+    use actix_utils::future::{ready, Ready};
+
+    use super::*;
+    use crate::Service;
+
+    // pseudo-doctest for Transform trait
+    pub struct TimeoutTransform {
+        timeout: Duration,
+    }
+
+    // pseudo-doctest for Transform trait
+    impl<S: Service<Req>, Req> Transform<S, Req> for TimeoutTransform {
+        type Response = S::Response;
+        type Error = S::Error;
+        type InitError = S::Error;
+        type Transform = Timeout<S>;
+        type Future = Ready<Result<Self::Transform, Self::InitError>>;
+
+        fn new_transform(&self, service: S) -> Self::Future {
+            ready(Ok(Timeout {
+                service,
+                _timeout: self.timeout,
+            }))
+        }
+    }
+
+    // pseudo-doctest for Transform trait
+    pub struct Timeout<S> {
+        service: S,
+        _timeout: Duration,
+    }
+
+    // pseudo-doctest for Transform trait
+    impl<S: Service<Req>, Req> Service<Req> for Timeout<S> {
+        type Response = S::Response;
+        type Error = S::Error;
+        type Future = S::Future;
+
+        crate::forward_ready!(service);
+
+        fn call(&self, req: Req) -> Self::Future {
+            self.service.call(req)
+        }
+    }
+}

actix-tls/Cargo.toml

@@ -42,8 +42,8 @@ uri = ["http"]
 [dependencies]
 actix-codec = "0.4.0-beta.1"
 actix-rt = { version = "2.2.0", default-features = false }
-actix-service = "2.0.0-beta.5"
-actix-utils = "3.0.0-beta.2"
+actix-service = "2.0.0"
+actix-utils = "3.0.0"
 
 derive_more = "0.99.5"
 futures-core = { version = "0.3.7", default-features = false, features = ["alloc"] }
@@ -64,7 +64,7 @@ tokio-native-tls = { version = "0.3", optional = true }
 
 [dev-dependencies]
 actix-rt = "2.2.0"
-actix-server = "2.0.0-beta.3"
+actix-server = "2.0.0-beta.5"
 bytes = "1"
 env_logger = "0.8"
 futures-util = { version = "0.3.7", default-features = false, features = ["sink"] }

actix-tls/examples/tcp-rustls.rs

@@ -31,7 +31,7 @@ use std::{
 
 use actix_rt::net::TcpStream;
 use actix_server::Server;
-use actix_service::pipeline_factory;
+use actix_service::ServiceFactoryExt as _;
 use actix_tls::accept::rustls::{Acceptor as RustlsAcceptor, TlsStream};
 use futures_util::future::ok;
 use log::info;
@@ -39,14 +39,9 @@ use rustls::{
     internal::pemfile::certs, internal::pemfile::rsa_private_keys, NoClientAuth, ServerConfig,
 };
 
-#[derive(Debug)]
-struct ServiceState {
-    num: Arc<AtomicUsize>,
-}
-
 #[actix_rt::main]
 async fn main() -> io::Result<()> {
-    env::set_var("RUST_LOG", "actix=trace,basic=trace");
+    env::set_var("RUST_LOG", "info");
     env_logger::init();
 
     let mut tls_config = ServerConfig::new(NoClientAuth::new());
@@ -73,7 +68,8 @@ async fn main() -> io::Result<()> {
             let count = Arc::clone(&count);
 
             // Set up TLS service factory
-            pipeline_factory(tls_acceptor.clone())
+            tls_acceptor
+                .clone()
                 .map_err(|err| println!("Rustls error: {:?}", err))
                 .and_then(move |stream: TlsStream<TcpStream>| {
                     let num = count.fetch_add(1, Ordering::Relaxed);

actix-tls/src/connect/resolve.rs

@@ -56,7 +56,7 @@ pub enum Resolver {
 /// An interface for custom async DNS resolvers.
 ///
 /// # Usage
-/// ```rust
+/// ```
 /// use std::net::SocketAddr;
 ///
 /// use actix_tls::connect::{Resolve, Resolver};

actix-tracing/Cargo.toml

@@ -16,9 +16,9 @@ name = "actix_tracing"
 path = "src/lib.rs"
 
 [dependencies]
-actix-service = "2.0.0-beta.5"
+actix-service = "2.0.0"
+actix-utils = "3.0.0"
 
-futures-util = { version = "0.3.7", default-features = false }
 tracing = "0.1"
 tracing-futures = "0.2"
 

actix-tracing/src/lib.rs

@@ -9,7 +9,7 @@ use core::marker::PhantomData;
 use actix_service::{
     apply, ApplyTransform, IntoServiceFactory, Service, ServiceFactory, Transform,
 };
-use futures_util::future::{ok, Either, Ready};
+use actix_utils::future::{ok, Either, Ready};
 use tracing_futures::{Instrument, Instrumented};
 
 /// A `Service` implementation that automatically enters/exits tracing spans
@@ -48,9 +48,9 @@ where
             .clone()
             .map(|span| tracing::span!(parent: &span, tracing::Level::INFO, "future"))
         {
-            Either::Right(fut.instrument(span))
+            Either::right(fut.instrument(span))
         } else {
-            Either::Left(fut)
+            Either::left(fut)
         }
     }
 }

actix-utils/CHANGES.md

@@ -3,6 +3,16 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 3.0.0 - 2021-04-16
+* No significant changes from `3.0.0-beta.4`.
+
+
+## 3.0.0-beta.4 - 2021-04-01
+* Add `future::Either` type. [#305]
+
+[#305]: https://github.com/actix/actix-net/pull/305
+
+
 ## 3.0.0-beta.3 - 2021-04-01
 * Moved `mpsc` to own crate `local-channel`. [#301]
 * Moved `task::LocalWaker` to own crate `local-waker`. [#301]

actix-utils/Cargo.toml

@@ -1,14 +1,14 @@
 [package]
 name = "actix-utils"
-version = "3.0.0-beta.3"
+version = "3.0.0"
 authors = [
     "Nikolay Kim <fafhrd91@gmail.com>",
     "Rob Ede <robjtede@icloud.com>",
 ]
-description = "Utilities for the Actix ecosystem"
+description = "Various utilities used in the Actix ecosystem"
 keywords = ["network", "framework", "async", "futures"]
-repository = "https://github.com/actix/actix-net.git"
 categories = ["network-programming", "asynchronous"]
+repository = "https://github.com/actix/actix-net"
 license = "MIT OR Apache-2.0"
 edition = "2018"
 
@@ -17,6 +17,7 @@ name = "actix_utils"
 path = "src/lib.rs"
 
 [dependencies]
+pin-project-lite = "0.2"
 local-waker = "0.1"
 
 [dev-dependencies]

actix-utils/src/future/either.rs

@@ -0,0 +1,91 @@
+//! A symmetric either future.
+
+use core::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use pin_project_lite::pin_project;
+
+pin_project! {
+    /// Combines two different futures that have the same output type.
+    ///
+    /// Construct variants with [`Either::left`] and [`Either::right`].
+    ///
+    /// # Examples
+    /// ```
+    /// use actix_utils::future::{ready, Ready, Either};
+    ///
+    /// # async fn run() {
+    /// let res = Either::<_, Ready<usize>>::left(ready(42));
+    /// assert_eq!(res.await, 42);
+    ///
+    /// let res = Either::<Ready<usize>, _>::right(ready(43));
+    /// assert_eq!(res.await, 43);
+    /// # }
+    /// ```
+    #[project = EitherProj]
+    #[derive(Debug, Clone)]
+    pub enum Either<L, R> {
+        /// A value of type `L`.
+        #[allow(missing_docs)]
+        Left { #[pin] value: L },
+
+        /// A value of type `R`.
+        #[allow(missing_docs)]
+        Right { #[pin] value: R },
+    }
+}
+
+impl<L, R> Either<L, R> {
+    /// Creates new `Either` using left variant.
+    pub fn left(value: L) -> Either<L, R> {
+        Either::Left { value }
+    }
+
+    /// Creates new `Either` using right variant.
+    pub fn right(value: R) -> Either<L, R> {
+        Either::Right { value }
+    }
+}
+
+impl<T> Either<T, T> {
+    /// Unwraps into inner value when left and right have a common type.
+    pub fn into_inner(self) -> T {
+        match self {
+            Either::Left { value } => value,
+            Either::Right { value } => value,
+        }
+    }
+}
+
+impl<L, R> Future for Either<L, R>
+where
+    L: Future,
+    R: Future<Output = L::Output>,
+{
+    type Output = L::Output;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match self.project() {
+            EitherProj::Left { value } => value.poll(cx),
+            EitherProj::Right { value } => value.poll(cx),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::future::{ready, Ready};
+
+    #[actix_rt::test]
+    async fn test_either() {
+        let res = Either::<_, Ready<usize>>::left(ready(42));
+        assert_eq!(res.await, 42);
+
+        let res = Either::<Ready<usize>, _>::right(ready(43));
+        assert_eq!(res.await, 43);
+    }
+}

actix-utils/src/future/mod.rs

@@ -1,7 +1,9 @@
 //! Asynchronous values.
 
+mod either;
 mod poll_fn;
 mod ready;
 
+pub use self::either::Either;
 pub use self::poll_fn::{poll_fn, PollFn};
 pub use self::ready::{err, ok, ready, Ready};

actix-utils/src/future/poll_fn.rs

@@ -7,7 +7,7 @@ use core::{
     task::{Context, Poll},
 };
 
-/// Create a future driven by the provided function that receives a task context.
+/// Creates a future driven by the provided function that receives a task context.
 pub fn poll_fn<F, T>(f: F) -> PollFn<F>
 where
     F: FnMut(&mut Context<'_>) -> Poll<T>,

actix-utils/src/future/ready.rs

@@ -69,7 +69,7 @@ pub fn ready<T>(val: T) -> Ready<T> {
     Ready { val: Some(val) }
 }
 
-/// Create a future that is immediately ready with a success value.
+/// Creates a future that is immediately ready with a success value.
 ///
 /// # Examples
 /// ```no_run
@@ -84,7 +84,7 @@ pub fn ok<T, E>(val: T) -> Ready<Result<T, E>> {
     Ready { val: Some(Ok(val)) }
 }
 
-/// Create a future that is immediately ready with an error value.
+/// Creates a future that is immediately ready with an error value.
 ///
 /// # Examples
 /// ```no_run

actix-utils/src/lib.rs

@@ -1,4 +1,4 @@
-//! Various utilities for the Actix ecosystem.
+//! Various utilities used in the Actix ecosystem.
 
 #![deny(rust_2018_idioms, nonstandard_style)]
 #![warn(missing_docs)]