Pause pruning while snapshotting (#11178)

* WIP. Typos and logging.

* Format todos

* Pause pruning while a snapshot is under way
Logs, docs and todos

* Allocate memory for the full chunk

* Name snapshotting threads

* Ensure `taking_snapshot` is set to false whenever and however `take_snapshot`returns
Rename `take_at` to `request_snapshot_at`
Cleanup

* Let "in_progress" deletion fail
Fix tests

* Just use an atomic

* Review grumbles

* Finish the sentence

* Resolve a few todos and clarify comments.

* Calculate progress rate since last update

* Lockfile

* Fix tests

* typo

* Reinstate default snapshotting frequency
Cut down on the logging noise

* address grumble

* Log memory use with `journal_size()` and explain why.

Author: dvdplm

Cargo.lock

@@ -51,6 +51,7 @@ rayon = "1.1"
 registrar = { path = "../util/registrar" }
 rlp = "0.4.0"
 rustc-hex = "2"
+scopeguard = "1.0.0"
 serde = "1.0"
 serde_derive = "1.0"
 snapshot = { path = "snapshot" }

ethcore/Cargo.toml

@@ -270,14 +270,12 @@ where
 			ClientIoMessage::TakeSnapshot(num) => {
 				let client = self.client.clone();
 				let snapshot = self.snapshot.clone();
-
 				let res = thread::Builder::new().name("Periodic Snapshot".into()).spawn(move || {
 					if let Err(e) = snapshot.take_snapshot(&*client, num) {
 						match e {
 							EthcoreError::Snapshot(SnapshotError::SnapshotAborted) => info!("Snapshot aborted"),
 							_ => warn!("Failed to take snapshot at block #{}: {}", num, e),
 						}
-
 					}
 				});
 

ethcore/service/src/service.rs

@@ -33,6 +33,7 @@ rand_xorshift = "0.2"
 parking_lot = "0.9"
 rlp = "0.4.2"
 rlp_derive = { path = "../../util/rlp-derive" }
+scopeguard = "1.0.0"
 snappy = { package = "parity-snappy", version ="0.1.0" }
 state-db = { path = "../state-db" }
 trie-db = "0.15.0"

ethcore/snapshot/Cargo.toml

@@ -48,7 +48,7 @@ fn encoding_basic() {
 
 	let thin_rlp = ::rlp::encode(&account);
 	assert_eq!(::rlp::decode::<BasicAccount>(&thin_rlp).unwrap(), account);
-	let p = Progress::default();
+	let p = Progress::new();
 	let fat_rlps = to_fat_rlps(&keccak(&addr), &account, &AccountDB::from_hash(db.as_hash_db(), keccak(addr)), &mut Default::default(), usize::max_value(), usize::max_value(), &p).unwrap();
 	let fat_rlp = Rlp::new(&fat_rlps[0]).at(1).unwrap();
 	assert_eq!(from_fat_rlp(&mut AccountDBMut::from_hash(db.as_hash_db_mut(), keccak(addr)), fat_rlp, H256::zero()).unwrap().0, account);
@@ -69,7 +69,7 @@ fn encoding_version() {
 
 	let thin_rlp = ::rlp::encode(&account);
 	assert_eq!(::rlp::decode::<BasicAccount>(&thin_rlp).unwrap(), account);
-	let p = Progress::default();
+	let p = Progress::new();
 	let fat_rlps = to_fat_rlps(&keccak(&addr), &account, &AccountDB::from_hash(db.as_hash_db(), keccak(addr)), &mut Default::default(), usize::max_value(), usize::max_value(), &p).unwrap();
 	let fat_rlp = Rlp::new(&fat_rlps[0]).at(1).unwrap();
 	assert_eq!(from_fat_rlp(&mut AccountDBMut::from_hash(db.as_hash_db_mut(), keccak(addr)), fat_rlp, H256::zero()).unwrap().0, account);
@@ -96,7 +96,7 @@ fn encoding_storage() {
 	let thin_rlp = ::rlp::encode(&account);
 	assert_eq!(::rlp::decode::<BasicAccount>(&thin_rlp).unwrap(), account);
 
-	let p = Progress::default();
+	let p = Progress::new();
 
 	let fat_rlp = to_fat_rlps(&keccak(&addr), &account, &AccountDB::from_hash(db.as_hash_db(), keccak(addr)), &mut Default::default(), usize::max_value(), usize::max_value(), &p).unwrap();
 	let fat_rlp = Rlp::new(&fat_rlp[0]).at(1).unwrap();
@@ -124,7 +124,7 @@ fn encoding_storage_split() {
 	let thin_rlp = ::rlp::encode(&account);
 	assert_eq!(::rlp::decode::<BasicAccount>(&thin_rlp).unwrap(), account);
 
-	let p = Progress::default();
+	let p = Progress::new();
 	let fat_rlps = to_fat_rlps(&keccak(addr), &account, &AccountDB::from_hash(db.as_hash_db(), keccak(addr)), &mut Default::default(), 500, 1000, &p).unwrap();
 	let mut root = KECCAK_NULL_RLP;
 	let mut restored_account = None;
@@ -170,8 +170,8 @@ fn encoding_code() {
 	};
 
 	let mut used_code = HashSet::new();
-	let p1 = Progress::default();
-	let p2 = Progress::default();
+	let p1 = Progress::new();
+	let p2 = Progress::new();
 	let fat_rlp1 = to_fat_rlps(&keccak(&addr1), &account1, &AccountDB::from_hash(db.as_hash_db(), keccak(addr1)), &mut used_code, usize::max_value(), usize::max_value(), &p1).unwrap();
 	let fat_rlp2 = to_fat_rlps(&keccak(&addr2), &account2, &AccountDB::from_hash(db.as_hash_db(), keccak(addr2)), &mut used_code, usize::max_value(), usize::max_value(), &p2).unwrap();
 	assert_eq!(used_code.len(), 1);

ethcore/snapshot/snapshot-tests/src/account.rs

@@ -27,7 +27,8 @@ use client_traits::ChainInfo;
 use common_types::{
 	ids::BlockId,
 	basic_account::BasicAccount,
-	errors::EthcoreError
+	errors::EthcoreError,
+	snapshot::Progress,
 };
 use engine::Engine;
 use ethcore::client::Client;
@@ -145,7 +146,7 @@ pub fn snap(client: &Client) -> (Box<dyn SnapshotReader>, TempDir) {
 	let tempdir = TempDir::new("").unwrap();
 	let path = tempdir.path().join("file");
 	let writer = PackedWriter::new(&path).unwrap();
-	let progress = Default::default();
+	let progress = Progress::new();
 
 	let hash = client.chain_info().best_block_hash;
 	client.take_snapshot(writer, BlockId::Hash(hash), &progress).unwrap();

ethcore/snapshot/snapshot-tests/src/helpers.rs

@@ -74,7 +74,7 @@ fn chunk_and_restore(amount: u64) {
 		&bc,
 		best_hash,
 		&writer,
-		&Progress::default()
+		&Progress::new()
 	).unwrap();
 
 	let manifest = ManifestData {

ethcore/snapshot/snapshot-tests/src/proof_of_work.rs

@@ -278,7 +278,7 @@ fn keep_ancient_blocks() {
 		&bc,
 		best_hash,
 		&writer,
-		&Progress::default()
+		&Progress::new()
 	).unwrap();
 	let state_db = client.state_db().journal_db().boxed_clone();
 	let start_header = bc.block_header_data(&best_hash).unwrap();
@@ -287,7 +287,7 @@ fn keep_ancient_blocks() {
 		state_db.as_hash_db(),
 		&state_root,
 		&writer,
-		&Progress::default(),
+		&Progress::new(),
 		None,
 		0
 	).unwrap();

ethcore/snapshot/snapshot-tests/src/service.rs

@@ -62,7 +62,7 @@ fn snap_and_restore() {
 
 	let mut state_hashes = Vec::new();
 	for part in 0..SNAPSHOT_SUBPARTS {
-		let mut hashes = chunk_state(&old_db, &state_root, &writer, &Progress::default(), Some(part), 0).unwrap();
+		let mut hashes = chunk_state(&old_db, &state_root, &writer, &Progress::new(), Some(part), 0).unwrap();
 		state_hashes.append(&mut hashes);
 	}
 
@@ -133,7 +133,7 @@ fn get_code_from_prev_chunk() {
 	let mut make_chunk = |acc, hash| {
 		let mut db = journaldb::new_memory_db();
 		AccountDBMut::from_hash(&mut db, hash).insert(EMPTY_PREFIX, &code[..]);
-		let p = Progress::default();
+		let p = Progress::new();
 		let fat_rlp = to_fat_rlps(&hash, &acc, &AccountDB::from_hash(&db, hash), &mut used_code, usize::max_value(), usize::max_value(), &p).unwrap();
 		let mut stream = RlpStream::new_list(1);
 		stream.append_raw(&fat_rlp[0], 1);
@@ -178,7 +178,7 @@ fn checks_flag() {
 	let state_root = producer.state_root();
 	let writer = Mutex::new(PackedWriter::new(&snap_file).unwrap());
 
-	let state_hashes = chunk_state(&old_db, &state_root, &writer, &Progress::default(), None, 0).unwrap();
+	let state_hashes = chunk_state(&old_db, &state_root, &writer, &Progress::new(), None, 0).unwrap();
 
 	writer.into_inner().finish(ManifestData {
 		version: 2,

ethcore/snapshot/snapshot-tests/src/state.rs

@@ -41,8 +41,8 @@ impl Oracle for TestOracle {
 
 struct TestBroadcast(Option<u64>);
 impl Broadcast for TestBroadcast {
-	fn take_at(&self, num: Option<u64>) {
-		if num != self.0 {
+	fn request_snapshot_at(&self, num: u64) {
+		if Some(num) != self.0 {
 			panic!("Watcher broadcast wrong number. Expected {:?}, found {:?}", self.0, num);
 		}
 	}