pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-12 21:41:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Pin states in memory so that they are not pruned away while still referenced (#2761)

Browse Source 
* State pinning in client

* Canonicalization queue

* Fixed prioritization queue

* possible fix of "hash mismatch"

* Check for pinned discarded states

* Release state before finalization

* Style

* Style
This commit is contained in:
Arkadiy Paronyan
2019-06-04 10:01:12 +02:00
committed by Gavin Wood
parent 6ce7c1c8c8
commit 3b26453047
13 changed files with 392 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/core/client/db/src/lib.rs
+106 -11
View File
@@ -72,6 +72,97 @@ const MIN_BLOCKS_TO_KEEP_CHANGES_TRIES_FOR: u32 = 32768;
/// DB-backed patricia trie state, transaction type is an overlay of changes to commit.
pub type DbState = state_machine::TrieBackend<Arc<state_machine::Storage<Blake2Hasher>>, Blake2Hasher>;
pub struct RefTrackingState<Block: BlockT> {
state: DbState,
storage: Arc<StorageDb<Block>>,
parent_hash: Option<Block::Hash>,
}
impl<B: BlockT> RefTrackingState<B> {
fn new(state: DbState, storage: Arc<StorageDb<B>>, parent_hash: Option<B::Hash>) -> RefTrackingState<B> {
if let Some(hash) = &parent_hash {
storage.state_db.pin(hash);
}
RefTrackingState {
state,
parent_hash,
storage,
}
}
}
impl<B: BlockT> Drop for RefTrackingState<B> {
fn drop(&mut self) {
if let Some(hash) = &self.parent_hash {
self.storage.state_db.unpin(hash);
}
}
}
impl<B: BlockT> StateBackend<Blake2Hasher> for RefTrackingState<B> {
type Error = <DbState as StateBackend<Blake2Hasher>>::Error;
type Transaction = <DbState as StateBackend<Blake2Hasher>>::Transaction;
type TrieBackendStorage = <DbState as StateBackend<Blake2Hasher>>::TrieBackendStorage;
fn storage(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
self.state.storage(key)
}
fn storage_hash(&self, key: &[u8]) -> Result<Option<H256>, Self::Error> {
self.state.storage_hash(key)
}
fn child_storage(&self, storage_key: &[u8], key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
self.state.child_storage(storage_key, key)
}
fn exists_storage(&self, key: &[u8]) -> Result<bool, Self::Error> {
self.state.exists_storage(key)
}
fn exists_child_storage(&self, storage_key: &[u8], key: &[u8]) -> Result<bool, Self::Error> {
self.state.exists_child_storage(storage_key, key)
}
fn for_keys_with_prefix<F: FnMut(&[u8])>(&self, prefix: &[u8], f: F) {
self.state.for_keys_with_prefix(prefix, f)
}
fn for_keys_in_child_storage<F: FnMut(&[u8])>(&self, storage_key: &[u8], f: F) {
self.state.for_keys_in_child_storage(storage_key, f)
}
fn storage_root<I>(&self, delta: I) -> (H256, Self::Transaction)
where
I: IntoIterator<Item=(Vec<u8>, Option<Vec<u8>>)>
{
self.state.storage_root(delta)
}
fn child_storage_root<I>(&self, storage_key: &[u8], delta: I) -> (Vec<u8>, bool, Self::Transaction)
where
I: IntoIterator<Item=(Vec<u8>, Option<Vec<u8>>)>,
{
self.state.child_storage_root(storage_key, delta)
}
fn pairs(&self) -> Vec<(Vec<u8>, Vec<u8>)> {
self.state.pairs()
}
fn keys(&self, prefix: &[u8]) -> Vec<Vec<u8>> {
self.state.keys(prefix)
}
fn child_keys(&self, child_key: &[u8], prefix: &[u8]) -> Vec<Vec<u8>> {
self.state.child_keys(child_key, prefix)
}
fn as_trie_backend(&mut self) -> Option<&state_machine::TrieBackend<Self::TrieBackendStorage, Blake2Hasher>> {
self.state.as_trie_backend()
}
}
/// Database settings.
pub struct DatabaseSettings {
/// Cache size in bytes. If `None` default is used.
@@ -270,7 +361,7 @@ impl<Block: BlockT> client::blockchain::ProvideCache<Block> for BlockchainDb<Blo
/// Database transaction
pub struct BlockImportOperation<Block: BlockT, H: Hasher> {
old_state: CachingState<Blake2Hasher, DbState, Block>,
old_state: CachingState<Blake2Hasher, RefTrackingState<Block>, Block>,
db_updates: PrefixedMemoryDB<H>,
storage_updates: Vec<(Vec<u8>, Option<Vec<u8>>)>,
changes_trie_updates: MemoryDB<H>,
@@ -295,7 +386,7 @@ impl<Block> client::backend::BlockImportOperation<Block, Blake2Hasher>
for BlockImportOperation<Block, Blake2Hasher>
where Block: BlockT<Hash=H256>,
{
type State = CachingState<Blake2Hasher, DbState, Block>;
type State = CachingState<Blake2Hasher, RefTrackingState<Block>, Block>;
fn state(&self) -> Result<Option<&Self::State>, client::error::Error> {
Ok(Some(&self.old_state))
@@ -922,6 +1013,8 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
let changes_trie_updates = operation.changes_trie_updates;
self.changes_tries_storage.commit(&mut transaction, changes_trie_updates);
let cache = operation.old_state.release(); // release state reference so that it can be finalized
if finalized {
// TODO: ensure best chain contains this block.
@@ -953,7 +1046,7 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
meta_updates.push((hash, number, pending_block.leaf_state.is_best(), finalized));
Some((number, hash, enacted, retracted, displaced_leaf, is_best))
Some((number, hash, enacted, retracted, displaced_leaf, is_best, cache))
} else {
None
};
@@ -975,7 +1068,7 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
let write_result = self.storage.db.write(transaction).map_err(db_err);
if let Some((number, hash, enacted, retracted, displaced_leaf, is_best)) = imported {
if let Some((number, hash, enacted, retracted, displaced_leaf, is_best, mut cache)) = imported {
if let Err(e) = write_result {
let mut leaves = self.blockchain.leaves.write();
let mut undo = leaves.undo();
@@ -990,7 +1083,7 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
return Err(e)
}
operation.old_state.sync_cache(
cache.sync_cache(
&enacted,
&retracted,
operation.storage_updates,
@@ -1090,7 +1183,7 @@ impl<Block> client::backend::AuxStore for Backend<Block> where Block: BlockT<Has
impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> where Block: BlockT<Hash=H256> {
type BlockImportOperation = BlockImportOperation<Block, Blake2Hasher>;
type Blockchain = BlockchainDb<Block>;
type State = CachingState<Blake2Hasher, DbState, Block>;
type State = CachingState<Blake2Hasher, RefTrackingState<Block>, Block>;
type ChangesTrieStorage = DbChangesTrieStorage<Block>;
fn begin_operation(&self) -> Result<Self::BlockImportOperation, client::error::Error> {
@@ -1217,7 +1310,8 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
BlockId::Hash(h) if h == Default::default() => {
let genesis_storage = DbGenesisStorage::new();
let root = genesis_storage.0.clone();
let state = DbState::new(Arc::new(genesis_storage), root);
let db_state = DbState::new(Arc::new(genesis_storage), root);
let state = RefTrackingState::new(db_state, self.storage.clone(), None);
return Ok(CachingState::new(state, self.shared_cache.clone(), None));
},
_ => {}
@@ -1228,7 +1322,8 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
let hash = hdr.hash();
if !self.storage.state_db.is_pruned(&hash, (*hdr.number()).saturated_into::<u64>()) {
let root = H256::from_slice(hdr.state_root().as_ref());
let state = DbState::new(self.storage.clone(), root);
let db_state = DbState::new(self.storage.clone(), root);
let state = RefTrackingState::new(db_state, self.storage.clone(), Some(hash.clone()));
Ok(CachingState::new(state, self.shared_cache.clone(), Some(hash)))
} else {
Err(client::error::Error::UnknownBlock(format!("State already discarded for {:?}", block)))
@@ -1243,10 +1338,10 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
!self.storage.state_db.is_pruned(hash, number.saturated_into::<u64>())
}
fn destroy_state(&self, mut state: Self::State) -> Result<(), client::error::Error> {
if let Some(hash) = state.parent_hash.clone() {
fn destroy_state(&self, state: Self::State) -> Result<(), client::error::Error> {
if let Some(hash) = state.cache.parent_hash.clone() {
let is_best = || self.blockchain.meta.read().best_hash == hash;
state.sync_cache(&[], &[], vec![], None, None, is_best);
state.release().sync_cache(&[], &[], vec![], None, None, is_best);
}
Ok(())
}
substrate/core/client/db/src/storage_cache.rs
+93 -71
View File
@@ -99,6 +99,17 @@ struct LocalCache<H: Hasher> {
hashes: HashMap<StorageKey, Option<H::Out>>,
}
/// Cache changes.
pub struct CacheChanges<H: Hasher, B: Block> {
/// Shared canonical state cache.
shared_cache: SharedCache<B, H>,
/// Local cache of values for this state.
local_cache: RwLock<LocalCache<H>>,
/// Hash of the block on top of which this instance was created or
/// `None` if cache is disabled
pub parent_hash: Option<B::Hash>,
}
/// State abstraction.
/// Manages shared global state cache which reflects the canonical
/// state as it is on the disk.
@@ -109,56 +120,11 @@ struct LocalCache<H: Hasher> {
pub struct CachingState<H: Hasher, S: StateBackend<H>, B: Block> {
/// Backing state.
state: S,
/// Shared canonical state cache.
shared_cache: SharedCache<B, H>,
/// Local cache of values for this state.
local_cache: RwLock<LocalCache<H>>,
/// Hash of the block on top of which this instance was created or
/// `None` if cache is disabled
pub parent_hash: Option<B::Hash>,
/// Cache data.
pub cache: CacheChanges<H, B>
}
impl<H: Hasher, S: StateBackend<H>, B: Block> CachingState<H, S, B> {
/// Create a new instance wrapping generic State and shared cache.
pub fn new(state: S, shared_cache: SharedCache<B, H>, parent_hash: Option<B::Hash>) -> CachingState<H, S, B> {
CachingState {
state,
shared_cache,
local_cache: RwLock::new(LocalCache {
storage: Default::default(),
hashes: Default::default(),
}),
parent_hash: parent_hash,
}
}
fn storage_insert(cache: &mut Cache<B, H>, k: StorageValue, v: Option<StorageValue>) {
if let Some(v_) = &v {
while cache.storage_used_size + v_.len() > cache.shared_cache_size {
// pop until space constraint satisfied
match cache.storage.remove_lru() {
Some((_, Some(popped_v))) =>
cache.storage_used_size = cache.storage_used_size - popped_v.len(),
Some((_, None)) => continue,
None => break,
};
}
cache.storage_used_size = cache.storage_used_size + v_.len();
}
cache.storage.insert(k, v);
}
fn storage_remove(
storage: &mut LruCache<StorageKey, Option<StorageValue>>,
k: &StorageKey,
storage_used_size: &mut usize,
) {
let v = storage.remove(k);
if let Some(Some(v_)) = v {
*storage_used_size = *storage_used_size - v_.len();
}
}
impl<H: Hasher, B: Block> CacheChanges<H, B> {
/// Propagate local cache into the shared cache and synchronize
/// the shared cache with the best block state.
/// This function updates the shared cache by removing entries
@@ -189,7 +155,7 @@ impl<H: Hasher, S: StateBackend<H>, B: Block> CachingState<H, S, B> {
m.is_canon = true;
for a in &m.storage {
trace!("Reverting enacted key {:?}", a);
CachingState::<H, S, B>::storage_remove(&mut cache.storage, a, &mut cache.storage_used_size);
CacheChanges::<H, B>::storage_remove(&mut cache.storage, a, &mut cache.storage_used_size);
}
false
} else {
@@ -205,7 +171,7 @@ impl<H: Hasher, S: StateBackend<H>, B: Block> CachingState<H, S, B> {
m.is_canon = false;
for a in &m.storage {
trace!("Retracted key {:?}", a);
CachingState::<H, S, B>::storage_remove(&mut cache.storage, a, &mut cache.storage_used_size);
CacheChanges::<H, B>::storage_remove(&mut cache.storage, a, &mut cache.storage_used_size);
}
false
} else {
@@ -228,7 +194,7 @@ impl<H: Hasher, S: StateBackend<H>, B: Block> CachingState<H, S, B> {
if is_best {
trace!("Committing {} local, {} hashes, {} modified entries", local_cache.storage.len(), local_cache.hashes.len(), changes.len());
for (k, v) in local_cache.storage.drain() {
CachingState::<H, S, B>::storage_insert(cache, k, v);
CacheChanges::<H, B>::storage_insert(cache, k, v);
}
for (k, v) in local_cache.hashes.drain() {
cache.hashes.insert(k, v);
@@ -248,7 +214,7 @@ impl<H: Hasher, S: StateBackend<H>, B: Block> CachingState<H, S, B> {
modifications.insert(k.clone());
if is_best {
cache.hashes.remove(&k);
CachingState::<H, S, B>::storage_insert(cache, k, v);
CacheChanges::<H, B>::storage_insert(cache, k, v);
}
}
// Save modified storage. These are ordered by the block number.
@@ -272,6 +238,50 @@ impl<H: Hasher, S: StateBackend<H>, B: Block> CachingState<H, S, B> {
}
}
fn storage_insert(cache: &mut Cache<B, H>, k: StorageValue, v: Option<StorageValue>) {
if let Some(v_) = &v {
while cache.storage_used_size + v_.len() > cache.shared_cache_size {
// pop until space constraint satisfied
match cache.storage.remove_lru() {
Some((_, Some(popped_v))) =>
cache.storage_used_size = cache.storage_used_size - popped_v.len(),
Some((_, None)) => continue,
None => break,
};
}
cache.storage_used_size = cache.storage_used_size + v_.len();
}
cache.storage.insert(k, v);
}
fn storage_remove(
storage: &mut LruCache<StorageKey, Option<StorageValue>>,
k: &StorageKey,
storage_used_size: &mut usize,
) {
let v = storage.remove(k);
if let Some(Some(v_)) = v {
*storage_used_size = *storage_used_size - v_.len();
}
}
}
impl<H: Hasher, S: StateBackend<H>, B: Block> CachingState<H, S, B> {
/// Create a new instance wrapping generic State and shared cache.
pub fn new(state: S, shared_cache: SharedCache<B, H>, parent_hash: Option<B::Hash>) -> CachingState<H, S, B> {
CachingState {
state,
cache: CacheChanges {
shared_cache,
local_cache: RwLock::new(LocalCache {
storage: Default::default(),
hashes: Default::default(),
}),
parent_hash: parent_hash,
},
}
}
/// Check if the key can be returned from cache by matching current block parent hash against canonical
/// state and filtering out entries modified in later blocks.
fn is_allowed(
@@ -312,6 +322,11 @@ impl<H: Hasher, S: StateBackend<H>, B: Block> CachingState<H, S, B> {
trace!("Cache lookup skipped for {:?}: parent hash is unknown", key);
false
}
/// Dispose state and return cache data.
pub fn release(self) -> CacheChanges<H, B> {
self.cache
}
}
impl<H: Hasher, S: StateBackend<H>, B:Block> StateBackend<H> for CachingState<H, S, B> {
@@ -320,13 +335,13 @@ impl<H: Hasher, S: StateBackend<H>, B:Block> StateBackend<H> for CachingState<H,
type TrieBackendStorage = S::TrieBackendStorage;
fn storage(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
let local_cache = self.local_cache.upgradable_read();
let local_cache = self.cache.local_cache.upgradable_read();
if let Some(entry) = local_cache.storage.get(key).cloned() {
trace!("Found in local cache: {:?}", key);
return Ok(entry)
}
let mut cache = self.shared_cache.lock();
if Self::is_allowed(key, &self.parent_hash, &cache.modifications) {
let mut cache = self.cache.shared_cache.lock();
if Self::is_allowed(key, &self.cache.parent_hash, &cache.modifications) {
if let Some(entry) = cache.storage.get_mut(key).map(|a| a.clone()) {
trace!("Found in shared cache: {:?}", key);
return Ok(entry)
@@ -339,13 +354,13 @@ impl<H: Hasher, S: StateBackend<H>, B:Block> StateBackend<H> for CachingState<H,
}
fn storage_hash(&self, key: &[u8]) -> Result<Option<H::Out>, Self::Error> {
let local_cache = self.local_cache.upgradable_read();
let local_cache = self.cache.local_cache.upgradable_read();
if let Some(entry) = local_cache.hashes.get(key).cloned() {
trace!("Found hash in local cache: {:?}", key);
return Ok(entry)
}
let mut cache = self.shared_cache.lock();
if Self::is_allowed(key, &self.parent_hash, &cache.modifications) {
let mut cache = self.cache.shared_cache.lock();
if Self::is_allowed(key, &self.cache.parent_hash, &cache.modifications) {
if let Some(entry) = cache.hashes.get_mut(key).map(|a| a.clone()) {
trace!("Found hash in shared cache: {:?}", key);
return Ok(entry)
@@ -405,8 +420,8 @@ impl<H: Hasher, S: StateBackend<H>, B:Block> StateBackend<H> for CachingState<H,
self.state.child_keys(child_key, prefix)
}
fn try_into_trie_backend(self) -> Option<TrieBackend<Self::TrieBackendStorage, H>> {
self.state.try_into_trie_backend()
fn as_trie_backend(&mut self) -> Option<&TrieBackend<Self::TrieBackendStorage, H>> {
self.state.as_trie_backend()
}
}
@@ -436,22 +451,22 @@ mod tests {
// blocks [ 3a(c) 2a(c) 2b 1b 1a(c) 0 ]
// state [ 5 5 4 3 2 2 ]
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(root_parent.clone()));
s.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], Some(h0.clone()), Some(0), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], Some(h0.clone()), Some(0), || true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h0.clone()));
s.sync_cache(&[], &[], vec![], Some(h1a.clone()), Some(1), || true);
s.cache.sync_cache(&[], &[], vec![], Some(h1a.clone()), Some(1), || true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h0.clone()));
s.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], Some(h1b.clone()), Some(1), || false);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], Some(h1b.clone()), Some(1), || false);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1b.clone()));
s.sync_cache(&[], &[], vec![(key.clone(), Some(vec![4]))], Some(h2b.clone()), Some(2), || false);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![4]))], Some(h2b.clone()), Some(2), || false);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1a.clone()));
s.sync_cache(&[], &[], vec![(key.clone(), Some(vec![5]))], Some(h2a.clone()), Some(2), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![5]))], Some(h2a.clone()), Some(2), || true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h2a.clone()));
s.sync_cache(&[], &[], vec![], Some(h3a.clone()), Some(3), || true);
s.cache.sync_cache(&[], &[], vec![], Some(h3a.clone()), Some(3), || true);
let s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h3a.clone()));
assert_eq!(s.storage(&key).unwrap().unwrap(), vec![5]);
@@ -468,7 +483,14 @@ mod tests {
// reorg to 3b
// blocks [ 3b(c) 3a 2a 2b(c) 1b 1a 0 ]
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h2b.clone()));
s.sync_cache(&[h1b.clone(), h2b.clone(), h3b.clone()], &[h1a.clone(), h2a.clone(), h3a.clone()], vec![], Some(h3b.clone()), Some(3), || true);
s.cache.sync_cache(
&[h1b.clone(), h2b.clone(), h3b.clone()],
&[h1a.clone(), h2a.clone(), h3a.clone()],
vec![],
Some(h3b.clone()),
Some(3),
|| true
);
let s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h3a.clone()));
assert!(s.storage(&key).unwrap().is_none());
}
@@ -482,11 +504,11 @@ mod tests {
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(root_parent.clone()));
let key = H256::random()[..].to_vec();
s.sync_cache(&[], &[], vec![(key.clone(), Some(vec![1, 2, 3]))], Some(h0.clone()), Some(0), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![1, 2, 3]))], Some(h0.clone()), Some(0), || true);
assert_eq!(shared.lock().used_storage_cache_size(), 3 /* bytes */);
let key = H256::random()[..].to_vec();
s.sync_cache(&[], &[], vec![(key.clone(), Some(vec![1, 2]))], Some(h0.clone()), Some(0), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![1, 2]))], Some(h0.clone()), Some(0), || true);
assert_eq!(shared.lock().used_storage_cache_size(), 5 /* bytes */);
}
@@ -499,11 +521,11 @@ mod tests {
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(root_parent.clone()));
let key = H256::random()[..].to_vec();
s.sync_cache(&[], &[], vec![(key.clone(), Some(vec![1, 2, 3, 4]))], Some(h0.clone()), Some(0), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![1, 2, 3, 4]))], Some(h0.clone()), Some(0), || true);
assert_eq!(shared.lock().used_storage_cache_size(), 4 /* bytes */);
let key = H256::random()[..].to_vec();
s.sync_cache(&[], &[], vec![(key.clone(), Some(vec![1, 2]))], Some(h0.clone()), Some(0), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![1, 2]))], Some(h0.clone()), Some(0), || true);
assert_eq!(shared.lock().used_storage_cache_size(), 2 /* bytes */);
}
}