Updated NodesManager to create shared cache between all nodes

Author: vladvildanov

redis/cluster.py

@@ -9,7 +9,7 @@
 from redis._parsers import CommandsParser, Encoder
 from redis._parsers.helpers import parse_scan
 from redis.backoff import default_backoff
-from redis.cache import CacheConfig, CacheInterface
+from redis.cache import CacheConfig, CacheInterface, CacheFactoryInterface, CacheFactory
 from redis.client import CaseInsensitiveDict, PubSub, Redis
 from redis.commands import READ_COMMANDS, RedisClusterCommands
 from redis.commands.helpers import list_or_args
@@ -1327,6 +1327,7 @@ def __init__(
         address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
         cache: Optional[CacheInterface] = None,
         cache_config: Optional[CacheConfig] = None,
+        cache_factory: Optional[CacheFactoryInterface] = None,
         **kwargs,
     ):
         self.nodes_cache = {}
@@ -1339,8 +1340,9 @@ def __init__(
         self._dynamic_startup_nodes = dynamic_startup_nodes
         self.connection_pool_class = connection_pool_class
         self.address_remap = address_remap
-        self.cache = cache
-        self.cache_config = cache_config
+        self._cache = cache
+        self._cache_config = cache_config
+        self._cache_factory = cache_factory
         self._moved_exception = None
         self.connection_kwargs = kwargs
         self.read_load_balancer = LoadBalancer()
@@ -1484,15 +1486,13 @@ def create_redis_node(self, host, port, **kwargs):
             # Create a redis node with a costumed connection pool
             kwargs.update({"host": host})
             kwargs.update({"port": port})
-            kwargs.update({"cache": self.cache})
-            kwargs.update({"cache_config": self.cache_config})
+            kwargs.update({"cache": self._cache})
             r = Redis(connection_pool=self.connection_pool_class(**kwargs))
         else:
             r = Redis(
                 host=host,
                 port=port,
-                cache=self.cache,
-                cache_config=self.cache_config,
+                cache=self._cache,
                 **kwargs,
             )
         return r
@@ -1624,6 +1624,12 @@ def initialize(self):
                 f"one reachable node: {str(exception)}"
             ) from exception
 
+        if self._cache is None and self._cache_config is not None:
+            if self._cache_factory is None:
+                self._cache = CacheFactory(self._cache_config).get_cache()
+            else:
+                self._cache = self._cache_factory.get_cache()
+
         # Create Redis connections to all nodes
         self.create_redis_connections(list(tmp_nodes_cache.values()))
 

redis/connection.py

@@ -1468,13 +1468,7 @@ def disconnect(self, inuse_connections: bool = True) -> None:
             for connection in connections:
                 connection.disconnect()
 
-        # Send an event to stop scheduled healthcheck execution.
-        if self._hc_cancel_event is not None and not self._hc_cancel_event.is_set():
-            self._hc_cancel_event.set()
-
-        # Joins healthcheck thread on disconnect.
-        if self._hc_thread is not None and not self._hc_thread.is_alive():
-            self._hc_thread.join()
+            self.stop_scheduled_healthcheck()
 
     def close(self) -> None:
         """Close the pool, disconnecting all connections"""
@@ -1496,6 +1490,15 @@ def run_scheduled_healthcheck(self) -> None:
                 self._perform_health_check, hc_interval, self._hc_cancel_event
             )
 
+    def stop_scheduled_healthcheck(self) -> None:
+        # Send an event to stop scheduled healthcheck execution.
+        if self._hc_cancel_event is not None and not self._hc_cancel_event.is_set():
+            self._hc_cancel_event.set()
+
+        # Joins healthcheck thread on disconnect.
+        if self._hc_thread is not None and not self._hc_thread.is_alive():
+            self._hc_thread.join()
+
     def _perform_health_check(self, done: threading.Event) -> None:
         self._checkpid()
         with self._lock:
@@ -1670,10 +1673,4 @@ def disconnect(self):
         for connection in self._connections:
             connection.disconnect()
 
-        # Send an event to stop scheduled healthcheck execution.
-        if self._hc_cancel_event is not None and not self._hc_cancel_event.is_set():
-            self._hc_cancel_event.set()
-
-        # Joins healthcheck thread on disconnect.
-        if self._hc_thread is not None and not self._hc_thread.is_alive():
-            self._hc_thread.join()
+        self.stop_scheduled_healthcheck()

tests/test_cache.py

@@ -374,6 +374,8 @@ def test_get_from_cache(self, r):
             cache.get(CacheKey(command="GET", redis_keys=("foo",))).cache_value
             == b"barbar"
         )
+        # Make sure that cache is shared between nodes.
+        assert cache == r.nodes_manager.get_node_from_slot(1).redis_connection.get_cache()
 
     @pytest.mark.parametrize(
         "r",