Merge pull request #3 from jtgeibel/tokio-threadpool

Switch from `futures-cpupool` to `tokio-threadpool`

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "conduit-hyper"
-version = "0.1.3"
+version = "0.2.0-a.0"
 authors = ["Justin Geibel <jtgeibel@gmail.com>"]
 license = "MIT OR Apache-2.0"
 description = "Host a conduit based web application on a hyper server"
@@ -11,8 +11,12 @@ 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-router = "0.8"
+tokio = "0.1"

examples/server.rs

@@ -0,0 +1,57 @@
+#![deny(warnings, clippy::all)]
+
+use conduit::{Handler, Request, Response};
+use conduit_hyper::Server;
+use conduit_router::RouteBuilder;
+use futures::Future;
+use tokio::runtime;
+
+use std::collections::HashMap;
+use std::io::{Cursor, Error};
+use std::thread::sleep;
+
+fn main() {
+    let app = build_conduit_handler();
+    let addr = ([127, 0, 0, 1], 12345).into();
+    let server = Server::bind(&addr, app).map_err(|e| {
+        eprintln!("server error: {}", e);
+    });
+
+    let mut rt = runtime::Builder::new()
+        // Set the max number of concurrent requests (tokio defaults to 100)
+        .blocking_threads(2)
+        .build()
+        .unwrap();
+    rt.spawn(server);
+    rt.shutdown_on_idle().wait().unwrap();
+}
+
+fn build_conduit_handler() -> impl Handler {
+    let mut router = RouteBuilder::new();
+    router.get("/", endpoint);
+    router.get("/panic", panic);
+    router
+}
+
+fn endpoint(_: &mut dyn Request) -> Result<Response, Error> {
+    let body = "Hello world!";
+
+    sleep(std::time::Duration::from_secs(2));
+
+    let mut headers = HashMap::new();
+    headers.insert(
+        "Content-Type".to_string(),
+        vec!["text/plain; charset=utf-8".to_string()],
+    );
+    headers.insert("Content-Length".to_string(), vec![body.len().to_string()]);
+    Ok(Response {
+        status: (200, "OK"),
+        headers,
+        body: Box::new(Cursor::new(body)),
+    })
+}
+
+fn panic(_: &mut dyn Request) -> Result<Response, Error> {
+    // For now, connection is immediately closed
+    panic!("message");
+}

src/lib.rs

@@ -1,5 +1,57 @@
 #![deny(warnings, clippy::all, missing_debug_implementations)]
 
+//! A wrapper for integrating `hyper 0.12` with a `conduit 0.8` blocking application stack.
+//!
+//! A `conduit::Handler` is allowed to block so the `Server` must be spawned on the (default)
+//! multi-threaded `Runtime` which allows (by default) 100 concurrent blocking threads.  Any excess
+//! requests will asynchronously wait for an available blocking thread.
+//!
+//! # Examples
+//!
+//! Try out the example with `cargo run --example server`.
+//!
+//! Typical usage:
+//!
+//! ```no_run
+//! use conduit::Handler;
+//! use conduit_hyper::Server;
+//! use futures::Future;
+//! use tokio::runtime::Runtime;
+//!
+//! fn main() {
+//!     let app = build_conduit_handler();
+//!     let addr = ([127, 0, 0, 1], 12345).into();
+//!     let server = Server::bind(&addr, app).map_err(|e| {
+//!         eprintln!("server error: {}", e);
+//!     });
+//!
+//!     let mut rt = Runtime::new().unwrap();
+//!     rt.spawn(server);
+//!     rt.shutdown_on_idle().wait().unwrap();
+//! }
+//!
+//! fn build_conduit_handler() -> impl Handler {
+//!     // ...
+//! #     Endpoint()
+//! }
+//! #
+//! # use std::{collections, error, io};
+//! #
+//! # use conduit::{Request, Response};
+//! #
+//! # struct Endpoint();
+//! #
+//! # impl Handler for Endpoint {
+//! #     fn call(&self, _: &mut dyn Request) -> Result<Response, Box<dyn error::Error + Send>> {
+//! #         Ok(Response {
+//! #             status: (200, "OK"),
+//! #             headers: collections::HashMap::new(),
+//! #             body: Box::new(io::Cursor::new("")),
+//! #         })
+//! #     }
+//! # }
+//! ```
+
 #[cfg(test)]
 mod tests;
 
@@ -9,13 +61,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 +133,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 +223,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 +275,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 +287,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 +320,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`