Skip to content
This repository was archived by the owner on Nov 15, 2023. It is now read-only.

Reduce shared state cache contention #9321

Merged
1 commit merged into from
Jul 14, 2021
Merged

Reduce shared state cache contention #9321

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions client/db/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1643,7 +1643,7 @@ impl<Block: BlockT> Backend<Block> {
self.changes_tries_storage.post_commit(changes_trie_cache_ops);

if let Some((enacted, retracted)) = cache_update {
self.shared_cache.lock().sync(&enacted, &retracted);
self.shared_cache.write().sync(&enacted, &retracted);
}

for m in meta_updates {
Expand Down Expand Up @@ -2052,7 +2052,7 @@ impl<Block: BlockT> sc_client_api::backend::Backend<Block> for Backend<Block> {
);
let database_cache = MemorySize::from_bytes(0);
let state_cache = MemorySize::from_bytes(
(*&self.shared_cache).lock().used_storage_cache_size(),
(*&self.shared_cache).read().used_storage_cache_size(),
);
let state_db = self.storage.state_db.memory_info();

Expand Down
69 changes: 39 additions & 30 deletions client/db/src/storage_cache.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
use std::collections::{VecDeque, HashSet, HashMap};
use std::sync::Arc;
use std::hash::Hash as StdHash;
use parking_lot::{Mutex, RwLock, RwLockUpgradableReadGuard};
use parking_lot::{RwLock, RwLockUpgradableReadGuard};
use linked_hash_map::{LinkedHashMap, Entry};
use hash_db::Hasher;
use sp_runtime::traits::{Block as BlockT, Header, HashFor, NumberFor};
Expand Down Expand Up @@ -222,7 +222,7 @@ impl<B: BlockT> Cache<B> {
}
}

pub type SharedCache<B> = Arc<Mutex<Cache<B>>>;
pub type SharedCache<B> = Arc<RwLock<Cache<B>>>;

/// Fix lru storage size for hash (small 64ko).
const FIX_LRU_HASH_SIZE: usize = 65_536;
Expand All @@ -234,7 +234,7 @@ pub fn new_shared_cache<B: BlockT>(
) -> SharedCache<B> {
let top = child_ratio.1.saturating_sub(child_ratio.0);
Arc::new(
Mutex::new(
RwLock::new(
Cache {
lru_storage: LRUMap(
LinkedHashMap::new(), 0, shared_cache_size * top / child_ratio.1
Expand Down Expand Up @@ -337,7 +337,7 @@ impl<B: BlockT> CacheChanges<B> {
commit_number: Option<NumberFor<B>>,
is_best: bool,
) {
let mut cache = self.shared_cache.lock();
let mut cache = self.shared_cache.write();
trace!(
"Syncing cache, id = (#{:?}, {:?}), parent={:?}, best={}",
commit_number,
Expand Down Expand Up @@ -527,12 +527,15 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>> for Cachin

return Ok(entry)
}
let mut cache = self.cache.shared_cache.lock();
if Self::is_allowed(Some(key), None, &self.cache.parent_hash, &cache.modifications) {
if let Some(entry) = cache.lru_storage.get(key).map(|a| a.clone()) {
trace!("Found in shared cache: {:?}", HexDisplay::from(&key));
self.usage.tally_key_read(key, entry.as_ref(), true);
return Ok(entry)
{
let cache = self.cache.shared_cache.upgradable_read();
if Self::is_allowed(Some(key), None, &self.cache.parent_hash, &cache.modifications) {
let mut cache = RwLockUpgradableReadGuard::upgrade(cache);
if let Some(entry) = cache.lru_storage.get(key).map(|a| a.clone()) {
trace!("Found in shared cache: {:?}", HexDisplay::from(&key));
self.usage.tally_key_read(key, entry.as_ref(), true);
return Ok(entry)
}
}
}
trace!("Cache miss: {:?}", HexDisplay::from(&key));
Expand All @@ -548,11 +551,14 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>> for Cachin
trace!("Found hash in local cache: {:?}", HexDisplay::from(&key));
return Ok(entry)
}
let mut cache = self.cache.shared_cache.lock();
if Self::is_allowed(Some(key), None, &self.cache.parent_hash, &cache.modifications) {
if let Some(entry) = cache.lru_hashes.get(key).map(|a| a.0.clone()) {
trace!("Found hash in shared cache: {:?}", HexDisplay::from(&key));
return Ok(entry)
{
let cache = self.cache.shared_cache.upgradable_read();
if Self::is_allowed(Some(key), None, &self.cache.parent_hash, &cache.modifications) {
let mut cache = RwLockUpgradableReadGuard::upgrade(cache);
if let Some(entry) = cache.lru_hashes.get(key).map(|a| a.0.clone()) {
trace!("Found hash in shared cache: {:?}", HexDisplay::from(&key));
return Ok(entry)
}
}
}
trace!("Cache hash miss: {:?}", HexDisplay::from(&key));
Expand All @@ -574,13 +580,16 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>> for Cachin
self.usage.tally_child_key_read(&key, entry, true)
)
}
let mut cache = self.cache.shared_cache.lock();
if Self::is_allowed(None, Some(&key), &self.cache.parent_hash, &cache.modifications) {
if let Some(entry) = cache.lru_child_storage.get(&key).map(|a| a.clone()) {
trace!("Found in shared cache: {:?}", key);
return Ok(
self.usage.tally_child_key_read(&key, entry, true)
)
{
let cache = self.cache.shared_cache.upgradable_read();
if Self::is_allowed(None, Some(&key), &self.cache.parent_hash, &cache.modifications) {
let mut cache = RwLockUpgradableReadGuard::upgrade(cache);
if let Some(entry) = cache.lru_child_storage.get(&key).map(|a| a.clone()) {
trace!("Found in shared cache: {:?}", key);
return Ok(
self.usage.tally_child_key_read(&key, entry, true)
)
}
}
}
trace!("Cache miss: {:?}", key);
Expand Down Expand Up @@ -1274,7 +1283,7 @@ mod tests {
true,
);
// 32 key, 3 byte size
assert_eq!(shared.lock().used_storage_cache_size(), 35 /* bytes */);
assert_eq!(shared.read().used_storage_cache_size(), 35 /* bytes */);

let key = H256::random()[..].to_vec();
s.cache.sync_cache(
Expand All @@ -1287,7 +1296,7 @@ mod tests {
true,
);
// 35 + (2 * 32) key, 2 byte size
assert_eq!(shared.lock().used_storage_cache_size(), 101 /* bytes */);
assert_eq!(shared.read().used_storage_cache_size(), 101 /* bytes */);
}

#[test]
Expand All @@ -1313,7 +1322,7 @@ mod tests {
true,
);
// 32 key, 4 byte size
assert_eq!(shared.lock().used_storage_cache_size(), 36 /* bytes */);
assert_eq!(shared.read().used_storage_cache_size(), 36 /* bytes */);

let key = H256::random()[..].to_vec();
s.cache.sync_cache(
Expand All @@ -1326,7 +1335,7 @@ mod tests {
true,
);
// 32 key, 2 byte size
assert_eq!(shared.lock().used_storage_cache_size(), 34 /* bytes */);
assert_eq!(shared.read().used_storage_cache_size(), 34 /* bytes */);
}

#[test]
Expand Down Expand Up @@ -1379,7 +1388,7 @@ mod tests {

// Restart (or unknown block?), clear caches.
{
let mut cache = s.cache.shared_cache.lock();
let mut cache = s.cache.shared_cache.write();
let cache = &mut *cache;
cache.lru_storage.clear();
cache.lru_hashes.clear();
Expand Down Expand Up @@ -1426,7 +1435,7 @@ mod tests {
Some(1),
true,
);
assert_eq!(shared.lock().lru_storage.get(&key).unwrap(), &Some(vec![1]));
assert_eq!(shared.write().lru_storage.get(&key).unwrap(), &Some(vec![1]));

let mut s = CachingState::new(
InMemoryBackend::<BlakeTwo256>::default(),
Expand All @@ -1445,7 +1454,7 @@ mod tests {
false,
);

assert_eq!(shared.lock().lru_storage.get(&key).unwrap(), &Some(vec![1]));
assert_eq!(shared.write().lru_storage.get(&key).unwrap(), &Some(vec![1]));

let mut s = CachingState::new(
InMemoryBackend::<BlakeTwo256>::default(),
Expand Down Expand Up @@ -1800,7 +1809,7 @@ mod qc {

std::mem::swap(fork_chain, &mut new_fork);

self.shared.lock().sync(&retracted, &enacted);
self.shared.write().sync(&retracted, &enacted);

self.head_state(
self.canon.last()
Expand Down