[edge-infer] vllm_omni.windows: host-OS compatibility plugin for native Windows

One package that makes the POSIX assumptions measured in W0-W8 hold in
every process vLLM / vLLM-Omni starts, and reports what it did:
- provides fcntl (flock/lockf on msvcrt.locking; LOCK_SH served as
  exclusive) and uvloop (asyncio.run on a selector loop) when absent;
- installs WindowsSelectorEventLoopPolicy and a signal.signal fallback
  for loop.add_signal_handler on the Windows loop classes;
- replaces whole vLLM's get_open_zmq_ipc_path / find_loaded_library when
  the wheel lacks the branch patch and the upstream body hash is known;
- verifies the wheel-only sites (sentinel poll, int64 seed) and reports
  wheel_required at the exact site instead of failing downstream;
- shm_dir() replaces the /dev/shm literals; os.geteuid provided.
Activated from vllm_omni/patch.py and as vllm.general_plugins /
vllm_omni.general_plugins entry points (idempotent; no-op off Windows).
 prints the audit; the M0 placement
report embeds it. 16 Linux tests through injected primitives.
Design: analysis/design_windows_compat_plugin_20260921.md

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
tzhouamandClaude Fable 5.1 committed 2026-09-21 14:02:01 +08:00
1 parent 97d97b6503
commit 00510ce38b
15 files changed
+1050 -2

No files matched your search

+6
View File
@@ -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]
View File
Whitespace-only changes.
+269
View File
@@ -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"] == {}
@@ -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:
+15
View File
@@ -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]:
+3 -1
View File
@@ -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
+7
View File
@@ -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
+72
View File
@@ -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
+36
View File
@@ -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())
+97
View File
@@ -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
+54
View File
@@ -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"
+38
View File
@@ -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)
+78
View File
@@ -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
+191
View File
@@ -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"
+180
View File
@@ -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