From 502dd7bed0f112d8559de498c880cc84800d2441 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Fri, 18 Sep 2026 19:04:36 +0800 Subject: [PATCH] [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 --- tests/edge/local/test_e2e_serve_text.py | 99 ++++++++++++++++++++----- vllm_omni/benchmarks/omniinteract.py | 19 ++++- vllm_omni/engine/stage_phase_lock.py | 45 +++++++++-- 3 files changed, 134 insertions(+), 29 deletions(-) diff --git a/tests/edge/local/test_e2e_serve_text.py b/tests/edge/local/test_e2e_serve_text.py index e99286c79..a84e8a528 100644 --- a/tests/edge/local/test_e2e_serve_text.py +++ b/tests/edge/local/test_e2e_serve_text.py @@ -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 --omni`` as a subprocess, through ``tests.helpers.runtime.OmniServer`` -(the same launcher the online-serving suite uses) -- and talks to it over HTTP +serve --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: diff --git a/vllm_omni/benchmarks/omniinteract.py b/vllm_omni/benchmarks/omniinteract.py index b821d0235..842a3dc6a 100644 --- a/vllm_omni/benchmarks/omniinteract.py +++ b/vllm_omni/benchmarks/omniinteract.py @@ -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: diff --git a/vllm_omni/engine/stage_phase_lock.py b/vllm_omni/engine/stage_phase_lock.py index 2a723fdd0..44493f649 100644 --- a/vllm_omni/engine/stage_phase_lock.py +++ b/vllm_omni/engine/stage_phase_lock.py @@ -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