diff --git a/.github/workflows/ci-cd.yml b/.github/workflows/ci-cd.yml index 1c283b68619..90f06d9d492 100644 --- a/.github/workflows/ci-cd.yml +++ b/.github/workflows/ci-cd.yml @@ -558,7 +558,7 @@ jobs: done exit 1 - name: Run benchmarks - uses: CodSpeedHQ/action@v5.0.3 + uses: CodSpeedHQ/action@v5.2.1 with: mode: instrumentation run: python -Im pytest --no-cov -vvvvv --codspeed --durations=30 --timeout=0 diff --git a/CHANGES/13561.contrib.rst b/CHANGES/13561.contrib.rst new file mode 100644 index 00000000000..7066f0b6516 --- /dev/null +++ b/CHANGES/13561.contrib.rst @@ -0,0 +1,3 @@ +Added benchmarks for reading masked WebSocket messages and fixed the +existing read benchmarks, which stopped measuring the parser after the +eighth large frame due to the queue limit -- by :user:`bdraco`. diff --git a/tests/test_benchmarks_http_websocket.py b/tests/test_benchmarks_http_websocket.py index 1ccbca316cb..e19b4e6091c 100644 --- a/tests/test_benchmarks_http_websocket.py +++ b/tests/test_benchmarks_http_websocket.py @@ -5,7 +5,7 @@ import pytest -from aiohttp._websocket.helpers import MSG_SIZE, PACK_LEN3 +from aiohttp._websocket.helpers import MSG_SIZE, PACK_LEN1, PACK_LEN3, websocket_mask from aiohttp._websocket.reader import WebSocketDataQueue from aiohttp.base_protocol import BaseProtocol from aiohttp.helpers import DEFAULT_CHUNK_SIZE @@ -18,27 +18,51 @@ BenchmarkFixture = pytest_codspeed.BenchmarkFixture +# Large enough that a benchmark run never crosses the queue limit; hitting it +# would engage read backpressure and silently stop the parser mid-benchmark. +READ_QUEUE_LIMIT = 2**24 +MASK = b"\x9a\x3c\x71\xe5" + + +def _make_reader(event_loop: asyncio.AbstractEventLoop) -> WebSocketReader: + protocol = BaseProtocol(event_loop) + # A WebSocket connection is always upgraded; without this, backpressure + # would hit ``assert self._parser is not None`` in pause_reading(). + protocol._upgraded = True + queue = WebSocketDataQueue(protocol, READ_QUEUE_LIMIT, loop=event_loop) + return WebSocketReader( + queue, max_msg_size=DEFAULT_CHUNK_SIZE, compress=True, decode_text=True + ) + + +def _masked_frame(opcode: WSMsgType, payload: bytes) -> bytes: + """Build a client-to-server frame with a masked payload.""" + masked = bytearray(payload) + websocket_mask(MASK, masked) + first_byte = 0x80 | opcode.value + length = len(payload) + assert length < 126 or length > 2**16 + if length < 126: + header = PACK_LEN1(first_byte, 0x80 | length) + else: + header = PACK_LEN3(first_byte, 0x80 | 127, length) + return header + MASK + bytes(masked) + + def test_read_large_binary_websocket_messages( event_loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture ) -> None: """Read one hundred large binary websocket messages.""" - queue = WebSocketDataQueue( - BaseProtocol(event_loop), DEFAULT_CHUNK_SIZE, loop=event_loop - ) - reader = WebSocketReader( - queue, max_msg_size=DEFAULT_CHUNK_SIZE, compress=True, decode_text=True - ) - # PACK3 has a minimum message length of 2**16 bytes. message = b"x" * ((2**16) + 1) msg_length = len(message) first_byte = 0x80 | 0 | WSMsgType.BINARY.value header = PACK_LEN3(first_byte, 127, msg_length) raw_message = header + message - feed_data = reader.feed_data @benchmark def _run() -> None: + feed_data = _make_reader(event_loop).feed_data for _ in range(100): feed_data(raw_message) @@ -47,22 +71,42 @@ def test_read_one_hundred_websocket_text_messages( event_loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture ) -> None: """Benchmark reading 100 WebSocket text messages.""" - queue = WebSocketDataQueue( - BaseProtocol(event_loop), DEFAULT_CHUNK_SIZE, loop=event_loop - ) - reader = WebSocketReader( - queue, max_msg_size=DEFAULT_CHUNK_SIZE, compress=True, decode_text=True - ) raw_message = ( b'\x81~\x01!{"id":1,"src":"shellyplugus-c049ef8c30e4","dst":"aios-1453812500' b'8","result":{"name":null,"id":"shellyplugus-c049ef8c30e4","mac":"C049EF8C30E' b'4","slot":1,"model":"SNPL-00116US","gen":2,"fw_id":"20231219-133953/1.1.0-g3' b'4b5d4f","ver":"1.1.0","app":"PlugUS","auth_en":false,"auth_domain":null}}' ) - feed_data = reader.feed_data @benchmark def _run() -> None: + feed_data = _make_reader(event_loop).feed_data + for _ in range(100): + feed_data(raw_message) + + +def test_read_one_hundred_masked_websocket_text_messages( + event_loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture +) -> None: + """Read 100 small masked text messages, as a server receives them.""" + raw_message = _masked_frame(WSMsgType.TEXT, b'{"id":1,"type":"ping"}') + + @benchmark + def _run() -> None: + feed_data = _make_reader(event_loop).feed_data + for _ in range(100): + feed_data(raw_message) + + +def test_read_one_hundred_masked_large_binary_websocket_messages( + event_loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture +) -> None: + """Read 100 large masked binary messages, as a server receives them.""" + raw_message = _masked_frame(WSMsgType.BINARY, b"x" * ((2**16) + 1)) + + @benchmark + def _run() -> None: + feed_data = _make_reader(event_loop).feed_data for _ in range(100): feed_data(raw_message) diff --git a/tests/test_web_server.py b/tests/test_web_server.py index 25c9b14f9bb..b04815539be 100644 --- a/tests/test_web_server.py +++ b/tests/test_web_server.py @@ -296,10 +296,12 @@ async def handler(request: web.BaseRequest) -> NoReturn: async def test_handler_cancellation(unused_port_socket: socket.socket) -> None: event = asyncio.Event() + started = asyncio.Event() sock = unused_port_socket port = sock.getsockname()[1] async def on_request(request: web.Request) -> web.Response: + started.set() try: await asyncio.sleep(10) except asyncio.CancelledError: @@ -320,11 +322,11 @@ async def on_request(request: web.Request) -> web.Response: try: assert runner.server.handler_cancellation, "Flag was not propagated" - async with client.ClientSession( - timeout=client.ClientTimeout(total=0.15) - ) as sess: - with pytest.raises(asyncio.TimeoutError): - await sess.get(f"http://127.0.0.1:{port}/") + _, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write(b"GET / HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n") + await writer.drain() + await asyncio.wait_for(started.wait(), timeout=5) + writer.close() with suppress(asyncio.TimeoutError): await asyncio.wait_for(event.wait(), timeout=1) @@ -336,13 +338,12 @@ async def on_request(request: web.Request) -> web.Response: async def test_no_handler_cancellation(unused_port_socket: socket.socket) -> None: timeout_event = asyncio.Event() done_event = asyncio.Event() + started = asyncio.Event() sock = unused_port_socket port = sock.getsockname()[1] - started = False async def on_request(request: web.Request) -> web.Response: - nonlocal started - started = True + started.set() await asyncio.wait_for(timeout_event.wait(), timeout=5) done_event.set() return web.Response() @@ -357,17 +358,16 @@ async def on_request(request: web.Request) -> web.Response: await site.start() try: - async with client.ClientSession( - timeout=client.ClientTimeout(total=0.2) - ) as sess: - with pytest.raises(asyncio.TimeoutError): - await sess.get(f"http://127.0.0.1:{port}/") + _, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write(b"GET / HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n") + await writer.drain() + await asyncio.wait_for(started.wait(), timeout=5) + writer.close() await asyncio.sleep(0.1) timeout_event.set() with suppress(asyncio.TimeoutError): await asyncio.wait_for(done_event.wait(), timeout=1) - assert started assert done_event.is_set() finally: await asyncio.gather(runner.shutdown(), site.stop())