Handle reorganizations in the state cache

Author: arkpar

ethcore/src/client/client.rs

@@ -304,8 +304,7 @@ impl Client {
 		// Enact Verified Block
 		let parent = chain_has_parent.unwrap();
 		let last_hashes = self.build_last_hashes(header.parent_hash().clone());
-		let is_canon = header.parent_hash() == &chain.best_block_hash();
-		let db = if is_canon { self.state_db.lock().boxed_clone_canon() } else { self.state_db.lock().boxed_clone() };
+		let db = self.state_db.lock().boxed_clone_canon(&header.parent_hash());
 
 		let enact_result = enact_verified(block, engine, self.tracedb.read().tracing_enabled(), db, &parent, last_hashes, self.factories.clone());
 		if let Err(e) = enact_result {
@@ -443,8 +442,8 @@ impl Client {
 			.collect();
 
 		//let traces = From::from(block.traces().clone().unwrap_or_else(Vec::new));
-
 		let mut batch = DBTransaction::new(&self.db.read());
+
 		// CHECK! I *think* this is fine, even if the state_root is equal to another
 		// already-imported block of the same number.
 		// TODO: Prove it with a test.
@@ -459,6 +458,8 @@ impl Client {
 			enacted: route.enacted.clone(),
 			retracted: route.retracted.len()
 		});
+		let is_canon = route.omitted.len() == 0;
+		state.update_cache(&route.enacted, &route.retracted, is_canon);
 		// Final commit to the DB
 		self.db.read().write_buffered(batch);
 		chain.commit();
@@ -533,9 +534,11 @@ impl Client {
 
 	/// Get a copy of the best block's state.
 	pub fn state(&self) -> State {
+		let header = self.best_block_header();
+		let header = HeaderView::new(&header);
 		State::from_existing(
-			self.state_db.lock().boxed_clone(),
-			HeaderView::new(&self.best_block_header()).state_root(),
+			self.state_db.lock().boxed_clone_canon(&header.hash()),
+			header.state_root(),
 			self.engine.account_start_nonce(),
 			self.factories.clone())
 		.expect("State root of best block header always valid.")

ethcore/src/state/mod.rs

@@ -486,7 +486,7 @@ impl State {
 		let mut addresses = self.cache.borrow_mut();
 		trace!("Committing cache {:?} entries", addresses.len());
 		for (address, a) in addresses.drain().filter(|&(_, ref a)| !a.is_dirty()) {
-			self.db.cache_account(address, a.account);
+			self.db.cache_account(address, a.account, a.state == AccountState::Commited);
 		}
 	}
 

ethcore/src/state_db.rs

@@ -14,17 +14,21 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::collections::{VecDeque, HashSet};
 use lru_cache::LruCache;
 use util::journaldb::JournalDB;
 use util::hash::{H256};
 use util::hashdb::HashDB;
 use state::Account;
+use header::BlockNumber;
 use util::{Arc, Address, Database, DBTransaction, UtilError, Mutex, Hashable};
 use bloom_journal::{Bloom, BloomJournal};
 use db::COL_ACCOUNT_BLOOM;
 use byteorder::{LittleEndian, ByteOrder};
 
 const STATE_CACHE_ITEMS: usize = 65536;
+const STATE_CACHE_BLOCKS: usize = 8;
+
 
 pub const ACCOUNT_BLOOM_SPACE: usize = 1048576;
 pub const DEFAULT_ACCOUNT_PRESET: usize = 1000000;
@@ -34,6 +38,24 @@ pub const ACCOUNT_BLOOM_HASHCOUNT_KEY: &'static [u8] = b"account_hash_count";
 struct AccountCache {
 	/// DB Account cache. `None` indicates that account is known to be missing.
 	accounts: LruCache<Address, Option<Account>>,
+	/// Accounts changed in recent blocks. Ordered by block number.
+	modifications: VecDeque<BlockChanges>,
+}
+
+struct CacheQueueItem {
+	address: Address,
+	account: Option<Account>,
+	modified: bool,
+}
+
+#[derive(Debug)]
+// Accumulates a list of accounts changed in a block.
+struct BlockChanges {
+	number: BlockNumber,
+	hash: H256,
+	parent: H256,
+	accounts: HashSet<Address>,
+	is_canon: bool,
 }
 
 /// State database abstraction.
@@ -43,27 +65,23 @@ struct AccountCache {
 /// on commit.
 /// For non-canonical clones cache is cleared on commit.
 pub struct StateDB {
+	// Backing database.
 	db: Box<JournalDB>,
+	// Shared canonical state cache.
 	account_cache: Arc<Mutex<AccountCache>>,
-	cache_overlay: Vec<(Address, Option<Account>)>,
-	is_canon: bool,
+	// Local pending cache changes.
+	cache_overlay: Vec<CacheQueueItem>,
+	// Shared account bloom. Does not handle chain reorganizations.
 	account_bloom: Arc<Mutex<Bloom>>,
+	// Hash of the block on top of which this instance was created
+	parent_hash: Option<H256>,
+	// Hash of the committing block
+	commit_hash: Option<H256>,
+	// Number of the committing block
+	commit_number: Option<BlockNumber>,
 }
 
 impl StateDB {
-
-	/// 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)),
-		}
-	}
-
 	/// Loads accounts bloom from the database
 	/// This bloom is used to handle request for the non-existant account fast
 	pub fn load_bloom(db: &Database) -> Bloom {
@@ -91,6 +109,23 @@ impl StateDB {
 		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),
+				modifications: VecDeque::new(),
+			})),
+			cache_overlay: Vec::new(),
+			account_bloom: Arc::new(Mutex::new(bloom)),
+			parent_hash: None,
+			commit_hash: None,
+			commit_number: None,
+		}
+	}
+
 	pub fn check_account_bloom(&self, address: &Address) -> bool {
 		trace!(target: "account_bloom", "Check account bloom: {:?}", address);
 		let bloom = self.account_bloom.lock();
@@ -125,14 +160,103 @@ impl StateDB {
  			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();
-		} else {
-			self.clear_cache();
-		}
+		self.commit_hash = Some(id.clone());
+		self.commit_number = Some(now);
 		Ok(records)
 	}
 
