solana_runtime/
status_cache.rs

1use {
2    log::*,
3    rand::{thread_rng, Rng},
4    serde::Serialize,
5    solana_accounts_db::ancestors::Ancestors,
6    solana_sdk::{
7        clock::{Slot, MAX_RECENT_BLOCKHASHES},
8        hash::Hash,
9    },
10    std::{
11        collections::{hash_map::Entry, HashMap, HashSet},
12        sync::{Arc, Mutex},
13    },
14};
15
16pub const MAX_CACHE_ENTRIES: usize = MAX_RECENT_BLOCKHASHES;
17const CACHED_KEY_SIZE: usize = 20;
18
19// Store forks in a single chunk of memory to avoid another lookup.
20pub type ForkStatus<T> = Vec<(Slot, T)>;
21type KeySlice = [u8; CACHED_KEY_SIZE];
22type KeyMap<T> = HashMap<KeySlice, ForkStatus<T>>;
23// Map of Hash and status
24pub type Status<T> = Arc<Mutex<HashMap<Hash, (usize, Vec<(KeySlice, T)>)>>>;
25// A Map of hash + the highest fork it's been observed on along with
26// the key offset and a Map of the key slice + Fork status for that key
27type KeyStatusMap<T> = HashMap<Hash, (Slot, usize, KeyMap<T>)>;
28
29// A map of keys recorded in each fork; used to serialize for snapshots easily.
30// Doesn't store a `SlotDelta` in it because the bool `root` is usually set much later
31type SlotDeltaMap<T> = HashMap<Slot, Status<T>>;
32
33// The statuses added during a slot, can be used to build on top of a status cache or to
34// construct a new one. Usually derived from a status cache's `SlotDeltaMap`
35pub type SlotDelta<T> = (Slot, bool, Status<T>);
36
37#[cfg_attr(feature = "frozen-abi", derive(AbiExample))]
38#[derive(Clone, Debug)]
39pub struct StatusCache<T: Serialize + Clone> {
40    cache: KeyStatusMap<T>,
41    roots: HashSet<Slot>,
42    /// all keys seen during a fork/slot
43    slot_deltas: SlotDeltaMap<T>,
44}
45
46impl<T: Serialize + Clone> Default for StatusCache<T> {
47    fn default() -> Self {
48        Self {
49            cache: HashMap::default(),
50            // 0 is always a root
51            roots: HashSet::from([0]),
52            slot_deltas: HashMap::default(),
53        }
54    }
55}
56
57impl<T: Serialize + Clone + PartialEq> PartialEq for StatusCache<T> {
58    fn eq(&self, other: &Self) -> bool {
59        self.roots == other.roots
60            && self
61                .cache
62                .iter()
63                .all(|(hash, (slot, key_index, hash_map))| {
64                    if let Some((other_slot, other_key_index, other_hash_map)) =
65                        other.cache.get(hash)
66                    {
67                        if slot == other_slot && key_index == other_key_index {
68                            return hash_map.iter().all(|(slice, fork_map)| {
69                                if let Some(other_fork_map) = other_hash_map.get(slice) {
70                                    // all this work just to compare the highest forks in the fork map
71                                    // per entry
72                                    return fork_map.last() == other_fork_map.last();
73                                }
74                                false
75                            });
76                        }
77                    }
78                    false
79                })
80    }
81}
82
83impl<T: Serialize + Clone> StatusCache<T> {
84    pub fn clear_slot_entries(&mut self, slot: Slot) {
85        let slot_deltas = self.slot_deltas.remove(&slot);
86        if let Some(slot_deltas) = slot_deltas {
87            let slot_deltas = slot_deltas.lock().unwrap();
88            for (blockhash, (_, key_list)) in slot_deltas.iter() {
89                // Any blockhash that exists in self.slot_deltas must also exist
90                // in self.cache, because in self.purge_roots(), when an entry
91                // (b, (max_slot, _, _)) is removed from self.cache, this implies
92                // all entries in self.slot_deltas < max_slot are also removed
93                if let Entry::Occupied(mut o_blockhash_entries) = self.cache.entry(*blockhash) {
94                    let (_, _, all_hash_maps) = o_blockhash_entries.get_mut();
95
96                    for (key_slice, _) in key_list {
97                        if let Entry::Occupied(mut o_key_list) = all_hash_maps.entry(*key_slice) {
98                            let key_list = o_key_list.get_mut();
99                            key_list.retain(|(updated_slot, _)| *updated_slot != slot);
100                            if key_list.is_empty() {
101                                o_key_list.remove_entry();
102                            }
103                        } else {
104                            panic!(
105                                "Map for key must exist if key exists in self.slot_deltas, slot: {slot}"
106                            )
107                        }
108                    }
109
110                    if all_hash_maps.is_empty() {
111                        o_blockhash_entries.remove_entry();
112                    }
113                } else {
114                    panic!("Blockhash must exist if it exists in self.slot_deltas, slot: {slot}")
115                }
116            }
117        }
118    }
119
120    /// Check if the key is in any of the forks in the ancestors set and
121    /// with a certain blockhash.
122    pub fn get_status<K: AsRef<[u8]>>(
123        &self,
124        key: K,
125        transaction_blockhash: &Hash,
126        ancestors: &Ancestors,
127    ) -> Option<(Slot, T)> {
128        let map = self.cache.get(transaction_blockhash)?;
129        let (_, index, keymap) = map;
130        let max_key_index = key.as_ref().len().saturating_sub(CACHED_KEY_SIZE + 1);
131        let index = (*index).min(max_key_index);
132        let key_slice: &[u8; CACHED_KEY_SIZE] =
133            arrayref::array_ref![key.as_ref(), index, CACHED_KEY_SIZE];
134        if let Some(stored_forks) = keymap.get(key_slice) {
135            let res = stored_forks
136                .iter()
137                .find(|(f, _)| ancestors.contains_key(f) || self.roots.contains(f))
138                .cloned();
139            if res.is_some() {
140                return res;
141            }
142        }
143        None
144    }
145
146    /// Search for a key with any blockhash
147    /// Prefer get_status for performance reasons, it doesn't need
148    /// to search all blockhashes.
149    pub fn get_status_any_blockhash<K: AsRef<[u8]>>(
150        &self,
151        key: K,
152        ancestors: &Ancestors,
153    ) -> Option<(Slot, T)> {
154        let keys: Vec<_> = self.cache.keys().copied().collect();
155
156        for blockhash in keys.iter() {
157            trace!("get_status_any_blockhash: trying {}", blockhash);
158            let status = self.get_status(&key, blockhash, ancestors);
159            if status.is_some() {
160                return status;
161            }
162        }
163        None
164    }
165
166    /// Add a known root fork.  Roots are always valid ancestors.
167    /// After MAX_CACHE_ENTRIES, roots are removed, and any old keys are cleared.
168    pub fn add_root(&mut self, fork: Slot) {
169        self.roots.insert(fork);
170        self.purge_roots();
171    }
172
173    pub fn roots(&self) -> &HashSet<Slot> {
174        &self.roots
175    }
176
177    /// Insert a new key for a specific slot.
178    pub fn insert<K: AsRef<[u8]>>(
179        &mut self,
180        transaction_blockhash: &Hash,
181        key: K,
182        slot: Slot,
183        res: T,
184    ) {
185        let max_key_index = key.as_ref().len().saturating_sub(CACHED_KEY_SIZE + 1);
186
187        // Get the cache entry for this blockhash.
188        let (max_slot, key_index, hash_map) =
189            self.cache.entry(*transaction_blockhash).or_insert_with(|| {
190                let key_index = thread_rng().gen_range(0..max_key_index + 1);
191                (slot, key_index, HashMap::new())
192            });
193
194        // Update the max slot observed to contain txs using this blockhash.
195        *max_slot = std::cmp::max(slot, *max_slot);
196
197        // Grab the key slice.
198        let key_index = (*key_index).min(max_key_index);
199        let mut key_slice = [0u8; CACHED_KEY_SIZE];
200        key_slice.clone_from_slice(&key.as_ref()[key_index..key_index + CACHED_KEY_SIZE]);
201
202        // Insert the slot and tx result into the cache entry associated with
203        // this blockhash and keyslice.
204        let forks = hash_map.entry(key_slice).or_default();
205        forks.push((slot, res.clone()));
206
207        self.add_to_slot_delta(transaction_blockhash, slot, key_index, key_slice, res);
208    }
209
210    pub fn purge_roots(&mut self) {
211        if self.roots.len() > MAX_CACHE_ENTRIES {
212            if let Some(min) = self.roots.iter().min().cloned() {
213                self.roots.remove(&min);
214                self.cache.retain(|_, (fork, _, _)| *fork > min);
215                self.slot_deltas.retain(|slot, _| *slot > min);
216            }
217        }
218    }
219
220    /// Clear for testing
221    pub fn clear(&mut self) {
222        for v in self.cache.values_mut() {
223            v.2 = HashMap::new();
224        }
225
226        self.slot_deltas
227            .iter_mut()
228            .for_each(|(_, status)| status.lock().unwrap().clear());
229    }
230
231    /// Get the statuses for all the root slots
232    pub fn root_slot_deltas(&self) -> Vec<SlotDelta<T>> {
233        self.roots()
234            .iter()
235            .map(|root| {
236                (
237                    *root,
238                    true, // <-- is_root
239                    self.slot_deltas.get(root).cloned().unwrap_or_default(),
240                )
241            })
242            .collect()
243    }
244
245    // replay deltas into a status_cache allows "appending" data
246    pub fn append(&mut self, slot_deltas: &[SlotDelta<T>]) {
247        for (slot, is_root, statuses) in slot_deltas {
248            statuses
249                .lock()
250                .unwrap()
251                .iter()
252                .for_each(|(tx_hash, (key_index, statuses))| {
253                    for (key_slice, res) in statuses.iter() {
254                        self.insert_with_slice(tx_hash, *slot, *key_index, *key_slice, res.clone())
255                    }
256                });
257            if *is_root {
258                self.add_root(*slot);
259            }
260        }
261    }
262
263    pub fn from_slot_deltas(slot_deltas: &[SlotDelta<T>]) -> Self {
264        // play all deltas back into the status cache
265        let mut me = Self::default();
266        me.append(slot_deltas);
267        me
268    }
269
270    fn insert_with_slice(
271        &mut self,
272        transaction_blockhash: &Hash,
273        slot: Slot,
274        key_index: usize,
275        key_slice: [u8; CACHED_KEY_SIZE],
276        res: T,
277    ) {
278        let hash_map =
279            self.cache
280                .entry(*transaction_blockhash)
281                .or_insert((slot, key_index, HashMap::new()));
282        hash_map.0 = std::cmp::max(slot, hash_map.0);
283
284        let forks = hash_map.2.entry(key_slice).or_default();
285        forks.push((slot, res.clone()));
286
287        self.add_to_slot_delta(transaction_blockhash, slot, key_index, key_slice, res);
288    }
289
290    // Add this key slice to the list of key slices for this slot and blockhash
291    // combo.
292    fn add_to_slot_delta(
293        &mut self,
294        transaction_blockhash: &Hash,
295        slot: Slot,
296        key_index: usize,
297        key_slice: [u8; CACHED_KEY_SIZE],
298        res: T,
299    ) {
300        let mut fork_entry = self.slot_deltas.entry(slot).or_default().lock().unwrap();
301        let (_key_index, hash_entry) = fork_entry
302            .entry(*transaction_blockhash)
303            .or_insert((key_index, vec![]));
304        hash_entry.push((key_slice, res))
305    }
306}
307
308#[cfg(test)]
309mod tests {
310    use {
311        super::*,
312        solana_sdk::{hash::hash, signature::Signature},
313    };
314
315    type BankStatusCache = StatusCache<()>;
316
317    #[test]
318    fn test_empty_has_no_sigs() {
319        let sig = Signature::default();
320        let blockhash = hash(Hash::default().as_ref());
321        let status_cache = BankStatusCache::default();
322        assert_eq!(
323            status_cache.get_status(sig, &blockhash, &Ancestors::default()),
324            None
325        );
326        assert_eq!(
327            status_cache.get_status_any_blockhash(sig, &Ancestors::default()),
328            None
329        );
330    }
331
332    #[test]
333    fn test_find_sig_with_ancestor_fork() {
334        let sig = Signature::default();
335        let mut status_cache = BankStatusCache::default();
336        let blockhash = hash(Hash::default().as_ref());
337        let ancestors = vec![(0, 1)].into_iter().collect();
338        status_cache.insert(&blockhash, sig, 0, ());
339        assert_eq!(
340            status_cache.get_status(sig, &blockhash, &ancestors),
341            Some((0, ()))
342        );
343        assert_eq!(
344            status_cache.get_status_any_blockhash(sig, &ancestors),
345            Some((0, ()))
346        );
347    }
348
349    #[test]
350    fn test_find_sig_without_ancestor_fork() {
351        let sig = Signature::default();
352        let mut status_cache = BankStatusCache::default();
353        let blockhash = hash(Hash::default().as_ref());
354        let ancestors = Ancestors::default();
355        status_cache.insert(&blockhash, sig, 1, ());
356        assert_eq!(status_cache.get_status(sig, &blockhash, &ancestors), None);
357        assert_eq!(status_cache.get_status_any_blockhash(sig, &ancestors), None);
358    }
359
360    #[test]
361    fn test_find_sig_with_root_ancestor_fork() {
362        let sig = Signature::default();
363        let mut status_cache = BankStatusCache::default();
364        let blockhash = hash(Hash::default().as_ref());
365        let ancestors = Ancestors::default();
366        status_cache.insert(&blockhash, sig, 0, ());
367        status_cache.add_root(0);
368        assert_eq!(
369            status_cache.get_status(sig, &blockhash, &ancestors),
370            Some((0, ()))
371        );
372    }
373
374    #[test]
375    fn test_insert_picks_latest_blockhash_fork() {
376        let sig = Signature::default();
377        let mut status_cache = BankStatusCache::default();
378        let blockhash = hash(Hash::default().as_ref());
379        let ancestors = vec![(0, 0)].into_iter().collect();
380        status_cache.insert(&blockhash, sig, 0, ());
381        status_cache.insert(&blockhash, sig, 1, ());
382        for i in 0..(MAX_CACHE_ENTRIES + 1) {
383            status_cache.add_root(i as u64);
384        }
385        assert!(status_cache
386            .get_status(sig, &blockhash, &ancestors)
387            .is_some());
388    }
389
390    #[test]
391    fn test_root_expires() {
392        let sig = Signature::default();
393        let mut status_cache = BankStatusCache::default();
394        let blockhash = hash(Hash::default().as_ref());
395        let ancestors = Ancestors::default();
396        status_cache.insert(&blockhash, sig, 0, ());
397        for i in 0..(MAX_CACHE_ENTRIES + 1) {
398            status_cache.add_root(i as u64);
399        }
400        assert_eq!(status_cache.get_status(sig, &blockhash, &ancestors), None);
401    }
402
403    #[test]
404    fn test_clear_signatures_sigs_are_gone() {
405        let sig = Signature::default();
406        let mut status_cache = BankStatusCache::default();
407        let blockhash = hash(Hash::default().as_ref());
408        let ancestors = Ancestors::default();
409        status_cache.insert(&blockhash, sig, 0, ());
410        status_cache.add_root(0);
411        status_cache.clear();
412        assert_eq!(status_cache.get_status(sig, &blockhash, &ancestors), None);
413    }
414
415    #[test]
416    fn test_clear_signatures_insert_works() {
417        let sig = Signature::default();
418        let mut status_cache = BankStatusCache::default();
419        let blockhash = hash(Hash::default().as_ref());
420        let ancestors = Ancestors::default();
421        status_cache.add_root(0);
422        status_cache.clear();
423        status_cache.insert(&blockhash, sig, 0, ());
424        assert!(status_cache
425            .get_status(sig, &blockhash, &ancestors)
426            .is_some());
427    }
428
429    #[test]
430    fn test_signatures_slice() {
431        let sig = Signature::default();
432        let mut status_cache = BankStatusCache::default();
433        let blockhash = hash(Hash::default().as_ref());
434        status_cache.clear();
435        status_cache.insert(&blockhash, sig, 0, ());
436        let (_, index, sig_map) = status_cache.cache.get(&blockhash).unwrap();
437        let sig_slice: &[u8; CACHED_KEY_SIZE] =
438            arrayref::array_ref![sig.as_ref(), *index, CACHED_KEY_SIZE];
439        assert!(sig_map.get(sig_slice).is_some());
440    }
441
442    #[test]
443    fn test_slot_deltas() {
444        let sig = Signature::default();
445        let mut status_cache = BankStatusCache::default();
446        let blockhash = hash(Hash::default().as_ref());
447        status_cache.clear();
448        status_cache.insert(&blockhash, sig, 0, ());
449        assert!(status_cache.roots().contains(&0));
450        let slot_deltas = status_cache.root_slot_deltas();
451        let cache = StatusCache::from_slot_deltas(&slot_deltas);
452        assert_eq!(cache, status_cache);
453        let slot_deltas = cache.root_slot_deltas();
454        let cache = StatusCache::from_slot_deltas(&slot_deltas);
455        assert_eq!(cache, status_cache);
456    }
457
458    #[test]
459    fn test_roots_deltas() {
460        let sig = Signature::default();
461        let mut status_cache = BankStatusCache::default();
462        let blockhash = hash(Hash::default().as_ref());
463        let blockhash2 = hash(blockhash.as_ref());
464        status_cache.insert(&blockhash, sig, 0, ());
465        status_cache.insert(&blockhash, sig, 1, ());
466        status_cache.insert(&blockhash2, sig, 1, ());
467        for i in 0..(MAX_CACHE_ENTRIES + 1) {
468            status_cache.add_root(i as u64);
469        }
470        assert_eq!(status_cache.slot_deltas.len(), 1);
471        assert!(status_cache.slot_deltas.contains_key(&1));
472        let slot_deltas = status_cache.root_slot_deltas();
473        let cache = StatusCache::from_slot_deltas(&slot_deltas);
474        assert_eq!(cache, status_cache);
475    }
476
477    #[test]
478    #[allow(clippy::assertions_on_constants)]
479    fn test_age_sanity() {
480        assert!(MAX_CACHE_ENTRIES <= MAX_RECENT_BLOCKHASHES);
481    }
482
483    #[test]
484    fn test_clear_slot_signatures() {
485        let sig = Signature::default();
486        let mut status_cache = BankStatusCache::default();
487        let blockhash = hash(Hash::default().as_ref());
488        let blockhash2 = hash(blockhash.as_ref());
489        status_cache.insert(&blockhash, sig, 0, ());
490        status_cache.insert(&blockhash, sig, 1, ());
491        status_cache.insert(&blockhash2, sig, 1, ());
492
493        let mut ancestors0 = Ancestors::default();
494        ancestors0.insert(0, 0);
495        let mut ancestors1 = Ancestors::default();
496        ancestors1.insert(1, 0);
497
498        // Clear slot 0 related data
499        assert!(status_cache
500            .get_status(sig, &blockhash, &ancestors0)
501            .is_some());
502        status_cache.clear_slot_entries(0);
503        assert!(status_cache
504            .get_status(sig, &blockhash, &ancestors0)
505            .is_none());
506        assert!(status_cache
507            .get_status(sig, &blockhash, &ancestors1)
508            .is_some());
509        assert!(status_cache
510            .get_status(sig, &blockhash2, &ancestors1)
511            .is_some());
512
513        // Check that the slot delta for slot 0 is gone, but slot 1 still
514        // exists
515        assert!(!status_cache.slot_deltas.contains_key(&0));
516        assert!(status_cache.slot_deltas.contains_key(&1));
517
518        // Clear slot 1 related data
519        status_cache.clear_slot_entries(1);
520        assert!(status_cache.slot_deltas.is_empty());
521        assert!(status_cache
522            .get_status(sig, &blockhash, &ancestors1)
523            .is_none());
524        assert!(status_cache
525            .get_status(sig, &blockhash2, &ancestors1)
526            .is_none());
527        assert!(status_cache.cache.is_empty());
528    }
529
530    // Status cache uses a random key offset for each blockhash. Ensure that shorter
531    // keys can still be used if the offset if greater than the key length.
532    #[test]
533    fn test_different_sized_keys() {
534        let mut status_cache = BankStatusCache::default();
535        let ancestors = vec![(0, 0)].into_iter().collect();
536        let blockhash = Hash::default();
537        for _ in 0..100 {
538            let blockhash = hash(blockhash.as_ref());
539            let sig_key = Signature::default();
540            let hash_key = Hash::new_unique();
541            status_cache.insert(&blockhash, sig_key, 0, ());
542            status_cache.insert(&blockhash, hash_key, 0, ());
543            assert!(status_cache
544                .get_status(sig_key, &blockhash, &ancestors)
545                .is_some());
546            assert!(status_cache
547                .get_status(hash_key, &blockhash, &ancestors)
548                .is_some());
549        }
550    }
551}
solana_runtime/status_cache.rs

solana_runtime/
status_cache.rs
