Repository: redis/redis-py
Pull request #2360 (2022-08-25 14:24:43)

Dev/no can read
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

The `asyncio.Connection.can_read()` API is unnecessary in async-io, since polling a connection is best done with timeout.
Keeping this api adds complexity in the lower layers of the asyncio.Connection since it requires reading data and then "unreading" them for subsequent `read` operations.

Further, this code is almost exclusively used by assertion macros, to make sure that there is no data present on a  connection pulled from a connection pool.  For this use case, it is fine that the read data is discarded.

The only remaining use case was in the `PubSub` implementation, but that can be done differently.

Renaming the method to `can_read_destructive()` and simplifying it, getting rid of the "keep any read data for next read" logic, paves the way for further simplifications of the asyncio connection code, and other pull requests in the same ares (getting rid of locking, buffering, simplifying timeouts, etc)

The PR contains elements from Pull Requests #2356 and #2104 as necessary for it to work properly

codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2360?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
> Merging [#2360](https://codecov.io/gh/redis/redis-py/pull/2360?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (955bdee) into [master](https://codecov.io/gh/redis/redis-py/commit/ab5d407a03192bd8ab63a1bfc891db9c04330265?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (ab5d407) will **decrease** coverage by `13.55%`.
> The diff coverage is `80.00%`.

```diff
@@             Coverage Diff             @@
##           master    #2360       +/-   ##
===========================================
- Coverage   92.19%   78.64%   -13.56%     
===========================================
  Files         111      111               
  Lines       28782    28753       -29     
===========================================
- Hits        26536    22612     -3924     
- Misses       2246     6141     +3895     
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2360?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [tests/test\_asyncio/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY2x1c3Rlci5weQ==) | `18.88% <0.00%> (-78.40%)` | :arrow_down: |
| [redis/asyncio/connection.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jb25uZWN0aW9uLnB5) | `77.77% <78.26%> (-8.34%)` | :arrow_down: |
| [redis/asyncio/client.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbGllbnQucHk=) | `91.99% <100.00%> (-0.28%)` | :arrow_down: |
| [tests/test\_asyncio/test\_connection\_pool.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY29ubmVjdGlvbl9wb29sLnB5) | `99.40% <100.00%> (ø)` | |
| [tests/test\_asyncio/test\_pubsub.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfcHVic3ViLnB5) | `99.32% <100.00%> (ø)` | |
| [tests/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9jbHVzdGVyLnB5) | `18.58% <0.00%> (-78.27%)` | :arrow_down: |
| [redis/asyncio/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbHVzdGVyLnB5) | `16.52% <0.00%> (-73.77%)` | :arrow_down: |
| [redis/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2x1c3Rlci5weQ==) | `18.81% <0.00%> (-71.04%)` | :arrow_down: |
| [redis/asyncio/parser.py](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9wYXJzZXIucHk=) | `17.30% <0.00%> (-63.47%)` | :arrow_down: |
| ... and [34 more](https://codecov.io/gh/redis/redis-py/pull/2360/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)

dvora-h: @kristjanvalur Can you resolve the conflicts with master?


kristjanvalur: No problem, it is best that I resolve these myself anyway.

Patch:
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 9c8caae0..c13054b2 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -24,6 +24,8 @@ from typing import (
     cast,
 )
 
+import async_timeout
+
 from redis.asyncio.connection import (
     Connection,
     ConnectionPool,
@@ -754,15 +756,21 @@ class PubSub:
 
         await self.check_health()
 
-        async def try_read():
-            if not block:
-                if not await conn.can_read(timeout=timeout):
+        if not conn.is_connected:
+            await conn.connect()
+
+        if not block:
+
+            async def read_with_timeout():
+                try:
+                    async with async_timeout.timeout(timeout):
+                        return await conn.read_response()
+                except asyncio.TimeoutError:
                     return None
-            else:
-                await conn.connect()
-            return await conn.read_response()
 
-        response = await self._execute(conn, try_read)
+            response = await self._execute(conn, read_with_timeout)
+        else:
+            response = await self._execute(conn, conn.read_response)
 
         if conn.health_check_interval and response == self.health_check_response:
             # ignore the health check message as user might not expect it
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 64848f4f..53b41af7 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -208,7 +208,7 @@ class BaseParser:
     def on_connect(self, connection: "Connection"):
         raise NotImplementedError()
 
-    async def can_read(self, timeout: float) -> bool:
+    async def can_read_destructive(self) -> bool:
         raise NotImplementedError()
 
     async def read_response(
@@ -286,9 +286,9 @@ class SocketBuffer:
                 return False
             raise ConnectionError(f"Error while reading from socket: {ex.args}")
 
-    async def can_read(self, timeout: float) -> bool:
+    async def can_read_destructive(self) -> bool:
         return bool(self.length) or await self._read_from_socket(
-            timeout=timeout, raise_on_timeout=False
+            timeout=0, raise_on_timeout=False
         )
 
     async def read(self, length: int) -> bytes:
@@ -386,8 +386,8 @@ class PythonParser(BaseParser):
             self._buffer = None
         self.encoder = None
 
-    async def can_read(self, timeout: float):
-        return self._buffer and bool(await self._buffer.can_read(timeout))
+    async def can_read_destructive(self):
+        return self._buffer and bool(await self._buffer.can_read_destructive())
 
     async def read_response(
         self, disable_decoding: bool = False
@@ -444,9 +444,7 @@ class PythonParser(BaseParser):
 class HiredisParser(BaseParser):
     """Parser class for connections using Hiredis"""
 
-    __slots__ = BaseParser.__slots__ + ("_next_response", "_reader", "_socket_timeout")
-
-    _next_response: bool
+    __slots__ = BaseParser.__slots__ + ("_reader", "_socket_timeout")
 
     def __init__(self, socket_read_size: int):
         if not HIREDIS_AVAILABLE:
@@ -466,23 +464,18 @@ class HiredisParser(BaseParser):
             kwargs["errors"] = connection.encoder.encoding_errors
 
         self._reader = hiredis.Reader(**kwargs)
-        self._next_response = False
         self._socket_timeout = connection.socket_timeout
 
     def on_disconnect(self):
         self._stream = None
         self._reader = None
-        self._next_response = False
 
-    async def can_read(self, timeout: float):
+    async def can_read_destructive(self):
         if not self._stream or not self._reader:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-
-        if self._next_response is False:
-            self._next_response = self._reader.gets()
-        if self._next_response is False:
-            return await self.read_from_socket(timeout=timeout, raise_on_timeout=False)
-        return True
+        if self._reader.gets():
+            return True
+        return await self.read_from_socket(timeout=0, raise_on_timeout=False)
 
     async def read_from_socket(
         self,
@@ -523,12 +516,6 @@ class HiredisParser(BaseParser):
             self.on_disconnect()
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
 
-        # _next_response might be cached from a can_read() call
-        if self._next_response is not False:
-            response = self._next_response
-            self._next_response = False
-            return response
-
         response = self._reader.gets()
         while response is False:
             await self.read_from_socket()
@@ -925,12 +912,10 @@ class Connection:
             self.pack_command(*args), check_health=kwargs.get("check_health", True)
         )
 
-    async def can_read(self, timeout: float = 0):
+    async def can_read_destructive(self):
         """Poll the socket to see if there's data that can be read."""
-        if not self.is_connected:
-            await self.connect()
         try:
-            return await self._parser.can_read(timeout)
+            return await self._parser.can_read_destructive()
         except OSError as e:
             await self.disconnect(nowait=True)
             raise ConnectionError(
@@ -957,6 +942,10 @@ class Connection:
             raise ConnectionError(
                 f"Error while reading from {self.host}:{self.port} : {e.args}"
             )
+        except asyncio.CancelledError:
+            # need this check for 3.7, where CancelledError
+            # is subclass of Exception, not BaseException
+            raise
         except Exception:
             await self.disconnect(nowait=True)
             raise
@@ -1498,12 +1487,12 @@ class ConnectionPool:
             # pool before all data has been read or the socket has been
             # closed. either way, reconnect and verify everything is good.
             try:
-                if await connection.can_read():
+                if await connection.can_read_destructive():
                     raise ConnectionError("Connection has data") from None
             except ConnectionError:
                 await connection.disconnect()
                 await connection.connect()
-                if await connection.can_read():
+                if await connection.can_read_destructive():
                     raise ConnectionError("Connection not ready") from None
         except BaseException:
             # release the connection back to the pool so that we don't
@@ -1699,12 +1688,12 @@ class BlockingConnectionPool(ConnectionPool):
             # pool before all data has been read or the socket has been
             # closed. either way, reconnect and verify everything is good.
             try:
-                if await connection.can_read():
+                if await connection.can_read_destructive():
                     raise ConnectionError("Connection has data") from None
             except ConnectionError:
                 await connection.disconnect()
                 await connection.connect()
-                if await connection.can_read():
+                if await connection.can_read_destructive():
                     raise ConnectionError("Connection not ready") from None
         except BaseException:
             # release the connection back to the pool so that we don't leak it


Test patch:
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index 88cfb1fc..f1bbe422 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -433,7 +433,7 @@ class TestRedisClusterObj:
                     Connection,
                     send_packed_command=mock.DEFAULT,
                     connect=mock.DEFAULT,
-                    can_read=mock.DEFAULT,
+                    can_read_destructive=mock.DEFAULT,
                 ) as mocks:
                     # simulate 7006 as a failed node
                     def execute_command_mock(self, *args, **options):
@@ -473,7 +473,7 @@ class TestRedisClusterObj:
                     execute_command.successful_calls = 0
                     execute_command.failed_calls = 0
                     initialize.side_effect = initialize_mock
-                    mocks["can_read"].return_value = False
+                    mocks["can_read_destructive"].return_value = False
                     mocks["send_packed_command"].return_value = "MOCK_OK"
                     mocks["connect"].return_value = None
                     with mock.patch.object(
@@ -514,7 +514,7 @@ class TestRedisClusterObj:
             send_command=mock.DEFAULT,
             read_response=mock.DEFAULT,
             _connect=mock.DEFAULT,
-            can_read=mock.DEFAULT,
+            can_read_destructive=mock.DEFAULT,
             on_connect=mock.DEFAULT,
         ) as mocks:
             with mock.patch.object(
@@ -546,7 +546,7 @@ class TestRedisClusterObj:
                 mocks["send_command"].return_value = True
                 mocks["read_response"].return_value = "OK"
                 mocks["_connect"].return_value = True
-                mocks["can_read"].return_value = False
+                mocks["can_read_destructive"].return_value = False
                 mocks["on_connect"].return_value = True
 
                 # Create a cluster with reading from replications
diff --git a/tests/test_asyncio/test_connection_pool.py b/tests/test_asyncio/test_connection_pool.py
index d281608e..35f23f44 100644
--- a/tests/test_asyncio/test_connection_pool.py
+++ b/tests/test_asyncio/test_connection_pool.py
@@ -103,7 +103,7 @@ class DummyConnection(Connection):
     async def disconnect(self):
         pass
 
-    async def can_read(self, timeout: float = 0):
+    async def can_read_destructive(self, timeout: float = 0):
         return False
 
 
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 32268fe5..86584e47 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -847,7 +847,7 @@ class TestPubSubAutoReconnect:
                     self.state = 1
                     with mock.patch.object(self.pubsub.connection, "_parser") as m:
                         m.read_response.side_effect = socket.error
-                        m.can_read.side_effect = socket.error
+                        m.can_read_destructive.side_effect = socket.error
                         # wait until task noticies the disconnect until we
                         # undo the patch
                         await self.cond.wait_for(lambda: self.state >= 2)