Allow timeout to work when reading with nowait (#5854) (#7292)

(Note this depends on and extends #5853)

When reading in a loop while the buffer is being constantly filled, the
timeout does not work as there are no calls to `_wait()` where the timer
is used.

I don't know if this edge case is enough to be worried about, but have
put together an initial attempt at fixing it.
I'm not sure if this is really the right solution, but can atleast be
used as as a discussion on ways to improve this.

This can't be backported as this changes the public API (one of the
functions is now async).

Related #5851.

---------

Co-authored-by: pre-commit-ci[bot]
<66853113+pre-commit-ci[bot]@users.noreply.github.com>
(cherry picked from commit 80e2bde)

<!-- Thank you for your contribution! -->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

## Are there changes in behavior for the user?

<!-- Outline any notable behaviour for the end users. -->

## Related issue number

<!-- Are there any issues opened that will be resolved by merging this
change? -->

## Checklist

- [ ] I think the code is well written
- [ ] Unit tests for the changes exist
- [ ] Documentation reflects the changes
- [ ] If you provide code modification, please add yourself to
`CONTRIBUTORS.txt`
  * The format is &lt;Name&gt; &lt;Surname&gt;.
  * Please keep alphabetical order, the file is sorted by names.
- [ ] Add a new news fragment into the `CHANGES` folder
  * name it `<issue_id>.<type>` for example (588.bugfix)
* if you don't have an `issue_id` change it to the pr id after creating
the pr
  * ensure type is one of the following:
    * `.feature`: Signifying a new feature.
    * `.bugfix`: Signifying a bug fix.
    * `.doc`: Signifying a documentation improvement.
    * `.removal`: Signifying a deprecation or removal of public API.
* `.misc`: A ticket has been closed, but it is not of interest to users.
* Make sure to use full sentences with correct case and punctuation, for
example: "Fix issue with non-ascii contents in doctest text files."

CHANGES/5854.bugfix

@@ -0,0 +1 @@
+Fixed client timeout not working when incoming data is always available without waiting -- by :user:`Dreamsorcerer`.

aiohttp/helpers.py

@@ -682,7 +682,8 @@ def __call__(self) -> None:
 
 
 class BaseTimerContext(ContextManager["BaseTimerContext"]):
-    pass
+    def assert_timeout(self) -> None:
+        """Raise TimeoutError if timeout has been exceeded."""
 
 
 class TimerNoop(BaseTimerContext):
@@ -706,6 +707,11 @@ def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
         self._tasks: List[asyncio.Task[Any]] = []
         self._cancelled = False
 
+    def assert_timeout(self) -> None:
+        """Raise TimeoutError if timer has already been cancelled."""
+        if self._cancelled:
+            raise asyncio.TimeoutError from None
+
     def __enter__(self) -> BaseTimerContext:
         task = current_task(loop=self._loop)
 

aiohttp/streams.py

@@ -4,7 +4,7 @@
 from typing import Awaitable, Callable, Deque, Generic, List, Optional, Tuple, TypeVar
 
 from .base_protocol import BaseProtocol
-from .helpers import BaseTimerContext, set_exception, set_result
+from .helpers import BaseTimerContext, TimerNoop, set_exception, set_result
 from .log import internal_logger
 from .typedefs import Final
 
@@ -116,7 +116,7 @@ def __init__(
         self._waiter: Optional[asyncio.Future[None]] = None
         self._eof_waiter: Optional[asyncio.Future[None]] = None
         self._exception: Optional[BaseException] = None
-        self._timer = timer
+        self._timer = TimerNoop() if timer is None else timer
         self._eof_callbacks: List[Callable[[], None]] = []
 
     def __repr__(self) -> str:
@@ -291,10 +291,7 @@ async def _wait(self, func_name: str) -> None:
 
         waiter = self._waiter = self._loop.create_future()
         try:
-            if self._timer:
-                with self._timer:
-                    await waiter
-            else:
+            with self._timer:
                 await waiter
         finally:
             self._waiter = None
@@ -485,8 +482,9 @@ def _read_nowait_chunk(self, n: int) -> bytes:
 
     def _read_nowait(self, n: int) -> bytes:
         """Read not more than n bytes, or whole buffer if n == -1"""
-        chunks = []
+        self._timer.assert_timeout()
 
+        chunks = []
         while self._buffer:
             chunk = self._read_nowait_chunk(n)
             chunks.append(chunk)

tests/test_client_functional.py

@@ -3034,6 +3034,30 @@ async def handler(request):
             await resp.read()
 
 
+async def test_timeout_with_full_buffer(aiohttp_client) -> None:
+    async def handler(request):
+        """Server response that never ends and always has more data available."""
+        resp = web.StreamResponse()
+        await resp.prepare(request)
+        while True:
+            await resp.write(b"1" * 1000)
+            await asyncio.sleep(0.01)
+
+    async def request(client):
+        timeout = aiohttp.ClientTimeout(total=0.5)
+        async with await client.get("/", timeout=timeout) as resp:
+            with pytest.raises(asyncio.TimeoutError):
+                async for data in resp.content.iter_chunked(1):
+                    await asyncio.sleep(0.01)
+
+    app = web.Application()
+    app.add_routes([web.get("/", handler)])
+
+    client = await aiohttp_client(app)
+    # wait_for() used just to ensure that a failing test doesn't hang.
+    await asyncio.wait_for(request(client), 1)
+
+
 async def test_read_bufsize_session_default(aiohttp_client) -> None:
     async def handler(request):
         return web.Response(body=b"1234567")