perf: replace RwLock<HashMap/HashSet> with DashMap/DashSet (#21692)

Co-authored-by: Georgios Konstantopoulos <me@gakonst.com>
Co-authored-by: joshieDo <93316087+joshieDo@users.noreply.github.com>

Cargo.lock

@@ -2902,6 +2902,7 @@ dependencies = [
  "lock_api",
  "once_cell",
  "parking_lot_core",
+ "serde",
 ]
 
 [[package]]
@@ -8233,11 +8234,11 @@ dependencies = [
  "alloy-chains",
  "alloy-primitives",
  "alloy-rlp",
+ "dashmap",
  "data-encoding",
  "enr",
  "hickory-resolver",
  "linked_hash_set",
- "parking_lot",
  "rand 0.9.2",
  "reth-chainspec",
  "reth-ethereum-forks",
@@ -10999,7 +11000,7 @@ dependencies = [
  "alloy-provider",
  "alloy-rpc-types",
  "alloy-rpc-types-engine",
- "parking_lot",
+ "dashmap",
  "reth-chainspec",
  "reth-db-api",
  "reth-errors",

crates/net/dns/Cargo.toml

@@ -30,12 +30,12 @@ tokio-stream.workspace = true
 hickory-resolver = { workspace = true, features = ["tokio"] }
 
 # misc
+dashmap = { workspace = true, features = ["inline"] }
 data-encoding.workspace = true
 linked_hash_set.workspace = true
 schnellru.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
-parking_lot.workspace = true
 serde = { workspace = true, optional = true }
 serde_with = { workspace = true, optional = true }
 
@@ -56,9 +56,9 @@ serde = [
     "alloy-primitives/serde",
     "enr/serde",
     "linked_hash_set/serde",
-    "parking_lot/serde",
     "rand/serde",
     "secp256k1/serde",
     "hickory-resolver/serde",
     "reth-ethereum-forks/serde",
+    "dashmap/serde",
 ]

crates/net/dns/src/resolver.rs

@@ -1,9 +1,9 @@
 //! Perform DNS lookups
 
+use dashmap::DashMap;
 use hickory_resolver::name_server::ConnectionProvider;
 pub use hickory_resolver::{ResolveError, TokioResolver};
-use parking_lot::RwLock;
-use std::{collections::HashMap, future::Future};
+use std::future::Future;
 use tracing::trace;
 
 /// A type that can lookup DNS entries
@@ -72,25 +72,25 @@ impl Resolver for DnsResolver {
 
 /// A [Resolver] that uses an in memory map to lookup entries
 #[derive(Debug, Default)]
-pub struct MapResolver(RwLock<HashMap<String, String>>);
+pub struct MapResolver(DashMap<String, String>);
 
 // === impl MapResolver ===
 
 impl MapResolver {
     /// Inserts a key-value pair into the map.
     pub fn insert(&self, k: String, v: String) -> Option<String> {
-        self.0.write().insert(k, v)
+        self.0.insert(k, v)
     }
 
     /// Returns the value corresponding to the key
     pub fn get(&self, k: &str) -> Option<String> {
-        self.0.read().get(k).cloned()
+        self.0.get(k).map(|entry| entry.value().clone())
     }
 
     /// Removes a key from the map, returning the value at the key if the key was previously in the
     /// map.
     pub fn remove(&self, k: &str) -> Option<String> {
-        self.0.write().remove(k)
+        self.0.remove(k).map(|(_, v)| v)
     }
 }
 

crates/storage/provider/src/providers/state/overlay.rs

@@ -1,6 +1,6 @@
 use alloy_primitives::{BlockNumber, B256};
+use dashmap::DashMap;
 use metrics::{Counter, Histogram};
-use parking_lot::RwLock;
 use reth_chain_state::LazyOverlay;
 use reth_db_api::DatabaseError;
 use reth_errors::{ProviderError, ProviderResult};
@@ -22,7 +22,6 @@ use reth_trie_db::{
     ChangesetCache, DatabaseHashedCursorFactory, DatabaseHashedPostState, DatabaseTrieCursorFactory,
 };
 use std::{
-    collections::{hash_map::Entry, HashMap},
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -102,7 +101,7 @@ pub struct OverlayStateProviderFactory<F> {
     metrics: OverlayStateProviderMetrics,
     /// A cache which maps `db_tip -> Overlay`. If the db tip changes during usage of the factory
     /// then a new entry will get added to this, but in most cases only one entry is present.
-    overlay_cache: Arc<RwLock<HashMap<BlockNumber, Overlay>>>,
+    overlay_cache: Arc<DashMap<BlockNumber, Overlay>>,
 }
 
 impl<F> OverlayStateProviderFactory<F> {
@@ -419,17 +418,16 @@ where
         let db_tip_block = self.get_db_tip_block_number(provider)?;
 
         // If the overlay is present in the cache then return it directly.
-        if let Some(overlay) = self.overlay_cache.as_ref().read().get(&db_tip_block) {
-            return Ok(overlay.clone());
+        if let Some(entry) = self.overlay_cache.get(&db_tip_block) {
+            return Ok(entry.value().clone());
         }
 
-        // If the overlay is not present then we need to calculate a new one. We grab a write lock,
-        // and then check the cache again in case some other thread populated the cache since we
-        // checked with the read-lock. If still not present we calculate and populate.
+        // If the overlay is not present then we need to calculate a new one.
+        // DashMap's entry API handles the race condition internally.
         let mut cache_miss = false;
-        let overlay = match self.overlay_cache.as_ref().write().entry(db_tip_block) {
-            Entry::Occupied(entry) => entry.get().clone(),
-            Entry::Vacant(entry) => {
+        let overlay = match self.overlay_cache.entry(db_tip_block) {
+            dashmap::Entry::Occupied(entry) => entry.get().clone(),
+            dashmap::Entry::Vacant(entry) => {
                 cache_miss = true;
                 let overlay = self.calculate_overlay(provider, db_tip_block)?;
                 entry.insert(overlay.clone());

crates/storage/rpc-provider/Cargo.toml

@@ -40,7 +40,7 @@ tokio = { workspace = true, features = ["sync", "macros", "rt-multi-thread"] }
 
 # other
 tracing.workspace = true
-parking_lot.workspace = true
+dashmap = { workspace = true, features = ["inline"] }
 
 # revm
 revm.workspace = true

crates/storage/rpc-provider/src/lib.rs

@@ -27,13 +27,11 @@
 use alloy_consensus::{constants::KECCAK_EMPTY, BlockHeader};
 use alloy_eips::{BlockHashOrNumber, BlockNumberOrTag};
 use alloy_network::{primitives::HeaderResponse, BlockResponse};
-use alloy_primitives::{
-    map::HashMap, Address, BlockHash, BlockNumber, StorageKey, TxHash, TxNumber, B256, U256,
-};
+use alloy_primitives::{Address, BlockHash, BlockNumber, StorageKey, TxHash, TxNumber, B256, U256};
 use alloy_provider::{ext::DebugApi, network::Network, Provider};
 use alloy_rpc_types::{AccountInfo, BlockId};
 use alloy_rpc_types_engine::ForkchoiceState;
-use parking_lot::RwLock;
+use dashmap::DashMap;
 use reth_chainspec::{ChainInfo, ChainSpecProvider};
 use reth_db_api::{
     mock::{DatabaseMock, TxMock},
@@ -912,7 +910,7 @@ where
     /// Cached bytecode for accounts
     ///
     /// Since the state provider is short-lived, we don't worry about memory leaks.
-    code_store: RwLock<HashMap<B256, Bytecode>>,
+    code_store: DashMap<B256, Bytecode>,
     /// Whether to use Reth-specific RPC methods for better performance
     reth_rpc_support: bool,
 }
@@ -942,7 +940,7 @@ impl<P: Clone, Node: NodeTypes, N> RpcBlockchainStateProvider<P, Node, N> {
             network: std::marker::PhantomData,
             chain_spec: None,
             compute_state_root: false,
-            code_store: RwLock::new(HashMap::default()),
+            code_store: Default::default(),
             reth_rpc_support: true,
         }
     }
@@ -960,7 +958,7 @@ impl<P: Clone, Node: NodeTypes, N> RpcBlockchainStateProvider<P, Node, N> {
             network: std::marker::PhantomData,
             chain_spec: Some(chain_spec),
             compute_state_root: false,
-            code_store: RwLock::new(HashMap::default()),
+            code_store: Default::default(),
             reth_rpc_support: true,
         }
     }
@@ -982,7 +980,7 @@ impl<P: Clone, Node: NodeTypes, N> RpcBlockchainStateProvider<P, Node, N> {
             network: self.network,
             chain_spec: self.chain_spec.clone(),
             compute_state_root: self.compute_state_root,
-            code_store: RwLock::new(HashMap::default()),
+            code_store: Default::default(),
             reth_rpc_support: self.reth_rpc_support,
         }
     }
@@ -1038,9 +1036,7 @@ impl<P: Clone, Node: NodeTypes, N> RpcBlockchainStateProvider<P, Node, N> {
             let code_hash = account_info.code_hash();
             if code_hash != KECCAK_EMPTY {
                 // Insert code into the cache
-                self.code_store
-                    .write()
-                    .insert(code_hash, Bytecode::new_raw(account_info.code.clone()));
+                self.code_store.insert(code_hash, Bytecode::new_raw(account_info.code.clone()));
             }
 
             Ok(account_info)
@@ -1119,7 +1115,7 @@ where
 {
     fn bytecode_by_hash(&self, code_hash: &B256) -> Result<Option<Bytecode>, ProviderError> {
         if !self.reth_rpc_support {
-            return Ok(self.code_store.read().get(code_hash).cloned());
+            return Ok(self.code_store.get(code_hash).map(|entry| entry.value().clone()));
         }
 
         self.block_on_async(async {