stroblme
f52d9c3077
_serve_once treated a cancelled call future as finished, so an event loop cancelling all tasks at once (shutdown) never sent the worker a cancel and the call's thread kept asyncio.run waiting for SILENCE_S. The cameras audio test now removes its voice flow and device, the unpaired-frame test waits for its first frame before the unpair, and test_connector no longer inherits the controller an app test bound to demand.
250 lines
9.9 KiB
Python
250 lines
9.9 KiB
Python
"""Service: a server kept running on a worker while its flow is live.
|
|
|
|
A Python node is a call that returns, in a process the worker kills afterwards,
|
|
so a model it loads is loaded again on every message. A service is the other
|
|
shape: one long call that runs a command, such as a speech server, and keeps
|
|
it up. Other nodes reach it over HTTP at the url it publishes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import shlex
|
|
import uuid
|
|
from collections.abc import Iterable
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import httpx
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from fluksio.core.config import settings
|
|
from fluksio.flow.messages import MessageSpec
|
|
from fluksio.flow.nodes.base import Node
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
from fluksio.flow.remote import RemoteWorker, RemoteWorkerHub
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#: Seconds between restarts, one step per start in a row that never got
|
|
#: healthy. A service that served and then died starts again after the first.
|
|
BACKOFF = settings.FLOW_RESTART_BACKOFF
|
|
#: Starts in a row that never answered healthy before the node reads down.
|
|
DOWN_AFTER = 3
|
|
#: How often a starting server's health is asked, and how long one ask may take.
|
|
HEALTH_POLL_S = 1.0
|
|
HEALTH_TIMEOUT_S = 2.0
|
|
#: How often to look again when no worker carries the label.
|
|
WORKER_POLL_S = 2.0
|
|
#: How long a stopping flow waits for the worker to say the server is gone, so
|
|
#: a flow started again finds the port free. Under the supervisor's grace.
|
|
STOP_WAIT_S = 4.0
|
|
|
|
|
|
class ServiceNode(Node):
|
|
"""Keep a server running on a worker and publish the url it listens on.
|
|
|
|
The command runs on a worker matched by label, in one call that takes none
|
|
of the worker's ``--parallel`` slots, and starts again whenever it ends.
|
|
Every output port carries the url once the server answers its health
|
|
check, and an empty string the moment the call ends.
|
|
"""
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
command: str = Field(
|
|
description=(
|
|
"The server to run, such as `parrot serve --host 0.0.0.0 "
|
|
"--port {port}`. {port} and {weights} are filled in."
|
|
)
|
|
)
|
|
port: int = Field(
|
|
default=8100, ge=1, le=65535, description="The port the server listens on."
|
|
)
|
|
worker: str = Field(
|
|
description="The label of the worker to run it on, such as gpu."
|
|
)
|
|
health: str = Field(
|
|
default="/health",
|
|
description="The path that answers 200 once the server is ready.",
|
|
)
|
|
env: dict[str, str] = Field(
|
|
default_factory=dict,
|
|
description="Environment variables for the server.",
|
|
)
|
|
weights: str = Field(
|
|
default="",
|
|
pattern=r"^(sha256:[0-9a-f]{64})?$",
|
|
description=(
|
|
"An artifact digest, fetched to the worker. Its local path "
|
|
"is {weights} in the command."
|
|
),
|
|
)
|
|
|
|
__slots__ = ("cfg", "argv", "hub", "_task", "_url")
|
|
|
|
def __init__(
|
|
self,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
self.cfg = self.Params.model_validate(params or {})
|
|
self.argv = shlex.split(self.cfg.command.replace("{port}", str(self.cfg.port)))
|
|
if not self.argv:
|
|
raise ValueError("a service needs a command")
|
|
if "{weights}" in self.cfg.command and not self.cfg.weights:
|
|
raise ValueError("the command uses {weights}, but no weights are set")
|
|
# Set from the app on start; a test sets it directly.
|
|
self.hub: RemoteWorkerHub | None = None
|
|
self._task: asyncio.Task[None] | None = None
|
|
self._url = ""
|
|
super().__init__(
|
|
f=self._never_called,
|
|
requires=requires,
|
|
provides=provides,
|
|
params=params,
|
|
name=name or "service",
|
|
)
|
|
|
|
@staticmethod
|
|
def _never_called(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
|
|
"""Its flow starts it, not a message, and it publishes on its own."""
|
|
return None
|
|
|
|
# -- lifecycle ------------------------------------------------------------
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
if app is not None:
|
|
self.hub = getattr(app.state, "worker_hub", None)
|
|
# Spawned on every start, as the websocket node's: the supervisor
|
|
# ignores a duplicate.
|
|
if self._task is not None and not self._task.done():
|
|
return
|
|
self._task = self._run_supervised("service", self._keep_running)
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
if self._task is not None:
|
|
await self._cancel_task(self._task)
|
|
self._task = None
|
|
|
|
async def _keep_running(self) -> None:
|
|
"""Run the server, and run it again whenever it ends."""
|
|
failures = 0
|
|
while True:
|
|
worker = self.hub.pick(self.cfg.worker) if self.hub is not None else None
|
|
if worker is None:
|
|
self.report_health(
|
|
"degraded", f"no worker labelled '{self.cfg.worker}' is attached"
|
|
)
|
|
await asyncio.sleep(WORKER_POLL_S)
|
|
continue
|
|
self.report_health("degraded", f"starting on '{worker.name}'")
|
|
healthy, reason = await self._serve_once(worker)
|
|
# Only starts that never got healthy lead to down: one that served
|
|
# for a while is restarted as often as it dies.
|
|
failures = 0 if healthy else failures + 1
|
|
delay = BACKOFF[min(failures, len(BACKOFF) - 1)]
|
|
logger.warning("Service '%s' ended: %s", self.id, reason)
|
|
self.report_health(
|
|
"down" if failures >= DOWN_AFTER else "degraded",
|
|
f"{reason}; restarting in {delay:g}s",
|
|
)
|
|
await asyncio.sleep(delay)
|
|
|
|
async def _serve_once(self, worker: RemoteWorker) -> tuple[bool, str]:
|
|
"""One call on this worker, until it ends: whether it got healthy, and why."""
|
|
call_id = f"{self.id}:{uuid.uuid4().hex[:8]}"
|
|
host = f"[{worker.host}]" if ":" in worker.host else worker.host
|
|
url = f"http://{host}:{self.cfg.port}"
|
|
payload = {
|
|
"op": "serve",
|
|
"call_id": call_id,
|
|
"argv": self.argv,
|
|
"env": self.cfg.env,
|
|
"weights": self.cfg.weights,
|
|
}
|
|
# On a thread, as every call is, and holding no slot.
|
|
call = asyncio.ensure_future(
|
|
asyncio.to_thread(worker.request, payload, 0, slot=False)
|
|
)
|
|
# Retrieved however it ends, so one that outlives a stop is not
|
|
# reported at exit.
|
|
call.add_done_callback(lambda done: done.cancelled() or done.exception())
|
|
healthy = False
|
|
try:
|
|
healthy = await self._await_health(call, worker.name, url)
|
|
if healthy:
|
|
self._url = url
|
|
await asyncio.to_thread(self._publish, url)
|
|
self.report_health("ok")
|
|
await asyncio.wait({call})
|
|
error = call.result().get("error") or {}
|
|
return healthy, str(error.get("message") or "the service stopped")
|
|
except Exception as exc:
|
|
# The worker went away or went silent, or the url was refused.
|
|
return healthy, f"{type(exc).__name__}: {exc}"
|
|
finally:
|
|
# A cancelled future is done while its thread still waits for the
|
|
# worker, which is how an event loop cancelling every task at once
|
|
# (a shutdown) reaches here: only an answer means it has ended.
|
|
if not call.done() or call.cancelled():
|
|
# The flow stopped, or this side failed. Ending the call kills
|
|
# the server's process group on the worker, and its answer says
|
|
# that is done, so the next start finds the port free.
|
|
await asyncio.to_thread(worker.cancel, call_id)
|
|
await asyncio.wait({call}, timeout=STOP_WAIT_S)
|
|
await asyncio.to_thread(self._withdraw)
|
|
|
|
async def _await_health(
|
|
self, call: asyncio.Future[Any], worker: str, url: str
|
|
) -> bool:
|
|
"""Ask the server's health until it answers 200, or the call ends first."""
|
|
async with httpx.AsyncClient(timeout=HEALTH_TIMEOUT_S) as client:
|
|
while not call.done():
|
|
problem = await self._probe(client, url + self.cfg.health)
|
|
if problem is None:
|
|
return True
|
|
self.report_health(
|
|
"degraded",
|
|
f"starting on '{worker}': {self.cfg.health} {problem}",
|
|
)
|
|
await asyncio.wait({call}, timeout=HEALTH_POLL_S)
|
|
return False
|
|
|
|
@staticmethod
|
|
async def _probe(client: httpx.AsyncClient, url: str) -> str | None:
|
|
"""None once the url answers 200, else what it did instead."""
|
|
try:
|
|
response = await client.get(url)
|
|
except httpx.HTTPError as exc:
|
|
return f"is not answering ({type(exc).__name__})"
|
|
if response.status_code == 200:
|
|
return None
|
|
return f"answers {response.status_code}"
|
|
|
|
# -- the url ----------------------------------------------------------------
|
|
|
|
def _publish(self, url: str) -> None:
|
|
self.emit({spec.port: url for spec in self.output_ports})
|
|
|
|
def _withdraw(self) -> None:
|
|
"""Empty the url the moment the server is gone.
|
|
|
|
Unless its value is gone already: a deleted flow's values are cleared
|
|
before its nodes stop, and writing the url would bring one back.
|
|
"""
|
|
if not self._url:
|
|
return
|
|
self._url = ""
|
|
pipeline = self._pipeline
|
|
if pipeline is not None and any(
|
|
pipeline.state.get(spec.name) for spec in self.output_ports
|
|
):
|
|
self._publish("")
|