Use Rayon instead of spawn_blocking for WSGI

rwf/Cargo.toml

@@ -8,7 +8,7 @@ description = "Comprehensive framework for building web applications in Rust"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-wsgi = ["pyo3"]
+wsgi = ["pyo3", "rayon"]
 default = []
 
 [dependencies]
@@ -37,6 +37,7 @@ regex = "1"
 sha1 = "0.10"
 toml = "0.8"
 pyo3 = { version = "0.22", features = ["auto-initialize"], optional = true }
+rayon = { version = "1", optional = true }
 
 [dev-dependencies]
 tempdir = "0.3"

rwf/src/controller/wsgi.rs

@@ -3,51 +3,61 @@ use crate::http::{wsgi::WsgiRequest, Request, Response};
 
 use async_trait::async_trait;
 use pyo3::prelude::*;
+use rayon::{ThreadPool, ThreadPoolBuilder};
+use tokio::sync::oneshot::channel;
 use tokio::time::{timeout, Duration};
 
 pub struct WsgiController {
     path: &'static str,
     timeout: Duration,
+    pool: ThreadPool,
 }
 
 impl WsgiController {
     pub fn new(path: &'static str) -> Self {
         WsgiController {
             path,
             timeout: Duration::from_secs(60),
+            pool: ThreadPoolBuilder::new().num_threads(2).build().unwrap(),
         }
     }
 
     pub fn timeout(mut self, timeout: Duration) -> Self {
         self.timeout = timeout;
         self
     }
+
+    pub fn max_threads(mut self, threads: usize) -> Self {
+        self.pool = ThreadPoolBuilder::new()
+            .num_threads(threads)
+            .build()
+            .unwrap();
+        self
+    }
 }
 
 #[async_trait]
 impl Controller for WsgiController {
     async fn handle(&self, request: &Request) -> Result<Response, Error> {
         let request = WsgiRequest::from_request(request)?;
         let path = self.path;
+        let (tx, rx) = channel();
+
+        self.pool.spawn(move || {
+            let response = Python::with_gil(|py| {
+                // Import is cached.
+                let module = PyModule::import_bound(py, path).unwrap();
+                let application: Py<PyAny> = module.getattr("application").unwrap().into();
+                request.send(&application)
+            })
+            .unwrap();
+
+            tx.send(response).unwrap();
+        });
 
         // TODO: spawn blocking tasks cannot be aborted.
         // This only aborts waiting for the result to be returned.
-        match timeout(
-            self.timeout,
-            tokio::task::spawn_blocking(move || {
-                let response = Python::with_gil(|py| {
-                    // Import is cached.
-                    let module = PyModule::import_bound(py, path).unwrap();
-                    let application: Py<PyAny> = module.getattr("application").unwrap().into();
-                    request.send(&application)
-                })
-                .unwrap();
-
-                response
-            }),
-        )
-        .await?
-        {
+        match timeout(self.timeout, rx).await? {
             Ok(response) => Ok(response.to_response()?),
             Err(e) => Ok(Response::internal_error(e)),
         }