Don't ceil small timeouts (#5091)

CHANGES/4850.feature

@@ -0,0 +1 @@
+Don't ceil timeouts that are smaller than 5 seconds.

aiohttp/helpers.py

@@ -494,18 +494,20 @@ def _weakref_handle(info):  # type: ignore
             getattr(ob, name)()
 
 
-def weakref_handle(ob, name, timeout, loop, ceil_timeout=True):  # type: ignore
+def weakref_handle(ob, name, timeout, loop):  # type: ignore
     if timeout is not None and timeout > 0:
         when = loop.time() + timeout
-        if ceil_timeout:
+        if timeout >= 5:
             when = ceil(when)
 
         return loop.call_at(when, _weakref_handle, (weakref.ref(ob), name))
 
 
 def call_later(cb, timeout, loop):  # type: ignore
     if timeout is not None and timeout > 0:
-        when = ceil(loop.time() + timeout)
+        when = loop.time() + timeout
+        if timeout > 5:
+            when = ceil(when)
         return loop.call_at(when, cb)
 
 
@@ -527,9 +529,12 @@ def close(self) -> None:
         self._callbacks.clear()
 
     def start(self) -> Optional[asyncio.Handle]:
-        if self._timeout is not None and self._timeout > 0:
-            at = ceil(self._loop.time() + self._timeout)
-            return self._loop.call_at(at, self.__call__)
+        timeout = self._timeout
+        if timeout is not None and timeout > 0:
+            when = self._loop.time() + timeout
+            if timeout >= 5:
+                when = ceil(when)
+            return self._loop.call_at(when, self.__call__)
         else:
             return None
 
@@ -612,8 +617,13 @@ def __enter__(self) -> async_timeout.timeout:
             if self._task is None:
                 raise RuntimeError(
                     'Timeout context manager should be used inside a task')
+            now = self._loop.time()
+            delay = self._timeout
+            when = now + delay
+            if delay > 5:
+                when = ceil(when)
             self._cancel_handler = self._loop.call_at(
-                ceil(self._loop.time() + self._timeout), self._cancel_task)
+                when, self._cancel_task)
         return self
 
 

docs/client_quickstart.rst

@@ -453,3 +453,20 @@ Thus the default timeout is::
 
    aiohttp.ClientTimeout(total=5*60, connect=None,
                          sock_connect=None, sock_read=None)
+
+.. note::
+
+   *aiohttp* **ceils** timeout if the value is equal or greater than 5
+   seconds. The timeout expires at the next integer second greater than
+   ``current_time + timeout``.
+
+   The ceiling is done for the sake of optimization, when many concurrent tasks
+   are scheduled to wake-up at the almost same but different absolute times. It
+   leads to very many event loop wakeups, which kills performance.
+
+   The optimization shifts absolute wakeup times by scheduling them to exactly
+   the same time as other neighbors, the loop wakes up once-per-second for
+   timeout expiration.
+
+   Smaller timeouts are not rounded to help testing; in the real life network
+   timeouts usually greater than tens of seconds.

docs/spelling_wordlist.txt

@@ -309,6 +309,8 @@ utils
 uvloop
 vcvarsall
 waituntil
+wakeup
+wakeups
 webapp
 websocket
 websocket’s

requirements/ci-wheel.txt

@@ -18,8 +18,8 @@ yarl==1.4.2
 
 # required c-ares will not build on windows and has build problems on Macos Python<3.7
 aiodns==2.0.0; sys_platform=="linux" or sys_platform=="darwin" and python_version>="3.7"
-cryptography==2.9.2; platform_machine!="i686" and python_version<"3.8" # no 32-bit wheels; no python 3.9 wheels yet
+cryptography==2.9.2; platform_machine!="i686" and python_version<"3.9" # no 32-bit wheels; no python 3.9 wheels yet
 trustme==0.6.0; platform_machine!="i686"    # no 32-bit wheels
 codecov==2.1.10
