Merge branch 'master' into kristjan/interrupted-send-receive

Author: kristjanvalur

.github/workflows/integration.yaml

@@ -51,6 +51,7 @@ jobs:
      timeout-minutes: 30
      strategy:
        max-parallel: 15
+       fail-fast: false
        matrix:
          python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8', 'pypy-3.9']
          test-type: ['standalone', 'cluster']
@@ -108,6 +109,7 @@ jobs:
     name: Install package from commit hash
     runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8', 'pypy-3.9']
     steps:

redis/asyncio/client.py

@@ -500,28 +500,37 @@ async def _disconnect_raise(self, conn: Connection, error: Exception):
         ):
             raise error
 
-    # COMMAND EXECUTION AND PROTOCOL PARSING
-    async def execute_command(self, *args, **options):
-        """Execute a command and return a parsed response"""
-        await self.initialize()
-        pool = self.connection_pool
-        command_name = args[0]
-        conn = self.connection or await pool.get_connection(command_name, **options)
-
-        if self.single_connection_client:
-            await self._single_conn_lock.acquire()
+    async def _try_send_command_parse_response(self, conn, *args, **options):
         try:
             return await conn.retry.call_with_retry(
                 lambda: self._send_command_parse_response(
-                    conn, command_name, *args, **options
+                    conn, args[0], *args, **options
                 ),
                 lambda error: self._disconnect_raise(conn, error),
             )
+        except asyncio.CancelledError:
+            await conn.disconnect(nowait=True)
+            raise
         finally:
             if self.single_connection_client:
                 self._single_conn_lock.release()
             if not self.connection:
-                await pool.release(conn)
+                await self.connection_pool.release(conn)
+
+    # COMMAND EXECUTION AND PROTOCOL PARSING
+    async def execute_command(self, *args, **options):
+        """Execute a command and return a parsed response"""
+        await self.initialize()
+        pool = self.connection_pool
+        command_name = args[0]
+        conn = self.connection or await pool.get_connection(command_name, **options)
+
+        if self.single_connection_client:
+            await self._single_conn_lock.acquire()
+
+        return await asyncio.shield(
+            self._try_send_command_parse_response(conn, *args, **options)
+        )
 
     async def parse_response(
         self, connection: Connection, command_name: Union[str, bytes], **options
@@ -702,6 +711,11 @@ async def reset(self):
             self.pending_unsubscribe_patterns = set()
 
     def close(self) -> Awaitable[NoReturn]:
+        # In case a connection property does not yet exist
+        # (due to a crash earlier in the Redis() constructor), return
+        # immediately as there is nothing to clean-up.
+        if not hasattr(self, "connection"):
+            return
         return self.reset()
 
     async def on_connect(self, connection: Connection):
@@ -760,10 +774,18 @@ async def _disconnect_raise_connect(self, conn, error):
         is not a TimeoutError. Otherwise, try to reconnect
         """
         await conn.disconnect()
+
         if not (conn.retry_on_timeout and isinstance(error, TimeoutError)):
             raise error
         await conn.connect()
 
+    async def _try_execute(self, conn, command, *arg, **kwargs):
+        try:
+            return await command(*arg, **kwargs)
+        except asyncio.CancelledError:
+            await conn.disconnect()
+            raise
+
     async def _execute(self, conn, command, *args, **kwargs):
         """
         Connect manually upon disconnection. If the Redis server is down,
@@ -772,9 +794,11 @@ async def _execute(self, conn, command, *args, **kwargs):
         called by the # connection to resubscribe us to any channels and
         patterns we were previously listening to
         """
-        return await conn.retry.call_with_retry(
-            lambda: command(*args, **kwargs),
-            lambda error: self._disconnect_raise_connect(conn, error),
+        return await asyncio.shield(
+            conn.retry.call_with_retry(
+                lambda: self._try_execute(conn, command, *args, **kwargs),
+                lambda error: self._disconnect_raise_connect(conn, error),
+            )
         )
 
     async def parse_response(self, block: bool = True, timeout: float = 0):
@@ -1178,6 +1202,18 @@ async def _disconnect_reset_raise(self, conn, error):
             await self.reset()
             raise
 
+    async def _try_send_command_parse_response(self, conn, *args, **options):
+        try:
+            return await conn.retry.call_with_retry(
+                lambda: self._send_command_parse_response(
+                    conn, args[0], *args, **options
+                ),
+                lambda error: self._disconnect_reset_raise(conn, error),
+            )
+        except asyncio.CancelledError:
+            await conn.disconnect()
+            raise
+
     async def immediate_execute_command(self, *args, **options):
         """
         Execute a command immediately, but don't auto-retry on a
@@ -1193,12 +1229,8 @@ async def immediate_execute_command(self, *args, **options):
                 command_name, self.shard_hint
             )
             self.connection = conn
