src/genro_asgi_multiworker_spa/orchestration/template_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 template process: build the group engine once, then fork workers off it.16 17 ``python -m genro_asgi_multiworker_spa.orchestration.template_entry`` is what a GroupHandler18 spawns, one per group, named ``template-<group>``. It builds the expensive thing19 its group's workers all need — the ``group_engine`` — and from then on every20 worker of that group is a ``fork`` of this process, so the engine is not built21 again and not copied: the children find it already in their own memory.22 23 **No asyncio here, on purpose.** A ``WorkerEntry`` runs ``asyncio.run``, and a24 child forked out of a running loop inherits that loop as *running* — the call25 would raise — over an epoll the parent shares. A synchronous template deletes26 that whole class of trouble instead of working around it, and the measurements27 the design rests on (one thread, no open channels, the macOS fork traps mute)28 were taken on a process of exactly this shape.29 30 **Everything travels as JSON lines on the pipes**, and nothing in an environment31 variable. The first line ``stdin`` carries is this process's own configuration::32 33 {"name": "template-standard",34 "engine_factory": "some.module:SomeFactory",35 "kwargs": {"whatever the factory needs": "..."}}36 37 ``engine_factory`` is a ``module.path:ClassName`` reference to a class the38 deployment provides; it is instantiated with ``kwargs`` and asked for the engine39 by calling ``build_group_engine()`` on it. Both keys are mandatory: a template40 with no factory has nothing to share, so a short line is a contract violation41 and ends the process.42 43 Every line after the first is one fork request, and it carries a worker's whole44 spawn payload — the same object the handler already writes for a spawned child,45 unchanged. The answer is one line back on ``stdout``: the child's pid, or an46 error when the fork was refused.47 48 **The child never receives the payload — it already has it.** ``fork`` copies49 the memory, so the request the parent had just parsed is in the child's own50 variable. That is also why the engine crosses at all: an object cannot travel in51 an environment variable, and here it does not have to.52 53 Three things happen in the child before anything else, in this order: it takes a54 session of its own (``setsid``, or a kill of its process group would take the55 template and every sibling with it); it closes the inherited pipes (a child56 holding the write end of ``stdout`` open would mask the template's death, whose57 clean signal is the EOF the GroupHandler reads); and only then does it live the58 ordinary worker life, ``WorkerEntry`` with the payload and the engine.59 60 ``stderr`` is left alone — it is not a pipe — so the logs of the template and of61 every child go where the GroupHandler's own go.62 63 **The answer channel is nobody's stdout.** At birth (real pipes only, never the64 injected ones of a test) the template duplicates its stdout descriptor and65 keeps the duplicate as the channel, then points ``stdout`` itself at66 ``stderr``: the engine build runs arbitrary deployment code, and a ``print()``67 in it must land in the logs — landing in the channel made the GroupHandler68 read it as an answer, and the first forks of every container start failed with69 «Extra data» until the build's lines were consumed. The forked child closes70 the duplicate with the other pipes, and its own stdout — like the template's —71 speaks to the logs.72 """73 74 from __future__ import annotations75 76 import gc77 import importlib78 import json79 import logging80 import os81 import signal82 import sys83 import threading84 from typing import Any85 86 from genro_toolbox.smartasync import set_sync87 88 from .worker_entry import WorkerEntry89 90 #: The keys the first line cannot omit: without a factory there is nothing to share.91 REQUIRED_KEYS = ("name", "engine_factory")92 93 __all__ = ["REQUIRED_KEYS", "TemplateEntry"]94 95 96 class TemplateEntry:97 """One template's whole life: read the first line, build the engine, fork on demand.98 99 Args:100 pipe_in: the line source; ``sys.stdin`` when omitted.101 pipe_out: where the answers go; ``sys.stdout`` when omitted.102 """103 104 def __init__(self, pipe_in: Any = None, pipe_out: Any = None) -> None:105 # Before any factory runs, and before any loop exists (D22): a storage106 # node reached under an unpinned loop hands back a coroutine, not a value.107 set_sync()108 self.pipe_in = sys.stdin if pipe_in is None else pipe_in109 if pipe_out is None:110 # The answer channel is a DUPLICATE of the real stdout, taken for111 # this process alone, and stdout itself is pointed at stderr —112 # BEFORE the engine is built, because the factory runs arbitrary113 # deployment code (the bridge's builds a whole legacy site) and114 # every print() of that build used to land IN the channel: the115 # GroupHandler read those lines as answers and the first forks of116 # every container start failed with «Extra data» (diagnosis of117 # 2026-08-28). From here on a print anywhere in this process — the118 # build, the frozen engine, a forked child before it closes its119 # copy — goes to the logs, and the channel carries answers only.120 # Only for the REAL stdout: a test that injects its pipes gets121 # them verbatim, and no descriptor of the test runner is touched.122 self.pipe_out = os.fdopen(os.dup(sys.stdout.fileno()), "w")123 os.dup2(sys.stderr.fileno(), sys.stdout.fileno())124 else:125 self.pipe_out = pipe_out126 self.name = ""127 self.group_engine: Any = None128 self.logger = logging.getLogger(__name__)129 130 def read_launch(self) -> dict[str, Any]:131 """Read and validate the first line; a violation ends the process.132 133 Returns:134 The launch configuration as the GroupHandler wrote it.135 136 Raises:137 SystemExit: the pipe closed first, the line is not a JSON object, or138 a key nothing here may invent is missing.139 """140 raw = self.pipe_in.readline()141 if not raw:142 raise SystemExit("the pipe closed before the launch line arrived")143 try:144 config = json.loads(raw)145 except ValueError as exc:146 raise SystemExit(f"the launch line is not valid JSON: {exc}") from None147 if not isinstance(config, dict):148 raise SystemExit(149 f"the launch line must be a JSON object, got {type(config).__name__}"150 )151 missing = [key for key in REQUIRED_KEYS if not config.get(key)]152 if missing:153 raise SystemExit(f"the launch line is missing {', '.join(missing)}")154 return config155 156 def load_class(self, dotted: str) -> type:157 """Resolve a ``module.path:ClassName`` reference to the class object.158 159 Args:160 dotted: the reference as the launch line carries it.161 162 Returns:163 The class.164 165 Raises:166 SystemExit: the reference is not in that form.167 """168 module_path, _, class_name = dotted.partition(":")169 if not module_path or not class_name:170 raise SystemExit(f"engine_factory must be 'module.path:ClassName', got {dotted!r}")171 return getattr(importlib.import_module(module_path), class_name)172 173 def build_group_engine(self, config: dict[str, Any]) -> Any:174 """Ask the declared factory for this group's engine.175 176 Args:177 config: the launch configuration.178 179 Returns:180 Whatever the factory built. This process never looks inside it.181 """182 factory_class = self.load_class(config["engine_factory"])183 return factory_class(**(config.get("kwargs") or {})).build_group_engine()184 185 @property186 def live_thread_count(self) -> int:187 """How many threads are alive in this process; one is the fork invariant.188 189 A property of the PROCESS, not of this object: a template is a fresh190 process and answers 1, while the same code inside a test runner answers191 whatever that runner left running.192 """193 return threading.active_count()194 195 def freeze_heap(self) -> None:196 """Put everything alive out of the collector's reach, before the first fork.197 198 The engine is the biggest thing this process will ever hold, and the199 children get it for free — until the collector walks that inherited graph200 looking for cycles. Walking it writes to every object it touches, and a201 written page stops being shared: the child pays for the engine one page at202 a time, for a walk that can find nothing, because nothing here is garbage.203 ``gc.freeze`` moves what exists now into the permanent generation, which204 the collector never visits, and the children inherit that with the fork.205 206 Measured on the bridge's own site, 2026-08-24: 98 MB per worker after 200207 requests instead of 153. What a child allocates while serving is tracked208 and collected as always — only what the template built is frozen.209 210 The price, declared: among frozen objects no cycle is ever collected211 again, so two of them that point at each other and become unreachable stay.212 Reference counting still frees what it can, and the engine lives as long as213 the process does.214 """215 gc.freeze()216 self.logger.info(217 "Template %s: %s objects frozen, %s left for the collector",218 self.name,219 gc.get_freeze_count(),220 len(gc.get_objects()),221 )222 223 def reap_children(self, signum: int, frame: Any) -> None:224 """Collect the children that have died, and nothing else.225 226 The template is their parent only on paper: it buries them so they do not227 stay zombies, and never reads a status, decides, or reports. Installed as228 the ``SIGCHLD`` handler; a blocking read it interrupts resumes on its own.229 """230 while True:231 try:232 pid, _ = os.waitpid(-1, os.WNOHANG)233 except ChildProcessError:234 return235 if pid == 0:236 return237 238 def reply(self, answer: dict[str, Any]) -> None:239 """Write one answer line and push it out, so the buffer is empty at the next fork."""240 self.pipe_out.write(json.dumps(answer) + "\n")241 self.pipe_out.flush()242 243 def fork_worker(self, payload: dict[str, Any]) -> None:244 """Fork one worker for this payload, or refuse out loud; answers either way.245 246 Args:247 payload: the worker's spawn payload, as the handler wrote it.248 249 The refusal is not a silent check: a second thread at fork time is what250 arms the macOS traps this design is only safe without, so a template that251 has grown one says so and forks nothing.252 """253 alive = self.live_thread_count254 if alive != 1:255 self.reply({"error": f"{self.name}: {alive} threads alive, refusing to fork"})256 return257 pid = os.fork()258 if pid == 0:259 self.live_as_worker(payload)260 self.logger.info("Template %s: forked %s (pid %s)", self.name, payload["name"], pid)261 self.reply({"pid": pid})262 263 def live_as_worker(self, payload: dict[str, Any]) -> None:264 """Leave the template behind, then become the worker this payload names.265 266 Args:267 payload: the worker's spawn payload, already in this process's memory.268 269 Never returns. Two steps in this order: a session of its own, so a kill of270 this worker's process group cannot reach the template or a sibling; then271 the inherited pipes closed, so this child cannot mask the template's death272 by holding the write end open.273 """274 os.setsid()275 self.pipe_in.close()276 self.pipe_out.close()277 self.become_worker(payload)278 279 def become_worker(self, payload: dict[str, Any]) -> None:280 """Run the ordinary worker life and exit with its code.281 282 Args:283 payload: the worker's spawn payload.284 285 Never returns. The exit is the ordinary one, not ``os._exit``, so whatever286 the engine registered to run at exit still runs.287 """288 entry = WorkerEntry(config=payload, group_engine=self.group_engine)289 sys.exit(entry.run())290 291 def serve(self) -> None:292 """Build the engine, then fork on every line until the pipe ends.293 294 Sets ``name`` and ``group_engine``. Returns when the pipe closed: the295 GroupHandler is gone, and a template with nobody to serve has nothing to296 do — the workers already forked keep serving without it.297 """298 config = self.read_launch()299 self.name = config["name"]300 self.group_engine = self.build_group_engine(config)301 self.freeze_heap()302 signal.signal(signal.SIGCHLD, self.reap_children)303 self.logger.info("Template %s: engine built, waiting (pid %s)", self.name, os.getpid())304 while True:305 raw = self.pipe_in.readline()306 if not raw:307 self.logger.info("Template %s: its pipe ended, leaving", self.name)308 return309 self.fork_worker(json.loads(raw))310 311 def run(self) -> int:312 """Run the whole life; 0 when the pipe ended."""313 self.serve()314 return 0315 316 317 def main() -> int:318 """``python -m genro_asgi_multiworker_spa.orchestration.template_entry``: the launch shell."""319 logging.basicConfig(level=os.environ.get("GENRO_ASGI_LOG_LEVEL", "INFO"))320 return TemplateEntry().run()321 322 323 if __name__ == "__main__":324 sys.exit(main())