-uvloop==0.12.1; platform_system!="Windows" and implementation_name=="cpython" and python_version<"3.7" # MagicStack/uvloop#14
+uvloop==0.14.0; platform_system!="Windows" and implementation_name=="cpython" and python_version<"3.9" # MagicStack/uvloop#14
 idna-ssl==1.1.0; python_version<"3.7"

tests/test_client_functional.py

@@ -30,10 +30,6 @@ def fname(here):
     return here / 'conftest.py'
 
 
-def ceil(val):
-    return val
-
-
 async def test_keepalive_two_requests_success(
         aiohttp_client) -> None:
     async def handler(request):
@@ -570,7 +566,6 @@ async def handler(request):
 
 
 async def test_timeout_on_reading_headers(aiohttp_client, mocker) -> None:
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
 
     async def handler(request):
         resp = web.StreamResponse()
@@ -589,8 +584,6 @@ async def handler(request):
 async def test_timeout_on_conn_reading_headers(aiohttp_client, mocker) -> None:
     # tests case where user did not set a connection timeout
 
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
-
     async def handler(request):
         resp = web.StreamResponse()
         await asyncio.sleep(0.1)
@@ -608,7 +601,6 @@ async def handler(request):
 
 
 async def test_timeout_on_session_read_timeout(aiohttp_client, mocker) -> None:
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
 
     async def handler(request):
         resp = web.StreamResponse()
@@ -630,7 +622,6 @@ async def handler(request):
 
 
 async def test_read_timeout_between_chunks(aiohttp_client, mocker) -> None:
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
 
     async def handler(request):
         resp = aiohttp.web.StreamResponse()
@@ -656,7 +647,6 @@ async def handler(request):
 
 
 async def test_read_timeout_on_reading_chunks(aiohttp_client, mocker) -> None:
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
 
     async def handler(request):
         resp = aiohttp.web.StreamResponse()
@@ -682,7 +672,6 @@ async def handler(request):
 async def test_timeout_on_reading_data(aiohttp_client, mocker) -> None:
     loop = asyncio.get_event_loop()
 
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
     fut = loop.create_future()
 
     async def handler(request):
@@ -704,7 +693,6 @@ async def handler(request):
 
 
 async def test_timeout_none(aiohttp_client, mocker) -> None:
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
 
     async def handler(request):
         resp = web.StreamResponse()

tests/test_client_ws_functional.py

@@ -7,14 +7,6 @@
 from aiohttp import hdrs, web
 
 
-@pytest.fixture
-def ceil(mocker):
-    def ceil(val):
-        return val
-
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
-
-
 async def test_send_recv_text(aiohttp_client) -> None:
 
     async def handler(request):
@@ -528,7 +520,7 @@ async def handler(request):
     await resp.close()
 
 
-async def test_heartbeat(aiohttp_client, ceil) -> None:
+async def test_heartbeat(aiohttp_client) -> None:
     ping_received = False
 
     async def handler(request):
@@ -553,7 +545,7 @@ async def handler(request):
     assert ping_received
 
 
-async def test_heartbeat_no_pong(aiohttp_client, ceil) -> None:
+async def test_heartbeat_no_pong(aiohttp_client) -> None:
     ping_received = False
 
     async def handler(request):

tests/test_helpers.py

@@ -4,6 +4,7 @@
 import os
 import platform
 import tempfile
+from math import modf
 from unittest import mock
 
 import pytest
@@ -295,6 +296,18 @@ def test_timeout_handle(loop) -> None:
     assert not handle._callbacks
 
 
+def test_when_timeout_smaller_second(loop) -> None:
+    timeout = 0.1
+    timer = loop.time() + timeout
+
+    handle = helpers.TimeoutHandle(loop, timeout)
+    when = handle.start()._when
+    handle.close()
+
+    assert isinstance(when, float)
+    assert f"{when:.3f}" == f"{timer:.3f}"
+
+
 def test_timeout_handle_cb_exc(loop) -> None:
     handle = helpers.TimeoutHandle(loop, 10.2)
     cb = mock.Mock()
