Added test cases and scheduler dependency

Author: vladvildanov

redis/client.py

@@ -15,7 +15,6 @@
     _RedisCallbacksRESP3,
     bool_ok,
 )
-from redis.cache import CacheMixin
 from redis.commands import (
     CoreCommands,
     RedisModuleCommands,
@@ -86,7 +85,7 @@ class AbstractRedis:
     pass
 
 
-class Redis(RedisModuleCommands, CoreCommands, SentinelCommands, CacheMixin):
+class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     """
     Implementation of the Redis protocol.
 

redis/connection.py

@@ -11,6 +11,8 @@
 from time import time
 from typing import Any, Callable, List, Optional, Type, Union
 from urllib.parse import parse_qs, unquote, urlparse
+
+from apscheduler.schedulers.background import BackgroundScheduler
 from cachetools import TTLCache, Cache, LRUCache
 from cachetools.keys import hashkey
 from redis.cache import CacheConfiguration
@@ -788,15 +790,21 @@ def send_command(self, *args, **kwargs):
         if self._cache.get(self._current_command_hash):
             return
 
-        # Send command over socket only if it's read-only command that not yet cached.
+        # Set temporary entry as a status to prevent race condition from another connection.
+        self._cache[self._current_command_hash] = "caching-in-progress"
+
+        # Send command over socket only if it's allowed read-only command that not yet cached.
         self._conn.send_command(*args, **kwargs)
 
     def can_read(self, timeout=0):
         return self._conn.can_read(timeout)
 
     def read_response(self, disable_decoding=False, *, disconnect_on_error=True, push_request=False):
-        # Check if command response exists in a cache.
-        if self._current_command_hash in self._cache:
+        # Check if command response exists in a cache and it's not in progress.
+        if (
+                self._current_command_hash in self._cache
+                and self._cache[self._current_command_hash] != "caching-in-progress"
+        ):
             return self._cache[self._current_command_hash]
 
         response = self._conn.read_response(
@@ -805,8 +813,12 @@ def read_response(self, disable_decoding=False, *, disconnect_on_error=True, pus
             push_request=push_request
         )
 
-        # Check if command that was sent is write command to prevent caching of write replies.
-        if response is None or self._current_command_hash is None:
+        # If response is None prevent from caching and remove temporary cache entry.
+        if response is None:
+            self._cache.pop(self._current_command_hash)
+            return response
+        # Prevent not-allowed command from caching.
+        elif self._current_command_hash is None:
             return response
 
         # Create separate mapping for keys or add current response to associated keys.
@@ -817,7 +829,12 @@ def read_response(self, disable_decoding=False, *, disconnect_on_error=True, pus
             else:
                 self._keys_mapping[key] = [self._current_command_hash]
 
-        self._cache[self._current_command_hash] = response
+        cache_entry = self._cache.get(self._current_command_hash, None)
+
+        # Cache only responses that still valid and wasn't invalidated by another connection in meantime
+        if cache_entry is not None:
+            self._cache[self._current_command_hash] = response
+
         return response
 
     def pack_command(self, *args):
@@ -1218,22 +1235,28 @@ def __init__(
         self.max_connections = max_connections
         self._cache = None
         self._cache_conf = None
+        self._scheduler = None
 
         if connection_kwargs.get("use_cache"):
             if connection_kwargs.get("protocol") not in [3, "3"]:
                 raise RedisError("Client caching is only supported with RESP version 3")
 
             self._cache_conf = CacheConfiguration(**self.connection_kwargs)
 
-            if self.connection_kwargs.get("cache"):
-                self._cache = self.connection_kwargs.get("cache")
+            cache = self.connection_kwargs.get("cache")
+            if cache is not None:
+                self._cache = cache
             else:
                 self._cache = TTLCache(self.connection_kwargs["cache_size"], self.connection_kwargs["cache_ttl"])
 
-            connection_kwargs.pop("use_cache", None)
-            connection_kwargs.pop("cache_size", None)
-            connection_kwargs.pop("cache_ttl", None)
-            connection_kwargs.pop("cache", None)
+            # self.scheduler = BackgroundScheduler()
+            # self.scheduler.add_job(self._perform_health_check, "interval", seconds=2)
+            # self.scheduler.start()
+
+        connection_kwargs.pop("use_cache", None)
+        connection_kwargs.pop("cache_size", None)
+        connection_kwargs.pop("cache_ttl", None)
+        connection_kwargs.pop("cache", None)
 
         # a lock to protect the critical section in _checkpid().
         # this lock is acquired when the process id changes, such as
@@ -1246,6 +1269,10 @@ 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__}"
@@ -1432,6 +1459,16 @@ def set_retry(self, retry: "Retry") -> None:
         for conn in self._in_use_connections:
             conn.retry = retry
 
+    def _perform_health_check(self) -> None:
+        self._checkpid()
+        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):
     """

requirements.txt

@@ -1,2 +1,3 @@
 async-timeout>=4.0.3
 cachetools
+apscheduler

tests/test_cache.py

@@ -1,35 +1,190 @@
 import time
 
+import pytest
+from cachetools import TTLCache, LRUCache, LFUCache
+
+import redis
 from redis import Redis, RedisCluster
+from redis.utils import HIREDIS_AVAILABLE
+from tests.conftest import _get_client
+
+
+@pytest.fixture()
+def r(request):
+    use_cache = request.param.get("use_cache", False)
+    cache = request.param.get("cache")
+    kwargs = request.param.get("kwargs", {})
+    protocol = request.param.get("protocol", 3)
+    single_connection_client = request.param.get("single_connection_client", False)
+    with _get_client(
+            redis.Redis,
+            request,
+            protocol=protocol,
+            single_connection_client=single_connection_client,
+            use_cache=use_cache,
+            cache=cache,
+            **kwargs,
+    ) as client:
+        yield client, cache
+
+
+@pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
+class TestCache:
+    @pytest.mark.parametrize("r", [{"cache": TTLCache(128, 300), "use_cache": True}], indirect=True)
+    @pytest.mark.onlynoncluster
+    def test_get_from_cache(self, r, r2):
+        r, cache = r
+        # add key to redis
+        r.set("foo", "bar")
+        # get key from redis and save in local cache
+        assert r.get("foo") == b"bar"
+        # get key from local cache
+        assert cache.get(("GET", "foo")) == b"bar"
+        # change key in redis (cause invalidation)
+        r2.set("foo", "barbar")
+        # Retrieves a new value from server and cache it
+        assert r.get("foo") == b"barbar"
+        # Make sure that new value was cached
+        assert cache.get(("GET", "foo")) == b"barbar"
+
+    @pytest.mark.parametrize(
+        "r",
+        [{"cache": LRUCache(3), "use_cache": True}],
+        indirect=True,
+    )
+    def test_cache_lru_eviction(self, r):
+        r, cache = r
+        # add 3 keys to redis
+        r.set("foo", "bar")
+        r.set("foo2", "bar2")
+        r.set("foo3", "bar3")
+        # get 3 keys from redis and save in local cache
+        assert r.get("foo") == b"bar"
+        assert r.get("foo2") == b"bar2"
+        assert r.get("foo3") == b"bar3"
+        # get the 3 keys from local cache
+        assert cache.get(("GET", "foo")) == b"bar"
+        assert cache.get(("GET", "foo2")) == b"bar2"
+        assert cache.get(("GET", "foo3")) == b"bar3"
+        # add 1 more key to redis (exceed the max size)
+        r.set("foo4", "bar4")
+        assert r.get("foo4") == b"bar4"
+        # the first key is not in the local cache anymore
+        assert cache.get(("GET", "foo")) is None
+
+    @pytest.mark.parametrize("r", [{"cache": TTLCache(maxsize=128, ttl=1), "use_cache": True}], indirect=True)
+    def test_cache_ttl(self, r, cache):
+        r, cache = r
+        # add key to redis
+        r.set("foo", "bar")
+        # get key from redis and save in local cache
+        assert r.get("foo") == b"bar"
+        # get key from local cache
+        assert cache.get(("GET", "foo")) == b"bar"
+        # wait for the key to expire
+        time.sleep(1)
+        # the key is not in the local cache anymore
+        assert cache.get(("GET", "foo")) is None
 
+    @pytest.mark.parametrize(
+        "r",
+        [{"cache": LFUCache(3), "use_cache": True}],
+        indirect=True,
+    )
+    def test_cache_lfu_eviction(self, r, cache):
+        r, cache = r
+        # add 3 keys to redis
+        r.set("foo", "bar")
+        r.set("foo2", "bar2")
+        r.set("foo3", "bar3")
+        # get 3 keys from redis and save in local cache
+        assert r.get("foo") == b"bar"
+        assert r.get("foo2") == b"bar2"
+        assert r.get("foo3") == b"bar3"
+        # change the order of the keys in the cache
+        assert cache.get(("GET", "foo")) == b"bar"
+        assert cache.get(("GET", "foo")) == b"bar"
+        assert cache.get(("GET", "foo3")) == b"bar3"
+        # add 1 more key to redis (exceed the max size)
+        r.set("foo4", "bar4")
+        assert r.get("foo4") == b"bar4"
+        # test the eviction policy
+        assert cache.currsize == 3
+        assert cache.get(("GET", "foo")) == b"bar"
+        assert cache.get(("GET", "foo2")) is None
 
-def test_standalone_cached_get_and_set():
-    r = Redis(use_cache=True, protocol=3)
-    assert r.set("key", 5)
-    assert r.get("key") == b"5"
+    @pytest.mark.parametrize(
+        "r",
+        [{"cache": LRUCache(maxsize=128), "use_cache": True}],
+        indirect=True,
+    )
+    def test_cache_ignore_not_allowed_command(self, r):
+        r, cache = r
+        # add fields to hash
+        assert r.hset("foo", "bar", "baz")
+        # get random field
+        assert r.hrandfield("foo") == b"bar"
+        assert cache.get(("HRANDFIELD", "foo")) is None
 
-    r2 = Redis(protocol=3)
-    r2.set("key", "foo")
+    @pytest.mark.parametrize(
+        "r",
+        [{"cache": LRUCache(maxsize=128), "use_cache": True}],
+        indirect=True,
+    )
+    def test_cache_invalidate_all_related_responses(self, r, cache):
+        r, cache = r
+        # Add keys
+        assert r.set("foo", "bar")
+        assert r.set("bar", "foo")
 
-    time.sleep(0.5)
+        # Make sure that replies was cached
+        assert r.mget("foo", "bar") == [b"bar", b"foo"]
+        assert cache.get(("MGET", "foo", "bar")) == [b"bar", b"foo"]
 
-    after_invalidation = r.get("key")
-    print(f'after invalidation {after_invalidation}')
-    assert after_invalidation == b"foo"
+        # Invalidate one of the keys and make sure that all associated cached entries was removed
+        assert r.set("foo", "baz")
+        assert r.get("foo") == b"baz"
+        assert cache.get(("MGET", "foo", "bar")) is None
+        assert cache.get(("GET", "foo")) == b"baz"
 
+    @pytest.mark.parametrize(
+        "r",
+        [{"cache": LRUCache(maxsize=128), "use_cache": True}],
+        indirect=True,
+    )
+    def test_cache_flushed_on_server_flush(self, r, cache):
+        r, cache = r
+        # Add keys
+        assert r.set("foo", "bar")
+        assert r.set("bar", "foo")
+        assert r.set("baz", "bar")
 
-def test_cluster_cached_get_and_set():
-    cluster_url = "redis://localhost:16379/0"
+        # Make sure that replies was cached
+        assert r.get("foo") == b"bar"
+        assert r.get("bar") == b"foo"
+        assert r.get("baz") == b"bar"
+        assert cache.get(("GET", "foo")) == b"bar"
+        assert cache.get(("GET", "bar")) == b"foo"
+        assert cache.get(("GET", "baz")) == b"bar"
 
-    r = RedisCluster.from_url(cluster_url, use_cache=True, protocol=3)
-    assert r.set("key", 5)
-    assert r.get("key") == b"5"
+        # Flush server and trying to access cached entry
+        assert r.flushall()
+        assert r.get("foo") is None
+        assert cache.currsize == 0
 
-    r2 = RedisCluster.from_url(cluster_url, use_cache=True, protocol=3)
-    r2.set("key", "foo")
 
-    time.sleep(0.5)
-    
-    after_invalidation = r.get("key")
-    print(f'after invalidation {after_invalidation}')
-    assert after_invalidation == b"foo"
+# def test_cluster_cached_get_and_set():
+#     cluster_url = "redis://localhost:16379/0"
+#
+#     r = RedisCluster.from_url(cluster_url, use_cache=True, protocol=3)
+#     assert r.set("key", 5)
+#     assert r.get("key") == b"5"
+#
+#     r2 = RedisCluster.from_url(cluster_url, use_cache=True, protocol=3)
+#     r2.set("key", "foo")
+#
+#     time.sleep(0.5)
+#
+#     after_invalidation = r.get("key")
+#     print(f'after invalidation {after_invalidation}')
+#     assert after_invalidation == b"foo"