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.

Cargo.toml

@@ -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", branch = "parity-2.2" }
 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" }

ethcore/stratum/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", branch = "parity-2.2" }
+jsonrpc-macros = { git = "https://github.com/paritytech/jsonrpc.git", branch = "parity-2.2" }
+jsonrpc-tcp-server = { git = "https://github.com/paritytech/jsonrpc.git", branch = "parity-2.2" }
 log = "0.4"
 parking_lot = "0.6"
 

hash-fetch/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/src/client.rs

@@ -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/lib.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;

ipfs/Cargo.toml

@@ -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", branch = "parity-2.2" }
+jsonrpc-http-server = { git = "https://github.com/paritytech/jsonrpc.git", branch = "parity-2.2" }
 rlp = { version = "0.3.0", features = ["ethereum"] }
 cid = "0.3"
 multihash = "0.8"

ipfs/src/error.rs

@@ -30,6 +30,42 @@ pub enum ServerError {
 	InvalidInterface
 }
 
+
+/// Handle IO errors (ports taken when starting the server).
+impl From<::std::io::Error> for ServerError {
+	fn from(err: ::std::io::Error) -> ServerError {
+		ServerError::IoError(err)
+	}
+}
+
+impl From<http::hyper::error::Error> for ServerError {
+	fn from(err: http::hyper::error::Error) -> ServerError {
+		ServerError::Other(err)
+	}
+}
+
+impl From<ServerError> for String {
+	fn from(err: ServerError) -> String {
+		match err {
+			ServerError::IoError(err) => err.to_string(),
+			ServerError::Other(err) => err.to_string(),
+			ServerError::InvalidInterface => "Invalid --ipfs-api-interface parameter".into(),
+		}
+	}
+}
+
+impl ::std::fmt::Display for ServerError {
+	fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+        match self {
+        	ServerError::IoError(err) => write!(f, "Io Error: {}", err),
+        	ServerError::Other(err) => write!(f, "Other error: {}", err),
+        	ServerError::InvalidInterface => write!(f, "Invalid interface"),
+        }
+    }
+}
+
+impl ::std::error::Error for ServerError {}
+
 #[derive(Debug, PartialEq)]
 pub enum Error {
 	CidParsingFailed,
@@ -71,35 +107,4 @@ impl From<multihash::Error> for Error {
 	fn from(_: multihash::Error) -> Error {
 		Error::CidParsingFailed
 	}
-}
-
-/// Handle IO errors (ports taken when starting the server).
-impl From<::std::io::Error> for ServerError {
-	fn from(err: ::std::io::Error) -> ServerError {
-		ServerError::IoError(err)
-	}
-}
-
-impl From<http::hyper::error::Error> for ServerError {
-	fn from(err: http::hyper::error::Error) -> ServerError {
-		ServerError::Other(err)
-	}
-}
-
-impl From<ServerError> for String {
-	fn from(err: ServerError) -> String {
-		match err {
-			ServerError::IoError(err) => err.to_string(),
-			ServerError::Other(err) => err.to_string(),
-			ServerError::InvalidInterface => "Invalid --ipfs-api-interface parameter".into(),
-		}
-	}
-}
-
-impl ::std::fmt::Display for ServerError {
-	fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
-        write!(f, "{:?}", self)
-    }
-}
-
-impl ::std::error::Error for ServerError {}
+}

miner/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/src/gas_price_calibrator.rs

@@ -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/lib.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/work_notify.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()

parity/lib.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;