[edge-infer W8] Serving e2e launches the server itself; no module-level fcntl on the serve import path
test_e2e_serve_text.py no longer imports tests.helpers.runtime (it pulls the media/diffusion test helpers and therefore diffusers, absent from the Windows venv); a small Popen launcher with a psutil process-tree kill replaces OmniServer for this one test. vllm-omni serve died at import on Windows: entrypoints/cli imports benchmarks.patch -> benchmarks/omniinteract.py, which did 'import fcntl' at module scope. Its output lock is rebuilt on the fork's _filelock_compat primitive (fcntl on POSIX, msvcrt on Windows). engine/stage_phase_lock.py (parallel_stage_init's SH/EX device locks) gets the same treatment: fcntl optional, modes spelled locally, and on Windows both modes map to the exclusive byte lock -- more serialising, never less safe. Measured: the serving e2e's server exited with ModuleNotFoundError: No module named 'fcntl' on both Windows venvs. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
1 parent
d4c097531d
commit
502dd7bed0
3 files changed
+134
-29
No files matched your search
@@ -4,8 +4,10 @@
|
||||
|
||||
``test_e2e_local_text.py`` drives ``LocalTextEngine`` in-process. This module
|
||||
drives the *server* path instead -- ``python -m vllm_omni.entrypoints.cli.main
|
||||
serve <model> --omni`` as a subprocess, through ``tests.helpers.runtime.OmniServer``
|
||||
(the same launcher the online-serving suite uses) -- and talks to it over HTTP
|
||||
serve <model> --omni`` as a subprocess, launched here with a small launcher of
|
||||
its own (``tests.helpers.runtime.OmniServer`` would do, but importing it pulls
|
||||
the media/diffusion test helpers and therefore ``diffusers``, which the AR/text
|
||||
path and a Windows venv without the media deps do not have) -- and talks to it over HTTP
|
||||
with plain ``requests``: health, model listing, non-streaming chat, streaming
|
||||
chat, and a client that disconnects mid-stream after which the server must
|
||||
still answer. Everything runs on one local device; no network access.
|
||||
@@ -18,9 +20,14 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
import psutil
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
@@ -57,24 +64,62 @@ def model_dir() -> str:
|
||||
return _find_model()
|
||||
|
||||
|
||||
def _free_port(host: str = "127.0.0.1") -> int:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.bind((host, 0))
|
||||
return sock.getsockname()[1]
|
||||
|
||||
|
||||
def _kill_tree(proc: subprocess.Popen, timeout: float = 30) -> None:
|
||||
"""Terminate the server and every child (engine core, workers), on any OS."""
|
||||
try:
|
||||
parent = psutil.Process(proc.pid)
|
||||
except psutil.NoSuchProcess:
|
||||
return
|
||||
children = parent.children(recursive=True)
|
||||
for p in [*children, parent]:
|
||||
try:
|
||||
p.terminate()
|
||||
except psutil.NoSuchProcess:
|
||||
pass
|
||||
_, alive = psutil.wait_procs([*children, parent], timeout=timeout)
|
||||
for p in alive:
|
||||
try:
|
||||
p.kill()
|
||||
except psutil.NoSuchProcess:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def server(model_dir: str) -> Iterator[tuple[str, str]]:
|
||||
"""Start the Omni server on the local text plan's settings; yield (base_url, model)."""
|
||||
from tests.helpers.runtime import OmniServer
|
||||
|
||||
env = {
|
||||
"VLLM_NO_USAGE_STATS": "1",
|
||||
"DO_NOT_TRACK": "1",
|
||||
"HF_HUB_OFFLINE": "1",
|
||||
# The same sampler setting LocalTextEngine applies to itself
|
||||
# (vllm_omni/edge/local/engine.py). Without it the serve path picks the
|
||||
# flashinfer top-k/top-p sampler, whose JIT needs nvcc at runtime; a
|
||||
# venv without a CUDA toolkit (the WSL reference venv) dies in
|
||||
# profile_run with "Could not find nvcc". The local plan never wants a
|
||||
# runtime compiler on the device it admits.
|
||||
"VLLM_USE_FLASHINFER_SAMPLER": "0",
|
||||
}
|
||||
serve_args = [
|
||||
env = os.environ.copy()
|
||||
env.update(
|
||||
{
|
||||
"VLLM_NO_USAGE_STATS": "1",
|
||||
"DO_NOT_TRACK": "1",
|
||||
"HF_HUB_OFFLINE": "1",
|
||||
# The same sampler setting LocalTextEngine applies to itself
|
||||
# (vllm_omni/edge/local/engine.py). Without it the serve path picks the
|
||||
# flashinfer top-k/top-p sampler, whose JIT needs nvcc at runtime; a
|
||||
# venv without a CUDA toolkit (the WSL reference venv) dies in
|
||||
# profile_run with "Could not find nvcc". The local plan never wants a
|
||||
# runtime compiler on the device it admits.
|
||||
"VLLM_USE_FLASHINFER_SAMPLER": "0",
|
||||
}
|
||||
)
|
||||
host, port = "127.0.0.1", _free_port()
|
||||
cmd = [
|
||||
sys.executable,
|
||||
"-m",
|
||||
"vllm_omni.entrypoints.cli.main",
|
||||
"serve",
|
||||
model_dir,
|
||||
"--host",
|
||||
host,
|
||||
"--port",
|
||||
str(port),
|
||||
"--omni",
|
||||
"--max-model-len",
|
||||
str(MAX_MODEL_LEN),
|
||||
"--max-num-seqs",
|
||||
@@ -88,9 +133,23 @@ def server(model_dir: str) -> Iterator[tuple[str, str]]:
|
||||
"900",
|
||||
"--disable-log-stats",
|
||||
]
|
||||
srv = OmniServer(model_dir, serve_args, env_dict=env, use_omni=True)
|
||||
with srv:
|
||||
yield f"http://{srv.host}:{srv.port}", model_dir
|
||||
repo_root = Path(__file__).resolve().parents[3]
|
||||
proc = subprocess.Popen(cmd, env=env, cwd=str(repo_root))
|
||||
try:
|
||||
deadline = time.time() + 1200
|
||||
while True:
|
||||
if proc.poll() is not None:
|
||||
pytest.fail(f"server exited with code {proc.returncode} before opening its port")
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.settimeout(1)
|
||||
if sock.connect_ex((host, port)) == 0:
|
||||
break
|
||||
if time.time() > deadline:
|
||||
pytest.fail("server did not open its port within 1200 s")
|
||||
time.sleep(2)
|
||||
yield f"http://{host}:{port}", model_dir
|
||||
finally:
|
||||
_kill_tree(proc)
|
||||
|
||||
|
||||
def _wait_healthy(base_url: str, timeout: float = 300) -> None:
|
||||
|
||||
@@ -9,7 +9,6 @@ import asyncio
|
||||
import binascii
|
||||
import contextlib
|
||||
import copy
|
||||
import fcntl
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
@@ -549,13 +548,27 @@ def _output_dir(root: Path, case: OmniInteractCase) -> Path:
|
||||
|
||||
@contextlib.contextmanager
|
||||
def omniinteract_output_lock(root: Path) -> Iterator[None]:
|
||||
# [edge-infer W8] was a module-level ``import fcntl`` + blocking ``flock``.
|
||||
# ``fcntl`` does not exist on Windows, and this module is imported by
|
||||
# ``vllm_omni.entrypoints.cli`` (through ``benchmarks.patch``), so every
|
||||
# ``vllm-omni serve`` died at import there. The blocking exclusive lock is
|
||||
# rebuilt from the non-blocking primitive in ``_filelock_compat`` (fcntl on
|
||||
# POSIX, msvcrt on Windows) with a short retry; semantics are unchanged.
|
||||
from vllm_omni._filelock_compat import flock_exclusive_nb, funlock
|
||||
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
with (root / ARTIFACT_LOCK_FILE).open("a+b") as lock_file:
|
||||
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
|
||||
fd = lock_file.fileno()
|
||||
while True:
|
||||
try:
|
||||
flock_exclusive_nb(fd)
|
||||
break
|
||||
except BlockingIOError:
|
||||
time.sleep(0.05)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
|
||||
funlock(fd)
|
||||
|
||||
|
||||
def _atomic_replace(path: Path, writer: Callable[[Path], None]) -> None:
|
||||
|
||||
@@ -30,9 +30,13 @@ is importable and unit-testable without a GPU; torch/platform imports are lazy.
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import fcntl
|
||||
import os
|
||||
import time
|
||||
|
||||
try:
|
||||
import fcntl
|
||||
except ImportError: # [edge-infer W8] Windows: no fcntl; see _try_lock below.
|
||||
fcntl = None # type: ignore[assignment]
|
||||
from collections.abc import Iterator
|
||||
|
||||
from vllm.logger import init_logger
|
||||
@@ -43,6 +47,35 @@ from vllm_omni.engine.stage_init_utils import (
|
||||
record_lock_holder_pid,
|
||||
)
|
||||
|
||||
# Lock modes, spelled without fcntl so the module imports on Windows.
|
||||
LOCK_SH = getattr(fcntl, "LOCK_SH", 1)
|
||||
LOCK_EX = getattr(fcntl, "LOCK_EX", 2)
|
||||
|
||||
|
||||
def _try_lock(fd: int, mode: int) -> None:
|
||||
"""Non-blocking lock in *mode*; raises BlockingIOError when held elsewhere.
|
||||
|
||||
[edge-infer W8] On POSIX this is ``fcntl.flock(fd, mode | LOCK_NB)``. On
|
||||
Windows ``msvcrt.locking`` has no shared mode, so both SH and EX become the
|
||||
exclusive byte lock from ``_filelock_compat``: strictly more serialising
|
||||
than POSIX (shared holders now queue behind each other), never less safe.
|
||||
"""
|
||||
if fcntl is not None:
|
||||
fcntl.flock(fd, mode | fcntl.LOCK_NB)
|
||||
return
|
||||
from vllm_omni._filelock_compat import flock_exclusive_nb
|
||||
|
||||
flock_exclusive_nb(fd)
|
||||
|
||||
|
||||
def _unlock(fd: int) -> None:
|
||||
if fcntl is not None:
|
||||
fcntl.flock(fd, fcntl.LOCK_UN)
|
||||
return
|
||||
from vllm_omni._filelock_compat import funlock
|
||||
|
||||
funlock(fd)
|
||||
|
||||
logger = init_logger(__name__)
|
||||
|
||||
# Default upper bound on how long a single phase will wait for its device locks
|
||||
@@ -149,14 +182,14 @@ class DevicePhaseLock:
|
||||
fd, writable = open_device_lock_file(device_init_lock_path(device_id, self._lock_dir))
|
||||
while True:
|
||||
try:
|
||||
fcntl.flock(fd, mode | fcntl.LOCK_NB)
|
||||
_try_lock(fd, mode)
|
||||
break
|
||||
except BlockingIOError:
|
||||
if time.monotonic() > deadline:
|
||||
os.close(fd)
|
||||
raise DeviceLockTimeoutError(
|
||||
f"Timed out after {self._timeout_s:.0f}s acquiring "
|
||||
f"{'EX' if mode == fcntl.LOCK_EX else 'SH'} lock on device {device_id}"
|
||||
f"{'EX' if mode == LOCK_EX else 'SH'} lock on device {device_id}"
|
||||
)
|
||||
time.sleep(0.01)
|
||||
record_lock_holder_pid(fd, writable)
|
||||
@@ -172,7 +205,7 @@ class DevicePhaseLock:
|
||||
if not self._device_ids:
|
||||
yield
|
||||
return
|
||||
fds = self._acquire(fcntl.LOCK_SH)
|
||||
fds = self._acquire(LOCK_SH)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
@@ -184,7 +217,7 @@ class DevicePhaseLock:
|
||||
if not self._device_ids:
|
||||
yield
|
||||
return
|
||||
fds = self._acquire(fcntl.LOCK_EX)
|
||||
fds = self._acquire(LOCK_EX)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
@@ -194,7 +227,7 @@ class DevicePhaseLock:
|
||||
def _release_fds(fds: list[int]) -> None:
|
||||
for fd in fds:
|
||||
try:
|
||||
fcntl.flock(fd, fcntl.LOCK_UN)
|
||||
_unlock(fd)
|
||||
os.close(fd)
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
|
||||
Reference in new issue
Block a user