Avoid an extra "can_read" call and use timeout directly.

Author: kristjanvalur

redis/asyncio/client.py

@@ -754,9 +754,8 @@ async def parse_response(self, block: bool = True, timeout: float = 0):
 
         await self.check_health()
 
-        if not block and not await self._execute(conn, conn.can_read, timeout=timeout):
-            return None
-        response = await self._execute(conn, conn.read_response)
+        read_timeout = None if block else timeout
+        response = await self._execute(conn, conn.read_response, timeout=read_timeout)
 
         if conn.health_check_interval and response == self.health_check_response:
             # ignore the health check message as user might not expect it

redis/asyncio/connection.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextlib
 import copy
 import enum
 import errno
@@ -56,6 +57,19 @@
 if HIREDIS_AVAILABLE:
     import hiredis
 
+if sys.version_info[:2] >= (3, 10):
+    nullcontext = contextlib.nullcontext()
+else:
+
+    class NullContext:
+        async def __aenter__(self):
+            pass
+
+        async def __aexit__(self, *args):
+            pass
+
+    nullcontext = NullContext()
+
 NONBLOCKING_EXCEPTION_ERROR_NUMBERS = {
     BlockingIOError: errno.EWOULDBLOCK,
     ssl.SSLWantReadError: 2,
@@ -918,55 +932,27 @@ async def can_read(self, timeout: float = 0):
                 f"Error while reading from {self.host}:{self.port}: {e.args}"
             )
 
-    async def read_response(self, disable_decoding: bool = False):
-        """Read the response from a previously sent command"""
-        try:
-            async with self._lock:
-                if self.socket_timeout:
-                    async with async_timeout.timeout(self.socket_timeout):
-                        response = await self._parser.read_response(
-                            disable_decoding=disable_decoding
-                        )
-                else:
-                    response = await self._parser.read_response(
-                        disable_decoding=disable_decoding
-                    )
-        except asyncio.TimeoutError:
-            await self.disconnect()
-            raise TimeoutError(f"Timeout reading from {self.host}:{self.port}")
-        except OSError as e:
-            await self.disconnect()
-            raise ConnectionError(
-                f"Error while reading from {self.host}:{self.port} : {e.args}"
-            )
-        except BaseException:
-            await self.disconnect()
-            raise
-
-        if self.health_check_interval:
-            if sys.version_info[0:2] == (3, 6):
-                func = asyncio.get_event_loop
-            else:
-                func = asyncio.get_running_loop
-            self.next_health_check = func().time() + self.health_check_interval
-
-        if isinstance(response, ResponseError):
-            raise response from None
-        return response
-
-    async def read_response_without_lock(self, disable_decoding: bool = False):
+    async def read_response(
+        self,
+        disable_decoding: bool = False,
+        timeout: Optional[float] = None,
+        with_lock: bool = True,
+    ):
         """Read the response from a previously sent command"""
+        read_timeout = timeout if timeout is not None else self.socket_timeout
+        lock_ctxt = self._lock if with_lock else nullcontext
+        if not self.is_connected:
+            await self.connect()
         try:
-            if self.socket_timeout:
-                async with async_timeout.timeout(self.socket_timeout):
-                    response = await self._parser.read_response(
-                        disable_decoding=disable_decoding
-                    )
-            else:
+            async with lock_ctxt, async_timeout.timeout(read_timeout):
                 response = await self._parser.read_response(
                     disable_decoding=disable_decoding
                 )
         except asyncio.TimeoutError:
+            if timeout is not None:
+                # user requested timeout, return None
+                return None
+            # it was a self.socket_timeout error.
             await self.disconnect()
             raise TimeoutError(f"Timeout reading from {self.host}:{self.port}")
         except OSError as e:
@@ -989,6 +975,14 @@ async def read_response_without_lock(self, disable_decoding: bool = False):
             raise response from None
         return response
 
+    async def read_response_without_lock(
+        self, disable_decoding: bool = False, timeout: Optional[float] = None
+    ):
+        """Read the response from a previously sent command"""
+        return await self.read_response(
+            disable_decoding=disable_decoding, timeout=timeout, with_lock=False
+        )
+
     def pack_command(self, *args: EncodableT) -> List[bytes]:
         """Pack a series of arguments into the Redis protocol"""
         output = []