[Frontend] Watch frontend processes during engine startup (#43417)
Signed-off-by: Bugen Zhao <i@bugenzhao.com> Signed-off-by: Nick Hill <nickhill123@gmail.com> Co-authored-by: OpenAI Codex <codex@openai.com> Co-authored-by: Nick Hill <nickhill123@gmail.com>
Bugen Zhao committed
Aug 7, 2026 at 02:17 UTC
7b9f2dad8920f115c1caea36e096e43c04c3da68
5 files changed
+152
-42
tests/v1/engine/test_core_engine_actor_manager.py
+2
-6
@@ -308,12 +308,8 @@ def test_ray_dp_addresses_resolved_before_actor_creation(
308
executor_class=_DummyExecutor,
309
log_stats=False,
310
addresses=addresses,
311
- ) as (
312
- engine_manager,
313
- _coordinator,
314
- _addresses_out,
315
- _tensor_queue,
316
- ):
311
+ ) as engine_launch:
312
+ engine_manager = engine_launch.engine_manager
313
assert isinstance(engine_manager, CoreEngineActorManager)
314
315
# API-server children bind to the pre-allocated ports.
tests/v1/engine/test_startup_watch_processes.py
new
+68
@@ -0,0 +1,68 @@
1
+# SPDX-License-Identifier: Apache-2.0
2
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+
4
+from multiprocessing import connection
5
+from types import SimpleNamespace
6
+
7
+import pytest
8
+import zmq
9
+
10
+from vllm.v1.engine.utils import (
11
+ CoreEngine,
12
+ CoreEngineLaunch,
13
+ EngineZmqAddresses,
14
+ wait_for_engine_startup,
15
+)
16
+
17
+pytestmark = pytest.mark.skip_global_cleanup
18
+
19
+
20
+class _FinishedProcess:
21
+ name = "RustFrontend"
22
+
23
+ def __init__(self, sentinel):
24
+ self.sentinel = sentinel
25
+
26
+ @property
27
+ def exitcode(self):
28
+ return 1
29
+
30
+
31
+def test_wait_for_engine_startup_reports_watched_process_exit():
32
+ ctx = zmq.Context()
33
+ handshake_socket = ctx.socket(zmq.ROUTER)
34
+ recv, send = connection.Pipe(duplex=False)
35
+ send.close()
36
+
37
+ parallel_config = SimpleNamespace(
38
+ data_parallel_size_local=1,
39
+ data_parallel_hybrid_lb=False,
40
+ data_parallel_external_lb=False,
41
+ )
42
+
43
+ try:
44
+ launch = CoreEngineLaunch(
45
+ engine_manager=None,
46
+ coordinator=None,
47
+ addresses=EngineZmqAddresses(inputs=[], outputs=[]),
48
+ tensor_queue=None,
49
+ )
50
+ launch.watched_frontend_processes = [_FinishedProcess(recv)]
51
+ with pytest.raises(RuntimeError) as exc_info:
52
+ wait_for_engine_startup(
53
+ handshake_socket,
54
+ [CoreEngine()],
55
+ parallel_config, # type: ignore[arg-type]
56
+ coordinated_dp=False,
57
+ cache_config=None, # type: ignore[arg-type]
58
+ launch=launch,
59
+ )
60
+ finally:
61
+ recv.close()
62
+ handshake_socket.close(linger=0)
63
+ ctx.term()
64
+
65
+ assert "Frontend process failed during engine core initialization" in str(
66
+ exc_info.value
67
+ )
68
+ assert "Failed frontend proc(s): {'RustFrontend': 1}" in str(exc_info.value)
vllm/entrypoints/cli/serve.py
+12
-7
@@ -326,12 +326,12 @@ def run_multi_api_server(args: argparse.Namespace):
326
defer_api_server_ports=not (rust_frontend_path or is_ray_dp),
327
)
328
329
- with launch_core_engines(vllm_config, executor_class, log_stats, addresses) as (
330
- local_engine_manager,
331
- coordinator,
332
- addresses,
333
- tensor_queue,
334
- ):
329
+ with launch_core_engines(
330
+ vllm_config, executor_class, log_stats, addresses
331
+ ) as engine_launch:
332
+ local_engine_manager = engine_launch.engine_manager
333
+ coordinator = engine_launch.coordinator
334
+ addresses = engine_launch.addresses
335
stats_update_address = (
336
coordinator.get_stats_publish_address() if coordinator else None
337
)
@@ -364,7 +364,7 @@ def run_multi_api_server(args: argparse.Namespace):
364
input_addresses=addresses.inputs,
365
output_addresses=addresses.outputs,
366
stats_update_address=stats_update_address,
367
- tensor_queue=tensor_queue,
367
+ tensor_queue=engine_launch.tensor_queue,
368
)
369
370
if not is_ray_dp:
@@ -377,6 +377,11 @@ def run_multi_api_server(args: argparse.Namespace):
377
addresses.inputs = actual_inputs
378
addresses.outputs = actual_outputs
379
380
+ # Set frontend processes to watch during engine startup.
381
+ # If any of these processes exit before the engines are up, the engine startup
382
+ # will be aborted with an error.
383
+ engine_launch.watched_frontend_processes = api_server_manager.processes
384
+
385
# Wait for API servers.
386
try:
387
wait_for_completion_or_failure(
vllm/v1/engine/core_client.py
+6
-3
@@ -608,9 +608,12 @@ class MPClient(EngineCoreClient):
608
609
with launch_core_engines(
610
vllm_config, executor_class, log_stats, addresses
611
- ) as (engine_manager, coordinator, addresses, tensor_queue):
612
- self.resources.coordinator = coordinator
613
- self.resources.engine_manager = engine_manager
611
+ ) as engine_launch:
612
+ self.resources.coordinator = engine_launch.coordinator
613
+ self.resources.engine_manager = engine_launch.engine_manager
614
+ coordinator = engine_launch.coordinator
615
+ addresses = engine_launch.addresses
616
+ tensor_queue = engine_launch.tensor_queue
617
618
self.stats_update_address = addresses.frontend_stats_publish_address
619
if coordinator is not None:
vllm/v1/engine/utils.py
+64
-26
@@ -5,10 +5,10 @@ import contextlib
5
import os
6
import threading
7
import weakref
8
-from collections.abc import Callable, Iterator
8
+from collections.abc import Callable, Iterator, Sequence
9
from dataclasses import dataclass
10
from enum import Enum, auto
11
-from multiprocessing import Process, connection
11
+from multiprocessing import connection
12
from multiprocessing.process import BaseProcess
13
from multiprocessing.queues import Queue
14
from typing import TYPE_CHECKING, cast
@@ -32,7 +32,7 @@ from vllm.utils.system_utils import get_mp_context
32
from vllm.v1.engine.coordinator import DPCoordinator
33
from vllm.v1.executor import Executor
34
from vllm.v1.executor.ray_utils import WORKER_SPECIFIC_ENV_VARS
35
-from vllm.v1.utils import get_engine_client_zmq_addr, shutdown
35
+from vllm.v1.utils import _SubprocessWrapper, get_engine_client_zmq_addr, shutdown
36
37
if TYPE_CHECKING:
38
from ray.util.placement_group import PlacementGroup
@@ -1050,20 +1050,29 @@ def get_engine_zmq_addresses(
1050
)
1051
1052
1053
+FrontendProcess = BaseProcess | _SubprocessWrapper
1054
+
1055
+
1056
+@dataclass
1057
+class CoreEngineLaunch:
1058
+ """Resources and startup barrier for launched engine processes."""
1059
+
1060
+ engine_manager: CoreEngineProcManager | CoreEngineActorManager | None
1061
+ coordinator: DPCoordinator | None
1062
+ addresses: EngineZmqAddresses
1063
+ tensor_queue: Queue | None
1064
+ # Frontend processes to watch during engine startup; may be assigned by
1065
+ # the caller before the startup barrier runs on context manager exit.
1066
+ watched_frontend_processes: Sequence[FrontendProcess] = ()
1067
+
1068
+
1069
@contextlib.contextmanager
1070
def launch_core_engines(
1071
vllm_config: VllmConfig,
1072
executor_class: type[Executor],
1073
log_stats: bool,
1074
addresses: EngineZmqAddresses,
1059
-) -> Iterator[
1060
- tuple[
1061
- CoreEngineProcManager | CoreEngineActorManager | None,
1062
- DPCoordinator | None,
1063
- EngineZmqAddresses,
1064
- Queue | None,
1065
- ]
1066
-]:
1075
+) -> Iterator[CoreEngineLaunch]:
1076
"""Launch engine and DP coordinator processes as needed."""
1077
1078
parallel_config = vllm_config.parallel_config
@@ -1119,7 +1128,9 @@ def launch_core_engines(
1128
log_stats=log_stats,
1129
)
1130
1122
- yield engine_actor_manager, coordinator, addresses, tensor_queue
1131
+ yield CoreEngineLaunch(
1132
+ engine_actor_manager, coordinator, addresses, tensor_queue
1133
+ )
1134
return
1135
1136
if offline_mode:
@@ -1188,30 +1199,27 @@ def launch_core_engines(
1199
else:
1200
local_engine_manager = None
1201
1191
- yield local_engine_manager, coordinator, addresses, tensor_queue
1192
-
1193
- # Now wait for engines to start.
1202
+ launch = CoreEngineLaunch(
1203
+ local_engine_manager, coordinator, addresses, tensor_queue
1204
+ )
1205
+ yield launch
1206
wait_for_engine_startup(
1207
handshake_socket,
1196
- addresses,
1208
engines_to_handshake,
1209
parallel_config,
1210
dp_size > 1 and vllm_config.model_config.is_moe,
1211
vllm_config.cache_config,
1201
- local_engine_manager,
1202
- coordinator.proc if coordinator else None,
1212
+ launch,
1213
)
1214
1215
1216
def wait_for_engine_startup(
1217
handshake_socket: zmq.Socket,
1208
- addresses: EngineZmqAddresses,
1218
core_engines: list[CoreEngine],
1219
parallel_config: ParallelConfig,
1220
coordinated_dp: bool,
1221
cache_config: CacheConfig,
1213
- proc_manager: CoreEngineProcManager | None,
1214
- coord_process: Process | None,
1222
+ launch: CoreEngineLaunch,
1223
):
1224
# Wait for engine core process(es) to send ready messages.
1225
local_count = parallel_config.data_parallel_size_local
@@ -1226,11 +1234,21 @@ def wait_for_engine_startup(
1234
and not parallel_config.data_parallel_external_lb
1235
)
1236
1229
- if proc_manager is not None:
1230
- for sentinel in proc_manager.sentinels():
1237
+ # 1. Engine processes
1238
+ if isinstance(launch.engine_manager, CoreEngineProcManager):
1239
+ for sentinel in launch.engine_manager.sentinels():
1240
poller.register(sentinel, zmq.POLLIN)
1241
+ # 2. DP Coordinator process, if present
1242
+ coord_process = launch.coordinator.proc if launch.coordinator else None
1243
if coord_process is not None:
1244
poller.register(coord_process.sentinel, zmq.POLLIN)
1245
+ # 3. Watched frontend processes, if any
1246
+ frontend_process_by_fd: dict[int, FrontendProcess] = {}
1247
+ for proc in launch.watched_frontend_processes:
1248
+ fd = proc.sentinel if isinstance(proc.sentinel, int) else proc.sentinel.fileno()
1249
+ frontend_process_by_fd[fd] = proc
1250
+ poller.register(fd, zmq.POLLIN)
1251
+
1252
while any(conn_pending) or any(start_pending):
1253
events = poller.poll(STARTUP_POLL_PERIOD_MS)
1254
if not events:
@@ -1246,14 +1264,34 @@ def wait_for_engine_startup(
1264
)
1265
continue
1266
if len(events) > 1 or events[0][0] != handshake_socket:
1249
- # One of the local core processes exited.
1250
- finished = proc_manager.finished_procs() if proc_manager else {}
1267
+ # One of the local core, coordinator, or watched frontend processes exited.
1268
+ if isinstance(launch.engine_manager, CoreEngineProcManager):
1269
+ finished = launch.engine_manager.finished_procs()
1270
+ else:
1271
+ finished = {}
1272
if coord_process is not None and coord_process.exitcode is not None:
1273
finished[coord_process.name] = coord_process.exitcode
1274
+ failed_frontend_procs = {
1275
+ proc.name: proc.exitcode
1276
+ for fd, proc in frontend_process_by_fd.items()
1277
+ if proc.exitcode is not None
1278
+ or any(event_fd == fd for event_fd, _ in events)
1279
+ }
1280
+ if failed_frontend_procs and not finished:
1281
+ raise RuntimeError(
1282
+ "Frontend process failed during engine core initialization. "
1283
+ "See root cause above. "
1284
+ f"Failed frontend proc(s): {failed_frontend_procs}"
1285
+ )
1286
raise RuntimeError(
1287
"Engine core initialization failed. "
1288
"See root cause above. "
1289
f"Failed core proc(s): {finished}"
1290
+ + (
1291
+ f", failed frontend proc(s): {failed_frontend_procs}"
1292
+ if failed_frontend_procs
1293
+ else ""
1294
+ )
1295
)
1296
1297
# Receive HELLO and READY messages from the input socket.
@@ -1293,7 +1331,7 @@ def wait_for_engine_startup(
1331
# Send init message with DP config info.
1332
init_message = msgspec.msgpack.encode(
1333
EngineHandshakeMetadata(
1296
- addresses=addresses,
1334
+ addresses=launch.addresses,
1335
parallel_config={
1336
k: getattr(parallel_config, k)
1337
for k in (