diff --git a/tests/test_dispatch_stage_threading.py b/tests/test_dispatch_stage_threading.py new file mode 100644 index 00000000..bc1eed9b --- /dev/null +++ b/tests/test_dispatch_stage_threading.py @@ -0,0 +1,141 @@ +# Copyright (c) 2026 Code Infinity +# SPDX-License-Identifier: MPL-2.0 +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at https://mozilla.org/MPL/2.0/. + +"""THE DISPATCH PIPELINE STAGES MUST NOT BLOCK THE EVENT LOOP EITHER. + +ADR-0074 (test_sync_handler_threading.py) threaded the ROUTE HANDLER so a +blocking `def` route no longer freezes the loop. But only the handler was +offloaded: every dispatch STAGE still runs on the loop (server.py, the +`for stage in _*_STAGES` runners). Two of those stages do blocking network I/O +whose cost the handler fix never touched: + + * session LOAD (_autostart_session -> Session.start(sid) -> handler.read) + * session SAVE (_stage_session_save -> session.save() -> handler.write) + +With a network session backend (Redis/Valkey/memcached/MongoDB) each is a +blocking `socket.recv`/`sendall` round-trip ON the loop, so even a perfectly +async route serialises every request through its session I/O. + +These tests PROVE it the way ADR-0074 did — a real request through the REAL +`server.handle` dispatch, a real clock, real threads, and a real file-backed +session handler with a deliberate latency standing in for a slow backend (the +same instrument as ADR-0074's `time.sleep` route; it is a REAL handler doing +REAL file I/O plus a measured delay, not a mock of a backend's protocol). + +They are RED on today's code (stages on the loop) and GREEN once the blocking +stages are offloaded with `asyncio.to_thread`, exactly as the handler was. +""" +import asyncio +import threading +import time + +import pytest + +import tina4_python.core.server as server +from tina4_python.core.request import Request +from tina4_python.core.middleware import Middleware +from tina4_python.core.router import Router, get as route_get +from tina4_python import session as session_mod +from tina4_python.session import Session, FileSessionHandler, session_cookie_name + +DELAY = 0.3 # the latency each session read/write stands in for +N = 4 # concurrent requests for the serialization proof + + +class SlowFileSessionHandler(FileSessionHandler): + """A REAL file-backed session handler with a deliberate per-op latency. + + read()/write() persist to a real file (the parent's behaviour) and then + sleep DELAY to represent a slow network backend's round-trip. The sleep is + a real blocking call - the instrument, not a mock - so a stage that runs it + on the loop genuinely blocks the loop for DELAY, which is the thing measured. + """ + + def read(self, session_id): + time.sleep(DELAY) + return super().read(session_id) + + def write(self, session_id, data, ttl=0): + time.sleep(DELAY) + return super().write(session_id, data, ttl) + + +@pytest.fixture(autouse=True) +def _app(monkeypatch, tmp_path): + Router.clear() + Middleware.reset() + # Inject the REAL slow file handler into the dispatch's own Session() calls. + slow = SlowFileSessionHandler(str(tmp_path)) + monkeypatch.setattr(Session, "_resolve_handler", staticmethod(lambda: slow)) + + @route_get("/sess") + async def _probe(request, response): # async route: the handler is NOT the blocker + request.session.set("touched", "1") # make the save stage persist (blocking write) + return response("ok") + + yield + Router.clear() + Middleware.reset() + + +def _request(): + # A cookie makes autostart do a blocking READ; the route's write makes the + # save stage do a blocking WRITE. Both land in the dispatch, on the loop. + cookie = f"{session_cookie_name()}=abc123".encode() + scope = { + "type": "http", "method": "GET", "path": "/sess", "query_string": b"", + "scheme": "http", "headers": [(b"cookie", cookie)], "client": ("127.0.0.1", 1), + } + return server.handle(Request.from_scope(scope, b"")) + + +def test_session_io_does_not_stall_the_event_loop(): + """While one request's session read+write run, the loop must keep ticking. + + RED today: the session stages run on the loop, so the ticker is frozen for + ~2*DELAY and records almost nothing. GREEN once they are offloaded. + """ + ticks = [] + + async def ticker(): + for _ in range(200): + ticks.append(time.monotonic()) + await asyncio.sleep(0.01) + + async def drive(): + t = asyncio.ensure_future(ticker()) + await asyncio.sleep(0) # let the ticker start + await _request() + t.cancel() + + asyncio.run(drive()) + + assert len(ticks) > 20, ( + f"the loop ticked only {len(ticks)} times while one request did its " + f"session read+write (~{2 * DELAY:.1f}s). The session stages ran ON the " + "loop, so every other in-flight request was frozen with it." + ) + + +def test_concurrent_requests_do_not_serialize_on_session_io(): + """N concurrent requests must overlap their session I/O, not queue on it. + + Serial (today): N * 2 * DELAY. Concurrent (after offloading): ~2 * DELAY. + The assertion sits well below the serial time so it is RED now, GREEN after, + and the ratio it prints is the performance improvement. + """ + async def drive(): + start = time.monotonic() + await asyncio.gather(*(_request() for _ in range(N))) + return time.monotonic() - start + + elapsed = asyncio.run(drive()) + serial = N * 2 * DELAY + assert elapsed < serial * 0.6, ( + f"{N} concurrent requests took {elapsed:.2f}s; serial would be ~{serial:.2f}s " + f"and fully concurrent ~{2 * DELAY:.2f}s. They serialised on session I/O " + "running on the event loop." + ) diff --git a/tina4_python/core/server.py b/tina4_python/core/server.py index 162cdc86..4e8d6513 100644 --- a/tina4_python/core/server.py +++ b/tina4_python/core/server.py @@ -2923,6 +2923,29 @@ def _stage_head_strip(ctx: DispatchContext) -> None: _stage_head_strip, ) +#: Response stages that do BLOCKING I/O and must run off the event loop (via +#: asyncio.to_thread in the dispatch runner), so one request's session write or +#: log write never freezes the loop for every other in-flight request. The rest +#: of _RESPONSE_STAGES is CPU-only and stays on the loop. See the LOAD offload in +#: handle() and the handler threading in _invoke_handler (ADR-0074). +_OFFLOAD_RESPONSE_STAGES = frozenset({_stage_request_log, _stage_session_save}) + + +async def _run_response_stage(stage, ctx: "DispatchContext") -> None: + """Run one response stage, OFF the event loop if it does blocking I/O. + + Session SAVE (backend write) and the request-log file write block; offload + those to a worker thread so one request's I/O never freezes the loop for the + rest. The CPU-only stages (headers, CORS, toolbar, HEAD strip) stay on the + loop. ctx is request-local, so running a stage in a thread is safe. The + branch lives HERE, not in handle(), so the dispatch runner stays flat + (test_dispatch_pipeline::test_the_god_function_does_not_come_back). + """ + if stage in _OFFLOAD_RESPONSE_STAGES: + await asyncio.to_thread(stage, ctx) + else: + stage(ctx) + def _resolve_request_id(request: Request) -> str: """Honour a well-formed inbound X-Request-ID (sanitized), else generate one. @@ -2962,7 +2985,13 @@ async def handle(request: Request) -> Response: from tina4_python.frond.engine import set_request_session, reset_request_session session_token = set_request_session(None) try: - _init_session(request) + # Session LOAD reads the backend (Redis/Valkey/memcached/MongoDB do a + # blocking socket round-trip; the file backend a blocking file read). + # Offload it so it does not freeze the loop — the handler was threaded + # for the same reason (ADR-0074), but only the handler was. The Session + # and its backend connection are request-local, so a worker thread adds + # no cross-request sharing. + await asyncio.to_thread(_init_session, request) # form_token() binds each token to this request's session (CSRF). set_request_session(request.session) @@ -2988,7 +3017,7 @@ async def handle(request: Request) -> Response: break for stage in _RESPONSE_STAGES: - stage(ctx) + await _run_response_stage(stage, ctx) return ctx.response finally: @@ -3825,6 +3854,7 @@ def _find_available_port(start: int, max_tries: int = 10) -> int: import socket try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(("127.0.0.1", start)) s.close() return start