Make run_pending_tasks to evict more entries from the cache at once

Add unit tests.

src/common.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 pub(crate) mod builder_utils;
 pub(crate) mod concurrent;
 pub(crate) mod deque;
@@ -10,6 +12,11 @@ pub(crate) mod timer_wheel;
 #[cfg(test)]
 pub(crate) mod test_utils;
 
+use self::concurrent::constants::{
+    DEFAULT_EVICTION_BATCH_SIZE, DEFAULT_MAINTENANCE_TASK_TIMEOUT_MILLIS,
+    DEFAULT_MAX_LOG_SYNC_REPEATS,
+};
+
 // Note: `CacheRegion` cannot have more than four enum variants. This is because
 // `crate::{sync,unsync}::DeqNodes` uses a `tagptr::TagNonNull<DeqNode<T>, 2>`
 // pointer, where the 2-bit tag is `CacheRegion`.
@@ -56,6 +63,54 @@ impl PartialEq<usize> for CacheRegion {
     }
 }
 
+#[derive(Clone, Debug)]
+pub(crate) struct HousekeeperConfig {
+    /// The timeout duration for the `run_pending_tasks` method. This is a safe-guard
+    /// to prevent cache read/write operations (that may call `run_pending_tasks`
+    /// internally) from being blocked for a long time when the user wrote a slow
+    /// eviction listener closure.
+    ///
+    /// Used only when the eviction listener closure is set for the cache instance.
+    ///
+    /// Default: `DEFAULT_MAINTENANCE_TASK_TIMEOUT_MILLIS`
+    pub(crate) maintenance_task_timeout: Duration,
+    /// The maximum repeat count for receiving operation logs from the read and write
+    /// log channels. Default: `MAX_LOG_SYNC_REPEATS`.
+    pub(crate) max_log_sync_repeats: usize,
+    /// The batch size of entries to be processed by each internal eviction method.
+    /// Default: `EVICTION_BATCH_SIZE`.
+    pub(crate) eviction_batch_size: usize,
+}
+
+impl Default for HousekeeperConfig {
+    fn default() -> Self {
+        Self {
+            maintenance_task_timeout: Duration::from_millis(
+                DEFAULT_MAINTENANCE_TASK_TIMEOUT_MILLIS,
+            ),
+            max_log_sync_repeats: DEFAULT_MAX_LOG_SYNC_REPEATS,
+            eviction_batch_size: DEFAULT_EVICTION_BATCH_SIZE,
+        }
+    }
+}
+
+impl HousekeeperConfig {
+    #[cfg(test)]
+    pub(crate) fn new(
+        maintenance_task_timeout: Option<Duration>,
+        max_log_sync_repeats: Option<usize>,
+        eviction_batch_size: Option<usize>,
+    ) -> Self {
+        Self {
+            maintenance_task_timeout: maintenance_task_timeout.unwrap_or(Duration::from_millis(
+                DEFAULT_MAINTENANCE_TASK_TIMEOUT_MILLIS,
+            )),
+            max_log_sync_repeats: max_log_sync_repeats.unwrap_or(DEFAULT_MAX_LOG_SYNC_REPEATS),
+            eviction_batch_size: eviction_batch_size.unwrap_or(DEFAULT_EVICTION_BATCH_SIZE),
+        }
+    }
+}
+
 // Ensures the value fits in a range of `128u32..=u32::MAX`.
 pub(crate) fn sketch_capacity(max_capacity: u64) -> u32 {
     max_capacity.try_into().unwrap_or(u32::MAX).max(128)

src/common/concurrent/constants.rs

@@ -1,19 +1,23 @@
-pub(crate) const MAX_SYNC_REPEATS: usize = 4;
-pub(crate) const PERIODICAL_SYNC_INITIAL_DELAY_MILLIS: u64 = 300;
+pub(crate) const DEFAULT_MAX_LOG_SYNC_REPEATS: usize = 4;
+pub(crate) const LOG_SYNC_INTERVAL_MILLIS: u64 = 300;
 
 pub(crate) const READ_LOG_FLUSH_POINT: usize = 64;
-pub(crate) const READ_LOG_SIZE: usize = READ_LOG_FLUSH_POINT * (MAX_SYNC_REPEATS + 2); // 384
-
 pub(crate) const WRITE_LOG_FLUSH_POINT: usize = 64;
-pub(crate) const WRITE_LOG_SIZE: usize = WRITE_LOG_FLUSH_POINT * (MAX_SYNC_REPEATS + 2); // 384
+
+// 384 elements
+pub(crate) const READ_LOG_CH_SIZE: usize =
+    READ_LOG_FLUSH_POINT * (DEFAULT_MAX_LOG_SYNC_REPEATS + 2);
+
+// 384 elements
+pub(crate) const WRITE_LOG_CH_SIZE: usize =
+    WRITE_LOG_FLUSH_POINT * (DEFAULT_MAX_LOG_SYNC_REPEATS + 2);
 
 // TODO: Calculate the batch size based on the number of entries in the cache (or an
 // estimated number of entries to evict)
-pub(crate) const EVICTION_BATCH_SIZE: usize = WRITE_LOG_SIZE;
-pub(crate) const INVALIDATION_BATCH_SIZE: usize = WRITE_LOG_SIZE;
+pub(crate) const DEFAULT_EVICTION_BATCH_SIZE: usize = WRITE_LOG_CH_SIZE;
 
 /// The default timeout duration for the `run_pending_tasks` method.
-pub(crate) const DEFAULT_RUN_PENDING_TASKS_TIMEOUT_MILLIS: u64 = 100;
+pub(crate) const DEFAULT_MAINTENANCE_TASK_TIMEOUT_MILLIS: u64 = 100;
 
 #[cfg(feature = "sync")]
 pub(crate) const WRITE_RETRY_INTERVAL_MICROS: u64 = 50;

src/common/concurrent/housekeeper.rs

@@ -1,13 +1,11 @@
-use super::constants::{
-    DEFAULT_RUN_PENDING_TASKS_TIMEOUT_MILLIS, MAX_SYNC_REPEATS,
-    PERIODICAL_SYNC_INITIAL_DELAY_MILLIS,
-};
+use super::constants::LOG_SYNC_INTERVAL_MILLIS;
 
 use super::{
     atomic_time::AtomicInstant,
     constants::{READ_LOG_FLUSH_POINT, WRITE_LOG_FLUSH_POINT},
 };
 use crate::common::time::{CheckedTimeOps, Instant};
+use crate::common::HousekeeperConfig;
 
 use parking_lot::{Mutex, MutexGuard};
 use std::{
@@ -17,19 +15,24 @@ use std::{
 
 pub(crate) trait InnerSync {
     /// Runs the pending tasks. Returns `true` if there are more entries to evict.
-    fn run_pending_tasks(&self, max_sync_repeats: usize, timeout: Option<Duration>) -> bool;
+    fn run_pending_tasks(
+        &self,
+        timeout: Option<Duration>,
+        max_log_sync_repeats: usize,
+        eviction_batch_size: usize,
+    ) -> bool;
 
     fn now(&self) -> Instant;
 }
 
 pub(crate) struct Housekeeper {
     run_lock: Mutex<()>,
     run_after: AtomicInstant,
-    /// A flag to indicate if the last `run_pending_tasks` call left more entries to
-    /// evict.
+    /// A flag to indicate if the last call on `run_pending_tasks` method left some
+    /// entries to evict.
     ///
     /// Used only when the eviction listener closure is set for this cache instance
-    /// because, if not, `run_pending_tasks` will never leave more entries to evict.
+    /// because, if not, `run_pending_tasks` will never leave entries to evict.
     more_entries_to_evict: Option<AtomicBool>,
     /// The timeout duration for the `run_pending_tasks` method. This is a safe-guard
     /// to prevent cache read/write operations (that may call `run_pending_tasks`
@@ -38,17 +41,21 @@ pub(crate) struct Housekeeper {
     ///
     /// Used only when the eviction listener closure is set for this cache instance.
     maintenance_task_timeout: Option<Duration>,
+    /// The maximum repeat count for receiving operation logs from the read and write
+    /// log channels. Default: `MAX_LOG_SYNC_REPEATS`.
+    max_log_sync_repeats: usize,
+    /// The batch size of entries to be processed by each internal eviction method.
+    /// Default: `EVICTION_BATCH_SIZE`.
+    eviction_batch_size: usize,
     auto_run_enabled: AtomicBool,
 }
 
 impl Housekeeper {
-    pub(crate) fn new(is_eviction_listener_enabled: bool) -> Self {
+    pub(crate) fn new(is_eviction_listener_enabled: bool, config: HousekeeperConfig) -> Self {
         let (more_entries_to_evict, maintenance_task_timeout) = if is_eviction_listener_enabled {
             (
                 Some(AtomicBool::new(false)),
-                Some(Duration::from_millis(
-                    DEFAULT_RUN_PENDING_TASKS_TIMEOUT_MILLIS,
-                )),
+                Some(config.maintenance_task_timeout),
             )
         } else {
             (None, None)
@@ -59,6 +66,8 @@ impl Housekeeper {
             run_after: AtomicInstant::new(Self::sync_after(Instant::now())),
             more_entries_to_evict,
             maintenance_task_timeout,
+            max_log_sync_repeats: config.max_log_sync_repeats,
+            eviction_batch_size: config.eviction_batch_size,
             auto_run_enabled: AtomicBool::new(true),
         }
     }
@@ -109,12 +118,14 @@ impl Housekeeper {
         let now = cache.now();
         self.run_after.set_instant(Self::sync_after(now));
         let timeout = self.maintenance_task_timeout;
-        let more_to_evict = cache.run_pending_tasks(MAX_SYNC_REPEATS, timeout);
+        let repeats = self.max_log_sync_repeats;
+        let batch_size = self.eviction_batch_size;
+        let more_to_evict = cache.run_pending_tasks(timeout, repeats, batch_size);
         self.set_more_entries_to_evict(more_to_evict);
     }
 
     fn sync_after(now: Instant) -> Instant {
-        let dur = Duration::from_millis(PERIODICAL_SYNC_INITIAL_DELAY_MILLIS);
+        let dur = Duration::from_millis(LOG_SYNC_INTERVAL_MILLIS);
         let ts = now.checked_add(dur);
         // Assuming that `now` is current wall clock time, this should never fail at
         // least next millions of years.