add background task

Signed-off-by: Shoham Elias <shohame@amazon.com>

Author: shohamazon

glide-core/cache/Cargo.toml

@@ -12,3 +12,5 @@ moka = { version = "0.12", features = ["sync"] }
 dashmap = "6"
 logger_core = { path = "../../logger_core" }
 once_cell = "1"
+parking_lot = "0.12"
+tokio = { version = "1", features = ["rt", "time", "macros"] }

glide-core/cache/src/lib.rs

@@ -1,16 +1,20 @@
 // Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
 use dashmap::DashMap;
-use logger_core::log_info;
+use logger_core::{log_debug, log_info};
 use moka::policy::EvictionPolicy as MokaEvictionPolicy;
 use moka::sync::Cache;
 use once_cell::sync::Lazy;
 use redis::Value;
 use std::sync::{Arc, Weak};
 use std::time::Duration;
+use tokio::task::JoinHandle;
 
 static CACHE_REGISTRY: Lazy<DashMap<String, Weak<Cache<Vec<u8>, Value>>>> =
     Lazy::new(|| DashMap::new());
 
+static HOUSEKEEPING_HANDLE: Lazy<parking_lot::Mutex<Option<JoinHandle<()>>>> =
+    Lazy::new(|| parking_lot::Mutex::new(None));
+
 /// Cache eviction policy
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EvictionPolicy {
@@ -96,6 +100,9 @@ pub fn get_or_create_cache(
     // Store weak reference in registry
     CACHE_REGISTRY.insert(cache_id.to_string(), Arc::downgrade(&cache));
 
+    // Start housekeeping task if this is the first cache
+    start_cache_housekeeping();
+
     cache
 }
 
@@ -174,14 +181,82 @@ fn calculate_value_additional_data(value: &Value) -> usize {
 /// Returns the total size of the cache entry (key + value) in bytes
 fn cache_entry_weigher(key: &Vec<u8>, value: &Value) -> u32 {
     let total_size = key.len() + calculate_value_size(value);
-    println!(
-        "Cache entry weigher: key_len = {}, value_size = {}",
-        key.len(),
-        calculate_value_size(value)
-    );
     total_size.try_into().unwrap_or(u32::MAX)
 }
 
+/// Periodically runs pending tasks for all live caches
+async fn periodic_cache_housekeeping(interval_duration: Duration) {
+    log_info(
+        "cache_housekeeping",
+        format!(
+            "Started cache housekeeping task (interval: {:?})",
+            interval_duration
+        ),
+    );
+
+    loop {
+        tokio::time::sleep(interval_duration).await;
+
+        let mut live_count = 0;
+        let mut dead_keys = Vec::new();
+
+        // Process all caches
+        for entry in CACHE_REGISTRY.iter() {
+            match entry.value().upgrade() {
+                Some(cache) => {
+                    cache.run_pending_tasks();
+                    live_count += 1;
+                }
+                None => {
+                    // Cache is dead, mark for removal
+                    dead_keys.push(entry.key().clone());
+                }
+            }
+        }
+
+        // Clean up dead cache entries
+        for key in dead_keys {
+            CACHE_REGISTRY.remove(&key);
+            log_debug(
+                "cache_housekeeping",
+                format!("Removed dead cache entry: {}", key),
+            );
+        }
+
+        // If no live caches remain, stop the housekeeping task
+        if live_count == 0 && CACHE_REGISTRY.is_empty() {
+            log_info(
+                "cache_housekeeping",
+                "No live caches remaining, stopping housekeeping task",
+            );
+            break;
+        }
+
+        log_debug(
+            "cache_housekeeping",
+            format!("Processed {} live caches", live_count),
+        );
+    }
+
+    log_info("cache_housekeeping", "Cache housekeeping task stopped");
+}
+
+/// Start the cache housekeeping background task
+pub fn start_cache_housekeeping() {
+    let mut handle_guard = HOUSEKEEPING_HANDLE.lock();
+
+    // Don't start if already running
+    if handle_guard.is_some() {
+        log_debug("cache_housekeeping", "Housekeeping task already running");
+        return;
+    }
+
+    let task = tokio::spawn(periodic_cache_housekeeping(Duration::from_millis(500)));
+    *handle_guard = Some(task);
+
+    log_info("cache_housekeeping", "Started cache housekeeping task");
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

python/glide-async/python/glide/__init__.py

@@ -158,6 +158,8 @@
     VectorFieldAttributesHnsw,
     VectorType,
     json_batch,
+    ClientSideCache,
+    EvictionPolicy,
 )
 
 from .async_commands import ft, glide_json
@@ -385,4 +387,7 @@ def __getattr__(self, name):
     "FtAggregateSortProperty",
     "FtProfileOptions",
     "QueryType",
+    # Cache
+    "ClientSideCache",
+    "EvictionPolicy",
 ]

python/glide-shared/glide_shared/__init__.py

@@ -168,6 +168,7 @@
     SlotKeyRoute,
     SlotType,
 )