+	/// Update canonical cache. This should be called after the block has been commited and the
+	/// blockchain route has ben calculated.
+	/// `retracted` is the list of the retracted block hashes.
+	pub fn update_cache(&mut self, enacted: &[H256], retracted: &[H256], is_best: bool) {
+		trace!("commit id = (#{:?}, {:?}), parent={:?}, best={}", self.commit_number, self.commit_hash, self.parent_hash, is_best);
+		let mut cache = self.account_cache.lock();
+		let mut cache = &mut *cache;
+
+		// Clean changes from re-enacted and retracted blocks
+		let mut clear = false;
+		for block in enacted.iter().filter(|h| self.commit_hash.as_ref().map_or(false, |p| *h != p)) {
+			clear = clear || {
+				if let Some(ref mut m) = cache.modifications.iter_mut().find(|ref m| &m.hash == block) {
+					trace!("reverting enacted block {:?}", block);
+					m.is_canon = true;
+					for a in &m.accounts {
+						trace!("reverting enacted address {:?}", a);
+						cache.accounts.remove(a);
+					}
+					false
+				} else {
+					true
+				}
+			};
+		}
+
+		for block in retracted {
+			clear = clear || {
+				if let Some(ref mut m) = cache.modifications.iter_mut().find(|ref m| &m.hash == block) {
+					trace!("retracting block {:?}", block);
+					m.is_canon = false;
+					for a in &m.accounts {
+						trace!("retracted address {:?}", a);
+						cache.accounts.remove(a);
+					}
+					false
+				} else {
+					true
+				}
+			};
+		}
+		if clear {
+			// We don't know anything about the block; clear everything
+			trace!("wiping cache");
+			cache.accounts.clear();
+			cache.modifications.clear();
+		}
+
+		// Apply cache changes only if committing on top of the latest canonical state
+		// blocks are ordered by number and only one block with a given number is marked as canonical
+		// (contributed to canonical state cache)
+		if let (Some(ref number), Some(ref hash), Some(ref parent)) = (self.commit_number, self.commit_hash, self.parent_hash) {
+			if cache.modifications.len() == STATE_CACHE_BLOCKS {
+				cache.modifications.pop_back();
+			}
+			let mut modifications = HashSet::new();
+			trace!("committing {} cache entries", self.cache_overlay.len());
+			for account in self.cache_overlay.drain(..) {
+				if account.modified {
+					modifications.insert(account.address.clone());
+				}
+				if is_best {
+					if let Some(&mut Some(ref mut existing)) = cache.accounts.get_mut(&account.address) {
+						if let Some(new) = account.account {
+							if account.modified {
+								existing.merge_with(new);
+							}
+							continue;
+						}
+					}
+					cache.accounts.insert(account.address, account.account);
+				}
+			}
+
+			// Save modified accounts. These are ordered by the block number.
+			let block_changes = BlockChanges {
+				accounts: modifications,
+				number: *number,
+				hash: hash.clone(),
+				is_canon: is_best,
+				parent: parent.clone(),
+			};
+			let insert_at = cache.modifications.iter().enumerate().find(|&(_, ref m)| m.number < *number).map(|(i, _)| i);
+			trace!("inserting modifications at {:?}", insert_at);
+			if let Some(insert_at) = insert_at {
+				cache.modifications.insert(insert_at, block_changes);
+			} else {
+				cache.modifications.push_back(block_changes);
+			}
+		}
+	}
+
 	/// Returns an interface to HashDB.
 	pub fn as_hashdb(&self) -> &HashDB {
 		self.db.as_hashdb()
@@ -149,19 +273,23 @@ impl StateDB {
 			db: self.db.boxed_clone(),
 			account_cache: self.account_cache.clone(),
 			cache_overlay: Vec::new(),
-			is_canon: false,
 			account_bloom: self.account_bloom.clone(),
+			parent_hash: None,
+			commit_hash: None,
+			commit_number: None,
 		}
 	}
 
 	/// Clone the database for a canonical state.
