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

Merged
merged 18 commits into from
May 16, 2023
Merged
Show file tree
Hide file tree
Changes from 14 commits
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
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 check_timeout(self) -> None:
Dreamsorcerer marked this conversation as resolved.
Show resolved Hide resolved
"""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 check_timeout(self) -> None:
"""Raise TimeoutError if timer has already been cancelled."""
if self._cancelled:
raise asyncio.TimeoutError from None
Dreamsorcerer marked this conversation as resolved.
Show resolved Hide resolved

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 = timer if timer else TimerNoop()
Dreamsorcerer marked this conversation as resolved.
Show resolved Hide resolved
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.check_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)
with pytest.raises(asyncio.TimeoutError):
async with await client.get("/", timeout=timeout) as resp:
Dreamsorcerer marked this conversation as resolved.
Show resolved Hide resolved
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() timeout should not be reached.
Dreamsorcerer marked this conversation as resolved.
Show resolved Hide resolved
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