forest/db/
blockstore_with_read_cache.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use crate::prelude::*;
5use crate::utils::cache::SizeTrackingCache;
6use std::sync::atomic::{self, AtomicUsize};
7
8#[auto_impl::auto_impl(&, Arc)]
9pub trait BlockstoreReadCache {
10    fn get(&self, k: &Cid) -> Option<Vec<u8>>;
11
12    fn put(&self, k: Cid, block: Vec<u8>);
13}
14
15pub type DefaultBlockstoreReadCache = SizeTrackingCache<CidWrapper, Vec<u8>>;
16
17impl BlockstoreReadCache for SizeTrackingCache<CidWrapper, Vec<u8>> {
18    fn get(&self, k: &Cid) -> Option<Vec<u8>> {
19        self.deref().get(k)
20    }
21
22    fn put(&self, k: Cid, block: Vec<u8>) {
23        self.insert(k.into(), block);
24    }
25}
26
27pub trait BlockstoreReadCacheStats {
28    fn hit(&self) -> usize;
29
30    fn track_hit(&self);
31
32    fn miss(&self) -> usize;
33
34    fn track_miss(&self);
35}
36
37#[derive(Debug, Default)]
38pub struct DefaultBlockstoreReadCacheStats {
39    hit: crossbeam_utils::CachePadded<AtomicUsize>,
40    miss: crossbeam_utils::CachePadded<AtomicUsize>,
41}
42
43impl BlockstoreReadCacheStats for DefaultBlockstoreReadCacheStats {
44    fn hit(&self) -> usize {
45        self.hit.load(atomic::Ordering::Relaxed)
46    }
47
48    fn track_hit(&self) {
49        self.hit.fetch_add(1, atomic::Ordering::Relaxed);
50    }
51
52    fn miss(&self) -> usize {
53        self.miss.load(atomic::Ordering::Relaxed)
54    }
55
56    fn track_miss(&self) {
57        self.miss.fetch_add(1, atomic::Ordering::Relaxed);
58    }
59}
60
61#[derive(derive_more::Constructor)]
62pub struct BlockstoreWithReadCache<
63    DB: Blockstore,
64    CACHE: BlockstoreReadCache,
65    STATS: BlockstoreReadCacheStats,
66> {
67    inner: DB,
68    cache: CACHE,
69    stats: Option<STATS>,
70}
71
72impl<DB: Blockstore, CACHE: BlockstoreReadCache, STATS: BlockstoreReadCacheStats>
73    BlockstoreWithReadCache<DB, CACHE, STATS>
74{
75    pub fn stats(&self) -> Option<&STATS> {
76        self.stats.as_ref()
77    }
78}
79
80impl<DB: Blockstore, CACHE: BlockstoreReadCache, STATS: BlockstoreReadCacheStats> Blockstore
81    for BlockstoreWithReadCache<DB, CACHE, STATS>
82{
83    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
84        if let Some(cached) = self.cache.get(k) {
85            self.stats.as_ref().map(BlockstoreReadCacheStats::track_hit);
86            Ok(Some(cached))
87        } else {
88            let block = self.inner.get(k)?;
89            self.stats
90                .as_ref()
91                .map(BlockstoreReadCacheStats::track_miss);
92            if let Some(block) = &block {
93                self.cache.put(*k, block.clone());
94            }
95            Ok(block)
96        }
97    }
98
99    fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
100        self.inner.put_keyed(k, block)
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107    use crate::{db::MemoryDB, utils::rand::forest_rng};
108    use fvm_ipld_encoding::DAG_CBOR;
109    use multihash_codetable::Code::Blake2b256;
110    use multihash_codetable::MultihashDigest as _;
111    use rand::Rng as _;
112    use std::sync::Arc;
113
114    #[test]
115    fn test_blockstore_read_cache() {
116        const N_RECORDS: usize = 4;
117        const CACHE_SIZE: usize = 2;
118        let mem_db = Arc::new(MemoryDB::default());
119        let mut records = Vec::with_capacity(N_RECORDS);
120        for _ in 0..N_RECORDS {
121            let mut record = [0; 1024];
122            forest_rng().fill(&mut record);
123            let key = Cid::new_v1(DAG_CBOR, Blake2b256.digest(record.as_slice()));
124            mem_db.put_keyed(&key, &record).unwrap();
125            records.push((key, record));
126        }
127        let cache = Arc::new(DefaultBlockstoreReadCache::new_without_metrics_registry(
128            "test_blockstore_read_cache",
129            CACHE_SIZE.try_into().unwrap(),
130        ));
131        let db = BlockstoreWithReadCache::new(
132            mem_db.clone(),
133            cache.clone(),
134            Some(DefaultBlockstoreReadCacheStats::default()),
135        );
136
137        assert_eq!(cache.len(), 0);
138        assert_eq!(db.stats().unwrap().hit(), 0);
139        assert_eq!(db.stats().unwrap().miss(), 0);
140
141        for (i, (k, v)) in records.iter().enumerate() {
142            assert_eq!(&db.get(k).unwrap().unwrap(), v);
143
144            assert_eq!(cache.len(), CACHE_SIZE.min(i + 1));
145            assert_eq!(db.stats().unwrap().hit(), i);
146            assert_eq!(db.stats().unwrap().miss(), i + 1);
147
148            assert_eq!(&db.get(k).unwrap().unwrap(), v);
149
150            assert_eq!(cache.len(), CACHE_SIZE.min(i + 1));
151            assert_eq!(db.stats().unwrap().hit(), i + 1);
152            assert_eq!(db.stats().unwrap().miss(), i + 1);
153        }
154
155        let (k0, v0) = &records[0];
156
157        assert_eq!(&db.get(k0).unwrap().unwrap(), v0);
158
159        assert_eq!(cache.len(), CACHE_SIZE);
160        assert_eq!(db.stats().unwrap().hit(), 4);
161        assert_eq!(db.stats().unwrap().miss(), 5);
162
163        assert_eq!(&db.get(k0).unwrap().unwrap(), v0);
164
165        assert_eq!(cache.len(), CACHE_SIZE);
166        assert_eq!(db.stats().unwrap().hit(), 5);
167        assert_eq!(db.stats().unwrap().miss(), 5);
168    }
169}
forest/db/blockstore_with_read_cache.rs

forest/db/
blockstore_with_read_cache.rs
