Switch from futures-cpupool to tokio-threadpool

This is a breaking change as the server must now be spawned on a
multi-threaded `tokio::runtime::Runtime`.  Tokio can manage the blocking
threads more efficiently than we can with a separate pool.

Author: jtgeibel

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "conduit-hyper"
-version = "0.1.3"
+version = "0.2.0-a.0"
 authors = ["Justin Geibel <[email protected]>"]
 license = "MIT OR Apache-2.0"
 description = "Host a conduit based web application on a hyper server"
@@ -11,8 +11,13 @@ edition = "2018"
 [dependencies]
 conduit = "0.8"
 futures = "0.1"
-futures-cpupool = "0.1"
 hyper = "0.12"
 http = "0.1"
 log = "0.4"
 semver = "0.5" # Must match version in conduit for now
+tokio-threadpool = "0.1.12"
+
+[dev-dependencies]
+conduit = "0.8"
+conduit-router = "0.8"
+tokio = "0.1"

src/lib.rs

@@ -9,13 +9,27 @@ use std::path::{Component, Path, PathBuf};
 use std::sync::Arc;
 
 use futures::{future, Future, Stream};
-use futures_cpupool::CpuPool;
-use hyper::{Body, Chunk, Method, Request, Response, Server, StatusCode, Version};
+use hyper::{Body, Chunk, Method, Request, Response, StatusCode, Version};
 use log::error;
 
 // Consumers of this library need access to this particular version of `semver`
 pub use semver;
 
+/// A builder for a `hyper::Server`
+#[derive(Debug)]
+pub struct Server;
+
+impl Server {
+    /// Bind a handler to an address
+    pub fn bind<H: conduit::Handler>(
+        addr: &SocketAddr,
+        handler: H,
+    ) -> hyper::Server<hyper::server::conn::AddrIncoming, Service<H>> {
+        let service = Service::new(handler);
+        hyper::Server::bind(&addr).serve(service)
+    }
+}
+
 #[derive(Debug)]
 struct Parts(http::request::Parts);
 
@@ -67,7 +81,7 @@ struct ConduitRequest {
     parts: Parts,
     path: String,
     body: Cursor<Chunk>,
-    extensions: conduit::Extensions,
+    extensions: conduit::Extensions, // makes struct non-Send
 }
 
 impl conduit::Request for ConduitRequest {
@@ -157,8 +171,34 @@ impl conduit::Request for ConduitRequest {
     }
 }
 
