Bloom upgrade in beta (#2609)

* bloom expand & upgrade logic

* remove arc

* progress tracker

* comment about magic number

* progress tick initial

ethcore/src/migrations/account_bloom.rs

@@ -17,7 +17,7 @@
 //! Bloom upgrade
 
 use client::{DB_COL_EXTRA, DB_COL_HEADERS, DB_NO_OF_COLUMNS, DB_COL_STATE, DB_COL_ACCOUNT_BLOOM};
-use state_db::{ACCOUNT_BLOOM_SPACE, DEFAULT_ACCOUNT_PRESET, StateDB, ACCOUNT_BLOOM_HASHCOUNT_KEY};
+use state_db::{ACCOUNT_BLOOM_SPACE, DEFAULT_ACCOUNT_PRESET, StateDB, ACCOUNT_BLOOM_HASHCOUNT_KEY, ACCOUNT_BLOOM_SPACE_KEY};
 use util::trie::TrieDB;
 use views::HeaderView;
 use bloomfilter::Bloom;
@@ -26,6 +26,7 @@ use util::journaldb;
 use util::{H256, FixedHash, BytesConvertable};
 use util::{Database, DatabaseConfig, DBTransaction, CompactionProfile};
 use std::path::Path;
+use byteorder::{LittleEndian, ByteOrder};
 
 fn check_bloom_exists(db: &Database) -> bool {
 	let hash_count_entry = db.get(DB_COL_ACCOUNT_BLOOM, ACCOUNT_BLOOM_HASHCOUNT_KEY)
@@ -34,10 +35,22 @@ fn check_bloom_exists(db: &Database) -> bool {
 	hash_count_entry.is_some()
 }
 
+fn check_space_match(db: &Database) -> Result<(), usize> {
+	let db_space = db.get(DB_COL_ACCOUNT_BLOOM, ACCOUNT_BLOOM_SPACE_KEY)
+		.expect("Low-level database error")
+		.map(|val| LittleEndian::read_u64(&val[..]) as usize)
+		// this was the initial size of the bloom which was not written in the database
+		.unwrap_or(1048576);
+
+	if db_space == ACCOUNT_BLOOM_SPACE { Ok(()) } else { Err(db_space) }
+}
+
 /// Account bloom upgrade routine. If bloom already present, does nothing.
 /// If database empty (no best block), does nothing.
 /// Can be called on upgraded database with no issues (will do nothing).
 pub fn upgrade_account_bloom(db_path: &Path) -> Result<(), Error> {
+	let mut progress = ::util::migration::Progress::default();
+
 	let path = try!(db_path.to_str().ok_or(Error::MigrationImpossible));
 	trace!(target: "migration", "Account bloom upgrade at {:?}", db_path);
 
@@ -67,14 +80,34 @@ pub fn upgrade_account_bloom(db_path: &Path) -> Result<(), Error> {
 	};
 	let state_root = HeaderView::new(&best_block_header).state_root();
 
-	if check_bloom_exists(&source) {
-		// bloom already exists, nothing to do
-		trace!(target: "migration", "Bloom already present, skipping");
-		return Ok(())
+	let db = ::std::sync::Arc::new(source);
+	let batch = DBTransaction::new(&db);
+
+	if check_bloom_exists(&*db) {
+		match check_space_match(&*db) {
+			Ok(_) => {
+				// bloom already exists and desired and stored spaces match, nothing to do
+				trace!(target: "migration", "Bloom already present of the right space, skipping");
+				return Ok(())
+			},
+			Err(wrong_size) => {
+				// nullify existing bloom entries
+				trace!(target: "migration", "Clearing old bloom of space {}", &wrong_size);
+				let mut key = [0u8; 8];
+				for idx in 0..(wrong_size as u64/8) {
+					LittleEndian::write_u64(&mut key, idx);
+					try!(batch.put(DB_COL_ACCOUNT_BLOOM, &key, &[0u8; 8]));
+
+					if idx % 10000 == 1 { progress.tick(); };
+				}
+
+				LittleEndian::write_u64(&mut key, ACCOUNT_BLOOM_SPACE as u64);
+				try!(batch.put(DB_COL_ACCOUNT_BLOOM, ACCOUNT_BLOOM_SPACE_KEY, &key));
+			},
+		}
 	}
 
-	println!("Adding accounts bloom (one-time upgrade)");
-	let db = ::std::sync::Arc::new(source);
+	println!("Adding/expanding accounts bloom (one-time upgrade)");
 	let bloom_journal = {
 		let mut bloom = Bloom::new(ACCOUNT_BLOOM_SPACE, DEFAULT_ACCOUNT_PRESET);
 		// no difference what algorithm is passed, since there will be no writes
@@ -86,6 +119,7 @@ pub fn upgrade_account_bloom(db_path: &Path) -> Result<(), Error> {
 		for (ref account_key, _) in account_trie.iter() {
 			let account_key_hash = H256::from_slice(&account_key);
 			bloom.set(account_key_hash.as_slice());
+			progress.tick();
 		}
 
 		bloom.drain_journal()
@@ -93,12 +127,10 @@ pub fn upgrade_account_bloom(db_path: &Path) -> Result<(), Error> {
 
 	trace!(target: "migration", "Generated {} bloom updates", bloom_journal.entries.len());
 
-	let batch = DBTransaction::new(&db);
 	try!(StateDB::commit_bloom(&batch, bloom_journal).map_err(|_| Error::Custom("Failed to commit bloom".to_owned())));
 	try!(db.write(batch));
 
 	trace!(target: "migration", "Finished bloom update");
 
-
 	Ok(())
 }

ethcore/src/state_db.rs

@@ -87,10 +87,11 @@ pub struct StateDB {
 	commit_number: Option<BlockNumber>,
 }
 
-pub const ACCOUNT_BLOOM_SPACE: usize = 1048576;
-pub const DEFAULT_ACCOUNT_PRESET: usize = 1000000;
+pub const ACCOUNT_BLOOM_SPACE: usize = 4 * 1048576;
+pub const DEFAULT_ACCOUNT_PRESET: usize = 100 * 1000000;
 
 pub const ACCOUNT_BLOOM_HASHCOUNT_KEY: &'static [u8] = b"account_hash_count";
+pub const ACCOUNT_BLOOM_SPACE_KEY: &'static [u8] = b"account_space";
 
 impl StateDB {
 	/// Loads accounts bloom from the database
@@ -138,13 +139,11 @@ impl StateDB {
 	}
 
 	pub fn check_account_bloom(&self, address: &Address) -> bool {
-		trace!(target: "account_bloom", "Check account bloom: {:?}", address);
 		let bloom = self.account_bloom.lock();
 		bloom.check(address.sha3().as_slice())
 	}
 
 	pub fn note_account_bloom(&self, address: &Address) {
-		trace!(target: "account_bloom", "Note account bloom: {:?}", address);
 		let mut bloom = self.account_bloom.lock();
 		bloom.set(address.sha3().as_slice());
 	}