[Engine] Workers cannot outlive a killed EngineCore
Workers set PR_SET_PDEATHSIG(SIGKILL) against their EngineCore parent with a getppid race check. Start methods other than fork/spawn are refused before any worker is created. Subprocess tests cover fork, spawn, graceful SIGTERM and the refusal. Co-Authored-By: Hinata <hinata@idihore.id> Signed-off-by: MOVZX <movzx@yahoo.com>
This commit is contained in:
@@ -5,6 +5,7 @@ import os
|
||||
import pickle
|
||||
import queue
|
||||
import signal
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
@@ -75,6 +76,44 @@ from vllm.v1.worker.worker_base import WorkerWrapperBase
|
||||
logger = init_logger(__name__)
|
||||
|
||||
|
||||
_WORKER_PARENT_DEATH_SIGNAL = signal.SIGKILL
|
||||
_PR_SET_PDEATHSIG = 1
|
||||
|
||||
|
||||
def _install_worker_parent_death_signal(expected_parent_pid: int) -> None:
|
||||
"""Have Linux kill this worker if its EngineCore parent exits."""
|
||||
if not sys.platform.startswith("linux"):
|
||||
return
|
||||
|
||||
import ctypes
|
||||
|
||||
libc = ctypes.CDLL(None, use_errno=True)
|
||||
prctl = libc.prctl
|
||||
prctl.argtypes = [
|
||||
ctypes.c_int,
|
||||
ctypes.c_ulong,
|
||||
ctypes.c_ulong,
|
||||
ctypes.c_ulong,
|
||||
ctypes.c_ulong,
|
||||
]
|
||||
prctl.restype = ctypes.c_int
|
||||
if prctl(
|
||||
_PR_SET_PDEATHSIG,
|
||||
int(_WORKER_PARENT_DEATH_SIGNAL),
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
) != 0:
|
||||
error_number = ctypes.get_errno()
|
||||
raise OSError(error_number, os.strerror(error_number))
|
||||
|
||||
# The parent can exit between spawning this process and the prctl call.
|
||||
# Once prctl is installed, a later exit is covered by the kernel.
|
||||
if os.getppid() != expected_parent_pid:
|
||||
os.kill(os.getpid(), _WORKER_PARENT_DEATH_SIGNAL)
|
||||
os._exit(1)
|
||||
|
||||
|
||||
class FutureWrapper(Future):
|
||||
def __init__(
|
||||
self,
|
||||
@@ -715,6 +754,15 @@ class WorkerProc:
|
||||
inherited_fds: list[int] | None = None,
|
||||
) -> UnreadyWorkerProcHandle:
|
||||
context = get_mp_context()
|
||||
# Workers arm PR_SET_PDEATHSIG against this process and verify it via
|
||||
# getppid(). Under forkserver the OS parent is the forkserver, so a
|
||||
# healthy worker would kill itself; refuse before creating anything.
|
||||
start_method = context.get_start_method()
|
||||
if start_method not in ("fork", "spawn"):
|
||||
raise RuntimeError(
|
||||
"vLLM worker processes require the 'fork' or 'spawn' "
|
||||
f"multiprocessing start method (got {start_method!r})"
|
||||
)
|
||||
# Ready pipe to communicate readiness from child to parent
|
||||
ready_reader, ready_writer = context.Pipe(duplex=False)
|
||||
# Death pipe to let child detect parent process exit
|
||||
@@ -723,6 +771,7 @@ class WorkerProc:
|
||||
inherited_fds = inherited_fds.copy()
|
||||
inherited_fds.extend((ready_reader.fileno(), death_writer.fileno()))
|
||||
process_kwargs = {
|
||||
"expected_parent_pid": os.getpid(),
|
||||
"vllm_config": vllm_config,
|
||||
"local_rank": local_rank,
|
||||
"rank": rank,
|
||||
@@ -858,6 +907,9 @@ class WorkerProc:
|
||||
if logging_config := getattr(kwargs["vllm_config"], "logging_config", None):
|
||||
configure_logging(logging_config)
|
||||
|
||||
expected_parent_pid = kwargs.pop("expected_parent_pid")
|
||||
_install_worker_parent_death_signal(expected_parent_pid)
|
||||
|
||||
# Signal handler used for graceful termination.
|
||||
# SystemExit exception is only raised once to allow this and worker
|
||||
# processes to terminate without error
|
||||
|
||||
Reference in New Issue
Block a user