feat: support in-memory cache online resize

Cargo.toml

@@ -22,6 +22,7 @@ license = "Apache-2.0"
 readme = "README.md"
 
 [workspace.dependencies]
+ahash = "0.8"
 bytesize = { package = "foyer-bytesize", version = "2" }
 clap = { version = "4", features = ["derive"] }
 crossbeam = "0.8"
@@ -37,6 +38,7 @@ test-log = { version = "0.2", default-features = false, features = [
     "trace",
     "color",
 ] }
+thiserror = "2"
 tokio = { package = "madsim-tokio", version = "0.2", features = [
     "rt",
     "rt-multi-thread",

examples/Cargo.toml

@@ -14,7 +14,7 @@ publish = false
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-ahash = "0.8"
+ahash = { workspace = true }
 anyhow = "1"
 chrono = "0.4"
 equivalent = { workspace = true }

foyer-cli/Cargo.toml

@@ -17,7 +17,7 @@ publish = false
 anyhow = "1"
 bytesize = { workspace = true }
 clap = { workspace = true }
-thiserror = "1"
+thiserror = { workspace = true }
 
 [dev-dependencies]
 

foyer-common/Cargo.toml

@@ -13,6 +13,7 @@ readme = { workspace = true }
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+ahash = { workspace = true }
 bytes = "1"
 cfg-if = "1"
 fastrace = { workspace = true }

foyer-common/src/hasher.rs

@@ -0,0 +1,103 @@
+//  Copyright 2024 foyer Project Authors
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use std::hash::{BuildHasher, Hasher};
+
+pub use ahash::RandomState as AhashRandomState;
+
+/// A hasher return u64 mod result.
+#[derive(Debug, Default)]
+pub struct ModRandomState {
+    state: u64,
+}
+
+impl Hasher for ModRandomState {
+    fn finish(&self) -> u64 {
+        self.state
+    }
+
+    fn write(&mut self, bytes: &[u8]) {
+        for byte in bytes {
+            self.state = (self.state << 8) + *byte as u64;
+        }
+    }
+
+    fn write_u8(&mut self, i: u8) {
+        self.write(&[i])
+    }
+
+    fn write_u16(&mut self, i: u16) {
+        self.write(&i.to_be_bytes())
+    }
+
+    fn write_u32(&mut self, i: u32) {
+        self.write(&i.to_be_bytes())
+    }
+
+    fn write_u64(&mut self, i: u64) {
+        self.write(&i.to_be_bytes())
+    }
+
+    fn write_u128(&mut self, i: u128) {
+        self.write(&i.to_be_bytes())
+    }
+
+    fn write_usize(&mut self, i: usize) {
+        self.write(&i.to_be_bytes())
+    }
+
+    fn write_i8(&mut self, i: i8) {
+        self.write_u8(i as u8)
+    }
+
+    fn write_i16(&mut self, i: i16) {
+        self.write_u16(i as u16)
+    }
+
+    fn write_i32(&mut self, i: i32) {
+        self.write_u32(i as u32)
+    }
+
+    fn write_i64(&mut self, i: i64) {
+        self.write_u64(i as u64)
+    }
+
+    fn write_i128(&mut self, i: i128) {
+        self.write_u128(i as u128)
+    }
+
+    fn write_isize(&mut self, i: isize) {
+        self.write_usize(i as usize)
+    }
+}
+
+impl BuildHasher for ModRandomState {
+    type Hasher = Self;
+
+    fn build_hasher(&self) -> Self::Hasher {
+        Self::default()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_mod_hasher() {
+        for i in 0..255u8 {
+            assert_eq!(i, ModRandomState::default().hash_one(i) as u8,)
+        }
+    }
+}

foyer-common/src/lib.rs

@@ -30,6 +30,8 @@ pub mod countdown;
 pub mod event;
 /// Future extensions.
 pub mod future;
+/// Provisioned hashers.
+pub mod hasher;
 /// The shared metrics for foyer.
 pub mod metrics;
 /// Extensions for [`std::option::Option`].

foyer-memory/Cargo.toml

@@ -13,7 +13,7 @@ readme = { workspace = true }
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-ahash = "0.8"
+ahash = { workspace = true }
 bitflags = "2"
 cmsketch = "0.2.1"
 equivalent = { workspace = true }
@@ -27,6 +27,7 @@ parking_lot = { workspace = true }
 paste = "1"
 pin-project = "1"
 serde = { workspace = true }
+thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = "0.1"
 

foyer-memory/src/cache.rs

@@ -36,6 +36,7 @@
     },
     raw::{FetchMark, FetchState, RawCache, RawCacheConfig, RawCacheEntry, RawFetch, Weighter},
     record::CacheHint,
+    Result,
 };
 
 pub type FifoCache<K, V, S = RandomState> = RawCache<Fifo<K, V>, S>;
@@ -513,6 +514,17 @@
     V: Value,
     S: HashBuilder,
 {
+    /// Update capacity and evict overflowed entries.
+    #[fastrace::trace(name = "foyer::memory::cache::resize")]
+    pub fn resize(&self, capacity: usize) -> Result<()> {
+        match self {
+            Cache::Fifo(cache) => cache.resize(capacity),
+            Cache::S3Fifo(cache) => cache.resize(capacity),
+            Cache::Lru(cache) => cache.resize(capacity),
+            Cache::Lfu(cache) => cache.resize(capacity),
+        }
+    }
+
     /// Insert cache entry to the in-memory cache.
     #[fastrace::trace(name = "foyer::memory::cache::insert")]
     pub fn insert(&self, key: K, value: V) -> CacheEntry<K, V, S> {

foyer-memory/src/error.rs

@@ -0,0 +1,53 @@
+//  Copyright 2024 foyer Project Authors
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use std::fmt::Display;
+
+/// In-memory cache error.
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    /// Multiple error list.
+    #[error(transparent)]
+    Multiple(MultipleError),
+    /// Config error.
+    #[error("config error: {0}")]
+    ConfigError(String),
+}
+
+impl Error {
+    /// Combine multiple errors into one error.
+    pub fn multiple(errs: Vec<Error>) -> Self {
+        Self::Multiple(MultipleError(errs))
+    }
+}
+
+#[derive(thiserror::Error, Debug)]
+pub struct MultipleError(Vec<Error>);
+
+impl Display for MultipleError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "multiple errors: [")?;
+        if let Some((last, errs)) = self.0.as_slice().split_last() {
+            for err in errs {
+                write!(f, "{}, ", err)?;
+            }
+            write!(f, "{}", last)?;
+        }
+        write!(f, "]")?;
+        Ok(())
+    }
+}
+
+/// In-memory cache result.
+pub type Result<T> = std::result::Result<T, Error>;