+/// Owned data consumed by the worker thread
+///
+/// `ConduitRequest` cannot be sent between threads, so the input data is
+/// captured on a core thread and taken by the worker thread.
+struct RequestInfo(Option<(Parts, Chunk)>);
+
+impl RequestInfo {
+    /// Save the request info that can be sent between threads
+    fn new(parts: http::request::Parts, body: Chunk) -> Self {
+        let tuple = (Parts(parts), body);
+        Self(Some(tuple))
+    }
+
+    /// Take back the request info
+    ///
+    /// Call this from the worker thread to obtain ownership of the `Send` data
+    ///
+    /// # Panics
+    ///
+    /// Panics if called more than once on a value
+    fn take(&mut self) -> (Parts, Chunk) {
+        self.0.take().expect("called take multiple times")
+    }
+}
+
 impl ConduitRequest {
-    fn new(parts: Parts, body: Chunk) -> ConduitRequest {
+    fn new(info: &mut RequestInfo) -> Self {
+        let (parts, body) = info.take();
         let path = parts.0.uri.path().to_string();
         let path = Path::new(&path);
         let path = path
@@ -183,7 +223,7 @@ impl ConduitRequest {
             .to_string_lossy()
             .to_string(); // non-Unicode is replaced with U+FFFD REPLACEMENT CHARACTER
 
-        ConduitRequest {
+        Self {
             parts,
             path,
             body: Cursor::new(body),
@@ -195,15 +235,13 @@ impl ConduitRequest {
 /// Serve a `conduit::Handler` on a thread pool
 #[derive(Debug)]
 pub struct Service<H> {
-    pool: CpuPool,
     handler: Arc<H>,
 }
 
 // #[derive(Clone)] results in cloning a ref, and not the Service
 impl<H> Clone for Service<H> {
     fn clone(&self) -> Self {
         Service {
-            pool: self.pool.clone(),
             handler: self.handler.clone(),
         }
     }
@@ -230,39 +268,32 @@ impl<H: conduit::Handler> hyper::service::Service for Service<H> {
 
     /// Returns a future which buffers the response body and then calls the conduit handler from a thread pool
     fn call(&mut self, request: Request<Self::ReqBody>) -> Self::Future {
-        let pool = self.pool.clone();
         let handler = self.handler.clone();
 
         let (parts, body) = request.into_parts();
         let response = body.concat2().and_then(move |full_body| {
-            pool.spawn_fn(move || {
-                let mut request = ConduitRequest::new(Parts(parts), full_body);
-                let response = handler
-                    .call(&mut request)
-                    .map(good_response)
-                    .unwrap_or_else(|e| error_response(e.description()));
-
-                Ok(response)
+            let mut request_info = RequestInfo::new(parts, full_body);
+            future::poll_fn(move || {
+                tokio_threadpool::blocking(|| {
+                    let mut request = ConduitRequest::new(&mut request_info);
+                    handler
+                        .call(&mut request)
+                        .map(good_response)
+                        .unwrap_or_else(|e| error_response(e.description()))
+                })
+                .map_err(|_| panic!("the threadpool shut down"))
             })
         });
         Box::new(response)
     }
 }
 
 impl<H: conduit::Handler> Service<H> {
-    /// Create a multi-threaded `Service` from a `Handler`
-    pub fn new(handler: H, threads: usize) -> Service<H> {
+    fn new(handler: H) -> Self {
         Service {
-            pool: CpuPool::new(threads),
             handler: Arc::new(handler),
         }
     }
-
-    /// Run the `Service` bound to a given `SocketAddr`
-    pub fn run(&self, addr: SocketAddr) {
-        let server = Server::bind(&addr).serve(self.clone());
-        hyper::rt::run(server.map_err(|e| error!("Server error: {}", e)));
-    }
 }
 
 /// Builds a `hyper::Response` given a `conduit:Response`

src/tests.rs

@@ -73,12 +73,25 @@ fn build_headers(msg: &str) -> HashMap<String, Vec<String>> {
     headers
 }
 
+fn build_threadpool() -> tokio::runtime::Runtime {
+    tokio::runtime::Builder::new()
+        .core_threads(1)
+        .blocking_threads(1)
+        .build()
+        .unwrap()
+}
+
 fn simulate_request<H: Handler>(handler: H) -> hyper::Response<hyper::Body> {
     use hyper::service::{NewService, Service};
 
-    let new_service = super::Service::new(handler, 1);
-    let mut service = new_service.new_service().wait().unwrap();
-    service.call(hyper::Request::default()).wait().unwrap()
+    let mut pool = build_threadpool();
+
+    pool.block_on(futures::lazy(|| {
+        let new_service = super::Service::new(handler);
+        let mut service = new_service.new_service().wait().unwrap();
+        service.call(hyper::Request::default()).wait()
+    }))
+    .unwrap()
 }
 
 fn into_chunk(resp: hyper::Response<hyper::Body>) -> hyper::Chunk {
@@ -122,19 +135,25 @@ fn recover_from_panic() {
 fn normalize_path() {
     use hyper::service::{NewService, Service};
 
-    let new_service = super::Service::new(AssertPathNormalized, 1);
-    let mut service = new_service.new_service().wait().unwrap();
-    let req = hyper::Request::put("//removed/.././.././normalized")
-        .body(hyper::Body::default())
-        .unwrap();
-    let resp = service.call(req).wait().unwrap();
-    assert_eq!(resp.status(), 200);
-    assert_eq!(resp.headers().len(), 1);
-
-    let req = hyper::Request::put("//normalized")
-        .body(hyper::Body::default())
-        .unwrap();
-    let resp = service.call(req).wait().unwrap();
-    assert_eq!(resp.status(), 200);
-    assert_eq!(resp.headers().len(), 1);
+    let mut pool = build_threadpool();
+
+    pool.block_on(futures::lazy(|| {
+        let new_service = super::Service::new(AssertPathNormalized);
+        let mut service = new_service.new_service().wait().unwrap();
+        let req = hyper::Request::put("//removed/.././.././normalized")
+            .body(hyper::Body::default())
+            .unwrap();
+        let resp = service.call(req).wait().unwrap();
+        assert_eq!(resp.status(), 200);
+        assert_eq!(resp.headers().len(), 1);
+
+        let req = hyper::Request::put("//normalized")
+            .body(hyper::Body::default())
+            .unwrap();
+        let resp = service.call(req).wait().unwrap();
+        assert_eq!(resp.status(), 200);
+        assert_eq!(resp.headers().len(), 1);
+        Ok::<_, ()>(())
+    }))
+    .unwrap()
 }