Skip to content

Commit 7fbee6b

Browse files
authored
Merge pull request #92 from QuixiAI/fix/sigterm-during-load
Honor SIGTERM/SIGINT during engine startup and model load
2 parents 7860975 + 8534232 commit 7fbee6b

7 files changed

Lines changed: 310 additions & 15 deletions

File tree

Lines changed: 204 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,204 @@
1+
# SPDX-License-Identifier: Apache-2.0
2+
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3+
"""A stop request (SIGTERM/SIGINT) during engine startup must end the server.
4+
5+
Regression tests for a server that ignored SIGTERM while the model was
6+
loading: the API server's startup handler raised KeyboardInterrupt, the
7+
engine-handshake socket context swallowed it, and the API server went on to
8+
wait up to VLLM_ENGINE_READY_TIMEOUT_S for an engine that had been stopped.
9+
Multiproc workers whose parent died mid-load kept loading as orphans.
10+
11+
None of these tests need a GPU.
12+
"""
13+
14+
import multiprocessing
15+
import signal
16+
import subprocess
17+
import sys
18+
import textwrap
19+
import time
20+
21+
import pytest
22+
import zmq
23+
24+
from vllm.utils.network_utils import zmq_socket_ctx
25+
from vllm.v1.engine.utils import poll_engine_socket_or_exit
26+
27+
pytestmark = pytest.mark.cpu_test
28+
29+
# Generous next to the ~1 s these take when they work; far below the 60 s
30+
# that the stand-in "model load" or READY wait would take when they don't.
31+
EXIT_DEADLINE_S = 15.0
32+
33+
34+
def _ipc_path(tmp_path, name: str) -> str:
35+
return f"ipc://{tmp_path}/{name}"
36+
37+
38+
def _start(script: str, *args: str) -> subprocess.Popen:
39+
proc = subprocess.Popen(
40+
[sys.executable, "-c", textwrap.dedent(script), *args],
41+
stdout=subprocess.PIPE,
42+
stderr=subprocess.STDOUT,
43+
text=True,
44+
)
45+
assert proc.stdout is not None
46+
for line in proc.stdout:
47+
if line.strip() == "STARTED":
48+
return proc
49+
proc.wait()
50+
pytest.fail(f"child exited before startup (rc={proc.returncode})")
51+
52+
53+
def _time_to_exit(proc: subprocess.Popen) -> float:
54+
start = time.monotonic()
55+
try:
56+
proc.wait(timeout=EXIT_DEADLINE_S)
57+
except subprocess.TimeoutExpired:
58+
proc.kill()
59+
proc.wait()
60+
pytest.fail(f"process still running {EXIT_DEADLINE_S}s after the signal")
61+
return time.monotonic() - start
62+
63+
64+
def test_zmq_socket_ctx_propagates_keyboard_interrupt(tmp_path):
65+
with (
66+
pytest.raises(KeyboardInterrupt),
67+
zmq_socket_ctx(_ipc_path(tmp_path, "ki"), zmq.ROUTER, bind=True),
68+
):
69+
raise KeyboardInterrupt("terminated")
70+
71+
72+
# Mirrors the API server: run_server() turns SIGTERM/SIGINT into
73+
# KeyboardInterrupt, launch_core_engines() waits for the engine handshake
74+
# inside zmq_socket_ctx(), and MPClient then waits for the engine's READY.
75+
_API_SERVER_STARTUP = """
76+
import signal, sys, time
77+
import zmq
78+
from vllm.utils.network_utils import zmq_socket_ctx
79+
80+
def _interrupt_init(*_):
81+
raise KeyboardInterrupt("terminated")
82+
83+
signal.signal(signal.SIGTERM, _interrupt_init)
84+
signal.signal(signal.SIGINT, _interrupt_init)
85+
86+
with zmq_socket_ctx(sys.argv[1], zmq.ROUTER, bind=True) as handshake:
87+
print("STARTED", flush=True)
88+
handshake.poll(timeout=60_000) # engine is loading the model
89+
time.sleep(60) # the READY wait that follows a completed handshake
90+
print("STARTUP CONTINUED AFTER STOP REQUEST", flush=True)
91+
"""
92+
93+
94+
@pytest.mark.parametrize("sig", [signal.SIGTERM, signal.SIGINT])
95+
def test_stop_signal_during_engine_handshake_exits(tmp_path, sig):
96+
proc = _start(_API_SERVER_STARTUP, _ipc_path(tmp_path, "handshake"))
97+
time.sleep(0.5)
98+
proc.send_signal(sig)
99+
_time_to_exit(proc)
100+
assert proc.returncode != 0
101+
assert proc.stdout is not None
102+
assert "STARTUP CONTINUED" not in proc.stdout.read()
103+
104+
105+
def _mp_context():
106+
return multiprocessing.get_context("spawn")
107+
108+
109+
def test_ready_wait_fails_fast_when_engine_exits(tmp_path):
110+
ctx = zmq.Context()
111+
try:
112+
sock = ctx.socket(zmq.PULL)
113+
sock.bind(_ipc_path(tmp_path, "ready"))
114+
engine = _mp_context().Process(target=time.sleep, args=(0.2,))
115+
engine.start()
116+
start = time.monotonic()
117+
with pytest.raises(RuntimeError, match="Engine core initialization failed"):
118+
poll_engine_socket_or_exit(sock, [engine], timeout_ms=60_000)
119+
assert time.monotonic() - start < EXIT_DEADLINE_S
120+
engine.join()
121+
finally:
122+
ctx.destroy(linger=0)
123+
124+
125+
def test_ready_wait_returns_message_or_timeout(tmp_path):
126+
ctx = zmq.Context()
127+
try:
128+
path = _ipc_path(tmp_path, "ready-ok")
129+
sock = ctx.socket(zmq.PULL)
130+
sock.bind(path)
131+
engine = _mp_context().Process(target=time.sleep, args=(30,))
132+
engine.start()
133+
try:
134+
assert not poll_engine_socket_or_exit(sock, [engine], timeout_ms=50)
135+
push = ctx.socket(zmq.PUSH)
136+
push.connect(path)
137+
push.send(b"READY")
138+
assert poll_engine_socket_or_exit(sock, [engine], timeout_ms=5_000)
139+
assert sock.recv() == b"READY"
140+
finally:
141+
engine.kill()
142+
engine.join()
143+
finally:
144+
ctx.destroy(linger=0)
145+
146+
147+
# Mirrors worker_main(): the worker's SIGTERM handler raises SystemExit, and
148+
# the death pipe is the only way it learns that its parent (the engine core)
149+
# is gone. "ignore" stands in for a main thread stuck in a native call.
150+
_WORKER_STARTUP = """
151+
import signal, sys, threading, time
152+
from multiprocessing import Pipe
153+
import vllm.v1.executor.multiproc_executor as mpe
154+
155+
if sys.argv[1] == "ignore":
156+
signal.signal(signal.SIGTERM, signal.SIG_IGN)
157+
mpe._ORPHANED_WORKER_EXIT_GRACE_S = 0.5
158+
else:
159+
def _handler(*_):
160+
raise SystemExit(3)
161+
signal.signal(signal.SIGTERM, _handler)
162+
163+
death_reader, death_writer = Pipe(duplex=False)
164+
mpe.WorkerProc.monitor_death_pipe(
165+
death_reader, threading.Event(), threading.Event(), []
166+
)
167+
print("STARTED", flush=True)
168+
death_writer.close() # the parent exits while this worker loads weights
169+
time.sleep(60)
170+
print("WORKER OUTLIVED ITS PARENT", flush=True)
171+
"""
172+
173+
174+
@pytest.mark.parametrize("main_thread", ["interruptible", "ignore"])
175+
def test_worker_exits_when_parent_dies_during_startup(main_thread):
176+
proc = _start(_WORKER_STARTUP, main_thread)
177+
_time_to_exit(proc)
178+
assert proc.returncode == (3 if main_thread == "interruptible" else 1)
179+
assert proc.stdout is not None
180+
assert "OUTLIVED" not in proc.stdout.read()
181+
182+
183+
def test_worker_death_pipe_after_startup_only_stops_queues():
184+
"""Once startup is done, parent exit keeps the existing behaviour: the
185+
worker's queues are shut down and the busy loop exits on its own."""
186+
import threading
187+
188+
from vllm.v1.executor.multiproc_executor import WorkerProc
189+
190+
class _Queue:
191+
def __init__(self):
192+
self.stopped = threading.Event()
193+
194+
def shutdown(self):
195+
self.stopped.set()
196+
197+
reader, writer = multiprocessing.Pipe(duplex=False)
198+
shutdown_requested, startup_done = threading.Event(), threading.Event()
199+
startup_done.set()
200+
queue = _Queue()
201+
WorkerProc.monitor_death_pipe(reader, shutdown_requested, startup_done, [queue])
202+
writer.close()
203+
assert queue.stopped.wait(5)
204+
assert shutdown_requested.is_set()