foyer-memory/src/eviction/fifo.rs

@@ -19,7 +19,10 @@ use intrusive_collections::{intrusive_adapter, LinkedList, LinkedListAtomicLink}
 use serde::{Deserialize, Serialize};
 
 use super::{Eviction, Op};
-use crate::record::{CacheHint, Record};
+use crate::{
+    error::Result,
+    record::{CacheHint, Record},
+};
 
 /// Fifo eviction algorithm config.
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -77,7 +80,9 @@ where
         }
     }
 
-    fn update(&mut self, _: usize, _: &Self::Config) {}
+    fn update(&mut self, _: usize, _: Option<&Self::Config>) -> Result<()> {
+        Ok(())
+    }
 
     fn push(&mut self, record: Arc<Record<Self>>) {
         record.set_in_eviction(true);

foyer-memory/src/eviction/lfu.rs

@@ -23,7 +23,10 @@
 use serde::{Deserialize, Serialize};
 
 use super::{Eviction, Op};
-use crate::record::{CacheHint, Record};
+use crate::{
+    error::{Error, Result},
+    record::{CacheHint, Record},
+};
 
 /// w-TinyLFU eviction algorithm config.
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -135,6 +138,8 @@
 
     step: usize,
     decay: usize,
+
+    config: LfuConfig,
 }
 
 impl<K, V> Lfu<K, V>
@@ -203,6 +208,8 @@
             config.window_capacity_ratio + config.protected_capacity_ratio
         );
 
+        let config = config.clone();
+
         let window_weight_capacity = (capacity as f64 * config.window_capacity_ratio) as usize;
         let protected_weight_capacity = (capacity as f64 * config.protected_capacity_ratio) as usize;
         let frequencies = CMSketchU16::new(config.cmsketch_eps, config.cmsketch_confidence);
@@ -220,38 +227,48 @@
             frequencies,
             step: 0,
             decay,
+            config,
         }
     }
 
-    fn update(&mut self, capacity: usize, config: &Self::Config) {
-        if config.window_capacity_ratio <= 0.0 || config.window_capacity_ratio >= 1.0 {
-            tracing::error!(
-                "window_capacity_ratio must be in (0, 1), given: {}, new config ignored",
-                config.window_capacity_ratio
-            );
-        }
+    fn update(&mut self, capacity: usize, config: Option<&Self::Config>) -> Result<()> {
+        if let Some(config) = config {
+            let mut msgs = vec![];
+            if config.window_capacity_ratio <= 0.0 || config.window_capacity_ratio >= 1.0 {
+                msgs.push(format!(
+                    "window_capacity_ratio must be in (0, 1), given: {}, new config ignored",
+                    config.window_capacity_ratio
+                ));
+            }
+            if config.protected_capacity_ratio <= 0.0 || config.protected_capacity_ratio >= 1.0 {
+                msgs.push(format!(
+                    "protected_capacity_ratio must be in (0, 1), given: {}, new config ignored",
+                    config.protected_capacity_ratio
+                ));
+            }
+            if config.window_capacity_ratio + config.protected_capacity_ratio >= 1.0 {
+                msgs.push(format!(
+                    "must guarantee: window_capacity_ratio + protected_capacity_ratio < 1, given: {}, new config ignored",
+                    config.window_capacity_ratio + config.protected_capacity_ratio
+                ));
+            }
 
-        if config.protected_capacity_ratio <= 0.0 || config.protected_capacity_ratio >= 1.0 {
-            tracing::error!(
-                "protected_capacity_ratio must be in (0, 1), given: {}, new config ignored",
-                config.protected_capacity_ratio
-            );
-        }
+            if !msgs.is_empty() {
+                return Err(Error::ConfigError(msgs.join(" | ")));
+            }
 
-        if config.window_capacity_ratio + config.protected_capacity_ratio >= 1.0 {
-            tracing::error!(
-                "must guarantee: window_capacity_ratio + protected_capacity_ratio < 1, given: {}, new config ignored",
-                config.window_capacity_ratio + config.protected_capacity_ratio
-            )
+            self.config = config.clone();
         }
 
         // TODO(MrCroxx): Raise a warn log the cmsketch args updates is not supported yet if it is modified.
 
-        let window_weight_capacity = (capacity as f64 * config.window_capacity_ratio) as usize;
-        let protected_weight_capacity = (capacity as f64 * config.protected_capacity_ratio) as usize;
+        let window_weight_capacity = (capacity as f64 * self.config.window_capacity_ratio) as usize;
+        let protected_weight_capacity = (capacity as f64 * self.config.protected_capacity_ratio) as usize;
 
         self.window_weight_capacity = window_weight_capacity;
         self.protected_weight_capacity = protected_weight_capacity;
+
+        Ok(())
     }
 
     /// Push a new record to `window`.