Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
Allow timeout to work when reading with nowait (#5854)
(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)
Dreamsorcerer committed May 16, 2023
commit 5359a0aec1891aaab62b741ae9d009b5e4a8c56d
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
@@ -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)

12 changes: 5 additions & 7 deletions aiohttp/streams.py
Original file line number Diff line number Diff line change
@@ -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)
24 changes: 24 additions & 0 deletions tests/test_client_functional.py
Original file line number Diff line number Diff line change
@@ -3034,6 +3034,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) -> None:
async def handler(request):
return web.Response(body=b"1234567")