move threadpool to separate crate

actix-threadpool/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2019-03-28
+
+* Move threadpool to separate crate

actix-threadpool/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "actix-threadpool"
+version = "0.1.0"
+authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+description = "Actix thread pool for sync code"
+keywords = ["actix", "network", "framework", "async", "futures"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-threadpool/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+edition = "2018"
+workspace = ".."
+
+[lib]
+name = "actix_threadpool"
+path = "src/lib.rs"
+
+[dependencies]
+derive_more = "0.14"
+futures = "0.1.25"
+parking_lot = "0.7"
+lazy_static = "1.2"
+log = "0.4"
+num_cpus = "1.10"
+threadpool = "1.7"

actix-threadpool/src/lib.rs

@@ -0,0 +1,88 @@
+//! Thread pool for blocking operations
+
+use std::fmt;
+
+use derive_more::Display;
+use futures::sync::oneshot;
+use futures::{Async, Future, Poll};
+use parking_lot::Mutex;
+use threadpool::ThreadPool;
+
+/// Env variable for default cpu pool size
+const ENV_CPU_POOL_VAR: &str = "ACTIX_THREADPOOL";
+
+lazy_static::lazy_static! {
+    pub(crate) static ref DEFAULT_POOL: Mutex<ThreadPool> = {
+        let default = match std::env::var(ENV_CPU_POOL_VAR) {
+            Ok(val) => {
+                if let Ok(val) = val.parse() {
+                    val
+                } else {
+                    log::error!("Can not parse ACTIX_THREADPOOL value");
+                    num_cpus::get() * 5
+                }
+            }
+            Err(_) => num_cpus::get() * 5,
+        };
+        Mutex::new(
+            threadpool::Builder::new()
+                .thread_name("actix-web".to_owned())
+                .num_threads(default)
+                .build(),
+        )
+    };
+}
+
+thread_local! {
+    static POOL: ThreadPool = {
+        DEFAULT_POOL.lock().clone()
+    };
+}
+
+/// Blocking operation execution error
+#[derive(Debug, Display)]
+pub enum BlockingError<E: fmt::Debug> {
+    #[display(fmt = "{:?}", _0)]
+    Error(E),
+    #[display(fmt = "Thread pool is gone")]
+    Canceled,
+}
+
+/// Execute blocking function on a thread pool, returns future that resolves
+/// to result of the function execution.
+pub fn run<F, I, E>(f: F) -> CpuFuture<I, E>
+where
+    F: FnOnce() -> Result<I, E> + Send + 'static,
+    I: Send + 'static,
+    E: Send + fmt::Debug + 'static,
+{
+    let (tx, rx) = oneshot::channel();
+    POOL.with(|pool| {
+        pool.execute(move || {
+            if !tx.is_canceled() {
+                let _ = tx.send(f());
+            }
+        })
+    });
+
+    CpuFuture { rx }
+}
+
+/// Blocking operation completion future. It resolves with results
+/// of blocking function execution.
+pub struct CpuFuture<I, E> {
+    rx: oneshot::Receiver<Result<I, E>>,
+}
+
+impl<I, E: fmt::Debug> Future for CpuFuture<I, E> {
+    type Item = I;
+    type Error = BlockingError<E>;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let res = futures::try_ready!(self.rx.poll().map_err(|_| BlockingError::Canceled));
+        match res {
+            Ok(val) => Ok(Async::Ready(val)),
+            Err(err) => Err(BlockingError::Error(err)),
+        }
+    }
+}