Consolidate Tokio Runtime use, remove CpuPool.

* Rename and move the `tokio_reactor` crate (`util/reactor`) to
  `tokio_runtime` (`util/runtime`).

* Rename `EventLoop` to `Runtime`.

    - Rename `EventLoop::spawn` to `Runtime::with_default_thread_count`.

    - Add the `Runtime::with_thread_count` method.

    - Rename `Remote` to `Executor`.

* Remove uses of `CpuPool` and spawn all tasks via the `Runtime` executor
  instead.

* Other changes related to `CpuPool` removal:

    - Remove `Reservations::with_pool`. `::new` now takes an `Executor` as an argument.

    - Remove `SenderReservations::with_pool`. `::new` now takes an `Executor` as an argument.

Author: c0gent

Cargo.lock

@@ -28,10 +28,9 @@ serde = "1.0"
 serde_json = "1.0"
 serde_derive = "1.0"
 futures = "0.1"
-futures-cpupool = "0.1"
 fdlimit = "0.1"
 ctrlc = { git = "https://github.com/paritytech/rust-ctrlc.git" }
-jsonrpc-core = { git = "https://github.com/c0gent/jsonrpc.git", branch = "c0gent-hyper" }
+jsonrpc-core = { git = "https://github.com/paritytech/jsonrpc.git", rev = "7a59776" }
 ethcore = { path = "ethcore", features = ["parity"] }
 parity-bytes = "0.1"
 ethcore-io = { path = "util/io" }
@@ -51,7 +50,7 @@ rpc-cli = { path = "rpc_cli" }
 parity-hash-fetch = { path = "hash-fetch" }
 parity-ipfs-api = { path = "ipfs" }
 parity-local-store = { path = "local-store" }
-parity-reactor = { path = "util/reactor" }
+parity-runtime = { path = "util/runtime" }
 parity-rpc = { path = "rpc" }
 parity-rpc-client = { path = "rpc_client" }
 parity-updater = { path = "updater" }

Cargo.toml

@@ -8,9 +8,9 @@ authors = ["Parity Technologies <[email protected]>"]
 [dependencies]
 ethereum-types = "0.4"
 keccak-hash = "0.1"
-jsonrpc-core = { git = "https://github.com/c0gent/jsonrpc.git", branch = "c0gent-hyper" }
-jsonrpc-macros = { git = "https://github.com/c0gent/jsonrpc.git", branch = "c0gent-hyper" }
-jsonrpc-tcp-server = { git = "https://github.com/c0gent/jsonrpc.git", branch = "c0gent-hyper" }
+jsonrpc-core = { git = "https://github.com/paritytech/jsonrpc.git", rev = "7a59776" }
+jsonrpc-macros = { git = "https://github.com/paritytech/jsonrpc.git", rev = "7a59776" }
+jsonrpc-tcp-server = { git = "https://github.com/paritytech/jsonrpc.git", rev = "7a59776" }
 log = "0.4"
 parking_lot = "0.6"
 

ethcore/stratum/Cargo.toml

@@ -8,7 +8,6 @@ authors = ["Parity Technologies <[email protected]>"]
 
 [dependencies]
 futures = "0.1"
-futures-cpupool = "0.1"
 log = "0.4"
 mime = "0.3"
 mime_guess = "2.0.0-alpha.2"
@@ -17,7 +16,7 @@ rustc-hex = "1.0"
 fetch = { path = "../util/fetch" }
 parity-bytes = "0.1"
 ethereum-types = "0.4"
-parity-reactor = { path = "../util/reactor" }
+parity-runtime = { path = "../util/runtime" }
 keccak-hash = "0.1"
 registrar = { path = "../registrar" }
 

hash-fetch/Cargo.toml

@@ -23,9 +23,8 @@ use std::path::PathBuf;
 
 use hash::keccak_buffer;
 use fetch::{self, Fetch};
-use futures_cpupool::CpuPool;
 use futures::{Future, IntoFuture};