-
-        return await conn.retry.call_with_retry(
-            lambda: self._send_command_parse_response(
-                conn, command_name, *args, **options
-            ),
-            lambda error: self._disconnect_reset_raise(conn, error),
+        return await asyncio.shield(
+            self._try_send_command_parse_response(conn, *args, **options)
         )
 
     def pipeline_execute_command(self, *args, **options):
@@ -1366,6 +1398,19 @@ async def _disconnect_raise_reset(self, conn: Connection, error: Exception):
             await self.reset()
             raise
 
+    async def _try_execute(self, conn, execute, stack, raise_on_error):
+        try:
+            return await conn.retry.call_with_retry(
+                lambda: execute(conn, stack, raise_on_error),
+                lambda error: self._disconnect_raise_reset(conn, error),
+            )
+        except asyncio.CancelledError:
+            # not supposed to be possible, yet here we are
+            await conn.disconnect(nowait=True)
+            raise
+        finally:
+            await self.reset()
+
     async def execute(self, raise_on_error: bool = True):
         """Execute all the commands in the current pipeline"""
         stack = self.command_stack
@@ -1387,10 +1432,11 @@ async def execute(self, raise_on_error: bool = True):
         conn = cast(Connection, conn)
 
         try:
-            return await conn.retry.call_with_retry(
-                lambda: execute(conn, stack, raise_on_error),
-                lambda error: self._disconnect_raise_reset(conn, error),
+            return await asyncio.shield(
+                self._try_execute(conn, execute, stack, raise_on_error)
             )
+        except RuntimeError:
+            await self.reset()
         finally:
             await self.reset()
 

redis/asyncio/cluster.py

@@ -1002,12 +1002,33 @@ async def execute_command(self, *args: Any, **kwargs: Any) -> Any:
         await connection.send_packed_command(connection.pack_command(*args), False)
 
         # Read response
+        return await asyncio.shield(
+            self._parse_and_release(connection, args[0], **kwargs)
+        )
+
+    async def _parse_and_release(self, connection, *args, **kwargs):
         try:
-            return await self.parse_response(connection, args[0], **kwargs)
+            return await self.parse_response(connection, *args, **kwargs)
+        except asyncio.CancelledError:
+            # should not be possible
+            await connection.disconnect(nowait=True)
+            raise
         finally:
-            # Release connection
             self._free.append(connection)
 
+    async def _try_parse_response(self, cmd, connection, ret):
+        try:
+            cmd.result = await asyncio.shield(
+                self.parse_response(connection, cmd.args[0], **cmd.kwargs)
+            )
+        except asyncio.CancelledError:
+            await connection.disconnect(nowait=True)
+            raise
+        except Exception as e:
+            cmd.result = e
+            ret = True
+        return ret
+
     async def execute_pipeline(self, commands: List["PipelineCommand"]) -> bool:
         # Acquire connection
         connection = self.acquire_connection()
@@ -1020,13 +1041,7 @@ async def execute_pipeline(self, commands: List["PipelineCommand"]) -> bool:
         # Read responses
         ret = False
         for cmd in commands:
-            try:
-                cmd.result = await self.parse_response(
-                    connection, cmd.args[0], **cmd.kwargs
-                )
-            except Exception as e:
-                cmd.result = e
-                ret = True
+            ret = await asyncio.shield(self._try_parse_response(cmd, connection, ret))
 
         # Release connection
         self._free.append(connection)

redis/asyncio/connection.py

@@ -25,7 +25,9 @@
 )
 from urllib.parse import ParseResult, parse_qs, unquote, urlparse
 
-if sys.version_info.major >= 3 and sys.version_info.minor >= 11:
+# the functionality is available in 3.11.x but has a major issue before
+# 3.11.3. See https://github.com/redis/redis-py/issues/2633
+if sys.version_info >= (3, 11, 3):
     from asyncio import timeout as async_timeout
 else:
     from async_timeout import timeout as async_timeout

redis/connection.py

@@ -1167,8 +1167,9 @@ def _connect(self):
 class UnixDomainSocketConnection(AbstractConnection):
     "Manages UDS communication to and from a Redis server"
 
-    def __init__(self, path="", **kwargs):
+    def __init__(self, path="", socket_timeout=None, **kwargs):
         self.path = path
+        self.socket_timeout = socket_timeout
         super().__init__(**kwargs)
 
     def repr_pieces(self):

setup.py

@@ -8,7 +8,7 @@
     long_description_content_type="text/markdown",
     keywords=["Redis", "key-value store", "database"],
     license="MIT",
-    version="4.5.2",
+    version="4.5.4",
     packages=find_packages(
         include=[
             "redis",
@@ -34,7 +34,7 @@
     install_requires=[
         'importlib-metadata >= 1.0; python_version < "3.8"',
         'typing-extensions; python_version<"3.8"',
-        'async-timeout>=4.0.2; python_version<"3.11"',
+        'async-timeout>=4.0.2; python_version<="3.11.2"',
     ],
     classifiers=[
         "Development Status :: 5 - Production/Stable",