RPC cpu pool (#6023)

* RPC cpu pool.

* introduce optional thread pool when processing RPC requests.

* Bump jsonrpc.

* Removing boxes.

* Fix CLI tests.

Cargo.lock

@@ -1061,7 +1061,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 [[package]]
 name = "jsonrpc-core"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
  "futures 0.1.11 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1073,7 +1073,7 @@ dependencies = [
 [[package]]
 name = "jsonrpc-http-server"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
  "hyper 0.10.0-a.0 (git+https://github.com/paritytech/hyper)",
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
@@ -1086,7 +1086,7 @@ dependencies = [
 [[package]]
 name = "jsonrpc-ipc-server"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
  "bytes 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
@@ -1099,7 +1099,7 @@ dependencies = [
 [[package]]
 name = "jsonrpc-macros"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
  "jsonrpc-pubsub 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
@@ -1109,7 +1109,7 @@ dependencies = [
 [[package]]
 name = "jsonrpc-minihttp-server"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
  "bytes 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
@@ -1124,7 +1124,7 @@ dependencies = [
 [[package]]
 name = "jsonrpc-pubsub"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
  "log 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1134,8 +1134,9 @@ dependencies = [
 [[package]]
 name = "jsonrpc-server-utils"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
+ "bytes 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "globset 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
  "log 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1146,7 +1147,7 @@ dependencies = [
 [[package]]
 name = "jsonrpc-tcp-server"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
  "bytes 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
@@ -1160,11 +1161,13 @@ dependencies = [
 [[package]]
 name = "jsonrpc-ws-server"
 version = "7.0.0"
-source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#d12476f42ee672fa9d023f66fcfa5981d9aaba3a"
+source = "git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7#5e79be8a098cdda221713992f4a46b41a1d4d8f0"
 dependencies = [
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
  "jsonrpc-server-utils 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
  "log 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "parking_lot 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "slab 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "ws 0.7.1 (git+https://github.com/tomusdrw/ws-rs)",
 ]
 
@@ -1775,6 +1778,7 @@ dependencies = [
  "ethsync 1.7.0",
  "fetch 0.1.0",
  "futures 0.1.11 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-cpupool 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
  "jsonrpc-http-server 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
  "jsonrpc-ipc-server 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",

parity/cli/mod.rs

@@ -180,8 +180,10 @@ usage! {
 			or |c: &Config| otry!(c.rpc).apis.as_ref().map(|vec| vec.join(",")),
 		flag_jsonrpc_hosts: String = "none",
 			or |c: &Config| otry!(c.rpc).hosts.as_ref().map(|vec| vec.join(",")),
-		flag_jsonrpc_threads: Option<usize> = None,
+		flag_jsonrpc_server_threads: Option<usize> = None,
-			or |c: &Config| otry!(c.rpc).threads.map(Some),
+			or |c: &Config| otry!(c.rpc).server_threads.map(Some),
+		flag_jsonrpc_threads: usize = 0usize,
+			or |c: &Config| otry!(c.rpc).processing_threads,
 
 		// WS
 		flag_no_ws: bool = false,
@@ -468,7 +470,8 @@ struct Rpc {
 	cors: Option<String>,
 	apis: Option<Vec<String>>,
 	hosts: Option<Vec<String>>,
-	threads: Option<usize>,
+	server_threads: Option<usize>,
+	processing_threads: Option<usize>,
 }
 
 #[derive(Default, Debug, PartialEq, Deserialize)]
@@ -749,7 +752,8 @@ mod tests {
 			flag_jsonrpc_cors: Some("null".into()),
 			flag_jsonrpc_apis: "web3,eth,net,parity,traces,rpc,secretstore".into(),
 			flag_jsonrpc_hosts: "none".into(),
-			flag_jsonrpc_threads: None,
+			flag_jsonrpc_server_threads: None,
+			flag_jsonrpc_threads: 0,
 
 			// WS
 			flag_no_ws: false,
@@ -977,7 +981,8 @@ mod tests {
 				cors: None,
 				apis: None,
 				hosts: None,
-				threads: None,
+				server_threads: None,
+				processing_threads: None,
 			}),
 			ipc: Some(Ipc {
 				disable: None,

parity/cli/usage.txt

@@ -176,9 +176,12 @@ API and Console Options:
                                    is additional security against some attack
                                    vectors. Special options: "all", "none",
                                    (default: {flag_jsonrpc_hosts}).
-  --jsonrpc-threads THREADS        Enables experimental faster implementation of JSON-RPC server.
+  --jsonrpc-server-threads NUM     Enables experimental faster implementation of JSON-RPC server.
                                    Requires Dapps server to be disabled
-                                   using --no-dapps. (default: {flag_jsonrpc_threads:?})
+                                   using --no-dapps. (default: {flag_jsonrpc_server_threads:?})
+  --jsonrpc-threads THREADS        Turn on additional processing threads in all RPC servers.
+                                   Setting this to non-zero value allows parallel cpu-heavy queries
+                                   execution. (default: {flag_jsonrpc_threads})
 
   --no-ws                          Disable the WebSockets server. (default: {flag_no_ws})
   --ws-port PORT                   Specify the port portion of the WebSockets server

parity/configuration.rs

@@ -137,7 +137,7 @@ impl Configuration {
 		let secretstore_conf = self.secretstore_config()?;
 		let format = self.format()?;
 
-		if self.args.flag_jsonrpc_threads.is_some() && dapps_conf.enabled {
+		if self.args.flag_jsonrpc_server_threads.is_some() && dapps_conf.enabled {
 			dapps_conf.enabled = false;
 			writeln!(&mut stderr(), "Warning: Disabling Dapps server because fast RPC server was enabled.").expect("Error writing to stderr.")
 		}
@@ -825,11 +825,12 @@ impl Configuration {
 			},
 			hosts: self.rpc_hosts(),
 			cors: self.rpc_cors(),
-			threads: match self.args.flag_jsonrpc_threads {
+			server_threads: match self.args.flag_jsonrpc_server_threads {
 				Some(threads) if threads > 0 => Some(threads),
 				None => None,
-				_ => return Err("--jsonrpc-threads number needs to be positive.".into()),
+				_ => return Err("--jsonrpc-server-threads number needs to be positive.".into()),
-			}
+			},
+			processing_threads: self.args.flag_jsonrpc_threads,
 		};
 
 		Ok(conf)

parity/rpc.rs

@@ -30,6 +30,7 @@ use rpc_apis::{self, ApiSet};
 
 pub use parity_rpc::{IpcServer, HttpServer, RequestMiddleware};
 pub use parity_rpc::ws::Server as WsServer;
+pub use parity_rpc::informant::CpuPool;
 
 
 pub const DAPPS_DOMAIN: &'static str = "web3.site";
@@ -42,7 +43,8 @@ pub struct HttpConfiguration {
 	pub apis: ApiSet,
 	pub cors: Option<Vec<String>>,
 	pub hosts: Option<Vec<String>>,
-	pub threads: Option<usize>,
+	pub server_threads: Option<usize>,
+	pub processing_threads: usize,
 }
 
 impl HttpConfiguration {
@@ -63,7 +65,8 @@ impl Default for HttpConfiguration {
 			apis: ApiSet::UnsafeContext,
 			cors: None,
 			hosts: Some(Vec::new()),
-			threads: None,
+			server_threads: None,
+			processing_threads: 0,
 		}
 	}
 }
@@ -94,7 +97,8 @@ impl From<UiConfiguration> for HttpConfiguration {
 			apis: rpc_apis::ApiSet::SafeContext,
 			cors: None,
 			hosts: conf.hosts,
-			threads: None,
+			server_threads: None,
+			processing_threads: 0,
 		}
 	}
 }
@@ -176,6 +180,7 @@ pub struct Dependencies<D: rpc_apis::Dependencies> {
 	pub apis: Arc<D>,
 	pub remote: TokioRemote,
 	pub stats: Arc<RpcStats>,
+	pub pool: Option<CpuPool>,
 }
 
 pub fn new_ws<D: rpc_apis::Dependencies>(
@@ -192,11 +197,12 @@ pub fn new_ws<D: rpc_apis::Dependencies>(
 	let addr = url.parse().map_err(|_| format!("Invalid WebSockets listen host/port given: {}", url))?;
 
 
-	let full_handler = setup_apis(rpc_apis::ApiSet::SafeContext, deps);
+	let pool = deps.pool.clone();
+	let full_handler = setup_apis(rpc_apis::ApiSet::SafeContext, deps, pool.clone());
 	let handler = {
 		let mut handler = MetaIoHandler::with_middleware((
 			rpc::WsDispatcher::new(full_handler),
-			Middleware::new(deps.stats.clone(), deps.apis.activity_notifier())
+			Middleware::new(deps.stats.clone(), deps.apis.activity_notifier(), pool)
 		));
 		let apis = conf.apis.list_apis();
 		deps.apis.extend_with_set(&mut handler, &apis);
@@ -252,7 +258,8 @@ pub fn new_http<D: rpc_apis::Dependencies>(
 	let http_address = (conf.interface, conf.port);
 	let url = format!("{}:{}", http_address.0, http_address.1);
 	let addr = url.parse().map_err(|_| format!("Invalid {} listen host/port given: {}", id, url))?;
-	let handler = setup_apis(conf.apis, deps);
+	let pool = deps.pool.clone();
+	let handler = setup_apis(conf.apis, deps, pool);
 	let remote = deps.remote.clone();
 
 	let cors_domains = into_domains(conf.cors);
@@ -265,7 +272,7 @@ pub fn new_http<D: rpc_apis::Dependencies>(
 		handler,
 		remote,
 		rpc::RpcExtractor,
-		match (conf.threads, middleware) {
+		match (conf.server_threads, middleware) {
 			(Some(threads), None) => rpc::HttpSettings::Threads(threads),
 			(None, middleware) => rpc::HttpSettings::Dapps(middleware),
 			(Some(_), Some(_)) => {
@@ -291,7 +298,8 @@ pub fn new_ipc<D: rpc_apis::Dependencies>(
 		return Ok(None);
 	}
 
-	let handler = setup_apis(conf.apis, dependencies);
+	let pool = dependencies.pool.clone();
+	let handler = setup_apis(conf.apis, dependencies, pool);
 	let remote = dependencies.remote.clone();
 	match rpc::start_ipc(&conf.socket_addr, handler, remote, rpc::RpcExtractor) {
 		Ok(server) => Ok(Some(server)),
@@ -318,11 +326,11 @@ fn with_domain(items: Option<Vec<String>>, domain: &str, addresses: &[Option<(St
 	})
 }
 
-fn setup_apis<D>(apis: ApiSet, deps: &Dependencies<D>) -> MetaIoHandler<Metadata, Middleware<D::Notifier>>
+fn setup_apis<D>(apis: ApiSet, deps: &Dependencies<D>, pool: Option<CpuPool>) -> MetaIoHandler<Metadata, Middleware<D::Notifier>>
 	where D: rpc_apis::Dependencies
 {
 	let mut handler = MetaIoHandler::with_middleware(
-		Middleware::new(deps.stats.clone(), deps.apis.activity_notifier())
+		Middleware::new(deps.stats.clone(), deps.apis.activity_notifier(), pool)
 	);
 	let apis = apis.list_apis();
 	deps.apis.extend_with_set(&mut handler, &apis);

parity/run.rs

@@ -316,6 +316,11 @@ fn execute_light(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) ->
 		apis: deps_for_rpc_apis.clone(),
 		remote: event_loop.raw_remote(),
 		stats: rpc_stats.clone(),
+		pool: if cmd.http_conf.processing_threads > 0 {
+			Some(rpc::CpuPool::new(cmd.http_conf.processing_threads))
+		} else {
+			None
+		},
 	};
 
 	// start rpc servers
@@ -663,6 +668,12 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
 		apis: deps_for_rpc_apis.clone(),
 		remote: event_loop.raw_remote(),
 		stats: rpc_stats.clone(),
+		pool: if cmd.http_conf.processing_threads > 0 {
+			Some(rpc::CpuPool::new(cmd.http_conf.processing_threads))
+		} else {
+			None
+		},
+
 	};
 
 	// start rpc servers

rpc/Cargo.toml

@@ -10,6 +10,7 @@ authors = ["Parity Technologies <admin@parity.io>"]
 [dependencies]
 cid = "0.2"
 futures = "0.1"
+futures-cpupool = "0.1"
 log = "0.3"
 multihash ="0.6"
 order-stat = "0.1"

rpc/src/lib.rs

@@ -23,6 +23,7 @@
 extern crate cid;
 extern crate crypto as rust_crypto;
 extern crate futures;
+extern crate futures_cpupool;
 extern crate multihash;
 extern crate order_stat;
 extern crate rand;

rpc/src/v1/extractors.rs

@@ -217,18 +217,26 @@ impl<M: core::Middleware<Metadata>> WsDispatcher<M> {
 }
 
 impl<M: core::Middleware<Metadata>> core::Middleware<Metadata> for WsDispatcher<M> {
-	fn on_request<F>(&self, request: core::Request, meta: Metadata, process: F) -> core::FutureResponse where
+	type Future = core::futures::future::Either<
-		F: FnOnce(core::Request, Metadata) -> core::FutureResponse,
+		M::Future,
+		core::FutureResponse,
+	>;
+
+	fn on_request<F, X>(&self, request: core::Request, meta: Metadata, process: F) -> Self::Future where
+		F: FnOnce(core::Request, Metadata) -> X,
+		X: core::futures::Future<Item=Option<core::Response>, Error=()> + Send + 'static,
 	{
+		use self::core::futures::future::Either::{A, B};
+
 		let use_full = match &meta.origin {
 			&Origin::Signer { .. } => true,
 			_ => false,
 		};
 
 		if use_full {
-			self.full_handler.handle_rpc_request(request, meta)
+			A(self.full_handler.handle_rpc_request(request, meta))
 		} else {
-			process(request, meta)
+			B(process(request, meta).boxed())
 		}
 	}
 }

rpc/src/v1/informant.rs

@@ -21,10 +21,13 @@ use std::sync::Arc;
 use std::sync::atomic::{self, AtomicUsize};
 use std::time;
 use futures::Future;
+use futures_cpupool as pool;
 use jsonrpc_core as rpc;
 use order_stat;
 use util::RwLock;
 
+pub use self::pool::CpuPool;
+
 const RATE_SECONDS: usize = 10;
 const STATS_SAMPLES: usize = 60;
 
@@ -184,14 +187,16 @@ pub trait ActivityNotifier: Send + Sync + 'static {
 pub struct Middleware<T: ActivityNotifier = ClientNotifier> {
 	stats: Arc<RpcStats>,
 	notifier: T,
+	pool: Option<CpuPool>,
 }
 
 impl<T: ActivityNotifier> Middleware<T> {
 	/// Create new Middleware with stats counter and activity notifier.
-	pub fn new(stats: Arc<RpcStats>, notifier: T) -> Self {
+	pub fn new(stats: Arc<RpcStats>, notifier: T, pool: Option<CpuPool>) -> Self {
 		Middleware {
-			stats: stats,
+			stats,
-			notifier: notifier,
+			notifier,
+			pool,
 		}
 	}
 
@@ -201,19 +206,32 @@ impl<T: ActivityNotifier> Middleware<T> {
 }
 
 impl<M: rpc::Metadata, T: ActivityNotifier> rpc::Middleware<M> for Middleware<T> {
-	fn on_request<F>(&self, request: rpc::Request, meta: M, process: F) -> rpc::FutureResponse where
+	type Future = rpc::futures::future::Either<
-		F: FnOnce(rpc::Request, M) -> rpc::FutureResponse,
+		pool::CpuFuture<Option<rpc::Response>, ()>,
+		rpc::FutureResponse,
+	>;
+
+	fn on_request<F, X>(&self, request: rpc::Request, meta: M, process: F) -> Self::Future where
+		F: FnOnce(rpc::Request, M) -> X,
+		X: rpc::futures::Future<Item=Option<rpc::Response>, Error=()> + Send + 'static,
 	{
+		use self::rpc::futures::future::Either::{A, B};
+
 		let start = time::Instant::now();
-		let response = process(request, meta);
 
 		self.notifier.active();
+		self.stats.count_request();
+
 		let stats = self.stats.clone();
-		stats.count_request();
+		let future = process(request, meta).map(move |res| {
-		response.map(move |res| {
 			stats.add_roundtrip(Self::as_micro(start.elapsed()));
 			res
-		}).boxed()
+		});
+
+		match self.pool {
+			Some(ref pool) => A(pool.spawn(future)),
+			None => B(future.boxed()),
+		}
 	}
 }