Merge pull request #12 from fjall-rs/generic-blob-cache

Generic blob cache

Author: marvin-j97

Cargo.toml

@@ -28,7 +28,6 @@ byteview = { version = "0.6.1" }
 interval-heap = "0.0.5"
 log = "0.4.22"
 path-absolutize = "3.1.1"
-quick_cache = { version = "0.6.5", default-features = false }
 rustc-hash = "2.0.0"
 serde = { version = "1.0.215", optional = true, features = ["derive"] }
 tempfile = "3.12.0"

benches/value_log.rs

@@ -1,10 +1,70 @@
 use criterion::{criterion_group, criterion_main, Criterion};
 use rand::{Rng, RngCore};
-use std::sync::Arc;
+use std::{
+    collections::BTreeMap,
+    sync::{Arc, RwLock},
+};
 use value_log::{
-    BlobCache, Compressor, Config, IndexReader, IndexWriter, MockIndex, MockIndexWriter, ValueLog,
+    BlobCache, Compressor, Config, IndexReader, IndexWriter, UserKey, UserValue, ValueHandle,
+    ValueLog, ValueLogId,
 };
 
+type MockIndexInner = RwLock<BTreeMap<UserKey, (ValueHandle, u32)>>;
+
+/// Mock in-memory index
+#[allow(clippy::module_name_repetitions)]
+#[derive(Clone, Default)]
+pub struct MockIndex(Arc<MockIndexInner>);
+
+impl std::ops::Deref for MockIndex {
+    type Target = MockIndexInner;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl MockIndex {
+    /// Remove item
+    pub fn remove(&self, key: &[u8]) {
+        self.0.write().expect("lock is poisoned").remove(key);
+    }
+}
+
+impl IndexReader for MockIndex {
+    fn get(&self, key: &[u8]) -> std::io::Result<Option<ValueHandle>> {
+        Ok(self
+            .read()
+            .expect("lock is poisoned")
+            .get(key)
+            .map(|(vhandle, _)| vhandle)
+            .cloned())
+    }
+}
+
+/// Used for tests only
+#[allow(clippy::module_name_repetitions)]
+pub struct MockIndexWriter(pub MockIndex);
+
+impl IndexWriter for MockIndexWriter {
+    fn insert_indirect(
+        &mut self,
+        key: &[u8],
+        value: ValueHandle,
+        size: u32,
+    ) -> std::io::Result<()> {
+        self.0
+            .write()
+            .expect("lock is poisoned")
+            .insert(key.into(), (value, size));
+        Ok(())
+    }
+
+    fn finish(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
 #[derive(Clone, Default)]
 struct NoCompressor;
 
@@ -18,6 +78,17 @@ impl Compressor for NoCompressor {
     }
 }
 
+#[derive(Clone)]
+struct NoCacher;
+
+impl BlobCache for NoCacher {
+    fn get(&self, _: ValueLogId, _: &ValueHandle) -> Option<UserValue> {
+        None
+    }
+
+    fn insert(&self, _: ValueLogId, _: &ValueHandle, _: UserValue) {}
+}
+
 fn prefetch(c: &mut Criterion) {
     let mut group = c.benchmark_group("prefetch range");
 
@@ -30,11 +101,11 @@ fn prefetch(c: &mut Criterion) {
     let folder = tempfile::tempdir().unwrap();
     let vl_path = folder.path();
 
-    let value_log = ValueLog::open(vl_path, Config::<NoCompressor>::default()).unwrap();
+    let value_log = ValueLog::open(vl_path, Config::<_, NoCompressor>::new(NoCacher)).unwrap();
 
     let mut writer = value_log.get_writer().unwrap();
 
-    let mut rng = rand::thread_rng();
+    let mut rng: rand::prelude::ThreadRng = rand::rng();
 
     for key in (0u64..2_000_000).map(u64::to_be_bytes) {
         let mut data = vec![0u8; item_size];
@@ -51,11 +122,11 @@ fn prefetch(c: &mut Criterion) {
 
     value_log.register_writer(writer).unwrap();
 
-    let mut rng = rand::thread_rng();
+    let mut rng = rand::rng();
 
     group.bench_function(format!("{range_size}x{item_size}B - no prefetch"), |b| {
         b.iter(|| {
-            let start = rng.gen_range(0u64..1_999_000);
+            let start = rng.random_range(0u64..1_999_000);
 
             for x in start..(start + range_size) {
                 let vhandle = index.get(&x.to_be_bytes()).unwrap().unwrap();
@@ -67,7 +138,7 @@ fn prefetch(c: &mut Criterion) {
 
     group.bench_function(format!("{range_size}x{item_size}B - with prefetch"), |b| {
         b.iter(|| {
-            let start = rng.gen_range(0u64..1_999_000);
+            let start = rng.random_range(0u64..1_999_000);
 
             {
                 let vhandle = index.get(&start.to_be_bytes()).unwrap().unwrap();
@@ -113,65 +184,11 @@ fn load_value(c: &mut Criterion) {
         let folder = tempfile::tempdir().unwrap();
         let vl_path = folder.path();
 
-        let value_log = ValueLog::open(
-            vl_path,
-            Config::<NoCompressor>::default()
-                .blob_cache(Arc::new(BlobCache::with_capacity_bytes(0))),
-        )
-        .unwrap();
-
-        let mut writer = value_log.get_writer().unwrap();
-
-        let mut rng = rand::thread_rng();
-
-        for size in sizes {
-            let key = size.to_string();
-
-            let mut data = vec![0u8; size];
-            rng.fill_bytes(&mut data);
-
-            index_writer
-                .insert_indirect(
-                    key.as_bytes(),
-                    writer.get_next_value_handle(),
-                    data.len() as u32,
-                )
-                .unwrap();
-
-            writer.write(key.as_bytes(), &data).unwrap();
-        }
-
-        value_log.register_writer(writer).unwrap();
-
-        for size in sizes {
-            let key = size.to_string();
-            let vhandle = index.get(key.as_bytes()).unwrap().unwrap();
-
-            group.bench_function(format!("{size} bytes (uncached)"), |b| {
-                b.iter(|| {
-                    value_log.get(&vhandle).unwrap().unwrap();
-                })
-            });
-        }
-    }
-
-    {
-        let index = MockIndex::default();
-        let mut index_writer = MockIndexWriter(index.clone());
-
-        let folder = tempfile::tempdir().unwrap();
-        let vl_path = folder.path();
-
-        let value_log = ValueLog::open(
-            vl_path,
-            Config::<NoCompressor>::default()
-                .blob_cache(Arc::new(BlobCache::with_capacity_bytes(64 * 1_024 * 1_024))),
-        )
-        .unwrap();
+        let value_log = ValueLog::open(vl_path, Config::<_, NoCompressor>::new(NoCacher)).unwrap();
 
         let mut writer = value_log.get_writer().unwrap();
 
-        let mut rng = rand::thread_rng();
+        let mut rng = rand::rng();
 
         for size in sizes {
             let key = size.to_string();
@@ -196,10 +213,7 @@ fn load_value(c: &mut Criterion) {
             let key = size.to_string();
             let vhandle = index.get(key.as_bytes()).unwrap().unwrap();
 
-            // NOTE: Warm up cache
-            value_log.get(&vhandle).unwrap().unwrap();
-
-            group.bench_function(format!("{size} bytes (cached)"), |b| {
+            group.bench_function(format!("{size} bytes"), |b| {
                 b.iter(|| {
                     value_log.get(&vhandle).unwrap().unwrap();
                 })

src/blob_cache.rs

@@ -2,106 +2,17 @@
 // This source code is licensed under both the Apache 2.0 and MIT License
 // (found in the LICENSE-* files in the repository)
 
-use crate::{value::UserValue, value_log::ValueLogId, ValueHandle};
-use quick_cache::{sync::Cache, Equivalent, Weighter};
-
-type Item = UserValue;
-
-#[derive(Eq, std::hash::Hash, PartialEq)]
-pub struct CacheKey(ValueLogId, ValueHandle);
-
-impl Equivalent<CacheKey> for (ValueLogId, &ValueHandle) {
-    fn equivalent(&self, key: &CacheKey) -> bool {
-        self.0 == key.0 && self.1 == &key.1
-    }
-}
-
-impl From<(ValueLogId, ValueHandle)> for CacheKey {
-    fn from((vid, vhandle): (ValueLogId, ValueHandle)) -> Self {
-        Self(vid, vhandle)
-    }
-}
-
-#[derive(Clone)]
-struct BlobWeighter;
-
-impl Weighter<CacheKey, Item> for BlobWeighter {
-    #[allow(clippy::cast_possible_truncation)]
-    fn weight(&self, _: &CacheKey, blob: &Item) -> u64 {
-        blob.len() as u64
-    }
-}
+use crate::{value_log::ValueLogId, UserValue, ValueHandle};
 
 /// Blob cache, in which blobs are cached in-memory
 /// after being retrieved from disk
 ///
 /// This speeds up consecutive accesses to the same blobs, improving
 /// read performance for hot data.
-pub struct BlobCache {
-    // NOTE: rustc_hash performed best: https://fjall-rs.github.io/post/fjall-2-1
-    /// Concurrent cache implementation
-    data: Cache<CacheKey, Item, BlobWeighter, rustc_hash::FxBuildHasher>,
-
-    /// Capacity in bytes
-    capacity: u64,
-}
-
-impl std::fmt::Debug for BlobCache {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "BlobCache<cap: {} bytes>", self.capacity)
-    }
-}
-
-impl BlobCache {
-    /// Creates a new block cache with roughly `n` bytes of capacity.
-    #[must_use]
-    pub fn with_capacity_bytes(bytes: u64) -> Self {
-        use quick_cache::sync::DefaultLifecycle;
-
-        #[allow(clippy::default_trait_access)]
-        let quick_cache = Cache::with(
-            10_000,
-            bytes,
-            BlobWeighter,
-            Default::default(),
-            DefaultLifecycle::default(),
-        );
-
-        Self {
-            data: quick_cache,
-            capacity: bytes,
-        }
-    }
-
-    pub(crate) fn insert(&self, key: CacheKey, value: UserValue) {
-        self.data.insert(key, value);
-    }
-
-    pub(crate) fn get(&self, vlog_id: ValueLogId, vhandle: &ValueHandle) -> Option<Item> {
-        self.data.get(&(vlog_id, vhandle))
-    }
-
-    /// Returns the cache capacity in bytes.
-    #[must_use]
-    pub fn capacity(&self) -> u64 {
-        self.capacity
-    }
-
-    /// Returns the size in bytes.
-    #[must_use]
-    pub fn size(&self) -> u64 {
-        self.data.weight()
-    }
-
-    /// Returns the number of cached blocks.
-    #[must_use]
-    pub fn len(&self) -> usize {
-        self.data.len()
-    }
+pub trait BlobCache: Clone {
+    /// Caches a blob.
+    fn insert(&self, vlog_id: ValueLogId, vhandle: &ValueHandle, value: UserValue);
 
-    /// Returns `true` if there are no cached blocks.
-    #[must_use]
-    pub fn is_empty(&self) -> bool {
-        self.len() == 0
-    }
+    /// Retrieves a blob from the cache, or `None` if it could not be found.
+    fn get(&self, vlog_id: ValueLogId, vhandle: &ValueHandle) -> Option<UserValue>;
 }

src/config.rs

@@ -3,33 +3,29 @@
 // (found in the LICENSE-* files in the repository)
 
 use crate::{blob_cache::BlobCache, compression::Compressor};
-use std::sync::Arc;
 
 /// Value log configuration
-pub struct Config<C: Compressor + Clone> {
+pub struct Config<BC: BlobCache, C: Compressor + Clone> {
     /// Target size of vLog segments
     pub(crate) segment_size_bytes: u64,
 
     /// Blob cache to use
-    pub(crate) blob_cache: Arc<BlobCache>,
+    pub(crate) blob_cache: BC,
 
     /// Compression to use
     pub(crate) compression: C,
 }
 
-impl<C: Compressor + Clone + Default> Default for Config<C> {
-    fn default() -> Self {
+impl<BC: BlobCache, C: Compressor + Clone + Default> Config<BC, C> {
+    /// Creates a new configuration builder.
+    pub fn new(blob_cache: BC) -> Self {
         Self {
-            segment_size_bytes: /* 256 MiB */ 256 * 1_024 * 1_024,
-            blob_cache: Arc::new(BlobCache::with_capacity_bytes(
-                /* 16 MiB */ 16 * 1_024 * 1_024,
-            )),
-            compression: C::default(),
+            blob_cache,
+            compression: Default::default(),
+            segment_size_bytes: 128 * 1_024 * 1_024,
         }
     }
-}
 
-impl<C: Compressor + Clone> Config<C> {
     /// Sets the compression & decompression scheme.
     #[must_use]
     pub fn compression(mut self, compressor: C) -> Self {
@@ -41,10 +37,8 @@ impl<C: Compressor + Clone> Config<C> {
     ///
     /// You can create a global [`BlobCache`] and share it between multiple
     /// value logs to cap global cache memory usage.
-    ///
-    /// Defaults to a blob cache with 16 MiB of capacity *per value log*.
     #[must_use]
-    pub fn blob_cache(mut self, blob_cache: Arc<BlobCache>) -> Self {
+    pub fn blob_cache(mut self, blob_cache: BC) -> Self {
         self.blob_cache = blob_cache;
         self
     }