diff --git a/pyproject.toml b/pyproject.toml index 5e42e1749..922ac04ac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -160,6 +160,12 @@ vllm-omni = "vllm_omni.entrypoints.cli.main:main" # only import vllm, not vllm_omni) will still pick up the omni archs. [project.entry-points."vllm.general_plugins"] vllm_omni_register_models = "vllm_omni.engine.arg_utils:register_omni_models_to_vllm" +# [edge-infer] Host-OS compatibility for native Windows (no-op elsewhere); vLLM runs +# general plugins in the API server, every engine-core child and every worker. +vllm_omni_windows = "vllm_omni.windows:activate" + +[project.entry-points."vllm_omni.general_plugins"] +vllm_omni_windows = "vllm_omni.windows:activate" [tool.setuptools.packages.find] diff --git a/tests/windows/__init__.py b/tests/windows/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/windows/test_windows_compat.py b/tests/windows/test_windows_compat.py new file mode 100644 index 000000000..e32d0f461 --- /dev/null +++ b/tests/windows/test_windows_compat.py @@ -0,0 +1,269 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""``vllm_omni.windows`` on any OS: the shim logic through injected primitives, +the no-op contract off Windows, the report schema, and the marker/hash rules +that decide whether a vLLM function is replaced or left alone.""" + +from __future__ import annotations + +import asyncio +import os +import sys +import tempfile + +import pytest + +from vllm_omni.windows import activate, is_active, is_windows, report +from vllm_omni.windows import aio, shims, vllm_patches +from vllm_omni.windows.paths import shm_dir, shm_path + + +# ---- fcntl shim ------------------------------------------------------------ + + +class FakeLocking: + """Byte-range primitive with msvcrt.locking's contract, held state in memory.""" + + NBLCK, UNLCK = 2, 0 + + def __init__(self) -> None: + self.held: set[int] = set() + self.calls: list[tuple[int, int]] = [] + + def __call__(self, fd: int, mode: int, nbytes: int) -> None: + self.calls.append((fd, mode)) + if mode == self.NBLCK: + if fd in self.held: + raise OSError(36, "Resource deadlock avoided") + self.held.add(fd) + elif mode == self.UNLCK: + self.held.discard(fd) + + +@pytest.fixture +def lock_fd(): + fd, path = tempfile.mkstemp() + yield fd + os.close(fd) + os.unlink(path) + + +def _shim(locking: FakeLocking, sleeps: list[float]): + return shims.FcntlShim(locking, FakeLocking.NBLCK, FakeLocking.UNLCK, sleep=sleeps.append, poll_interval=0.01) + + +def test_flock_exclusive_then_unlock(lock_fd): + locking, sleeps = FakeLocking(), [] + s = _shim(locking, sleeps) + s.flock(lock_fd, shims.LOCK_EX) + assert lock_fd in locking.held + s.flock(lock_fd, shims.LOCK_UN) + assert lock_fd not in locking.held + assert sleeps == [] + + +def test_flock_non_blocking_raises_blockingioerror_when_held(lock_fd): + locking, sleeps = FakeLocking(), [] + locking.held.add(lock_fd) + s = _shim(locking, sleeps) + with pytest.raises(BlockingIOError): + s.flock(lock_fd, shims.LOCK_EX | shims.LOCK_NB) + assert sleeps == [] + + +def test_flock_blocking_waits_until_released(lock_fd): + locking = FakeLocking() + locking.held.add(lock_fd) + sleeps: list[float] = [] + + def sleep(dt: float) -> None: # the other holder lets go after two polls + sleeps.append(dt) + if len(sleeps) == 2: + locking.held.discard(lock_fd) + + s = shims.FcntlShim(locking, FakeLocking.NBLCK, FakeLocking.UNLCK, sleep=sleep, poll_interval=0.01) + s.flock(lock_fd, shims.LOCK_EX) + assert len(sleeps) == 2 and lock_fd in locking.held + + +def test_shared_lock_is_served_as_exclusive_and_counted(lock_fd): + locking, sleeps = FakeLocking(), [] + s = _shim(locking, sleeps) + s.flock(lock_fd, shims.LOCK_SH) + assert s.shared_served_as_exclusive == 1 + assert lock_fd in locking.held + + +def test_lock_offset_leaves_file_position_alone(lock_fd): + os.write(lock_fd, b"12345\n") + os.lseek(lock_fd, 3, os.SEEK_SET) + s = _shim(FakeLocking(), []) + s.flock(lock_fd, shims.LOCK_EX) + assert os.lseek(lock_fd, 0, os.SEEK_CUR) == 3 + + +def test_fcntl_shim_module_shape(): + mod = _shim(FakeLocking(), []).as_module() + assert mod.__name__ == "fcntl" and mod.__vllm_omni_windows_shim__ + assert (mod.LOCK_SH, mod.LOCK_EX, mod.LOCK_NB, mod.LOCK_UN) == (1, 2, 4, 8) + assert callable(mod.flock) and callable(mod.lockf) + + +# ---- uvloop shim ----------------------------------------------------------- + + +def test_uvloop_shim_runs_a_coroutine(): + mod = shims.make_uvloop_shim() + assert mod.__vllm_omni_windows_shim__ + + async def main(): + await asyncio.sleep(0) + return 42 + + assert mod.run(main()) == 42 + loop = mod.new_event_loop() + try: + assert isinstance(loop, asyncio.AbstractEventLoop) + finally: + loop.close() + + +def test_provide_leaves_a_real_module_alone(): + # ``os`` is always real; provide() must not touch it. + assert shims.provide("os", lambda: None) == "real" + assert sys.modules["os"] is os + + +def test_provide_installs_and_then_reports_already(): + name = "_vllm_omni_windows_test_missing_module_" + sys.modules.pop(name, None) + try: + assert shims.provide(name, shims.make_uvloop_shim) == "provided" + assert shims.provide(name, shims.make_uvloop_shim) == "already" + assert sys.modules[name].__vllm_omni_windows_shim__ + finally: + sys.modules.pop(name, None) + + +# ---- signal fallback ------------------------------------------------------- + + +class FakeLoop: + def __init__(self) -> None: + self.scheduled = [] + self.__dict__["_x"] = None + + def call_soon_threadsafe(self, cb, *args): + self.scheduled.append((cb, args)) + + +def test_signal_fallback_registers_and_fires_via_call_soon_threadsafe(monkeypatch): + import signal as _signal + + installed = {} + + def fake_signal(sig, handler): + installed[sig] = handler + + monkeypatch.setattr(_signal, "signal", fake_signal) + + def original_add(self, sig, cb, *args): + raise NotImplementedError + + def original_remove(self, sig): + raise NotImplementedError + + add, remove = aio.make_signal_fallback(original_add, original_remove) + loop = FakeLoop() + hits = [] + add(loop, _signal.SIGINT, hits.append, "x") + assert _signal.SIGINT in installed + installed[_signal.SIGINT](_signal.SIGINT, None) # the OS delivers the signal + assert loop.scheduled == [(hits.append, ("x",))] + assert remove(loop, _signal.SIGINT) is True + assert remove(loop, _signal.SIGINT) is False # nothing left, original says NotImplemented + + +def test_signal_fallback_prefers_the_native_implementation(): + calls = [] + + def original_add(self, sig, cb, *args): + calls.append(sig) + + add, _ = aio.make_signal_fallback(original_add, lambda self, sig: True) + add(FakeLoop(), 2, lambda: None) + assert calls == [2] + + +# ---- vLLM patch rules ------------------------------------------------------ + + +def test_upstream_hashes_are_the_pinned_bodies_or_the_branch_marker(): + """On the installed vLLM each replaceable function is either the known upstream body or already patched.""" + vllm = pytest.importorskip("vllm") + from vllm.utils import network_utils, system_utils + + for name, fn in (("get_open_zmq_ipc_path", network_utils.get_open_zmq_ipc_path), + ("find_loaded_library", system_utils.find_loaded_library)): + marked = vllm_patches._has_mark(fn) + known = vllm_patches._src_hash(fn) in vllm_patches.KNOWN_UPSTREAM[name] + assert marked or known, f"{name} on vllm {vllm.__version__}: neither branch-patched nor a known upstream body" + + +def test_verify_reports_wheel_required_on_an_unpatched_source(monkeypatch): + class Fake: + pass + + import types + + m = types.ModuleType("fake_vllm_site") + + def wait_for_engine_startup(): + return "poller.register(sentinel)" + + m.wait_for_engine_startup = wait_for_engine_startup + monkeypatch.setitem(sys.modules, "fake_vllm_site", m) + import inspect + + monkeypatch.setattr(inspect, "getsource", lambda obj: "poller.register(sentinel)") + assert vllm_patches._verify("fake_vllm_site", "wait_for_engine_startup", "_win_sentinels") == "wheel_required" + assert vllm_patches._verify("fake_vllm_site", "wait_for_engine_startup", "sentinel") == "already" + + +# ---- paths ----------------------------------------------------------------- + + +def test_shm_dir_exists_and_path_joins(): + d = shm_dir() + assert os.path.isdir(d) + assert shm_path("x.lock").startswith(d) + if os.path.isdir("/dev/shm"): + assert d == "/dev/shm" + + +# ---- activation contract --------------------------------------------------- + + +def test_activate_is_idempotent_and_a_noop_off_windows(): + first = activate() + second = activate() + assert first == second + if not is_windows(): + assert first == {"platform": "n/a"} + assert not is_active() + else: + assert is_active() + assert set(first) >= {"fcntl", "uvloop", "event_loop_policy", "signal_handlers", "get_open_zmq_ipc_path"} + + +def test_report_schema(): + rep = report() + for key in ("platform", "active", "activations", "steps", "asyncio", "zmq", "shm_dir", "vllm", "native_modules", "optional_deps"): + assert key in rep, key + assert rep["activations"] >= 1 + assert isinstance(rep["optional_deps"], dict) and "fcntl" in rep["optional_deps"] + if rep["vllm"]: + if is_windows(): + assert rep["vllm"]["wheel_sites"], "on Windows the wheel sites must be reported" + else: + assert rep["vllm"]["wheel_sites"] == {} diff --git a/vllm_omni/distributed/omni_connectors/connectors/cuda_ipc_connector.py b/vllm_omni/distributed/omni_connectors/connectors/cuda_ipc_connector.py index 73b11701e..45b51f423 100644 --- a/vllm_omni/distributed/omni_connectors/connectors/cuda_ipc_connector.py +++ b/vllm_omni/distributed/omni_connectors/connectors/cuda_ipc_connector.py @@ -42,7 +42,10 @@ from vllm_omni.distributed.omni_connectors.connectors.shm_connector import Share logger = init_logger(__name__) MARKER = "__cuda_ipc__" -ACK_DIR = "/dev/shm" +# [edge-infer] /dev/shm where it exists (Linux, WSL); a per-user directory on Windows. +from vllm_omni.windows.paths import shm_dir as _shm_dir + +ACK_DIR = _shm_dir() def _ack_path(put_key: str) -> str: diff --git a/vllm_omni/edge/local/engine.py b/vllm_omni/edge/local/engine.py index 636fb7faf..452993134 100644 --- a/vllm_omni/edge/local/engine.py +++ b/vllm_omni/edge/local/engine.py @@ -114,6 +114,18 @@ def _host_kind() -> str: return "linux" +def _host_compat_report() -> dict[str, Any] | None: + """``vllm_omni.windows.report()`` on Windows, so a record says what made the run possible.""" + from vllm_omni.windows import is_windows, report + + if not is_windows(): + return None + try: + return report() + except Exception as exc: # the audit must never take the engine down + return {"error": f"{type(exc).__name__}: {exc}"} + + def gpu_attribution_for_tree() -> dict[str, Any]: """Try to attribute GPU bytes to this process tree, and say why when it fails. @@ -765,6 +777,9 @@ class LocalTextEngine: "tree_rss_delta_at_load": self.load_memory.get("tree_rss_delta"), **gpu_attribution_for_tree(), }, + # [edge-infer] On native Windows, which compatibility layer carried this + # process (shims, patches, wheel-provided sites). None elsewhere. + "host_compat": _host_compat_report(), } def report_usage(self) -> dict[str, Any]: diff --git a/vllm_omni/edge/step_stats.py b/vllm_omni/edge/step_stats.py index 11f68dbfc..7c7838411 100644 --- a/vllm_omni/edge/step_stats.py +++ b/vllm_omni/edge/step_stats.py @@ -237,7 +237,9 @@ def instrument_sampler(sampler: Any) -> None: def shm_lockfile_age_ms(key: str, now: float | None = None) -> float | None: """Age of the SharedMemoryConnector lock file for ``key`` (written at put time).""" try: - st = os.stat(f"/dev/shm/shm_{key}_lockfile.lock") + from vllm_omni.windows.paths import shm_path + + st = os.stat(shm_path(f"shm_{key}_lockfile.lock")) except OSError: return None return ((now if now is not None else time.time()) - st.st_mtime) * 1000.0 diff --git a/vllm_omni/patch.py b/vllm_omni/patch.py index 045c87b38..9c8da61f2 100644 --- a/vllm_omni/patch.py +++ b/vllm_omni/patch.py @@ -3,6 +3,13 @@ import os import sys from functools import cached_property +# [edge-infer] Host-OS compatibility must precede every vLLM import below: it +# provides `fcntl`/`uvloop` and sets the event-loop policy on native Windows. +# No-op on other platforms. Design: analysis/design_windows_compat_plugin_20260921.md +from vllm_omni.windows import activate as _activate_windows_compat # isort:skip + +_activate_windows_compat() + import torch from aenum import extend_enum from vllm.config import ModelConfig as _OriginalModelConfig diff --git a/vllm_omni/windows/__init__.py b/vllm_omni/windows/__init__.py new file mode 100644 index 000000000..95d9abcf1 --- /dev/null +++ b/vllm_omni/windows/__init__.py @@ -0,0 +1,72 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""``vllm_omni.windows`` -- host-OS compatibility so vLLM and vLLM-Omni run on native Windows. + +Design: ``analysis/design_windows_compat_plugin_20260921.md`` (edge-infer). In +one sentence: the POSIX assumptions that killed Windows runs in W0–W8 are a +handful of classes repeated across vLLM, vLLM-Omni and model code; this +package provides the missing modules (``fcntl``, ``uvloop``), fixes the +process state Windows gets wrong (event loop policy, signal handlers), replaces +whole the two vLLM functions whose entire body is platform logic, and +*reports* the sites that must come from the Windows wheel. + +Three doors, all idempotent, so import order does not matter: + +* ``vllm_omni/patch.py`` calls :func:`activate` first thing; +* the ``vllm.general_plugins`` entry point ``vllm_omni_windows`` -- vLLM calls + it in the API server, every engine-core child and every worker; +* the ``vllm_omni.general_plugins`` entry point of the same name. + +Off Windows every function here is a no-op that reports ``n/a``. +""" + +from __future__ import annotations + +import logging +import sys + +from vllm_omni.windows.report import LEDGER, report + +logger = logging.getLogger(__name__) + +__all__ = ["activate", "is_active", "is_windows", "report"] + + +def is_windows() -> bool: + return sys.platform == "win32" + + +def is_active() -> bool: + return bool(LEDGER["active"]) + + +def activate() -> dict[str, str]: + """Apply every layer once per process. Returns the per-step outcome.""" + LEDGER["activations"] += 1 + if not is_windows(): + LEDGER["steps"] = {"platform": "n/a"} + return LEDGER["steps"] + if LEDGER["active"]: + return LEDGER["steps"] + + from vllm_omni.windows import aio, paths, shims, vllm_patches + + steps: dict[str, str] = {} + # 1. modules that do not exist here -- before anything can import them + steps["fcntl"] = shims.provide("fcntl", shims.make_msvcrt_fcntl) + steps["uvloop"] = shims.provide("uvloop", shims.make_uvloop_shim) + # 2. process state + steps["event_loop_policy"] = aio.install_selector_policy() + steps["signal_handlers"] = aio.patch_signal_handlers() + steps["os_geteuid"] = paths.provide_geteuid() + # 3. vLLM functions replaced whole (only when the wheel lacks the branch patch) + steps["get_open_zmq_ipc_path"] = vllm_patches.patch_get_open_zmq_ipc_path() + steps["find_loaded_library"] = vllm_patches.patch_find_loaded_library() + LEDGER["steps"] = steps + LEDGER["active"] = True + logger.info("[vllm_omni.windows] active: %s", ", ".join(f"{k}={v}" for k, v in steps.items())) + fc = sys.modules.get("fcntl") + impl = getattr(fc, "_impl", None) + if impl is not None and getattr(impl, "shared_served_as_exclusive", 0): + logger.info("[vllm_omni.windows] LOCK_SH served as exclusive (msvcrt has no shared locks)") + return steps diff --git a/vllm_omni/windows/__main__.py b/vllm_omni/windows/__main__.py new file mode 100644 index 000000000..01422220b --- /dev/null +++ b/vllm_omni/windows/__main__.py @@ -0,0 +1,36 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""``python -m vllm_omni.windows report`` -- print the audit as JSON (exit 1 if a wheel site is missing).""" + +from __future__ import annotations + +import argparse +import json +import sys + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(prog="python -m vllm_omni.windows") + ap.add_argument("command", choices=["report"], nargs="?", default="report") + ap.add_argument("--json", dest="json_path", help="also write the report here") + args = ap.parse_args(argv) + + from vllm_omni.windows import activate, report + + activate() + rep = report() + text = json.dumps(rep, indent=2, default=str) + print(text) + if args.json_path: + with open(args.json_path, "w", encoding="utf-8") as fh: + fh.write(text + "\n") + sites = (rep.get("vllm") or {}).get("wheel_sites") or {} + missing = sorted(k for k, v in sites.items() if v in ("wheel_required", "unknown_upstream")) + if missing: + print(f"wheel_required: {', '.join(missing)}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/vllm_omni/windows/aio.py b/vllm_omni/windows/aio.py new file mode 100644 index 000000000..49a79ef7c --- /dev/null +++ b/vllm_omni/windows/aio.py @@ -0,0 +1,97 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""asyncio on Windows: the selector loop, and signal handlers that exist. + +``loop.add_signal_handler`` is Unix-only; on Windows every loop raises +``NotImplementedError``. vLLM's ``serve_http`` registers SIGINT/SIGTERM through +it, so the API server died right after the engine came up (W8, measured). The +fallback registers a plain ``signal.signal`` handler that hops back onto the +loop with ``call_soon_threadsafe`` -- the same shutdown path, registered the +only way Windows allows. Signals Windows does not have raise the original +error, so nothing is silently ignored. +""" + +from __future__ import annotations + +import asyncio +import signal +import sys +from collections.abc import Callable +from typing import Any + +from vllm_omni.windows.shims import ensure_selector_policy + +_MARK = "__vllm_omni_windows_signal_fallback__" + +_WINDOWS_SIGNALS = { + getattr(signal, name) for name in ("SIGINT", "SIGTERM", "SIGBREAK", "SIGABRT") if hasattr(signal, name) +} + + +def make_signal_fallback(original_add: Callable, original_remove: Callable): + """Build ``add_signal_handler``/``remove_signal_handler`` replacements. + + Returned functions try the loop's own implementation first (a real selector + loop on Unix keeps its native behaviour), and fall back to + ``signal.signal`` when the loop says ``NotImplementedError``. + """ + + def add_signal_handler(self, sig: int, callback: Callable, *args: Any) -> None: + try: + return original_add(self, sig, callback, *args) + except NotImplementedError: + if sig not in _WINDOWS_SIGNALS: + raise + + def _handler(signum, frame): # runs on the main thread, outside the loop + self.call_soon_threadsafe(callback, *args) + + signal.signal(sig, _handler) + registry = self.__dict__.setdefault(_MARK, {}) + registry[sig] = _handler + + def remove_signal_handler(self, sig: int) -> bool: + registry = self.__dict__.get(_MARK) or {} + if sig in registry: + signal.signal(sig, signal.SIG_DFL) + del registry[sig] + return True + try: + return original_remove(self, sig) + except NotImplementedError: + return False + + add_signal_handler.__vllm_omni_windows__ = True # type: ignore[attr-defined] + remove_signal_handler.__vllm_omni_windows__ = True # type: ignore[attr-defined] + return add_signal_handler, remove_signal_handler + + +def patch_signal_handlers() -> str: + """Install the fallback on the loop classes Windows uses. Returns already|patched|n/a.""" + if sys.platform != "win32": + return "n/a" + targets = [cls for cls in (getattr(asyncio, "ProactorEventLoop", None), getattr(asyncio, "SelectorEventLoop", None)) if cls] + if not targets: + return "n/a" + if all(getattr(cls.add_signal_handler, "__vllm_omni_windows__", False) for cls in targets): + return "already" + for cls in targets: + if getattr(cls.add_signal_handler, "__vllm_omni_windows__", False): + continue + add, remove = make_signal_fallback(cls.add_signal_handler, cls.remove_signal_handler) + cls.add_signal_handler = add # type: ignore[method-assign] + cls.remove_signal_handler = remove # type: ignore[method-assign] + return "patched" + + +def install_selector_policy() -> str: + if sys.platform != "win32": + return "n/a" + return "set" if ensure_selector_policy() else "already" + + +def running_loop_class() -> str | None: + try: + return type(asyncio.get_running_loop()).__name__ + except RuntimeError: + return None diff --git a/vllm_omni/windows/paths.py b/vllm_omni/windows/paths.py new file mode 100644 index 000000000..75b16383f --- /dev/null +++ b/vllm_omni/windows/paths.py @@ -0,0 +1,54 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Filesystem places POSIX code hard-codes, and the one ``os`` function Windows lacks.""" + +from __future__ import annotations + +import os +import sys +import tempfile +from pathlib import Path + +_POSIX_SHM = "/dev/shm" + + +def shm_dir() -> str: + """The directory for ``/dev/shm``-style lock and ack files. + + ``/dev/shm`` where it exists (Linux, WSL). On Windows a per-user directory + under ``%LOCALAPPDATA%`` (falling back to the temp dir), created on first + use. Callers that used to write ``f"/dev/shm/..."`` go through this so the + path is the same on every platform that has one and a real one elsewhere. + """ + if os.path.isdir(_POSIX_SHM): + return _POSIX_SHM + base = os.environ.get("LOCALAPPDATA") or tempfile.gettempdir() + path = Path(base) / "vllm_omni" / "shm" + path.mkdir(parents=True, exist_ok=True) + return str(path) + + +def shm_path(name: str) -> str: + return os.path.join(shm_dir(), name) + + +def provide_geteuid() -> str: + """``os.geteuid``/``os.getuid`` do not exist on Windows. + + ``host_weight_runtime`` compares ``st_uid`` against ``os.geteuid()`` for a + "did we create this" check; Windows ``os.stat`` reports ``st_uid == 0``, so + returning 0 makes the comparison mean the same thing it can mean there. + """ + if sys.platform != "win32": + return "n/a" + if hasattr(os, "geteuid"): + return "already" if getattr(os.geteuid, "__vllm_omni_windows__", False) else "real" + + def geteuid() -> int: + return 0 + + geteuid.__vllm_omni_windows__ = True # type: ignore[attr-defined] + os.geteuid = geteuid # type: ignore[attr-defined] + if not hasattr(os, "getuid"): + os.getuid = geteuid # type: ignore[attr-defined] + return "provided" diff --git a/vllm_omni/windows/procs.py b/vllm_omni/windows/procs.py new file mode 100644 index 000000000..96752d689 --- /dev/null +++ b/vllm_omni/windows/procs.py @@ -0,0 +1,38 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Process helpers that work on every OS (no ``os.killpg``, no ``kill -9``).""" + +from __future__ import annotations + +import subprocess + + +def kill_tree(pid: int, *, timeout: float = 30.0) -> dict[str, int]: + """Terminate ``pid`` and every descendant; escalate to kill after ``timeout``. + + Returns counts: ``{"terminated": n, "killed": m}``. Missing processes are + not an error (the tree may already be gone). + """ + import psutil + + try: + parent = psutil.Process(pid) + except psutil.NoSuchProcess: + return {"terminated": 0, "killed": 0} + procs = [*parent.children(recursive=True), parent] + for p in procs: + try: + p.terminate() + except psutil.NoSuchProcess: + pass + _, alive = psutil.wait_procs(procs, timeout=timeout) + for p in alive: + try: + p.kill() + except psutil.NoSuchProcess: + pass + return {"terminated": len(procs) - len(alive), "killed": len(alive)} + + +def kill_popen_tree(proc: subprocess.Popen, *, timeout: float = 30.0) -> dict[str, int]: + return kill_tree(proc.pid, timeout=timeout) diff --git a/vllm_omni/windows/report.py b/vllm_omni/windows/report.py new file mode 100644 index 000000000..1bbd4bb93 --- /dev/null +++ b/vllm_omni/windows/report.py @@ -0,0 +1,78 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""The audit: what the Windows layer did in this process, and what it could not do. + +Everything ``activate()`` does lands in ``LEDGER``; ``report()`` adds the +facts a Windows record needs next to a result (zmq transport, loop policy, +which vLLM sites came from the wheel, which native modules are present, which +optional dependencies import). Nothing here changes state. +""" + +from __future__ import annotations + +import asyncio +import importlib +import importlib.util +import sys +from typing import Any + +LEDGER: dict[str, Any] = {"active": False, "steps": {}, "activations": 0} + +OPTIONAL_DEPS = ("torchaudio", "soundfile", "diffusers", "flashinfer", "pynvml", "uvloop", "fcntl") + + +def _importable(name: str) -> str: + mod = sys.modules.get(name) + if mod is not None and getattr(mod, "__vllm_omni_windows_shim__", False): + return "shim" + try: + return "yes" if importlib.util.find_spec(name) is not None else "no" + except (ImportError, ValueError): + return "no" + + +def _native_modules() -> list[str]: + try: + import vllm + except ImportError: + return [] + from pathlib import Path + + root = Path(vllm.__file__).parent + suffixes = (".pyd", ".so") + return sorted( + str(p.relative_to(root)).replace("\\", "/").split(".", 1)[0] + for p in root.rglob("*") + if p.suffix in suffixes and "third_party" not in p.parts + ) + + +def report() -> dict[str, Any]: + from vllm_omni.windows import aio, paths, vllm_patches + + out: dict[str, Any] = { + "platform": sys.platform, + "active": LEDGER["active"], + "activations": LEDGER["activations"], + "steps": dict(LEDGER["steps"]), + } + out["asyncio"] = { + "policy": type(asyncio.get_event_loop_policy()).__name__, + "running_loop": aio.running_loop_class(), + } + try: + import zmq + + out["zmq"] = {"version": zmq.zmq_version(), "has_ipc": bool(zmq.has("ipc"))} + except ImportError: + out["zmq"] = None + out["shm_dir"] = paths.shm_dir() if sys.platform == "win32" else None + try: + import vllm + + out["vllm"] = {"version": vllm.__version__, "wheel_sites": vllm_patches.verify_wheel_sites()} + except ImportError: + out["vllm"] = None + out["native_modules"] = _native_modules() + out["optional_deps"] = {name: _importable(name) for name in OPTIONAL_DEPS} + return out diff --git a/vllm_omni/windows/shims.py b/vllm_omni/windows/shims.py new file mode 100644 index 000000000..a60b4055b --- /dev/null +++ b/vllm_omni/windows/shims.py @@ -0,0 +1,191 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Modules that do not exist on Windows, provided so ``import`` succeeds. + +Two of them account for most Windows import deaths measured in W0–W8: + +* ``fcntl`` — file locks. vLLM-Omni and model code call ``fcntl.flock`` for + device-init locks, weight-store locks, media caches. Windows has + ``msvcrt.locking`` instead: byte-range, mandatory, exclusive only. +* ``uvloop`` — no Windows wheels. Every ``uvloop.run(...)`` entry point can run + on ``asyncio.run`` provided the loop is a *selector* loop (zmq.asyncio needs + ``add_reader``; the Windows default Proactor loop has none). + +Both shims are only installed when the real module is absent, and both are +built on injectable primitives so the logic is testable on Linux. +""" + +from __future__ import annotations + +import asyncio +import os +import sys +import time +import types +from collections.abc import Callable + +# --------------------------------------------------------------------------- +# fcntl +# --------------------------------------------------------------------------- + +# POSIX values, so code that does arithmetic on them (``LOCK_EX | LOCK_NB``) +# behaves identically. +LOCK_SH = 1 +LOCK_EX = 2 +LOCK_NB = 4 +LOCK_UN = 8 + +# The locked byte lives far past any content a caller writes into a lock file +# (PID lines and the like), so the lock never overlaps data. Same design as +# ``vllm_omni._filelock_compat`` (W1). +_LOCK_OFFSET = 1 << 30 + + +class FcntlShim: + """``flock``/``lockf`` on a byte-range primitive. + + ``locking(fd, mode, nbytes)`` follows ``msvcrt.locking``'s contract: mode + ``NBLCK`` takes a non-blocking exclusive lock and raises ``OSError`` when + held elsewhere; ``UNLCK`` releases. Shared locks (``LOCK_SH``) are served + as exclusive because the primitive has none -- strictly more serialising + than POSIX, never less safe; this is said once per process in the log. + """ + + def __init__( + self, + locking: Callable[[int, int, int], None], + nblck: int, + unlck: int, + *, + sleep: Callable[[float], None] = time.sleep, + poll_interval: float = 0.01, + ) -> None: + self._locking = locking + self._nblck = nblck + self._unlck = unlck + self._sleep = sleep + self._poll_interval = poll_interval + self.shared_served_as_exclusive = 0 + + def _at_lock_offset(self, fd: int, fn: Callable[[], None]) -> None: + here = os.lseek(fd, 0, os.SEEK_CUR) + os.lseek(fd, _LOCK_OFFSET, os.SEEK_SET) + try: + fn() + finally: + os.lseek(fd, here, os.SEEK_SET) + + def flock(self, fd: int, operation: int) -> None: + fd = fd if isinstance(fd, int) else fd.fileno() + if operation & LOCK_UN: + self._at_lock_offset(fd, lambda: self._locking(fd, self._unlck, 1)) + return + if operation & LOCK_SH and not operation & LOCK_EX: + self.shared_served_as_exclusive += 1 + non_blocking = bool(operation & LOCK_NB) + while True: + try: + self._at_lock_offset(fd, lambda: self._locking(fd, self._nblck, 1)) + return + except OSError as exc: + if non_blocking: + raise BlockingIOError(str(exc)) from exc + self._sleep(self._poll_interval) + + def lockf(self, fd: int, cmd: int, len: int = 0, start: int = 0, whence: int = 0) -> None: # noqa: A002 + # POSIX lockf's F_LOCK/F_TLOCK/F_ULOCK spell the same three states. + self.flock(fd, cmd) + + def as_module(self) -> types.ModuleType: + mod = types.ModuleType("fcntl") + mod.__doc__ = "fcntl shim from vllm_omni.windows (msvcrt.locking underneath)" + mod.LOCK_SH, mod.LOCK_EX, mod.LOCK_NB, mod.LOCK_UN = LOCK_SH, LOCK_EX, LOCK_NB, LOCK_UN + # lockf's command constants, for callers that use them by name. + mod.F_ULOCK, mod.F_LOCK, mod.F_TLOCK, mod.F_TEST = LOCK_UN, LOCK_EX, LOCK_EX | LOCK_NB, LOCK_NB + mod.flock = self.flock + mod.lockf = self.lockf + mod.__vllm_omni_windows_shim__ = True + mod._impl = self + return mod + + +def make_msvcrt_fcntl() -> types.ModuleType: + import msvcrt # Windows only + + return FcntlShim(msvcrt.locking, msvcrt.LK_NBLCK, msvcrt.LK_UNLCK).as_module() + + +# --------------------------------------------------------------------------- +# uvloop +# --------------------------------------------------------------------------- + + +def selector_policy_class() -> type | None: + return getattr(asyncio, "WindowsSelectorEventLoopPolicy", None) + + +def ensure_selector_policy() -> bool: + """Make new event loops selector loops. Returns True if it changed the policy.""" + policy = selector_policy_class() + if policy is None: # not Windows + return False + if isinstance(asyncio.get_event_loop_policy(), policy): + return False + asyncio.set_event_loop_policy(policy()) + return True + + +def make_uvloop_shim() -> types.ModuleType: + """A ``uvloop`` whose ``run``/``install``/``new_event_loop`` are asyncio on a selector loop.""" + mod = types.ModuleType("uvloop") + mod.__doc__ = "uvloop shim from vllm_omni.windows: asyncio on a selector event loop" + mod.__version__ = "0.0.0+vllm_omni_windows_shim" + + def install() -> None: + ensure_selector_policy() + + def new_event_loop() -> asyncio.AbstractEventLoop: + ensure_selector_policy() + return asyncio.new_event_loop() + + def run(main, **kwargs): + ensure_selector_policy() + return asyncio.run(main, **kwargs) + + mod.install = install + mod.new_event_loop = new_event_loop + mod.run = run + mod.EventLoopPolicy = selector_policy_class() or asyncio.DefaultEventLoopPolicy + mod.Loop = getattr(asyncio, "SelectorEventLoop", asyncio.AbstractEventLoop) + mod.__vllm_omni_windows_shim__ = True + return mod + + +# --------------------------------------------------------------------------- +# installation +# --------------------------------------------------------------------------- + + +def _real_module_importable(name: str) -> bool: + if name in sys.modules: + return not getattr(sys.modules[name], "__vllm_omni_windows_shim__", False) + try: + __import__(name) + return True + except ImportError: + return False + + +def provide(name: str, factory: Callable[[], types.ModuleType]) -> str: + """Install ``factory()`` as ``sys.modules[name]`` unless a real module exists. + + Returns ``"real"`` (nothing done), ``"provided"`` (shim installed now) or + ``"already"`` (our shim was installed earlier in this process). + """ + existing = sys.modules.get(name) + if existing is not None and getattr(existing, "__vllm_omni_windows_shim__", False): + return "already" + if _real_module_importable(name): + return "real" + sys.modules[name] = factory() + return "provided" diff --git a/vllm_omni/windows/vllm_patches.py b/vllm_omni/windows/vllm_patches.py new file mode 100644 index 000000000..1ec53166c --- /dev/null +++ b/vllm_omni/windows/vllm_patches.py @@ -0,0 +1,180 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Runtime patches of vLLM control-plane functions, and verification of the rest. + +The rule (design §4): a function may be replaced *whole* at runtime only when +its entire body is platform logic and its upstream body is one we recognise by +hash. Sites whose fix sits in the middle of a larger function (the engine +startup sentinel poll, the sampler seed dtype) must come from the wheel -- the +Windows branch carries them -- and are only *verified* here, so a run on an +unpatched wheel is reported as ``wheel_required`` at the exact site instead of +failing somewhere downstream. + +Every outcome string used here: ``already`` (the wheel carries the branch +patch, marker found), ``patched`` (replaced now), ``unknown_upstream`` (no +marker and the upstream body is not one we know -- left alone, reported), +``wheel_required`` (verify-only site, marker absent), ``not_needed`` (another +layer covers it), ``n/a`` (off Windows or module absent). +""" + +from __future__ import annotations + +import hashlib +import inspect +import sys +from typing import Any + +BRANCH_MARK = "edge-infer" # every branch patch carries "[edge-infer W? windows patch]" + +# sha256[:16] of inspect.getsource() of the *upstream* function, measured on the +# PyPI wheels vllm==0.28.0 and vllm==0.29.0 (identical bodies at both tags). +KNOWN_UPSTREAM = { + "get_open_zmq_ipc_path": {"1d9aeb608d4c5dbc"}, + "find_loaded_library": {"b7916f0a8ebae56c"}, +} + + +def _src_hash(obj: Any) -> str | None: + try: + return hashlib.sha256(inspect.getsource(obj).encode()).hexdigest()[:16] + except (OSError, TypeError): + return None + + +def _has_mark(obj: Any) -> bool: + try: + return BRANCH_MARK in inspect.getsource(obj) + except (OSError, TypeError): + return False + + +# --------------------------------------------------------------------------- +# replaced whole +# --------------------------------------------------------------------------- + + +def _win_get_open_zmq_ipc_path() -> str: + """The W0 body: libzmq on Windows has no ipc:// transport; loopback TCP is the local option. + + This drops the filesystem permissions a Unix socket carries -- a loopback + port is reachable by any local user -- so it is a fallback conditional on + ipc being truly unavailable, never a preference. + """ + import zmq + + from vllm import envs + from vllm.utils.network_utils import get_open_port, get_tcp_uri + + if not zmq.has("ipc"): + return get_tcp_uri("127.0.0.1", get_open_port()) + from uuid import uuid4 + + return f"ipc://{envs.VLLM_RPC_BASE_PATH}/{uuid4()}" + + +def _make_win_find_loaded_library(original): + def find_loaded_library(lib_name: str) -> str | None: + """On Windows an already-imported extension module's __file__ IS the loaded path (W0).""" + if sys.platform != "win32": + return original(lib_name) + for mod in list(sys.modules.values()): + f = getattr(mod, "__file__", None) + if not f: + continue + base = f.replace("\\", "/").rsplit("/", maxsplit=1)[-1] + if base.startswith((f"{lib_name}.", f"{lib_name}-")): + return f + return None + + find_loaded_library.__vllm_omni_windows__ = True # type: ignore[attr-defined] + return find_loaded_library + + +def _replace(module, name: str, replacement) -> str: + current = getattr(module, name, None) + if current is None: + return "n/a" + if getattr(current, "__vllm_omni_windows__", False): + return "already" + if _has_mark(current): + return "already" + if _src_hash(current) not in KNOWN_UPSTREAM.get(name, set()): + return "unknown_upstream" + setattr(module, name, replacement) + return "patched" + + +def patch_get_open_zmq_ipc_path() -> str: + if sys.platform != "win32": + return "n/a" + try: + from vllm.utils import network_utils + except ImportError: + return "n/a" + _win_get_open_zmq_ipc_path.__vllm_omni_windows__ = True # type: ignore[attr-defined] + status = _replace(network_utils, "get_open_zmq_ipc_path", _win_get_open_zmq_ipc_path) + if status == "patched": + # Callers that imported the name directly (``from ... import get_open_zmq_ipc_path``) + # before activation keep the old binding; the ones that matter import lazily + # (Omni's stage_engine_startup) or after plugin load (vLLM's own engine code). + pass + return status + + +def patch_find_loaded_library() -> str: + if sys.platform != "win32": + return "n/a" + try: + from vllm.utils import system_utils + except ImportError: + return "n/a" + current = getattr(system_utils, "find_loaded_library", None) + if current is None: + return "n/a" + if getattr(current, "__vllm_omni_windows__", False) or _has_mark(current): + return "already" + if _src_hash(current) not in KNOWN_UPSTREAM["find_loaded_library"]: + return "unknown_upstream" + system_utils.find_loaded_library = _make_win_find_loaded_library(current) + return "patched" + + +# --------------------------------------------------------------------------- +# verify only (the fix must be in the wheel) +# --------------------------------------------------------------------------- + + +def _verify(import_path: str, attr: str | None, needle: str) -> str: + try: + module = __import__(import_path, fromlist=["_"]) + except ImportError: + return "n/a" + obj = getattr(module, attr) if attr else module + try: + src = inspect.getsource(obj) + except (OSError, TypeError): + return "n/a" + return "already" if needle in src else "wheel_required" + + +def verify_wheel_sites() -> dict[str, str]: + """The sites a runtime plugin cannot fix; each must be `already` on Windows.""" + if sys.platform != "win32": + return {} + out = { + # v1/engine/utils.py: zmq cannot poll a process HANDLE; 500 ms poll + exitcode checks. + "wait_for_engine_startup_sentinel_poll": _verify("vllm.v1.engine.utils", "wait_for_engine_startup", "_win_sentinels"), + # sample/states.py: numpy's default int is int32 on Windows; int64 seed bounds overflow. + "sampler_seed_int64": _verify("vllm.v1.worker.gpu.sample.states", None, "dtype=np.int64"), + } + # v1/utils.py binds uvloop lazily on the branch; with the uvloop shim present the + # module-level import would succeed anyway, so absence is "not_needed", not a failure. + status = _verify("vllm.v1.utils", None, "__getattr__") + if status == "wheel_required": + status = "not_needed" if "uvloop" in sys.modules else "wheel_required" + out["v1_utils_uvloop_lazy"] = status + # launcher.serve_http: the branch has a signal.signal fallback; the loop-class patch + # (aio.patch_signal_handlers) covers an unpatched wheel too. + status = _verify("vllm.entrypoints.launchers.launcher", "serve_http", "signal-handler") + out["serve_http_signal_fallback"] = "already" if status == "already" else ("not_needed" if status == "wheel_required" else status) + return out