diff --git a/docs/asgi.md b/docs/asgi.md index a0f49d2..529d768 100644 --- a/docs/asgi.md +++ b/docs/asgi.md @@ -8,9 +8,9 @@ The module `oxyroute.asgi` builds a minimal RSGI-shaped `scope` / `protocol` and ## How responses get back to ASGI -The RSGI response helpers on the protocol object are **synchronous** from the Rust side. The ASGI implementation schedules `send` coroutines on the **ASGI process’s** event loop: that loop is captured in the protocol and native code uses `asyncio.run_coroutine_threadsafe` + `result()` to run each `send` on it. +The RSGI response helpers on the protocol object are **synchronous** from the Rust side. The ASGI implementation bridges this through a thread-safe outgoing queue: sync `response_*` calls enqueue ASGI messages from the worker thread, and an async drain task on the main loop performs `send(...)` in order. -**Why the bridge does not `await` on that loop directly:** Awaiting the native `handle_rsgi` coroutine on the same loop that must drain `run_coroutine_threadsafe` can **deadlock** (the loop is blocked in `await` and never runs the scheduled `send` tasks). The bridge therefore runs `handle_rsgi` in a **thread-pool** worker using `asyncio.run()` on a **separate** event loop, while the protocol still targets the main ASGI loop for `send` scheduling. +**Why the bridge does not `await` on that loop directly:** Awaiting the native `handle_rsgi` coroutine on the same loop that must process outgoing messages can **deadlock**. The bridge therefore runs `handle_rsgi` in a **thread-pool** worker using `asyncio.run()` on a **separate** event loop, while the main ASGI loop drains the queued `http.response.*` events. **Implications:** This bridge is a **practical adapter**, not a full reimplementation of RSGI under every ASGI host. Prefer **RSGI/Granian** for production with OxyRoute if you can. With **Uvicorn**, use a **single worker** process unless you are sure the combination is safe in your app (the usual `workers=N` + in-process `asyncio` footguns can still apply to third-party `asyncio` use). diff --git a/oxyroute/asgi.py b/oxyroute/asgi.py index 811d19f..99716de 100644 --- a/oxyroute/asgi.py +++ b/oxyroute/asgi.py @@ -168,18 +168,22 @@ def _norm_headers_asgi( class _RsgiProtocol: - __slots__ = ("_body", "_loop", "_send", "_status", "status") + __slots__ = ("_body", "_loop", "_queue", "_status", "status") - def __init__(self, body: bytes, send: Any, main_loop: asyncio.AbstractEventLoop) -> None: + def __init__( + self, + body: bytes, + queue: asyncio.Queue[dict[str, Any] | None], + main_loop: asyncio.AbstractEventLoop, + ) -> None: self._body = body - self._send = send + self._queue = queue self._loop = main_loop self._status: int | None = None self.status = 200 - def _run_send(self, coro: Any) -> None: - f = asyncio.run_coroutine_threadsafe(coro, self._loop) - f.result() + def _enqueue(self, message: dict[str, Any]) -> None: + self._loop.call_soon_threadsafe(self._queue.put_nowait, message) def __call__(self) -> Any: async def _body() -> bytes: @@ -192,66 +196,54 @@ def response_str(self, status: int, headers: list, body: str) -> None: self.status = int(status) b = body.encode("utf-8") h = _norm_headers_asgi(headers) - - async def _go() -> None: - await self._send( - { - "type": "http.response.start", - "status": int(status), - "headers": h, - } - ) - await self._send( - { - "type": "http.response.body", - "body": b, - } - ) - - self._run_send(_go()) + self._enqueue( + { + "type": "http.response.start", + "status": int(status), + "headers": h, + } + ) + self._enqueue( + { + "type": "http.response.body", + "body": b, + } + ) def response_bytes(self, status: int, headers: list, body: bytes) -> None: self._status = int(status) self.status = int(status) h = _norm_headers_asgi(headers) - - async def _go() -> None: - await self._send( - { - "type": "http.response.start", - "status": int(status), - "headers": h, - } - ) - await self._send( - { - "type": "http.response.body", - "body": body, - } - ) - - self._run_send(_go()) + self._enqueue( + { + "type": "http.response.start", + "status": int(status), + "headers": h, + } + ) + self._enqueue( + { + "type": "http.response.body", + "body": body, + } + ) def response_empty(self, status: int, headers: list) -> None: self._status = int(status) self.status = int(status) h = _norm_headers_asgi(headers) - - async def _go() -> None: - await self._send( - { - "type": "http.response.start", - "status": int(status), - "headers": h, - } - ) - await self._send( - { - "type": "http.response.body", - } - ) - - self._run_send(_go()) + self._enqueue( + { + "type": "http.response.start", + "status": int(status), + "headers": h, + } + ) + self._enqueue( + { + "type": "http.response.body", + } + ) async def asgi_to_rsgi( @@ -295,7 +287,17 @@ async def asgi_to_rsgi( hdrs, ) loop = asyncio.get_running_loop() - proto = _RsgiProtocol(body, send, loop) + queue: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue() + + async def _drain_outgoing() -> None: + while True: + msg = await queue.get() + if msg is None: + return + await send(msg) + + drain_task = asyncio.create_task(_drain_outgoing()) + proto = _RsgiProtocol(body, queue, loop) await loop.run_in_executor( None, _run_handle_rsgi_blocking, @@ -303,6 +305,8 @@ async def asgi_to_rsgi( rscope, proto, ) + await queue.put(None) + await drain_task def build_asgi_caller(framework_app: Any) -> Callable[..., Any]: