Spaces:
Running
Running
File size: 4,746 Bytes
085f660 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | """Check four vLLM/PyTorch binary roles against Eggcracker's exact release pins."""
from __future__ import annotations
import argparse
import json
import os
import stat
import sys
from collections.abc import Callable, Sequence
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from lumi_eggcracker import __version__, elfmarkers
from lumi_eggcracker.elfmarkers import (
PYTORCH_ATEN_EVIDENCE_ID,
PYTORCH_BRIDGE_EVIDENCE_ID,
VLLM_EXTENSION_EVIDENCE_ID,
VLLM_PYTHON_EVIDENCE_ID,
RuntimeEvidence,
)
SCHEMA = "lumi-eggcracker.vllm-compatibility.v1"
PROFILE = "content.safetensors-vllm"
DescriptorInspector = Callable[[int], RuntimeEvidence | None]
def _absolute_path(parser: argparse.ArgumentParser, value: str) -> Path:
path = Path(value)
if not path.is_absolute():
parser.error("vLLM and PyTorch binary paths must be absolute")
return path
def _inspect_regular_path(
path: Path, inspector: DescriptorInspector
) -> RuntimeEvidence | None:
try:
metadata = path.lstat()
except OSError:
return None
if not stat.S_ISREG(metadata.st_mode):
return None
flags = (
os.O_RDONLY
| getattr(os, "O_CLOEXEC", 0)
| getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_NONBLOCK", 0)
)
try:
descriptor = os.open(path, flags)
except OSError:
return None
try:
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
return None
return inspector(descriptor)
except OSError:
return None
finally:
os.close(descriptor)
def compatibility(
pytorch_bridge: Path,
pytorch_aten: Path,
vllm_python: Path,
vllm_extension: Path,
) -> dict[str, object]:
evidence = {
"pytorch_bridge_identity": _inspect_regular_path(
pytorch_bridge, elfmarkers._inspect_pytorch_descriptor
),
"pytorch_aten_identity": _inspect_regular_path(
pytorch_aten, elfmarkers._inspect_pytorch_descriptor
),
"vllm_python_identity": _inspect_regular_path(
vllm_python, elfmarkers._inspect_vllm_python_descriptor
),
"vllm_extension_identity": _inspect_regular_path(
vllm_extension, elfmarkers._inspect_vllm_extension_descriptor
),
}
expected = {
"pytorch_bridge_identity": PYTORCH_BRIDGE_EVIDENCE_ID,
"pytorch_aten_identity": PYTORCH_ATEN_EVIDENCE_ID,
"vllm_python_identity": VLLM_PYTHON_EVIDENCE_ID,
"vllm_extension_identity": VLLM_EXTENSION_EVIDENCE_ID,
}
checks = {
role: (
"SUPPORTED"
if value is not None and value.evidence_id == expected[role]
else "UNSUPPORTED"
)
for role, value in evidence.items()
}
supported = all(value == "SUPPORTED" for value in checks.values())
return {
"checks": checks,
"limitations": [
"BINARY_IDENTITY_ONLY",
"SAFETENSORS_MODEL_NOT_INSPECTED",
"LIVE_TOPOLOGY_NOT_PROVEN",
"EXECUTION_CONTEXT_NOT_QUALIFIED",
"COMPATIBILITY_CHECK_NOT_CONTAINMENT_EVIDENCE",
"COMPATIBILITY_CHECK_NOT_ADOPTION_EVIDENCE",
],
"result": "SUPPORTED" if supported else "UNSUPPORTED",
"schema_version": SCHEMA,
"target_profile": PROFILE,
"version": __version__,
}
def parser() -> argparse.ArgumentParser:
value = argparse.ArgumentParser(
description=(
"Check exact vLLM and PyTorch binary identities without installing or starting "
"Eggcracker."
)
)
value.add_argument(
"--pytorch-bridge", required=True, help="absolute path to the PyTorch bridge ELF"
)
value.add_argument(
"--pytorch-aten", required=True, help="absolute path to the PyTorch ATen ELF"
)
value.add_argument(
"--vllm-python", required=True, help="absolute path to the vLLM CPython ELF"
)
value.add_argument(
"--vllm-extension", required=True, help="absolute path to the vLLM extension ELF"
)
return value
def main(argv: Sequence[str] | None = None) -> int:
argument_parser = parser()
args = argument_parser.parse_args(argv)
result = compatibility(
_absolute_path(argument_parser, args.pytorch_bridge),
_absolute_path(argument_parser, args.pytorch_aten),
_absolute_path(argument_parser, args.vllm_python),
_absolute_path(argument_parser, args.vllm_extension),
)
print(json.dumps(result, indent=2, sort_keys=True))
return 0 if result["result"] == "SUPPORTED" else 2
if __name__ == "__main__":
raise SystemExit(main())
|