-use parity_reactor::Remote;
+use parity_runtime::Executor;
 use urlhint::{URLHintContract, URLHint, URLHintResult};
 use registrar::{RegistrarClient, Asynchronous};
 use ethereum_types::H256;
@@ -109,21 +108,19 @@ fn validate_hash(path: PathBuf, hash: H256, body: fetch::BodyReader) -> Result<P
 
 /// Default Hash-fetching client using on-chain contract to resolve hashes to URLs.
 pub struct Client<F: Fetch + 'static = fetch::Client> {
-	pool: CpuPool,
 	contract: URLHintContract,
 	fetch: F,
-	remote: Remote,
+	executor: Executor,
 	random_path: Arc<Fn() -> PathBuf + Sync + Send>,
 }
 
 impl<F: Fetch + 'static> Client<F> {
 	/// Creates new instance of the `Client` given on-chain contract client, fetch service and task runner.
-	pub fn with_fetch(contract: Arc<RegistrarClient<Call=Asynchronous>>, pool: CpuPool, fetch: F, remote: Remote) -> Self {
+	pub fn with_fetch(contract: Arc<RegistrarClient<Call=Asynchronous>>, fetch: F, executor: Executor) -> Self {
 		Client {
-			pool,
 			contract: URLHintContract::new(contract),
 			fetch: fetch,
-			remote: remote,
+			executor: executor,
 			random_path: Arc::new(random_temp_path),
 		}
 	}
@@ -135,7 +132,6 @@ impl<F: Fetch + 'static> HashFetch for Client<F> {
 
 		let random_path = self.random_path.clone();
 		let remote_fetch = self.fetch.clone();
-		let pool = self.pool.clone();
 		let future = self.contract.resolve(hash)
 			.map_err(|e| { warn!("Error resolving URL: {}", e); Error::NoResolution })
 			.and_then(|maybe_url| maybe_url.ok_or(Error::NoResolution))
@@ -162,7 +158,7 @@ impl<F: Fetch + 'static> HashFetch for Client<F> {
 					Ok(response)
 				}
 			})
-			.and_then(move |response| pool.spawn_fn(move || {
+			.and_then(move |response| {
 				debug!(target: "fetch", "Content fetched, validating hash ({:?})", hash);
 				let path = random_path();
 				let res = validate_hash(path.clone(), hash, fetch::BodyReader::new(response));
@@ -172,10 +168,10 @@ impl<F: Fetch + 'static> HashFetch for Client<F> {
 					let _ = fs::remove_file(&path);
 				}
 				res
-			}))
+			})
 			.then(move |res| { on_done(res); Ok(()) as Result<(), ()> });
 
-		self.remote.spawn(future);
+		self.executor.spawn(future);
 	}
 }
 
