terminus_store/storage/
cache.rs

1use super::layer::*;
2use crate::layer::*;
3use async_trait::async_trait;
4use std::collections::HashMap;
5use std::io;
6use std::path::Path;
7use std::sync::{Arc, RwLock, Weak};
8use tdb_succinct::{StringDict, TypedDict};
9
10pub trait LayerCache: 'static + Send + Sync {
11    fn get_layer_from_cache(&self, name: [u32; 5]) -> Option<Arc<InternalLayer>>;
12    fn cache_layer(&self, layer: Arc<InternalLayer>);
13
14    fn invalidate(&self, name: [u32; 5]);
15}
16
17pub struct NoCache;
18
19impl LayerCache for NoCache {
20    fn get_layer_from_cache(&self, _name: [u32; 5]) -> Option<Arc<InternalLayer>> {
21        None
22    }
23
24    fn cache_layer(&self, _layer: Arc<InternalLayer>) {}
25
26    fn invalidate(&self, _name: [u32; 5]) {}
27}
28
29lazy_static! {
30    pub static ref NOCACHE: Arc<dyn LayerCache> = Arc::new(NoCache);
31}
32
33// locking isn't really ideal but the lock window will be relatively small so it shouldn't hurt performance too much except on heavy updates.
34// ideally we should be using some concurrent hashmap implementation instead.
35// furthermore, there should be some logic to remove stale entries, like a periodic pass. right now, there isn't.
36#[derive(Default)]
37pub struct LockingHashMapLayerCache {
38    cache: RwLock<HashMap<[u32; 5], Weak<InternalLayer>>>,
39}
40
41impl LockingHashMapLayerCache {
42    pub fn new() -> Self {
43        Default::default()
44    }
45}
46
47impl LayerCache for LockingHashMapLayerCache {
48    fn get_layer_from_cache(&self, name: [u32; 5]) -> Option<Arc<InternalLayer>> {
49        let cache = self
50            .cache
51            .read()
52            .expect("rwlock read should always succeed");
53
54        let result = cache.get(&name).map(|c| c.to_owned());
55        std::mem::drop(cache);
56
57        match result {
58            None => None,
59            Some(weak) => match weak.upgrade() {
60                None => {
61                    self.cache
62                        .write()
63                        .expect("rwlock write should always succeed")
64                        .remove(&name);
65                    None
66                }
67                Some(result) => Some(result),
68            },
69        }
70    }
71
72    fn cache_layer(&self, layer: Arc<InternalLayer>) {
73        let mut cache = self
74            .cache
75            .write()
76            .expect("rwlock write should always succeed");
77        cache.insert(layer.name(), Arc::downgrade(&layer));
78    }
79
80    fn invalidate(&self, name: [u32; 5]) {
81        // the dumb way - we just delete the thing from cache forcing a refresh
82        let mut cache = self
83            .cache
84            .write()
85            .expect("rwlock read should always succeed");
86
87        cache.remove(&name);
88    }
89}
90
91#[derive(Clone)]
92pub struct CachedLayerStore {
93    pub(crate) inner: Arc<dyn LayerStore>,
94    pub(crate) cache: Arc<dyn LayerCache>,
95}
96
97impl CachedLayerStore {
98    pub fn new<S: LayerStore, C: LayerCache>(inner: S, cache: C) -> CachedLayerStore {
99        CachedLayerStore {
100            inner: Arc::new(inner),
101            cache: Arc::new(cache),
102        }
103    }
104
105    pub fn invalidate(&self, name: [u32; 5]) {
106        self.cache.invalidate(name);
107    }
108}
109
110#[async_trait]
111impl LayerStore for CachedLayerStore {
112    async fn layers(&self) -> io::Result<Vec<[u32; 5]>> {
113        self.inner.layers().await
114    }
115
116    async fn get_layer(&self, name: [u32; 5]) -> io::Result<Option<Arc<InternalLayer>>> {
117        self.inner
118            .get_layer_with_cache(name, self.cache.clone())
119            .await
120    }
121
122    async fn get_layer_with_cache(
123        &self,
124        name: [u32; 5],
125        cache: Arc<dyn LayerCache>,
126    ) -> io::Result<Option<Arc<InternalLayer>>> {
127        self.inner.get_layer_with_cache(name, cache).await
128    }
129
130    async fn finalize_layer(&self, name: [u32; 5]) -> io::Result<()> {
131        self.inner.finalize_layer(name).await
132    }
133
134    async fn get_layer_parent_name(&self, name: [u32; 5]) -> io::Result<Option<[u32; 5]>> {
135        // is layer in cache? if so, we can use the cached version
136        if let Some(layer) = self.cache.get_layer_from_cache(name) {
137            Ok(layer.parent_name())
138        } else {
139            self.inner.get_layer_parent_name(name).await
140        }
141    }
142
143    async fn get_node_dictionary(&self, name: [u32; 5]) -> io::Result<Option<StringDict>> {
144        // is layer in cache? if so, we can use the cached version
145        if let Some(layer) = self.cache.get_layer_from_cache(name) {
146            // unless it is a rollup
147            if !layer.is_rollup() {
148                return Ok(Some(layer.node_dictionary().clone()));
149            }
150        }
151
152        self.inner.get_node_dictionary(name).await
153    }
154
155    async fn get_predicate_dictionary(&self, name: [u32; 5]) -> io::Result<Option<StringDict>> {
156        // is layer in cache? if so, we can use the cached version
157        if let Some(layer) = self.cache.get_layer_from_cache(name) {
158            // unless it is a rollup
159            if !layer.is_rollup() {
160                return Ok(Some(layer.predicate_dictionary().clone()));
161            }
162        }
163
164        self.inner.get_predicate_dictionary(name).await
165    }
166
167    async fn get_value_dictionary(&self, name: [u32; 5]) -> io::Result<Option<TypedDict>> {
168        // is layer in cache? if so, we can use the cached version
169        if let Some(layer) = self.cache.get_layer_from_cache(name) {
170            // unless it is a rollup
171            if !layer.is_rollup() {
172                return Ok(Some(layer.value_dictionary().clone()));
173            }
174        }
175
176        self.inner.get_value_dictionary(name).await
177    }
178
179    async fn get_node_count(&self, name: [u32; 5]) -> io::Result<Option<u64>> {
180        // is layer in cache? if so, we can use the cached version
181        if let Some(layer) = self.cache.get_layer_from_cache(name) {
182            // unless it is a rollup
183            if !layer.is_rollup() {
184                return Ok(Some(layer.node_dictionary().num_entries() as u64));
185            }
186        }
187
188        self.inner.get_node_count(name).await
189    }
190
191    async fn get_predicate_count(&self, name: [u32; 5]) -> io::Result<Option<u64>> {
192        // is layer in cache? if so, we can use the cached version
193        if let Some(layer) = self.cache.get_layer_from_cache(name) {
194            // unless it is a rollup
195            if !layer.is_rollup() {
196                return Ok(Some(layer.predicate_dictionary().num_entries() as u64));
197            }
198        }
199
200        self.inner.get_value_count(name).await
201    }
202
203    async fn get_value_count(&self, name: [u32; 5]) -> io::Result<Option<u64>> {
204        // is layer in cache? if so, we can use the cached version
205        if let Some(layer) = self.cache.get_layer_from_cache(name) {
206            // unless it is a rollup
207            if !layer.is_rollup() {
208                return Ok(Some(layer.value_dictionary().num_entries() as u64));
209            }
210        }
211
212        self.inner.get_value_count(name).await
213    }
214
215    async fn get_node_value_idmap(&self, name: [u32; 5]) -> io::Result<Option<IdMap>> {
216        // is layer in cache? if so, we can use the cached version
217        if let Some(layer) = self.cache.get_layer_from_cache(name) {
218            // unless it is a rollup
219            if !layer.is_rollup() {
220                return Ok(Some(layer.node_value_id_map().clone()));
221            }
222        }
223
224        self.inner.get_node_value_idmap(name).await
225    }
226
227    async fn get_predicate_idmap(&self, name: [u32; 5]) -> io::Result<Option<IdMap>> {
228        // is layer in cache? if so, we can use the cached version
229        if let Some(layer) = self.cache.get_layer_from_cache(name) {
230            // unless it is a rollup
231            if !layer.is_rollup() {
232                return Ok(Some(layer.predicate_id_map().clone()));
233            }
234        }
235
236        self.inner.get_predicate_idmap(name).await
237    }
238
239    async fn create_base_layer(&self) -> io::Result<Box<dyn LayerBuilder>> {
240        self.inner.create_base_layer().await
241    }
242
243    async fn create_child_layer(&self, parent: [u32; 5]) -> io::Result<Box<dyn LayerBuilder>> {
244        self.inner
245            .create_child_layer_with_cache(parent, self.cache.clone())
246            .await
247    }
248
249    async fn create_child_layer_with_cache(
250        &self,
251        parent: [u32; 5],
252        cache: Arc<dyn LayerCache>,
253    ) -> io::Result<Box<dyn LayerBuilder>> {
254        self.inner
255            .create_child_layer_with_cache(parent, cache)
256            .await
257    }
258
259    async fn perform_rollup(&self, layer: Arc<InternalLayer>) -> io::Result<[u32; 5]> {
260        self.inner.perform_rollup(layer).await
261    }
262
263    async fn perform_rollup_upto_with_cache(
264        &self,
265        layer: Arc<InternalLayer>,
266        upto: [u32; 5],
267        cache: Arc<dyn LayerCache>,
268    ) -> io::Result<[u32; 5]> {
269        self.inner
270            .perform_rollup_upto_with_cache(layer, upto, cache)
271            .await
272    }
273
274    async fn perform_rollup_upto(
275        &self,
276        layer: Arc<InternalLayer>,
277        upto: [u32; 5],
278    ) -> io::Result<[u32; 5]> {
279        self.inner
280            .perform_rollup_upto_with_cache(layer, upto, self.cache.clone())
281            .await
282    }
283
284    async fn perform_imprecise_rollup_upto_with_cache(
285        &self,
286        layer: Arc<InternalLayer>,
287        upto: [u32; 5],
288        cache: Arc<dyn LayerCache>,
289    ) -> io::Result<[u32; 5]> {
290        self.inner
291            .perform_imprecise_rollup_upto_with_cache(layer, upto, cache)
292            .await
293    }
294
295    async fn perform_imprecise_rollup_upto(
296        &self,
297        layer: Arc<InternalLayer>,
298        upto: [u32; 5],
299    ) -> io::Result<[u32; 5]> {
300        self.inner
301            .perform_imprecise_rollup_upto_with_cache(layer, upto, self.cache.clone())
302            .await
303    }
304
305    async fn register_rollup(&self, layer: [u32; 5], rollup: [u32; 5]) -> io::Result<()> {
306        // when registering a rollup layer, we need to make sure that
307        // the cached version is updated as well.
308        self.inner.register_rollup(layer, rollup).await?;
309        self.cache.invalidate(layer);
310
311        Ok(())
312    }
313
314    async fn rollup_upto(&self, layer: Arc<InternalLayer>, upto: [u32; 5]) -> io::Result<[u32; 5]> {
315        let cache = self.cache.clone();
316        self.rollup_upto_with_cache(layer, upto, cache).await
317    }
318
319    async fn squash(&self, layer: Arc<InternalLayer>) -> io::Result<[u32; 5]> {
320        self.inner.squash(layer).await
321    }
322
323    async fn squash_upto(&self, layer: Arc<InternalLayer>, upto: [u32; 5]) -> io::Result<[u32; 5]> {
324        self.inner.squash_upto(layer, upto).await
325    }
326
327    async fn merge_base_layer(&self, layers: &[[u32; 5]], temp_dir: &Path) -> io::Result<[u32; 5]> {
328        self.inner.merge_base_layer(layers, temp_dir).await
329    }
330
331    async fn layer_is_ancestor_of(
332        &self,
333        descendant: [u32; 5],
334        ancestor: [u32; 5],
335    ) -> io::Result<bool> {
336        self.inner.layer_is_ancestor_of(descendant, ancestor).await
337    }
338
339    async fn triple_addition_exists(
340        &self,
341        layer: [u32; 5],
342        subject: u64,
343        predicate: u64,
344        object: u64,
345    ) -> io::Result<bool> {
346        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
347            if !cached.is_rollup() {
348                return Ok(cached.internal_triple_addition_exists(subject, predicate, object));
349            }
350        }
351
352        self.inner
353            .triple_addition_exists(layer, subject, predicate, object)
354            .await
355    }
356
357    async fn triple_removal_exists(
358        &self,
359        layer: [u32; 5],
360        subject: u64,
361        predicate: u64,
362        object: u64,
363    ) -> io::Result<bool> {
364        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
365            if !cached.is_rollup() {
366                return Ok(cached.internal_triple_removal_exists(subject, predicate, object));
367            }
368        }
369
370        self.inner
371            .triple_removal_exists(layer, subject, predicate, object)
372            .await
373    }
374
375    async fn triple_additions(
376        &self,
377        layer: [u32; 5],
378    ) -> io::Result<OptInternalLayerTripleSubjectIterator> {
379        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
380            if !cached.is_rollup() {
381                return Ok(cached.internal_triple_additions());
382            }
383        }
384
385        self.inner.triple_additions(layer).await
386    }
387
388    async fn triple_removals(
389        &self,
390        layer: [u32; 5],
391    ) -> io::Result<OptInternalLayerTripleSubjectIterator> {
392        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
393            if !cached.is_rollup() {
394                return Ok(cached.internal_triple_removals());
395            }
396        }
397
398        self.inner.triple_removals(layer).await
399    }
400
401    async fn triple_additions_s(
402        &self,
403        layer: [u32; 5],
404        subject: u64,
405    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
406        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
407            if !cached.is_rollup() {
408                return Ok(cached.internal_triple_additions_s(subject));
409            }
410        }
411
412        self.inner.triple_additions_s(layer, subject).await
413    }
414
415    async fn triple_removals_s(
416        &self,
417        layer: [u32; 5],
418        subject: u64,
419    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
420        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
421            if !cached.is_rollup() {
422                return Ok(cached.internal_triple_removals_s(subject));
423            }
424        }
425
426        self.inner.triple_removals_s(layer, subject).await
427    }
428
429    async fn triple_additions_sp(
430        &self,
431        layer: [u32; 5],
432        subject: u64,
433        predicate: u64,
434    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
435        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
436            if !cached.is_rollup() {
437                return Ok(cached.internal_triple_additions_sp(subject, predicate));
438            }
439        }
440
441        self.inner
442            .triple_additions_sp(layer, subject, predicate)
443            .await
444    }
445
446    async fn triple_removals_sp(
447        &self,
448        layer: [u32; 5],
449        subject: u64,
450        predicate: u64,
451    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
452        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
453            if !cached.is_rollup() {
454                return Ok(cached.internal_triple_removals_sp(subject, predicate));
455            }
456        }
457
458        self.inner
459            .triple_removals_sp(layer, subject, predicate)
460            .await
461    }
462
463    async fn triple_additions_p(
464        &self,
465        layer: [u32; 5],
466        predicate: u64,
467    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
468        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
469            if !cached.is_rollup() {
470                return Ok(Box::new(cached.internal_triple_additions_p(predicate))
471                    as Box<dyn Iterator<Item = _> + Send>);
472            }
473        }
474
475        self.inner.triple_additions_p(layer, predicate).await
476    }
477
478    async fn triple_removals_p(
479        &self,
480        layer: [u32; 5],
481        predicate: u64,
482    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
483        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
484            if !cached.is_rollup() {
485                return Ok(Box::new(cached.internal_triple_removals_p(predicate))
486                    as Box<dyn Iterator<Item = _> + Send>);
487            }
488        }
489
490        self.inner.triple_removals_p(layer, predicate).await
491    }
492
493    async fn triple_additions_o(
494        &self,
495        layer: [u32; 5],
496        object: u64,
497    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
498        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
499            if !cached.is_rollup() {
500                return Ok(cached.internal_triple_additions_o(object));
501            }
502        }
503
504        self.inner.triple_additions_o(layer, object).await
505    }
506
507    async fn triple_removals_o(
508        &self,
509        layer: [u32; 5],
510        object: u64,
511    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
512        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
513            if !cached.is_rollup() {
514                return Ok(cached.internal_triple_removals_o(object));
515            }
516        }
517
518        self.inner.triple_removals_o(layer, object).await
519    }
520
521    async fn triple_layer_addition_count(&self, layer: [u32; 5]) -> io::Result<usize> {
522        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
523            if !cached.is_rollup() {
524                return Ok(cached.internal_triple_layer_addition_count());
525            }
526        }
527
528        self.inner.triple_layer_addition_count(layer).await
529    }
530
531    async fn triple_layer_removal_count(&self, layer: [u32; 5]) -> io::Result<usize> {
532        if let Some(cached) = self.cache.get_layer_from_cache(layer) {
533            if !cached.is_rollup() {
534                return Ok(cached.internal_triple_layer_removal_count());
535            }
536        }
537
538        self.inner.triple_layer_removal_count(layer).await
539    }
540
541    async fn retrieve_layer_stack_names(&self, name: [u32; 5]) -> io::Result<Vec<[u32; 5]>> {
542        self.inner.retrieve_layer_stack_names(name).await
543    }
544
545    async fn retrieve_layer_stack_names_upto(
546        &self,
547        name: [u32; 5],
548        upto: [u32; 5],
549    ) -> io::Result<Vec<[u32; 5]>> {
550        self.inner.retrieve_layer_stack_names_upto(name, upto).await
551    }
552}
553
554#[cfg(test)]
555pub mod tests {
556    use super::*;
557    use crate::storage::directory::*;
558    use crate::storage::memory::*;
559    use tempfile::tempdir;
560
561    fn cached_layer_eq(layer1: &dyn Layer, layer2: &dyn Layer) -> bool {
562        // a trait object consists of two parts, a pointer to the concrete data, followed by a vtable.
563        // we consider two layers equal if that first part, the pointer to the concrete data, is equal.
564        unsafe {
565            let ptr1 = *(layer1 as *const dyn Layer as *const usize);
566            let ptr2 = *(layer2 as *const dyn Layer as *const usize);
567            ptr1 == ptr2
568        }
569    }
570
571    #[tokio::test]
572    async fn cached_memory_layer_store_returns_same_layer_multiple_times() {
573        let store = CachedLayerStore::new(MemoryLayerStore::new(), LockingHashMapLayerCache::new());
574        let mut builder = store.create_base_layer().await.unwrap();
575        let base_name = builder.name();
576
577        builder.add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"));
578        builder.add_value_triple(ValueTriple::new_string_value("pig", "says", "oink"));
579        builder.add_value_triple(ValueTriple::new_string_value("duck", "says", "quack"));
580
581        builder.commit_boxed().await.unwrap();
582
583        builder = store.create_child_layer(base_name).await.unwrap();
584        let child_name = builder.name();
585
586        builder.remove_value_triple(ValueTriple::new_string_value("duck", "says", "quack"));
587        builder.add_value_triple(ValueTriple::new_node("cow", "likes", "pig"));
588
589        builder.commit_boxed().await.unwrap();
590
591        let layer1 = store.get_layer(child_name).await.unwrap().unwrap();
592        let layer2 = store.get_layer(child_name).await.unwrap().unwrap();
593
594        let base_layer = store.cache.get_layer_from_cache(base_name).unwrap();
595        let base_layer_2 = store.get_layer(base_name).await.unwrap().unwrap();
596
597        assert!(cached_layer_eq(&*layer1, &*layer2));
598        assert!(cached_layer_eq(&*base_layer, &*base_layer_2));
599    }
600
601    #[tokio::test]
602    async fn cached_directory_layer_store_returns_same_layer_multiple_times() {
603        let dir = tempdir().unwrap();
604        let store = CachedLayerStore::new(
605            DirectoryLayerStore::new(dir.path()),
606            LockingHashMapLayerCache::new(),
607        );
608        let mut builder = store.create_base_layer().await.unwrap();
609        let base_name = builder.name();
610
611        builder.add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"));
612        builder.add_value_triple(ValueTriple::new_string_value("pig", "says", "oink"));
613        builder.add_value_triple(ValueTriple::new_string_value("duck", "says", "quack"));
614
615        builder.commit_boxed().await.unwrap();
616
617        builder = store.create_child_layer(base_name).await.unwrap();
618        let child_name = builder.name();
619
620        builder.remove_value_triple(ValueTriple::new_string_value("duck", "says", "quack"));
621        builder.add_value_triple(ValueTriple::new_node("cow", "likes", "pig"));
622
623        builder.commit_boxed().await.unwrap();
624
625        let layer1 = store.get_layer(child_name).await.unwrap().unwrap();
626        let layer2 = store.get_layer(child_name).await.unwrap().unwrap();
627
628        let base_layer = store.cache.get_layer_from_cache(base_name).unwrap();
629        let base_layer_2 = store.get_layer(base_name).await.unwrap().unwrap();
630
631        assert!(cached_layer_eq(&*layer1, &*layer2));
632        assert!(cached_layer_eq(&*base_layer, &*base_layer_2));
633    }
634
635    #[tokio::test]
636    async fn cached_layer_store_forgets_entries_when_they_are_dropped() {
637        let store = CachedLayerStore::new(MemoryLayerStore::new(), LockingHashMapLayerCache::new());
638        let mut builder = store.create_base_layer().await.unwrap();
639        let base_name = builder.name();
640
641        builder.add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"));
642        builder.add_value_triple(ValueTriple::new_string_value("pig", "says", "oink"));
643        builder.add_value_triple(ValueTriple::new_string_value("duck", "says", "quack"));
644
645        builder.commit_boxed().await.unwrap();
646
647        let layer = store.get_layer(base_name).await.unwrap().unwrap();
648        let weak = Arc::downgrade(&layer);
649
650        // we expect 2 weak pointers, the one we made above and the one stored in cache
651        assert_eq!(2, Arc::weak_count(&layer));
652
653        // forget the layers
654        std::mem::drop(layer);
655
656        // according to our weak reference, there's no longer any strong reference around
657        assert!(weak.upgrade().is_none());
658
659        // retrieving the same layer again works just fine
660        let layer = store.get_layer(base_name).await.unwrap().unwrap();
661
662        // and only has one weak pointer pointing to it, the newly cached one
663        assert_eq!(1, Arc::weak_count(&layer));
664    }
665
666    #[test]
667    fn retrieve_layer_stack_names_retrieves_correctly() {
668        //let store = CachedLayerStore::new(MemoryLayerStore::new());
669        //let builder = store.create_base_layer().wait().unwrap();
670    }
671}
terminus_store/storage/cache.rs

terminus_store/storage/
cache.rs
