# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""GenroAsgiWorker - a minimal single-app ASGI server.
The unit a commander spawns to distribute load: one uvicorn process serving
exactly one app. Unlike :class:`AsgiServer` it owns no multi-app registry,
middleware/plugin system, sessions, storage or config orchestration — only the
bare minimum to serve one app:
- a ``Dispatcher`` (which, with a single app on the empty mount, always resolves
to that app),
- a ``ServerLifespan`` (so the app's ``on_startup``/``on_shutdown`` run),
- a ``WsxHandler`` terminating the websocket connections the commander pipes
down (issue #44): the standing connection lives in the process that owns the
user's pages, so server-initiated push is the worker's own business,
- the dual parent-child relationship with the app (``app.server -> worker``),
- the app-facing server surface the app's CANONICAL dispatch needs: a
``RequestRegistry``, an empty ``db_registry`` and a None-returning
``authenticate`` (no server-level auth: whoever fronts the worker owns it).
The app is mounted on the empty mount ('') so the dispatcher's root fallback
always reaches it. It is the executor in the daemon-less worker model: the
commander forwards requests to it over HTTP; it just runs the app.
Two serving paths:
- ``run()`` — the blocking CLI path (``uvicorn.run``), unchanged;
- ``start()`` / ``wait_stopped()`` / ``request_stop()`` — the async path for a
parent that interleaves the serving with other tasks (the pool child entry):
uvicorn runs as a task and, with ``port=0``, the OS-assigned port is read
back from the bound socket, so the worker can announce where it listens.
"""
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING, Any
import uvicorn
from genro_routes import RoutingClass # type: ignore[import-untyped]
from .dispatcher import Dispatcher
from ..executors import ThreadExecutor
from ..lifespan import ServerLifespan
from ..request import RequestRegistry
from ..wsx import WsxHandler
if TYPE_CHECKING:
from ..types import Receive, Scope, Send
__all__ = ["GenroAsgiWorker"]
[docs]
class GenroAsgiWorker(RoutingClass):
"""Minimal ASGI server hosting a single app, run by uvicorn.
Attributes:
app: The single mounted application (an ASGI callable).
apps: One-entry dict {'': app}, so Dispatcher/ServerLifespan — which
iterate ``server.apps`` — see the app on the root fallback mount.
host: Bind host.
port: Bind port.
logger: Logger for worker messages.
lifespan: ServerLifespan driving the app's on_startup/on_shutdown.
dispatcher: Dispatcher delegating every request to the single app.
wsx_handler: WsxHandler terminating websocket connections (the
dispatcher's fallback for an app with no ``serve_websocket`` hook).
executor: Owned ThreadExecutor — the dispatch for blocking work, whose
metrics report this worker's pressure (busy/total/queue/occupancy).
The app reaches it via the dual relationship (``app.server.executor``).
request_registry: RequestRegistry the app's canonical dispatch creates
and unregisters its requests on.
db_registry: Always empty — the minimal worker registers no databases;
the app's ``get_db`` falls back to its own registry.
server: The uvicorn Server of the async path (None until ``start()``).
serve_task: The task running ``server.serve()`` on the async path.
"""
__slots__ = (
"app",
"apps",
"host",
"port",
"logger",
"lifespan",
"dispatcher",
"wsx_handler",
"executor",
"request_registry",
"db_registry",
"server",
"serve_task",
)
[docs]
def __init__(
self,
app: Any,
host: str = "127.0.0.1",
port: int = 8000,
max_workers: int | None = None,
) -> None:
"""Initialize the worker around a single app.
Args:
app: The application to serve (mounted on the empty mount).
host: Bind host.
port: Bind port.
max_workers: Thread pool size for the dispatch executor (default:
ThreadPoolExecutor's own, ``min(32, cpu + 4)``).
"""
self.app = app
self.host = host
self.port = port
self.logger = logging.getLogger("genro_asgi.worker")
self.executor = ThreadExecutor(name="worker", max_workers=max_workers)
self.request_registry = RequestRegistry()
self.db_registry: dict[str, Any] = {}
self.lifespan = ServerLifespan(self)
self.dispatcher = Dispatcher(self)
self.wsx_handler = WsxHandler(self)
app.mount_name = ""
self.attach_instance(app) # dual relationship: app.server -> self
self.apps = {"": app}
self.server: uvicorn.Server | None = None
self.serve_task: asyncio.Task[None] | None = None
[docs]
def authenticate(self, scope: Scope) -> dict[str, Any] | None:
"""No server-level auth on the minimal worker: every identity is None.
The app's canonical dispatch delegates here (the app-facing server
protocol); whoever fronts the worker (the commander, a gateway) owns the
authentication of what it forwards.
"""
return None
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Handle an ASGI event: lifespan to ServerLifespan, the rest to the
Dispatcher (websocket scopes fall back to the WsxHandler there).
``/_metrics`` (and ``/_commands/...``) are served by the mounted app's
own canonical routing: they reach the Dispatcher like any other request
and fall through to the single app on the empty mount.
"""
if scope["type"] == "lifespan":
await self.lifespan(scope, receive, send)
else:
await self.dispatcher(scope, receive, send)
[docs]
def run(self) -> None:
"""Run the worker using uvicorn (single process, no reload)."""
self.logger.info(f"Starting worker on {self.host}:{self.port}")
try:
uvicorn.run(self, host=self.host, port=self.port)
finally:
self.executor.shutdown()
[docs]
async def start(self) -> None:
"""Serve as a task in the running loop; ``self.port`` becomes the bound port.
The async counterpart of ``run()``: uvicorn's ``serve()`` runs as a task
and this returns once the server has started, with the OS-assigned port
(``port=0``) read back from the bound socket — the same discovery
``ChannelHub.start()`` does for its TCP variant.
"""
self.server = uvicorn.Server(uvicorn.Config(self, host=self.host, port=self.port))
self.serve_task = asyncio.create_task(self.server.serve())
while not self.server.started:
if self.serve_task.done():
self.serve_task.result()
raise RuntimeError("uvicorn stopped before startup completed")
await asyncio.sleep(0.02)
self.port = self.server.servers[0].sockets[0].getsockname()[1]
self.logger.info(f"Worker serving on {self.host}:{self.port}")
[docs]
def request_stop(self) -> None:
"""Ask the serving task to exit (uvicorn drains connections and returns)."""
if self.server is None:
raise RuntimeError("worker not started")
self.server.should_exit = True
[docs]
async def wait_stopped(self) -> None:
"""Block until the serving task ends, then release the dispatch executor."""
if self.serve_task is None:
raise RuntimeError("worker not started")
try:
await self.serve_task
finally:
self.executor.shutdown()
def __repr__(self) -> str:
return f"GenroAsgiWorker(app={type(self.app).__name__}, {self.host}:{self.port})"
if __name__ == "__main__":
pass