Added support for RedisCluster and multi-threaded test cases

Author: vladvildanov

redis/cluster.py

@@ -11,7 +11,6 @@
 from redis._parsers import CommandsParser, Encoder
 from redis._parsers.helpers import parse_scan
 from redis.backoff import default_backoff
-from redis.cache import CacheMixin
 from redis.client import CaseInsensitiveDict, PubSub, Redis
 from redis.commands import READ_COMMANDS, RedisClusterCommands
 from redis.commands.helpers import list_or_args
@@ -446,7 +445,7 @@ def replace_default_node(self, target_node: "ClusterNode" = None) -> None:
                     self.nodes_manager.default_node = random.choice(replicas)
 
 
-class RedisCluster(AbstractRedisCluster, RedisClusterCommands, CacheMixin):
+class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
     @classmethod
     def from_url(cls, url, **kwargs):
         """
@@ -504,7 +503,10 @@ def __init__(
         dynamic_startup_nodes: bool = True,
         url: Optional[str] = None,
         address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
-        use_cache: Optional[bool] = False,
+        use_cache: bool = False,
+        cache: Optional[Cache] = None,
+        cache_size: int = 128,
+        cache_ttl: int = 300,
         **kwargs,
     ):
         """
@@ -628,6 +630,10 @@ def __init__(
             kwargs.get("encoding_errors", "strict"),
             kwargs.get("decode_responses", False),
         )
+        protocol = kwargs.get("protocol", None)
+        if use_cache and protocol not in [3, "3"]:
+            raise RedisError("Client caching is only supported with RESP version 3")
+
         self.cluster_error_retry_attempts = cluster_error_retry_attempts
         self.command_flags = self.__class__.COMMAND_FLAGS.copy()
         self.node_flags = self.__class__.NODE_FLAGS.copy()
@@ -641,6 +647,9 @@ def __init__(
             dynamic_startup_nodes=dynamic_startup_nodes,
             address_remap=address_remap,
             use_cache=use_cache,
+            cache=cache,
+            cache_size=cache_size,
+            cache_ttl=cache_ttl,
             **kwargs,
         )
 
@@ -649,11 +658,6 @@ def __init__(
         )
         self.result_callbacks = CaseInsensitiveDict(self.__class__.RESULT_CALLBACKS)
 
-        protocol = kwargs.get("protocol", None)
-        if use_cache and protocol not in [3, "3"]:
-            raise RedisError("Client caching is only supported with RESP version 3")
-        CacheMixin.__init__(self, use_cache, None)
-
         self.commands_parser = CommandsParser(self)
         self._lock = threading.Lock()
 
@@ -1057,8 +1061,6 @@ def _parse_target_nodes(self, target_nodes):
         return nodes
 
     def execute_command(self, *args, **options):
-        if self.use_cache:
-            return self.cached_call(self._execute_command, *args, **options)
         return self._internal_execute_command(*args, **options)
 
     def _internal_execute_command(self, *args, **kwargs):
@@ -1163,7 +1165,7 @@ def _execute_command(self, target_node, *args, **kwargs):
                     connection.send_command("ASKING")
                     redis_node.parse_response(connection, "ASKING", **kwargs)
                     asking = False
-                connection.send_command(*args)
+                connection.send_command(*args, **kwargs)
                 response = redis_node.parse_response(connection, command, **kwargs)
                 if command in self.cluster_response_callbacks:
                     response = self.cluster_response_callbacks[command](
@@ -1317,7 +1319,7 @@ def reset(self) -> None:
         self.primary_to_idx.clear()
 
 
-class NodesManager(CacheMixin):
+class NodesManager():
     def __init__(
         self,
         startup_nodes,
@@ -1327,8 +1329,10 @@ def __init__(
         dynamic_startup_nodes=True,
         connection_pool_class=ConnectionPool,
         address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
-        use_cache: Optional[bool] = False,
+        use_cache: bool = False,
         cache: Optional[Cache] = None,
+        cache_size: int = 128,
+        cache_ttl: int = 300,
         **kwargs,
     ):
         self.nodes_cache = {}
@@ -1342,13 +1346,15 @@ def __init__(
         self.connection_pool_class = connection_pool_class
         self.address_remap = address_remap
         self.use_cache = use_cache
+        self.cache = cache
+        self.cache_size = cache_size
+        self.cache_ttl = cache_ttl
         self._moved_exception = None
         self.connection_kwargs = kwargs
         self.read_load_balancer = LoadBalancer()
         if lock is None:
             lock = threading.Lock()
         self._lock = lock
-        CacheMixin.__init__(self, use_cache, None, cache)
         self.initialize()
 
     def get_node(self, host=None, port=None, node_name=None):
@@ -1486,9 +1492,21 @@ 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})
-            r = Redis(connection_pool=self.connection_pool_class(**kwargs), use_cache=self.use_cache, cache=self.cache)
+            kwargs.update({"use_cache": self.use_cache})
+            kwargs.update({"cache": self.cache})
+            kwargs.update({"cache_size": self.cache_size})
+            kwargs.update({"cache_ttl": self.cache_ttl})
+            r = Redis(connection_pool=self.connection_pool_class(**kwargs))
         else:
-            r = Redis(host=host, port=port, use_cache=self.use_cache, cache=self.cache, **kwargs)
+            r = Redis(
+                host=host,
+                port=port,
+                use_cache=self.use_cache,
+                cache=self.cache,
+                cache_size=self.cache_size,
+                cache_ttl=self.cache_ttl,
+                **kwargs,
+            )
         return r
 
     def _get_or_create_cluster_node(self, host, port, role, tmp_nodes_cache):

redis/connection.py

@@ -8,7 +8,7 @@
 from abc import abstractmethod
 from itertools import chain
 from queue import Empty, Full, LifoQueue
-from time import time
+from time import time, sleep
 from typing import Any, Callable, List, Optional, Type, Union
 from urllib.parse import parse_qs, unquote, urlparse
 
@@ -756,6 +756,7 @@ def on_connect(self):
         self._conn.on_connect()
 
     def disconnect(self, *args):
+        self._cache.clear()
         self._conn.disconnect(*args)
 
     def check_health(self):
@@ -1235,7 +1236,7 @@ def __init__(
         self.max_connections = max_connections
         self._cache = None
         self._cache_conf = None
-        self._scheduler = None
+        self.scheduler = None
 
         if connection_kwargs.get("use_cache"):
             if connection_kwargs.get("protocol") not in [3, "3"]:
@@ -1249,9 +1250,9 @@ def __init__(
             else:
                 self._cache = TTLCache(self.connection_kwargs["cache_size"], self.connection_kwargs["cache_ttl"])
 
-            # self.scheduler = BackgroundScheduler()
-            # self.scheduler.add_job(self._perform_health_check, "interval", seconds=2)
-            # self.scheduler.start()
+            self.scheduler = BackgroundScheduler()
+            self.scheduler.add_job(self._perform_health_check, "interval", seconds=2, id="cache_health_check")
+            self.scheduler.start()
 
         connection_kwargs.pop("use_cache", None)
         connection_kwargs.pop("cache_size", None)
@@ -1269,10 +1270,6 @@ def __init__(
         self._fork_lock = threading.Lock()
         self.reset()
 
-    def __del__(self):
-        if self._scheduler is not None:
-            self.scheduler.shutdown()
-
     def __repr__(self) -> (str, str):
         return (
             f"<{type(self).__module__}.{type(self).__name__}"
@@ -1464,10 +1461,8 @@ def _perform_health_check(self) -> None:
         with self._lock:
             while self._available_connections:
                 conn = self._available_connections.pop()
-                self._in_use_connections.add(conn)
                 conn.send_command('PING')
                 conn.read_response()
-                self.release(conn)
 
 
 class BlockingConnectionPool(ConnectionPool):

tests/conftest.py

@@ -16,7 +16,7 @@
 from redis.retry import Retry
 
 REDIS_INFO = {}
-default_redis_url = "redis://localhost:6379/0"
+default_redis_url = "redis://localhost:6372/0"
 default_protocol = "2"
 default_redismod_url = "redis://localhost:6479"