diff --git a/Cargo.lock b/Cargo.lock index 9391d686..56bcfd04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2401,6 +2401,7 @@ dependencies = [ "agave-feature-set", "ahash", "arc-swap", + "crossbeam-queue", "derive_more", "flume", "magicblock-accountsdb", diff --git a/Cargo.toml b/Cargo.toml index f807321d..a782470a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -55,6 +55,7 @@ bytemuck = "1.25.2" cfg-if = "1.0.4" clonetree = "0.0.2" criterion = "0.8.2" +crossbeam-queue = "0.3.14" derive_more = "2.1.1" env_logger = "0.11.11" fjall = { version = "3.1.10", default-features = false } diff --git a/engine/tests/recency.rs b/engine/tests/recency.rs new file mode 100644 index 00000000..a2b16e4d --- /dev/null +++ b/engine/tests/recency.rs @@ -0,0 +1,91 @@ +//! Recency eviction under RPC fan-out: accounts read on every request must +//! stay resident while each request also touches many one-off accounts. +//! +//! `missing_accounts` returns exactly the accounts a synchronizing caller +//! fetches upstream, so its size per round is the upstream fetch cost. The +//! test drains evictions the way the validator does: claim the cached eviction +//! under the account lease and delete the account. +#![cfg(test)] + +use engine::{ + Engine, + testkit::{Pacing, TestEngine}, +}; +use keeper::testkit::{Dirs, keeper_builder}; +use solana_account::{AccountBuilder, AccountMode}; +use solana_pubkey::Pubkey; +use tokio::sync::mpsc::UnboundedReceiver; + +const CAPACITY: usize = 256; +const HOT: usize = 50; +const ONE_OFF_PER_ROUND: usize = 300; +const BATCH: usize = 100; +const ROUNDS: usize = 15; +/// Rounds allowed for hot accounts to graduate from probation. Each round's +/// one-off traffic exceeds the cache, so a hot account evicted before its +/// first hit reaches main only if its ghost entry survives the round (about +/// two thirds of the time here). Stragglers shrink geometrically, so after +/// this many rounds the chance any of the 50 remains is about 0.1%. +const WARMUP_ROUNDS: usize = 12; + +/// Materializes every requested account the engine reports missing, then +/// applies the resulting evictions. Returns how many accounts were fetched. +async fn request( + engine: &Engine, + evictions: &mut UnboundedReceiver, + keys: &[Pubkey], + owner: Pubkey, +) -> usize { + let missing = engine.missing_accounts(keys).await.unwrap(); + let fetched = missing.len(); + for accessor in missing { + let account = AccountBuilder::default() + .lamports(2_000_000) + .owner(owner) + .mode(AccountMode::Mirrored) + .slot(1) + .data(vec![1]); + accessor.materialize(account, None).await.unwrap(); + } + while let Ok(evicted) = evictions.try_recv() { + let accessor = engine.account(evicted).await.unwrap(); + if let Some(accessor) = accessor.into_cached_eviction() { + accessor.delete().await.unwrap(); + } + } + fetched +} + +/// Proves hot accounts stop being re-fetched once they graduate from +/// probation, while every request also fans out to one-off accounts that +/// exceed the whole cache. An LRU re-fetches most of the hot set every round. +#[tokio::test(flavor = "multi_thread")] +async fn hot_accounts_survive_one_off_fanout() { + let dirs = Dirs::default(); + let mut builder = keeper_builder(&dirs); + builder.accountsdb.lru_capacity = CAPACITY; + let te = TestEngine::from_builder(dirs, builder, Pacing::External).await; + let mut evictions = te.accounts().subscribe_evictions().unwrap(); + let owner = Pubkey::new_unique(); + let hot: Vec<_> = (0..HOT).map(|_| Pubkey::new_unique()).collect(); + + let mut refetched_per_round = Vec::with_capacity(ROUNDS); + for _ in 0..ROUNDS { + refetched_per_round.push(request(&te, &mut evictions, &hot, owner).await); + let one_off: Vec<_> = (0..ONE_OFF_PER_ROUND).map(|_| Pubkey::new_unique()).collect(); + for batch in one_off.chunks(BATCH) { + assert_eq!( + request(&te, &mut evictions, batch, owner).await, + batch.len(), + "one-off accounts are always fetched" + ); + } + } + + println!("hot accounts fetched per round: {refetched_per_round:?}"); + assert!( + refetched_per_round.iter().skip(WARMUP_ROUNDS).all(|count| *count == 0), + "hot accounts are still evicted by one-off traffic and re-fetched upstream \ + after {WARMUP_ROUNDS} warm-up rounds: {refetched_per_round:?}" + ); +} diff --git a/keeper/Cargo.toml b/keeper/Cargo.toml index 6005393e..e2f4e3d7 100644 --- a/keeper/Cargo.toml +++ b/keeper/Cargo.toml @@ -24,6 +24,7 @@ nucleus = { workspace = true, features = ["config", "notifier", "runtime"] } ahash = { workspace = true } arc-swap = { workspace = true } +crossbeam-queue = { workspace = true } derive_more = { workspace = true, features = ["from"] } flume = { workspace = true } oneshot = { workspace = true } diff --git a/keeper/src/accessor.rs b/keeper/src/accessor.rs index 27cffa97..b0c7a1b8 100644 --- a/keeper/src/accessor.rs +++ b/keeper/src/accessor.rs @@ -47,9 +47,12 @@ pub struct AccountsAccessor<'a> { } impl<'a> AccountsAccessor<'a> { - /// Promotes an existing recency entry without loading or pinning the account. + /// Records a read hit on a resident recency entry without acquiring a + /// lease or loading the account. Lock-free; absent accounts are ignored. + /// + /// Frequently touched accounts are retained over accounts read once. pub fn touch(&self, pubkey: &Pubkey) { - self.keeper.caches.accounts.lru.get_sync(pubkey); + self.keeper.caches.accounts.recency.touch(pubkey); } /// Waits for exclusive mutation ownership of `pubkey` and records the diff --git a/keeper/src/cache.rs b/keeper/src/cache.rs index 4d062578..8c35d03c 100644 --- a/keeper/src/cache.rs +++ b/keeper/src/cache.rs @@ -10,13 +10,14 @@ use ledger::{ }; use nucleus::{Slot, ledger::Block, notifier::EventNotifier}; use parking_lot::Mutex; -use scc::{HashCache, HashMap, hash_cache::Entry as CacheEntry, hash_map::Entry}; +use scc::{HashMap, hash_map::Entry}; use solana_account::AccountMode; use solana_hash::Hash as SolanaHash; use solana_pubkey::Pubkey; use crate::{ metrics, + s3fifo::{Evicted, S3Fifo}, subscriptions::{Subscription, Unicast}, }; @@ -68,8 +69,8 @@ impl Caches { /// Account access cache along with per-account mutation reservations. pub(crate) struct AccountCache { - /// Accounts that do not require persistence, ordered by recent access. - pub(crate) lru: HashCache, + /// Resident accounts that do not require persistence, under S3-FIFO eviction. + pub(crate) recency: S3Fifo, /// In-flight account mutations keyed by account pubkey. pub(crate) reservations: HashMap, RandomState>, /// Pubkeys evicted when a materialization displaces a cold account. @@ -79,9 +80,8 @@ pub(crate) struct AccountCache { impl AccountCache { /// Creates mutation coordination with the requested recency capacity. pub(crate) fn new(capacity: usize) -> Self { - let lru = HashCache::with_capacity_and_hasher(capacity, capacity, Default::default()); Self { - lru, + recency: S3Fifo::new(capacity), reservations: Default::default(), evictions: Unicast::new(Subscription::Evictions), } @@ -137,7 +137,7 @@ impl AccountLease { /// Updates recency after materializing the account in `mode`. pub fn materialized(&mut self, mode: AccountMode) { - if let Some(pubkey) = self.cache.track(self.pubkey, mode) { + for pubkey in self.cache.track(self.pubkey, mode) { self.cache.evictions.send(|| pubkey); } } @@ -150,7 +150,7 @@ impl AccountLease { /// Returns whether an earlier recency eviction still applies to the /// observed account and cache entry. pub fn cached_eviction_applies(&self) -> bool { - self.evictable() && !self.cache.lru.contains_sync(&self.pubkey) + self.evictable() && !self.cache.recency.contains(&self.pubkey) } } @@ -163,40 +163,32 @@ impl Drop for AccountLease { } impl AccountCache { - /// Tracks a materialized account and returns one displaced from recency. - fn track(&self, pubkey: Pubkey, mode: AccountMode) -> Option { + /// Tracks a materialized account and returns the accounts it displaced. + fn track(&self, pubkey: Pubkey, mode: AccountMode) -> Evicted { if !mode.requires_persistence() { return self.admit_recency(pubkey); } self.remove_recency(&pubkey); - None + Evicted::new() } - /// Promotes an existing entry or admits `pubkey`, returning any displaced - /// account. - fn admit_recency(&self, pubkey: Pubkey) -> Option { - let entry = match self.lru.entry_sync(pubkey) { - CacheEntry::Occupied(entry) => { - // Entry access does not promote recency; release its lock and - // use the cache access API for that operation. - drop(entry); - self.lru.get_sync(&pubkey); - return None; - } - CacheEntry::Vacant(entry) => entry, - }; - let (evicted, _) = entry.put_entry(()); - if evicted.is_some() { - metrics::account_cache_eviction(); - } else { + /// Counts a hit on a resident entry or admits `pubkey`, returning any + /// displaced accounts. + fn admit_recency(&self, pubkey: Pubkey) -> Evicted { + let admission = self.recency.admit(pubkey); + if admission.inserted { metrics::account_cache_insert(); } - evicted.map(|entry| entry.0) + for _ in &admission.evicted { + metrics::account_cache_eviction(); + metrics::account_cache_remove(); + } + admission.evicted } /// Removes `pubkey` from recency and records an actual removal. fn remove_recency(&self, pubkey: &Pubkey) { - if self.lru.remove_sync(pubkey).is_some() { + if self.recency.remove(pubkey) { metrics::account_cache_remove(); } } diff --git a/keeper/src/lib.rs b/keeper/src/lib.rs index 20470f86..21f113ea 100644 --- a/keeper/src/lib.rs +++ b/keeper/src/lib.rs @@ -62,6 +62,7 @@ pub mod builder; mod cache; pub mod error; mod metrics; +mod s3fifo; mod subscriptions; mod util; diff --git a/keeper/src/s3fifo.rs b/keeper/src/s3fifo.rs new file mode 100644 index 00000000..43efc57f --- /dev/null +++ b/keeper/src/s3fifo.rs @@ -0,0 +1,423 @@ +//! Lock-free S3-FIFO residency cache. +//! +//! S3-FIFO (Yang et al., SOSP '23) keeps three FIFO structures: +//! +//! - **small**: probation for newly admitted keys, about 10% of capacity; +//! - **main**: keys that were hit while on probation, or readmitted from ghost; +//! - **ghost**: fingerprints of keys recently evicted from small. +//! +//! A hit only bumps a saturating 2-bit counter on the entry, so the read path +//! is one lock-free index lookup plus at most one atomic compare-exchange. Queue +//! order changes only on admission and eviction. Keys seen once leave from +//! small without displacing frequently hit keys in main, which makes the cache +//! resistant to bursts of one-off keys. +//! +//! Concurrency: keys are spread by hash over independent shards so admissions +//! and evictions on different shards never touch the same queue. Each shard +//! indexes entries in an [`scc::HashIndex`] (lock-free reads, per-bucket +//! writes) and orders them in lock-free [`SegQueue`]s. Ghost is a fixed table +//! of atomic fingerprints: one store per eviction, one compare-exchange per +//! readmission, no allocation. Colliding fingerprints overwrite each other, +//! which only makes ghost readmission approximate. +//! +//! Entries removed from the index stay in their queue as stale entries and are +//! skipped when popped. Queue occupancy, stale entries included, is bounded by +//! [`QUEUE_BOUND_FACTOR`] times the shard capacity; a push over the bound first +//! retires the queue head. + +use std::{ + hash::Hash, + num::NonZeroUsize, + sync::{ + Arc, + atomic::{AtomicBool, AtomicIsize, AtomicU8, AtomicU64, Ordering}, + }, +}; + +use ahash::RandomState; +use crossbeam_queue::SegQueue; +use scc::HashIndex; +use smallvec::SmallVec; + +/// Saturation point of the per-entry hit counter. +const MAX_FREQ: u8 = 3; +/// Share of capacity, in percent, reserved for the probationary small queue. +const SMALL_PERCENT: usize = 10; +/// Queue occupancy bound, including stale entries, as a multiple of capacity. +const QUEUE_BOUND_FACTOR: usize = 2; +/// Maximum live queue entries one eviction examines before evicting the +/// current candidate regardless of its hit counter. Stale entries are dropped +/// when popped and do not count, since they are never requeued. Without +/// concurrent hits, an eviction ends within `MAX_FREQ` passes over main; this +/// bounds the work when concurrent hits keep re-warming entries. +const EVICTION_PROBES: usize = 512; +/// Smallest shard capacity; smaller caches use fewer shards so per-shard +/// queues stay large enough for the small/main split to be meaningful. +const MIN_SHARD_CAPACITY: usize = 256; +/// Shards per available CPU, to keep contention on any one queue low. +const SHARDS_PER_CPU: usize = 4; + +/// Keys displaced by one admission; usually zero or one. +pub(crate) type Evicted = SmallVec<[K; 1]>; + +/// Outcome of [`S3Fifo::admit`]. +pub(crate) struct Admission { + /// Whether `key` became resident; `false` means it was already resident + /// and the admission counted as a hit. + pub(crate) inserted: bool, + /// Keys evicted to keep the cache within capacity. + pub(crate) evicted: Evicted, +} + +/// Bounded lock-free S3-FIFO residency set, sharded by key hash. +pub(crate) struct S3Fifo { + hasher: RandomState, + shards: Box<[Shard]>, +} + +impl S3Fifo +where + K: Copy + Eq + Hash + 'static, +{ + /// Creates an empty cache holding at most `capacity` keys (minimum one). + pub(crate) fn new(capacity: usize) -> Self { + let capacity = capacity.max(1); + let cpus = std::thread::available_parallelism().map_or(1, NonZeroUsize::get); + let count = cpus.saturating_mul(SHARDS_PER_CPU).min(capacity / MIN_SHARD_CAPACITY); + Self::with_shards(capacity, count) + } + + /// Creates an empty cache splitting `capacity` over `count` shards. + fn with_shards(capacity: usize, count: usize) -> Self { + let capacity = capacity.max(1); + let count = count.clamp(1, capacity); + let (base, extra) = (capacity / count, capacity % count); + let shards = + (0..count).map(|shard| Shard::new(base + usize::from(shard < extra))).collect(); + Self { + hasher: RandomState::new(), + shards, + } + } + + /// Records a hit on `key` if it is resident, without changing queue order. + /// Lock-free. Returns whether `key` is resident. + pub(crate) fn touch(&self, key: &K) -> bool { + self.shard(self.hash(key)).touch(key) + } + + /// Returns whether `key` is resident. Lock-free. + pub(crate) fn contains(&self, key: &K) -> bool { + self.shard(self.hash(key)).index.contains(key) + } + + /// Records a hit on a resident `key`, or admits it and evicts keys as + /// needed to stay within capacity. + pub(crate) fn admit(&self, key: K) -> Admission { + let hash = self.hash(&key); + self.shard(hash).admit(key, hash, &self.hasher) + } + + /// Removes `key`, returning whether it was resident. Its queue entry is + /// left behind as stale and skipped when popped. + pub(crate) fn remove(&self, key: &K) -> bool { + self.shard(self.hash(key)).remove(key) + } + + fn hash(&self, key: &K) -> u64 { + self.hasher.hash_one(key) + } + + fn shard(&self, hash: u64) -> &Shard { + // Ghost slots use the low bits; pick shards from the high bits. + let high = usize::try_from(hash >> 32).unwrap_or_default(); + &self.shards[high % self.shards.len()] + } + + /// Returns the number of resident keys. + #[cfg(test)] + pub(crate) fn len(&self) -> usize { + self.shards.iter().map(Shard::len).sum() + } + + /// Number of queued entries in small and main, stale ones included. + #[cfg(test)] + pub(crate) fn queued(&self) -> (usize, usize) { + self.shards.iter().fold((0, 0), |(small, main), shard| { + (small + shard.small.len(), main + shard.main.len()) + }) + } + + /// Sum of shard queue bounds, shared by small and main. + #[cfg(test)] + pub(crate) fn queue_bound(&self) -> usize { + self.shards.iter().map(|shard| shard.queue_bound).sum() + } + + /// Number of keys in the indexes; equals [`Self::len`] when quiescent. + #[cfg(test)] + pub(crate) fn indexed(&self) -> usize { + self.shards.iter().map(|shard| shard.index.len()).sum() + } + + /// Creates a cache with an explicit shard count. + #[cfg(test)] + pub(crate) fn with_shard_count(capacity: usize, count: usize) -> Self { + Self::with_shards(capacity, count) + } + + /// Number of shards. + #[cfg(test)] + pub(crate) fn shard_count(&self) -> usize { + self.shards.len() + } +} + +/// One resident key shared between the index and its current queue. +struct Node { + key: K, + freq: AtomicU8, + /// Set after the node leaves the index; queues skip it when popped. + removed: AtomicBool, +} + +impl Node { + fn new(key: K) -> Self { + Self { + key, + freq: AtomicU8::new(0), + removed: AtomicBool::new(false), + } + } + + /// Records one hit. A lost race under contention drops the increment, + /// which only makes the frequency estimate more conservative. + fn hit(&self) { + let freq = self.freq.load(Ordering::Relaxed); + if freq < MAX_FREQ { + let _ = + self.freq.compare_exchange(freq, freq + 1, Ordering::Relaxed, Ordering::Relaxed); + } + } + + fn is_warm(&self) -> bool { + self.freq.load(Ordering::Relaxed) > 0 + } + + /// Decrements the hit counter, returning whether it was non-zero. + fn cool(&self) -> bool { + self.freq + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |freq| { + freq.checked_sub(1) + }) + .is_ok() + } + + fn is_removed(&self) -> bool { + self.removed.load(Ordering::Acquire) + } +} + +/// Queue a popped node came from. +#[derive(Clone, Copy)] +enum Source { + Small, + Main, +} + +/// Independent S3-FIFO instance over a hash partition of the keys. +struct Shard { + index: HashIndex>, RandomState>, + small: SegQueue>>, + main: SegQueue>>, + /// Fingerprints of keys recently evicted from small; zero marks empty. + ghost: Box<[AtomicU64]>, + /// Resident entries. Signed because a concurrent removal can be counted + /// before the admission it removes. + len: AtomicIsize, + capacity: isize, + small_target: usize, + queue_bound: usize, +} + +impl Shard +where + K: Copy + Eq + Hash + 'static, +{ + fn new(capacity: usize) -> Self { + let capacity = capacity.max(1); + let small_target = (capacity * SMALL_PERCENT / 100).max(1); + let ghost_slots = capacity.saturating_sub(small_target).max(1).next_power_of_two(); + Self { + index: HashIndex::with_hasher(RandomState::new()), + small: SegQueue::new(), + main: SegQueue::new(), + ghost: (0..ghost_slots).map(|_| AtomicU64::new(0)).collect(), + len: AtomicIsize::new(0), + capacity: isize::try_from(capacity).unwrap_or(isize::MAX), + small_target, + queue_bound: capacity.saturating_mul(QUEUE_BOUND_FACTOR), + } + } + + fn touch(&self, key: &K) -> bool { + self.index.peek_with(key, |_, node| node.hit()).is_some() + } + + fn admit(&self, key: K, hash: u64, hasher: &RandomState) -> Admission { + let mut evicted = Evicted::new(); + if self.touch(&key) { + return Admission { inserted: false, evicted }; + } + let node = Arc::new(Node::new(key)); + if self.index.insert_sync(key, node.clone()).is_err() { + // A concurrent admission of the same key won; count it as a hit. + self.touch(&key); + return Admission { inserted: false, evicted }; + } + let len = self.len.fetch_add(1, Ordering::AcqRel) + 1; + if self.take_ghost(hash) { + self.push_main(node, &mut evicted); + } else { + self.push_small(node, hasher, &mut evicted); + } + if evicted.is_empty() && len > self.capacity { + self.evict_one(hasher, &mut evicted); + } + Admission { inserted: true, evicted } + } + + fn remove(&self, key: &K) -> bool { + let mut removed = None; + self.index.remove_if_sync(key, |node| { + removed = Some(node.clone()); + true + }); + let Some(node) = removed else { + return false; + }; + node.removed.store(true, Ordering::Release); + self.len.fetch_sub(1, Ordering::AcqRel); + true + } + + /// Evicts one resident key, examining at most [`EVICTION_PROBES`] live + /// entries. Ends early if both queues run out. + fn evict_one(&self, hasher: &RandomState, evicted: &mut Evicted) { + let mut probes = 0; + while let Some((node, source)) = self.pop_candidate() { + if node.is_removed() { + continue; + } + probes += 1; + let force = probes >= EVICTION_PROBES; + match source { + Source::Small if !force && node.is_warm() => self.push_main(node, evicted), + Source::Small => { + self.evict_node(&node, Some(hasher), evicted); + } + Source::Main if !force && node.cool() => self.push_main(node, evicted), + Source::Main => { + self.evict_node(&node, None, evicted); + } + } + if !evicted.is_empty() { + return; + } + } + } + + /// Pops from small while it holds at least its target share, otherwise + /// from main, falling back to the other queue when one is empty. + fn pop_candidate(&self) -> Option<(Arc>, Source)> { + let small = || self.small.pop().map(|node| (node, Source::Small)); + let main = || self.main.pop().map(|node| (node, Source::Main)); + if self.small.len() >= self.small_target { + small().or_else(main) + } else { + main().or_else(small) + } + } + + /// Appends `node` to small, retiring heads while the queue is at its bound. + fn push_small(&self, node: Arc>, hasher: &RandomState, evicted: &mut Evicted) { + while self.small.len() >= self.queue_bound { + let Some(head) = self.small.pop() else { + break; + }; + if head.is_removed() { + continue; + } + if head.is_warm() { + self.push_main(head, evicted); + } else { + self.evict_node(&head, Some(hasher), evicted); + } + } + self.small.push(node); + } + + /// Appends `node` to main, evicting heads while the queue is at its bound. + /// + /// Main only reaches its bound when stale entries or concurrent overshoot + /// fill it, so evicting a live head here regardless of frequency is rare. + fn push_main(&self, node: Arc>, evicted: &mut Evicted) { + while self.main.len() >= self.queue_bound { + let Some(head) = self.main.pop() else { + break; + }; + if !head.is_removed() { + self.evict_node(&head, None, evicted); + } + } + self.main.push(node); + } + + /// Removes `node` from the index if it is still the resident entry for + /// its key and records the key in `evicted`. With `ghost`, also remembers + /// the key for readmission into main. + fn evict_node( + &self, + node: &Arc>, + ghost: Option<&RandomState>, + evicted: &mut Evicted, + ) -> bool { + let key = node.key; + if !self.index.remove_if_sync(&key, |resident| Arc::ptr_eq(resident, node)) { + return false; + } + node.removed.store(true, Ordering::Release); + self.len.fetch_sub(1, Ordering::AcqRel); + if let Some(hasher) = ghost { + let hash = hasher.hash_one(key); + self.ghost_slot(hash).store(fingerprint(hash), Ordering::Relaxed); + } + evicted.push(key); + true + } + + /// Consumes the ghost fingerprint for `hash`, returning whether it was set. + fn take_ghost(&self, hash: u64) -> bool { + let slot = self.ghost_slot(hash); + let fingerprint = fingerprint(hash); + slot.load(Ordering::Relaxed) == fingerprint + && slot + .compare_exchange(fingerprint, 0, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + fn ghost_slot(&self, hash: u64) -> &AtomicU64 { + let low = usize::try_from(hash & u64::from(u32::MAX)).unwrap_or_default(); + &self.ghost[low & (self.ghost.len() - 1)] + } + + #[cfg(test)] + fn len(&self) -> usize { + self.len.load(Ordering::Acquire).max(0).unsigned_abs() + } +} + +/// Non-zero ghost fingerprint of a key hash. +fn fingerprint(hash: u64) -> u64 { + hash | 1 +} diff --git a/keeper/src/tests/caches.rs b/keeper/src/tests/caches.rs index e922f3f9..92a9ee54 100644 --- a/keeper/src/tests/caches.rs +++ b/keeper/src/tests/caches.rs @@ -135,7 +135,7 @@ async fn account_lease_coordinates_recency_and_waiters() { drop(lease); drop(waiter.await); - assert_eq!(cache.lru.get_sync(&pk).is_some(), tracked, "{mode:?}"); + assert_eq!(cache.recency.contains(&pk), tracked, "{mode:?}"); } let cache = Arc::new(AccountCache::new(256)); @@ -143,7 +143,7 @@ async fn account_lease_coordinates_recency_and_waiters() { let mut lease = cache.lock(pk).await; lease.materialized(AccountMode::Mirrored); drop(lease); - assert!(cache.lru.get_sync(&pk).is_some(), "mirror is admitted"); + assert!(cache.recency.contains(&pk), "mirror is admitted"); let mut eviction = cache.lock(pk).await; eviction.observed = Some((AccountMode::Mirrored, 0)); assert!( @@ -152,17 +152,14 @@ async fn account_lease_coordinates_recency_and_waiters() { ); drop(eviction); assert!( - cache.lru.get_sync(&pk).is_some(), + cache.recency.contains(&pk), "rejecting stale eviction leaves recency unchanged" ); let mut lease = cache.lock(pk).await; lease.materialized(AccountMode::Delegated); drop(lease); - assert!( - cache.lru.get_sync(&pk).is_none(), - "delegation removes recency" - ); + assert!(!cache.recency.contains(&pk), "delegation removes recency"); let mut lease = cache.lock(pk).await; lease.observed = Some((AccountMode::Mirrored, 0)); @@ -173,8 +170,5 @@ async fn account_lease_coordinates_recency_and_waiters() { lease.materialized(AccountMode::Mirrored); lease.deleted(); drop(lease); - assert!( - cache.lru.get_sync(&pk).is_none(), - "deletion removes recency" - ); + assert!(!cache.recency.contains(&pk), "deletion removes recency"); } diff --git a/keeper/src/tests/mod.rs b/keeper/src/tests/mod.rs index 11a3dfae..da3993f3 100644 --- a/keeper/src/tests/mod.rs +++ b/keeper/src/tests/mod.rs @@ -6,6 +6,7 @@ mod caches; mod recovery; +mod s3fifo; mod subscriptions; use crate::testkit::{TestKeeper, signed_tx}; diff --git a/keeper/src/tests/s3fifo.rs b/keeper/src/tests/s3fifo.rs new file mode 100644 index 00000000..502d3142 --- /dev/null +++ b/keeper/src/tests/s3fifo.rs @@ -0,0 +1,226 @@ +//! S3-FIFO residency policy: capacity bound, frequency-aware retention, ghost +//! readmission, stale-entry bounds, and concurrent accounting. + +use std::{collections::HashSet, sync::Arc}; + +use solana_account::AccountMode; +use solana_pubkey::Pubkey; + +use crate::{cache::AccountCache, s3fifo::S3Fifo}; + +/// Admits `keys` in order and returns every key reported as evicted. +fn admit_all(cache: &S3Fifo, keys: impl IntoIterator) -> Vec { + keys.into_iter().flat_map(|key| cache.admit(key).evicted).collect() +} + +/// Proves residency never exceeds capacity and every displaced key is +/// reported exactly once and is no longer resident. +#[test] +fn s3fifo_bounds_residency_and_reports_every_eviction() { + let cache = S3Fifo::new(16); + let evicted = admit_all(&cache, 0..64); + + assert_eq!(cache.len(), 16); + assert_eq!(evicted.len(), 48); + let evicted: HashSet<_> = evicted.into_iter().collect(); + assert_eq!(evicted.len(), 48, "each key is reported once"); + for key in 0..64 { + assert_ne!( + cache.contains(&key), + evicted.contains(&key), + "key {key} is either resident or reported evicted" + ); + } +} + +/// Proves keys hit while resident survive a scan of one-off keys many times +/// larger than the cache, which an LRU would flush entirely. +#[test] +fn s3fifo_retains_hot_keys_through_a_scan() { + let cache = S3Fifo::new(100); + let hot = 0..50; + admit_all(&cache, hot.clone()); + for key in hot.clone() { + assert!(cache.touch(&key)); + assert!( + !cache.admit(key).inserted, + "admitting a resident key is a hit" + ); + } + + let evicted = admit_all(&cache, 1_000..11_000); + + assert!( + hot.clone().all(|key| cache.contains(&key)), + "hot keys stay resident" + ); + assert!( + evicted.iter().all(|key| *key >= 1_000), + "only scan keys are evicted" + ); + assert_eq!(cache.len(), 100); +} + +/// Proves a key evicted from probation and admitted again soon after skips +/// probation, so it outlives later one-off keys. +#[test] +fn s3fifo_readmits_recent_ghosts_into_main() { + let cache = S3Fifo::new(10); + admit_all(&cache, 0..=10); + assert!( + !cache.contains(&0), + "the coldest probationary key is evicted" + ); + + assert!(cache.admit(0).inserted); + // Without ghost readmission, key 0 would reach the probation head within + // ten admissions and be evicted. + admit_all(&cache, 100..120); + + assert!(cache.contains(&0), "a ghost readmission lands in main"); +} + +/// Proves touching an absent key neither admits it nor disturbs residency. +#[test] +fn s3fifo_touch_ignores_absent_keys() { + let cache = S3Fifo::new(4); + admit_all(&cache, 0..4); + + assert!(!cache.touch(&99)); + assert!(!cache.contains(&99)); + assert_eq!(cache.len(), 4); +} + +/// Proves stale queue entries left by removals stay within the queue bound and +/// never count against residency. +#[test] +fn s3fifo_removal_churn_keeps_queues_bounded() { + let cache = S3Fifo::new(8); + for key in 0..1_000 { + assert!(cache.admit(key % 3).inserted); + assert!(cache.remove(&(key % 3))); + assert!(!cache.remove(&(key % 3)), "a second removal finds nothing"); + let (small, main) = cache.queued(); + assert!(small + main <= 2 * cache.queue_bound()); + } + assert_eq!(cache.len(), 0); + + let evicted = admit_all(&cache, 100..108); + assert!( + evicted.is_empty(), + "stale entries do not displace live keys" + ); + assert_eq!(cache.len(), 8); +} + +/// Proves an admission over capacity still evicts a live key when more stale +/// entries than the probe budget sit ahead of it in the queue. +#[test] +fn s3fifo_stale_backlog_does_not_block_eviction() { + const CAPACITY: usize = 1_024; + let cache = S3Fifo::with_shard_count(CAPACITY, 1); + let capacity = u64::try_from(CAPACITY).unwrap(); + for key in 0..capacity { + assert!(cache.admit(key).inserted); + assert!(cache.remove(&key)); + } + admit_all(&cache, capacity..2 * capacity); + assert_eq!(cache.len(), CAPACITY); + + let admission = cache.admit(2 * capacity); + + assert_eq!(admission.evicted.len(), 1, "one live key is displaced"); + assert_eq!(cache.len(), CAPACITY, "residency stays within capacity"); +} + +/// Proves concurrent admissions, hits, and removals keep the resident count, +/// the indexes, and the reported evictions consistent and within bounds, for +/// both a single shard and a sharded cache. +#[test] +fn s3fifo_concurrent_accounting_is_consistent() { + const THREADS: u64 = 8; + for capacity in [64, 4_096] { + let cache = S3Fifo::new(capacity); + if capacity > 256 { + assert!(cache.shard_count() > 1, "large caches are sharded"); + } + let keys = u64::try_from(capacity * 8).unwrap(); + + let (inserted, evicted, removed) = std::thread::scope(|scope| { + let workers: Vec<_> = (0..THREADS) + .map(|thread| { + let cache = &cache; + scope.spawn(move || churn(cache, thread, keys)) + }) + .collect(); + workers + .into_iter() + .map(|worker| worker.join().unwrap()) + .fold((0, 0, 0), |(i, e, r), (wi, we, wr)| { + (i + wi, e + we, r + wr) + }) + }); + + assert_eq!(cache.len(), cache.indexed(), "counters match the indexes"); + assert_eq!( + inserted - evicted - removed, + cache.len(), + "every exit is reported" + ); + assert!(cache.len() <= capacity); + let (small, main) = cache.queued(); + let slack = usize::try_from(THREADS).unwrap() * cache.shard_count(); + assert!(small <= cache.queue_bound() + slack); + assert!(main <= cache.queue_bound() + slack); + } +} + +/// Runs a random mix of admissions, hits, and removals over `keys` keys and +/// returns the observed (inserted, evicted, removed) counts. +fn churn(cache: &S3Fifo, thread: u64, keys: u64) -> (usize, usize, usize) { + const OPS: u64 = 20_000; + let (mut inserted, mut evicted, mut removed) = (0usize, 0usize, 0usize); + let mut state = 0x9E37_79B9_7F4A_7C15 ^ (thread + 1); + for _ in 0..OPS { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + let key = state % keys; + match state % 20 { + 0 => removed += usize::from(cache.remove(&key)), + 1..=5 => { + cache.touch(&key); + } + _ => { + let admission = cache.admit(key); + inserted += usize::from(admission.inserted); + evicted += admission.evicted.len(); + } + } + } + (inserted, evicted, removed) +} + +/// Proves materializing past capacity publishes the displaced account to the +/// eviction subscriber, and a later eviction check for it still applies. +#[tokio::test] +async fn account_cache_publishes_displaced_accounts() { + let cache = Arc::new(AccountCache::new(2)); + let mut evictions = cache.evictions.subscribe().unwrap(); + let keys: Vec<_> = (0..3).map(|_| Pubkey::new_unique()).collect(); + for key in &keys { + let mut lease = cache.lock(*key).await; + lease.observed = Some((AccountMode::Mirrored, 0)); + lease.materialized(AccountMode::Mirrored); + } + + let evicted = evictions.try_recv().expect("one account is displaced"); + assert_eq!(evicted, keys[0]); + assert!( + evictions.try_recv().is_err(), + "only one account is displaced" + ); + let mut lease = cache.lock(evicted).await; + lease.observed = Some((AccountMode::Mirrored, 0)); + assert!(lease.cached_eviction_applies()); +} diff --git a/nucleus/src/config.rs b/nucleus/src/config.rs index 45da2d6f..7a5c4e75 100644 --- a/nucleus/src/config.rs +++ b/nucleus/src/config.rs @@ -42,11 +42,11 @@ impl>> From for Authority { pub struct AccountsDBParams { /// Accounts database root directory. pub directory: PathBuf, - /// Requested maximum number of resolved account pubkeys retained for - /// recency tracking and eviction notifications. + /// Maximum number of resolved non-authoritative account pubkeys retained + /// for recency tracking and eviction notifications (minimum one). /// - /// The cache uses at least 256 slots, rounds larger capacities up to a - /// power of two, and may evict earlier under bucket pressure. + /// Eviction follows S3-FIFO: accounts hit while resident are retained over + /// accounts seen once. Concurrent admissions may briefly exceed the bound. pub lru_capacity: usize, }