From c92f611f7c2b0c9acd44a58ae2fa38ca2f6bf0e7 Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:38:12 -0700 Subject: [PATCH 1/9] [None][fix] serve: make multi-frontend metrics, profiling and liveness launcher-owned With num_serve_frontends > 1 every frontend drained the engine's iteration stats and kept its own Prometheus registry, so /metrics and /prometheus/metrics reported a random frontend's share; /start_profile returned 500 on attached frontends; the Anthropic batch store was per-process; and attached frontends never noticed a dead launcher or engine and kept answering /health 200. - The launcher's uvicorn also listens on a Unix socket in the multi-frontend ipc dir, whose path reaches the attached frontends through the spawn environment; they forward /metrics, /kv_cache_events, /start_profile, /stop_profile and /v1/messages/batches* to it (tensorrt_llm/serve/multi_frontend.py). - Only the launcher runs the iteration-stats collector; it also polls on a 1 s cadence (off the event loop) since attached-frontend requests never wake it. - The launcher exports one PROMETHEUS_MULTIPROC_DIR before spawning the children so request counters/histograms aggregate across frontends; config-info gauges are logged by the launcher only. - An attached-frontend watchdog turns a vanished parent or failing launcher /health into a fatal executor error plus graceful shutdown. Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- tensorrt_llm/commands/serve.py | 88 +++++++- tensorrt_llm/serve/multi_frontend.py | 204 ++++++++++++++++++ tensorrt_llm/serve/openai_server.py | 122 +++++++++-- .../integration/test_lists/test-db/l0_cpu.yml | 1 + .../test_serve_multi_frontend_helpers.py | 172 +++++++++++++++ 5 files changed, 569 insertions(+), 18 deletions(-) create mode 100644 tensorrt_llm/serve/multi_frontend.py create mode 100644 tests/unittest/llmapi/test_serve_multi_frontend_helpers.py diff --git a/tensorrt_llm/commands/serve.py b/tensorrt_llm/commands/serve.py index 0873ddcf5251..4ad68671d071 100644 --- a/tensorrt_llm/commands/serve.py +++ b/tensorrt_llm/commands/serve.py @@ -32,7 +32,7 @@ from pathlib import Path from types import FrameType from typing import (TYPE_CHECKING, Any, Dict, Iterator, NoReturn, Optional, - Sequence, Set) + Sequence, Set, Tuple) import click import torch @@ -73,6 +73,9 @@ from tensorrt_llm.serve._frontend_processes import ( _init_multi_frontend_mode, _signal_frontend_ready, _spawn_attached_frontends, _terminate_attached_frontends) +from tensorrt_llm.serve.multi_frontend import (LAUNCHER_UDS_ENV, + LAUNCHER_UDS_NAME, + MultiFrontendServing) from tensorrt_llm.serve.tool_parser import ToolParserFactory from tensorrt_llm.serve.tool_parser.tool_parser_factory import ( MODEL_TYPE_TO_TOOL_PARSER, resolve_auto_tool_parser) @@ -517,6 +520,56 @@ def _provision_kv_cache_pool(llm_args: dict, yield +def _bind_launcher_uds(llm) -> Tuple[socket.socket, str]: + """Bind the launcher's Unix socket inside the multi-frontend ipc dir. + + uvicorn serves it next to the TCP socket, so attached frontends can + forward launcher-owned routes (/metrics, /kv_cache_events, profiling, + message batches) to the one process that owns the engine. The ipc dir + is private to the launcher's user and removed by the executor proxy. + """ + ipc_dir = getattr(getattr(llm, "_executor", None), + "_multi_frontend_ipc_dir", None) + if not ipc_dir: + raise ValueError( + "num_serve_frontends > 1 requires the classic IPC executor " + "proxy in multi-frontend mode") + path = os.path.join(ipc_dir, LAUNCHER_UDS_NAME) + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + sock.bind(path) + except OSError: + sock.close() + raise + return sock, path + + +def _share_prometheus_multiproc_dir(llm, ipc_dir: str) -> None: + """Give every frontend one PROMETHEUS_MULTIPROC_DIR before spawning. + + prometheus_client's multiprocess mode aggregates the .db files under + that directory. Left to OpenAIServer, each process would create its own + after the children already exist, and every /prometheus/metrics scrape + would report a single frontend's share of the request counters. + + The directory lives inside the multi-frontend ipc dir, which the + executor proxy removes at shutdown. It is deliberately not created via + set_prometheus_multiproc_dir(): that helper keeps only its most recent + TemporaryDirectory alive, so its second call (OpenAIServer.__init__) + would garbage-collect, and thereby delete, a shared directory created + by the first while every process still writes into it. + """ + args = getattr(llm, "args", None) + if args is None or os.environ.get("PROMETHEUS_MULTIPROC_DIR"): + return + if (getattr(args, "return_perf_metrics", False) + or getattr(args, "perf_metrics_output_dir", None)): + path = tempfile.mkdtemp(prefix="prometheus_", dir=ipc_dir) + os.environ["PROMETHEUS_MULTIPROC_DIR"] = path + logger.info( + f"Shared PROMETHEUS_MULTIPROC_DIR for all frontends: {path}") + + def launch_server( host: str, port: int, @@ -548,6 +601,17 @@ def launch_server( model = served_model_name or llm_args["model"] multi_frontend = _init_multi_frontend_mode(llm_args, multi_frontend_enabled) + # An attached frontend forwards launcher-owned routes to the launcher's + # Unix socket, whose path the launcher hands down when spawning it (see + # serve/multi_frontend.py). The launcher is this process's parent. + multi_frontend_serving = None + if multi_frontend.is_attached_frontend: + attached_launcher_uds = os.environ.pop(LAUNCHER_UDS_ENV, None) + if attached_launcher_uds is not None: + multi_frontend_serving = MultiFrontendServing( + launcher_uds=attached_launcher_uds, + is_launcher=False, + launcher_pid=os.getppid()) if enable_runtime_control_endpoints and multi_frontend.num_frontends > 1: raise click.BadParameter( "Runtime control endpoints require num_serve_frontends=1 because " @@ -625,12 +689,23 @@ def launch_server( # server construction, middleware registration, and runtime, or a # failure in between leaks the child processes. frontend_children = [] + sockets = [s] + launcher_uds_sock = None + launcher_uds_path = None try: if multi_frontend.is_launcher: + launcher_uds_sock, launcher_uds_path = _bind_launcher_uds( + llm) + _share_prometheus_multiproc_dir( + llm, os.path.dirname(launcher_uds_path)) frontend_children = _spawn_attached_frontends( llm, multi_frontend.num_frontends, + extra_env={LAUNCHER_UDS_ENV: launcher_uds_path}, report_failure=_report_observed_child_failure) + multi_frontend_serving = MultiFrontendServing( + launcher_uds=launcher_uds_path, is_launcher=True) + sockets.append(launcher_uds_sock) server = OpenAIServer( generator=llm, @@ -647,7 +722,8 @@ def launch_server( internal_disagg_auth_key=internal_disagg_auth_key, enable_runtime_control_endpoints= enable_runtime_control_endpoints, - runtime_control_api_key=runtime_control_api_key) + runtime_control_api_key=runtime_control_api_key, + multi_frontend_serving=multi_frontend_serving) _apply_fastapi_middlewares(server.app, middleware) # Optionally disable GC (default: not disabled) @@ -655,10 +731,16 @@ def launch_server( gc.disable() _signal_frontend_ready(multi_frontend) - uvloop.run(server(host, port, sockets=[s])) + uvloop.run(server(host, port, sockets=sockets)) finally: if frontend_children: _terminate_attached_frontends(frontend_children) + if launcher_uds_sock is not None: + launcher_uds_sock.close() + try: + os.unlink(launcher_uds_path) + except OSError: + pass def launch_mm_encoder_server( diff --git a/tensorrt_llm/serve/multi_frontend.py b/tensorrt_llm/serve/multi_frontend.py new file mode 100644 index 000000000000..58a4adf90d5f --- /dev/null +++ b/tensorrt_llm/serve/multi_frontend.py @@ -0,0 +1,204 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Helpers for trtllm-serve multi-frontend serving (``num_serve_frontends > 1``). + +Several frontend processes share one executor and one ``SO_REUSEPORT`` port, +and a client cannot choose which of them answers. Anything that needs exactly +one owner per server therefore lives in the launcher (frontend 0): draining +the engine's iteration statistics and KV-cache events, runtime profiling, and +the in-memory Anthropic Message Batches store. The launcher's uvicorn also +listens on a Unix domain socket in the multi-frontend ipc directory; attached +frontends forward those routes to it (:class:`LauncherForwarder`) and watch it +(:class:`AttachedFrontendWatchdog`) so that a dead launcher or engine does not +leave them answering ``/health`` 200 with nothing behind them. +""" + +import asyncio +import os +from dataclasses import dataclass +from typing import Callable, Optional + +import aiohttp +from fastapi import Request +from fastapi.responses import JSONResponse, Response + +from tensorrt_llm.logger import logger + +LAUNCHER_UDS_NAME = "launcher.sock" +# Carries the launcher's Unix socket path to the attached frontends it spawns. +LAUNCHER_UDS_ENV = "TLLM_SERVE_LAUNCHER_UDS" + +# Routes an attached frontend hands to the launcher, as FastAPI route +# templates. They are registered before the regular routes, and FastAPI +# matches in registration order, so they shadow the local handlers. The +# catch-all covers /v1/messages/batches/{batch_id}[/cancel|/results]. +FORWARDED_ROUTES = ( + ("GET", "/metrics"), + ("POST", "/kv_cache_events"), + ("POST", "/start_profile"), + ("POST", "/stop_profile"), + ("GET", "/v1/messages/batches"), + ("POST", "/v1/messages/batches"), + ("GET", "/v1/messages/batches/{rest:path}"), + ("POST", "/v1/messages/batches/{rest:path}"), + ("DELETE", "/v1/messages/batches/{rest:path}"), +) + +_FORWARDED_REQUEST_HEADERS = ("content-type", "accept", "authorization") + + +@dataclass(frozen=True) +class MultiFrontendServing: + """This process's role in a multi-frontend server. + + ``None`` in place of an instance means single-frontend serving. + """ + + launcher_uds: str + is_launcher: bool + # The launcher's pid as seen by an attached frontend (its parent). + launcher_pid: Optional[int] = None + + @property + def is_attached(self) -> bool: + return not self.is_launcher + + +class LauncherForwarder: + """Forward HTTP requests from an attached frontend to the launcher.""" + + def __init__(self, uds_path: str, timeout_s: float = 120.0): + self._uds_path = uds_path + self._timeout = aiohttp.ClientTimeout(total=timeout_s) + self._session: Optional[aiohttp.ClientSession] = None + + async def _get_session(self) -> aiohttp.ClientSession: + # Created lazily so it binds to the serving event loop. + if self._session is None or self._session.closed: + # force_close: no idle keep-alive connections to race against a + # launcher that is shutting down; these routes are low-rate. + self._session = aiohttp.ClientSession( + connector=aiohttp.UnixConnector(path=self._uds_path, force_close=True), + timeout=self._timeout, + ) + return self._session + + async def forward(self, request: Request) -> Response: + """FastAPI endpoint: replay ``request`` against the launcher verbatim.""" + target = request.url.path + if request.url.query: + target = f"{target}?{request.url.query}" + headers = { + k: v for k, v in request.headers.items() if k.lower() in _FORWARDED_REQUEST_HEADERS + } + body = await request.body() + try: + session = await self._get_session() + # The host is required by the URL syntax and ignored by the + # Unix connector. + async with session.request( + request.method, f"http://launcher{target}", data=body, headers=headers + ) as resp: + content = await resp.read() + return Response( + content=content, + status_code=resp.status, + media_type=resp.headers.get("Content-Type"), + ) + except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as e: + logger.warning( + f"Forwarding {request.method} {target} to the launcher frontend failed: {e!r}" + ) + return JSONResponse( + status_code=503, + content={ + "error": f"The launcher frontend that serves {request.url.path} " + f"is unavailable: {e!r}" + }, + ) + + async def get_status(self, path: str, timeout_s: float = 5.0) -> Optional[int]: + """HTTP status of ``GET path`` on the launcher, ``None`` if unreachable.""" + try: + session = await self._get_session() + async with session.get( + f"http://launcher{path}", timeout=aiohttp.ClientTimeout(total=timeout_s) + ) as resp: + await resp.read() + return resp.status + except (aiohttp.ClientError, asyncio.TimeoutError, OSError): + return None + + async def close(self) -> None: + if self._session is not None and not self._session.closed: + await self._session.close() + self._session = None + + +class AttachedFrontendWatchdog: + """Detect a lost launcher or engine from an attached frontend. + + An attached frontend owns no worker processes, so the only signals that + the engine is gone are its parent (the launcher) disappearing, or the + launcher's own ``/health`` failing, which happens once the launcher's + executor recorded a fatal engine error. Either way ``mark_dead`` is + called exactly once; the caller records the fatal error on its executor + (failing in-flight requests fast) and starts server shutdown. + """ + + def __init__( + self, + forwarder: LauncherForwarder, + launcher_pid: int, + mark_dead: Callable[[BaseException], None], + *, + ppid_interval_s: float = 1.0, + health_interval_s: float = 5.0, + health_failures: int = 3, + ): + self._forwarder = forwarder + self._launcher_pid = launcher_pid + self._mark_dead = mark_dead + self._ppid_interval = ppid_interval_s + self._health_interval = health_interval_s + self._health_failures = health_failures + + async def run(self) -> None: + loop = asyncio.get_running_loop() + failures = 0 + next_health = loop.time() + self._health_interval + while True: + await asyncio.sleep(self._ppid_interval) + if os.getppid() != self._launcher_pid: + self._die(f"launcher frontend (pid {self._launcher_pid}) exited") + return + if loop.time() < next_health: + continue + next_health = loop.time() + self._health_interval + status = await self._forwarder.get_status("/health") + if status == 200: + failures = 0 + continue + failures += 1 + if failures >= self._health_failures: + self._die( + f"launcher frontend /health failed {failures} " + f"consecutive times (last status: {status})" + ) + return + + def _die(self, why: str) -> None: + logger.error(f"Attached frontend lost its engine: {why}") + self._mark_dead(RuntimeError(f"attached frontend lost its engine: {why}")) diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 4d0a9c22b315..b10b0e33c4d7 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -93,6 +93,10 @@ from tensorrt_llm.serve.encode_batcher import (EncodeBatcher, InputTooLongError, QueueFullError) from tensorrt_llm.serve.metadata_server import create_metadata_server +from tensorrt_llm.serve.multi_frontend import (FORWARDED_ROUTES, + AttachedFrontendWatchdog, + LauncherForwarder, + MultiFrontendServing) from tensorrt_llm.serve.openai_protocol import ( ChatCompletionNamedToolChoiceParam, ChatCompletionRequest, ChatCompletionResponse, ChatCompletionResponseChoice, @@ -597,7 +601,8 @@ def __init__( enable_rl_control_endpoints: bool = False, rl_control_api_key: Optional[str] = None, enable_runtime_control_endpoints: bool = False, - runtime_control_api_key: Optional[str] = None): + runtime_control_api_key: Optional[str] = None, + multi_frontend_serving: Optional[MultiFrontendServing] = None): if enable_rl_control_endpoints: logger.warning( "RL control endpoints are deprecated; use generic runtime " @@ -725,6 +730,22 @@ def __init__( ) if server_role is not None else "server" self._perf_metrics_writer = PerfMetricsJsonlWriter( perf_metrics_output_dir, server_kind) + # Multi-frontend serving (num_serve_frontends > 1): the launcher is + # the sole consumer of the engine's iteration stats / KV events and + # the only owner of profiling and the batch store; attached + # frontends forward those routes to it over its Unix socket and + # watch it for engine or launcher death (see serve/multi_frontend.py). + self._multi_frontend = multi_frontend_serving + self._launcher_forwarder: Optional[LauncherForwarder] = None + self._launcher_watchdog_task: Optional[asyncio.Task] = None + if multi_frontend_serving is not None and multi_frontend_serving.is_attached: + self._launcher_forwarder = LauncherForwarder( + multi_frontend_serving.launcher_uds) + # Requests finishing on attached frontends never wake the launcher's + # collector, so the launcher also polls on a fixed cadence. + self._iteration_stats_poll_interval: Optional[float] = ( + 1.0 if multi_frontend_serving is not None + and multi_frontend_serving.is_launcher else None) self._iteration_stats_collector_task = None self._iteration_stats_wakeup_event = asyncio.Event() # Bounded snapshot of iteration stats for the GET /metrics handler. @@ -792,7 +813,18 @@ async def lifespan(app: FastAPI): # Start background iteration stats collector if metrics are enabled # The PyTorch backend args include `enable_iter_perf_stats`. - if self.metrics_collector and getattr( + # An attached frontend never drains the engine queue: the + # launcher owns it and /metrics is forwarded there. + if self._launcher_forwarder is not None: + self._launcher_watchdog_task = asyncio.create_task( + AttachedFrontendWatchdog( + self._launcher_forwarder, + self._multi_frontend.launcher_pid, + self._on_launcher_lost).run()) + logger.info( + "Attached frontend: forwarding launcher-owned routes " + f"to {self._multi_frontend.launcher_uds}") + elif self.metrics_collector and getattr( self.generator.args, "enable_iter_perf_stats", True): # The background loop becomes the sole consumer of the # engine stats queue; /metrics reads from a tee buffer @@ -825,6 +857,14 @@ async def lifespan(app: FastAPI): yield await self._perf_metrics_writer.close() + if self._launcher_watchdog_task is not None: + self._launcher_watchdog_task.cancel() + try: + await self._launcher_watchdog_task + except asyncio.CancelledError: + pass + if self._launcher_forwarder is not None: + await self._launcher_forwarder.close() if self.embedding_batcher is not None: await self.embedding_batcher.shutdown() logger.info("Stopped encode dynamic batcher") @@ -889,6 +929,14 @@ async def validation_exception_handler(request, exc): self._init_embedding_batcher() self.register_embedding_routes() else: + if self._launcher_forwarder is not None: + # Registered first so they shadow the local handlers below + # (FastAPI matches in registration order). + for method, path in FORWARDED_ROUTES: + self.app.add_api_route(path, + self._launcher_forwarder.forward, + methods=[method], + include_in_schema=False) self.register_routes() if self._collect_perf_metrics: @@ -1020,7 +1068,11 @@ def _init_llm(self, chat_template: Optional[str] = None): request_inference_time_buckets=( pmc.request_inference_time_buckets if pmc else None), ) - self._log_config_info_metrics() + # With a shared PROMETHEUS_MULTIPROC_DIR every process would + # export its own per-pid config-info series; the launcher's + # suffices (attached frontends share its engine and config). + if self._launcher_forwarder is None: + self._log_config_info_metrics() @staticmethod def _ensure_post_processor_hook_supported( @@ -1723,6 +1775,22 @@ async def data_transceiver_state(self) -> JSONResponse: base64.b64encode(state).decode("utf-8") }) + def _on_launcher_lost(self, error: BaseException) -> None: + """Watchdog callback of an attached frontend: fail fast, then shut down. + + Mirrors the fatal-error branch of ``health``: record the error on the + executor so in-flight and new requests raise EngineDeadError instead + of hanging, then raise SIGINT once for uvicorn's graceful shutdown. + """ + executor = getattr(self.generator, '_executor', None) + if executor is None: + return + if getattr(executor, '_fatal_error', None) is None: + executor._set_fatal_error(error) + if not getattr(executor, 'doing_shutdown', True): + _record_generator_termination(self.generator) + signal.raise_signal(signal.SIGINT) + async def health(self) -> Response: if self._check_health(): return Response(status_code=200) @@ -1943,8 +2011,10 @@ async def _extract_metrics(self, res: RequestOutput, raw_request: Request): self.metrics_collector.log_request_metrics_dict( res.metrics_dict) # Note: Iteration stats are collected by the background _iteration_stats_collector_loop task - # Wake up the stats collector to drain iteration stats - if getattr(self.generator.args, "enable_iter_perf_stats", True): + # Wake up the stats collector to drain iteration stats (an + # attached frontend has no collector: the launcher drains). + if self._iteration_stats_collector_task is not None and getattr( + self.generator.args, "enable_iter_perf_stats", True): self._iteration_stats_wakeup_event.set() async def _create_chat_response( @@ -1995,22 +2065,37 @@ async def _iteration_stats_collector_loop(self): try: logger.info("Iteration stats collector loop started") while True: - # Wait for signal that requests have completed and stats may be available - await self._iteration_stats_wakeup_event.wait() + # Wait for signal that requests have completed and stats may + # be available. A multi-frontend launcher also wakes on a + # fixed cadence: requests finishing on attached frontends + # never set this event, yet their iterations land in the + # engine queue this loop alone drains. + woken_by_timer = False + try: + await asyncio.wait_for( + self._iteration_stats_wakeup_event.wait(), + timeout=getattr(self, "_iteration_stats_poll_interval", + None)) + except asyncio.TimeoutError: + woken_by_timer = True # Clear the event for next wakeup self._iteration_stats_wakeup_event.clear() # Drain all available iteration stats and log each one to Prometheus. try: - async for llm_stat in self.generator.get_stats_async( - timeout=0.5): - self.metrics_collector.log_iteration_stats(llm_stat) - # Tee into the /metrics snapshot buffer so the HTTP - # handler can serve without competing for the engine - # queue (nvbug 6102381). - if self._iteration_stats_buffer is not None: - self._iteration_stats_buffer.append(llm_stat) + if woken_by_timer: + # Idle poll: the stats RPC would otherwise wait out + # its timeout inside a synchronous .remote() call and + # stall this event loop, so fetch once without + # waiting, off the loop. + for llm_stat in await asyncio.to_thread( + self.generator.get_stats, 0): + self._record_iteration_stat(llm_stat) + else: + async for llm_stat in self.generator.get_stats_async( + timeout=0.5): + self._record_iteration_stat(llm_stat) except Exception as e: # Log errors but continue collecting stats logger.error(f"Error collecting iteration stats: {e}", @@ -2045,6 +2130,13 @@ def _apply_spec_decode_stats_opt_in(self, postproc_args.spec_decode_num_spec_tokens = getattr( self, "_spec_decode_num_spec_tokens", None) + def _record_iteration_stat(self, llm_stat: dict) -> None: + self.metrics_collector.log_iteration_stats(llm_stat) + # Tee into the /metrics snapshot buffer so the HTTP handler can serve + # without competing for the engine queue (nvbug 6102381). + if self._iteration_stats_buffer is not None: + self._iteration_stats_buffer.append(llm_stat) + async def openai_chat(self, request: ChatCompletionRequest, raw_request: Request) -> Response: diff --git a/tests/integration/test_lists/test-db/l0_cpu.yml b/tests/integration/test_lists/test-db/l0_cpu.yml index 3ed9f4085675..f37ab4984d5f 100644 --- a/tests/integration/test_lists/test-db/l0_cpu.yml +++ b/tests/integration/test_lists/test-db/l0_cpu.yml @@ -252,6 +252,7 @@ l0_cpu: - unittest/llmapi/test_runtime_memory_api.py - unittest/llmapi/test_sampling_params.py - unittest/llmapi/test_serialization.py + - unittest/llmapi/test_serve_multi_frontend_helpers.py - unittest/llmapi/test_serve_report_addr.py - unittest/llmapi/test_serve_sleep_mode.py - unittest/llmapi/test_session_prefetcher.py diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py new file mode 100644 index 000000000000..832011e28b11 --- /dev/null +++ b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py @@ -0,0 +1,172 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""CPU coverage for trtllm-serve's multi-frontend helpers. + +An attached frontend forwards launcher-owned routes to the launcher over a +Unix socket and watches the launcher for engine or process death. Both are +exercised here against a fake launcher; the GPU path is covered end to end by +the serving integration tests. +""" + +import asyncio +import os +import threading +from contextlib import asynccontextmanager + +import pytest +from aiohttp import web +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from tensorrt_llm.serve.multi_frontend import ( + FORWARDED_ROUTES, + AttachedFrontendWatchdog, + LauncherForwarder, +) + +pytestmark = pytest.mark.cpu_only + + +class _FakeLauncher: + """aiohttp app on a Unix socket standing in for the launcher's uvicorn.""" + + def __init__(self, path: str, health_status: int = 200) -> None: + self.path = path + self.health_status = health_status + self.seen = [] + self._loop = asyncio.new_event_loop() + self._ready = threading.Event() + self._thread = threading.Thread(target=self._serve, daemon=True) + self._thread.start() + assert self._ready.wait(10), "fake launcher did not start" + + def _serve(self) -> None: + asyncio.set_event_loop(self._loop) + + async def echo(request: web.Request) -> web.Response: + self.seen.append((request.method, request.path_qs, await request.text())) + return web.json_response( + {"method": request.method, "path": request.path_qs}, + status=201 if request.method == "POST" else 200, + ) + + async def health(_: web.Request) -> web.Response: + return web.Response(status=self.health_status) + + app = web.Application() + app.router.add_get("/health", health) + app.router.add_route("*", "/{tail:.*}", echo) + self._runner = web.AppRunner(app) + self._loop.run_until_complete(self._runner.setup()) + self._loop.run_until_complete(web.UnixSite(self._runner, self.path).start()) + self._ready.set() + self._loop.run_forever() + + def close(self) -> None: + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join(10) + self._loop.run_until_complete(self._runner.cleanup()) + self._loop.close() + + +def _attached_app(forwarder: LauncherForwarder) -> FastAPI: + """The forwarded routes as OpenAIServer registers them, plus a local one.""" + + @asynccontextmanager + async def lifespan(_: FastAPI): + yield + await forwarder.close() + + app = FastAPI(lifespan=lifespan) + for method, path in FORWARDED_ROUTES: + app.add_api_route(path, forwarder.forward, methods=[method]) + + @app.get("/health") + async def local_health(): + return {"local": True} + + return app + + +def test_forwarder_replays_launcher_owned_routes(tmp_path) -> None: + launcher = _FakeLauncher(str(tmp_path / "launcher.sock")) + try: + with TestClient(_attached_app(LauncherForwarder(launcher.path))) as client: + r = client.get("/metrics?limit=3") + assert r.status_code == 200 + assert r.json() == {"method": "GET", "path": "/metrics?limit=3"} + + r = client.post("/v1/messages/batches/b1/cancel", json={"reason": "x"}) + assert r.status_code == 201 + assert r.json()["path"] == "/v1/messages/batches/b1/cancel" + method, path, body = launcher.seen[-1] + assert (method, path) == ("POST", "/v1/messages/batches/b1/cancel") + assert '"reason"' in body + + # Routes outside FORWARDED_ROUTES stay local. + assert client.get("/health").json() == {"local": True} + finally: + launcher.close() + + +def test_forwarder_reports_unreachable_launcher(tmp_path) -> None: + with TestClient(_attached_app(LauncherForwarder(str(tmp_path / "missing.sock")))) as client: + r = client.get("/metrics") + assert r.status_code == 503 + assert "unavailable" in r.json()["error"] + + +def test_watchdog_marks_dead_after_consecutive_health_failures(tmp_path) -> None: + launcher = _FakeLauncher(str(tmp_path / "launcher.sock"), health_status=503) + deaths = [] + + async def run() -> None: + forwarder = LauncherForwarder(launcher.path) + try: + watchdog = AttachedFrontendWatchdog( + forwarder, + os.getppid(), + deaths.append, + ppid_interval_s=0.01, + health_interval_s=0.02, + health_failures=2, + ) + await asyncio.wait_for(watchdog.run(), timeout=10) + finally: + await forwarder.close() + + try: + asyncio.run(run()) + finally: + launcher.close() + assert len(deaths) == 1 + assert "failed 2 consecutive" in str(deaths[0]) + + +def test_watchdog_marks_dead_when_launcher_process_is_gone(tmp_path) -> None: + deaths = [] + + async def run() -> None: + forwarder = LauncherForwarder(str(tmp_path / "none.sock")) + try: + # No process can have pid -1 as our parent: the launcher is gone. + watchdog = AttachedFrontendWatchdog(forwarder, -1, deaths.append, ppid_interval_s=0.01) + await asyncio.wait_for(watchdog.run(), timeout=5) + finally: + await forwarder.close() + + asyncio.run(run()) + assert len(deaths) == 1 + assert "exited" in str(deaths[0]) From 16ab0ad49f0fe13f1678d8aa68853d95240f19a0 Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Sun, 20 Sep 2026 00:40:15 -0700 Subject: [PATCH 2/9] [None][fix] serve: gate multi-frontend route ownership with a CPU test Every route the LLM OpenAIServer registers must be either launcher-owned (FORWARDED_ROUTES, forwarded by attached frontends) or explicitly listed as safe to serve from whichever frontend accepted the connection, with the reason. A new endpoint backed by per-process state that should be server-wide now fails the test instead of silently shipping K copies. Also catches forwarding templates that no longer match a route and stale allowlist entries. Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- .../integration/test_lists/test-db/l0_cpu.yml | 1 + .../test_serve_multi_frontend_route_gate.py | 168 ++++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py diff --git a/tests/integration/test_lists/test-db/l0_cpu.yml b/tests/integration/test_lists/test-db/l0_cpu.yml index f37ab4984d5f..4e1246422e81 100644 --- a/tests/integration/test_lists/test-db/l0_cpu.yml +++ b/tests/integration/test_lists/test-db/l0_cpu.yml @@ -253,6 +253,7 @@ l0_cpu: - unittest/llmapi/test_sampling_params.py - unittest/llmapi/test_serialization.py - unittest/llmapi/test_serve_multi_frontend_helpers.py + - unittest/llmapi/test_serve_multi_frontend_route_gate.py - unittest/llmapi/test_serve_report_addr.py - unittest/llmapi/test_serve_sleep_mode.py - unittest/llmapi/test_session_prefetcher.py diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py b/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py new file mode 100644 index 000000000000..bdf218211c79 --- /dev/null +++ b/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py @@ -0,0 +1,168 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Route-ownership gate for trtllm-serve multi-frontend serving. + +With ``num_serve_frontends > 1`` a client cannot choose which frontend process +answers, so every HTTP route of the LLM ``OpenAIServer`` must be classified: + +* **launcher-owned** -- backed by state that exists once per server (engine + stats queue, profiler, batch store). Attached frontends forward these to the + launcher; the list is ``tensorrt_llm.serve.multi_frontend.FORWARDED_ROUTES``. +* **per-frontend safe** -- correct (or deliberately per-process) when served by + whichever frontend accepted the connection; listed below with the reason. + +A new route that is in neither set fails this test, forcing the author to +decide instead of silently shipping a per-process copy of server-wide state. +""" + +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from starlette.routing import Match + +from tensorrt_llm.serve.multi_frontend import FORWARDED_ROUTES +from tensorrt_llm.serve.openai_server import OpenAIServer + +pytestmark = pytest.mark.cpu_only + +# (method, path template) -> why it is fine to answer from any frontend. +PER_FRONTEND_SAFE = { + ( + "GET", + "/health", + ): "liveness of the answering process; engine death propagates via the watchdog", + ("GET", "/health_generate"): "runs a real request through the shared executor", + ("GET", "/version"): "constant", + ("GET", "/v1/models"): "constant", + ("GET", "/server_info"): "static server/model configuration, identical in every frontend", + ("POST", "/v1/completions"): "request path: the whole point of multiple frontends", + ("POST", "/v1/chat/completions"): "request path", + ("POST", "/v1/messages"): "request path (Anthropic adapter)", + ("POST", "/v1/messages/count_tokens"): "tokenizer-only, identical in every frontend", + ( + "POST", + "/v1/responses", + ): "request path; the stateful store is disabled group-wide in multi-frontend mode", + ( + "GET", + "/v1/responses/{response_id}", + ): "always 404 in multi-frontend mode (store disabled group-wide)", + ( + "DELETE", + "/v1/responses/{response_id}", + ): "always 404 in multi-frontend mode (store disabled group-wide)", + ("POST", "/_internal/tokenize"): "tokenizer-only", + ("GET", "/energy_metrics"): "reads GPU energy counters; same devices from every process", + ("GET", "/v1/data_transceiver_state"): "fetched from the shared executor over RPC", + ("GET", "/steady_clock_offset"): "clock probe of the answering process", + ( + "POST", + "/steady_clock_offset", + ): "known limitation: calibrates the answering frontend only (documented)", + # RL control requires AsyncLLM, which is not the classic IPC executor path + # multi-frontend mode runs on; the two never coexist. + ("POST", "/release_memory"): "AsyncLLM-only; never registered together with attached frontends", + ("POST", "/resume_memory"): "AsyncLLM-only; never registered together with attached frontends", + ("POST", "/update_weights"): "AsyncLLM-only; never registered together with attached frontends", +} + + +def _llm_server_routes() -> list[tuple[str, str]]: + """(method, path) pairs the LLM server registers, without building an LLM.""" + server = object.__new__(OpenAIServer) + server.app = FastAPI() + server.generator = SimpleNamespace( + _executor=SimpleNamespace(resource_governor_queue=None), + args=SimpleNamespace(return_perf_metrics=False), + ) + server.use_harmony = False + server._enable_rl_control_endpoints = True + server.resource_governor = None + server.register_routes() + return sorted( + (method, route.path) + for route in server.app.routes + if hasattr(route, "methods") and route.methods + for method in route.methods + if method != "HEAD" + ) + + +def _forwarding_app() -> FastAPI: + app = FastAPI() + + async def _stub(): + return {} + + for method, path in FORWARDED_ROUTES: + app.add_api_route(path, _stub, methods=[method]) + return app + + +def _is_forwarded(app: FastAPI, method: str, path_template: str) -> bool: + # Concretise the registered template so the forwarding templates + # (including the {rest:path} catch-all) can be matched against it. + concrete = path_template + for name in ("{batch_id}", "{response_id}"): + concrete = concrete.replace(name, "x") + scope = {"type": "http", "method": method, "path": concrete, "path_params": {}} + return any(route.matches(scope)[0] == Match.FULL for route in app.routes) + + +def test_every_llm_route_is_classified() -> None: + fwd = _forwarding_app() + unclassified = [] + double = [] + for method, path in _llm_server_routes(): + forwarded = _is_forwarded(fwd, method, path) + safe = (method, path) in PER_FRONTEND_SAFE + if forwarded and safe: + double.append((method, path)) + elif not forwarded and not safe: + unclassified.append((method, path)) + assert not double, f"routes both forwarded and per-frontend-safe: {double}" + assert not unclassified, ( + "New OpenAIServer route(s) without a multi-frontend ownership decision: " + f"{unclassified}. Either add them to FORWARDED_ROUTES in " + "tensorrt_llm/serve/multi_frontend.py (state lives once per server, e.g. " + "an engine queue, the profiler or an in-memory store) or to " + "PER_FRONTEND_SAFE in this test with the reason they are correct when " + "served by whichever frontend accepted the connection." + ) + + +def test_every_forwarded_route_exists() -> None: + """A renamed route must not leave a dangling forwarding template.""" + registered = _llm_server_routes() + reg_app = FastAPI() + + async def _stub(): + return {} + + for method, path in registered: + reg_app.add_api_route(path, _stub, methods=[method]) + for method, template in FORWARDED_ROUTES: + probe = template.replace("{rest:path}", "x/cancel") + scope = {"type": "http", "method": method, "path": probe, "path_params": {}} + assert any(route.matches(scope)[0] == Match.FULL for route in reg_app.routes), ( + f"FORWARDED_ROUTES entry {method} {template} matches no registered route" + ) + + +def test_allowlist_has_no_stale_entries() -> None: + registered = set(_llm_server_routes()) + stale = [r for r in PER_FRONTEND_SAFE if r not in registered] + assert not stale, f"PER_FRONTEND_SAFE lists routes that no longer exist: {stale}" From 407aac7ad286feddf035264927838e471165f583 Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:28:11 -0700 Subject: [PATCH 3/9] [None][fix] serve: address review on launcher-owned multi-frontend routes - Forward every end-to-end request header to the launcher, dropping only hop-by-hop and framing headers. The allowlist dropped x-trtllm-msgpack, so a MessagePack body replayed verbatim was decoded as JSON, and auth headers a --middleware on the launcher checks again. - Count launcher /health failures only once the launcher has answered. The launcher starts listening after its attached frontends are ready and its own server is built, so a slow start (more than ~15 s) used to make every attached frontend shut itself down and leave one frontend. - Register the forwarding routes through OpenAIServer._register_launcher_forwarding_routes() and test their order against the real register_routes(); test _on_launcher_lost with and without a shutdown in progress, header forwarding, and a starting launcher. - Route gate: classify main's runtime-control routes and FastAPI's generated docs routes, and enumerate runtime-control routes too. The forwarding app no longer carries its own docs routes (they made the server's docs routes look forwarded), and each forwarding template is matched against the registered routes of its method (the fixed "x/cancel" probe matched no GET or DELETE route). Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- tensorrt_llm/serve/multi_frontend.py | 39 +++- tensorrt_llm/serve/openai_server.py | 23 +- .../test_serve_multi_frontend_helpers.py | 199 +++++++++++++++++- .../test_serve_multi_frontend_route_gate.py | 43 ++-- 4 files changed, 267 insertions(+), 37 deletions(-) diff --git a/tensorrt_llm/serve/multi_frontend.py b/tensorrt_llm/serve/multi_frontend.py index 58a4adf90d5f..a93402af2931 100644 --- a/tensorrt_llm/serve/multi_frontend.py +++ b/tensorrt_llm/serve/multi_frontend.py @@ -56,7 +56,27 @@ ("DELETE", "/v1/messages/batches/{rest:path}"), ) -_FORWARDED_REQUEST_HEADERS = ("content-type", "accept", "authorization") +# Request headers not replayed to the launcher: hop-by-hop headers (RFC 9110, +# section 7.6.1) and the framing/encoding headers aiohttp manages itself for +# the replayed request (the body it hands back is already decoded). Everything +# else arrives as the client sent it: x-trtllm-msgpack selects the body codec, +# and auth headers feed the --middleware the launcher runs again. +_NOT_FORWARDED_REQUEST_HEADERS = frozenset( + ( + "accept-encoding", + "connection", + "content-length", + "expect", + "host", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailer", + "transfer-encoding", + "upgrade", + ) +) @dataclass(frozen=True) @@ -100,9 +120,11 @@ async def forward(self, request: Request) -> Response: target = request.url.path if request.url.query: target = f"{target}?{request.url.query}" - headers = { - k: v for k, v in request.headers.items() if k.lower() in _FORWARDED_REQUEST_HEADERS - } + headers = [ + (k, v) + for k, v in request.headers.items() + if k.lower() not in _NOT_FORWARDED_REQUEST_HEADERS + ] body = await request.body() try: session = await self._get_session() @@ -178,6 +200,12 @@ def __init__( async def run(self) -> None: loop = asyncio.get_running_loop() failures = 0 + # The launcher builds its own server, and starts listening, only after + # every attached frontend reported ready, so /health failures count + # only once it has answered. A launcher that dies before that is + # caught by the parent check, and one that fails its startup + # terminates its attached frontends itself. + launcher_answered = False next_health = loop.time() + self._health_interval while True: await asyncio.sleep(self._ppid_interval) @@ -189,8 +217,11 @@ async def run(self) -> None: next_health = loop.time() + self._health_interval status = await self._forwarder.get_status("/health") if status == 200: + launcher_answered = True failures = 0 continue + if not launcher_answered: + continue failures += 1 if failures >= self._health_failures: self._die( diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index b10b0e33c4d7..069ac0cc0811 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -929,14 +929,7 @@ async def validation_exception_handler(request, exc): self._init_embedding_batcher() self.register_embedding_routes() else: - if self._launcher_forwarder is not None: - # Registered first so they shadow the local handlers below - # (FastAPI matches in registration order). - for method, path in FORWARDED_ROUTES: - self.app.add_api_route(path, - self._launcher_forwarder.forward, - methods=[method], - include_in_schema=False) + self._register_launcher_forwarding_routes() self.register_routes() if self._collect_perf_metrics: @@ -1775,6 +1768,20 @@ async def data_transceiver_state(self) -> JSONResponse: base64.b64encode(state).decode("utf-8") }) + def _register_launcher_forwarding_routes(self) -> None: + """On an attached frontend, hand launcher-owned routes to the launcher. + + Must run before register_routes(): FastAPI matches in registration + order, so these shadow the local handlers of the same routes. + """ + if self._launcher_forwarder is None: + return + for method, path in FORWARDED_ROUTES: + self.app.add_api_route(path, + self._launcher_forwarder.forward, + methods=[method], + include_in_schema=False) + def _on_launcher_lost(self, error: BaseException) -> None: """Watchdog callback of an attached frontend: fail fast, then shut down. diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py index 832011e28b11..8c0abb64b576 100644 --- a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py +++ b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py @@ -16,25 +16,32 @@ An attached frontend forwards launcher-owned routes to the launcher over a Unix socket and watches the launcher for engine or process death. Both are -exercised here against a fake launcher; the GPU path is covered end to end by -the serving integration tests. +exercised here against a fake launcher, together with the OpenAIServer hooks +that wire them in; spawning real frontends needs a GPU engine. """ import asyncio import os +import signal import threading +import time from contextlib import asynccontextmanager +from types import SimpleNamespace +import msgspec import pytest from aiohttp import web from fastapi import FastAPI from fastapi.testclient import TestClient +from starlette.routing import Match +from tensorrt_llm.serve import openai_server from tensorrt_llm.serve.multi_frontend import ( FORWARDED_ROUTES, AttachedFrontendWatchdog, LauncherForwarder, ) +from tensorrt_llm.serve.openai_server import OpenAIServer pytestmark = pytest.mark.cpu_only @@ -45,6 +52,7 @@ class _FakeLauncher: def __init__(self, path: str, health_status: int = 200) -> None: self.path = path self.health_status = health_status + self.health_hits = 0 self.seen = [] self._loop = asyncio.new_event_loop() self._ready = threading.Event() @@ -56,13 +64,16 @@ def _serve(self) -> None: asyncio.set_event_loop(self._loop) async def echo(request: web.Request) -> web.Response: - self.seen.append((request.method, request.path_qs, await request.text())) + self.seen.append( + (request.method, request.path_qs, await request.read(), dict(request.headers)) + ) return web.json_response( {"method": request.method, "path": request.path_qs}, status=201 if request.method == "POST" else 200, ) async def health(_: web.Request) -> web.Response: + self.health_hits += 1 return web.Response(status=self.health_status) app = web.Application() @@ -111,9 +122,9 @@ def test_forwarder_replays_launcher_owned_routes(tmp_path) -> None: r = client.post("/v1/messages/batches/b1/cancel", json={"reason": "x"}) assert r.status_code == 201 assert r.json()["path"] == "/v1/messages/batches/b1/cancel" - method, path, body = launcher.seen[-1] + method, path, body, _ = launcher.seen[-1] assert (method, path) == ("POST", "/v1/messages/batches/b1/cancel") - assert '"reason"' in body + assert b'"reason"' in body # Routes outside FORWARDED_ROUTES stay local. assert client.get("/health").json() == {"local": True} @@ -121,6 +132,38 @@ def test_forwarder_replays_launcher_owned_routes(tmp_path) -> None: launcher.close() +def test_forwarder_keeps_end_to_end_headers(tmp_path) -> None: + # The body is replayed byte for byte, so the header that says how to + # decode it (x-trtllm-msgpack) must travel with it, and so must auth + # headers a --middleware on the launcher checks again. + launcher = _FakeLauncher(str(tmp_path / "launcher.sock")) + body = msgspec.msgpack.encode({"requests": []}) + try: + with TestClient(_attached_app(LauncherForwarder(launcher.path))) as client: + r = client.post( + "/v1/messages/batches", + content=body, + headers={ + "content-type": "application/msgpack", + "x-trtllm-msgpack": "1", + "x-api-key": "secret", + }, + ) + assert r.status_code == 201 + finally: + launcher.close() + _, path, seen_body, headers = launcher.seen[-1] + headers = {k.lower(): v for k, v in headers.items()} + assert path == "/v1/messages/batches" + assert seen_body == body + assert headers["x-trtllm-msgpack"] == "1" + assert headers["x-api-key"] == "secret" + assert headers["content-type"] == "application/msgpack" + # Hop-by-hop and framing headers are the forwarder's own, not the client's. + assert headers["host"] != "testserver" + assert headers["content-length"] == str(len(body)) + + def test_forwarder_reports_unreachable_launcher(tmp_path) -> None: with TestClient(_attached_app(LauncherForwarder(str(tmp_path / "missing.sock")))) as client: r = client.get("/metrics") @@ -128,8 +171,15 @@ def test_forwarder_reports_unreachable_launcher(tmp_path) -> None: assert "unavailable" in r.json()["error"] +async def _wait_for(predicate, timeout: float = 10.0) -> None: + deadline = time.monotonic() + timeout + while not predicate(): + assert time.monotonic() < deadline, "condition not reached in time" + await asyncio.sleep(0.01) + + def test_watchdog_marks_dead_after_consecutive_health_failures(tmp_path) -> None: - launcher = _FakeLauncher(str(tmp_path / "launcher.sock"), health_status=503) + launcher = _FakeLauncher(str(tmp_path / "launcher.sock")) deaths = [] async def run() -> None: @@ -143,7 +193,11 @@ async def run() -> None: health_interval_s=0.02, health_failures=2, ) - await asyncio.wait_for(watchdog.run(), timeout=10) + task = asyncio.create_task(watchdog.run()) + await _wait_for(lambda: launcher.health_hits > 0) + assert not deaths + launcher.health_status = 503 + await asyncio.wait_for(task, timeout=10) finally: await forwarder.close() @@ -155,6 +209,47 @@ async def run() -> None: assert "failed 2 consecutive" in str(deaths[0]) +def test_watchdog_waits_for_a_starting_launcher(tmp_path) -> None: + # The launcher starts listening only after its attached frontends are + # ready; failures before it first answered must not kill the frontend. + path = str(tmp_path / "launcher.sock") + deaths = [] + + async def run() -> None: + forwarder = LauncherForwarder(path) + watchdog = AttachedFrontendWatchdog( + forwarder, + os.getppid(), + deaths.append, + ppid_interval_s=0.01, + health_interval_s=0.02, + health_failures=2, + ) + task = asyncio.create_task(watchdog.run()) + launcher = None + try: + # Nothing listens yet: ~25 probes, any 2 of which used to be fatal. + await asyncio.sleep(0.5) + assert not deaths and not task.done() + launcher = await asyncio.to_thread(_FakeLauncher, path, 503) + await _wait_for(lambda: launcher.health_hits >= 3) + assert not deaths and not task.done() + launcher.health_status = 200 + await _wait_for(lambda: launcher.health_hits >= 4) + launcher.health_status = 503 + await asyncio.wait_for(task, timeout=10) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await forwarder.close() + if launcher is not None: + await asyncio.to_thread(launcher.close) + + asyncio.run(run()) + assert len(deaths) == 1 + assert "failed 2 consecutive" in str(deaths[0]) + + def test_watchdog_marks_dead_when_launcher_process_is_gone(tmp_path) -> None: deaths = [] @@ -170,3 +265,93 @@ async def run() -> None: asyncio.run(run()) assert len(deaths) == 1 assert "exited" in str(deaths[0]) + + +def _attached_server(forwarder: LauncherForwarder) -> OpenAIServer: + """Just enough of an attached-frontend OpenAIServer to register routes.""" + server = object.__new__(OpenAIServer) + server.app = FastAPI() + server.generator = SimpleNamespace( + _executor=SimpleNamespace(resource_governor_queue=None), + args=SimpleNamespace(return_perf_metrics=False), + ) + server.use_harmony = False + server._enable_rl_control_endpoints = False + server._enable_runtime_control_endpoints = False + server.resource_governor = None + server._launcher_forwarder = forwarder + return server + + +def test_attached_server_forwards_before_local_handlers(tmp_path) -> None: + forwarder = LauncherForwarder(str(tmp_path / "launcher.sock")) + server = _attached_server(forwarder) + # The order OpenAIServer.__init__ registers them in. + server._register_launcher_forwarding_routes() + server.register_routes() + for method, template in FORWARDED_ROUTES: + path = template.replace("{rest:path}", "b1/cancel") + scope = {"type": "http", "method": method, "path": path, "path_params": {}} + first = next(r for r in server.app.routes if r.matches(scope)[0] == Match.FULL) + assert first.endpoint == forwarder.forward, f"{method} {path} is answered locally" + # Routes that are not launcher-owned keep their local handler. + scope = {"type": "http", "method": "GET", "path": "/health", "path_params": {}} + first = next(r for r in server.app.routes if r.matches(scope)[0] == Match.FULL) + assert first.endpoint == server.health + + +def test_single_frontend_server_registers_no_forwarding() -> None: + server = _attached_server(None) + before = list(server.app.routes) + server._register_launcher_forwarding_routes() + assert server.app.routes == before + + +class _FakeExecutor: + def __init__(self, doing_shutdown: bool) -> None: + self.doing_shutdown = doing_shutdown + self._fatal_error = None + self.fatal_errors = [] + + def _set_fatal_error(self, error: BaseException) -> None: + self.fatal_errors.append(error) + self._fatal_error = error + + +def _server_losing_launcher(monkeypatch, executor: _FakeExecutor): + server = object.__new__(OpenAIServer) + server.generator = SimpleNamespace(_executor=executor) + signals, terminations = [], [] + monkeypatch.setattr(openai_server.signal, "raise_signal", signals.append) + monkeypatch.setattr(openai_server, "_record_generator_termination", terminations.append) + return server, signals, terminations + + +@pytest.mark.parametrize("doing_shutdown", [False, True]) +def test_on_launcher_lost_fails_fast_then_shuts_down(monkeypatch, doing_shutdown: bool) -> None: + executor = _FakeExecutor(doing_shutdown) + server, signals, terminations = _server_losing_launcher(monkeypatch, executor) + + error = RuntimeError("attached frontend lost its engine: test") + server._on_launcher_lost(error) + + # In-flight and new requests fail with this error instead of hanging. + assert executor.fatal_errors == [error] + if doing_shutdown: + # Shutdown is already under way: no second SIGINT to force-quit it. + assert signals == [] and terminations == [] + else: + assert signals == [signal.SIGINT] + assert terminations == [server.generator] + + +def test_on_launcher_lost_keeps_an_earlier_fatal_error(monkeypatch) -> None: + executor = _FakeExecutor(doing_shutdown=False) + first = RuntimeError("engine error recorded first") + executor._fatal_error = first + server, signals, _ = _server_losing_launcher(monkeypatch, executor) + + server._on_launcher_lost(RuntimeError("attached frontend lost its engine: test")) + + assert executor.fatal_errors == [] and executor._fatal_error is first + assert signals == [signal.SIGINT] diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py b/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py index bdf218211c79..873174f1b07c 100644 --- a/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py +++ b/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py @@ -73,10 +73,20 @@ "/steady_clock_offset", ): "known limitation: calibrates the answering frontend only (documented)", # RL control requires AsyncLLM, which is not the classic IPC executor path - # multi-frontend mode runs on; the two never coexist. - ("POST", "/release_memory"): "AsyncLLM-only; never registered together with attached frontends", - ("POST", "/resume_memory"): "AsyncLLM-only; never registered together with attached frontends", + # multi-frontend mode runs on, and runtime control (--enable_sleep_mode) + # requires num_serve_frontends=1; neither coexists with attached frontends. + ( + "POST", + "/release_memory", + ): "RL / runtime control only; never coexists with attached frontends", + ("POST", "/resume_memory"): "RL / runtime control only; never coexists with attached frontends", ("POST", "/update_weights"): "AsyncLLM-only; never registered together with attached frontends", + ("GET", "/memory_status"): "runtime control requires num_serve_frontends=1", + # FastAPI's generated schema and docs pages: static, identical everywhere. + ("GET", "/openapi.json"): "generated from the route table, identical in every frontend", + ("GET", "/docs"): "static page over /openapi.json", + ("GET", "/docs/oauth2-redirect"): "static page", + ("GET", "/redoc"): "static page over /openapi.json", } @@ -89,7 +99,10 @@ def _llm_server_routes() -> list[tuple[str, str]]: args=SimpleNamespace(return_perf_metrics=False), ) server.use_harmony = False + # Mutually exclusive in OpenAIServer.__init__; enabling both here only + # enumerates every route either of them adds. server._enable_rl_control_endpoints = True + server._enable_runtime_control_endpoints = True server.resource_governor = None server.register_routes() return sorted( @@ -101,13 +114,15 @@ def _llm_server_routes() -> list[tuple[str, str]]: ) -def _forwarding_app() -> FastAPI: - app = FastAPI() +async def _stub(): + return {} - async def _stub(): - return {} - for method, path in FORWARDED_ROUTES: +def _forwarding_app(routes=FORWARDED_ROUTES) -> FastAPI: + # Without FastAPI's generated /docs, /openapi.json, ... routes, which would + # otherwise "match" the same routes of the server under test. + app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None) + for method, path in routes: app.add_api_route(path, _stub, methods=[method]) return app @@ -147,17 +162,9 @@ def test_every_llm_route_is_classified() -> None: def test_every_forwarded_route_exists() -> None: """A renamed route must not leave a dangling forwarding template.""" registered = _llm_server_routes() - reg_app = FastAPI() - - async def _stub(): - return {} - - for method, path in registered: - reg_app.add_api_route(path, _stub, methods=[method]) for method, template in FORWARDED_ROUTES: - probe = template.replace("{rest:path}", "x/cancel") - scope = {"type": "http", "method": method, "path": probe, "path_params": {}} - assert any(route.matches(scope)[0] == Match.FULL for route in reg_app.routes), ( + app = _forwarding_app([(method, template)]) + assert any(_is_forwarded(app, m, path) for m, path in registered), ( f"FORWARDED_ROUTES entry {method} {template} matches no registered route" ) From 35bef6961d5729ff527c1416c742294b33414a40 Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Sun, 18 Oct 2026 01:14:22 -0700 Subject: [PATCH 4/9] [None][fix] serve: make the disagg clock offset and server registration launcher-owned Two more pieces of server-wide state that multi-frontend serving kept once per process: - POST /steady_clock_offset: the disaggregated server calibrates a server once, so the offset reached a single frontend while every frontend stamps its own requests' perf metrics; across hosts, the other frontends' records stayed in the worker's own clock domain. The route is now launcher-owned: attached frontends forward it, the launcher publishes the offset in the multi-frontend ipc dir, and attached frontends apply it within 0.25 s (SharedClockOffset / follow_clock_offset). The request path never reads it. - Metadata-server and disagg-cluster registration: every frontend registered the server under its own process identity (llm_id, DisaggClusterWorker id), listing one server once per frontend -- a K-fold weight in the metadata router's server list and a prematurely satisfied minimal_instances in the disagg cluster. Only the launcher registers now; it reaches uvicorn after its attached frontends, so a registered server serves on every frontend. Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- tensorrt_llm/serve/multi_frontend.py | 76 ++++++++++- tensorrt_llm/serve/openai_server.py | 118 ++++++++++++----- .../test_serve_multi_frontend_helpers.py | 121 +++++++++++++++++- .../test_serve_multi_frontend_route_gate.py | 5 +- 4 files changed, 274 insertions(+), 46 deletions(-) diff --git a/tensorrt_llm/serve/multi_frontend.py b/tensorrt_llm/serve/multi_frontend.py index a93402af2931..a0cf36088139 100644 --- a/tensorrt_llm/serve/multi_frontend.py +++ b/tensorrt_llm/serve/multi_frontend.py @@ -17,16 +17,22 @@ Several frontend processes share one executor and one ``SO_REUSEPORT`` port, and a client cannot choose which of them answers. Anything that needs exactly one owner per server therefore lives in the launcher (frontend 0): draining -the engine's iteration statistics and KV-cache events, runtime profiling, and -the in-memory Anthropic Message Batches store. The launcher's uvicorn also -listens on a Unix domain socket in the multi-frontend ipc directory; attached -frontends forward those routes to it (:class:`LauncherForwarder`) and watch it -(:class:`AttachedFrontendWatchdog`) so that a dead launcher or engine does not -leave them answering ``/health`` 200 with nothing behind them. +the engine's iteration statistics and KV-cache events, runtime profiling, the +in-memory Anthropic Message Batches store, the disaggregated server's clock +offset, and the server's registration with a metadata server or disagg +cluster. The launcher's uvicorn also listens on a Unix domain socket in the +multi-frontend ipc directory; attached frontends forward those routes to it +(:class:`LauncherForwarder`), follow the clock offset it publishes there +(:class:`SharedClockOffset`), and watch it (:class:`AttachedFrontendWatchdog`) +so that a dead launcher or engine does not leave them answering ``/health`` +200 with nothing behind them. """ import asyncio +import contextlib +import math import os +import tempfile from dataclasses import dataclass from typing import Callable, Optional @@ -39,6 +45,8 @@ LAUNCHER_UDS_NAME = "launcher.sock" # Carries the launcher's Unix socket path to the attached frontends it spawns. LAUNCHER_UDS_ENV = "TLLM_SERVE_LAUNCHER_UDS" +# The clock offset the launcher publishes in the multi-frontend ipc dir. +CLOCK_OFFSET_NAME = "clock_offset" # Routes an attached frontend hands to the launcher, as FastAPI route # templates. They are registered before the regular routes, and FastAPI @@ -49,6 +57,7 @@ ("POST", "/kv_cache_events"), ("POST", "/start_profile"), ("POST", "/stop_profile"), + ("POST", "/steady_clock_offset"), ("GET", "/v1/messages/batches"), ("POST", "/v1/messages/batches"), ("GET", "/v1/messages/batches/{rest:path}"), @@ -169,6 +178,61 @@ async def close(self) -> None: self._session = None +class SharedClockOffset: + """The disaggregated server's clock offset, shared by all frontends. + + The disaggregated server measures this server's steady-clock offset and + POSTs it once, so it reaches a single frontend, which forwards it to the + launcher. Every frontend stamps its own requests' perf metrics with the + offset, though, so the launcher publishes it in the multi-frontend ipc dir + and the attached frontends follow it (:func:`follow_clock_offset`). + """ + + def __init__(self, ipc_dir: str): + self._path = os.path.join(ipc_dir, CLOCK_OFFSET_NAME) + self._seen = None + + def publish(self, offset: float) -> None: + """Launcher side: atomically replace the published offset.""" + fd, tmp = tempfile.mkstemp(prefix=f".{CLOCK_OFFSET_NAME}.", dir=os.path.dirname(self._path)) + try: + with os.fdopen(fd, "w") as f: + f.write(repr(float(offset))) + os.replace(tmp, self._path) + except BaseException: + with contextlib.suppress(OSError): + os.unlink(tmp) + raise + + def poll(self) -> Optional[float]: + """The published offset if it changed since the last poll, else None.""" + try: + with open(self._path) as f: + offset = float(f.read()) + except (OSError, ValueError): + return None + if not math.isfinite(offset) or offset == self._seen: + return None + self._seen = offset + return offset + + +async def follow_clock_offset( + shared: SharedClockOffset, apply: Callable[[float], None], interval_s: float = 0.25 +) -> None: + """Attached-frontend loop: apply every offset the launcher publishes. + + The disaggregated server calibrates a server when its router adds it, so + only requests routed within ``interval_s`` of that can still be stamped + with the previous offset; in exchange the request path never reads the file. + """ + while True: + offset = shared.poll() + if offset is not None: + apply(offset) + await asyncio.sleep(interval_s) + + class AttachedFrontendWatchdog: """Detect a lost launcher or engine from an attached frontend. diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 069ac0cc0811..797da0b1e471 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -96,7 +96,9 @@ from tensorrt_llm.serve.multi_frontend import (FORWARDED_ROUTES, AttachedFrontendWatchdog, LauncherForwarder, - MultiFrontendServing) + MultiFrontendServing, + SharedClockOffset, + follow_clock_offset) from tensorrt_llm.serve.openai_protocol import ( ChatCompletionNamedToolChoiceParam, ChatCompletionRequest, ChatCompletionResponse, ChatCompletionResponseChoice, @@ -646,7 +648,14 @@ def __init__( self._embedding_max_queue_size = embedding_max_queue_size self.embedding_batcher: Optional[EncodeBatcher] = None self.tool_parser = tool_parser - self.metadata_server = create_metadata_server(metadata_server_cfg) + # One server, one registration: attached frontends serve the + # launcher's port, so only the launcher advertises it to a metadata + # server or disagg cluster. Registering under each process's own + # identity listed one server once per frontend. + self._registers_server = (multi_frontend_serving is None + or multi_frontend_serving.is_launcher) + self.metadata_server = (create_metadata_server(metadata_server_cfg) + if self._registers_server else None) self.disagg_cluster_config = disagg_cluster_config self.multimodal_server_config = multimodal_server_config # Seed media-IO decode defaults from the model's input processor, then @@ -741,6 +750,13 @@ def __init__( if multi_frontend_serving is not None and multi_frontend_serving.is_attached: self._launcher_forwarder = LauncherForwarder( multi_frontend_serving.launcher_uds) + # POST /steady_clock_offset is launcher-owned; the launcher publishes + # the offset next to its socket and attached frontends follow it. + self._shared_clock_offset: Optional[SharedClockOffset] = ( + SharedClockOffset( + os.path.dirname(multi_frontend_serving.launcher_uds)) + if multi_frontend_serving is not None else None) + self._clock_offset_task: Optional[asyncio.Task] = None # Requests finishing on attached frontends never wake the launcher's # collector, so the launcher also polls on a fixed cadence. self._iteration_stats_poll_interval: Optional[float] = ( @@ -776,27 +792,7 @@ def __init__( @asynccontextmanager async def lifespan(app: FastAPI): await self._perf_metrics_writer.start() - if self.metadata_server is not None: - metadata = { - "model": self.model, - "version": VERSION, - "timestamp": datetime.now().isoformat(), - "server_role": server_role.name, - "url": self.binding_addr - } - # TODO: add more metadata - # Register with ETCD using the existing key format - self.metadata_server.put(f"trtllm/{self.generator.llm_id}", - metadata) - logger.info(f"trtllm/{self.generator.llm_id} is registered") - - if self.disagg_cluster_config: - self.disagg_cluster_storage = create_cluster_storage_client( - self.disagg_cluster_config.cluster_uri, - self.disagg_cluster_config.cluster_name) - self.disagg_cluster_worker = DisaggClusterWorker( - self.server_role, self.host, self.port, - self.disagg_cluster_config, self.disagg_cluster_storage) + self._register_server() # VisualGen has no args if not self._is_visual_gen: @@ -821,6 +817,9 @@ async def lifespan(app: FastAPI): self._launcher_forwarder, self._multi_frontend.launcher_pid, self._on_launcher_lost).run()) + self._clock_offset_task = asyncio.create_task( + follow_clock_offset(self._shared_clock_offset, + self._apply_shared_clock_offset)) logger.info( "Attached frontend: forwarding launcher-owned routes " f"to {self._multi_frontend.launcher_uds}") @@ -857,12 +856,13 @@ async def lifespan(app: FastAPI): yield await self._perf_metrics_writer.close() - if self._launcher_watchdog_task is not None: - self._launcher_watchdog_task.cancel() - try: - await self._launcher_watchdog_task - except asyncio.CancelledError: - pass + for task in (self._launcher_watchdog_task, self._clock_offset_task): + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass if self._launcher_forwarder is not None: await self._launcher_forwarder.close() if self.embedding_batcher is not None: @@ -878,11 +878,7 @@ async def lifespan(app: FastAPI): pass logger.info("Stopped background iteration stats collector task") - if self.metadata_server is not None: - self.metadata_server.remove(f"trtllm/{self.generator.llm_id}") - logger.info(f"trtllm/{self.generator.llm_id} is unregistered") - if self.disagg_cluster_worker: - await self.disagg_cluster_worker.deregister_worker() + await self._deregister_server() if self.resource_governor is not None: self.resource_governor.close() self.generator.shutdown() @@ -1782,6 +1778,48 @@ def _register_launcher_forwarding_routes(self) -> None: methods=[method], include_in_schema=False) + def _register_server(self) -> None: + """Advertise this server to a metadata server and/or disagg cluster. + + Attached frontends skip this (see ``_registers_server``); the disagg + cluster worker created here registers once uvicorn serves. + """ + if not self._registers_server: + return + if self.metadata_server is not None: + metadata = { + "model": self.model, + "version": VERSION, + "timestamp": datetime.now().isoformat(), + "server_role": self.server_role.name, + "url": self.binding_addr + } + # TODO: add more metadata + # Register with ETCD using the existing key format + self.metadata_server.put(f"trtllm/{self.generator.llm_id}", + metadata) + logger.info(f"trtllm/{self.generator.llm_id} is registered") + + if self.disagg_cluster_config: + self.disagg_cluster_storage = create_cluster_storage_client( + self.disagg_cluster_config.cluster_uri, + self.disagg_cluster_config.cluster_name) + self.disagg_cluster_worker = DisaggClusterWorker( + self.server_role, self.host, self.port, + self.disagg_cluster_config, self.disagg_cluster_storage) + + async def _deregister_server(self) -> None: + if self.metadata_server is not None: + self.metadata_server.remove(f"trtllm/{self.generator.llm_id}") + logger.info(f"trtllm/{self.generator.llm_id} is unregistered") + if self.disagg_cluster_worker: + await self.disagg_cluster_worker.deregister_worker() + + def _apply_shared_clock_offset(self, offset: float) -> None: + self._adjusted_steady_clock.set_reference_offset(offset) + logger.info(f"Following the launcher's steady clock offset between " + f"local and disagg server: {offset} second") + def _on_launcher_lost(self, error: BaseException) -> None: """Watchdog callback of an attached frontend: fail fast, then shut down. @@ -1972,6 +2010,18 @@ async def set_steady_clock_offset( logger.info( f"The steady clock offset between local and disagg server: {offset} second" ) + # Multi-frontend launcher (attached frontends forward this route + # here): every frontend stamps its own requests' perf metrics, so + # share the offset with them. + shared = getattr(self, "_shared_clock_offset", None) + if shared is not None: + try: + shared.publish(offset) + except OSError as e: + logger.error( + f"Failed to share the steady clock offset with the " + f"attached frontends: {e}") + return Response(status_code=500, content=str(e)) return Response(status_code=200) async def get_steady_clock_offset(self) -> JSONResponse: diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py index 8c0abb64b576..ca14751ed9c5 100644 --- a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py +++ b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py @@ -15,9 +15,11 @@ """CPU coverage for trtllm-serve's multi-frontend helpers. An attached frontend forwards launcher-owned routes to the launcher over a -Unix socket and watches the launcher for engine or process death. Both are -exercised here against a fake launcher, together with the OpenAIServer hooks -that wire them in; spawning real frontends needs a GPU engine. +Unix socket, follows the clock offset the launcher publishes, and watches the +launcher for engine or process death. These are exercised here against a fake +launcher, together with the OpenAIServer hooks that wire them in (including +that only the launcher registers the server); spawning real frontends needs a +GPU engine. """ import asyncio @@ -27,6 +29,7 @@ import time from contextlib import asynccontextmanager from types import SimpleNamespace +from unittest import mock import msgspec import pytest @@ -35,11 +38,15 @@ from fastapi.testclient import TestClient from starlette.routing import Match +from tensorrt_llm._utils import AdjustedSteadyClock +from tensorrt_llm.llmapi.disagg_utils import ServerRole from tensorrt_llm.serve import openai_server from tensorrt_llm.serve.multi_frontend import ( FORWARDED_ROUTES, AttachedFrontendWatchdog, LauncherForwarder, + SharedClockOffset, + follow_clock_offset, ) from tensorrt_llm.serve.openai_server import OpenAIServer @@ -355,3 +362,111 @@ def test_on_launcher_lost_keeps_an_earlier_fatal_error(monkeypatch) -> None: assert executor.fatal_errors == [] and executor._fatal_error is first assert signals == [signal.SIGINT] + + +def test_shared_clock_offset_publish_and_poll(tmp_path) -> None: + launcher = SharedClockOffset(str(tmp_path)) + follower = SharedClockOffset(str(tmp_path)) + assert follower.poll() is None # nothing published yet + launcher.publish(-12.5) + assert follower.poll() == -12.5 + assert follower.poll() is None # unchanged since the last poll + launcher.publish(-12.5) + assert follower.poll() is None # same value: nothing to apply + # Back-to-back publishes: the latest value wins, nothing is skipped. + launcher.publish(0.125) + launcher.publish(0.25) + assert follower.poll() == 0.25 + # A torn, foreign or non-finite file is ignored rather than applied. + for junk in ("not a number", "nan", "inf"): + (tmp_path / "clock_offset").write_text(junk) + assert follower.poll() is None + assert [p.name for p in tmp_path.iterdir()] == ["clock_offset"] + + +def test_follow_clock_offset_applies_each_published_offset(tmp_path) -> None: + launcher = SharedClockOffset(str(tmp_path)) + applied = [] + + async def run() -> None: + task = asyncio.create_task( + follow_clock_offset(SharedClockOffset(str(tmp_path)), applied.append, 0.01) + ) + try: + await asyncio.sleep(0.05) + assert applied == [] + launcher.publish(3.0) + await _wait_for(lambda: applied == [3.0]) + launcher.publish(-1.5) + await _wait_for(lambda: applied == [3.0, -1.5]) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + asyncio.run(run()) + + +def _clock_server(shared) -> OpenAIServer: + server = object.__new__(OpenAIServer) + server._adjusted_steady_clock = AdjustedSteadyClock(time_source=lambda: 100.0) + server._shared_clock_offset = shared + return server + + +def test_launcher_shares_the_disagg_clock_offset(tmp_path) -> None: + server = _clock_server(SharedClockOffset(str(tmp_path))) + response = asyncio.run(server.set_steady_clock_offset(offset=-7.25)) + assert response.status_code == 200 + assert server._adjusted_steady_clock.now() == 100.0 - 7.25 + assert SharedClockOffset(str(tmp_path)).poll() == -7.25 + + +def test_unshared_clock_offset_is_reported(tmp_path) -> None: + # The disaggregated server warns about misaligned timestamps on non-200. + server = _clock_server(SharedClockOffset(str(tmp_path / "gone"))) + response = asyncio.run(server.set_steady_clock_offset(offset=1.0)) + assert response.status_code == 500 + + +def test_single_frontend_sets_its_clock_only() -> None: + server = _clock_server(None) + response = asyncio.run(server.set_steady_clock_offset(offset=2.0)) + assert response.status_code == 200 + assert server._adjusted_steady_clock.now() == 102.0 + + +def _registering_server(registers: bool) -> OpenAIServer: + server = object.__new__(OpenAIServer) + server._registers_server = registers + server.metadata_server = mock.Mock() + server.disagg_cluster_config = SimpleNamespace(cluster_uri="etcd://h:2379", cluster_name="c") + server.disagg_cluster_worker = None + server.server_role = ServerRole.CONTEXT + server.model = "m" + server.binding_addr = "http://h:8000" + server.host, server.port = "h", 8000 + server.generator = SimpleNamespace(llm_id="h-1-0") + return server + + +@pytest.mark.parametrize("is_launcher", [True, False]) +def test_only_the_launcher_registers_the_server(monkeypatch, is_launcher: bool) -> None: + storage = mock.Mock() + worker_cls = mock.Mock() + monkeypatch.setattr(openai_server, "create_cluster_storage_client", storage) + monkeypatch.setattr(openai_server, "DisaggClusterWorker", worker_cls) + server = _registering_server(registers=is_launcher) + + server._register_server() + + if is_launcher: + key, metadata = server.metadata_server.put.call_args.args + assert key == "trtllm/h-1-0" and metadata["url"] == "http://h:8000" + worker_cls.assert_called_once() + assert server.disagg_cluster_worker is worker_cls.return_value + else: + # Attached frontends serve the launcher's port: one server, one entry. + server.metadata_server.put.assert_not_called() + storage.assert_not_called() + worker_cls.assert_not_called() + assert server.disagg_cluster_worker is None diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py b/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py index 873174f1b07c..3d045b70084f 100644 --- a/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py +++ b/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py @@ -67,11 +67,10 @@ ("POST", "/_internal/tokenize"): "tokenizer-only", ("GET", "/energy_metrics"): "reads GPU energy counters; same devices from every process", ("GET", "/v1/data_transceiver_state"): "fetched from the shared executor over RPC", - ("GET", "/steady_clock_offset"): "clock probe of the answering process", ( - "POST", + "GET", "/steady_clock_offset", - ): "known limitation: calibrates the answering frontend only (documented)", + ): "clock probe; every frontend reads the same host steady clock (the POST is forwarded)", # RL control requires AsyncLLM, which is not the classic IPC executor path # multi-frontend mode runs on, and runtime control (--enable_sleep_mode) # requires num_serve_frontends=1; neither coexists with attached frontends. From 6135889c23e14989bf8c4d2a12eab111a248d8f5 Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Sat, 24 Oct 2026 08:28:46 -0700 Subject: [PATCH 5/9] [None][fix] serve: read idle-poll iteration stats on the metrics thread Main now keeps /metrics stats reads off the default executor (#20698), on a dedicated metrics thread. The multi-frontend launcher's 1 s idle poll of the engine stats queue used asyncio.to_thread, i.e. that same default executor; run it on the metrics thread instead, which is idle while the collector loop owns the queue (/metrics then serves from the collector's buffer). Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- tensorrt_llm/serve/openai_server.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 797da0b1e471..f4e42cec9ee4 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -2145,9 +2145,13 @@ async def _iteration_stats_collector_loop(self): # Idle poll: the stats RPC would otherwise wait out # its timeout inside a synchronous .remote() call and # stall this event loop, so fetch once without - # waiting, off the loop. - for llm_stat in await asyncio.to_thread( - self.generator.get_stats, 0): + # waiting, off the loop. The metrics thread is idle + # while this loop owns the queue (/metrics then reads + # the buffer), and keeps the read off the default + # executor that request handling uses. + for llm_stat in await asyncio.get_running_loop( + ).run_in_executor(self._metrics_stats_executor, + self.generator.get_stats, 0): self._record_iteration_stat(llm_stat) else: async for llm_stat in self.generator.get_stats_async( From 000f4ba36fd96c5745361485e730b0664a71cc95 Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Sat, 24 Oct 2026 08:50:43 -0700 Subject: [PATCH 6/9] [None][test] serve: trim multi-frontend tests and comments Merge overlapping multi-frontend unit tests, drop ones that only restated an implementation detail, and shorten comments and docstrings in the multi-frontend code. No behavior change. Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- tensorrt_llm/commands/serve.py | 30 +- tensorrt_llm/serve/multi_frontend.py | 76 ++-- tensorrt_llm/serve/openai_server.py | 61 +--- .../test_serve_multi_frontend_helpers.py | 339 +++++------------- .../test_serve_multi_frontend_route_gate.py | 135 +++---- 5 files changed, 171 insertions(+), 470 deletions(-) diff --git a/tensorrt_llm/commands/serve.py b/tensorrt_llm/commands/serve.py index 4ad68671d071..ec2ad80d0445 100644 --- a/tensorrt_llm/commands/serve.py +++ b/tensorrt_llm/commands/serve.py @@ -521,12 +521,10 @@ def _provision_kv_cache_pool(llm_args: dict, def _bind_launcher_uds(llm) -> Tuple[socket.socket, str]: - """Bind the launcher's Unix socket inside the multi-frontend ipc dir. + """Bind the Unix socket attached frontends forward launcher-owned routes to. - uvicorn serves it next to the TCP socket, so attached frontends can - forward launcher-owned routes (/metrics, /kv_cache_events, profiling, - message batches) to the one process that owns the engine. The ipc dir - is private to the launcher's user and removed by the executor proxy. + It lives in the multi-frontend ipc dir, which is private to the user and + removed by the executor proxy. """ ipc_dir = getattr(getattr(llm, "_executor", None), "_multi_frontend_ipc_dir", None) @@ -545,19 +543,10 @@ def _bind_launcher_uds(llm) -> Tuple[socket.socket, str]: def _share_prometheus_multiproc_dir(llm, ipc_dir: str) -> None: - """Give every frontend one PROMETHEUS_MULTIPROC_DIR before spawning. - - prometheus_client's multiprocess mode aggregates the .db files under - that directory. Left to OpenAIServer, each process would create its own - after the children already exist, and every /prometheus/metrics scrape - would report a single frontend's share of the request counters. - - The directory lives inside the multi-frontend ipc dir, which the - executor proxy removes at shutdown. It is deliberately not created via - set_prometheus_multiproc_dir(): that helper keeps only its most recent - TemporaryDirectory alive, so its second call (OpenAIServer.__init__) - would garbage-collect, and thereby delete, a shared directory created - by the first while every process still writes into it. + """Give every frontend one PROMETHEUS_MULTIPROC_DIR before spawning them. + + Not via set_prometheus_multiproc_dir(): it keeps only its latest + TemporaryDirectory alive, so OpenAIServer's later call would delete this one. """ args = getattr(llm, "args", None) if args is None or os.environ.get("PROMETHEUS_MULTIPROC_DIR"): @@ -601,9 +590,8 @@ def launch_server( model = served_model_name or llm_args["model"] multi_frontend = _init_multi_frontend_mode(llm_args, multi_frontend_enabled) - # An attached frontend forwards launcher-owned routes to the launcher's - # Unix socket, whose path the launcher hands down when spawning it (see - # serve/multi_frontend.py). The launcher is this process's parent. + # An attached frontend gets the launcher's socket from its spawn env; the + # launcher is its parent. multi_frontend_serving = None if multi_frontend.is_attached_frontend: attached_launcher_uds = os.environ.pop(LAUNCHER_UDS_ENV, None) diff --git a/tensorrt_llm/serve/multi_frontend.py b/tensorrt_llm/serve/multi_frontend.py index a0cf36088139..ea7aa6bd4cea 100644 --- a/tensorrt_llm/serve/multi_frontend.py +++ b/tensorrt_llm/serve/multi_frontend.py @@ -14,18 +14,11 @@ # limitations under the License. """Helpers for trtllm-serve multi-frontend serving (``num_serve_frontends > 1``). -Several frontend processes share one executor and one ``SO_REUSEPORT`` port, -and a client cannot choose which of them answers. Anything that needs exactly -one owner per server therefore lives in the launcher (frontend 0): draining -the engine's iteration statistics and KV-cache events, runtime profiling, the -in-memory Anthropic Message Batches store, the disaggregated server's clock -offset, and the server's registration with a metadata server or disagg -cluster. The launcher's uvicorn also listens on a Unix domain socket in the -multi-frontend ipc directory; attached frontends forward those routes to it -(:class:`LauncherForwarder`), follow the clock offset it publishes there -(:class:`SharedClockOffset`), and watch it (:class:`AttachedFrontendWatchdog`) -so that a dead launcher or engine does not leave them answering ``/health`` -200 with nothing behind them. +The frontends share one executor and one ``SO_REUSEPORT`` port, so a client +cannot pick the process that answers. State that exists once per server lives +in the launcher (frontend 0), which also serves a Unix socket in the +multi-frontend ipc dir: attached frontends forward the routes backed by that +state to it, follow the clock offset it publishes, and watch it for death. """ import asyncio @@ -43,15 +36,11 @@ from tensorrt_llm.logger import logger LAUNCHER_UDS_NAME = "launcher.sock" -# Carries the launcher's Unix socket path to the attached frontends it spawns. LAUNCHER_UDS_ENV = "TLLM_SERVE_LAUNCHER_UDS" -# The clock offset the launcher publishes in the multi-frontend ipc dir. CLOCK_OFFSET_NAME = "clock_offset" -# Routes an attached frontend hands to the launcher, as FastAPI route -# templates. They are registered before the regular routes, and FastAPI -# matches in registration order, so they shadow the local handlers. The -# catch-all covers /v1/messages/batches/{batch_id}[/cancel|/results]. +# Launcher-owned routes. Attached frontends register them ahead of the local +# handlers, which they shadow since FastAPI matches in registration order. FORWARDED_ROUTES = ( ("GET", "/metrics"), ("POST", "/kv_cache_events"), @@ -65,11 +54,9 @@ ("DELETE", "/v1/messages/batches/{rest:path}"), ) -# Request headers not replayed to the launcher: hop-by-hop headers (RFC 9110, -# section 7.6.1) and the framing/encoding headers aiohttp manages itself for -# the replayed request (the body it hands back is already decoded). Everything -# else arrives as the client sent it: x-trtllm-msgpack selects the body codec, -# and auth headers feed the --middleware the launcher runs again. +# Hop-by-hop and framing headers, which aiohttp sets for the replayed request. +# The rest is forwarded: x-trtllm-msgpack selects the body codec, and auth +# headers feed the --middleware the launcher runs again. _NOT_FORWARDED_REQUEST_HEADERS = frozenset( ( "accept-encoding", @@ -97,8 +84,7 @@ class MultiFrontendServing: launcher_uds: str is_launcher: bool - # The launcher's pid as seen by an attached frontend (its parent). - launcher_pid: Optional[int] = None + launcher_pid: Optional[int] = None # attached frontends: their parent @property def is_attached(self) -> bool: @@ -114,10 +100,9 @@ def __init__(self, uds_path: str, timeout_s: float = 120.0): self._session: Optional[aiohttp.ClientSession] = None async def _get_session(self) -> aiohttp.ClientSession: - # Created lazily so it binds to the serving event loop. + # Lazy, to bind to the serving loop. These routes are low-rate, so no + # keep-alive connections to race against a launcher shutting down. if self._session is None or self._session.closed: - # force_close: no idle keep-alive connections to race against a - # launcher that is shutting down; these routes are low-rate. self._session = aiohttp.ClientSession( connector=aiohttp.UnixConnector(path=self._uds_path, force_close=True), timeout=self._timeout, @@ -137,8 +122,7 @@ async def forward(self, request: Request) -> Response: body = await request.body() try: session = await self._get_session() - # The host is required by the URL syntax and ignored by the - # Unix connector. + # The Unix connector ignores the host. async with session.request( request.method, f"http://launcher{target}", data=body, headers=headers ) as resp: @@ -179,13 +163,10 @@ async def close(self) -> None: class SharedClockOffset: - """The disaggregated server's clock offset, shared by all frontends. + """The disaggregated server's clock offset, published by the launcher. - The disaggregated server measures this server's steady-clock offset and - POSTs it once, so it reaches a single frontend, which forwards it to the - launcher. Every frontend stamps its own requests' perf metrics with the - offset, though, so the launcher publishes it in the multi-frontend ipc dir - and the attached frontends follow it (:func:`follow_clock_offset`). + The disaggregated server POSTs it once, but every frontend stamps its own + requests' perf metrics with it. """ def __init__(self, ipc_dir: str): @@ -220,12 +201,7 @@ def poll(self) -> Optional[float]: async def follow_clock_offset( shared: SharedClockOffset, apply: Callable[[float], None], interval_s: float = 0.25 ) -> None: - """Attached-frontend loop: apply every offset the launcher publishes. - - The disaggregated server calibrates a server when its router adds it, so - only requests routed within ``interval_s`` of that can still be stamped - with the previous offset; in exchange the request path never reads the file. - """ + """Attached-frontend loop: apply every offset the launcher publishes.""" while True: offset = shared.poll() if offset is not None: @@ -236,12 +212,9 @@ async def follow_clock_offset( class AttachedFrontendWatchdog: """Detect a lost launcher or engine from an attached frontend. - An attached frontend owns no worker processes, so the only signals that - the engine is gone are its parent (the launcher) disappearing, or the - launcher's own ``/health`` failing, which happens once the launcher's - executor recorded a fatal engine error. Either way ``mark_dead`` is - called exactly once; the caller records the fatal error on its executor - (failing in-flight requests fast) and starts server shutdown. + An attached frontend owns no workers: its only signals are the launcher + process exiting and the launcher's ``/health`` failing. ``mark_dead`` is + called once. """ def __init__( @@ -264,11 +237,8 @@ def __init__( async def run(self) -> None: loop = asyncio.get_running_loop() failures = 0 - # The launcher builds its own server, and starts listening, only after - # every attached frontend reported ready, so /health failures count - # only once it has answered. A launcher that dies before that is - # caught by the parent check, and one that fails its startup - # terminates its attached frontends itself. + # The launcher starts listening only after its attached frontends are + # ready, so failures count once it has answered. launcher_answered = False next_health = loop.time() + self._health_interval while True: diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index f4e42cec9ee4..3b14d999b839 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -648,10 +648,7 @@ def __init__( self._embedding_max_queue_size = embedding_max_queue_size self.embedding_batcher: Optional[EncodeBatcher] = None self.tool_parser = tool_parser - # One server, one registration: attached frontends serve the - # launcher's port, so only the launcher advertises it to a metadata - # server or disagg cluster. Registering under each process's own - # identity listed one server once per frontend. + # Attached frontends serve the launcher's port: only it registers. self._registers_server = (multi_frontend_serving is None or multi_frontend_serving.is_launcher) self.metadata_server = (create_metadata_server(metadata_server_cfg) @@ -739,26 +736,20 @@ def __init__( ) if server_role is not None else "server" self._perf_metrics_writer = PerfMetricsJsonlWriter( perf_metrics_output_dir, server_kind) - # Multi-frontend serving (num_serve_frontends > 1): the launcher is - # the sole consumer of the engine's iteration stats / KV events and - # the only owner of profiling and the batch store; attached - # frontends forward those routes to it over its Unix socket and - # watch it for engine or launcher death (see serve/multi_frontend.py). + # Multi-frontend serving: see serve/multi_frontend.py. self._multi_frontend = multi_frontend_serving self._launcher_forwarder: Optional[LauncherForwarder] = None self._launcher_watchdog_task: Optional[asyncio.Task] = None if multi_frontend_serving is not None and multi_frontend_serving.is_attached: self._launcher_forwarder = LauncherForwarder( multi_frontend_serving.launcher_uds) - # POST /steady_clock_offset is launcher-owned; the launcher publishes - # the offset next to its socket and attached frontends follow it. self._shared_clock_offset: Optional[SharedClockOffset] = ( SharedClockOffset( os.path.dirname(multi_frontend_serving.launcher_uds)) if multi_frontend_serving is not None else None) self._clock_offset_task: Optional[asyncio.Task] = None # Requests finishing on attached frontends never wake the launcher's - # collector, so the launcher also polls on a fixed cadence. + # collector, so it also polls. self._iteration_stats_poll_interval: Optional[float] = ( 1.0 if multi_frontend_serving is not None and multi_frontend_serving.is_launcher else None) @@ -809,8 +800,7 @@ async def lifespan(app: FastAPI): # Start background iteration stats collector if metrics are enabled # The PyTorch backend args include `enable_iter_perf_stats`. - # An attached frontend never drains the engine queue: the - # launcher owns it and /metrics is forwarded there. + # Attached frontends leave the engine queue to the launcher. if self._launcher_forwarder is not None: self._launcher_watchdog_task = asyncio.create_task( AttachedFrontendWatchdog( @@ -1057,9 +1047,7 @@ def _init_llm(self, chat_template: Optional[str] = None): request_inference_time_buckets=( pmc.request_inference_time_buckets if pmc else None), ) - # With a shared PROMETHEUS_MULTIPROC_DIR every process would - # export its own per-pid config-info series; the launcher's - # suffices (attached frontends share its engine and config). + # One config-info series per server, not per frontend process. if self._launcher_forwarder is None: self._log_config_info_metrics() @@ -1765,11 +1753,7 @@ async def data_transceiver_state(self) -> JSONResponse: }) def _register_launcher_forwarding_routes(self) -> None: - """On an attached frontend, hand launcher-owned routes to the launcher. - - Must run before register_routes(): FastAPI matches in registration - order, so these shadow the local handlers of the same routes. - """ + """Attached frontends: forward launcher-owned routes (before register_routes()).""" if self._launcher_forwarder is None: return for method, path in FORWARDED_ROUTES: @@ -1779,11 +1763,7 @@ def _register_launcher_forwarding_routes(self) -> None: include_in_schema=False) def _register_server(self) -> None: - """Advertise this server to a metadata server and/or disagg cluster. - - Attached frontends skip this (see ``_registers_server``); the disagg - cluster worker created here registers once uvicorn serves. - """ + """Register with the metadata server and/or disagg cluster (launcher only).""" if not self._registers_server: return if self.metadata_server is not None: @@ -1821,11 +1801,9 @@ def _apply_shared_clock_offset(self, offset: float) -> None: f"local and disagg server: {offset} second") def _on_launcher_lost(self, error: BaseException) -> None: - """Watchdog callback of an attached frontend: fail fast, then shut down. + """Watchdog callback: fail requests with EngineDeadError, then shut down. - Mirrors the fatal-error branch of ``health``: record the error on the - executor so in-flight and new requests raise EngineDeadError instead - of hanging, then raise SIGINT once for uvicorn's graceful shutdown. + Mirrors the fatal-error branch of ``health``. """ executor = getattr(self.generator, '_executor', None) if executor is None: @@ -2010,9 +1988,7 @@ async def set_steady_clock_offset( logger.info( f"The steady clock offset between local and disagg server: {offset} second" ) - # Multi-frontend launcher (attached frontends forward this route - # here): every frontend stamps its own requests' perf metrics, so - # share the offset with them. + # Multi-frontend launcher: every frontend stamps perf metrics with it. shared = getattr(self, "_shared_clock_offset", None) if shared is not None: try: @@ -2068,8 +2044,7 @@ async def _extract_metrics(self, res: RequestOutput, raw_request: Request): self.metrics_collector.log_request_metrics_dict( res.metrics_dict) # Note: Iteration stats are collected by the background _iteration_stats_collector_loop task - # Wake up the stats collector to drain iteration stats (an - # attached frontend has no collector: the launcher drains). + # Wake up the stats collector to drain iteration stats if self._iteration_stats_collector_task is not None and getattr( self.generator.args, "enable_iter_perf_stats", True): self._iteration_stats_wakeup_event.set() @@ -2123,10 +2098,7 @@ async def _iteration_stats_collector_loop(self): logger.info("Iteration stats collector loop started") while True: # Wait for signal that requests have completed and stats may - # be available. A multi-frontend launcher also wakes on a - # fixed cadence: requests finishing on attached frontends - # never set this event, yet their iterations land in the - # engine queue this loop alone drains. + # be available (a multi-frontend launcher also polls). woken_by_timer = False try: await asyncio.wait_for( @@ -2142,13 +2114,8 @@ async def _iteration_stats_collector_loop(self): # Drain all available iteration stats and log each one to Prometheus. try: if woken_by_timer: - # Idle poll: the stats RPC would otherwise wait out - # its timeout inside a synchronous .remote() call and - # stall this event loop, so fetch once without - # waiting, off the loop. The metrics thread is idle - # while this loop owns the queue (/metrics then reads - # the buffer), and keeps the read off the default - # executor that request handling uses. + # Idle poll: a non-blocking read on the metrics + # thread, idle while this loop owns the queue. for llm_stat in await asyncio.get_running_loop( ).run_in_executor(self._metrics_stats_executor, self.generator.get_stats, 0): diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py index ca14751ed9c5..b32cfee8d82c 100644 --- a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py +++ b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py @@ -12,15 +12,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""CPU coverage for trtllm-serve's multi-frontend helpers. - -An attached frontend forwards launcher-owned routes to the launcher over a -Unix socket, follows the clock offset the launcher publishes, and watches the -launcher for engine or process death. These are exercised here against a fake -launcher, together with the OpenAIServer hooks that wire them in (including -that only the launcher registers the server); spawning real frontends needs a -GPU engine. -""" +"""CPU tests for trtllm-serve multi-frontend serving, against a fake launcher.""" import asyncio import os @@ -46,7 +38,6 @@ AttachedFrontendWatchdog, LauncherForwarder, SharedClockOffset, - follow_clock_offset, ) from tensorrt_llm.serve.openai_server import OpenAIServer @@ -75,8 +66,7 @@ async def echo(request: web.Request) -> web.Response: (request.method, request.path_qs, await request.read(), dict(request.headers)) ) return web.json_response( - {"method": request.method, "path": request.path_qs}, - status=201 if request.method == "POST" else 200, + {"path": request.path_qs}, status=201 if request.method == "POST" else 200 ) async def health(_: web.Request) -> web.Response: @@ -99,83 +89,69 @@ def close(self) -> None: self._loop.close() -def _attached_app(forwarder: LauncherForwarder) -> FastAPI: - """The forwarded routes as OpenAIServer registers them, plus a local one.""" +def _attached_server(forwarder: LauncherForwarder) -> OpenAIServer: + """An attached-frontend OpenAIServer with just enough state to route.""" @asynccontextmanager async def lifespan(_: FastAPI): yield await forwarder.close() - app = FastAPI(lifespan=lifespan) - for method, path in FORWARDED_ROUTES: - app.add_api_route(path, forwarder.forward, methods=[method]) - - @app.get("/health") - async def local_health(): - return {"local": True} - - return app + server = object.__new__(OpenAIServer) + server.app = FastAPI(lifespan=lifespan) + server.generator = SimpleNamespace( + _executor=SimpleNamespace(resource_governor_queue=None), + args=SimpleNamespace(return_perf_metrics=False), + ) + server.use_harmony = False + server._enable_rl_control_endpoints = False + server._enable_runtime_control_endpoints = False + server.resource_governor = None + server._launcher_forwarder = forwarder + # The order OpenAIServer.__init__ registers them in. + server._register_launcher_forwarding_routes() + server.register_routes() + return server -def test_forwarder_replays_launcher_owned_routes(tmp_path) -> None: +def test_attached_server_forwards_launcher_owned_routes(tmp_path) -> None: launcher = _FakeLauncher(str(tmp_path / "launcher.sock")) - try: - with TestClient(_attached_app(LauncherForwarder(launcher.path))) as client: - r = client.get("/metrics?limit=3") - assert r.status_code == 200 - assert r.json() == {"method": "GET", "path": "/metrics?limit=3"} - - r = client.post("/v1/messages/batches/b1/cancel", json={"reason": "x"}) - assert r.status_code == 201 - assert r.json()["path"] == "/v1/messages/batches/b1/cancel" - method, path, body, _ = launcher.seen[-1] - assert (method, path) == ("POST", "/v1/messages/batches/b1/cancel") - assert b'"reason"' in body - - # Routes outside FORWARDED_ROUTES stay local. - assert client.get("/health").json() == {"local": True} - finally: - launcher.close() + forwarder = LauncherForwarder(launcher.path) + server = _attached_server(forwarder) + # Every launcher-owned route resolves to the forwarder before its local handler. + for method, template in FORWARDED_ROUTES: + path = template.replace("{rest:path}", "b1/cancel") + scope = {"type": "http", "method": method, "path": path, "path_params": {}} + first = next(r for r in server.app.routes if r.matches(scope)[0] == Match.FULL) + assert first.endpoint == forwarder.forward, f"{method} {path} is answered locally" -def test_forwarder_keeps_end_to_end_headers(tmp_path) -> None: - # The body is replayed byte for byte, so the header that says how to - # decode it (x-trtllm-msgpack) must travel with it, and so must auth - # headers a --middleware on the launcher checks again. - launcher = _FakeLauncher(str(tmp_path / "launcher.sock")) body = msgspec.msgpack.encode({"requests": []}) try: - with TestClient(_attached_app(LauncherForwarder(launcher.path))) as client: + with TestClient(server.app) as client: + r = client.get("/metrics?limit=3") + assert (r.status_code, r.json()) == (200, {"path": "/metrics?limit=3"}) + # The body is replayed byte for byte, so its codec header must travel + # with it, and so must auth headers a launcher --middleware checks. r = client.post( "/v1/messages/batches", content=body, - headers={ - "content-type": "application/msgpack", - "x-trtllm-msgpack": "1", - "x-api-key": "secret", - }, + headers={"x-trtllm-msgpack": "1", "x-api-key": "secret"}, ) assert r.status_code == 201 + assert client.get("/version").status_code == 200 # stays local finally: launcher.close() + assert len(launcher.seen) == 2 _, path, seen_body, headers = launcher.seen[-1] headers = {k.lower(): v for k, v in headers.items()} - assert path == "/v1/messages/batches" - assert seen_body == body - assert headers["x-trtllm-msgpack"] == "1" - assert headers["x-api-key"] == "secret" - assert headers["content-type"] == "application/msgpack" - # Hop-by-hop and framing headers are the forwarder's own, not the client's. - assert headers["host"] != "testserver" - assert headers["content-length"] == str(len(body)) - - -def test_forwarder_reports_unreachable_launcher(tmp_path) -> None: - with TestClient(_attached_app(LauncherForwarder(str(tmp_path / "missing.sock")))) as client: + assert (path, seen_body) == ("/v1/messages/batches", body) + assert headers["x-trtllm-msgpack"] == "1" and headers["x-api-key"] == "secret" + assert headers["host"] != "testserver" # hop-by-hop headers are not replayed + + with TestClient(server.app) as client: r = client.get("/metrics") - assert r.status_code == 503 - assert "unavailable" in r.json()["error"] + assert r.status_code == 503 and "unavailable" in r.json()["error"] async def _wait_for(predicate, timeout: float = 10.0) -> None: @@ -185,40 +161,7 @@ async def _wait_for(predicate, timeout: float = 10.0) -> None: await asyncio.sleep(0.01) -def test_watchdog_marks_dead_after_consecutive_health_failures(tmp_path) -> None: - launcher = _FakeLauncher(str(tmp_path / "launcher.sock")) - deaths = [] - - async def run() -> None: - forwarder = LauncherForwarder(launcher.path) - try: - watchdog = AttachedFrontendWatchdog( - forwarder, - os.getppid(), - deaths.append, - ppid_interval_s=0.01, - health_interval_s=0.02, - health_failures=2, - ) - task = asyncio.create_task(watchdog.run()) - await _wait_for(lambda: launcher.health_hits > 0) - assert not deaths - launcher.health_status = 503 - await asyncio.wait_for(task, timeout=10) - finally: - await forwarder.close() - - try: - asyncio.run(run()) - finally: - launcher.close() - assert len(deaths) == 1 - assert "failed 2 consecutive" in str(deaths[0]) - - -def test_watchdog_waits_for_a_starting_launcher(tmp_path) -> None: - # The launcher starts listening only after its attached frontends are - # ready; failures before it first answered must not kill the frontend. +def test_watchdog_counts_health_failures_only_after_the_launcher_answered(tmp_path) -> None: path = str(tmp_path / "launcher.sock") deaths = [] @@ -235,9 +178,9 @@ async def run() -> None: task = asyncio.create_task(watchdog.run()) launcher = None try: - # Nothing listens yet: ~25 probes, any 2 of which used to be fatal. + # The launcher starts listening after its attached frontends: ~25 + # failed probes against a socket nobody serves yet are not fatal. await asyncio.sleep(0.5) - assert not deaths and not task.done() launcher = await asyncio.to_thread(_FakeLauncher, path, 503) await _wait_for(lambda: launcher.health_hits >= 3) assert not deaths and not task.done() @@ -253,11 +196,10 @@ async def run() -> None: await asyncio.to_thread(launcher.close) asyncio.run(run()) - assert len(deaths) == 1 - assert "failed 2 consecutive" in str(deaths[0]) + assert len(deaths) == 1 and "failed 2 consecutive" in str(deaths[0]) -def test_watchdog_marks_dead_when_launcher_process_is_gone(tmp_path) -> None: +def test_watchdog_notices_a_gone_launcher_process(tmp_path) -> None: deaths = [] async def run() -> None: @@ -270,203 +212,90 @@ async def run() -> None: await forwarder.close() asyncio.run(run()) - assert len(deaths) == 1 - assert "exited" in str(deaths[0]) + assert len(deaths) == 1 and "exited" in str(deaths[0]) -def _attached_server(forwarder: LauncherForwarder) -> OpenAIServer: - """Just enough of an attached-frontend OpenAIServer to register routes.""" - server = object.__new__(OpenAIServer) - server.app = FastAPI() - server.generator = SimpleNamespace( - _executor=SimpleNamespace(resource_governor_queue=None), - args=SimpleNamespace(return_perf_metrics=False), +@pytest.mark.parametrize("doing_shutdown", [False, True]) +def test_on_launcher_lost_fails_fast_then_shuts_down(monkeypatch, doing_shutdown: bool) -> None: + fatal_errors = [] + executor = SimpleNamespace( + doing_shutdown=doing_shutdown, _fatal_error=None, _set_fatal_error=fatal_errors.append ) - server.use_harmony = False - server._enable_rl_control_endpoints = False - server._enable_runtime_control_endpoints = False - server.resource_governor = None - server._launcher_forwarder = forwarder - return server - - -def test_attached_server_forwards_before_local_handlers(tmp_path) -> None: - forwarder = LauncherForwarder(str(tmp_path / "launcher.sock")) - server = _attached_server(forwarder) - # The order OpenAIServer.__init__ registers them in. - server._register_launcher_forwarding_routes() - server.register_routes() - for method, template in FORWARDED_ROUTES: - path = template.replace("{rest:path}", "b1/cancel") - scope = {"type": "http", "method": method, "path": path, "path_params": {}} - first = next(r for r in server.app.routes if r.matches(scope)[0] == Match.FULL) - assert first.endpoint == forwarder.forward, f"{method} {path} is answered locally" - # Routes that are not launcher-owned keep their local handler. - scope = {"type": "http", "method": "GET", "path": "/health", "path_params": {}} - first = next(r for r in server.app.routes if r.matches(scope)[0] == Match.FULL) - assert first.endpoint == server.health - - -def test_single_frontend_server_registers_no_forwarding() -> None: - server = _attached_server(None) - before = list(server.app.routes) - server._register_launcher_forwarding_routes() - assert server.app.routes == before - - -class _FakeExecutor: - def __init__(self, doing_shutdown: bool) -> None: - self.doing_shutdown = doing_shutdown - self._fatal_error = None - self.fatal_errors = [] - - def _set_fatal_error(self, error: BaseException) -> None: - self.fatal_errors.append(error) - self._fatal_error = error - - -def _server_losing_launcher(monkeypatch, executor: _FakeExecutor): server = object.__new__(OpenAIServer) server.generator = SimpleNamespace(_executor=executor) signals, terminations = [], [] monkeypatch.setattr(openai_server.signal, "raise_signal", signals.append) monkeypatch.setattr(openai_server, "_record_generator_termination", terminations.append) - return server, signals, terminations - - -@pytest.mark.parametrize("doing_shutdown", [False, True]) -def test_on_launcher_lost_fails_fast_then_shuts_down(monkeypatch, doing_shutdown: bool) -> None: - executor = _FakeExecutor(doing_shutdown) - server, signals, terminations = _server_losing_launcher(monkeypatch, executor) error = RuntimeError("attached frontend lost its engine: test") server._on_launcher_lost(error) - # In-flight and new requests fail with this error instead of hanging. - assert executor.fatal_errors == [error] - if doing_shutdown: - # Shutdown is already under way: no second SIGINT to force-quit it. - assert signals == [] and terminations == [] - else: - assert signals == [signal.SIGINT] - assert terminations == [server.generator] - - -def test_on_launcher_lost_keeps_an_earlier_fatal_error(monkeypatch) -> None: - executor = _FakeExecutor(doing_shutdown=False) - first = RuntimeError("engine error recorded first") - executor._fatal_error = first - server, signals, _ = _server_losing_launcher(monkeypatch, executor) - - server._on_launcher_lost(RuntimeError("attached frontend lost its engine: test")) - - assert executor.fatal_errors == [] and executor._fatal_error is first - assert signals == [signal.SIGINT] + assert fatal_errors == [error] + # A shutdown already under way gets no second SIGINT (which would force-quit it). + assert signals == ([] if doing_shutdown else [signal.SIGINT]) + assert terminations == ([] if doing_shutdown else [server.generator]) def test_shared_clock_offset_publish_and_poll(tmp_path) -> None: - launcher = SharedClockOffset(str(tmp_path)) - follower = SharedClockOffset(str(tmp_path)) - assert follower.poll() is None # nothing published yet + launcher, follower = SharedClockOffset(str(tmp_path)), SharedClockOffset(str(tmp_path)) + assert follower.poll() is None launcher.publish(-12.5) assert follower.poll() == -12.5 - assert follower.poll() is None # unchanged since the last poll - launcher.publish(-12.5) - assert follower.poll() is None # same value: nothing to apply - # Back-to-back publishes: the latest value wins, nothing is skipped. + assert follower.poll() is None # unchanged launcher.publish(0.125) launcher.publish(0.25) - assert follower.poll() == 0.25 - # A torn, foreign or non-finite file is ignored rather than applied. - for junk in ("not a number", "nan", "inf"): + assert follower.poll() == 0.25 # the latest value wins + for junk in ("not a number", "nan"): (tmp_path / "clock_offset").write_text(junk) assert follower.poll() is None assert [p.name for p in tmp_path.iterdir()] == ["clock_offset"] -def test_follow_clock_offset_applies_each_published_offset(tmp_path) -> None: - launcher = SharedClockOffset(str(tmp_path)) - applied = [] - - async def run() -> None: - task = asyncio.create_task( - follow_clock_offset(SharedClockOffset(str(tmp_path)), applied.append, 0.01) - ) - try: - await asyncio.sleep(0.05) - assert applied == [] - launcher.publish(3.0) - await _wait_for(lambda: applied == [3.0]) - launcher.publish(-1.5) - await _wait_for(lambda: applied == [3.0, -1.5]) - finally: - task.cancel() - await asyncio.gather(task, return_exceptions=True) - - asyncio.run(run()) - - -def _clock_server(shared) -> OpenAIServer: +@pytest.mark.parametrize( + "role,status", + [("launcher", 200), ("launcher_without_ipc_dir", 500), ("single_frontend", 200)], +) +def test_set_steady_clock_offset(tmp_path, role: str, status: int) -> None: + shared = { + "launcher": SharedClockOffset(str(tmp_path)), + "launcher_without_ipc_dir": SharedClockOffset(str(tmp_path / "gone")), + "single_frontend": None, + }[role] server = object.__new__(OpenAIServer) server._adjusted_steady_clock = AdjustedSteadyClock(time_source=lambda: 100.0) server._shared_clock_offset = shared - return server - -def test_launcher_shares_the_disagg_clock_offset(tmp_path) -> None: - server = _clock_server(SharedClockOffset(str(tmp_path))) response = asyncio.run(server.set_steady_clock_offset(offset=-7.25)) - assert response.status_code == 200 - assert server._adjusted_steady_clock.now() == 100.0 - 7.25 - assert SharedClockOffset(str(tmp_path)).poll() == -7.25 - - -def test_unshared_clock_offset_is_reported(tmp_path) -> None: - # The disaggregated server warns about misaligned timestamps on non-200. - server = _clock_server(SharedClockOffset(str(tmp_path / "gone"))) - response = asyncio.run(server.set_steady_clock_offset(offset=1.0)) - assert response.status_code == 500 - -def test_single_frontend_sets_its_clock_only() -> None: - server = _clock_server(None) - response = asyncio.run(server.set_steady_clock_offset(offset=2.0)) - assert response.status_code == 200 - assert server._adjusted_steady_clock.now() == 102.0 + # A non-200 makes the disaggregated server warn about misaligned timestamps. + assert response.status_code == status + assert server._adjusted_steady_clock.now() == 92.75 + if role == "launcher": + assert SharedClockOffset(str(tmp_path)).poll() == -7.25 -def _registering_server(registers: bool) -> OpenAIServer: +@pytest.mark.parametrize("is_launcher", [True, False]) +def test_only_the_launcher_registers_the_server(monkeypatch, is_launcher: bool) -> None: + storage, worker_cls = mock.Mock(), mock.Mock() + monkeypatch.setattr(openai_server, "create_cluster_storage_client", storage) + monkeypatch.setattr(openai_server, "DisaggClusterWorker", worker_cls) server = object.__new__(OpenAIServer) - server._registers_server = registers + server._registers_server = is_launcher server.metadata_server = mock.Mock() server.disagg_cluster_config = SimpleNamespace(cluster_uri="etcd://h:2379", cluster_name="c") server.disagg_cluster_worker = None server.server_role = ServerRole.CONTEXT - server.model = "m" - server.binding_addr = "http://h:8000" + server.model, server.binding_addr = "m", "http://h:8000" server.host, server.port = "h", 8000 server.generator = SimpleNamespace(llm_id="h-1-0") - return server - - -@pytest.mark.parametrize("is_launcher", [True, False]) -def test_only_the_launcher_registers_the_server(monkeypatch, is_launcher: bool) -> None: - storage = mock.Mock() - worker_cls = mock.Mock() - monkeypatch.setattr(openai_server, "create_cluster_storage_client", storage) - monkeypatch.setattr(openai_server, "DisaggClusterWorker", worker_cls) - server = _registering_server(registers=is_launcher) server._register_server() if is_launcher: key, metadata = server.metadata_server.put.call_args.args assert key == "trtllm/h-1-0" and metadata["url"] == "http://h:8000" - worker_cls.assert_called_once() assert server.disagg_cluster_worker is worker_cls.return_value else: - # Attached frontends serve the launcher's port: one server, one entry. server.metadata_server.put.assert_not_called() storage.assert_not_called() worker_cls.assert_not_called() - assert server.disagg_cluster_worker is None diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py b/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py index 3d045b70084f..e477c020d314 100644 --- a/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py +++ b/tests/unittest/llmapi/test_serve_multi_frontend_route_gate.py @@ -12,19 +12,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""Route-ownership gate for trtllm-serve multi-frontend serving. +"""Every OpenAIServer route must have a multi-frontend owner. -With ``num_serve_frontends > 1`` a client cannot choose which frontend process -answers, so every HTTP route of the LLM ``OpenAIServer`` must be classified: - -* **launcher-owned** -- backed by state that exists once per server (engine - stats queue, profiler, batch store). Attached frontends forward these to the - launcher; the list is ``tensorrt_llm.serve.multi_frontend.FORWARDED_ROUTES``. -* **per-frontend safe** -- correct (or deliberately per-process) when served by - whichever frontend accepted the connection; listed below with the reason. - -A new route that is in neither set fails this test, forcing the author to -decide instead of silently shipping a per-process copy of server-wide state. +A route either is forwarded to the launcher (state that exists once per server, +``FORWARDED_ROUTES``) or is safe to answer from whichever frontend accepted the +connection (``PER_FRONTEND_SAFE`` below, with the reason). A new route in +neither set fails here, so its author has to decide. """ from types import SimpleNamespace @@ -38,54 +31,32 @@ pytestmark = pytest.mark.cpu_only -# (method, path template) -> why it is fine to answer from any frontend. PER_FRONTEND_SAFE = { - ( - "GET", - "/health", - ): "liveness of the answering process; engine death propagates via the watchdog", - ("GET", "/health_generate"): "runs a real request through the shared executor", + ("GET", "/health"): "liveness of this process; engine death reaches it via the watchdog", + ("GET", "/health_generate"): "a real request through the shared executor", ("GET", "/version"): "constant", ("GET", "/v1/models"): "constant", - ("GET", "/server_info"): "static server/model configuration, identical in every frontend", - ("POST", "/v1/completions"): "request path: the whole point of multiple frontends", + ("GET", "/server_info"): "static configuration", + ("POST", "/v1/completions"): "request path", ("POST", "/v1/chat/completions"): "request path", - ("POST", "/v1/messages"): "request path (Anthropic adapter)", - ("POST", "/v1/messages/count_tokens"): "tokenizer-only, identical in every frontend", - ( - "POST", - "/v1/responses", - ): "request path; the stateful store is disabled group-wide in multi-frontend mode", - ( - "GET", - "/v1/responses/{response_id}", - ): "always 404 in multi-frontend mode (store disabled group-wide)", - ( - "DELETE", - "/v1/responses/{response_id}", - ): "always 404 in multi-frontend mode (store disabled group-wide)", - ("POST", "/_internal/tokenize"): "tokenizer-only", - ("GET", "/energy_metrics"): "reads GPU energy counters; same devices from every process", - ("GET", "/v1/data_transceiver_state"): "fetched from the shared executor over RPC", - ( - "GET", - "/steady_clock_offset", - ): "clock probe; every frontend reads the same host steady clock (the POST is forwarded)", - # RL control requires AsyncLLM, which is not the classic IPC executor path - # multi-frontend mode runs on, and runtime control (--enable_sleep_mode) - # requires num_serve_frontends=1; neither coexists with attached frontends. - ( - "POST", - "/release_memory", - ): "RL / runtime control only; never coexists with attached frontends", - ("POST", "/resume_memory"): "RL / runtime control only; never coexists with attached frontends", - ("POST", "/update_weights"): "AsyncLLM-only; never registered together with attached frontends", - ("GET", "/memory_status"): "runtime control requires num_serve_frontends=1", - # FastAPI's generated schema and docs pages: static, identical everywhere. - ("GET", "/openapi.json"): "generated from the route table, identical in every frontend", - ("GET", "/docs"): "static page over /openapi.json", + ("POST", "/v1/messages"): "request path", + ("POST", "/v1/messages/count_tokens"): "tokenizer only", + ("POST", "/v1/responses"): "request path; the stateful store is off with several frontends", + ("GET", "/v1/responses/{response_id}"): "store is off with several frontends", + ("DELETE", "/v1/responses/{response_id}"): "store is off with several frontends", + ("POST", "/_internal/tokenize"): "tokenizer only", + ("GET", "/energy_metrics"): "the same GPUs from every process", + ("GET", "/v1/data_transceiver_state"): "read from the shared executor", + ("GET", "/steady_clock_offset"): "every frontend reads the same host clock", + # RL control needs AsyncLLM and runtime control needs one frontend. + ("POST", "/release_memory"): "never served by attached frontends", + ("POST", "/resume_memory"): "never served by attached frontends", + ("POST", "/update_weights"): "never served by attached frontends", + ("GET", "/memory_status"): "never served by attached frontends", + ("GET", "/openapi.json"): "generated, identical everywhere", + ("GET", "/docs"): "static page", ("GET", "/docs/oauth2-redirect"): "static page", - ("GET", "/redoc"): "static page over /openapi.json", + ("GET", "/redoc"): "static page", } @@ -98,8 +69,7 @@ def _llm_server_routes() -> list[tuple[str, str]]: args=SimpleNamespace(return_perf_metrics=False), ) server.use_harmony = False - # Mutually exclusive in OpenAIServer.__init__; enabling both here only - # enumerates every route either of them adds. + # Mutually exclusive in practice; both on to enumerate every route. server._enable_rl_control_endpoints = True server._enable_runtime_control_endpoints = True server.resource_governor = None @@ -107,7 +77,7 @@ def _llm_server_routes() -> list[tuple[str, str]]: return sorted( (method, route.path) for route in server.app.routes - if hasattr(route, "methods") and route.methods + if getattr(route, "methods", None) for method in route.methods if method != "HEAD" ) @@ -118,8 +88,7 @@ async def _stub(): def _forwarding_app(routes=FORWARDED_ROUTES) -> FastAPI: - # Without FastAPI's generated /docs, /openapi.json, ... routes, which would - # otherwise "match" the same routes of the server under test. + # No generated docs routes: they would "match" the server's own. app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None) for method, path in routes: app.add_api_route(path, _stub, methods=[method]) @@ -127,48 +96,26 @@ def _forwarding_app(routes=FORWARDED_ROUTES) -> FastAPI: def _is_forwarded(app: FastAPI, method: str, path_template: str) -> bool: - # Concretise the registered template so the forwarding templates - # (including the {rest:path} catch-all) can be matched against it. - concrete = path_template - for name in ("{batch_id}", "{response_id}"): - concrete = concrete.replace(name, "x") + concrete = path_template.replace("{batch_id}", "x").replace("{response_id}", "x") scope = {"type": "http", "method": method, "path": concrete, "path_params": {}} return any(route.matches(scope)[0] == Match.FULL for route in app.routes) def test_every_llm_route_is_classified() -> None: - fwd = _forwarding_app() - unclassified = [] - double = [] - for method, path in _llm_server_routes(): - forwarded = _is_forwarded(fwd, method, path) - safe = (method, path) in PER_FRONTEND_SAFE - if forwarded and safe: - double.append((method, path)) - elif not forwarded and not safe: - unclassified.append((method, path)) - assert not double, f"routes both forwarded and per-frontend-safe: {double}" + forwarding = _forwarding_app() + routes = _llm_server_routes() + forwarded = {r for r in routes if _is_forwarded(forwarding, *r)} + assert not forwarded & PER_FRONTEND_SAFE.keys(), "both forwarded and per-frontend safe" + unclassified = [r for r in routes if r not in forwarded and r not in PER_FRONTEND_SAFE] assert not unclassified, ( - "New OpenAIServer route(s) without a multi-frontend ownership decision: " - f"{unclassified}. Either add them to FORWARDED_ROUTES in " - "tensorrt_llm/serve/multi_frontend.py (state lives once per server, e.g. " - "an engine queue, the profiler or an in-memory store) or to " - "PER_FRONTEND_SAFE in this test with the reason they are correct when " - "served by whichever frontend accepted the connection." + f"Routes without a multi-frontend owner: {unclassified}. Add them to " + "FORWARDED_ROUTES (server-wide state) or to PER_FRONTEND_SAFE with the reason." ) -def test_every_forwarded_route_exists() -> None: - """A renamed route must not leave a dangling forwarding template.""" - registered = _llm_server_routes() +def test_no_stale_entries() -> None: + routes = _llm_server_routes() for method, template in FORWARDED_ROUTES: app = _forwarding_app([(method, template)]) - assert any(_is_forwarded(app, m, path) for m, path in registered), ( - f"FORWARDED_ROUTES entry {method} {template} matches no registered route" - ) - - -def test_allowlist_has_no_stale_entries() -> None: - registered = set(_llm_server_routes()) - stale = [r for r in PER_FRONTEND_SAFE if r not in registered] - assert not stale, f"PER_FRONTEND_SAFE lists routes that no longer exist: {stale}" + assert any(_is_forwarded(app, *r) for r in routes), f"{method} {template} matches nothing" + assert not PER_FRONTEND_SAFE.keys() - set(routes), "PER_FRONTEND_SAFE lists removed routes" From 3b26f18fd95901095cb598fea6df03b22e8fc87e Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Sat, 24 Oct 2026 20:00:26 -0700 Subject: [PATCH 7/9] [None][fix] serve: drop defensive getattr in multi-frontend code Access attributes that are always set directly: LLM._executor and LLM.args, BaseLlmArgs perf-metrics fields, the GenerationExecutor fatal-error and shutdown state, and OpenAIServer's own multi-frontend attributes. _set_fatal_error already keeps only the first error, and the stats collector only runs with enable_iter_perf_stats, so those checks were redundant. No behavior change. Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- tensorrt_llm/commands/serve.py | 10 ++++----- tensorrt_llm/serve/openai_server.py | 21 ++++++++----------- .../test_serve_multi_frontend_helpers.py | 4 +--- 3 files changed, 14 insertions(+), 21 deletions(-) diff --git a/tensorrt_llm/commands/serve.py b/tensorrt_llm/commands/serve.py index ec2ad80d0445..2503e043a0f8 100644 --- a/tensorrt_llm/commands/serve.py +++ b/tensorrt_llm/commands/serve.py @@ -526,8 +526,8 @@ def _bind_launcher_uds(llm) -> Tuple[socket.socket, str]: It lives in the multi-frontend ipc dir, which is private to the user and removed by the executor proxy. """ - ipc_dir = getattr(getattr(llm, "_executor", None), - "_multi_frontend_ipc_dir", None) + # Only the classic IPC proxy has the attribute. + ipc_dir = getattr(llm._executor, "_multi_frontend_ipc_dir", None) if not ipc_dir: raise ValueError( "num_serve_frontends > 1 requires the classic IPC executor " @@ -548,11 +548,9 @@ def _share_prometheus_multiproc_dir(llm, ipc_dir: str) -> None: Not via set_prometheus_multiproc_dir(): it keeps only its latest TemporaryDirectory alive, so OpenAIServer's later call would delete this one. """ - args = getattr(llm, "args", None) - if args is None or os.environ.get("PROMETHEUS_MULTIPROC_DIR"): + if os.environ.get("PROMETHEUS_MULTIPROC_DIR"): return - if (getattr(args, "return_perf_metrics", False) - or getattr(args, "perf_metrics_output_dir", None)): + if llm.args.return_perf_metrics or llm.args.perf_metrics_output_dir: path = tempfile.mkdtemp(prefix="prometheus_", dir=ipc_dir) os.environ["PROMETHEUS_MULTIPROC_DIR"] = path logger.info( diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 3b14d999b839..f9fd027d4f88 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -1805,12 +1805,11 @@ def _on_launcher_lost(self, error: BaseException) -> None: Mirrors the fatal-error branch of ``health``. """ - executor = getattr(self.generator, '_executor', None) - if executor is None: + executor = self.generator._executor + if executor is None: # the LLM already shut down return - if getattr(executor, '_fatal_error', None) is None: - executor._set_fatal_error(error) - if not getattr(executor, 'doing_shutdown', True): + executor._set_fatal_error(error) # keeps only the first error + if not executor.doing_shutdown: _record_generator_termination(self.generator) signal.raise_signal(signal.SIGINT) @@ -1989,10 +1988,9 @@ async def set_steady_clock_offset( f"The steady clock offset between local and disagg server: {offset} second" ) # Multi-frontend launcher: every frontend stamps perf metrics with it. - shared = getattr(self, "_shared_clock_offset", None) - if shared is not None: + if self._shared_clock_offset is not None: try: - shared.publish(offset) + self._shared_clock_offset.publish(offset) except OSError as e: logger.error( f"Failed to share the steady clock offset with the " @@ -2045,8 +2043,8 @@ async def _extract_metrics(self, res: RequestOutput, raw_request: Request): res.metrics_dict) # Note: Iteration stats are collected by the background _iteration_stats_collector_loop task # Wake up the stats collector to drain iteration stats - if self._iteration_stats_collector_task is not None and getattr( - self.generator.args, "enable_iter_perf_stats", True): + # The collector only runs with enable_iter_perf_stats. + if self._iteration_stats_collector_task is not None: self._iteration_stats_wakeup_event.set() async def _create_chat_response( @@ -2103,8 +2101,7 @@ async def _iteration_stats_collector_loop(self): try: await asyncio.wait_for( self._iteration_stats_wakeup_event.wait(), - timeout=getattr(self, "_iteration_stats_poll_interval", - None)) + timeout=self._iteration_stats_poll_interval) except asyncio.TimeoutError: woken_by_timer = True diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py index b32cfee8d82c..6145fd6fd445 100644 --- a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py +++ b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py @@ -218,9 +218,7 @@ async def run() -> None: @pytest.mark.parametrize("doing_shutdown", [False, True]) def test_on_launcher_lost_fails_fast_then_shuts_down(monkeypatch, doing_shutdown: bool) -> None: fatal_errors = [] - executor = SimpleNamespace( - doing_shutdown=doing_shutdown, _fatal_error=None, _set_fatal_error=fatal_errors.append - ) + executor = SimpleNamespace(doing_shutdown=doing_shutdown, _set_fatal_error=fatal_errors.append) server = object.__new__(OpenAIServer) server.generator = SimpleNamespace(_executor=executor) signals, terminations = [], [] From 23f3ceb5cf07f6d13cc1b1ba42bdd7929ae1f82b Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:26:09 -0700 Subject: [PATCH 8/9] [None][fix] serve: log an attached frontend that exits while serving The launcher only watched its attached frontends during startup. If one died later, the kernel dropped its socket from the shared port and the server kept serving on the others, with nothing reporting the lost frontend or the requests it was handling. A daemon thread now polls the children every second and logs an error with the frontend id, pid, exit code or signal and how many frontends remain; a clean exit (e.g. Ctrl-C reaching the process group) is logged at info. The launcher stops the watch before it terminates the group. Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- tensorrt_llm/commands/serve.py | 8 ++++- tensorrt_llm/serve/_frontend_processes.py | 32 +++++++++++++++++++ .../test_serve_multi_frontend_helpers.py | 28 +++++++++++++++- 3 files changed, 66 insertions(+), 2 deletions(-) diff --git a/tensorrt_llm/commands/serve.py b/tensorrt_llm/commands/serve.py index 2503e043a0f8..2aa36336f78c 100644 --- a/tensorrt_llm/commands/serve.py +++ b/tensorrt_llm/commands/serve.py @@ -72,7 +72,8 @@ from tensorrt_llm.serve import OpenAIDisaggServer, OpenAIServer from tensorrt_llm.serve._frontend_processes import ( _init_multi_frontend_mode, _signal_frontend_ready, - _spawn_attached_frontends, _terminate_attached_frontends) + _spawn_attached_frontends, _terminate_attached_frontends, + _watch_attached_frontends) from tensorrt_llm.serve.multi_frontend import (LAUNCHER_UDS_ENV, LAUNCHER_UDS_NAME, MultiFrontendServing) @@ -675,6 +676,7 @@ def launch_server( # server construction, middleware registration, and runtime, or a # failure in between leaks the child processes. frontend_children = [] + stop_watching_frontends = None sockets = [s] launcher_uds_sock = None launcher_uds_path = None @@ -689,6 +691,8 @@ def launch_server( multi_frontend.num_frontends, extra_env={LAUNCHER_UDS_ENV: launcher_uds_path}, report_failure=_report_observed_child_failure) + stop_watching_frontends = _watch_attached_frontends( + frontend_children) multi_frontend_serving = MultiFrontendServing( launcher_uds=launcher_uds_path, is_launcher=True) sockets.append(launcher_uds_sock) @@ -719,6 +723,8 @@ def launch_server( _signal_frontend_ready(multi_frontend) uvloop.run(server(host, port, sockets=sockets)) finally: + if stop_watching_frontends is not None: + stop_watching_frontends.set() if frontend_children: _terminate_attached_frontends(frontend_children) if launcher_uds_sock is not None: diff --git a/tensorrt_llm/serve/_frontend_processes.py b/tensorrt_llm/serve/_frontend_processes.py index 9ea351799f87..799e4e8eeab9 100644 --- a/tensorrt_llm/serve/_frontend_processes.py +++ b/tensorrt_llm/serve/_frontend_processes.py @@ -196,6 +196,38 @@ def _signal_frontend_ready(multi_frontend: MultiFrontendMode) -> None: os.close(fd) +def _watch_attached_frontends(children: list, interval_s: float = 1.0) -> threading.Event: + """Log an attached frontend that exits while the launcher serves. + + The kernel drops a dead frontend's socket from the shared port, so the + others keep serving and only its in-flight requests are lost; nothing else + reports it. Set the returned event before terminating the group, so the + expected exits stay quiet. + """ + stopping = threading.Event() + + def watch() -> None: + alive = dict(enumerate(children, start=1)) + while alive and not stopping.wait(interval_s): + for frontend_id, child in list(alive.items()): + code = child.poll() + if code is None or stopping.is_set(): + continue + del alive[frontend_id] + if code == 0: # e.g. Ctrl-C reaching the whole process group + logger.info(f"Attached serving frontend {frontend_id} (pid {child.pid}) exited") + continue + how = f"exit code {code}" if code > 0 else f"signal {-code}" + logger.error( + f"Attached serving frontend {frontend_id} (pid {child.pid}) exited with " + f"{how}; {len(alive) + 1} of {len(children) + 1} frontends keep serving, " + "and the requests it was handling are lost." + ) + + threading.Thread(target=watch, name="attached-frontend-watch", daemon=True).start() + return stopping + + def _terminate_attached_frontends(children: list) -> None: """Stop and reap the group with one shared grace deadline.""" for child in children: diff --git a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py index 6145fd6fd445..0a6f3b9055bb 100644 --- a/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py +++ b/tests/unittest/llmapi/test_serve_multi_frontend_helpers.py @@ -17,6 +17,8 @@ import asyncio import os import signal +import subprocess +import sys import threading import time from contextlib import asynccontextmanager @@ -32,7 +34,8 @@ from tensorrt_llm._utils import AdjustedSteadyClock from tensorrt_llm.llmapi.disagg_utils import ServerRole -from tensorrt_llm.serve import openai_server +from tensorrt_llm.serve import _frontend_processes, openai_server +from tensorrt_llm.serve._frontend_processes import _watch_attached_frontends from tensorrt_llm.serve.multi_frontend import ( FORWARDED_ROUTES, AttachedFrontendWatchdog, @@ -215,6 +218,29 @@ async def run() -> None: assert len(deaths) == 1 and "exited" in str(deaths[0]) +def test_launcher_logs_an_attached_frontend_that_dies(monkeypatch) -> None: + errors = [] + monkeypatch.setattr(_frontend_processes.logger, "error", errors.append) + sleeper = [sys.executable, "-c", "import time; time.sleep(60)"] + children = [subprocess.Popen(sleeper) for _ in range(2)] + try: + stopping = _watch_attached_frontends(children, interval_s=0.01) + children[0].kill() + deadline = time.monotonic() + 10 + while not errors and time.monotonic() < deadline: + time.sleep(0.01) + # Exits once the launcher terminates the group are expected. + stopping.set() + children[1].kill() + time.sleep(0.1) + finally: + for child in children: + child.kill() + child.wait() + assert len(errors) == 1 + assert "frontend 1 " in errors[0] and "signal 9" in errors[0] and "2 of 3" in errors[0] + + @pytest.mark.parametrize("doing_shutdown", [False, True]) def test_on_launcher_lost_fails_fast_then_shuts_down(monkeypatch, doing_shutdown: bool) -> None: fatal_errors = [] From 50a3d6d9e5125c0eb96b8f466b3ceb084d4ba992 Mon Sep 17 00:00:00 2001 From: Lance Liao <108499334+lancelly@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:30:30 -0700 Subject: [PATCH 9/9] [None][fix] serve: leave the launcher socket file to the ipc dir cleanup The launcher socket file lives in the multi-frontend ipc dir, which the executor proxy removes as a whole when it shuts down, so drop its separate unlink. The socket itself is still closed when the server exits. No behavior change. Signed-off-by: Lance Liao <108499334+lancelly@users.noreply.github.com> --- tensorrt_llm/commands/serve.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/tensorrt_llm/commands/serve.py b/tensorrt_llm/commands/serve.py index 2aa36336f78c..79f7ee8e0465 100644 --- a/tensorrt_llm/commands/serve.py +++ b/tensorrt_llm/commands/serve.py @@ -525,7 +525,7 @@ def _bind_launcher_uds(llm) -> Tuple[socket.socket, str]: """Bind the Unix socket attached frontends forward launcher-owned routes to. It lives in the multi-frontend ipc dir, which is private to the user and - removed by the executor proxy. + removed, socket file included, when the executor proxy shuts down. """ # Only the classic IPC proxy has the attribute. ipc_dir = getattr(llm._executor, "_multi_frontend_ipc_dir", None) @@ -679,7 +679,6 @@ def launch_server( stop_watching_frontends = None sockets = [s] launcher_uds_sock = None - launcher_uds_path = None try: if multi_frontend.is_launcher: launcher_uds_sock, launcher_uds_path = _bind_launcher_uds( @@ -728,11 +727,8 @@ def launch_server( if frontend_children: _terminate_attached_frontends(frontend_children) if launcher_uds_sock is not None: + # Its file goes with the ipc dir (see _bind_launcher_uds). launcher_uds_sock.close() - try: - os.unlink(launcher_uds_path) - except OSError: - pass def launch_mm_encoder_server(