global: massive columnar rework part 7

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<S, T>(
+    fn supply_source<S, T>(
         &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<S, T>(
+        &self,
+        name: &str,
+        version: Version,
+        source: &(impl ReadableCloneableVec<Height, S> + 'static),
+        compute: impl Fn(Height, S, Sats) -> T + Send + Sync + 'static,
+    ) -> impl TypedVec<I = Height, T = T> + ReadableVec<Height, T> + 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<S, T>(
         &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::*;

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<C: FiatType> AdditiveAggregateFiatPerBlock<C> {
                         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(),

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<C: FiatType> AdditiveAggregateFiatPerBlockCumulativeWithSums<C> {
                         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(),

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<T, M: StorageMode = Rw>
@@ -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 {

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()
         }

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(),
         }
     }

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(),
         }
     }

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<Vec<CacheEntry>>,
 }
 
-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<V>(&'static self, source: V) -> CachedVec<V>
+    where
+        V: TypedVec + ReadableVec<V::I, V::T> + 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<dyn Fn() + Send + Sync>,
 }
 
-static BUDGET: LruBudget = LruBudget {
-    remaining: AtomicUsize::new(MAX_CACHED),
-};
-static CACHES: Mutex<Vec<CacheEntry>> = 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<V>(source: V) -> CachedVec<V>
-where
-    V: TypedVec + ReadableVec<V::I, V::T> + 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();

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<I = Height, T = T> + ReadableVec<Height, T> + 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)
     }

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::<Identity<Cents>, 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,

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::<F>(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()
         }

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();