src/genro_asgi_multiworker_spa/orchestration/worker_entry.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 worker process: one child, one wire, one worker, nothing else.16 17 ``python -m genro_asgi_multiworker_spa.orchestration.worker_entry`` is what a WorkerHandler18 spawns. The child is **monolingual**: it speaks only its handler's socket — no19 HTTP port, no uvicorn, no second door. Its whole configuration arrives in the20 ``GENRO_ASGI_WORKER`` environment variable as one JSON object, the seven keys21 the handler writes::22 23 {"name": "standard_0001",24 "uds_url": "uds:/tmp/gnr_x/standard_0001.sock",25 "frozen_users_path": "/var/lib/gnr/frozen_users",26 "main_threadpool_size": 8,27 "aux_threadpool_size": 2,28 "worker_class": "genro_asgi_multiworker_spa.orchestration.spa_worker:SpaWorker",29 "kwargs": {"group": "standard"}}30 31 ``name``, ``uds_url`` and ``frozen_users_path`` are mandatory — the handler32 always knows all three — and a missing or malformed variable is a spawn33 contract violation: the process says so and exits, never guesses a default.34 ``worker_class`` is a ``module.path:ClassName`` reference and ``kwargs`` is the35 grammar that class was configured with; the real deployment names a subclass,36 because the base worker hosts no site.37 38 One thing never travels in that payload: the ``group_engine``. It is an object,39 the payload is JSON, and it is built once per group by the template process this40 child may have been forked from — so it arrives as a constructor argument41 instead, and ``build_worker`` puts it in the worker's own call only when there is42 one. A child spawned the ordinary way gets none, and the base worker does not43 declare it: it hosts no engine.44 45 **Storage is declared synchronous first thing.** The genro-storage nodes are46 ``smartasync``: under a running loop they hand back a coroutine instead of a47 value. The server pins the sync dispatch when it is built, and so does this48 child — before its loop exists, so every task it spawns inherits the pin.49 50 The life is one ``asyncio.run``: build the worker, connect to the socket it was51 given, present it (pid and config echo, the answer bringing the whole global52 store), then read envelopes until the wire ends. Two deaths reach that end. The53 worker leaves on its own — ``quit`` closed the wire from this side — and there54 is nothing left to do. Or the wire dies under it: the parent is gone, this55 process is an orphan, and the D8 self-defense runs — everybody into the deposit,56 then out. Either way the exit code is 0: whoever is watching reads a clean exit57 as "this child is gone", not as a crash to throttle.58 59 ``WorkerEntry`` is importable and testable on its own; the ``main()`` at the60 bottom is the thin ``python -m`` shell around it.61 """62 63 from __future__ import annotations64 65 import asyncio66 import importlib67 import json68 import logging69 import os70 import sys71 from typing import Any72 73 from genro_toolbox.smartasync import set_sync74 75 from genro_asgi.channel.frame import FrameStream76 from .freeze_handler import FreezeHandler77 from .spa_worker import SpaWorker78 from .worker_handler import WORKER_ENV_VAR79 80 #: The class the child builds when the payload names none: the base worker,81 #: which serves the protocol and hosts no site of its own.82 DEFAULT_WORKER_CLASS = "genro_asgi_multiworker_spa.orchestration.spa_worker:SpaWorker"83 84 #: The keys the child cannot invent for itself if they are missing.85 REQUIRED_KEYS = ("name", "uds_url", "frozen_users_path")86 87 __all__ = ["DEFAULT_WORKER_CLASS", "REQUIRED_KEYS", "WorkerEntry"]88 89 90 class WorkerEntry:91 """One spawned worker's whole life, driven by the ``GENRO_ASGI_WORKER`` payload.92 93 Args:94 config: the spawn payload; read from the environment when omitted.95 group_engine: the object the template built once for the whole group and96 this child inherited by fork; None when this child was spawned, and97 then nothing of the sort reaches the worker.98 """99 100 def __init__(101 self, config: dict[str, Any] | None = None, *, group_engine: Any = None102 ) -> None:103 # Before the loop exists, so every task of this process inherits it104 # (D22): a storage node reached under a loop that is not pinned hands105 # back a coroutine instead of a value.106 set_sync()107 self.config = self.read_config() if config is None else config108 self.name: str = self.config["name"]109 self.uds_url: str = self.config["uds_url"]110 self.frozen_users_path: str = self.config["frozen_users_path"]111 self.main_threadpool_size: int | None = self.config.get("main_threadpool_size")112 self.aux_threadpool_size: int | None = self.config.get("aux_threadpool_size")113 self.worker_class: str = self.config.get("worker_class") or DEFAULT_WORKER_CLASS114 self.kwargs: dict[str, Any] = self.config.get("kwargs") or {}115 self.group_engine = group_engine116 self.worker: SpaWorker | None = None117 self.logger = logging.getLogger(__name__)118 119 def read_config(self) -> dict[str, Any]:120 """Parse and validate the spawn payload; a violation ends the process.121 122 Returns:123 The payload as the handler wrote it.124 125 Raises:126 SystemExit: the variable is absent, unparsable, or short of a key127 nothing here may invent.128 """129 raw = os.environ.get(WORKER_ENV_VAR)130 if not raw:131 raise SystemExit(f"{WORKER_ENV_VAR} is not set: nothing to spawn")132 try:133 config = json.loads(raw)134 except ValueError as exc:135 raise SystemExit(f"{WORKER_ENV_VAR} is not valid JSON: {exc}") from None136 if not isinstance(config, dict):137 raise SystemExit(138 f"{WORKER_ENV_VAR} must be a JSON object, got {type(config).__name__}"139 )140 missing = [key for key in REQUIRED_KEYS if not config.get(key)]141 if missing:142 raise SystemExit(f"{WORKER_ENV_VAR} is missing {', '.join(missing)}")143 return config144 145 def load_class(self, dotted: str) -> type:146 """Resolve a ``module.path:ClassName`` reference to the class object.147 148 Args:149 dotted: the reference as the payload carries it.150 151 Returns:152 The class.153 154 Raises:155 SystemExit: the reference is not in that form.156 """157 module_path, _, class_name = dotted.partition(":")158 if not module_path or not class_name:159 raise SystemExit(f"worker_class must be 'module.path:ClassName', got {dotted!r}")160 return getattr(importlib.import_module(module_path), class_name)161 162 def build_worker(self) -> SpaWorker:163 """Build the configured worker with its deposit, its pools and its grammar.164 165 Returns:166 The worker, with no wire yet.167 168 The deposit is built HERE, on this side: nothing is handed a169 FreezeHandler over the channel, because the road to safety must not170 depend on the wire.171 172 ``group_engine`` joins the call only when this child has one, so a base173 worker — which does not declare it — is built by the same line.174 """175 worker_class = self.load_class(self.worker_class)176 kwargs = dict(self.kwargs)177 if self.group_engine is not None:178 kwargs["group_engine"] = self.group_engine179 return worker_class(180 self.name,181 freeze_handler=FreezeHandler(self.frozen_users_path),182 main_threadpool_size=self.main_threadpool_size,183 aux_threadpool_size=self.aux_threadpool_size,184 **kwargs,185 )186 187 async def connect(self) -> FrameStream:188 """Open the wire to the handler's socket.189 190 Returns:191 The frame codec over that connection.192 """193 reader, writer = await asyncio.open_unix_connection(self.uds_url.removeprefix("uds:"))194 return FrameStream(reader, writer)195 196 async def serve(self) -> None:197 """Build, present, serve — and leave when the wire dies first.198 199 Sets ``worker``. Returns when the worker has left: on its own decision,200 or because the wire under it was gone, which saves nothing. A worker that201 declared BOTH hosted seams dies before the wire exists: that is a202 contradiction, not a choice.203 """204 self.worker = self.build_worker()205 # The one configuration error the boot catches: two seams assigned at206 # once, which is a contradiction somebody declared. NO seam at all is207 # not an error — it is the base worker, which serves its orders and no208 # request, and whose http CALL is refused by the property when it comes.209 if self.worker.asgi_app is not None and self.worker.wsgi_app is not None:210 self.worker.hosted_app_seam211 self.worker.attach_stream(await self.connect())212 await self.worker.send_presentation(self.config)213 self.logger.info(214 "Worker %s: serving on %s (pid %s)", self.name, self.uds_url, os.getpid()215 )216 await self.worker.receive_frames()217 if not self.worker.exited:218 await self.worker.on_wire_lost()219 self.logger.info("Worker %s: exited cleanly", self.name)220 221 def run(self) -> int:222 """Run the whole life; 0 on either death."""223 asyncio.run(self.serve())224 return 0225 226 227 def main() -> int:228 """``python -m genro_asgi_multiworker_spa.orchestration.worker_entry``: the spawn shell."""229 logging.basicConfig(level=os.environ.get("GENRO_ASGI_LOG_LEVEL", "INFO"))230 return WorkerEntry().run()231 232 233 if __name__ == "__main__":234 sys.exit(main())