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

Fixed shared cache race on import (#4194)

Browse Source 
* Fixed is_best race on import

* Take import lock outside of backend

* Actually take the lock
This commit is contained in:
Arkadiy Paronyan
2019-11-25 13:38:37 +01:00
committed by Gavin Wood
parent 50b84a6438
commit d56d6163ef
8 changed files with 56 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/client/api/src/backend.rs
+2 -2
View File
@@ -33,7 +33,7 @@ use crate::{
};
use consensus::BlockOrigin;
use hash_db::Hasher;
use parking_lot::Mutex;
use parking_lot::RwLock;
/// In memory array of storage values.
pub type StorageCollection = Vec<(Vec<u8>, Option<Vec<u8>>)>;
@@ -310,7 +310,7 @@ pub trait Backend<Block, H>: AuxStore + Send + Sync where
/// the using components should acquire and hold the lock whenever they do
/// something that the import of a block would interfere with, e.g. importing
/// a new block or calculating the best head.
fn get_import_lock(&self) -> &Mutex<()>;
fn get_import_lock(&self) -> &RwLock<()>;
}
/// Changes trie storage that supports pruning.
substrate/client/api/src/blockchain.rs
+3 -3
View File
@@ -22,7 +22,7 @@ use sr_primitives::traits::{Block as BlockT, Header as HeaderT, NumberFor};
use sr_primitives::generic::BlockId;
use sr_primitives::Justification;
use log::warn;
use parking_lot::Mutex;
use parking_lot::RwLock;
use header_metadata::HeaderMetadata;
@@ -109,7 +109,7 @@ pub trait Backend<Block: BlockT>: HeaderBackend<Block> + HeaderMetadata<Block, E
&self,
target_hash: Block::Hash,
maybe_max_number: Option<NumberFor<Block>>,
import_lock: &Mutex<()>,
import_lock: &RwLock<()>,
) -> Result<Option<Block::Hash>> {
let target_header = {
match self.header(BlockId::Hash(target_hash))? {
@@ -130,7 +130,7 @@ pub trait Backend<Block: BlockT>: HeaderBackend<Block> + HeaderMetadata<Block, E
// ensure no blocks are imported during this code block.
// an import could trigger a reorg which could change the canonical chain.
// we depend on the canonical chain staying the same during this code block.
let _import_guard = import_lock.lock();
let _import_guard = import_lock.read();
let info = self.info();
substrate/client/db/src/lib.rs
+4 -4
View File
@@ -788,7 +788,7 @@ pub struct Backend<Block: BlockT> {
blockchain: BlockchainDb<Block>,
canonicalization_delay: u64,
shared_cache: SharedCache<Block, Blake2Hasher>,
import_lock: Mutex<()>,
import_lock: RwLock<()>,
is_archive: bool,
}
@@ -1251,7 +1251,7 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
operation.child_storage_updates,
Some(hash),
Some(number),
|| is_best,
is_best,
);
}
@@ -1531,13 +1531,13 @@ impl<Block> client_api::backend::Backend<Block, Blake2Hasher> for Backend<Block>
fn destroy_state(&self, state: Self::State) -> ClientResult<()> {
if let Some(hash) = state.cache.parent_hash.clone() {
let is_best = || self.blockchain.meta.read().best_hash == hash;
let is_best = self.blockchain.meta.read().best_hash == hash;
state.release().sync_cache(&[], &[], vec![], vec![], None, None, is_best);
}
Ok(())
}
fn get_import_lock(&self) -> &Mutex<()> {
fn get_import_lock(&self) -> &RwLock<()> {
&self.import_lock
}
}
substrate/client/db/src/storage_cache.rs
+28 -29
View File
@@ -309,7 +309,7 @@ impl<H: Hasher, B: BlockT> CacheChanges<H, B> {
/// that are invalidated by chain reorganization. `sync_cache`
/// should be called after the block has been committed and the
/// blockchain route has been calculated.
pub fn sync_cache<F: FnOnce() -> bool> (
pub fn sync_cache(
&mut self,
enacted: &[B::Hash],
retracted: &[B::Hash],
@@ -317,10 +317,9 @@ impl<H: Hasher, B: BlockT> CacheChanges<H, B> {
child_changes: ChildStorageCollection,
commit_hash: Option<B::Hash>,
commit_number: Option<<B::Header as Header>::Number>,
is_best: F,
is_best: bool,
) {
let mut cache = self.shared_cache.lock();
let is_best = is_best();
trace!("Syncing cache, id = (#{:?}, {:?}), parent={:?}, best={}", commit_number, commit_hash, self.parent_hash, is_best);
let cache = &mut *cache;
// Filter out commiting block if any.
@@ -621,22 +620,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));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], vec![], Some(h0), Some(0), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], vec![], Some(h0), Some(0), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h0));
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h1a), Some(1), || true);
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h1a), Some(1), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h0));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h1b), Some(1), || false);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h1b), Some(1), false);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1b.clone()));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![4]))], vec![], Some(h2b), Some(2), || false);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![4]))], vec![], Some(h2b), Some(2), false);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1a.clone()));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![5]))], vec![], Some(h2a), Some(2), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![5]))], vec![], Some(h2a), Some(2), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h2a));
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h3a), Some(3), || true);
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h3a), Some(3), true);
let s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h3a));
assert_eq!(s.storage(&key).unwrap().unwrap(), vec![5]);
@@ -660,7 +659,7 @@ mod tests {
vec![],
Some(h3b),
Some(3),
|| true,
true,
);
let s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h3a));
assert!(s.storage(&key).unwrap().is_none());
@@ -680,16 +679,16 @@ mod tests {
let shared = new_shared_cache::<Block, Blake2Hasher>(256*1024, (0,1));
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(root_parent));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], vec![], Some(h1), Some(1), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], vec![], Some(h1), Some(1), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1));
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h2a), Some(2), || true);
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h2a), Some(2), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h2b), Some(2), || false);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h2b), Some(2), false);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h2b));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h3b), Some(2), || false);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h3b), Some(2), false);
let s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h2a));
assert_eq!(s.storage(&key).unwrap().unwrap(), vec![2]);
@@ -708,19 +707,19 @@ mod tests {
let shared = new_shared_cache::<Block, Blake2Hasher>(256*1024, (0,1));
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(root_parent));
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h1), Some(1), || true);
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h1), Some(1), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1));
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h2a), Some(2), || true);
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h2a), Some(2), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h2a));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], vec![], Some(h3a), Some(3), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], vec![], Some(h3a), Some(3), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1));
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h2b), Some(2), || false);
s.cache.sync_cache(&[], &[], vec![], vec![], Some(h2b), Some(2), false);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h2b));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h3b), Some(3), || false);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h3b), Some(3), false);
let s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h3a));
assert_eq!(s.storage(&key).unwrap().unwrap(), vec![2]);
@@ -743,7 +742,7 @@ mod tests {
vec![],
Some(h0),
Some(0),
|| true,
true,
);
// 32 key, 3 byte size
assert_eq!(shared.lock().used_storage_cache_size(), 35 /* bytes */);
@@ -756,7 +755,7 @@ mod tests {
vec![(s_key.clone(), vec![(key.clone(), Some(vec![1, 2]))])],
Some(h0),
Some(0),
|| true,
true,
);
// 35 + (2 * 32) key, 2 byte size
assert_eq!(shared.lock().used_storage_cache_size(), 101 /* bytes */);
@@ -778,7 +777,7 @@ mod tests {
vec![],
Some(h0),
Some(0),
|| true,
true,
);
// 32 key, 4 byte size
assert_eq!(shared.lock().used_storage_cache_size(), 36 /* bytes */);
@@ -791,7 +790,7 @@ mod tests {
vec![],
Some(h0),
Some(0),
|| true,
true,
);
// 32 key, 2 byte size
assert_eq!(shared.lock().used_storage_cache_size(), 34 /* bytes */);
@@ -809,10 +808,10 @@ mod tests {
let shared = new_shared_cache::<Block, Blake2Hasher>(256*1024, (0, 1));
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(root_parent));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], vec![], Some(h0), Some(0), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![2]))], vec![], Some(h0), Some(0), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h0));
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h1), Some(1), || true);
s.cache.sync_cache(&[], &[], vec![(key.clone(), Some(vec![3]))], vec![], Some(h1), Some(1), true);
let mut s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1));
assert_eq!(s.storage(&key).unwrap(), Some(vec![3]));
@@ -831,7 +830,7 @@ mod tests {
s.cache.local_cache.write().storage.insert(key.clone(), Some(vec![42]));
// New value is propagated.
s.cache.sync_cache(&[], &[], vec![], vec![], None, None, || true);
s.cache.sync_cache(&[], &[], vec![], vec![], None, None, true);
let s = CachingState::new(InMemory::<Blake2Hasher>::default(), shared.clone(), Some(h1));
assert_eq!(s.storage(&key).unwrap(), None);
@@ -1051,7 +1050,7 @@ mod qc {
vec![],
Some(hash),
Some(total_h as u64),
|| false,
false,
);
state
@@ -1090,7 +1089,7 @@ mod qc {
vec![],
Some(hash),
Some(self.canon.len() as u64 + 1),
|| true,
true,
);
self.canon.push(next);
@@ -1138,7 +1137,7 @@ mod qc {
vec![],
Some(hash),
Some(height),
|| true,
true,
);
state
substrate/client/src/call_executor.rs
+12 -3
View File
@@ -96,7 +96,10 @@ where
None,
)
.map(|(result, _, _)| result)?;
self.backend.destroy_state(state)?;
{
let _lock = self.backend.get_import_lock().read();
self.backend.destroy_state(state)?;
}
Ok(return_data.into_encoded())
}
@@ -179,7 +182,10 @@ where
)
.map(|(result, _, _)| result)
}?;
self.backend.destroy_state(state)?;
{
let _lock = self.backend.get_import_lock().read();
self.backend.destroy_state(state)?;
}
Ok(result)
}
@@ -194,7 +200,10 @@ where
None,
);
let version = self.executor.runtime_version(&mut ext);
self.backend.destroy_state(state)?;
{
let _lock = self.backend.get_import_lock().read();
self.backend.destroy_state(state)?;
}
version.ok_or(error::Error::VersionInvalid.into())
}
substrate/client/src/client.rs
+1 -1
View File
@@ -710,7 +710,7 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
Err: From<error::Error>,
{
let inner = || {
let _import_lock = self.backend.get_import_lock().lock();
let _import_lock = self.backend.get_import_lock().write();
let mut op = ClientImportOperation {
op: self.backend.begin_operation()?,
substrate/client/src/in_mem.rs
+3 -3
View File
@@ -18,7 +18,7 @@
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use parking_lot::{RwLock, Mutex};
use parking_lot::RwLock;
use primitives::{ChangesTrieConfiguration, storage::well_known_keys};
use primitives::offchain::storage::{
InMemOffchainStorage as OffchainStorage
@@ -562,7 +562,7 @@ where
states: RwLock<HashMap<Block::Hash, InMemory<H>>>,
changes_trie_storage: ChangesTrieStorage<Block, H>,
blockchain: Blockchain<Block>,
import_lock: Mutex<()>,
import_lock: RwLock<()>,
}
impl<Block, H> Backend<Block, H>
@@ -712,7 +712,7 @@ where
Ok(Zero::zero())
}
fn get_import_lock(&self) -> &Mutex<()> {
fn get_import_lock(&self) -> &RwLock<()> {
&self.import_lock
}
}
substrate/client/src/light/backend.rs
+3 -3
View File
@@ -19,7 +19,7 @@
use std::collections::HashMap;
use std::sync::Arc;
use parking_lot::{RwLock, Mutex};
use parking_lot::RwLock;
use state_machine::{Backend as StateBackend, TrieBackend, backend::InMemory as InMemoryState, ChangesTrieTransaction};
use primitives::offchain::storage::InMemOffchainStorage;
@@ -49,7 +49,7 @@ const IN_MEMORY_EXPECT_PROOF: &str = "InMemory state backend has Void error type
pub struct Backend<S, H: Hasher> {
blockchain: Arc<Blockchain<S>>,
genesis_state: RwLock<Option<InMemoryState<H>>>,
import_lock: Mutex<()>,
import_lock: RwLock<()>,
}
/// Light block (header and justification) import operation.
@@ -216,7 +216,7 @@ impl<S, Block, H> ClientBackend<Block, H> for Backend<S, H> where
Err(ClientError::NotAvailableOnLightClient)
}
fn get_import_lock(&self) -> &Mutex<()> {
fn get_import_lock(&self) -> &RwLock<()> {
&self.import_lock
}
}