Merge pull request #145 from moka-rs/eviction-listener

Add support for eviction listener

.vscode/settings.json

@@ -1,8 +1,9 @@
 {
-    "rust-analyzer.cargo.features": ["future", "dash", "unstable-debug-counters"],
+    "rust-analyzer.cargo.features": ["future", "dash", "logging", "unstable-debug-counters"],
     "rust-analyzer.server.extraEnv": {
         "CARGO_TARGET_DIR": "target/ra"
     },
+    "editor.rulers": [85],
     "cSpell.words": [
         "aarch",
         "actix",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "moka"
-version = "0.8.6"
+version = "0.9.0"
 edition = "2018"
 rust-version = "1.51"
 
@@ -58,7 +58,7 @@ tagptr = "0.2"
 
 # Opt-out serde and stable_deref_trait features
 # https://github.com/Manishearth/triomphe/pull/5
-triomphe = { version = "0.1", default-features = false }
+triomphe = { version = "0.1.3", default-features = false }
 
 # Optional dependencies (enabled by default)
 quanta = { version = "0.10.0", optional = true }
@@ -83,11 +83,13 @@ log = { version = "0.4", optional = true }
 
 [dev-dependencies]
 actix-rt = { version = "2.7", default-features = false }
+anyhow = "1.0"
 async-std = { version = "1.11", features = ["attributes"] }
+env_logger = "0.9"
 getrandom = "0.2"
 reqwest = "0.11.11"
 skeptic = "0.13"
-tokio = { version = "1.19", features = ["rt-multi-thread", "macros", "sync", "time" ] }
+tokio = { version = "1.19", features = ["fs", "macros", "rt-multi-thread", "sync", "time" ] }
 
 [target.'cfg(trybuild)'.dev-dependencies]
 trybuild = "1.0"

README.md

@@ -53,6 +53,9 @@ algorithm to determine which entries to evict when the capacity is exceeded.
 - Supports expiration policies:
     - Time to live
     - Time to idle
+- Supports eviction listener, a callback function that will be called when an entry
+  is removed from the cache.
+
 
 [tiny-lfu]: https://github.com/moka-rs/moka/wiki#admission-and-eviction-policies
 
@@ -528,20 +531,23 @@ $ cargo +nightly -Z unstable-options --config 'build.rustdocflags="--cfg docsrs"
 ## Road Map
 
 - [x] `async` optimized caches. (`v0.2.0`)
-- [x] Size-aware eviction. (`v0.7.0` via
-      [#24](https://github.com/moka-rs/moka/pull/24))
-- [X] API stabilization. (Smaller core cache API, shorter names for frequently
-      used methods) (`v0.8.0` via [#105](https://github.com/moka-rs/moka/pull/105))
+- [x] Size-aware eviction. (`v0.7.0` via [#24][gh-pull-024])
+- [x] API stabilization. (Smaller core cache API, shorter names for frequently
+      used methods) (`v0.8.0` via [#105][gh-pull-105])
     - e.g.
     - `get_or_insert_with(K, F)` → `get_with(K, F)`
     - `get_or_try_insert_with(K, F)` → `try_get_with(K, F)`
     - `blocking_insert(K, V)` → `blocking().insert(K, V)`
     - `time_to_live()` → `policy().time_to_live()`
-- [ ] Notifications on eviction, etc.
+- [x] Notifications on eviction. (`v0.9.0` via [#145][gh-pull-145])
 - [ ] Cache statistics. (Hit rate, etc.)
 - [ ] Upgrade TinyLFU to Window-TinyLFU. ([details][tiny-lfu])
 - [ ] The variable (per-entry) expiration, using a hierarchical timer wheel.
 
+[gh-pull-024]: https://github.com/moka-rs/moka/pull/24
+[gh-pull-105]: https://github.com/moka-rs/moka/pull/105
+[gh-pull-145]: https://github.com/moka-rs/moka/pull/145
+
 
 ## About the Name
 

src/cht/segment.rs

@@ -193,6 +193,10 @@ impl<K, V, S> HashMap<K, V, S> {
         }
     }
 
+    pub(crate) fn actual_num_segments(&self) -> usize {
+        self.segments.len()
+    }
+
     /// Returns the number of elements in the map.
     ///
     /// # Safety
@@ -560,10 +564,6 @@ impl<K: Hash + Eq, V, S: BuildHasher> HashMap<K, V, S> {
     {
         bucket::hash(&self.build_hasher, key)
     }
-
-    pub(crate) fn actual_num_segments(&self) -> usize {
-        self.segments.len()
-    }
 }
 
 impl<K, V, S> Drop for HashMap<K, V, S> {

src/common/concurrent/thread_pool.rs

@@ -11,6 +11,8 @@ pub(crate) enum PoolName {
     Housekeeper,
     #[cfg(any(feature = "sync", feature = "future"))]
     Invalidator,
+    #[cfg(any(feature = "sync", feature = "future"))]
+    RemovalNotifier,
 }
 
 impl PoolName {
@@ -19,6 +21,8 @@ impl PoolName {
             PoolName::Housekeeper => "moka-housekeeper-{}",
             #[cfg(any(feature = "sync", feature = "future"))]
             PoolName::Invalidator => "moka-invalidator-{}",
+            #[cfg(any(feature = "sync", feature = "future"))]
+            PoolName::RemovalNotifier => "moka-notifier-{}",
         }
     }
 }

src/common/error.rs

@@ -1,5 +1,5 @@
 /// The error type for the functionalities around
-/// [`Cache#invalidate_entries_if`][invalidate-if] method.
+/// [`Cache::invalidate_entries_if`][invalidate-if] method.
 ///
 /// [invalidate-if]: ./sync/struct.Cache.html#method.invalidate_entries_if
 #[derive(thiserror::Error, Debug)]

src/future.rs

@@ -15,7 +15,7 @@ pub use {
 };
 
 /// The type of the unique ID to identify a predicate used by
-/// [`Cache#invalidate_entries_if`][invalidate-if] method.
+/// [`Cache::invalidate_entries_if`][invalidate-if] method.
 ///
 /// A `PredicateId` is a `String` of UUID (version 4).
 ///

src/future/builder.rs

@@ -1,5 +1,8 @@
 use super::Cache;
-use crate::common::{builder_utils, concurrent::Weigher};
+use crate::{
+    common::{builder_utils, concurrent::Weigher},
+    notification::{self, DeliveryMode, EvictionListener, RemovalCause},
+};
 
 use std::{
     collections::hash_map::RandomState,
@@ -13,7 +16,7 @@ use std::{
 ///
 /// [cache-struct]: ./struct.Cache.html
 ///
-/// # Examples
+/// # Example: Expirations
 ///
 /// ```rust
 /// // Cargo.toml
@@ -51,9 +54,12 @@ use std::{
 ///
 #[must_use]
 pub struct CacheBuilder<K, V, C> {
+    name: Option<String>,
     max_capacity: Option<u64>,
     initial_capacity: Option<usize>,
     weigher: Option<Weigher<K, V>>,
+    eviction_listener: Option<EvictionListener<K, V>>,
+    eviction_listener_conf: Option<notification::Configuration>,
     time_to_live: Option<Duration>,
     time_to_idle: Option<Duration>,
     invalidator_enabled: bool,
@@ -67,9 +73,12 @@ where
 {
     fn default() -> Self {
         Self {
+            name: None,
             max_capacity: None,
             initial_capacity: None,
             weigher: None,
+            eviction_listener: None,
+            eviction_listener_conf: None,
             time_to_live: None,
             time_to_idle: None,
             invalidator_enabled: false,
@@ -103,10 +112,13 @@ where
         let build_hasher = RandomState::default();
         builder_utils::ensure_expirations_or_panic(self.time_to_live, self.time_to_idle);
         Cache::with_everything(
+            self.name,
             self.max_capacity,
             self.initial_capacity,
             build_hasher,
             self.weigher,
+            self.eviction_listener,
+            self.eviction_listener_conf,
             self.time_to_live,
             self.time_to_idle,
             self.invalidator_enabled,
@@ -126,10 +138,13 @@ where
     {
         builder_utils::ensure_expirations_or_panic(self.time_to_live, self.time_to_idle);
         Cache::with_everything(
+            self.name,
             self.max_capacity,
             self.initial_capacity,
             hasher,
             self.weigher,
+            self.eviction_listener,
+            self.eviction_listener_conf,
             self.time_to_live,
             self.time_to_idle,
             self.invalidator_enabled,
@@ -138,6 +153,15 @@ where
 }
 
 impl<K, V, C> CacheBuilder<K, V, C> {
+    /// Sets the name of the cache. Currently the name is used for identification
+    /// only in logging messages.
+    pub fn name(self, name: &str) -> Self {
+        Self {
+            name: Some(name.to_string()),
+            ..self
+        }
+    }
+
     /// Sets the max capacity of the cache.
     pub fn max_capacity(self, max_capacity: u64) -> Self {
         Self {
@@ -154,7 +178,7 @@ impl<K, V, C> CacheBuilder<K, V, C> {
         }
     }
 
-    /// Sets the weigher closure of the cache.
+    /// Sets the weigher closure to the cache.
     ///
     /// The closure should take `&K` and `&V` as the arguments and returns a `u32`
     /// representing the relative size of the entry.
@@ -165,6 +189,35 @@ impl<K, V, C> CacheBuilder<K, V, C> {
         }
     }
 
+    /// Sets the eviction listener closure to the cache.
+    ///
+    /// The closure should take `Arc<K>`, `V` and [`RemovalCause`][removal-cause] as
+    /// the arguments. The [queued delivery mode][queued-mode] is used for the
+    /// listener.
+    ///
+    /// # Panics
+    ///
+    /// It is very important to make the listener closure not to panic. Otherwise,
+    /// the cache will stop calling the listener after a panic. This is an intended
+    /// behavior because the cache cannot know whether is is memory safe or not to
+    /// call the panicked lister again.
+    ///
+    /// [removal-cause]: ../notification/enum.RemovalCause.html
+    /// [queued-mode]: ../notification/enum.DeliveryMode.html#variant.Queued
+    pub fn eviction_listener_with_queued_delivery_mode(
+        self,
+        listener: impl Fn(Arc<K>, V, RemovalCause) + Send + Sync + 'static,
+    ) -> Self {
+        let conf = notification::Configuration::builder()
+            .delivery_mode(DeliveryMode::Queued)
+            .build();
+        Self {
+            eviction_listener: Some(Arc::new(listener)),
+            eviction_listener_conf: Some(conf),
+            ..self
+        }
+    }
+
     /// Sets the time to live of the cache.
     ///
     /// A cached entry will be expired after the specified duration past from