Removed background healthcheck, use connection reference approach instead

Author: vladvildanov

docs/resp3_features.rst

@@ -98,10 +98,4 @@ Enable caching with custom cache implementation:
 
 CacheImpl should implement a `CacheInterface` specified in `redis.cache` package.
 
-Explicit disconnect
-
-It's important to call `close()` to properly close the connection to server.
-For caching purposes, we're using a separate thread that performs health checks with configurable interval and it relies on
-`close()` to be called before the shutdown.
-
 More robust documentation soon will be available at `official Redis documentation <https://redis.io/docs/latest/>`_.

redis/cache.py

@@ -1,4 +1,5 @@
 import copy
+import weakref
 from abc import ABC, abstractmethod
 from collections import OrderedDict
 from enum import Enum
@@ -32,14 +33,21 @@ def __eq__(self, other):
 
 class CacheEntry:
     def __init__(
-        self, cache_key: CacheKey, cache_value: bytes, status: CacheEntryStatus
+        self,
+        cache_key: CacheKey,
+        cache_value: bytes,
+        status: CacheEntryStatus,
+        connection_ref,
     ):
         self.cache_key = cache_key
         self.cache_value = cache_value
         self.status = status
+        self.connection_ref = connection_ref
 
     def __hash__(self):
-        return hash((self.cache_key, self.cache_value, self.status))
+        return hash(
+            (self.cache_key, self.cache_value, self.status, self.connection_ref)
+        )
 
     def __eq__(self, other):
         return hash(self) == hash(other)
@@ -86,10 +94,6 @@ def get_max_size(self) -> int:
     def get_eviction_policy(self):
         pass
 
-    @abstractmethod
-    def get_health_check_interval(self) -> float:
-        pass
-
     @abstractmethod
     def is_exceeds_max_size(self, count: int) -> bool:
         pass
@@ -182,7 +186,7 @@ def get(self, key: CacheKey) -> Union[CacheEntry, None]:
             return None
 
         self._eviction_policy.touch(key)
-        return copy.deepcopy(entry)
+        return entry
 
     def delete_by_cache_keys(self, cache_keys: List[CacheKey]) -> List[bool]:
         response = []
@@ -360,12 +364,10 @@ def __init__(
         max_size: int = DEFAULT_MAX_SIZE,
         cache_class: Any = DEFAULT_CACHE_CLASS,
         eviction_policy: EvictionPolicy = DEFAULT_EVICTION_POLICY,
-        health_check_interval: float = 2.0,
     ):
         self._cache_class = cache_class
         self._max_size = max_size
         self._eviction_policy = eviction_policy
-        self._health_check_interval = health_check_interval
 
     def get_cache_class(self):
         return self._cache_class
@@ -376,9 +378,6 @@ def get_max_size(self) -> int:
     def get_eviction_policy(self) -> EvictionPolicy:
         return self._eviction_policy
 
-    def get_health_check_interval(self) -> float:
-        return self._health_check_interval
-
     def is_exceeds_max_size(self, count: int) -> bool:
         return count > self._max_size
 

redis/connection.py

@@ -35,7 +35,6 @@
     TimeoutError,
 )
 from .retry import Retry
-from .scheduler import Scheduler
 from .utils import (
     CRYPTOGRAPHY_AVAILABLE,
     HIREDIS_AVAILABLE,
@@ -741,12 +740,18 @@ class CacheProxyConnection(ConnectionInterface):
     MIN_ALLOWED_VERSION = "7.4.0"
     DEFAULT_SERVER_NAME = "redis"
 
-    def __init__(self, conn: ConnectionInterface, cache: CacheInterface):
+    def __init__(
+        self,
+        conn: ConnectionInterface,
+        cache: CacheInterface,
+        pool_lock: threading.Lock,
+    ):
         self.pid = os.getpid()
         self._conn = conn
         self.retry = self._conn.retry
         self.host = self._conn.host
         self.port = self._conn.port
+        self._pool_lock = pool_lock
         self._cache = cache
         self._cache_lock = threading.Lock()
         self._current_command_cache_key = None
@@ -824,9 +829,17 @@ def send_command(self, *args, **kwargs):
         )
 
         with self._cache_lock:
-            # If current command reply already cached
-            # prevent sending data over socket.
+            # We have to trigger invalidation processing in case if
+            # it was cached by another connection to avoid
+            # queueing invalidations in stale connections.
             if self._cache.get(self._current_command_cache_key):