‎vllm/entrypoints/cli/launch.py‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -125,12 +125,14 @@ def cmd_init() -> list[CLISubcommand]:
125125
async def run_launch_fastapi(args: argparse.Namespace) -> None:
126126
"""Run the online serving layer with FastAPI (no GPU inference)."""
127127

128-
# Interrupt initialization if SIGTERM arrives before uvicorn installs
129-
# its own signal handlers. Once uvicorn is running it replaces this.
128+
# Interrupt initialization if SIGTERM or SIGINT arrives before uvicorn
129+
# installs its own signal handlers. Once uvicorn is running it replaces
130+
# these; until then a blocked event loop would defer asyncio's SIGINT.
130131
def _interrupt_init(*_) -> None:
131132
raise KeyboardInterrupt("terminated")
132133

133134
signal.signal(signal.SIGTERM, _interrupt_init)
135+
signal.signal(signal.SIGINT, _interrupt_init)
134136

135137
# 1. Socket binding
136138
listen_address, sock = setup_server(args, reuse_port=False)

‎vllm/entrypoints/openai/api_server.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -806,12 +806,16 @@ async def run_server(args, **uvicorn_kwargs) -> None:
806806
bootstamp("api: run_server entry (imports done)")
807807
decorate_logs("APIServer", skip_if_decorated=True)
808808