+from .cache import ClientSideCache, EvictionPolicy
 
 __all__ = [
     # Client
@@ -327,4 +328,7 @@
     "FtAggregateSortProperty",
     "FtProfileOptions",
     "QueryType",
+    # Cache
+    "ClientSideCache",
+    "EvictionPolicy",
 ]

python/glide-shared/glide_shared/cache.py

@@ -78,6 +78,13 @@ def create(
         currently supported. This means cached values may become stale if updated on
         the server before the TTL expires.
 
+        **Note**: In order for 2 clients to share the same cache, they must be
+        created with the same `ClientSideCache` instance.
+        * Client's with different `ClientSideCache` instances will have separate caches,
+        even if the configurations are identical.
+        * Client's using different db's cannot share the same cache.
+        * Client's using different ACL users cannot share the same cache.
+
         Args:
             max_cache_kb (int): Maximum size of the cache in kilobytes (KB). This limits
                 the total memory used by cached keys and values. When this limit is reached,

python/glide-shared/glide_shared/config.py

@@ -355,6 +355,12 @@ class BaseClientConfiguration:
             If not set, connections are established immediately during client creation (equivalent to `False`).
         client_side_cache (Optional[ClientSideCache]): Configuration for client-side caching.
             See `ClientSideCache` for more information.
+            **Note**: In order for 2 clients to share the same cache, they must be
+                created with the same `ClientSideCache` instance.
+                * Client's with different `ClientSideCache` instances will have separate caches,
+                even if the configurations are identical.
+                * Client's using different db's cannot share the same cache.
+                * Client's using different ACL users cannot share the same cache.
     """
 
     def __init__(
@@ -600,6 +606,12 @@ class GlideClientConfiguration(BaseClientConfiguration):
             see `AdvancedGlideClientConfiguration`.
         client_side_cache (Optional[ClientSideCache]): Configuration for client-side caching.
             See `ClientSideCache` for more information.
+            **Note**: In order for 2 clients to share the same cache, they must be
+                created with the same `ClientSideCache` instance.
+                * Client's with different `ClientSideCache` instances will have separate caches,
+                even if the configurations are identical.
+                * Client's using different db's cannot share the same cache.
+                * Client's using different ACL users cannot share the same cache.
     """
 
     class PubSubChannelModes(IntEnum):
@@ -797,6 +809,12 @@ class GlideClusterClientConfiguration(BaseClientConfiguration):
             see `AdvancedGlideClusterClientConfiguration`.
         client_side_cache (Optional[ClientSideCache]): Configuration for client-side caching.
             See `ClientSideCache` for more information.
+            **Note**: In order for 2 clients to share the same cache, they must be
+                created with the same `ClientSideCache` instance.
+                * Client's with different `ClientSideCache` instances will have separate caches,
+                even if the configurations are identical.
+                * Client's using different db's cannot share the same cache.
+                * Client's using different ACL users cannot share the same cache.
 
     Note:
         Currently, the reconnection strategy in cluster mode is not configurable, and exponential backoff