Accounts bloom (#2357)

* proper bloom

* incremental bloom updates

* crate update

* return of the column

* fix n^2 byteorder write

* add notes to funs

* working bloom commits

* Optimizations

* bloom diag

* migration basic

* migration ongoing

* migration finalizing

* mingration api workarounds

* fix test_client setups

* snapshot bloom update

* review fixes

* just forward keys in the migration

* migration extra tracing

* fix migration path

* remove close pray

* review issues

ethcore/src/state_db.rs

@@ -18,8 +18,12 @@ use lru_cache::LruCache;
 use util::journaldb::JournalDB;
 use util::hash::{H256};
 use util::hashdb::HashDB;
-use util::{Arc, Address, DBTransaction, UtilError, Mutex};
+use util::{Arc, Address, DBTransaction, UtilError, Mutex, Hashable, BytesConvertable};
 use account::Account;
+use bloomfilter::{Bloom, BloomJournal};
+use util::Database;
+use client::DB_COL_ACCOUNT_BLOOM;
+use byteorder::{LittleEndian, ByteOrder};
 
 const STATE_CACHE_ITEMS: usize = 65536;
 
@@ -39,22 +43,86 @@ pub struct StateDB {
 	account_cache: Arc<Mutex<AccountCache>>,
 	cache_overlay: Vec<(Address, Option<Account>)>,
 	is_canon: bool,
+	account_bloom: Arc<Mutex<Bloom>>,
 }
 
+pub const ACCOUNT_BLOOM_SPACE: usize = 1048576;
+pub const DEFAULT_ACCOUNT_PRESET: usize = 1000000;
+
+pub const ACCOUNT_BLOOM_HASHCOUNT_KEY: &'static [u8] = b"account_hash_count";
+
 impl StateDB {
+	pub fn load_bloom(db: &Database) -> Bloom {
+		let hash_count_entry = db.get(DB_COL_ACCOUNT_BLOOM, ACCOUNT_BLOOM_HASHCOUNT_KEY)
+			.expect("Low-level database error");
+
+		if hash_count_entry.is_none() {
+			return Bloom::new(ACCOUNT_BLOOM_SPACE, DEFAULT_ACCOUNT_PRESET);
+		}
+		let hash_count_bytes = hash_count_entry.unwrap();
+		assert_eq!(hash_count_bytes.len(), 1);
+		let hash_count = hash_count_bytes[0];
+
+		let mut bloom_parts = vec![0u64; ACCOUNT_BLOOM_SPACE / 8];
+		let mut key = [0u8; 8];
+		for i in 0..ACCOUNT_BLOOM_SPACE / 8 {
+			LittleEndian::write_u64(&mut key, i as u64);
+			bloom_parts[i] = db.get(DB_COL_ACCOUNT_BLOOM, &key).expect("low-level database error")
+				.and_then(|val| Some(LittleEndian::read_u64(&val[..])))
+				.unwrap_or(0u64);
+		}
+
+		let bloom = Bloom::from_parts(&bloom_parts, hash_count as u32);
+		trace!(target: "account_bloom", "Bloom is {:?} full, hash functions count = {:?}", bloom.how_full(), hash_count);
+		bloom
+	}
+
 	/// Create a new instance wrapping `JournalDB`
 	pub fn new(db: Box<JournalDB>) -> StateDB {
+		let bloom = Self::load_bloom(db.backing());
 		StateDB {
 			db: db,
 			account_cache: Arc::new(Mutex::new(AccountCache { accounts: LruCache::new(STATE_CACHE_ITEMS) })),
 			cache_overlay: Vec::new(),
 			is_canon: false,
+			account_bloom: Arc::new(Mutex::new(bloom)),
 		}
 	}
 
+	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());
+	}
+
+	pub fn commit_bloom(batch: &DBTransaction, journal: BloomJournal) -> Result<(), UtilError> {
+		assert!(journal.hash_functions <= 255);
+		try!(batch.put(DB_COL_ACCOUNT_BLOOM, ACCOUNT_BLOOM_HASHCOUNT_KEY, &vec![journal.hash_functions as u8]));
+		let mut key = [0u8; 8];
+		let mut val = [0u8; 8];
+
+		for (bloom_part_index, bloom_part_value) in journal.entries {
+			LittleEndian::write_u64(&mut key, bloom_part_index as u64);
+			LittleEndian::write_u64(&mut val, bloom_part_value);
+			try!(batch.put(DB_COL_ACCOUNT_BLOOM, &key, &val));
+		}
+		Ok(())
+	}
+
 	/// Commit all recent insert operations and canonical historical commits' removals from the
 	/// old era to the backing database, reverting any non-canonical historical commit's inserts.
 	pub fn commit(&mut self, batch: &DBTransaction, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> {
+		{
+			let mut bloom_lock = self.account_bloom.lock();
+			try!(Self::commit_bloom(batch, bloom_lock.drain_journal()));
+		}
+
 		let records = try!(self.db.commit(batch, now, id, end));
 		if self.is_canon {
 			self.commit_cache();
@@ -81,6 +149,7 @@ impl StateDB {
 			account_cache: self.account_cache.clone(),
 			cache_overlay: Vec::new(),
 			is_canon: false,
+			account_bloom: self.account_bloom.clone(),
 		}
 	}
 
@@ -91,6 +160,7 @@ impl StateDB {
 			account_cache: self.account_cache.clone(),
 			cache_overlay: Vec::new(),
 			is_canon: true,
+			account_bloom: self.account_bloom.clone(),
 		}
 	}