gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-06-30 03:14:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use hashlink over lru for LruCache (#8911)

Browse Source 
Use the `LruCache` implementation provided by `hashlink` instead of the current `lru` one.
This is mostly a 1-to-1 swap with only slight API incompatibilities.
I have decided to leave some config files which previously used `NonZeroUsize` but they may not be required anymore and could potentially switch to `usize`.


Co-Authored-By: Mac L <mjladson@pm.me>
This commit is contained in:
Mac L
2026-06-16 10:54:11 +04:00
committed by GitHub
parent 41dff2d965
commit e0ff3b5709
28 changed files with 166 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
beacon_node/beacon_chain/Cargo.toml
View File

@@ -35,13 +35,13 @@ fixed_bytes = { workspace = true }
fork_choice = { workspace = true }
futures = { workspace = true }
genesis = { workspace = true }
hashlink = { workspace = true }
hex = { workspace = true }
int_to_bytes = { workspace = true }
itertools = { workspace = true }
kzg = { workspace = true }
lighthouse_version = { workspace = true }
logging = { workspace = true }
lru = { workspace = true }
merkle_proof = { workspace = true }
metrics = { workspace = true }
milhouse = { workspace = true }

13
beacon_node/beacon_chain/src/beacon_proposer_cache.rs
View File

@@ -10,21 +10,19 @@
use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
use fork_choice::ExecutionStatus;
use lru::LruCache;
use hashlink::lru_cache::LruCache;
use once_cell::sync::OnceCell;
use parking_lot::Mutex;
use safe_arith::SafeArith;
use smallvec::SmallVec;
use state_processing::state_advance::partial_state_advance;
use std::num::NonZeroUsize;
use std::sync::Arc;
use tracing::{debug, instrument};
use typenum::Unsigned;
use types::new_non_zero_usize;
use types::{BeaconState, BeaconStateError, ChainSpec, Epoch, EthSpec, Fork, Hash256, Slot};
/// The number of sets of proposer indices that should be cached.
const CACHE_SIZE: NonZeroUsize = new_non_zero_usize(16);
const CACHE_SIZE: usize = 16;
/// This value is fairly unimportant, it's used to avoid heap allocations. The result of it being
/// incorrect is non-substantial from a consensus perspective (and probably also from a
@@ -138,7 +136,8 @@ impl BeaconProposerCache {
) -> Arc<OnceCell<EpochBlockProposers>> {
let key = (epoch, shuffling_decision_block);
self.cache
.get_or_insert(key, || Arc::new(OnceCell::new()))
.entry(key)
.or_insert_with(|| Arc::new(OnceCell::new()))
.clone()
}
@@ -155,10 +154,10 @@ impl BeaconProposerCache {
fork: Fork,
) -> Result<(), BeaconStateError> {
let key = (epoch, shuffling_decision_block);
if !self.cache.contains(&key) {
if !self.cache.contains_key(&key) {
let epoch_proposers = EpochBlockProposers::new(epoch, fork, proposers);
self.cache
.put(key, Arc::new(OnceCell::with_value(epoch_proposers)));
.insert(key, Arc::new(OnceCell::with_value(epoch_proposers)));
}
Ok(())

9
beacon_node/beacon_chain/src/data_availability_checker.rs
View File

@@ -11,7 +11,6 @@ use slot_clock::SlotClock;
use std::collections::HashSet;
use std::fmt;
use std::fmt::Debug;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::time::Duration;
use task_executor::TaskExecutor;
@@ -20,7 +19,7 @@ use types::data::{BlobIdentifier, FixedBlobSidecarList, PartialDataColumn};
use types::{
BlobSidecar, BlobSidecarList, BlockImportSource, ChainSpec, DataColumnSidecar,
DataColumnSidecarList, Epoch, EthSpec, Hash256, PartialDataColumnSidecarError,
PartialDataColumnSidecarRef, SignedBeaconBlock, Slot, new_non_zero_usize,
PartialDataColumnSidecarRef, SignedBeaconBlock, Slot,
};
mod error;
@@ -49,7 +48,7 @@ pub use error::{Error as AvailabilityCheckError, ErrorCategory as AvailabilityCh
///
/// `PendingComponents` are now never removed from the cache manually are only removed via LRU
/// eviction to prevent race conditions (#7961), so we expect this cache to be full all the time.
const OVERFLOW_LRU_CAPACITY_NON_ZERO: NonZeroUsize = new_non_zero_usize(32);
const OVERFLOW_LRU_CAPACITY: usize = 32;
/// Cache to hold fully valid data that can't be imported to fork-choice yet. After Dencun hard-fork
/// blocks have a sidecar of data that is received separately from the network. We call the concept
@@ -124,13 +123,13 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
enable_partial_columns: bool,
) -> Result<Self, AvailabilityCheckError> {
let inner = DataAvailabilityCheckerInner::new(
OVERFLOW_LRU_CAPACITY_NON_ZERO,
OVERFLOW_LRU_CAPACITY,
custody_context.clone(),
spec.clone(),
)?;
let partial_assembler = if enable_partial_columns {
Some(Arc::new(PartialDataColumnAssembler::new(
OVERFLOW_LRU_CAPACITY_NON_ZERO,
OVERFLOW_LRU_CAPACITY,
)))
} else {
None

21
beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
View File

@@ -7,11 +7,10 @@ use crate::block_verification_types::{
use crate::data_availability_checker::{Availability, AvailabilityCheckError};
use crate::data_column_verification::KzgVerifiedCustodyDataColumn;
use crate::{BeaconChainTypes, BlockProcessStatus};
use lru::LruCache;
use hashlink::lru_cache::LruCache;
use parking_lot::{MappedRwLockReadGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
use ssz_types::RuntimeFixedVector;
use std::cmp::Ordering;
use std::num::NonZeroUsize;
use std::sync::Arc;
use tracing::{Span, debug, debug_span};
use types::data::BlobIdentifier;
@@ -365,7 +364,7 @@ pub(crate) enum ReconstructColumnsDecision<E: EthSpec> {
impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
pub fn new(
capacity: NonZeroUsize,
capacity: usize,
custody_context: Arc<CustodyContext<T::EthSpec>>,
spec: Arc<ChainSpec>,
) -> Result<Self, AvailabilityCheckError> {
@@ -565,7 +564,7 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
let mut write_lock = self.critical.write();
{
let pending_components = write_lock.get_or_insert_mut(block_root, || {
let pending_components = write_lock.entry(block_root).or_insert_with(|| {
PendingComponents::empty(block_root, self.spec.max_blobs_per_block(epoch) as usize)
});
update_fn(pending_components)?
@@ -672,7 +671,7 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
if let Some(BlockProcessStatus::NotValidated(_, _)) = self.get_cached_block(block_root) {
// If the block is execution invalid, this status is permanent and idempotent to this
// block_root. We drop its components (e.g. columns) because they will never be useful.
self.critical.write().pop(block_root);
self.critical.write().remove(block_root);
}
}
@@ -733,7 +732,7 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
}
// Now remove keys
for key in keys_to_remove {
write_lock.pop(&key);
write_lock.remove(&key);
}
Ok(())
@@ -765,7 +764,6 @@ mod test {
use store::{HotColdDB, ItemStore, StoreConfig, database::interface::BeaconNodeBackend};
use tempfile::{TempDir, tempdir};
use tracing::info;
use types::new_non_zero_usize;
use types::{DataColumnSubnetId, MinimalEthSpec};
const LOW_VALIDATOR_COUNT: usize = 32;
@@ -930,19 +928,14 @@ mod test {
let chain_db_path = tempdir().expect("should get temp dir");
let harness = get_fulu_chain(&chain_db_path).await;
let spec = harness.spec.clone();
let capacity_non_zero = new_non_zero_usize(capacity);
let custody_context = Arc::new(CustodyContext::new(
NodeCustodyType::Fullnode,
generate_data_column_indices_rand_order::<E>(),
&spec,
));
let cache = Arc::new(
DataAvailabilityCheckerInner::<T>::new(
capacity_non_zero,
custody_context,
spec.clone(),
)
.expect("should create cache"),
DataAvailabilityCheckerInner::<T>::new(capacity, custody_context, spec.clone())
.expect("should create cache"),
);
(harness, cache, chain_db_path)
}

3
beacon_node/beacon_chain/src/fetch_blobs/tests.rs
View File

@@ -9,7 +9,6 @@ use eth2::types::BlobsBundle;
use execution_layer::json_structures::{BlobAndProof, BlobAndProofV1, BlobAndProofV2};
use execution_layer::test_utils::generate_blobs;
use maplit::hashset;
use std::num::NonZeroUsize;
use std::sync::{Arc, Mutex};
use task_executor::test_utils::TestRuntime;
use types::{
@@ -339,7 +338,7 @@ fn mock_beacon_adapter(fork_name: ForkName, get_blobs_v3: bool) -> MockFetchBlob
let test_runtime = TestRuntime::default();
let spec = Arc::new(fork_name.make_genesis_spec(E::default_spec()));
let kzg = get_kzg(&spec);
let partial_assembler = PartialDataColumnAssembler::new(NonZeroUsize::new(32).unwrap());
let partial_assembler = PartialDataColumnAssembler::new(32);
let mut mock_adapter = MockFetchBlobsBeaconAdapter::default();
mock_adapter.expect_spec().return_const(spec.clone());

13
beacon_node/beacon_chain/src/light_client_server_cache.rs
View File

@@ -1,15 +1,14 @@
use crate::errors::BeaconChainError;
use crate::{BeaconChainTypes, BeaconStore, metrics};
use hashlink::lru_cache::LruCache;
use parking_lot::{Mutex, RwLock};
use safe_arith::SafeArith;
use ssz::Decode;
use std::num::NonZeroUsize;
use std::sync::Arc;
use store::DBColumn;
use store::KeyValueStore;
use tracing::debug;
use tree_hash::TreeHash;
use types::new_non_zero_usize;
use types::{
BeaconBlockRef, BeaconState, ChainSpec, Checkpoint, EthSpec, ForkName, Hash256,
LightClientBootstrap, LightClientFinalityUpdate, LightClientOptimisticUpdate,
@@ -19,7 +18,7 @@ use types::{
/// A prev block cache miss requires to re-generate the state of the post-parent block. Items in the
/// prev block cache are very small 32 * (6 + 1) = 224 bytes. 32 is an arbitrary number that
/// represents unlikely re-orgs, while keeping the cache very small.
const PREV_BLOCK_CACHE_SIZE: NonZeroUsize = new_non_zero_usize(32);
const PREV_BLOCK_CACHE_SIZE: usize = 32;
/// This cache computes light client messages ahead of time, required to satisfy p2p and API
/// requests. These messages include proofs on historical states, so on-demand computation is
@@ -39,7 +38,7 @@ pub struct LightClientServerCache<T: BeaconChainTypes> {
/// Caches the current sync committee,
latest_written_current_sync_committee: RwLock<Option<Arc<SyncCommittee<T::EthSpec>>>>,
/// Caches state proofs by block root
prev_block_cache: Mutex<lru::LruCache<Hash256, LightClientCachedData<T::EthSpec>>>,
prev_block_cache: Mutex<LruCache<Hash256, LightClientCachedData<T::EthSpec>>>,
/// Tracks the latest broadcasted finality update
latest_broadcasted_finality_update: RwLock<Option<LightClientFinalityUpdate<T::EthSpec>>>,
/// Tracks the latest broadcasted optimistic update
@@ -55,7 +54,7 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
latest_written_current_sync_committee: None.into(),
latest_broadcasted_finality_update: None.into(),
latest_broadcasted_optimistic_update: None.into(),
prev_block_cache: lru::LruCache::new(PREV_BLOCK_CACHE_SIZE).into(),
prev_block_cache: LruCache::new(PREV_BLOCK_CACHE_SIZE).into(),
}
}
@@ -74,7 +73,7 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
if fork_name.altair_enabled() {
// Persist in memory cache for a descendent block
let cached_data = LightClientCachedData::from_state(block_post_state)?;
self.prev_block_cache.lock().put(block_root, cached_data);
self.prev_block_cache.lock().insert(block_root, cached_data);
}
Ok(())
@@ -335,7 +334,7 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
// Insert value and return owned
self.prev_block_cache
.lock()
.put(*block_root, new_value.clone());
.insert(*block_root, new_value.clone());
Ok(new_value)
}

57
beacon_node/beacon_chain/src/partial_data_column_assembler.rs
View File

@@ -1,10 +1,9 @@
use crate::data_column_verification::{
KzgVerifiedCustodyDataColumn, KzgVerifiedCustodyPartialDataColumn,
};
use lru::LruCache;
use hashlink::lru_cache::LruCache;
use parking_lot::RwLock;
use std::collections::HashMap;
use std::num::NonZeroUsize;
use std::sync::Arc;
use tracing::error;
use types::core::{Epoch, EthSpec, Hash256};
@@ -44,7 +43,7 @@ pub struct PartialMergeResult<E: EthSpec> {
}
impl<E: EthSpec> PartialDataColumnAssembler<E> {
pub fn new(capacity: NonZeroUsize) -> Self {
pub fn new(capacity: usize) -> Self {
Self {
assemblies: RwLock::new(LruCache::new(capacity)),
}
@@ -55,7 +54,7 @@ impl<E: EthSpec> PartialDataColumnAssembler<E> {
pub fn init(&self, block_root: Hash256, header: Arc<PartialDataColumnHeader<E>>) -> bool {
let mut assemblies = self.assemblies.write();
if assemblies.contains(&block_root) {
if assemblies.contains_key(&block_root) {
return false;
}
@@ -65,7 +64,7 @@ impl<E: EthSpec> PartialDataColumnAssembler<E> {
columns: HashMap::new(),
};
assemblies.put(block_root, assembly);
assemblies.insert(block_root, assembly);
true
}
@@ -79,11 +78,13 @@ impl<E: EthSpec> PartialDataColumnAssembler<E> {
header: Arc<PartialDataColumnHeader<E>>,
) -> Option<PartialMergeResult<E>> {
let mut assemblies = self.assemblies.write();
let assembly = assemblies.get_or_insert_mut(block_root, || PartialAssembly {
header: header.clone(),
has_local_blobs: false,
columns: HashMap::new(),
});
let assembly = assemblies
.entry(block_root)
.or_insert_with(|| PartialAssembly {
header: header.clone(),
has_local_blobs: false,
columns: HashMap::new(),
});
let mut full_columns = Vec::new();
let mut updated_partials = Vec::new();
@@ -165,15 +166,17 @@ impl<E: EthSpec> PartialDataColumnAssembler<E> {
};
let mut assemblies = self.assemblies.write();
let assembly = assemblies.get_or_insert_mut(block_root, || PartialAssembly {
header: Arc::new(PartialDataColumnHeader {
kzg_commitments: fulu.kzg_commitments.clone(),
signed_block_header: fulu.signed_block_header.clone(),
kzg_commitments_inclusion_proof: fulu.kzg_commitments_inclusion_proof.clone(),
}),
has_local_blobs: false,
columns: Default::default(),
});
let assembly = assemblies
.entry(block_root)
.or_insert_with(|| PartialAssembly {
header: Arc::new(PartialDataColumnHeader {
kzg_commitments: fulu.kzg_commitments.clone(),
signed_block_header: fulu.signed_block_header.clone(),
kzg_commitments_inclusion_proof: fulu.kzg_commitments_inclusion_proof.clone(),
}),
has_local_blobs: false,
columns: Default::default(),
});
let prev = assembly
.columns
.insert(column.index(), AssemblyColumn::Complete(column.clone()));
@@ -215,11 +218,13 @@ impl<E: EthSpec> PartialDataColumnAssembler<E> {
header: &Arc<PartialDataColumnHeader<E>>,
) -> Vec<AssemblyColumn<E>> {
let mut assemblies = self.assemblies.write();
let assembly = assemblies.get_or_insert_mut(block_root, || PartialAssembly {
header: header.clone(),
has_local_blobs: true,
columns: Default::default(),
});
let assembly = assemblies
.entry(block_root)
.or_insert_with(|| PartialAssembly {
header: header.clone(),
has_local_blobs: true,
columns: Default::default(),
});
assembly.has_local_blobs = true;
@@ -253,7 +258,7 @@ impl<E: EthSpec> PartialDataColumnAssembler<E> {
}
for root in to_remove {
assemblies.pop(&root);
assemblies.remove(&root);
}
}
}
@@ -362,7 +367,7 @@ mod tests {
}
fn make_assembler() -> PartialDataColumnAssembler<E> {
PartialDataColumnAssembler::new(NonZeroUsize::new(16).unwrap())
PartialDataColumnAssembler::new(16)
}
// -- init and get_header tests --

15
beacon_node/beacon_chain/src/pending_payload_cache/mod.rs
View File

@@ -15,13 +15,12 @@ use crate::payload_envelope_verification::{
AvailabilityPendingExecutedEnvelope, AvailableExecutedEnvelope,
};
use crate::{BeaconChainTypes, CustodyContext, metrics};
use hashlink::lru_cache::LruCache;
use kzg::Kzg;
use lru::LruCache;
use parking_lot::{MappedRwLockReadGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
use std::collections::HashMap;
use std::fmt;
use std::fmt::Debug;
use std::num::NonZeroUsize;
use std::sync::Arc;
use tracing::{Span, debug, error, instrument};
use types::{
@@ -41,7 +40,6 @@ use crate::metrics::{
use crate::observed_data_sidecars::ObservationStrategy;
use pending_components::{PendingComponents, ReconstructColumnsDecision};
use types::SignedExecutionPayloadBid;
use types::new_non_zero_usize;
/// The LRU Cache stores `PendingComponents`, which store the block root, the execution payload bid, and its associated column data.
/// The execution payload bid stores the kzg commitments which we use to verify against incoming column data.
@@ -49,7 +47,7 @@ use types::new_non_zero_usize;
///
/// `PendingComponents` are now never removed from the cache manually and are only removed via LRU
/// eviction to prevent race conditions (#7961), so we expect this cache to be full all the time.
const AVAILABILITY_CACHE_CAPACITY: NonZeroUsize = new_non_zero_usize(32);
const AVAILABILITY_CACHE_CAPACITY: usize = 32;
/// This type is returned after adding a bid / column to the `DataAvailabilityChecker`.
///
@@ -206,7 +204,9 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
/// This will silently drop the bid if a bid for this block root already exists in the cache.
pub fn insert_bid(&self, block_root: Hash256, bid: Arc<SignedExecutionPayloadBid<T::EthSpec>>) {
let mut write_lock = self.availability_cache.write();
write_lock.get_or_insert_mut(block_root, || PendingComponents::new(block_root, bid));
write_lock
.entry(block_root)
.or_insert_with(|| PendingComponents::new(block_root, bid));
}
/// Perform KZG verification on RPC custody columns and insert them into the cache.
@@ -423,7 +423,8 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
{
let pending_components = write_lock
.get_or_insert_mut(block_root, || PendingComponents::new(block_root, bid));
.entry(block_root)
.or_insert_with(|| PendingComponents::new(block_root, bid));
update_fn(pending_components)
}
@@ -496,7 +497,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
}
}
for key in keys_to_remove {
write_lock.pop(&key);
write_lock.remove(&key);
}
Ok(())

28
beacon_node/beacon_chain/src/pre_finalization_cache.rs
View File

@@ -1,15 +1,13 @@
use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
use hashlink::lru_cache::LruCache;
use itertools::process_results;
use lru::LruCache;
use parking_lot::Mutex;
use std::num::NonZeroUsize;
use std::time::Duration;
use tracing::debug;
use types::Hash256;
use types::new_non_zero_usize;
const BLOCK_ROOT_CACHE_LIMIT: NonZeroUsize = new_non_zero_usize(512);
const LOOKUP_LIMIT: NonZeroUsize = new_non_zero_usize(8);
const BLOCK_ROOT_CACHE_LIMIT: usize = 512;
const LOOKUP_LIMIT: usize = 8;
const METRICS_TIMEOUT: Duration = Duration::from_millis(100);
/// Cache for rejecting attestations to blocks from before finalization.
@@ -49,13 +47,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
let mut cache = self.pre_finalization_block_cache.cache.lock();
// Check the cache to see if we already know this pre-finalization block root.
if cache.block_roots.contains(&block_root) {
if cache.block_roots.contains_key(&block_root) {
return Ok(true);
}
// Avoid repeating the disk lookup for blocks that are already subject to a network lookup.
// Sync will take care of de-duplicating the single block lookups.
if cache.in_progress_lookups.contains(&block_root) {
if cache.in_progress_lookups.contains_key(&block_root) {
return Ok(false);
}
@@ -68,19 +66,19 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
.map_err(BeaconChainError::BeaconStateError)
})?;
if is_recent_finalized_block {
cache.block_roots.put(block_root, ());
cache.block_roots.insert(block_root, ());
return Ok(true);
}
// 2. Check on disk.
if self.store.get_blinded_block(&block_root)?.is_some() {
cache.block_roots.put(block_root, ());
cache.block_roots.insert(block_root, ());
return Ok(true);
}
// 3. Check the network with a single block lookup.
cache.in_progress_lookups.put(block_root, ());
if cache.in_progress_lookups.len() == LOOKUP_LIMIT.get() {
cache.in_progress_lookups.insert(block_root, ());
if cache.in_progress_lookups.len() == LOOKUP_LIMIT {
// NOTE: we expect this to occur sometimes if a lot of blocks that we look up fail to be
// imported for reasons other than being pre-finalization. The cache will eventually
// self-repair in this case by replacing old entries with new ones until all the failed
@@ -95,8 +93,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
pub fn pre_finalization_block_rejected(&self, block_root: Hash256) {
// Future requests can know that this block is invalid without having to look it up again.
let mut cache = self.pre_finalization_block_cache.cache.lock();
cache.in_progress_lookups.pop(&block_root);
cache.block_roots.put(block_root, ());
cache.in_progress_lookups.remove(&block_root);
cache.block_roots.insert(block_root, ());
}
}
@@ -104,11 +102,11 @@ impl PreFinalizationBlockCache {
pub fn block_processed(&self, block_root: Hash256) {
// Future requests will find this block in fork choice, so no need to cache it in the
// ongoing lookup cache any longer.
self.cache.lock().in_progress_lookups.pop(&block_root);
self.cache.lock().in_progress_lookups.remove(&block_root);
}
pub fn contains(&self, block_root: Hash256) -> bool {
self.cache.lock().block_roots.contains(&block_root)
self.cache.lock().block_roots.contains_key(&block_root)
}
pub fn metrics(&self) -> Option<(usize, usize)> {