src/genro_asgi/pool.py

Source from this local checkout, regenerated when the reader rebuilds. Line links use #L<number>; a GitHub line range opens its first line.

1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 #     https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """The server's single thread pool for blocking work (SPECIFICATION.md §4, D2).16 17 ``WorkPool`` wraps one ``concurrent.futures.ThreadPoolExecutor`` and is held by18 the server as a dual parent-child (``self.server``). Async handlers stay on the19 event loop; only sync handlers reach the pool, dispatched through20 ``BaseServer.run_sync`` via ``loop.run_in_executor``.21 22 Lazily provisioned (invariant #1 — build lazily on the running loop): the23 executor is created on the first dispatch, never at boot, and torn down at24 lifespan shutdown only if it was ever provisioned.25 """26 27 from __future__ import annotations28 29 import asyncio30 import contextvars31 import os32 from concurrent.futures import ThreadPoolExecutor33 from typing import TYPE_CHECKING, Any, Callable34 35 if TYPE_CHECKING:36     from .server import BaseServer37 38 __all__ = ["WorkPool"]39 40 41 class WorkPool:42     """One thread pool for blocking (sync) handlers, owned by the server.43 44     Constructor kwarg: ``max_threads`` — the executor's ``max_workers``45     (``None`` uses the stdlib default: ``min(32, cpus + 4)``, where ``cpus``46     are the CPUs granted to the process on interpreters that have47     ``os.process_cpu_count``, the machine's on older ones). Threads are48     named ``genro-pool*`` so a handler can assert it ran off the loop.49     """50 51     def __init__(self, server: BaseServer, max_threads: int | None = None) -> None:52         self.server = server53         self._max_threads = max_threads54         self._executor: ThreadPoolExecutor | None = None55         self._busy = 056         self._total = 057 58     @property59     def provisioned(self) -> bool:60         """Whether the executor exists yet (a sync dispatch has happened)."""61         return self._executor is not None62 63     @property64     def metrics(self) -> dict[str, int]:65         """Pressure gauges of the pool: the slots that exist, the calls in flight.66 67         ``busy`` counts every ``run()`` entered and not yet exited — DEMAND, not68         slots held: past saturation the excess is queued inside the executor and69         still counts, so ``busy`` can exceed ``total`` (the consumers clamp).70 71         Zeros until the executor is provisioned — before the first sync dispatch72         there is nothing to measure, and reporting the configured size of a pool73         that does not exist would read as pressure that isn't there.74 75         ``total`` mirrors our own argument resolution, frozen at provision —76         never a private executor attribute.77         """78         if not self.provisioned:79             return {"total": 0, "busy": 0}80         return {"total": self._total, "busy": self._busy}81 82     @property83     def executor(self) -> ThreadPoolExecutor:84         """The pool's executor, created on first access (lazy provisioning).85 86         The moment of truth for ``total``: the slot count is resolved and87         frozen HERE, where the stdlib takes the same decision — from the CPUs88         granted to the process (``os.process_cpu_count``) on interpreters that89         have it, the machine's otherwise, exactly mirroring the executor's own90         default on each. A later affinity change cannot move the threads the91         pool already built, so the frozen number stays the true one.92         """93         if self._executor is None:94             cpus = getattr(os, "process_cpu_count", os.cpu_count)() or 195             self._total = self._max_threads if self._max_threads is not None else min(32, cpus + 4)96             self._executor = ThreadPoolExecutor(97                 max_workers=self._max_threads,98                 thread_name_prefix="genro-pool",99             )100         return self._executor101 102     async def run(self, fn: Callable[..., Any], *args: Any) -> Any:103         """Run blocking ``fn`` on a pool thread, provisioning on first call.104 105         The caller's context is copied into the worker thread (what106         ``asyncio.to_thread`` does), so a sync handler sees the loop-side107         ContextVars — e.g. the registry's current request.108         """109         loop = asyncio.get_running_loop()110         ctx = contextvars.copy_context()111         self._busy += 1112         try:113             return await loop.run_in_executor(self.executor, ctx.run, fn, *args)114         finally:115             self._busy -= 1116 117     def shutdown(self, wait: bool = True) -> None:118         """Tear the executor down — a no-op if it was never provisioned.119 120         Resets the lazy slot so a later dispatch re-provisions: a server121         reused through repeated ``serve()`` rounds keeps working.122         """123         if self.provisioned:124             self.executor.shutdown(wait=wait)125             self._executor = None126             self._total = 0