From 3a1c144d8890df05a195965a4fa71d67feac7cfe Mon Sep 17 00:00:00 2001 From: nym21 Date: Wed, 12 Aug 2026 11:00:37 +0200 Subject: [PATCH] global: massive columnar rework part 7 --- .../src/distribution/all_chain_sources.rs | 34 +++-- .../metrics/additive/aggregate/fiat.rs | 15 +-- .../metrics/aggregate/cumulative_fiat.rs | 17 +-- .../additive/without_amount_or_type.rs | 6 +- .../metrics/profitability/column_id.rs | 4 +- .../src/frameworks/coinflow/import.rs | 6 +- .../frameworks/cointime/aggregate/import.rs | 7 +- .../brk_computer/src/internal/cache_budget.rs | 118 +++++++++--------- .../per_block/computed/resolutions.rs | 4 +- .../internal/per_block/ratio/lazy_price.rs | 6 +- .../value/columnar_cumulative_rolling.rs | 4 +- crates/brk_computer/src/lib.rs | 2 +- 12 files changed, 124 insertions(+), 99 deletions(-) diff --git a/crates/brk_computer/src/distribution/all_chain_sources.rs b/crates/brk_computer/src/distribution/all_chain_sources.rs index 7fc12643e..63d666664 100644 --- a/crates/brk_computer/src/distribution/all_chain_sources.rs +++ b/crates/brk_computer/src/distribution/all_chain_sources.rs @@ -3,7 +3,7 @@ use vecdb::{ BinaryTransform, CachedBoxedVec, ReadableCloneableVec, ReadableVec, TypedVec, VecValue, }; -use crate::internal::{LazyIndexedVec, SatsToCents}; +use crate::internal::{CACHE_BUDGET, LazyIndexedVec, SatsToCents}; /// Shared handles to the pinned all-chain inputs. /// @@ -31,8 +31,7 @@ impl AllChainSources { } } - /// Lazily combines one ordinary source with the pinned all-supply cache. - pub fn with_supply( + fn supply_source( &self, name: &str, version: Version, @@ -52,8 +51,24 @@ impl AllChainSources { ) } - /// Lazily combines one ordinary source with market cap derived from the - /// pinned all-supply and spot-price caches. + /// Combines one ordinary source with pinned all supply, caching the result + /// when it becomes hot. + pub fn with_supply( + &self, + name: &str, + version: Version, + source: &(impl ReadableCloneableVec + 'static), + compute: impl Fn(Height, S, Sats) -> T + Send + Sync + 'static, + ) -> impl TypedVec + ReadableVec + Clone + 'static + where + S: VecValue, + T: VecValue, + { + CACHE_BUDGET.wrap(self.supply_source(name, version, source, compute)) + } + + /// Combines one ordinary source with market cap, caching only the final + /// result when it becomes hot. pub fn with_market_cap( &self, name: &str, @@ -65,14 +80,14 @@ impl AllChainSources { S: VecValue, T: VecValue, { - let with_supply = self.with_supply( + let with_supply = self.supply_source( &format!("{name}_with_supply"), Version::ZERO, source, |_, source, supply| WithSupply { source, supply }, ); - LazyIndexedVec::new( + let source = LazyIndexedVec::new( name, version, with_supply.read_only_boxed_clone(), @@ -84,7 +99,8 @@ impl AllChainSources { SatsToCents::apply(with_supply.supply, price), ) }, - ) + ); + CACHE_BUDGET.wrap(source) } } @@ -92,7 +108,7 @@ impl AllChainSources { mod tests { use vecdb::{ AnyStoredVec, CachedReadableVec, CachedVec, Database, EagerVec, ImportableVec, PcoVec, - ReadOnlyClone, WritableVec, + ReadOnlyClone, ReadableVec, WritableVec, }; use super::*; diff --git a/crates/brk_computer/src/distribution/metrics/additive/aggregate/fiat.rs b/crates/brk_computer/src/distribution/metrics/additive/aggregate/fiat.rs index a44459fc6..1ad506b89 100644 --- a/crates/brk_computer/src/distribution/metrics/additive/aggregate/fiat.rs +++ b/crates/brk_computer/src/distribution/metrics/additive/aggregate/fiat.rs @@ -13,7 +13,7 @@ use vecdb::{ use crate::{ indexes, - internal::{ColumnarPerBlock, FiatType, LazyFiatPerBlock, cache_wrap}, + internal::{CACHE_BUDGET, ColumnarPerBlock, FiatType, LazyFiatPerBlock}, }; #[derive(Deref, DerefMut, Traversable)] @@ -44,12 +44,13 @@ impl AdditiveAggregateFiatPerBlock { metric, ); let cents = match aggregate { - UTXOAggregateId::All => cache_wrap(source.sum_columns( - &format!("{name}_cents"), - version, - TermId::ALL.iter().copied(), - )) - .read_only_boxed_clone(), + UTXOAggregateId::All => CACHE_BUDGET + .wrap(source.sum_columns( + &format!("{name}_cents"), + version, + TermId::ALL.iter().copied(), + )) + .read_only_boxed_clone(), UTXOAggregateId::Sth => source .column(&format!("{name}_cents"), version, TermId::Short) .read_only_boxed_clone(), diff --git a/crates/brk_computer/src/distribution/metrics/aggregate/cumulative_fiat.rs b/crates/brk_computer/src/distribution/metrics/aggregate/cumulative_fiat.rs index a50f933ba..4b5753857 100644 --- a/crates/brk_computer/src/distribution/metrics/aggregate/cumulative_fiat.rs +++ b/crates/brk_computer/src/distribution/metrics/aggregate/cumulative_fiat.rs @@ -14,8 +14,8 @@ use vecdb::{ use crate::{ indexes, internal::{ - CachedWindowStartVec, ColumnarPerBlockCumulativeRolling, FiatType, - LazyFiatPerBlockCumulativeWithSums, Windows, cache_wrap, + CACHE_BUDGET, CachedWindowStartVec, ColumnarPerBlockCumulativeRolling, FiatType, + LazyFiatPerBlockCumulativeWithSums, Windows, }, }; @@ -53,12 +53,13 @@ impl AdditiveAggregateFiatPerBlockCumulativeWithSums { metric, ); let cumulative = match id { - UTXOAggregateId::All => cache_wrap(source.sum_columns( - &format!("{name}_cumulative_cents"), - version, - TermId::ALL.iter().copied(), - )) - .read_only_boxed_clone(), + UTXOAggregateId::All => CACHE_BUDGET + .wrap(source.sum_columns( + &format!("{name}_cumulative_cents"), + version, + TermId::ALL.iter().copied(), + )) + .read_only_boxed_clone(), UTXOAggregateId::Sth => source .column(&format!("{name}_cumulative_cents"), version, TermId::Short) .read_only_boxed_clone(), diff --git a/crates/brk_computer/src/distribution/metrics/columnar/additive/without_amount_or_type.rs b/crates/brk_computer/src/distribution/metrics/columnar/additive/without_amount_or_type.rs index e977babc0..e2e560804 100644 --- a/crates/brk_computer/src/distribution/metrics/columnar/additive/without_amount_or_type.rs +++ b/crates/brk_computer/src/distribution/metrics/columnar/additive/without_amount_or_type.rs @@ -15,7 +15,7 @@ use vecdb::{ }; use super::super::UTXORows; -use crate::internal::cache_wrap; +use crate::internal::CACHE_BUDGET; #[derive(Traversable)] pub struct UTXOColumnarMetricWithoutAmountOrType @@ -201,7 +201,9 @@ where where C: ColumnId, { - cache_wrap(source.sum_columns(name, version, columns)).read_only_boxed_clone() + CACHE_BUDGET + .wrap(source.sum_columns(name, version, columns)) + .read_only_boxed_clone() } pub fn min_len(&self) -> usize { diff --git a/crates/brk_computer/src/distribution/metrics/profitability/column_id.rs b/crates/brk_computer/src/distribution/metrics/profitability/column_id.rs index e6cf4614d..36decde48 100644 --- a/crates/brk_computer/src/distribution/metrics/profitability/column_id.rs +++ b/crates/brk_computer/src/distribution/metrics/profitability/column_id.rs @@ -7,7 +7,7 @@ use vecdb::{ ReadableColumnarVec, VecValue, }; -use crate::internal::cache_wrap; +use crate::internal::CACHE_BUDGET; const RANGE_COUNT: usize = ProfitabilityRangeId::ALL.len(); const COLUMN_COUNT: usize = TermId::ALL.len() * RANGE_COUNT; @@ -79,7 +79,7 @@ impl TermProfitabilityRangeId { }), ); if aggregate == UTXOAggregateId::All { - cache_wrap(source).read_only_boxed_clone() + CACHE_BUDGET.wrap(source).read_only_boxed_clone() } else { source.read_only_boxed_clone() } diff --git a/crates/brk_computer/src/frameworks/coinflow/import.rs b/crates/brk_computer/src/frameworks/coinflow/import.rs index d254cdf5b..3cb3a1e29 100644 --- a/crates/brk_computer/src/frameworks/coinflow/import.rs +++ b/crates/brk_computer/src/frameworks/coinflow/import.rs @@ -16,9 +16,8 @@ use super::{ use crate::{ indexes, internal::{ - ColumnarPerBlock, Identity, LazyColumnPerBlock, LazyColumnSpotValuePerBlock, + CACHE_BUDGET, ColumnarPerBlock, Identity, LazyColumnPerBlock, LazyColumnSpotValuePerBlock, LazyFiatPerBlock, LazyPerBlock, LazyPriceWithRatioPerBlock, LazySpotValuePerBlock, - cache_wrap, }, }; @@ -106,7 +105,8 @@ impl AggregateSources { { match aggregate.term() { Some(term) => source.column(name, version, term).read_only_boxed_clone(), - None => cache_wrap(source.sum_columns(name, version, TermId::ALL.iter().copied())) + None => CACHE_BUDGET + .wrap(source.sum_columns(name, version, TermId::ALL.iter().copied())) .read_only_boxed_clone(), } } diff --git a/crates/brk_computer/src/frameworks/cointime/aggregate/import.rs b/crates/brk_computer/src/frameworks/cointime/aggregate/import.rs index e71f804e0..e4169ce70 100644 --- a/crates/brk_computer/src/frameworks/cointime/aggregate/import.rs +++ b/crates/brk_computer/src/frameworks/cointime/aggregate/import.rs @@ -12,8 +12,8 @@ use super::{AwakeVecs, CohortVecs, DormantVecs, Sources, Vecs}; use crate::{ indexes, internal::{ - Identity, LazyFiatPerBlock, LazyPerBlock, LazyPriceWithRatioPerBlock, - LazySpotValuePerBlock, PerBlock, cache_wrap, + CACHE_BUDGET, Identity, LazyFiatPerBlock, LazyPerBlock, LazyPriceWithRatioPerBlock, + LazySpotValuePerBlock, PerBlock, }, }; @@ -59,7 +59,8 @@ impl Sources { { match aggregate.term() { Some(term) => source.column(name, version, term).read_only_boxed_clone(), - None => cache_wrap(source.sum_columns(name, version, TermId::ALL.iter().copied())) + None => CACHE_BUDGET + .wrap(source.sum_columns(name, version, TermId::ALL.iter().copied())) .read_only_boxed_clone(), } } diff --git a/crates/brk_computer/src/internal/cache_budget.rs b/crates/brk_computer/src/internal/cache_budget.rs index 37d6a8059..d7942b975 100644 --- a/crates/brk_computer/src/internal/cache_budget.rs +++ b/crates/brk_computer/src/internal/cache_budget.rs @@ -9,33 +9,66 @@ use vecdb::{CachedVec, CachedVecBudget, ReadableVec, TypedVec}; const MAX_CACHED: usize = 256; const MIN_ACCESSES: u64 = 2; -struct LruBudget { +pub struct CacheBudget { remaining: AtomicUsize, + caches: Mutex>, } -impl LruBudget { +impl CacheBudget { + const fn new() -> Self { + Self { + remaining: AtomicUsize::new(MAX_CACHED), + caches: Mutex::new(Vec::new()), + } + } + fn try_decrement(&self) -> bool { self.remaining .fetch_update(Relaxed, Relaxed, |n| if n > 0 { Some(n - 1) } else { None }) .is_ok() } -} -impl CachedVecBudget for LruBudget { - fn try_reserve(&self, access_count: u64) -> bool { - if access_count < MIN_ACCESSES { - return false; - } - if self.try_decrement() { - return true; - } - // Only evict if we're more popular than the least popular cached entry. - if evict_less_popular_than(access_count) { - self.try_decrement() + fn evict_less_popular_than(&self, threshold: u64) -> bool { + let caches = self.caches.lock(); + if let Some((index, _)) = caches + .iter() + .enumerate() + .filter(|(_, entry)| { + let count = entry.access_count.load(Relaxed); + count >= MIN_ACCESSES && count < threshold + }) + .min_by_key(|(_, entry)| entry.access_count.load(Relaxed)) + { + (caches[index].clear)(); + self.remaining.fetch_add(1, Relaxed); + true } else { false } } + + /// Wraps a source vec in this budget and registers it for eviction. + pub fn wrap(&'static self, source: V) -> CachedVec + where + V: TypedVec + ReadableVec + Clone + 'static, + { + let access_count = Arc::new(AtomicU64::new(0)); + let cached = CachedVec::wrap_budgeted(source, self, access_count.clone()); + let clone = cached.clone(); + self.caches.lock().push(CacheEntry { + access_count, + clear: Box::new(move || clone.clear()), + }); + cached + } + + /// Clears every registered vec and resets the budget. + pub fn clear(&self) { + for entry in self.caches.lock().iter() { + (entry.clear)(); + } + self.remaining.store(MAX_CACHED, Relaxed); + } } struct CacheEntry { @@ -43,49 +76,20 @@ struct CacheEntry { clear: Box, } -static BUDGET: LruBudget = LruBudget { - remaining: AtomicUsize::new(MAX_CACHED), -}; -static CACHES: Mutex> = Mutex::new(Vec::new()); - -fn evict_less_popular_than(threshold: u64) -> bool { - let caches = CACHES.lock(); - if let Some((idx, _)) = caches - .iter() - .enumerate() - .filter(|(_, e)| { - let c = e.access_count.load(Relaxed); - c >= MIN_ACCESSES && c < threshold - }) - .min_by_key(|(_, e)| e.access_count.load(Relaxed)) - { - (caches[idx].clear)(); - BUDGET.remaining.fetch_add(1, Relaxed); - true - } else { - false +impl CachedVecBudget for CacheBudget { + fn try_reserve(&self, access_count: u64) -> bool { + if access_count < MIN_ACCESSES { + return false; + } + if self.try_decrement() { + return true; + } + if self.evict_less_popular_than(access_count) { + self.try_decrement() + } else { + false + } } } -/// Wraps a source vec in a budgeted [`CachedVec`] and registers it for eviction. -pub fn cache_wrap(source: V) -> CachedVec -where - V: TypedVec + ReadableVec + Clone + 'static, -{ - let access_count = Arc::new(AtomicU64::new(0)); - let cached = CachedVec::wrap_budgeted(source, &BUDGET, access_count.clone()); - let clone = cached.clone(); - CACHES.lock().push(CacheEntry { - access_count, - clear: Box::new(move || clone.clear()), - }); - cached -} - -/// Clears all cached vecs and resets the budget. -pub fn cache_clear_all() { - for entry in CACHES.lock().iter() { - (entry.clear)(); - } - BUDGET.remaining.store(MAX_CACHED, Relaxed); -} +pub static CACHE_BUDGET: CacheBudget = CacheBudget::new(); diff --git a/crates/brk_computer/src/internal/per_block/computed/resolutions.rs b/crates/brk_computer/src/internal/per_block/computed/resolutions.rs index 4a87ebd55..21c19abed 100644 --- a/crates/brk_computer/src/internal/per_block/computed/resolutions.rs +++ b/crates/brk_computer/src/internal/per_block/computed/resolutions.rs @@ -14,7 +14,7 @@ use vecdb::{ use crate::{ indexes, - internal::{PerResolution, cache_wrap}, + internal::{CACHE_BUDGET, PerResolution}, }; /// Aggregation strategy for epoch-based indices (Halving, Epoch). @@ -127,7 +127,7 @@ where where V: TypedVec + ReadableVec + Clone + 'static, { - let cached = cache_wrap(height_source); + let cached = CACHE_BUDGET.wrap(height_source); let height_source = cached.read_only_boxed_clone(); Self::from_boxed_height_source(name, height_source, version, indexes) } diff --git a/crates/brk_computer/src/internal/per_block/ratio/lazy_price.rs b/crates/brk_computer/src/internal/per_block/ratio/lazy_price.rs index bb2f19cca..caef0a63d 100644 --- a/crates/brk_computer/src/internal/per_block/ratio/lazy_price.rs +++ b/crates/brk_computer/src/internal/per_block/ratio/lazy_price.rs @@ -4,7 +4,7 @@ use vecdb::{CachedBoxedVec, ReadableBoxedVec, ReadableCloneableVec, ReadableVec, use crate::{ indexes, - internal::{Identity, LazyIndexedVec, LazyPerBlock, LazyRatioPerBlock, Price}, + internal::{CACHE_BUDGET, Identity, LazyIndexedVec, LazyPerBlock, LazyRatioPerBlock, Price}, }; use super::price::price_ratio; @@ -34,13 +34,13 @@ impl LazyPriceWithRatioPerBlock { ); let price = Price::from_lazy_cents_source::, Cents>(name, version, &source); let ratio_version = version + Version::new(4); - let ppm_source = LazyIndexedVec::new( + let ppm_source = CACHE_BUDGET.wrap(LazyIndexedVec::new( &format!("{name}_ratio_ppm_source"), ratio_version, price.cents.height.read_only_boxed_clone(), spot_price.clone(), |_, price, spot| price_ratio(spot, price), - ); + )); let ratio = LazyRatioPerBlock::from_uncached_height_source( &format!("{name}_ratio"), ratio_version, diff --git a/crates/brk_computer/src/internal/per_block/value/columnar_cumulative_rolling.rs b/crates/brk_computer/src/internal/per_block/value/columnar_cumulative_rolling.rs index 86ddf2101..0d772740b 100644 --- a/crates/brk_computer/src/internal/per_block/value/columnar_cumulative_rolling.rs +++ b/crates/brk_computer/src/internal/per_block/value/columnar_cumulative_rolling.rs @@ -9,7 +9,7 @@ use vecdb::{ }; use crate::internal::{ - ColumnarPerBlockCumulativeRolling, StoredU64ToCents, StoredU64ToSats, cache_wrap, + CACHE_BUDGET, ColumnarPerBlockCumulativeRolling, StoredU64ToCents, StoredU64ToSats, }; #[derive(Deref, DerefMut, Traversable)] @@ -163,7 +163,7 @@ where }; let source = LazyVec::transformed::(name, version, raw); if budgeted { - cache_wrap(source).read_only_boxed_clone() + CACHE_BUDGET.wrap(source).read_only_boxed_clone() } else { source.read_only_boxed_clone() } diff --git a/crates/brk_computer/src/lib.rs b/crates/brk_computer/src/lib.rs index ee71225ff..30d237425 100644 --- a/crates/brk_computer/src/lib.rs +++ b/crates/brk_computer/src/lib.rs @@ -331,7 +331,7 @@ impl Computer { } pub fn compute(&mut self, indexer: &mut Indexer, exit: &Exit) -> Result<()> { - internal::cache_clear_all(); + internal::CACHE_BUDGET.clear(); let compute_start = Instant::now();