809-
# Interrupt initialization if SIGTERM arrives before uvicorn installs its
810-
# own signal handlers. Once uvicorn is running it replaces this.
809+
# Interrupt initialization if SIGTERM or SIGINT arrives before the HTTP
810+
# server installs its own signal handlers (serve_http replaces these).
811+
# Engine startup blocks the event loop, so asyncio's default SIGINT
812+
# handler, which only cancels the main task, cannot take effect until the
813+
# model has finished loading.
811814
def _interrupt_init(*_) -> None:
812815
raise KeyboardInterrupt("terminated")
813816

814817
signal.signal(signal.SIGTERM, _interrupt_init)
818+
signal.signal(signal.SIGINT, _interrupt_init)
815819

816820
listen_address, sock = setup_server(args, reuse_port=False)
817821
await run_server_worker(listen_address, sock, args, **uvicorn_kwargs)

‎vllm/utils/network_utils.py‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -325,7 +325,13 @@ def zmq_socket_ctx(
325325
identity: bytes | None = None,
326326
router_handover: bool = False,
327327
) -> Iterator[zmq.Socket]:
328-
"""Context manager for a ZMQ socket"""
328+
"""Context manager for a ZMQ socket.
329+
330+
Exceptions raised in the body, including KeyboardInterrupt, propagate.
331+
The API server turns SIGTERM/SIGINT during startup into KeyboardInterrupt;
332+
swallowing it here made the caller carry on as if the engine handshake
333+
had completed, so a stop request during model load was lost.
334+
"""
329335