-	pub fn boxed_clone_canon(&self) -> StateDB {
+	pub fn boxed_clone_canon(&self, parent: &H256) -> StateDB {
 		StateDB {
 			db: self.db.boxed_clone(),
 			account_cache: self.account_cache.clone(),
 			cache_overlay: Vec::new(),
-			is_canon: true,
 			account_bloom: self.account_bloom.clone(),
+			parent_hash: Some(parent.clone()),
+			commit_hash: None,
+			commit_number: None,
 		}
 	}
 
@@ -181,26 +309,17 @@ impl StateDB {
 	}
 
 	/// Enqueue cache change.
-	pub fn cache_account(&mut self, addr: Address, data: Option<Account>) {
-		self.cache_overlay.push((addr, data));
-	}
-
-	/// Apply pending cache changes.
-	fn commit_cache(&mut self) {
-		let mut cache = self.account_cache.lock();
-		for (address, account) in self.cache_overlay.drain(..) {
-			if let Some(&mut Some(ref mut existing)) = cache.accounts.get_mut(&address) {
-				if let Some(new) = account {
-					existing.merge_with(new);
-					continue;
-				}
-			}
-			cache.accounts.insert(address, account);
-		}
+	pub fn cache_account(&mut self, addr: Address, data: Option<Account>, modified: bool) {
+		self.cache_overlay.push(CacheQueueItem {
+			address: addr,
+			account: data,
+			modified: modified,
+		})
 	}
 
 	/// Clear the cache.
 	pub fn clear_cache(&mut self) {
+		trace!("Clearing cache");
 		self.cache_overlay.clear();
 		let mut cache = self.account_cache.lock();
 		cache.accounts.clear();
@@ -210,10 +329,10 @@ impl StateDB {
 	/// Returns 'None' if the state is non-canonical and cache is disabled
 	/// or if the account is not cached.
 	pub fn get_cached_account(&self, addr: &Address) -> Option<Option<Account>> {
-		if !self.is_canon {
+		let mut cache = self.account_cache.lock();
+		if !Self::is_allowed(addr, &self.parent_hash, &cache.modifications) {
 			return None;
 		}
-		let mut cache = self.account_cache.lock();
 		cache.accounts.get_mut(&addr).map(|a| a.as_ref().map(|a| a.clone_basic()))
 	}
 
@@ -222,11 +341,46 @@ impl StateDB {
 	/// or if the account is not cached.
 	pub fn get_cached<F, U>(&self, a: &Address, f: F) -> Option<U>
 		where F: FnOnce(Option<&mut Account>) -> U {
-		if !self.is_canon {
+		let mut cache = self.account_cache.lock();
+		if !Self::is_allowed(a, &self.parent_hash, &cache.modifications) {
 			return None;
 		}
-		let mut cache = self.account_cache.lock();
 		cache.accounts.get_mut(a).map(|c| f(c.as_mut()))
 	}
+
+	// Check if the account can be returned from cache by matching current block parent hash against canonical
+	// state and filtering out account modified in later blocks.
+	fn is_allowed(addr: &Address, parent_hash: &Option<H256>, modifications: &VecDeque<BlockChanges>) -> bool {
+		let mut parent = match *parent_hash {
+			None => {
+				trace!("cache lookup skipped for {:?}: no parent hash", addr);
+				return false;
+			}
+			Some(ref parent) => parent,
+		};
+		if modifications.is_empty() {
+			return true;
+		}
+		// Ignore all accounts modified in later blocks
+		// Modifications contains block ordered by the number
+		// We search for our parent in that list first and then for
+		// all its parent until we hit the canonical block,
+		// checking against all the intermediate modifications.
+		let mut iter = modifications.iter();
+		while let Some(ref m) = iter.next() {
+			if &m.hash == parent {
+				if m.is_canon {
+					return true;
+				}
+				parent = &m.parent;
+			}
+			if m.accounts.contains(addr) {
+				trace!("cache lookup skipped for {:?}: modified in a later block", addr);
+				return false;
+			}
+		}
+		trace!("cache lookup skipped for {:?}: parent hash is unknown", addr);
+		return false;
+	}
 }