Merge branch 'master' into ps_drop_python3_8_support

Author: petyaslavova

redis/client.py

@@ -368,6 +368,8 @@ def __init__(
         ]:
             raise RedisError("Client caching is only supported with RESP version 3")
 
+        # TODO: To avoid breaking changes during the bug fix, we have to keep non-reentrant lock.
+        # TODO: Remove this before next major version (7.0.0)
         self.single_connection_lock = threading.Lock()
         self.connection = None
         self._single_connection_client = single_connection_client
@@ -773,6 +775,9 @@ def __init__(
             self._event_dispatcher = EventDispatcher()
         else:
             self._event_dispatcher = event_dispatcher
+
+        # TODO: To avoid breaking changes during the bug fix, we have to keep non-reentrant lock.
+        # TODO: Remove this before next major version (7.0.0)
         self._lock = threading.Lock()
         if self.encoder is None:
             self.encoder = self.connection_pool.get_encoder()

redis/cluster.py

@@ -709,7 +709,7 @@ def __init__(
         self.result_callbacks = CaseInsensitiveDict(self.__class__.RESULT_CALLBACKS)
 
         self.commands_parser = CommandsParser(self)
-        self._lock = threading.Lock()
+        self._lock = threading.RLock()
 
     def __enter__(self):
         return self
@@ -1473,7 +1473,7 @@ def __init__(
         self.connection_kwargs = kwargs
         self.read_load_balancer = LoadBalancer()
         if lock is None:
-            lock = threading.Lock()
+            lock = threading.RLock()
         self._lock = lock
         if event_dispatcher is None:
             self._event_dispatcher = EventDispatcher()
@@ -2177,7 +2177,7 @@ def __init__(
             kwargs.get("decode_responses", False),
         )
         if lock is None:
-            lock = threading.Lock()
+            lock = threading.RLock()
         self._lock = lock
         self.parent_execute_command = super().execute_command
         self._execution_strategy: ExecutionStrategy = (

redis/connection.py

@@ -820,7 +820,7 @@ def __init__(
         self.credential_provider = conn.credential_provider
         self._pool_lock = pool_lock
         self._cache = cache
-        self._cache_lock = threading.Lock()
+        self._cache_lock = threading.RLock()
         self._current_command_cache_key = None
         self._current_options = None
         self.register_connect_callback(self._enable_tracking_callback)
@@ -1420,8 +1420,16 @@ def __init__(
         # object of this pool. subsequent threads acquiring this lock
         # will notice the first thread already did the work and simply
         # release the lock.
-        self._fork_lock = threading.Lock()
-        self._lock = threading.Lock()
+
+        self._fork_lock = threading.RLock()
+
+        if self.cache is None:
+            self._lock = threading.RLock()
+        else:
+            # TODO: To avoid breaking changes during the bug fix, we have to keep non-reentrant lock.
+            # TODO: Remove this before next major version (7.0.0)
+            self._lock = threading.Lock()
+
         self.reset()
 
     def __repr__(self) -> (str, str):

redis/typing.py

@@ -15,8 +15,6 @@
 
 if TYPE_CHECKING:
     from redis._parsers import Encoder
-    from redis.asyncio.connection import ConnectionPool as AsyncConnectionPool
-    from redis.connection import ConnectionPool
 
 
 Number = Union[int, float]
@@ -52,8 +50,6 @@
 
 
 class CommandsProtocol(Protocol):
-    connection_pool: Union["AsyncConnectionPool", "ConnectionPool"]
-
     def execute_command(self, *args, **options) -> ResponseT: ...