fix: Fixed potential deadlock from unexpected __del__ call

redis/client.py

@@ -369,6 +369,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
@@ -774,6 +776,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

@@ -710,7 +710,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
@@ -1474,7 +1474,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()
@@ -2178,7 +2178,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):