Remove secret_store runtimes. (#9888)

* Remove the independent runtimes from `KeyServerHttpListener` and
  `KeyServerCore` and instead require a `parity_runtime::Executor`
  to be passed upon creation of each.

* Remove the `threads` parameter from both `ClusterConfiguration` structs.

* Implement the `future::Executor` trait for `parity_runtime::Executor`.

* Update tests.
  - Update the `loop_until` function to instead use a oneshot to signal
    completion.
  - Modify the `make_key_servers` function to create and return a runtime.

parity/run.rs

@@ -739,7 +739,7 @@ fn execute_impl<Cr, Rr>(cmd: RunCmd, logger: Arc<RotatingLogger>, on_client_rq:
 		account_provider: account_provider,
 		accounts_passwords: &passwords,
 	};
-	let secretstore_key_server = secretstore::start(cmd.secretstore_conf.clone(), secretstore_deps)?;
+	let secretstore_key_server = secretstore::start(cmd.secretstore_conf.clone(), secretstore_deps, runtime.executor())?;
 
 	// the ipfs server
 	let ipfs_server = ipfs::start_server(cmd.ipfs_conf.clone(), client.clone())?;

parity/secretstore.rs

@@ -24,6 +24,7 @@ use ethcore::miner::Miner;
 use ethkey::{Secret, Public};
 use sync::SyncProvider;
 use ethereum_types::Address;
+use parity_runtime::Executor;
 
 /// This node secret key.
 #[derive(Debug, PartialEq, Clone)]
@@ -100,14 +101,14 @@ pub struct Dependencies<'a> {
 
 #[cfg(not(feature = "secretstore"))]
 mod server {
-	use super::{Configuration, Dependencies};
+	use super::{Configuration, Dependencies, Executor};
 
 	/// Noop key server implementation
 	pub struct KeyServer;
 
 	impl KeyServer {
 		/// Create new noop key server
-		pub fn new(_conf: Configuration, _deps: Dependencies) -> Result<Self, String> {
+		pub fn new(_conf: Configuration, _deps: Dependencies, _executor: Executor) -> Result<Self, String> {
 			Ok(KeyServer)
 		}
 	}
@@ -120,7 +121,7 @@ mod server {
 	use ethkey::KeyPair;
 	use ansi_term::Colour::{Red, White};
 	use db;
-	use super::{Configuration, Dependencies, NodeSecretKey, ContractAddress};
+	use super::{Configuration, Dependencies, NodeSecretKey, ContractAddress, Executor};
 
 	fn into_service_contract_address(address: ContractAddress) -> ethcore_secretstore::ContractAddress {
 		match address {
@@ -136,7 +137,7 @@ mod server {
 
 	impl KeyServer {
 		/// Create new key server
-		pub fn new(mut conf: Configuration, deps: Dependencies) -> Result<Self, String> {
+		pub fn new(mut conf: Configuration, deps: Dependencies, executor: Executor) -> Result<Self, String> {
 			let self_secret: Arc<ethcore_secretstore::NodeKeyPair> = match conf.self_secret.take() {
 				Some(NodeSecretKey::Plain(secret)) => Arc::new(ethcore_secretstore::PlainNodeKeyPair::new(
 					KeyPair::from_secret(secret).map_err(|e| format!("invalid secret: {}", e))?)),
@@ -179,7 +180,6 @@ mod server {
 				service_contract_doc_sretr_address: conf.service_contract_doc_sretr_address.map(into_service_contract_address),
 				acl_check_contract_address: conf.acl_check_contract_address.map(into_service_contract_address),
 				cluster_config: ethcore_secretstore::ClusterConfiguration {
-					threads: 4,
 					listener_address: ethcore_secretstore::NodeAddress {
 						address: conf.interface.clone(),
 						port: conf.port,
@@ -198,7 +198,7 @@ mod server {
 			cconf.cluster_config.nodes.insert(self_secret.public().clone(), cconf.cluster_config.listener_address.clone());
 
 			let db = db::open_secretstore_db(&conf.data_path)?;
-			let key_server = ethcore_secretstore::start(deps.client, deps.sync, deps.miner, self_secret, cconf, db)
+			let key_server = ethcore_secretstore::start(deps.client, deps.sync, deps.miner, self_secret, cconf, db, executor)
 				.map_err(|e| format!("Error starting KeyServer {}: {}", key_server_name, e))?;
 
 			Ok(KeyServer {
@@ -238,11 +238,11 @@ impl Default for Configuration {
 }
 
 /// Start secret store-related functionality
-pub fn start(conf: Configuration, deps: Dependencies) -> Result<Option<KeyServer>, String> {
+pub fn start(conf: Configuration, deps: Dependencies, executor: Executor) -> Result<Option<KeyServer>, String> {
 	if !conf.enabled {
 		return Ok(None);
 	}
 
-	KeyServer::new(conf, deps)
+	KeyServer::new(conf, deps, executor)
 		.map(|s| Some(s))
 }

secret_store/Cargo.toml

@@ -17,6 +17,7 @@ futures = "0.1"
 rustc-hex = "1.0"
 tiny-keccak = "1.4"
 tokio = "~0.1.11"
+parity-runtime = { path = "../util/runtime" }
 tokio-io = "0.1"
 tokio-service = "0.1"
 url = "1.0"

secret_store/src/key_server.rs

@@ -15,14 +15,11 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 use std::collections::BTreeSet;
-use std::thread;
 use std::sync::Arc;
-use std::sync::mpsc;
-use futures::{self, Future};
 use parking_lot::Mutex;
-use tokio::runtime;
 use crypto::DEFAULT_MAC;
 use ethkey::crypto;
+use parity_runtime::Executor;
 use super::acl_storage::AclStorage;
 use super::key_storage::KeyStorage;
 use super::key_server_set::KeyServerSet;
@@ -39,16 +36,16 @@ pub struct KeyServerImpl {
 
 /// Secret store key server data.
 pub struct KeyServerCore {
-	close: Option<futures::Complete<()>>,
-	handle: Option<thread::JoinHandle<()>>,
 	cluster: Arc<ClusterClient>,
 }
 
 impl KeyServerImpl {
 	/// Create new key server instance
-	pub fn new(config: &ClusterConfiguration, key_server_set: Arc<KeyServerSet>, self_key_pair: Arc<NodeKeyPair>, acl_storage: Arc<AclStorage>, key_storage: Arc<KeyStorage>) -> Result<Self, Error> {
+	pub fn new(config: &ClusterConfiguration, key_server_set: Arc<KeyServerSet>, self_key_pair: Arc<NodeKeyPair>,
+		acl_storage: Arc<AclStorage>, key_storage: Arc<KeyStorage>, executor: Executor) -> Result<Self, Error>
+	{
 		Ok(KeyServerImpl {
-			data: Arc::new(Mutex::new(KeyServerCore::new(config, key_server_set, self_key_pair, acl_storage, key_storage)?)),
+			data: Arc::new(Mutex::new(KeyServerCore::new(config, key_server_set, self_key_pair, acl_storage, key_storage, executor)?)),
 		})
 	}
 
@@ -175,9 +172,10 @@ impl MessageSigner for KeyServerImpl {
 }
 
 impl KeyServerCore {
-	pub fn new(config: &ClusterConfiguration, key_server_set: Arc<KeyServerSet>, self_key_pair: Arc<NodeKeyPair>, acl_storage: Arc<AclStorage>, key_storage: Arc<KeyStorage>) -> Result<Self, Error> {
+	pub fn new(config: &ClusterConfiguration, key_server_set: Arc<KeyServerSet>, self_key_pair: Arc<NodeKeyPair>,
+		acl_storage: Arc<AclStorage>, key_storage: Arc<KeyStorage>, executor: Executor) -> Result<Self, Error>
+	{
 		let config = NetClusterConfiguration {
-			threads: config.threads,
 			self_key_pair: self_key_pair.clone(),
 			listen_address: (config.listener_address.address.clone(), config.listener_address.port),
 			key_server_set: key_server_set,
@@ -188,45 +186,16 @@ impl KeyServerCore {
 			auto_migrate_enabled: config.auto_migrate_enabled,
 		};
 
-		let (stop, stopped) = futures::oneshot();
-		let (tx, rx) = mpsc::channel();
-		let handle = thread::Builder::new().name("KeyServerLoop".into()).spawn(move || {
-			let runtime_res = runtime::Builder::new()
-				.core_threads(config.threads)
-				.build();
-
-			let mut el = match runtime_res {
-				Ok(el) => el,
-				Err(e) => {
-					tx.send(Err(Error::Internal(format!("error initializing event loop: {}", e)))).expect("Rx is blocking upper thread.");
-					return;
-				},
-			};
-
-			let cluster = ClusterCore::new(el.executor(), config);
-			let cluster_client = cluster.and_then(|c| c.run().map(|_| c.client()));
-			tx.send(cluster_client.map_err(Into::into)).expect("Rx is blocking upper thread.");
-			let _ = el.block_on(futures::empty().select(stopped));
-
-			trace!(target: "secretstore_net", "{}: KeyServerLoop thread stopped", self_key_pair.public());
-		}).map_err(|e| Error::Internal(format!("{}", e)))?;
-		let cluster = rx.recv().map_err(|e| Error::Internal(format!("error initializing event loop: {}", e)))??;
+		let cluster = ClusterCore::new(executor, config)
+			.and_then(|c| c.run().map(|_| c.client()))
+			.map_err(|err| Error::from(err))?;
 
 		Ok(KeyServerCore {
-			close: Some(stop),
-			handle: Some(handle),
-			cluster: cluster,
+			cluster,
 		})
 	}
 }
 
-impl Drop for KeyServerCore {
-	fn drop(&mut self) {
-		self.close.take().map(|v| v.send(()));
-		self.handle.take().map(|h| h.join());
-	}
-}
-
 #[cfg(test)]
 pub mod tests {
 	use std::collections::BTreeSet;
@@ -243,6 +212,7 @@ pub mod tests {
 	use key_server_set::tests::MapKeyServerSet;
 	use key_server_cluster::math;
 	use ethereum_types::{H256, H520};
+	use parity_runtime::Runtime;
 	use types::{Error, Public, ClusterConfiguration, NodeAddress, RequestSignature, ServerKeyId,
 		EncryptedDocumentKey, EncryptedDocumentKeyShadow, MessageHash, EncryptedMessageSignature,
 		Requester, NodeId};
@@ -294,10 +264,9 @@ pub mod tests {
 		}
 	}
 
-	fn make_key_servers(start_port: u16, num_nodes: usize) -> (Vec<KeyServerImpl>, Vec<Arc<DummyKeyStorage>>) {
+	fn make_key_servers(start_port: u16, num_nodes: usize) -> (Vec<KeyServerImpl>, Vec<Arc<DummyKeyStorage>>, Runtime) {
 		let key_pairs: Vec<_> = (0..num_nodes).map(|_| Random.generate().unwrap()).collect();
 		let configs: Vec<_> = (0..num_nodes).map(|i| ClusterConfiguration {
-				threads: 1,
 				listener_address: NodeAddress {
 					address: "127.0.0.1".into(),
 					port: start_port + (i as u16),
@@ -316,11 +285,12 @@ pub mod tests {
 			.map(|(k, a)| (k.clone(), format!("{}:{}", a.address, a.port).parse().unwrap()))
 			.collect();
 		let key_storages = (0..num_nodes).map(|_| Arc::new(DummyKeyStorage::default())).collect::<Vec<_>>();
+		let runtime = Runtime::with_thread_count(4);
 		let key_servers: Vec<_> = configs.into_iter().enumerate().map(|(i, cfg)|
 			KeyServerImpl::new(&cfg, Arc::new(MapKeyServerSet::new(false, key_servers_set.clone())),
 				Arc::new(PlainNodeKeyPair::new(key_pairs[i].clone())),
 				Arc::new(DummyAclStorage::default()),
-				key_storages[i].clone()).unwrap()
+				key_storages[i].clone(), runtime.executor()).unwrap()
 		).collect();
 
 		// wait until connections are established. It is fast => do not bother with events here
@@ -350,13 +320,13 @@ pub mod tests {
 			}
 		}
 
-		(key_servers, key_storages)
+		(key_servers, key_storages, runtime)
 	}
 
 	#[test]
 	fn document_key_generation_and_retrievement_works_over_network_with_single_node() {
 		//::logger::init_log();
-		let (key_servers, _) = make_key_servers(6070, 1);
+		let (key_servers, _, runtime) = make_key_servers(6070, 1);
 
 		// generate document key
 		let threshold = 0;
@@ -372,12 +342,13 @@ pub mod tests {
 			let retrieved_key = crypto::ecies::decrypt(&secret, &DEFAULT_MAC, &retrieved_key).unwrap();
 			assert_eq!(retrieved_key, generated_key);
 		}
+		drop(runtime);
 	}
 
 	#[test]
 	fn document_key_generation_and_retrievement_works_over_network_with_3_nodes() {
 		//::logger::init_log();
-		let (key_servers, key_storages) = make_key_servers(6080, 3);
+		let (key_servers, key_storages, runtime) = make_key_servers(6080, 3);
 
 		let test_cases = [0, 1, 2];
 		for threshold in &test_cases {
@@ -399,12 +370,13 @@ pub mod tests {
 				assert!(key_share.encrypted_point.is_some());
 			}
 		}
+		drop(runtime);
 	}
 
 	#[test]
 	fn server_key_generation_and_storing_document_key_works_over_network_with_3_nodes() {
 		//::logger::init_log();
-		let (key_servers, _) = make_key_servers(6090, 3);
+		let (key_servers, _, runtime) = make_key_servers(6090, 3);
 
 		let test_cases = [0, 1, 2];
 		for threshold in &test_cases {
@@ -430,12 +402,13 @@ pub mod tests {
 				assert_eq!(retrieved_key, generated_key);
 			}
 		}
+		drop(runtime);
 	}
 
 	#[test]
 	fn server_key_generation_and_message_signing_works_over_network_with_3_nodes() {
 		//::logger::init_log();
-		let (key_servers, _) = make_key_servers(6100, 3);
+		let (key_servers, _, runtime) = make_key_servers(6100, 3);
 
 		let test_cases = [0, 1, 2];
 		for threshold in &test_cases {
@@ -455,12 +428,13 @@ pub mod tests {
 			// check signature
 			assert_eq!(math::verify_schnorr_signature(&server_public, &(signature_c, signature_s), &message_hash), Ok(true));
 		}
+		drop(runtime);
 	}
 
 	#[test]
 	fn decryption_session_is_delegated_when_node_does_not_have_key_share() {
 		//::logger::init_log();
-		let (key_servers, _) = make_key_servers(6110, 3);
+		let (key_servers, _, runtime) = make_key_servers(6110, 3);
 
 		// generate document key
 		let threshold = 0;
@@ -477,12 +451,13 @@ pub mod tests {
 		let retrieved_key = key_servers[0].restore_document_key(&document, &signature.into()).unwrap();
 		let retrieved_key = crypto::ecies::decrypt(&secret, &DEFAULT_MAC, &retrieved_key).unwrap();
 		assert_eq!(retrieved_key, generated_key);
+		drop(runtime);
 	}
 
 	#[test]
 	fn schnorr_signing_session_is_delegated_when_node_does_not_have_key_share() {
 		//::logger::init_log();
-		let (key_servers, _) = make_key_servers(6114, 3);
+		let (key_servers, _, runtime) = make_key_servers(6114, 3);
 		let threshold = 1;
 
 		// generate server key
@@ -503,12 +478,13 @@ pub mod tests {
 
 		// check signature
 		assert_eq!(math::verify_schnorr_signature(&server_public, &(signature_c, signature_s), &message_hash), Ok(true));
+		drop(runtime);
 	}
 
 	#[test]
 	fn ecdsa_signing_session_is_delegated_when_node_does_not_have_key_share() {
 		//::logger::init_log();
-		let (key_servers, _) = make_key_servers(6117, 4);
+		let (key_servers, _, runtime) = make_key_servers(6117, 4);
 		let threshold = 1;
 
 		// generate server key
@@ -528,6 +504,7 @@ pub mod tests {
 
 		// check signature
 		assert!(verify_public(&server_public, &signature.into(), &message_hash).unwrap());
+		drop(runtime);
 	}
 
 	#[test]

secret_store/src/key_server_cluster/client_sessions/generation_session.rs

@@ -1367,12 +1367,12 @@ pub mod tests {
 			let clusters_clone = clusters.clone();
 
 			// establish connections
-			loop_until(&mut core, CONN_TIMEOUT, move || clusters_clone.iter().all(all_connections_established));
+			loop_until(&core.executor(), CONN_TIMEOUT, move || clusters_clone.iter().all(all_connections_established));
 
 			// run session to completion
 			let session_id = SessionId::default();
 			let session = clusters[0].client().new_generation_session(session_id, Default::default(), Default::default(), threshold).unwrap();
-			loop_until(&mut core, SESSION_TIMEOUT, move || session.joint_public_and_secret().is_some());
+			loop_until(&core.executor(), SESSION_TIMEOUT, move || session.joint_public_and_secret().is_some());
 		}
 	}