Skip to content

Commit

Permalink
Allow timeout to work when reading with nowait (#5854)
Browse files Browse the repository at this point in the history
(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>
  • Loading branch information
Dreamsorcerer and pre-commit-ci[bot] authored May 16, 2023
1 parent 3187f6f commit 80e2bde
Show file tree
Hide file tree
Showing 4 changed files with 37 additions and 8 deletions.
1 change: 1 addition & 0 deletions CHANGES/5854.bugfix
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fixed client timeout not working when incoming data is always available without waiting -- by :user:`Dreamsorcerer`.
8 changes: 7 additions & 1 deletion aiohttp/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -676,7 +676,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):
Expand All @@ -700,6 +701,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 = asyncio.current_task(loop=self._loop)

Expand Down
12 changes: 5 additions & 7 deletions aiohttp/streams.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
from typing_extensions import Final

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

try: # pragma: no cover
Expand Down Expand Up @@ -122,7 +122,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:
Expand Down Expand Up @@ -297,10 +297,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
Expand Down Expand Up @@ -477,8 +474,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)
Expand Down
24 changes: 24 additions & 0 deletions tests/test_client_functional.py
Original file line number Diff line number Diff line change
Expand Up @@ -3022,6 +3022,30 @@ async def handler(request):
await resp.read()


async def test_timeout_with_full_buffer(aiohttp_client: Any) -> 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: Any) -> None:
async def handler(request):
return web.Response(body=b"1234567")
Expand Down

0 comments on commit 80e2bde

Please sign in to comment.