Added custom background scheduler, added unit testing

Author: vladvildanov

dev_requirements.txt

@@ -1,6 +1,5 @@
 black==24.3.0
 cachetools>=5.5.0
-apscheduler
 click==8.0.4
 flake8-isort
 flake8

redis/connection.py

@@ -8,11 +8,10 @@
 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
 
-from apscheduler.schedulers.background import BackgroundScheduler
 from cachetools import LRUCache
 from cachetools.keys import hashkey
 from redis.cache import (
@@ -21,6 +20,7 @@
     CacheInterface,
     CacheToolsFactory,
 )
+from . import scheduler
 
 from ._parsers import Encoder, _HiredisParser, _RESP2Parser, _RESP3Parser
 from .backoff import NoBackoff
@@ -36,6 +36,7 @@
     TimeoutError,
 )
 from .retry import Retry
+from .scheduler import Scheduler
 from .utils import (
     CRYPTOGRAPHY_AVAILABLE,
     HIREDIS_AVAILABLE,
@@ -1265,7 +1266,9 @@ def __init__(
         self.cache = None
         self._cache_conf = None
         self._cache_factory = cache_factory
-        self.scheduler = None
+        self._scheduler = None
+        self._hc_cancel_event = None
+        self._hc_thread = None
 
         if connection_kwargs.get("use_cache"):
             if connection_kwargs.get("protocol") not in [3, "3"]:
@@ -1286,14 +1289,7 @@ def __init__(
                 else:
                     self.cache = CacheToolsFactory(self._cache_conf).get_cache()
 
-            self.scheduler = BackgroundScheduler()
-            self.scheduler.add_job(
-                self._perform_health_check,
-                "interval",
-                seconds=2,
-                id="cache_health_check",
-            )
-            self.scheduler.start()
+            self._scheduler = Scheduler()
 
         connection_kwargs.pop("use_cache", None)
         connection_kwargs.pop("cache_eviction", None)
@@ -1312,6 +1308,16 @@ def __init__(
         self._fork_lock = threading.Lock()
         self.reset()
 
+        # Run scheduled healthcheck to avoid stale invalidations in idle connections.
+        if self.cache is not None and self._scheduler is not None:
+            self._hc_cancel_event = threading.Event()
+            self._hc_thread = self._scheduler.run_with_interval(
+                self._perform_health_check,
+                2,
+                self._hc_cancel_event
+            )
+
+
     def __repr__(self) -> (str, str):
         return (
             f"<{type(self).__module__}.{type(self).__name__}"
@@ -1491,6 +1497,14 @@ 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()
+
     def close(self) -> None:
         """Close the pool, disconnecting all connections"""
         self.disconnect()
@@ -1502,13 +1516,14 @@ def set_retry(self, retry: "Retry") -> None:
         for conn in self._in_use_connections:
             conn.retry = retry
 
-    def _perform_health_check(self) -> None:
+    def _perform_health_check(self, done: threading.Event) -> None:
         self._checkpid()
         with self._lock:
             while self._available_connections:
                 conn = self._available_connections.pop()
                 conn.send_command("PING")
                 conn.read_response()
+            done.set()
 
 
 class BlockingConnectionPool(ConnectionPool):

redis/scheduler.py

@@ -0,0 +1,61 @@
+import threading
+import time
+from typing import Callable
+
+
+class Scheduler:
+
+    def __init__(self, polling_period: float = 0.1):
+        """
+        :param polling_period: Period between polling operations.
+        Needs to detect when new job has to be scheduled.
+        """
+        self.polling_period = polling_period
+
+    def run_with_interval(
+            self,
+            func: Callable[[threading.Event, ...], None],
+            interval: float,
+            cancel: threading.Event,
+            args: tuple = (),
+    ) -> threading.Thread:
+        """
+        Run scheduled execution with given interval
+        in a separate thread until cancel event won't be set.
+        """
+        done = threading.Event()
+        thread = threading.Thread(target=self._run_timer, args=(func, interval, (done, *args), done, cancel))
+        thread.start()
+        return thread
+
+    def _get_timer(
+            self,
+            func: Callable[[threading.Event, ...], None],
+            interval: float,
+            args: tuple
+    ) -> threading.Timer:
+        timer = threading.Timer(interval=interval, function=func, args=args)
+        return timer
+
+    def _run_timer(
+            self,
+            func: Callable[[threading.Event, ...], None],
+            interval: float,
+            args: tuple,
+            done: threading.Event,
+            cancel: threading.Event
+    ):
+        timer = self._get_timer(func, interval, args)
+        timer.start()
+
+        while not cancel.is_set():
+            if done.is_set():
+                done.clear()
+                timer.join()
+                timer = self._get_timer(func, interval, args)
+                timer.start()
+            else:
+                time.sleep(self.polling_period)
+
+        timer.cancel()
+        timer.join()

requirements.txt

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

tests/conftest.py

@@ -1,17 +1,21 @@
 import argparse
 import random
+import threading
 import time
 from typing import Callable, TypeVar
 from unittest import mock
 from unittest.mock import Mock
 from urllib.parse import urlparse
 
 import pytest
+from _pytest import unittest
+
 import redis
 from packaging.version import Version
 from redis import Sentinel
 from redis.backoff import NoBackoff
-from redis.connection import Connection, SSLConnection, parse_url
+from redis.cache import CacheConfiguration, EvictionPolicy, CacheFactoryInterface, CacheInterface
+from redis.connection import Connection, SSLConnection, parse_url, ConnectionPool, ConnectionInterface
 from redis.exceptions import RedisClusterException
 from redis.retry import Retry
 from tests.ssl_utils import get_ssl_filename
@@ -537,6 +541,33 @@ def master_host(request):
     return parts.hostname, (parts.port or 6379)
 
 
+@pytest.fixture()
+def cache_conf() -> CacheConfiguration:
+    return CacheConfiguration(
+        cache_size=100,
+        cache_ttl=20,
+        cache_eviction=EvictionPolicy.TTL
+    )
+
+
+@pytest.fixture()
+def mock_cache_factory() -> CacheFactoryInterface:
+    mock_factory = Mock(spec=CacheFactoryInterface)
+    return mock_factory
+
+
+@pytest.fixture()
+def mock_cache() -> CacheInterface:
+    mock_cache = Mock(spec=CacheInterface)
+    return mock_cache
+
+
+@pytest.fixture()
+def mock_connection() -> ConnectionInterface:
+    mock_connection = Mock(spec=ConnectionInterface)
+    return mock_connection
+
+
 def wait_for_command(client, monitor, command, key=None):
     # issue a command with a key name that's local to this process.
     # if we find a command with our key before the command we're waiting

tests/test_cache.py

@@ -4,7 +4,7 @@
 import pytest
 import redis
 from cachetools import LFUCache, LRUCache, TTLCache
-from redis.cache import CacheToolsAdapter, EvictionPolicy
+from redis.cache import CacheToolsAdapter, EvictionPolicy, CacheConfiguration
 from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import _get_client, skip_if_resp_version
 
@@ -185,7 +185,7 @@ def test_get_from_cache_multithreaded(self, r):
         indirect=True,
     )
     @pytest.mark.onlynoncluster
-    def test_health_check_invalidate_cache(self, r, r2):
+    def test_health_check_invalidate_cache(self, r):
         cache = r.get_cache()
         # add key to redis
         r.set("foo", "bar")
@@ -194,7 +194,7 @@ def test_health_check_invalidate_cache(self, r, r2):
         # get key from local cache
         assert cache.get(("GET", "foo")) == b"bar"
         # change key in redis (cause invalidation)
-        r2.set("foo", "barbar")
+        r.set("foo", "barbar")
         # Wait for health check
         time.sleep(2)
         # Make sure that value was invalidated
@@ -1154,3 +1154,26 @@ def test_cache_invalidate_all_related_responses(self, r):
         assert r.get("foo") == b"baz"
         assert cache.get(("MGET", "foo", "bar")) is None
         assert cache.get(("GET", "foo")) == b"baz"
+
+
+class TestUnitCacheConfiguration:
+    TTL = 20
+    MAX_SIZE = 100
+    EVICTION_POLICY = EvictionPolicy.TTL
+
+    def test_get_ttl(self, cache_conf: CacheConfiguration):
+        assert self.TTL == cache_conf.get_ttl()
+
+    def test_get_max_size(self, cache_conf: CacheConfiguration):
+        assert self.MAX_SIZE == cache_conf.get_max_size()
+
+    def test_get_eviction_policy(self, cache_conf: CacheConfiguration):
+        assert self.EVICTION_POLICY == cache_conf.get_eviction_policy()
+
+    def test_is_exceeds_max_size(self, cache_conf: CacheConfiguration):
+        assert not cache_conf.is_exceeds_max_size(self.MAX_SIZE)
+        assert cache_conf.is_exceeds_max_size(self.MAX_SIZE + 1)
+
+    def test_is_allowed_to_cache(self, cache_conf: CacheConfiguration):
+        assert cache_conf.is_allowed_to_cache("GET")
+        assert not cache_conf.is_allowed_to_cache("SET")