Migrate actix-threadpool to std::future (#59)

* Migrate actix-threadpool to std::future

* Cosmetic refactor

- turn log::error! into log::warn! as it doesn't throw any error
- add Clone and Copy impls for Cancelled making it cheap to operate with
- apply rustfmt

* Bump up crate version to 0.2.0 and pre-fill its changelog

* Disable patching 'actix-threadpool' crate in global workspace as unnecessary

* Revert patching and fix 'actix-rt'

actix-rt/Cargo.toml

@@ -18,7 +18,7 @@ name = "actix_rt"
 path = "src/lib.rs"
 
 [dependencies]
-actix-threadpool = "0.1.1"
+actix-threadpool = "0.2"
 futures = "0.3.1"
 
 # TODO: Replace this with dependency on tokio-runtime once it is ready

actix-rt/src/arbiter.rs

@@ -4,11 +4,11 @@ use std::collections::HashMap;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::{fmt, thread};
 use std::pin::Pin;
-use std::task::Context;
+use std::task::{Context, Poll};
 
 use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
 use futures::channel::oneshot::{channel, Canceled, Sender};
-use futures::{future, Future, Poll, FutureExt, Stream};
+use futures::{future, Future, FutureExt, Stream};
 use tokio::runtime::current_thread::spawn;
 
 use crate::builder::Builder;

actix-threadpool/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [0.2.0] - 2019-??-??
+
+### Changed
+
+* Migrate to `std::future`
+
 ## [0.1.2] - 2019-08-05
 
 ### Changed

actix-threadpool/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-threadpool"
-version = "0.1.2"
+version = "0.2.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix thread pool for sync code"
 keywords = ["actix", "network", "framework", "async", "futures"]

actix-threadpool/src/lib.rs

@@ -1,34 +1,34 @@
 //! Thread pool for blocking operations
 
-use std::future::Future;
-use std::task::{Poll,Context};
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
 
 use derive_more::Display;
 use futures::channel::oneshot;
 use parking_lot::Mutex;
 use threadpool::ThreadPool;
-use std::pin::Pin;
 
-/// Env variable for default cpu pool size
+/// 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,
-        };
+        let num = std::env::var(ENV_CPU_POOL_VAR)
+            .map_err(|_| ())
+            .and_then(|val| {
+                val.parse().map_err(|_| log::warn!(
+                    "Can not parse {} value, using default",
+                    ENV_CPU_POOL_VAR,
+                ))
+            })
+            .unwrap_or_else(|_| num_cpus::get() * 5);
         Mutex::new(
             threadpool::Builder::new()
                 .thread_name("actix-web".to_owned())
-                .num_threads(default)
+                .num_threads(num)
                 .build(),
         )
     };
@@ -40,8 +40,8 @@ thread_local! {
     };
 }
 
-/// Blocking operation execution error
-#[derive(Debug, Display)]
+/// Error of blocking operation execution being cancelled.
+#[derive(Clone, Copy, Debug, Display)]
 #[display(fmt = "Thread pool is gone")]
 pub struct Cancelled;
 
@@ -71,13 +71,11 @@ pub struct CpuFuture<I> {
 }
 
 impl<I> Future for CpuFuture<I> {
-    type Output = Result<I,Cancelled>;
+    type Output = Result<I, Cancelled>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let rx =  unsafe{ self.map_unchecked_mut(|s|&mut  s.rx)};
-        let res =  futures::ready!(rx.poll(cx));
-
+        let rx = Pin::new(&mut Pin::get_mut(self).rx);
+        let res = futures::ready!(rx.poll(cx));
         Poll::Ready(res.map_err(|_| Cancelled))
     }
-
 }