@@ -333,14 +346,14 @@ def test_timer_context_no_task(loop) -> None:
 
 async def test_weakref_handle(loop) -> None:
     cb = mock.Mock()
-    helpers.weakref_handle(cb, 'test', 0.01, loop, False)
+    helpers.weakref_handle(cb, 'test', 0.01, loop)
     await asyncio.sleep(0.1)
     assert cb.test.called
 
 
 async def test_weakref_handle_weak(loop) -> None:
     cb = mock.Mock()
-    helpers.weakref_handle(cb, 'test', 0.01, loop, False)
+    helpers.weakref_handle(cb, 'test', 0.01, loop)
     del cb
     gc.collect()
     await asyncio.sleep(0.1)
@@ -373,6 +386,19 @@ def test_ceil_timeout_no_task(loop) -> None:
             pass
 
 
+async def test_ceil_timeout_round(loop) -> None:
+    with helpers.CeilTimeout(7.5, loop=loop) as cm:
+        frac, integer = modf(cm._cancel_handler.when())
+        assert frac == 0
+
+
+async def test_ceil_timeout_small(loop) -> None:
+    with helpers.CeilTimeout(1.1, loop=loop) as cm:
+        frac, integer = modf(cm._cancel_handler.when())
+        # a chance for exact integer with zero fraction is negligible
+        assert frac != 0
+
+
 # -------------------------------- ContentDisposition -------------------
 
 def test_content_disposition() -> None:

tests/test_web_protocol.py

@@ -97,14 +97,6 @@ def write(chunk):
     return transport
 
 
-@pytest.fixture
-def ceil(mocker):
-    def ceil(val):
-        return val
-
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
-
-
 async def test_shutdown(srv, transport) -> None:
     loop = asyncio.get_event_loop()
     assert transport is srv.transport
@@ -427,7 +419,7 @@ async def handle_request(request):
 
 
 async def test_lingering_timeout(
-    make_srv, transport, ceil, request_handler
+    make_srv, transport, request_handler
 ):
 
     async def handle_request(request):
@@ -516,7 +508,7 @@ async def test_handle_400(srv, buf, transport) -> None:
     assert b'400 Bad Request' in buf
 
 
-async def test_keep_alive(make_srv, transport, ceil) -> None:
+async def test_keep_alive(make_srv, transport) -> None:
     loop = asyncio.get_event_loop()
     srv = make_srv(keepalive_timeout=0.05)
     future = loop.create_future()

tests/test_web_websocket_functional.py

@@ -9,14 +9,6 @@
 from aiohttp.http import WSMsgType
 
 
-@pytest.fixture
-def ceil(mocker):
-    def ceil(val):
-        return val
-
-    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
-
-
 async def test_websocket_can_prepare(loop, aiohttp_client) -> None:
 
     async def handler(request):
@@ -516,7 +508,7 @@ async def handler(request):
     await closed
 
 
-async def aiohttp_client_close_handshake(loop, aiohttp_client, ceil):
+async def aiohttp_client_close_handshake(loop, aiohttp_client):
 
     closed = loop.create_future()
 
@@ -633,7 +625,7 @@ async def handler(request):
     assert raised
 
 
-async def test_heartbeat(loop, aiohttp_client, ceil) -> None:
+async def test_heartbeat(loop, aiohttp_client) -> None:
 
     async def handler(request):
         ws = web.WebSocketResponse(heartbeat=0.05)
@@ -654,7 +646,7 @@ async def handler(request):
     await ws.close()
 
 
-async def test_heartbeat_no_pong(loop, aiohttp_client, ceil) -> None:
+async def test_heartbeat_no_pong(loop, aiohttp_client) -> None:
 
     async def handler(request):
         ws = web.WebSocketResponse(heartbeat=0.05)