src/genro_asgi_multiworker_spa/orchestration/worker_handler.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 """WorkerHandler: the handler, which stays, and the process under it, which does not.16 17 A handler belongs to its group, carries a short name (``standard_0001``, minted by18 the group; short because the socket path has a hard system budget) and owns one19 socket. The process under it is replaceable: it can be killed, it can die on its20 own, it can be reborn on the same name and the same socket — and every placement21 pointing at the handler is untouched by all of that.22 23 **Four orders, and each verb carries its object.** ``launch_process`` opens the24 wire, spawns the child and waits for it to present itself; ``terminate_process``25 kills the process group and waits for the OS to bury it; ``quit_process`` asks26 the process to leave and waits for it to be gone; ``ping_process`` is one health27 beat, and it gives back what the child answered. Nothing here freezes a user,28 closes a tap or reads a policy: those belong one level up, and a handler that29 took them would be deciding for the pool.30 31 **The refusal is the answer.** ``assign_user`` is no order: it is the judgement a32 group asks for while it walks its workers looking for one that takes a user. The33 handler judges ITSELF — its own last photo against the setpoint its group carries34 — and says no by RAISING, the class of the refusal being the reason. It writes35 nothing: where a user lives is the group's map, and who is inside a process is36 what that process announces.37 38 **Low tolerance, and never two processes.** A mute beat is repeated ONCE past39 the timeout — against a lost packet, not against a sick worker — and then the40 process group is killed and its OS death awaited: SIGKILL, no escalation and no41 grace, because a grace period is the users waiting. Only after that death may a42 successor be launched: the wire is one, so a handler is never two processes. The43 declared price is that the slow-but-healthy dies and its users log in again,44 which is seconds of error instead of minutes of spinner.45 46 **The death is a STATE, not a mark posed from outside.** ``state`` carries one of47 five values and nobody but this handler writes it: ``starting`` (spawned, not yet48 presented), ``running``, ``quitting`` (asked to leave, draining, not coming49 back), ``quitted`` (died as it was ordered to, and the group has yet to consume50 the fact) and ``aborted`` (died with nobody waiting for it — the wild death).51 52 **The classification is the PURE WAIT.** An order to die parks a wait; the end of53 the wire resolves it. An end of wire WITH a live wait is the death somebody54 asked for; an end of wire without one is wild — no mark to set in advance, and55 none to give back. The wait is parked ONLY when a child is really on the wire: a56 wire with nobody on it reports nothing, and a wait left behind for a report that57 never comes would make the handler read its next wild death as ordered. Either58 way the handler writes the state, rings its group's wake and stops there: the59 group learns at that round, reading the state. This60 handler owns the list of its users (``hosted_users``), not the indexes of anybody61 else — the group unhooks it from the placement and the Commander, the single62 writer of the maps, prunes the traces, discards the parcels and removes the63 semaphores the dead one had announced. Which is why nothing in this module64 touches the deposit.65 66 **Every envelope goes to the chain, and the chain answers what goes down.** What67 arrives from below is handed to ``envelope_handler`` — the handler's own layer of68 the fold — which reads the photo, lets the levels above read the worker events,69 and gives back the payload for the envelope going down. So this handler carries70 no knowledge of what a worker event means, and the wire carries none either:71 the wire writes what it is handed. The one thing that answer carries today is the72 global store, whole, and only to a process presenting itself — which is the only73 one holding none of it.74 75 **No worker-owned counters here.** The handler holds ``worker_snapshot``, the last photo its76 process sent — filed by its own layer of the chain from whatever envelope carried77 it, so a live process has one from its very presentation: the gauges the judge78 reads are all in there EXCEPT CPU: the commander reads this process's cumulative79 CPU clock through psutil, by way of the handler, and keeps the two-reading anchor80 here. That temperature travels on no envelope. Aggregate counters still belong to the81 Commander, which is also the one that decides the orders worth counting.82 83 **The beat has no clock of its own.** ``ping_process`` is one beat and84 ``process_ping_interval`` is the cadence it is meant to be called at; the clock85 that calls it belongs to whoever governs the group. What the handler does say is86 whether it is worth beating at all: every envelope stamps the instant it87 arrived, so a process that has just answered traffic is not ``silent`` and its88 group leaves it alone. The handler's burial — the89 socket taken away — is ``WorkerConnector.stop()``, called by whoever closes the90 handler for good.91 92 Every order and every wild death leaves its line here through the module93 logger; the dedicated ``orchestration.log`` file, with its path, size and94 rotation, is grammar and is not built yet.95 """96 97 from __future__ import annotations98 99 import asyncio100 import json101 import logging102 import math103 import os104 import signal105 import subprocess106 import sys107 import time108 from pathlib import Path109 from typing import Any110 111 import psutil112 113 from .envelope_handler import WorkerEnvelopeHandler114 from .exceptions import (115 AssignmentRefused,116 NoRoomError,117 WorkerQuittingError,118 )119 from .worker_connector import ENVELOPE_SLOT_PRESENTATION, WorkerConnector120 from .worker_process import ForkedProcess, SpawnedProcess, WorkerProcess121 122 #: The environment variable the spawn payload travels in, as today.123 WORKER_ENV_VAR = "GENRO_ASGI_WORKER"124 125 #: The orders going DOWN are paths on the worker's own tree (#59, D59-14): the126 #: first segment names who ISSUES the order — ``group`` for the group's127 #: (orchestration), ``commander`` for the vertex's — and the last one the128 #: operation, a ``@route`` method of ``GroupOrders`` or ``CommanderOrders``.129 #:130 #: The health beat, and nothing else: it asks whether the process is alive, it131 #: does not ask for the photo, which rides every envelope on its own.132 PING_OP_PATH = "/group/ping"133 134 #: The debug door: evaluate one expression inside the child, repr back.135 EVAL_OP_PATH = "/commander/eval"136 137 #: The structured reading of a whole process: every register, JSON-safe, in138 #: one answer. Unlike the photo it is not periodic and nobody acts on it — it139 #: exists to be shown to a human.140 CENSUS_OP_PATH = "/commander/census"141 142 #: The switch of the observation: whether the process reports every register143 #: mutation of its own up the lane, as it happens. Off unless somebody is144 #: watching — an observer must not change what it observes.145 OBSERVE_OP_PATH = "/commander/observe"146 147 #: The routing key of the order to leave: the process drains and ends itself.148 #: Its answer comes back at once, carrying the photo with every user flagged for149 #: cession — the level above parks them all in one read.150 QUIT_OP_PATH = "/group/quit"151 152 #: The routing key that takes one user off the process, and the one that takes153 #: off a single connection of his. Each names the verb of ``SpaWorker`` that154 #: serves it, and carries that verb's own argument.155 DROP_USER_OP_PATH = "/group/drop_user"156 DROP_CONNECTION_OP_PATH = "/group/drop_connection"157 158 #: The routing key of the ordered freeze of ONE user: the worker waits for159 #: whatever holds him — a pull bringing him home, his calls in flight — parks160 #: him, and only then answers, so the REPLY IS the confirmation. A user this161 #: process does not host is refused out loud in that same REPLY.162 FREEZE_USER_OP_PATH = "/group/freeze_user"163 164 #: The routing key of the worker's OWN announcement: the envelope of what165 #: happened in this process while no CALL was being served — the transfer cycle166 #: of a quit — folded exactly as the envelope of a REPLY is. Every other worker167 #: event rides the REPLY of the CALL that caused it (owner, 2026-09-04). It is168 #: the group's operation: the worker announces to ITS group (#59, D59-14).169 ANNOUNCE_OP_PATH = "/group/announce"170 171 #: How long an ordered departure may take before this handler stops waiting for172 #: it, in seconds. Past it the process is killed and the death that follows is an173 #: abort like any other: whoever was leaving had its time.174 QUIT_TIMEOUT_SECONDS = 30.0175 176 #: Seconds between two beats of the same process — the cadence, not a clock.177 PROCESS_PING_INTERVAL = 5.0178 179 #: How long a process may stay mute before the beat counts as missed. Twice180 #: this, and the process is killed.181 PROCESS_PING_TIMEOUT = 10.0182 183 # How often the wait for an OS death re-reads the process: nothing signals that,184 # so it polls. The wait for the end of the wire does not — that one is a future.185 WAIT_POLL_INTERVAL = 0.05186 187 #: A temperature older than three intended samples is unavailable. One second188 #: is the floor, so a brief event-loop delay cannot withdraw a healthy gauge.189 CPU_TEMPERATURE_MIN_STALE_SECONDS = 1.0190 191 __all__ = [192 "ANNOUNCE_OP_PATH",193 "CENSUS_OP_PATH",194 "DROP_CONNECTION_OP_PATH",195 "DROP_USER_OP_PATH",196 "EVAL_OP_PATH",197 "FREEZE_USER_OP_PATH",198 "OBSERVE_OP_PATH",199 "PING_OP_PATH",200 "PROCESS_PING_INTERVAL",201 "PROCESS_PING_TIMEOUT",202 "QUIT_OP_PATH",203 "QUIT_TIMEOUT_SECONDS",204 "WORKER_ENV_VAR",205 "WorkerHandler",206 ]207 208 209 class WorkerHandler:210 """One handler of a group: its wire, its process, its users, its last photo.211 212 Args:213 group_handler: the group this handler belongs to; the end of a process is214 told to it as ``ping_now()``, it reads ``state`` at that round, and215 its ``envelope_handler`` is the way up for everything the process216 announces.217 name: the handler's name, minted by the group as ``<group>_<counter>``;218 it names the socket too, so it is short.219 instance_dir: the directory holding the sockets of this installation.220 frozen_users_path: the deposit root the child builds its own access to.221 entry_module: the module the child is started as (``python -m ...``).222 main_threadpool_size: the child's traffic pool size, None for its own223 default.224 aux_threadpool_size: the child's service pool size, much smaller.225 worker_class: the ``module:Class`` the child loads, None for its own.226 worker_kwargs: the grammar handed to that class; travels as ``kwargs``.227 executable: the interpreter to spawn with, this one by default.228 process_ping_interval: the cadence the beat is meant to be called at.229 process_ping_timeout: how long the process may stay mute per beat.230 """231 232 def __init__(233 self,234 group_handler: Any,235 name: str,236 *,237 instance_dir: str | Path,238 frozen_users_path: str | Path,239 entry_module: str,240 main_threadpool_size: int | None = None,241 aux_threadpool_size: int | None = None,242 worker_class: str | None = None,243 worker_kwargs: dict[str, Any] | None = None,244 executable: str | None = None,245 process_ping_interval: float = PROCESS_PING_INTERVAL,246 process_ping_timeout: float = PROCESS_PING_TIMEOUT,247 ) -> None:248 self.group_handler = group_handler249 self.name = name250 self.instance_dir = Path(instance_dir)251 self.frozen_users_path = Path(frozen_users_path)252 self.entry_module = entry_module253 self.main_threadpool_size = main_threadpool_size254 self.aux_threadpool_size = aux_threadpool_size255 self.worker_class = worker_class256 self.worker_kwargs = worker_kwargs or {}257 self.executable = executable or sys.executable258 self.process_ping_interval = process_ping_interval259 self.process_ping_timeout = process_ping_timeout260 self.process: WorkerProcess | None = None261 #: Where the process under this handler is in its life: ``starting``,262 #: ``running``, ``quitting``, ``quitted``, ``aborted``.263 #: Written only here; the group reads it at its round.264 self.state = "starting"265 #: The last photo the process sent, on whatever envelope carried it:266 #: memory, load, counts, per-connection clocks. Filed by this handler's267 #: own layer of the chain.268 self.worker_snapshot: dict[str, Any] | None = None269 #: The soft CPU admission (#43): True, this worker is a candidate for270 #: NEW users. The group's judge writes False when the smoothed271 #: ``cpu_temperature_percent`` crosses above ``cpu_admission_close_percent`` — the placement272 #: then skips this worker — and True again below273 #: ``cpu_admission_reopen_percent``. Over the threshold it stays closed;274 #: capacity is born only when concrete demand finds no open candidate.275 #: Sticky users are untouched, the memory veto in ``assign_user`` stands276 #: apart, and the state dies with the277 #: handler. State only — this handler decides nothing with it.278 self.cpu_admission_open = True279 #: When the placement last landed a user here (commander's monotonic280 #: clock), None before the first: the admission interval reads it.281 self.last_admission_monotonic: float | None = None282 #: The ``psutil.Process`` of the pid this handler owns, built once and283 #: rebuilt when the pid changes.284 self._process_probe: psutil.Process | None = None285 #: The last lightweight process reading as ``(process birth, cpu seconds,286 #: sample instant)``. The birth distinguishes a live worker from an287 #: unrelated process that later reused its pid.288 self._cpu_meter_reading: tuple[float, float, float] | None = None289 #: The worker's CPU share over the last meter interval, raw: telemetry290 #: only, no judge reads it. None until two readings of the same process291 #: exist: the first interval has no temperature yet.292 self.cpu_temperature_sample_percent: float | None = None293 #: The temperature the judges read: the raw samples through the group's294 #: asymmetric first-order filter (``cpu_heating_seconds`` up,295 #: ``cpu_cooling_seconds`` down), seeded by the first sample.296 self.cpu_temperature_percent: float | None = None297 #: When the temperature above was sampled, on the commander's monotonic298 #: clock, and the real width of the interval that produced it.299 self.cpu_temperature_sampled_at: float | None = None300 self.cpu_temperature_interval_seconds: float | None = None301 #: The offload condition last journaled for this worker, as302 #: ``(reason, subject)`` — ``single_user_overload`` and303 #: ``cpu_offload_no_active_candidate`` would otherwise repeat every304 #: beat for as long as they stand. Deduplication of the journal only,305 #: cleared when the worker leaves the offload picture; it dies with306 #: the handler and decides nothing.307 self.cpu_offload_condition: tuple[str, str | None] | None = None308 self.envelope_handler = WorkerEnvelopeHandler(self, group_handler.envelope_handler)309 self.connector = WorkerConnector(self, self.instance_dir / f"{name}.sock")310 self._logger = logging.getLogger(__name__)311 self._hosted_users: set[str] = set()312 self._death_wait: asyncio.Future[None] | None = None313 self._listening = False314 self._last_envelope_ts = 0.0315 self._running_since: float | None = None316 self._observation_switched = False317 self._observation_switch_tasks: set[asyncio.Task[Any]] = set()318 319 @property320 def life_seconds(self) -> float:321 """How long this worker has been serving, in seconds; 0.0 before it presented."""322 if self._running_since is None:323 return 0.0324 return time.monotonic() - self._running_since325 326 @property327 def requires_beat_ping(self) -> bool:328 """Whether nothing has been heard from this process for a whole cadence.329 330 Returns:331 True when the last envelope is older than ``process_ping_interval``.332 """333 return time.monotonic() - self._last_envelope_ts >= self.process_ping_interval334 335 @property336 def hosted_users(self) -> set[str]:337 """The users living on this handler's process; the fold is its single writer."""338 return self._hosted_users339 340 @property341 def spawn_payload(self) -> dict[str, Any]:342 """The child's whole configuration, as it travels JSON-encoded in ``GENRO_ASGI_WORKER``."""343 return {344 "name": self.name,345 "uds_url": self.connector.address,346 "frozen_users_path": str(self.frozen_users_path),347 "main_threadpool_size": self.main_threadpool_size,348 "aux_threadpool_size": self.aux_threadpool_size,349 "worker_class": self.worker_class,350 "kwargs": self.worker_kwargs,351 }352 353 def assign_user(self, user: str) -> None:354 """Judge whether this worker takes one more user, and refuse by raising.355 356 Args:357 user: the identity being placed here.358 359 Raises:360 WorkerQuittingError: its process is leaving or is gone.361 AssignmentRefused: it has not presented itself yet.362 NoRoomError: it already hosts ``worker_max_users`` placed users, or363 its memory is past ``worker_memory_admission_percent`` — the364 veto. The CPU admission is not judged here: the placement walks365 the open workers first and the CPU-closed ones only as its366 fallback, so this gate must let a closed worker take the user367 the memory allows.368 369 Nothing is written. The user count is read off ``user_worker_map``, the370 map the placement writes in the same breath — never ``hosted_users``,371 which the fold writes only when the worker has announced, and would let372 two rapid arrivals land on one worker before the first is on board.373 The admission interval is not judged here: it orders the placement's374 walk, it refuses nobody.375 """376 if self.state in ("quitting", "quitted", "aborted"):377 raise WorkerQuittingError(user, f"{self.name} is {self.state}")378 if self.state != "running":379 raise AssignmentRefused(user, f"{self.name} is {self.state}")380 policy = self.group_handler.policy381 placed = sum(382 1 for worker in self.group_handler.user_worker_map.values() if worker == self.name383 )384 if placed >= policy.worker_max_users:385 raise NoRoomError(user, f"{self.name} already hosts {placed} placed user(s)")386 memory_percent = self.group_handler.get_memory_occupancy_percent(self.worker_snapshot)387 if memory_percent > policy.worker_memory_admission_percent:388 raise NoRoomError(user, f"{self.name} stands at {memory_percent:.1f}% of memory")389 390 def get_process_cpu_reading(self) -> tuple[float, float] | None:391 """Read this worker's process birth and cumulative CPU seconds through psutil.392 393 Returns:394 ``(create time, cpu seconds)`` for the pid this handler owns, or395 None when there is no process or psutil cannot see it.396 397 The ``psutil.Process`` probe is built once per pid and reused. No398 command is spawned and no message is sent to the worker.399 """400 if self.process is None:401 return None402 try:403 if self._process_probe is None or self._process_probe.pid != self.process.pid:404 self._process_probe = psutil.Process(self.process.pid)405 times = self._process_probe.cpu_times()406 return self._process_probe.create_time(), times.user + times.system407 except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):408 return None409 410 def record_cpu_reading(411 self, reading: tuple[float, float] | None, *, sampled_at: float412 ) -> float | None:413 """Turn two lightweight process readings into this worker's temperature.414 415 Returns:416 The filtered temperature after this reading, or None when the417 reading did not produce one (first of a process, cleared, no time418 elapsed).419 420 The raw share of one core over the interval is kept as421 ``cpu_temperature_sample_percent``; ``cpu_temperature_percent`` moves422 towards it by ``1 - exp(-elapsed / tau)``, with ``tau`` the group's423 ``cpu_heating_seconds`` when the sample is hotter than the filtered424 value and ``cpu_cooling_seconds`` when it is colder. The first sample of425 a process seeds the filter.426 427 A missing row or a changed process birth clears the unfinished measure;428 neither invents a zero. The result is separate commander-side telemetry:429 it never changes the full photo, while CPU orchestration reads this430 channel explicitly.431 """432 if reading is None:433 self._cpu_meter_reading = None434 self.cpu_temperature_sample_percent = None435 self.cpu_temperature_percent = None436 self.cpu_temperature_sampled_at = None437 self.cpu_temperature_interval_seconds = None438 return None439 created_at, cpu_seconds = reading440 previous = self._cpu_meter_reading441 self._cpu_meter_reading = (created_at, cpu_seconds, sampled_at)442 if previous is None or previous[0] != created_at:443 self.cpu_temperature_sample_percent = None444 self.cpu_temperature_percent = None445 self.cpu_temperature_sampled_at = None446 self.cpu_temperature_interval_seconds = None447 return None448 elapsed = sampled_at - previous[2]449 if elapsed <= 0.0:450 return None451 burned = max(0.0, cpu_seconds - previous[1])452 sample = 100.0 * min(burned / elapsed, 1.0)453 current = self.cpu_temperature_percent454 if current is None:455 temperature = sample456 else:457 policy = self.group_handler.policy458 tau = policy.cpu_heating_seconds if sample > current else policy.cpu_cooling_seconds459 temperature = current + (1.0 - math.exp(-elapsed / tau)) * (sample - current)460 self.cpu_temperature_sample_percent = sample461 self.cpu_temperature_percent = temperature462 self.cpu_temperature_sampled_at = sampled_at463 self.cpu_temperature_interval_seconds = elapsed464 return temperature465 466 def get_cpu_temperature_percent(self) -> float | None:467 """Return the fresh commander-side temperature, never a stale value."""468 temperature = self.cpu_temperature_percent469 sampled_at = self.cpu_temperature_sampled_at470 cadence = self.group_handler.spa_commander.cpu_temperature_sample_seconds471 if temperature is None or sampled_at is None or cadence is None:472 return None473 stale_after = max(CPU_TEMPERATURE_MIN_STALE_SECONDS, 3.0 * cadence)474 if time.monotonic() - sampled_at > stale_after:475 return None476 return temperature477 478 def read_envelope(self, envelope: dict[str, Any]) -> dict[str, Any]:479 """Hand an envelope that arrived from the process to the chain.480 481 Args:482 envelope: the payload as it came off the wire.483 484 Returns:485 What goes back down, as the chain composed it — nothing at all when486 there is no envelope going the other way.487 488 Stamps the instant this process was last heard from. The envelope that489 carries the presentation — the ``pid`` slot, said at birth and never490 again — is told to the vertex through ``on_worker_presented``, the seam491 a consumer overrides to speak to a newborn process.492 """493 self._last_envelope_ts = time.monotonic()494 spa_commander = self.group_handler.spa_commander495 if not self._observation_switched and spa_commander.observation_watched:496 self._observation_switched = True497 self._fire_observation_switch()498 if ENVELOPE_SLOT_PRESENTATION in envelope:499 spa_commander.on_worker_presented(self)500 return self.envelope_handler(envelope)501 502 def _fire_observation_switch(self) -> None:503 """Turn this process's observation on without holding up the envelope it presented with."""504 task = asyncio.create_task(self.connector.call(OBSERVE_OP_PATH, {"on": True}))505 self._observation_switch_tasks.add(task)506 task.add_done_callback(self._observation_switch_tasks.discard)507 508 def serve_child_call(self, path: str, data: dict[str, Any]) -> Any:509 """Resolve a CALL the child placed on the lane on the group's dispatcher, and call it.510 511 Args:512 path: the routing key the child chose — ``/group/…`` or ``/commander/…``.513 data: its payload, handed over as the operation's keyword arguments.514 515 Returns:516 Whatever the operation answers, sync or awaitable; the wire awaits517 it if needed and puts it in the REPLY.518 519 Raises:520 NotFound: no operation at that path, on either tree; the wire turns521 it into an error REPLY, so the child is answered either way.522 TypeError: the payload does not fit the operation's signature —523 refused before the body runs, the same way.524 525 This handler is only the rung the call climbs: nothing is written here.526 """527 return self.group_handler.group_dispatcher.route.node(path)(**data)528 529 async def launch_process(self) -> None:530 """Open the wire if it is closed, spawn the child, wait for it to present itself.531 532 Raises:533 RuntimeError: a process is already alive under this handler.534 TimeoutError: the child never presented itself within535 ``process_ping_timeout``; it is killed before the raise.536 537 Sets ``process`` and ``state`` — ``starting`` while the child is on its538 way, ``running`` once it has presented itself — and binds the socket on539 the first launch.540 """541 if self.process is not None and self.process.alive:542 raise RuntimeError(543 f"WorkerHandler {self.name}: its process (pid {self.process.pid}) is still alive"544 )545 if not self._listening:546 await self.connector.start()547 self._listening = True548 self.state = "starting"549 self.process = await self.start_process()550 self._logger.info(551 "Worker %s: launched its process (pid %s) on %s",552 self.name,553 self.process.pid,554 self.connector.address,555 )556 try:557 await asyncio.wait_for(self.connector.wait_connected(), self.process_ping_timeout)558 except TimeoutError:559 self._logger.warning(560 "Worker %s: its process never presented itself in %.1fs — killing",561 self.name,562 self.process_ping_timeout,563 )564 await self.terminate_process()565 raise566 self.state = "running"567 self._running_since = time.monotonic()568 569 async def start_process(self) -> WorkerProcess:570 """Bring the process into the world, by fork when the group has a template.571 572 Returns:573 The process, however it was born.574 575 Raises:576 TemplateRefused: the group has a template and it did not fork.577 """578 template = self.group_handler.template579 if template is None:580 return self.spawn_process()581 return ForkedProcess(await template.fork_worker(self.spawn_payload))582 583 def spawn_process(self) -> SpawnedProcess:584 """Start a brand new interpreter with the payload in its environment.585 586 Returns:587 The spawned process.588 589 A program that starts fresh carries nothing of this one, so the payload590 has to travel in something the exec preserves — which is why this birth591 uses an environment variable and the forked one does not.592 """593 env = dict(os.environ)594 env[WORKER_ENV_VAR] = json.dumps(self.spawn_payload)595 return SpawnedProcess(596 subprocess.Popen(597 [self.executable, "-m", self.entry_module], env=env, start_new_session=True598 )599 )600 601 async def terminate_process(self) -> None:602 """Kill the process group and wait until the OS has buried it; clears ``process``.603 604 The wait is bounded by ``QUIT_TIMEOUT_SECONDS``, the same bound an ordered605 death already has. A spawned process always ends it, because this handler606 is the parent and reading ``alive`` buries it. A forked one may not: it607 stays a zombie — alive, to a pid — until its template collects it, and a608 template that has stopped collecting would hold this wait forever. Past609 the bound the handler says so and lets go.610 """611 process = self.process612 self._logger.info("Worker %s: killing its process (pid %s)", self.name, process.pid)613 self._kill_process_group()614 try:615 await asyncio.wait_for(self._wait_for_death(process), QUIT_TIMEOUT_SECONDS)616 except TimeoutError:617 self._logger.warning(618 "Worker %s: its process (pid %s) was killed but is still not buried "619 "after %.0fs — letting go of it",620 self.name,621 process.pid,622 QUIT_TIMEOUT_SECONDS,623 )624 self.process = None625 626 async def _wait_for_death(self, process: WorkerProcess) -> None:627 """Poll until that process is gone."""628 while process.alive:629 await asyncio.sleep(WAIT_POLL_INTERVAL)630 631 async def quit_process(self, freezer_path: str | None = None) -> None:632 """Ask the process to leave, and wait until it is gone.633 634 Args:635 freezer_path: where the parcels of this departure go, when they must636 not go to the working deposit — the reboot directory of a soft637 quit. None leaves the child on its own deposit.638 639 Sets ``quitting`` and parks the wait its death resolves. Past640 ``QUIT_TIMEOUT_SECONDS`` on either leg the wait is dropped and the process641 is killed, so the death that follows is an abort.642 """643 self._logger.info("Worker %s: asked to leave", self.name)644 self.state = "quitting"645 death = self._park_death_wait()646 try:647 await self.connector.call(648 QUIT_OP_PATH, {"freezer_path": freezer_path}, timeout=QUIT_TIMEOUT_SECONDS649 )650 await asyncio.wait_for(death, QUIT_TIMEOUT_SECONDS)651 except TimeoutError:652 self._death_wait = None653 self._logger.warning(654 "Worker %s: still here %.1fs after being asked to leave — killing its process",655 self.name,656 QUIT_TIMEOUT_SECONDS,657 )658 await self.terminate_process()659 660 async def ping_process(self) -> dict[str, Any] | None:661 """One health beat: are you alive? Kill the process if it stays mute.662 663 Returns:664 The payload the child answered with, or None when it answered neither665 beat and its process was killed for it.666 667 Acts on the process when it stays mute: the end of the wire writes the state.668 """669 for beat in (1, 2):670 try:671 return await self.connector.call(672 PING_OP_PATH, timeout=self.process_ping_timeout673 )674 except TimeoutError:675 self._logger.warning(676 "Worker %s: beat %s of 2 unanswered after %.1fs",677 self.name,678 beat,679 self.process_ping_timeout,680 )681 self._logger.warning("Worker %s: mute to both beats — killing its process", self.name)682 await self.terminate_process()683 return None684 685 def on_child_lost(self) -> None:686 """The wire died: the parked wait says whether anybody was expecting it.687 688 Sets ``state`` — ``quitted`` when a wait was live, ``aborted`` when the689 death was nobody's order — rings the group's wake, and gives the vertex690 back the store grant this process was holding, if it held one.691 """692 self.group_handler.spa_commander.commander_dispatcher.global_store.release_worker_lock(693 self.name694 )695 ordered = self._settle_death_wait()696 if ordered:697 self.state = "quitted"698 self._logger.info("Worker %s: its process left as it was asked to", self.name)699 else:700 self.state = "aborted"701 self._logger.warning(702 "Worker %s: WILD death of its process, %s users on board",703 self.name,704 len(self._hosted_users),705 )706 self.group_handler.ping_now()707 708 def _kill_process_group(self) -> None:709 """SIGKILL the child's whole process group; one already gone is the same outcome."""710 process = self.process711 if not process.alive:712 return713 try:714 os.killpg(os.getpgid(process.pid), signal.SIGKILL)715 except ProcessLookupError:716 pass717 718 def expect_death(self) -> None:719 """Say that the death about to happen is awaited, so it is not a wild one.720 721 Acts on the parked wait, whose being live is what ``on_child_lost`` reads722 to tell an ordered death from a wild one: what follows is ``quitted`` and723 no alarm is owed for it. The killing is the caller's own next step.724 """725 self._park_death_wait()726 727 def _park_death_wait(self) -> asyncio.Future[None]:728 """Park the wait an ordered death resolves; its being live IS the order."""729 self._death_wait = asyncio.get_running_loop().create_future()730 return self._death_wait731 732 def _settle_death_wait(self) -> bool:733 """Resolve the parked wait if one is live, and say whether one was.734 735 Returns:736 True when somebody was waiting; a wait already given up counts for nobody.737 """738 death = self._death_wait739 self._death_wait = None740 if death is None or death.done():741 return False742 death.set_result(None)743 return True