330336
ctx = zmq.Context() # type: ignore[attr-defined]
331337
try:
@@ -337,8 +343,5 @@ def zmq_socket_ctx(
337343
identity=identity,
338344
router_handover=router_handover,
339345
)
340-
except KeyboardInterrupt:
341-
logger.debug("Got Keyboard Interrupt.")
342-
343346
finally:
344347
ctx.destroy(linger=linger)

‎vllm/v1/engine/core_client.py‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
from concurrent.futures import Future
1313
from dataclasses import dataclass
1414
from multiprocessing.connection import Connection
15+
from multiprocessing.process import BaseProcess
1516
from multiprocessing.queues import Queue
1617
from threading import Thread
1718
from typing import Any, TypeAlias, TypeVar
@@ -57,6 +58,7 @@
5758
CoreEngineProcManager,
5859
get_engine_zmq_addresses,
5960
launch_core_engines,
61+
poll_engine_socket_or_exit,
6062
)
6163
from vllm.v1.executor import Executor
6264
from vllm.v1.fault_tolerance.engine_core_sentinel import FT_UTILITY_METHOD
@@ -631,9 +633,19 @@ def __init__(
631633
# Wait for ready messages from each engine on the input socket.
632634
identities = set(self.core_engines)
633635
sync_input_socket = zmq.Socket.shadow(self.input_socket)
636+
# Also watch the engine (and coordinator) processes this client
637+
# launched, so an engine that dies before sending READY fails
638+
# startup now instead of after VLLM_ENGINE_READY_TIMEOUT_S.
639+
watched_procs: list[BaseProcess] = []
640+
if isinstance(self.resources.engine_manager, CoreEngineProcManager):
641+
watched_procs.extend(self.resources.engine_manager.processes)
642+
if self.resources.coordinator is not None:
643+
watched_procs.append(self.resources.coordinator.proc)
634644
while identities:
635-
if not sync_input_socket.poll(
636-
timeout=VLLM_ENGINE_READY_TIMEOUT_S * 1000 # convert to ms
645+
if not poll_engine_socket_or_exit(
646+
sync_input_socket,
647+
watched_procs,
648+
timeout_ms=VLLM_ENGINE_READY_TIMEOUT_S * 1000,
637649
):
638650
raise TimeoutError(
639651
f"Timed out waiting for engine core processes to "

‎vllm/v1/engine/utils.py‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1219,6 +1219,33 @@ def launch_core_engines(
12191219
)
12201220

12211221

1222+
def poll_engine_socket_or_exit(
1223+
socket: zmq.Socket,
1224+
procs: list[BaseProcess],
1225+
timeout_ms: int,
1226+
) -> bool:
1227+
"""Wait for ``socket`` to become readable while watching ``procs``.
1228+
1229+
Returns True when a message is ready and False on timeout. Raises
1230+
RuntimeError as soon as any watched process exits first, so a caller
1231+
waiting on an engine that died -- crashed, or was terminated by a stop
1232+
request during model load -- fails now instead of after its timeout.
1233+
"""
1234+
poller = zmq.Poller()
1235+
poller.register(socket, zmq.POLLIN)
1236+
for proc in procs:
1237+
poller.register(proc.sentinel, zmq.POLLIN)
1238+
events = dict(poller.poll(timeout=timeout_ms))
1239+
finished = {proc.name: proc.exitcode for proc in procs if proc.sentinel in events}
1240+
if finished:
1241+
raise RuntimeError(
1242+
"Engine core initialization failed. "
1243+
"See root cause above. "
1244+
f"Failed core proc(s): {finished}"
1245+
)
1246+
return socket in events
1247+
1248+
12221249
def wait_for_engine_startup(
12231250
handshake_socket: zmq.Socket,
12241251
addresses: EngineZmqAddresses,

0 commit comments

Comments
 (0)