+                entry = self._cache.get(self._current_command_cache_key)
+
+                if entry.connection_ref != self._conn:
+                    with self._pool_lock:
+                        while entry.connection_ref.can_read():
+                            entry.connection_ref.read_response(push_request=True)
+
                 return
 
             # Set temporary entry value to prevent
@@ -836,6 +849,7 @@ def send_command(self, *args, **kwargs):
                     cache_key=self._current_command_cache_key,
                     cache_value=self.DUMMY_CACHE_VALUE,
                     status=CacheEntryStatus.IN_PROGRESS,
+                    connection_ref=self._conn,
                 )
             )
 
@@ -857,7 +871,9 @@ def read_response(
                 and self._cache.get(self._current_command_cache_key).status
                 != CacheEntryStatus.IN_PROGRESS
             ):
-                return self._cache.get(self._current_command_cache_key).cache_value
+                return copy.deepcopy(
+                    self._cache.get(self._current_command_cache_key).cache_value
+                )
 
         response = self._conn.read_response(
             disable_decoding=disable_decoding,
@@ -879,13 +895,9 @@ def read_response(
             # Cache only responses that still valid
             # and wasn't invalidated by another connection in meantime.
             if cache_entry is not None:
-                self._cache.set(
-                    CacheEntry(
-                        cache_key=self._current_command_cache_key,
-                        cache_value=response,
-                        status=CacheEntryStatus.VALID,
-                    )
-                )
+                cache_entry.status = CacheEntryStatus.VALID
+                cache_entry.cache_value = response
+                self._cache.set(cache_entry)
 
         return response
 
@@ -1284,9 +1296,6 @@ def __init__(
         self.max_connections = max_connections
         self.cache = None
         self._cache_factory = cache_factory
-        self._scheduler = None
-        self._hc_cancel_event = None
-        self._hc_thread = None
 
         if connection_kwargs.get("cache_config") or connection_kwargs.get("cache"):
             if connection_kwargs.get("protocol") not in [3, "3"]:
@@ -1307,8 +1316,6 @@ def __init__(
                         self.connection_kwargs.get("cache_config")
                     ).get_cache()
 
-            self._scheduler = Scheduler()
-
         connection_kwargs.pop("cache", None)
         connection_kwargs.pop("cache_config", None)
 
@@ -1322,7 +1329,6 @@ def __init__(
         # release the lock.
         self._fork_lock = threading.Lock()
         self.reset()
-        self.run_scheduled_healthcheck()
 
     def __repr__(self) -> (str, str):
         return (
@@ -1452,7 +1458,7 @@ def make_connection(self) -> "ConnectionInterface":
 
         if self.cache is not None:
             return CacheProxyConnection(
-                self.connection_class(**self.connection_kwargs), self.cache
+                self.connection_class(**self.connection_kwargs), self.cache, self._lock
             )
 
         return self.connection_class(**self.connection_kwargs)
@@ -1501,8 +1507,6 @@ def disconnect(self, inuse_connections: bool = True) -> None:
             for connection in connections:
                 connection.disconnect()
 
-            self.stop_scheduled_healthcheck()
-
     def close(self) -> None:
         """Close the pool, disconnecting all connections"""
         self.disconnect()
@@ -1514,33 +1518,6 @@ def set_retry(self, retry: "Retry") -> None:
         for conn in self._in_use_connections:
             conn.retry = retry
 
-    def run_scheduled_healthcheck(self) -> None:
-        # 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()
-            hc_interval = self.cache.get_config().get_health_check_interval()
-            self._hc_thread = self._scheduler.run_with_interval(
-                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:
-            while self._available_connections:
-                conn = self._available_connections.pop()
-                conn.send_command("PING")
-                conn.read_response()
-            done.set()
-
 
 class BlockingConnectionPool(ConnectionPool):
     """
@@ -1620,7 +1597,7 @@ def make_connection(self):
         "Make a fresh connection."
         if self.cache is not None:
             connection = CacheProxyConnection(
-                self.connection_class(**self.connection_kwargs), self.cache
+                self.connection_class(**self.connection_kwargs), self.cache, self._lock
             )
         else:
             connection = self.connection_class(**self.connection_kwargs)
@@ -1705,5 +1682,3 @@ def disconnect(self):
         self._checkpid()
         for connection in self._connections:
             connection.disconnect()
-
-        self.stop_scheduled_healthcheck()

redis/scheduler.py

@@ -115,7 +115,6 @@ def get_master_address(self):
             connection_pool = self.connection_pool_ref()
             if connection_pool is not None:
                 connection_pool.disconnect(inuse_connections=False)
-                connection_pool.run_scheduled_healthcheck()
         return master_address
 
     def rotate_slaves(self):