src/genro_asgi/tasks/manager.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 """TaskManager — the server-owned task backbone (spool + executor + hub, ◆D22).16 17 The spool and the executor are leaf pieces: the spool is a folder model on storage,18 the executor runs one resolved task. The manager is what makes them LIVE — a single19 object owned by the server (dual relationship: ``self.server``) that drives the20 fire-and-forget worker loop for the whole process. It owns:21 22 - ``spool`` — the ``LocalTaskExecutor``'s own spool over ``server.storage`` (the23 stateless seam: the same storage yields an equivalent spool, so the manager reuses24 the executor's rather than opening a second);25 - ``executor`` — the ``LocalTaskExecutor`` bound to the live server;26 - ``hub`` — the in-memory ``EventHub`` (the live progress courier for Phase 6);27 - ``scheduler`` — the ``TaskScheduler`` (the recurring loop, over ``task_store``);28 - ``task_store`` — the ``FileTaskStore`` (persistent schedules, over ``site:tasks``);29 - ``worker_id`` — the single logical worker of the mono-process core (``"local"``).30 31 ``start()``/``stop()`` are the lifecycle the server's lifespan hook calls32 (``TaskMixin.__call__``): ``start`` launches two tasks on the running loop — the33 fire-and-forget ``_worker_loop`` and the ``scheduler`` tick loop — and ``stop``34 cancels/awaits them both (in-flight executions are their own tasks and are left35 to finish). Each loop mirrors the same shape: a failing pass is logged and never36 kills the loop. Session GC is NOT a manager job: the store reaps expired37 sessions itself, delta-checked at ``create`` time (a ratified revision of core38 1e/◆D22 — the former ``_purge_loop`` is gone).39 40 The worker loop is FIRE-AND-FORGET on the event loop: it polls ``list_pending``,41 ``assign``s each task to ``worker_id``, and launches ``executor.execute`` as its own42 task. The D2 thread pool stays reserved for the blocking handler BODY inside43 ``execute`` (via ``server.run_sync``) — the loop itself never blocks the pool.44 Distributed dispatch (worker processes, a batch commander) is out of scope (D22).45 """46 47 from __future__ import annotations48 49 import asyncio50 import contextlib51 import logging52 from typing import TYPE_CHECKING, Any53 54 from .executor import WORKER_ID, LocalTaskExecutor55 from .hub import EventHub56 from .scheduler import TaskScheduler57 from .spool import ACTIVE58 from .store import FileTaskStore59 60 if TYPE_CHECKING:61 from ..server import BaseServer62 63 __all__ = ["TaskManager", "POLL_SECONDS"]64 65 POLL_SECONDS = 0.5 # how often the worker loop polls the pending queue66 67 68 class TaskManager:69 """Owns the spool/executor/hub and drives the fire-and-forget worker loop.70 71 Note:72 Bound to the live server (dual relationship: ``self.server``). The loop73 task and its event loop are held on the instance; ``running`` reports74 whether the loop task is live.75 """76 77 __slots__ = (78 "server", "executor", "hub", "task_store", "scheduler", "worker_id",79 "_loop", "_loop_task",80 )81 82 def __init__(self, server: BaseServer) -> None:83 """Bind to the live server and build the executor, hub, store and scheduler.84 85 ``server.tasks_config`` (the tuning dict the mixin peeled from a86 ``tasks=`` dict, empty otherwise) is applied here: ``mount`` overrides87 the store's by-keys choice, ``tick_seconds`` retunes the scheduler.88 """89 config = getattr(server, "tasks_config", None) or {}90 self.server = server91 self.executor = LocalTaskExecutor(server)92 self.hub = EventHub()93 self.task_store = FileTaskStore(server.storage, mount=config.get("mount"))94 self.scheduler = TaskScheduler(self)95 if config.get("tick_seconds") is not None:96 self.scheduler.tick_seconds = float(config["tick_seconds"])97 self.worker_id = WORKER_ID98 self._loop: asyncio.AbstractEventLoop | None = None99 self._loop_task: asyncio.Task[None] | None = None100 101 @property102 def spool(self) -> Any:103 """The task spool (the executor's own, over ``server.storage``)."""104 return self.executor.spool105 106 @property107 def running(self) -> bool:108 """Whether the worker loop task is currently live."""109 return self._loop_task is not None and not self._loop_task.done()110 111 # -- lifecycle (called by the server's lifespan hook) --112 113 def start(self) -> None:114 """Launch the worker and scheduler loops (lifespan startup)."""115 self._loop = asyncio.get_running_loop()116 self._loop_task = self._loop.create_task(self._worker_loop())117 self.scheduler.start()118 logging.getLogger(__name__).info("task manager started (worker %r)", self.worker_id)119 120 async def stop(self) -> None:121 """Cancel the worker and scheduler loops (lifespan shutdown).122 123 In-flight executions are their own tasks and are left to finish.124 """125 await self.scheduler.stop()126 if self._loop_task is not None:127 self._loop_task.cancel()128 with contextlib.suppress(asyncio.CancelledError):129 await self._loop_task130 self._loop_task = None131 logging.getLogger(__name__).info("task manager stopped")132 133 # -- the fire-and-forget worker loop --134 135 async def _worker_loop(self) -> None:136 """Poll pending forever; a failing poll is logged and never kills the loop."""137 while True:138 try:139 self._drain_pending()140 except Exception:141 logging.getLogger(__name__).exception("task worker poll failed")142 await asyncio.sleep(POLL_SECONDS)143 144 def _drain_pending(self) -> None:145 """Assign every pending task to the worker and launch its execution.146 147 Each task is claimed with an atomic ``assign`` (pending -> active) and its148 ``execute`` runs as its own loop task (fire-and-forget); the sync handler149 body inside ``execute`` goes to the pool, not this loop.150 """151 loop = self._loop152 assert loop is not None # set by start() before the loop runs153 for descriptor in self.spool.list_pending():154 task_id = descriptor["task_id"]155 self.spool.assign(task_id, self.worker_id)156 loop.create_task(self.executor.execute(task_id, self.worker_id))157 158 # -- the A<->C progress seam (spool = source of truth, hub = live courier) --159 160 def publish_progress(self, task_id: str, data: dict[str, Any]) -> None:161 """Write a progress snapshot AND publish it live, in one paired call.162 163 The pairing rule of the push channel: every ``progress.json`` write also164 fans out on the hub, keyed by the descriptor's launching ``session_id``165 (``None`` = no push channel, the spool write still happens). The task166 must be ACTIVE on this manager's worker — writing progress anywhere else167 would create a stray spool folder.168 169 Raises:170 LookupError: if the task is unknown or not in the active state.171 """172 descriptor = self.spool.get(task_id)173 if descriptor is None or descriptor["status"] != ACTIVE:174 raise LookupError(f"task not active: {task_id}")175 self.spool.write_progress(task_id, self.worker_id, data)176 session_id = descriptor.get("session_id")177 if session_id is not None:178 self.hub.publish(session_id, {"type": "progress", "task_id": task_id, "data": data})