From c2ddfb80590206f06a3d949296c0a7a6dc0c3744 Mon Sep 17 00:00:00 2001 From: yxf Date: Tue, 10 Nov 2026 20:09:19 +0800 Subject: [PATCH] fix(deployer): reject missing accelerator mappings - Publish completed device mappings atomically so concurrent workload creation sees consistent results. - Retry empty device discovery results and preserve policy-specific Kubernetes device mapping caches. - Reject Docker accelerator requests without mappings and cover missing devices and concurrent initialization. Signed-off-by: yxf --- gpustack_runtime/deployer/__types__.py | 56 +++++---- gpustack_runtime/deployer/docker.py | 70 ++++------- gpustack_runtime/deployer/kubernetes.py | 16 ++- .../deployer/test_visible_devices_ordering.py | 117 +++++++++++++++++- 4 files changed, 180 insertions(+), 79 deletions(-) diff --git a/gpustack_runtime/deployer/__types__.py b/gpustack_runtime/deployer/__types__.py index 53cb6ea..7b6c129 100644 --- a/gpustack_runtime/deployer/__types__.py +++ b/gpustack_runtime/deployer/__types__.py @@ -8,6 +8,7 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from enum import Enum +from threading import Lock from typing import TYPE_CHECKING, Any from dataclasses_json import dataclass_json @@ -1492,6 +1493,9 @@ class DevicesMaterial: """ +UNKNOWN_RUNTIME_ENV = "UNKNOWN_RUNTIME_VISIBLE_DEVICES" + + class Deployer(ABC): """ Base class for all deployers. @@ -1549,6 +1553,7 @@ def is_supported() -> bool: def __init__(self, name: str): self._name = name + self._materials_lock = Lock() def __enter__(self): return self @@ -1556,15 +1561,30 @@ def __enter__(self): def __exit__(self, exc_type, exc_value, traceback): self.close() - def _prepare(self): + def _prepare(self) -> dict[str, DevicesMaterial]: """ - Detect devices and prepare materials. + Build and cache device materials while the caller holds the cache lock. """ - if self._materials is not None: - return + materials = self._detect_materials() + if materials: + self._materials = materials + return materials - self._materials = {} + # Keep the generic fallback available without caching an empty detection. + ben = "UNKNOWN_BACKEND_VISIBLE_DEVICES" + return { + UNKNOWN_RUNTIME_ENV: DevicesMaterial( + manufacturer=ManufacturerEnum.UNKNOWN, + runtime_env=UNKNOWN_RUNTIME_ENV, + backend_env=[ben], + cdi="unknown.com/gpu", + runtime_values={"all": "all"}, + backend_values={ben: {"all": "all"}}, + ), + } + def _detect_materials(self) -> dict[str, DevicesMaterial]: + materials: dict[str, DevicesMaterial] = {} devices = detect_devices(fast=False) if self.allowed_mig_devices: devices = expand_mig_devices(devices) @@ -1617,7 +1637,7 @@ def _prepare(self): dev_numa_affinities[dev_index], ) - self._materials[ren] = DevicesMaterial( + materials[ren] = DevicesMaterial( manufacturer=manu, runtime_env=ren, backend_env=ben_list, @@ -1638,21 +1658,7 @@ def _prepare(self): cpus_affinities=dev_cpus_affinities, ) - if self._materials: - return - - # Fallback to unknown backend - ren = "UNKNOWN_RUNTIME_VISIBLE_DEVICES" - ben_list = ["UNKNOWN_BACKEND_VISIBLE_DEVICES"] - cdi = "unknown.com/gpu" - self._materials[ren] = DevicesMaterial( - manufacturer=ManufacturerEnum.UNKNOWN, - runtime_env=ren, - backend_env=ben_list, - cdi=cdi, - runtime_values={"all": "all"}, - backend_values={ben: {"all": "all"} for ben in ben_list}, - ) + return materials def _get_materials( self, @@ -1665,8 +1671,12 @@ def _get_materials( to corresponding devices materials. """ - self._prepare() - return self._materials + if self._materials is not None: + return self._materials + with self._materials_lock: + if self._materials is not None: + return self._materials + return self._prepare() def get_manufacturer( self, diff --git a/gpustack_runtime/deployer/docker.py b/gpustack_runtime/deployer/docker.py index 2a70d89..32687e8 100644 --- a/gpustack_runtime/deployer/docker.py +++ b/gpustack_runtime/deployer/docker.py @@ -29,6 +29,7 @@ from ..logging import debug_log_exception, debug_log_warning from .__types__ import ( DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS, + UNKNOWN_RUNTIME_ENV, Container, ContainerCheck, ContainerEnv, @@ -1076,53 +1077,34 @@ def _create_containers( manu, ) - # Request devices. if r_v == "all": - # Configure privileged. create_options["privileged"] = True - # Request all devices. - for ren in runtime_envs: + + for ren in runtime_envs: + if r_v == "all" or privileged: r_vs = self.get_runtime_visible_devices(ren, fmt) - # Request device via CDI. - if cdi: - if "device_requests" not in create_options: - create_options["device_requests"] = [] - create_options["device_requests"].append( - docker.types.DeviceRequest( - driver="cdi", - count=0, - device_ids=r_vs, - ), - ) - continue - # Request device via visible devices env. - create_options["environment"][ren] = ",".join(r_vs) - else: - # Request specific devices. - for ren in runtime_envs: - # Request all devices if privileged, - # otherwise, normalize requested devices. - if privileged: - r_vs = self.get_runtime_visible_devices(ren, fmt) - else: - r_vs = self.map_runtime_visible_devices( - ren, - resource_values, - fmt, - ) - # Request device via CDI. - if cdi: - if "device_requests" not in create_options: - create_options["device_requests"] = [] - create_options["device_requests"].append( - docker.types.DeviceRequest( - driver="cdi", - count=0, - device_ids=r_vs, - ), - ) - continue - # Request device via visible devices env. + else: + r_vs = self.map_runtime_visible_devices( + ren, + resource_values, + fmt, + ) + if not r_vs or (ren == UNKNOWN_RUNTIME_ENV and r_v != "all"): + msg = ( + f"Cannot create container {container_name}: " + f"no detected devices for {ren} " + f"(requested {r_k}={r_v})" + ) + raise OperationError(msg) + if cdi: + create_options.setdefault("device_requests", []).append( + docker.types.DeviceRequest( + driver="cdi", + count=0, + device_ids=r_vs, + ), + ) + else: create_options["environment"][ren] = ",".join(r_vs) # If not requesting all devices but privileged, diff --git a/gpustack_runtime/deployer/kubernetes.py b/gpustack_runtime/deployer/kubernetes.py index 44533fc..69128c3 100644 --- a/gpustack_runtime/deployer/kubernetes.py +++ b/gpustack_runtime/deployer/kubernetes.py @@ -1844,24 +1844,22 @@ def __init__(self): self._materials_cache: dict[str, dict[str, DevicesMaterial]] = {} self._materials_locks = {policy: Lock() for policy in ("env", "kdp")} - def _prepare(self): + def _get_materials(self) -> dict[str, DevicesMaterial]: """ Reuse device materials built for the same resolved injection policy. """ if self._materials is not None: - return + return self._materials policy = self._resolve_resource_injection_policy() with self._materials_locks[policy]: if policy not in self._materials_cache: self._runtime_uuid_values_allowed = policy != "kdp" - try: - super()._prepare() - except Exception: - # Incomplete detection must not become a reusable device map. - self._materials = None - raise - self._materials_cache[policy] = self._materials + materials = self._prepare() + if self._materials is not None: + self._materials_cache[policy] = self._materials + return materials self._materials = self._materials_cache[policy] + return self._materials @property def allowed_runtime_uuid_values(self) -> bool: diff --git a/tests/gpustack_runtime/deployer/test_visible_devices_ordering.py b/tests/gpustack_runtime/deployer/test_visible_devices_ordering.py index 00a6e4b..0e58baa 100644 --- a/tests/gpustack_runtime/deployer/test_visible_devices_ordering.py +++ b/tests/gpustack_runtime/deployer/test_visible_devices_ordering.py @@ -3,6 +3,8 @@ # Docker daemon, Podman socket or Kubernetes cluster. # ruff: noqa: SLF001 +from concurrent.futures import ThreadPoolExecutor +from threading import Event from types import SimpleNamespace import docker.errors @@ -18,6 +20,7 @@ ContainerResources, Deployer, DevicesMaterial, + OperationError, ) from gpustack_runtime.deployer.docker import DockerDeployer, DockerWorkloadPlan from gpustack_runtime.deployer.kubernetes import ( @@ -25,7 +28,7 @@ KubernetesWorkloadPlan, ) from gpustack_runtime.deployer.podman import PodmanDeployer, PodmanWorkloadPlan -from gpustack_runtime.detector import ManufacturerEnum +from gpustack_runtime.detector import Device, ManufacturerEnum _ORDERING_ENV = "CUDA_DEVICE_ORDER" @@ -62,7 +65,7 @@ _MIXED_MATERIALS = {**_NVIDIA_MATERIALS, **_AMD_MATERIALS} -def _deployer(cls, materials: dict[str, DevicesMaterial]): +def _deployer(cls, materials: dict[str, DevicesMaterial] | None): # Bypass the deployer's __init__, which reaches out to a live daemon or # API server, and pin the materials so _prepare() short-circuits instead # of detecting devices. @@ -106,7 +109,7 @@ def create(self, **kwargs): def _docker_container_envs( - materials: dict[str, DevicesMaterial], + materials: dict[str, DevicesMaterial] | None, container: Container, monkeypatch, ) -> list[tuple[str, str]]: @@ -133,6 +136,114 @@ def _docker_container_envs( return list(created[0]["environment"].items()) +@pytest.mark.parametrize( + "resource_key, requested", + [ + ("huawei.com/devices", "2,3"), + ("huawei.com/devices", "all"), + ("gpustack.ai/devices", "2,3"), + ], +) +def test_docker_rejects_request_without_device_material( + monkeypatch, + resource_key, + requested, +): + monkeypatch.setattr( + "gpustack_runtime.deployer.__types__.detect_devices", + lambda **_: [], + ) + container = _container({resource_key: requested}, privileged=True) + + with pytest.raises(OperationError, match=f"requested {resource_key}={requested}"): + _docker_container_envs(None, container, monkeypatch) + + +def test_docker_preserves_automap_all_fallback(monkeypatch): + monkeypatch.setattr( + "gpustack_runtime.deployer.__types__.detect_devices", + lambda **_: [], + ) + container = _container({"gpustack.ai/devices": "all"}) + + assert dict(_docker_container_envs(None, container, monkeypatch)) == { + "UNKNOWN_RUNTIME_VISIBLE_DEVICES": "all", + } + + +def test_docker_retries_device_detection_after_empty_result(monkeypatch): + devices = iter( + [ + [], + [ + Device( + manufacturer=ManufacturerEnum.ASCEND, + index=index, + name="Ascend 910B2", + uuid=str(index), + appendix={}, + ) + for index in range(4) + ], + ], + ) + monkeypatch.setattr( + "gpustack_runtime.deployer.__types__.detect_devices", + lambda **_: next(devices), + ) + deployer = _deployer(DockerDeployer, None) + runtime_env = "ASCEND_VISIBLE_DEVICES" + + assert deployer.get_runtime_visible_devices(runtime_env) == [] + assert deployer.get_runtime_visible_devices(runtime_env) == ["0", "1", "2", "3"] + assert deployer.map_backend_visible_devices([runtime_env], ["2", "3"]) == { + "ASCEND_RT_VISIBLE_DEVICES": "2,3", + "NPU_VISIBLE_DEVICES": "2,3", + } + + +def test_concurrent_device_mapping_waits_for_detection(monkeypatch): + detection_started = Event() + finish_detection = Event() + lookup_started = Event() + deployer = _deployer(DockerDeployer, None) + runtime_env = "ASCEND_VISIBLE_DEVICES" + + def detect(**_): + assert not detection_started.is_set() + detection_started.set() + assert finish_detection.wait(timeout=5) + return [ + Device( + manufacturer=ManufacturerEnum.ASCEND, + index=0, + name="Ascend 910B2", + uuid="0", + appendix={}, + ), + ] + + def lookup(): + lookup_started.set() + return deployer.map_backend_visible_devices([runtime_env], ["0"]) + + monkeypatch.setattr("gpustack_runtime.deployer.__types__.detect_devices", detect) + with ThreadPoolExecutor(max_workers=2) as executor: + first = executor.submit(deployer.get_runtime_visible_devices, runtime_env) + try: + assert detection_started.wait(timeout=5) + second = executor.submit(lookup) + assert lookup_started.wait(timeout=5) + assert deployer._materials is None + finally: + finish_detection.set() + assert first.result(timeout=5) == ["0"] + assert second.result(timeout=5) == { + "ASCEND_RT_VISIBLE_DEVICES": "0", + "NPU_VISIBLE_DEVICES": "0", + } + + def _podman_container_envs( materials: dict[str, DevicesMaterial], container: Container,