Multithreaded snapshot creation (#9239)

* Add Progress to Snapshot Secondary chunks creation

* Use half of CPUs to multithread snapshot creation

* Use env var to define number of threads

* info to debug logs

* Add Snapshot threads as CLI option

* Randomize chunks per thread

* Remove randomness, add debugging

* Add warning

* Add tracing

* Use parity-common fix seek branch

* Fix log

* Fix tests

* Fix tests

* PR Grumbles

* PR Grumble II

* Update Cargo.lock

* PR Grumbles

* Default snapshot threads to half number of CPUs

* Fix default snapshot threads // min 1

parity/cli/mod.rs

@@ -874,6 +874,10 @@ usage! {
 			"--no-periodic-snapshot",
 			"Disable automated snapshots which usually occur once every 10000 blocks.",
 
+			ARG arg_snapshot_threads: (Option<usize>) = None, or |c: &Config| c.snapshots.as_ref()?.processing_threads,
+			"--snapshot-threads=[NUM]",
+			"Enables multiple threads for snapshots creation.",
+
 		["Whisper Options"]
 			FLAG flag_whisper: (bool) = false, or |c: &Config| c.whisper.as_ref()?.enabled,
 			"--whisper",
@@ -1355,6 +1359,7 @@ struct Footprint {
 #[serde(deny_unknown_fields)]
 struct Snapshots {
 	disable_periodic: Option<bool>,
+	processing_threads: Option<usize>,
 }
 
 #[derive(Default, Debug, PartialEq, Deserialize)]
@@ -1788,6 +1793,7 @@ mod tests {
 			arg_export_state_at: "latest".into(),
 			arg_snapshot_at: "latest".into(),
 			flag_no_periodic_snapshot: false,
+			arg_snapshot_threads: None,
 
 			// -- Light options.
 			arg_on_demand_retry_count: Some(15),
@@ -2046,6 +2052,7 @@ mod tests {
 			}),
 			snapshots: Some(Snapshots {
 				disable_periodic: Some(true),
+				processing_threads: None,
 			}),
 			misc: Some(Misc {
 				logging: Some("own_tx=trace".into()),

parity/configuration.rs

@@ -30,8 +30,10 @@ use sync::{NetworkConfiguration, validate_node_url, self};
 use ethcore::ethstore::ethkey::{Secret, Public};
 use ethcore::client::{VMType};
 use ethcore::miner::{stratum, MinerOptions};
+use ethcore::snapshot::SnapshotConfiguration;
 use ethcore::verification::queue::VerifierSettings;
 use miner::pool;
+use num_cpus;
 
 use rpc::{IpcConfiguration, HttpConfiguration, WsConfiguration};
 use parity_rpc::NetworkSettings;
@@ -125,6 +127,7 @@ impl Configuration {
 		let update_policy = self.update_policy()?;
 		let logger_config = self.logger_config();
 		let ws_conf = self.ws_config()?;
+		let snapshot_conf = self.snapshot_config()?;
 		let http_conf = self.http_config()?;
 		let ipc_conf = self.ipc_config()?;
 		let net_conf = self.net_config()?;
@@ -298,6 +301,7 @@ impl Configuration {
 				file_path: self.args.arg_snapshot_file.clone(),
 				kind: snapshot::Kind::Take,
 				block_at: to_block_id(&self.args.arg_snapshot_at)?,
+				snapshot_conf: snapshot_conf,
 			};
 			Cmd::Snapshot(snapshot_cmd)
 		} else if self.args.cmd_restore {
@@ -314,6 +318,7 @@ impl Configuration {
 				file_path: self.args.arg_restore_file.clone(),
 				kind: snapshot::Kind::Restore,
 				block_at: to_block_id("latest")?, // unimportant.
+				snapshot_conf: snapshot_conf,
 			};
 			Cmd::Snapshot(restore_cmd)
 		} else if self.args.cmd_export_hardcoded_sync {
@@ -349,6 +354,7 @@ impl Configuration {
 				gas_price_percentile: self.args.arg_gas_price_percentile,
 				poll_lifetime: self.args.arg_poll_lifetime,
 				ws_conf: ws_conf,
+				snapshot_conf: snapshot_conf,
 				http_conf: http_conf,
 				ipc_conf: ipc_conf,
 				net_conf: net_conf,
@@ -374,7 +380,6 @@ impl Configuration {
 				private_tx_enabled,
 				name: self.args.arg_identity,
 				custom_bootnodes: self.args.arg_bootnodes.is_some(),
-				no_periodic_snapshot: self.args.flag_no_periodic_snapshot,
 				check_seal: !self.args.flag_no_seal_check,
 				download_old_blocks: !self.args.flag_no_ancient_blocks,
 				verifier_settings: verifier_settings,
@@ -892,6 +897,18 @@ impl Configuration {
 		Ok((provider_conf, encryptor_conf, self.args.flag_private_enabled))
 	}
 
+	fn snapshot_config(&self) -> Result<SnapshotConfiguration, String> {
+		let conf = SnapshotConfiguration {
+			no_periodic: self.args.flag_no_periodic_snapshot,
+			processing_threads: match self.args.arg_snapshot_threads {
+				Some(threads) if threads > 0 => threads,
+				_ => ::std::cmp::max(1, num_cpus::get() / 2),
+			},
+		};
+
+		Ok(conf)
+	}
+
 	fn network_settings(&self) -> Result<NetworkSettings, String> {
 		let http_conf = self.http_config()?;
 		let net_addresses = self.net_addresses()?;
@@ -1400,7 +1417,7 @@ mod tests {
 			name: "".into(),
 			custom_bootnodes: false,
 			fat_db: Default::default(),
-			no_periodic_snapshot: false,
+			snapshot_conf: Default::default(),
 			stratum: None,
 			check_seal: true,
 			download_old_blocks: true,

parity/run.rs

@@ -25,7 +25,7 @@ use ethcore::account_provider::{AccountProvider, AccountProviderSettings};
 use ethcore::client::{BlockId, CallContract, Client, Mode, DatabaseCompactionProfile, VMType, BlockChainClient, BlockInfo};
 use ethcore::ethstore::ethkey;
 use ethcore::miner::{stratum, Miner, MinerService, MinerOptions};
-use ethcore::snapshot;
+use ethcore::snapshot::{self, SnapshotConfiguration};
 use ethcore::spec::{SpecParams, OptimizeFor};
 use ethcore::verification::queue::VerifierSettings;
 use ethcore_logger::{Config as LogConfig, RotatingLogger};
@@ -119,7 +119,7 @@ pub struct RunCmd {
 	pub name: String,
 	pub custom_bootnodes: bool,
 	pub stratum: Option<stratum::Options>,
-	pub no_periodic_snapshot: bool,
+	pub snapshot_conf: SnapshotConfiguration,
 	pub check_seal: bool,
 	pub download_old_blocks: bool,
 	pub verifier_settings: VerifierSettings,
@@ -539,6 +539,7 @@ fn execute_impl<Cr, Rr>(cmd: RunCmd, logger: Arc<RotatingLogger>, on_client_rq:
 
 	client_config.queue.verifier_settings = cmd.verifier_settings;
 	client_config.transaction_verification_queue_size = ::std::cmp::max(2048, txpool_size / 4);
+	client_config.snapshot = cmd.snapshot_conf.clone();
 
 	// set up bootnodes
 	let mut net_conf = cmd.net_conf;
@@ -786,7 +787,7 @@ fn execute_impl<Cr, Rr>(cmd: RunCmd, logger: Arc<RotatingLogger>, on_client_rq:
 	});
 
 	// the watcher must be kept alive.
-	let watcher = match cmd.no_periodic_snapshot {
+	let watcher = match cmd.snapshot_conf.no_periodic {
 		true => None,
 		false => {
 			let sync = sync_provider.clone();

parity/snapshot.rs

@@ -22,7 +22,7 @@ use std::sync::Arc;
 
 use hash::keccak;
 use ethcore::account_provider::AccountProvider;
-use ethcore::snapshot::{Progress, RestorationStatus, SnapshotService as SS};
+use ethcore::snapshot::{Progress, RestorationStatus, SnapshotConfiguration, SnapshotService as SS};
 use ethcore::snapshot::io::{SnapshotReader, PackedReader, PackedWriter};
 use ethcore::snapshot::service::Service as SnapshotService;
 use ethcore::client::{Mode, DatabaseCompactionProfile, VMType};
@@ -62,6 +62,7 @@ pub struct SnapshotCommand {
 	pub file_path: Option<String>,
 	pub kind: Kind,
 	pub block_at: BlockId,
+	pub snapshot_conf: SnapshotConfiguration,
 }
 
 // helper for reading chunks from arbitrary reader and feeding them into the
@@ -165,7 +166,7 @@ impl SnapshotCommand {
 		execute_upgrades(&self.dirs.base, &db_dirs, algorithm, &self.compaction)?;
 
 		// prepare client config
-		let client_config = to_client_config(
+		let mut client_config = to_client_config(
 			&self.cache_config,
 			spec.name.to_lowercase(),
 			Mode::Active,
@@ -180,6 +181,8 @@ impl SnapshotCommand {
 			true,
 		);
 
+		client_config.snapshot = self.snapshot_conf;
+
 		let restoration_db_handler = db::restoration_db_handler(&client_path, &client_config);
 		let client_db = restoration_db_handler.open(&client_path)
 			.map_err(|e| format!("Failed to open database {:?}", e))?;