@@ -197,8 +193,7 @@ mod tests {
 	use rustc_hex::FromHex;
 	use std::sync::{Arc, mpsc};
 	use parking_lot::Mutex;
-	use futures_cpupool::CpuPool;
-	use parity_reactor::Remote;
+	use parity_runtime::Executor;
 	use urlhint::tests::{FakeRegistrar, URLHINT};
 	use super::{Error, Client, HashFetch, random_temp_path};
 
@@ -216,7 +211,7 @@ mod tests {
 		// given
 		let contract = Arc::new(FakeRegistrar::new());
 		let fetch = FakeFetch::new(None::<usize>);
-		let client = Client::with_fetch(contract.clone(), CpuPool::new(1), fetch, Remote::new_sync());
+		let client = Client::with_fetch(contract.clone(), fetch, Executor::new_sync());
 
 		// when
 		let (tx, rx) = mpsc::channel();
@@ -234,7 +229,7 @@ mod tests {
 		// given
 		let registrar = Arc::new(registrar());
 		let fetch = FakeFetch::new(None::<usize>);
-		let client = Client::with_fetch(registrar.clone(), CpuPool::new(1), fetch, Remote::new_sync());
+		let client = Client::with_fetch(registrar.clone(), fetch, Executor::new_sync());
 
 		// when
 		let (tx, rx) = mpsc::channel();
@@ -252,7 +247,7 @@ mod tests {
 		// given
 		let registrar = Arc::new(registrar());
 		let fetch = FakeFetch::new(Some(1));
-		let mut client = Client::with_fetch(registrar.clone(), CpuPool::new(1), fetch, Remote::new_sync());
+		let mut client = Client::with_fetch(registrar.clone(), fetch, Executor::new_sync());
 		let path = random_temp_path();
 		let path2 = path.clone();
 		client.random_path = Arc::new(move || path2.clone());
@@ -275,7 +270,7 @@ mod tests {
 		// given
 		let registrar = Arc::new(registrar());
 		let fetch = FakeFetch::new(Some(1));
-		let client = Client::with_fetch(registrar.clone(), CpuPool::new(1), fetch, Remote::new_sync());
+		let client = Client::with_fetch(registrar.clone(), fetch, Executor::new_sync());
 
 		// when
 		let (tx, rx) = mpsc::channel();

hash-fetch/src/client.rs

@@ -25,11 +25,10 @@ extern crate ethabi;
 extern crate parity_bytes as bytes;
 extern crate ethereum_types;
 extern crate futures;
-extern crate futures_cpupool;
 extern crate keccak_hash as hash;
 extern crate mime;
 extern crate mime_guess;
-extern crate parity_reactor;
+extern crate parity_runtime;
 extern crate rand;
 extern crate rustc_hex;
 extern crate registrar;

hash-fetch/src/lib.rs

@@ -9,8 +9,8 @@ authors = ["Parity Technologies <[email protected]>"]
 ethcore = { path = "../ethcore" }
 parity-bytes = "0.1"
 ethereum-types = "0.4"
-jsonrpc-core = { git = "https://github.com/c0gent/jsonrpc.git", branch = "c0gent-hyper" }
-jsonrpc-http-server = { git = "https://github.com/c0gent/jsonrpc.git", branch = "c0gent-hyper" }
+jsonrpc-core = { git = "https://github.com/paritytech/jsonrpc.git", rev = "7a59776" }
+jsonrpc-http-server = { git = "https://github.com/paritytech/jsonrpc.git", rev = "7a59776" }
 rlp = { version = "0.3.0", features = ["ethereum"] }
 cid = "0.3"
 multihash = "0.8"

ipfs/Cargo.toml

@@ -11,7 +11,6 @@ authors = ["Parity Technologies <[email protected]>"]
 ethash = { path = "../ethash", optional = true }
 fetch = { path = "../util/fetch", optional = true }
 hyper = { version = "0.12", optional = true }
-parity-reactor = { path = "../util/reactor", optional = true }
 url = { version = "1", optional = true }
 
 # Miner
@@ -20,7 +19,7 @@ error-chain = "0.12"
 ethcore-transaction = { path = "../ethcore/transaction" }
 ethereum-types = "0.4"
 futures = "0.1"
-futures-cpupool = "0.1"
+parity-runtime = { path = "../util/runtime" }
 heapsize = "0.4"
 keccak-hash = "0.1"
 linked-hash-map = "0.5"
@@ -37,4 +36,4 @@ ethkey = { path = "../ethkey" }
 rustc-hex = "1.0"
 
 [features]
-work-notify = ["ethash", "fetch", "hyper", "parity-reactor", "url"]
+work-notify = ["ethash", "fetch", "hyper", "url"]

miner/Cargo.toml

@@ -20,7 +20,7 @@ use std::time::{Instant, Duration};
 
 use ansi_term::Colour;
 use ethereum_types::U256;
-use futures_cpupool::CpuPool;
+use parity_runtime::Executor;
 use price_info::{Client as PriceInfoClient, PriceInfo};
 use price_info::fetch::Client as FetchClient;
 
@@ -43,7 +43,7 @@ pub struct GasPriceCalibrator {
 
 impl GasPriceCalibrator {
 	/// Create a new gas price calibrator.
-	pub fn new(options: GasPriceCalibratorOptions, fetch: FetchClient, p: CpuPool) -> GasPriceCalibrator {
+	pub fn new(options: GasPriceCalibratorOptions, fetch: FetchClient, p: Executor) -> GasPriceCalibrator {
 		GasPriceCalibrator {
 			options: options,
 			next_calibration: Instant::now(),

miner/src/gas_price_calibrator.rs

@@ -23,7 +23,7 @@ extern crate ansi_term;
 extern crate ethcore_transaction as transaction;
 extern crate ethereum_types;
 extern crate futures;
-extern crate futures_cpupool;
+extern crate parity_runtime;
 extern crate heapsize;
 extern crate keccak_hash as hash;
 extern crate linked_hash_map;

miner/src/lib.rs

@@ -18,12 +18,12 @@
 
 extern crate ethash;
 extern crate fetch;
-extern crate parity_reactor;
+extern crate parity_runtime;
 extern crate url;
 extern crate hyper;
 
 use self::fetch::{Fetch, Request, Client as FetchClient, Method};
-use self::parity_reactor::Remote;
+use self::parity_runtime::Executor;
 use self::ethash::SeedHashCompute;
 use self::url::Url;
 use self::hyper::header::{self, HeaderValue};
@@ -43,13 +43,13 @@ pub trait NotifyWork : Send + Sync {
 pub struct WorkPoster {
 	urls: Vec<Url>,
 	client: FetchClient,
-	remote: Remote,
+	executor: Executor,
 	seed_compute: Mutex<SeedHashCompute>,
 }
 
 impl WorkPoster {
 	/// Create new `WorkPoster`.
-	pub fn new(urls: &[String], fetch: FetchClient, remote: Remote) -> Self {
+	pub fn new(urls: &[String], fetch: FetchClient, executor: Executor) -> Self {
 		let urls = urls.into_iter().filter_map(|u| {
 			match Url::parse(u) {
 				Ok(url) => Some(url),
@@ -61,7 +61,7 @@ impl WorkPoster {
 		}).collect();
 		WorkPoster {
 			client: fetch,
-			remote: remote,
+			executor: executor,
 			urls: urls,
 			seed_compute: Mutex::new(SeedHashCompute::default()),
 		}
@@ -81,7 +81,7 @@ impl NotifyWork for WorkPoster {
 
 		for u in &self.urls {
 			let u = u.clone();
-			self.remote.spawn(self.client.fetch(
+			self.executor.spawn(self.client.fetch(
 				Request::new(u.clone(), Method::POST)
 					.with_header(header::CONTENT_TYPE, HeaderValue::from_static("application/json"))
 					.with_body(body.clone()), Default::default()

miner/src/work_notify.rs

@@ -25,7 +25,6 @@ extern crate clap;
 extern crate dir;
 extern crate env_logger;
 extern crate futures;
-extern crate futures_cpupool;
 extern crate atty;
 extern crate jsonrpc_core;
 extern crate num_cpus;
@@ -60,7 +59,7 @@ extern crate kvdb;
 extern crate parity_hash_fetch as hash_fetch;
 extern crate parity_ipfs_api;
 extern crate parity_local_store as local_store;
-extern crate parity_reactor;
+extern crate parity_runtime;
 extern crate parity_rpc;
 extern crate parity_updater as updater;
 extern crate parity_version;

parity/lib.rs

@@ -29,7 +29,7 @@ use light::TransactionQueue;
 
 use futures::{future, Future};
 
-use parity_reactor::Remote;
+use parity_runtime::Executor;
 
 use parking_lot::RwLock;
 
@@ -50,8 +50,8 @@ pub struct QueueCull<T> {
 	pub on_demand: Arc<OnDemand>,
 	/// The transaction queue.
 	pub txq: Arc<RwLock<TransactionQueue>>,
-	/// Event loop remote.
-	pub remote: Remote,
+	/// Event loop executor.
+	pub executor: Executor,
 }
 
 impl<T: LightChainClient + 'static> IoHandler<ClientIoMessage> for QueueCull<T> {
@@ -70,7 +70,7 @@ impl<T: LightChainClient + 'static> IoHandler<ClientIoMessage> for QueueCull<T>
 		let start_nonce = self.client.engine().account_start_nonce(best_header.number());
 
 		info!(target: "cull", "Attempting to cull queued transactions from {} senders.", senders.len());
-		self.remote.spawn_with_timeout(move || {
+		self.executor.spawn_with_timeout(move || {
 			let maybe_fetching = sync.with_context(move |ctx| {
 				// fetch the nonce of each sender in the queue.
 				let nonce_reqs = senders.iter()

parity/light_helpers/queue_cull.rs

@@ -21,7 +21,7 @@ use ethcore::client::Mode;
 use ethcore::ethereum;
 use ethcore::spec::{Spec, SpecParams};
 use ethereum_types::{U256, Address};
-use futures_cpupool::CpuPool;
+use parity_runtime::Executor;
 use hash_fetch::fetch::Client as FetchClient;
 use journaldb::Algorithm;
 use miner::gas_pricer::GasPricer;
@@ -256,7 +256,7 @@ impl Default for GasPricerConfig {
 }
 
 impl GasPricerConfig {
-	pub fn to_gas_pricer(&self, fetch: FetchClient, p: CpuPool) -> GasPricer {
+	pub fn to_gas_pricer(&self, fetch: FetchClient, p: Executor) -> GasPricer {
 		match *self {
 			GasPricerConfig::Fixed(u) => GasPricer::Fixed(u),
 			GasPricerConfig::Calibrated { usd_per_tx, recalibration_period, .. } => {

parity/params.rs

@@ -23,14 +23,13 @@ use dir::default_data_path;
 use dir::helpers::replace_home;
 use helpers::parity_ipc_path;
 use jsonrpc_core::MetaIoHandler;
-use parity_reactor::Remote;
+use parity_runtime::Executor;
 use parity_rpc::informant::{RpcStats, Middleware};
 use parity_rpc::{self as rpc, Metadata, DomainsValidation};
 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";
 
@@ -134,9 +133,8 @@ fn address(enabled: bool, bind_iface: &str, bind_port: u16, hosts: &Option<Vec<S
 
 pub struct Dependencies<D: rpc_apis::Dependencies> {
 	pub apis: Arc<D>,
-	pub remote: Remote,
+	pub executor: Executor,
 	pub stats: Arc<RpcStats>,
-	pub pool: Option<CpuPool>,
 }
 
 pub fn new_ws<D: rpc_apis::Dependencies>(
@@ -155,7 +153,7 @@ pub fn new_ws<D: rpc_apis::Dependencies>(
 	let handler = {
 		let mut handler = MetaIoHandler::with_middleware((
 			rpc::WsDispatcher::new(full_handler),
-			Middleware::new(deps.stats.clone(), deps.apis.activity_notifier(), deps.pool.clone())
+			Middleware::new(deps.stats.clone(), deps.apis.activity_notifier())
 		));
 		let apis = conf.apis.list_apis();
 		deps.apis.extend_with_set(&mut handler, &apis);
@@ -289,7 +287,7 @@ pub fn setup_apis<D>(apis: ApiSet, deps: &Dependencies<D>) -> MetaIoHandler<Meta
 	where D: rpc_apis::Dependencies
 {
 	let mut handler = MetaIoHandler::with_middleware(
-		Middleware::new(deps.stats.clone(), deps.apis.activity_notifier(), deps.pool.clone())
+		Middleware::new(deps.stats.clone(), deps.apis.activity_notifier())
 	);
 	let apis = apis.list_apis();
 	deps.apis.extend_with_set(&mut handler, &apis);