80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
|
"""Laufzeit-Fähigkeiten der Hardware (GPU-Erkennung) — eine Quelle für Admin + /api/config.
|
||
|
|
|
||
|
|
Die GPU-Maske `settings.gpus` wird beim Start an `CUDA_VISIBLE_DEVICES` gekoppelt
|
||
|
|
(`app/config.py::_apply_gpu_visibility`). Hier wird nur **erkannt** und nach Maske
|
||
|
|
**aufgelöst**; CUDA_VISIBLE_DEVICES wird zusätzlich respektiert, damit „angezeigt" =
|
||
|
|
„real nutzbar" gilt.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import os
|
||
|
|
import shutil
|
||
|
|
|
||
|
|
from app.config import settings, _parse_gpu_indices
|
||
|
|
|
||
|
|
|
||
|
|
async def detect_physical_gpus() -> list[dict]:
|
||
|
|
"""Physisch vorhandene NVIDIA-GPUs via nvidia-smi. [] wenn kein Tool/keine GPU/Fehler."""
|
||
|
|
if not shutil.which("nvidia-smi"):
|
||
|
|
return []
|
||
|
|
try:
|
||
|
|
proc = await asyncio.create_subprocess_exec(
|
||
|
|
"nvidia-smi",
|
||
|
|
"--query-gpu=index,name,memory.used,memory.total",
|
||
|
|
"--format=csv,noheader,nounits",
|
||
|
|
stdout=asyncio.subprocess.PIPE,
|
||
|
|
stderr=asyncio.subprocess.DEVNULL,
|
||
|
|
)
|
||
|
|
out, _ = await asyncio.wait_for(proc.communicate(), timeout=6.0)
|
||
|
|
if proc.returncode != 0:
|
||
|
|
return []
|
||
|
|
except (asyncio.TimeoutError, OSError):
|
||
|
|
return []
|
||
|
|
gpus: list[dict] = []
|
||
|
|
for ln in out.decode("utf-8", "replace").splitlines():
|
||
|
|
parts = [p.strip() for p in ln.split(",")]
|
||
|
|
if len(parts) == 4 and parts[0].isdigit():
|
||
|
|
used, total = int(parts[2]), int(parts[3])
|
||
|
|
gpus.append({
|
||
|
|
"index": int(parts[0]),
|
||
|
|
"name": parts[1],
|
||
|
|
"used_mib": used,
|
||
|
|
"total_mib": total,
|
||
|
|
"percent": round(used / total * 100) if total else 0,
|
||
|
|
})
|
||
|
|
return gpus
|
||
|
|
|
||
|
|
|
||
|
|
def _cuda_visible_mask() -> set[int] | None:
|
||
|
|
"""Indizes aus CUDA_VISIBLE_DEVICES; None = keine Einschränkung, set() = explizit keine."""
|
||
|
|
raw = os.environ.get("CUDA_VISIBLE_DEVICES")
|
||
|
|
if raw is None:
|
||
|
|
return None
|
||
|
|
if raw.strip() == "":
|
||
|
|
return set() # leer = keine GPU für CUDA-Anwendungen
|
||
|
|
return set(_parse_gpu_indices(raw))
|
||
|
|
|
||
|
|
|
||
|
|
async def resolve_gpus() -> list[dict]:
|
||
|
|
"""Sichtbare/nutzbare GPUs = physisch ∩ settings.gpus-Maske ∩ CUDA_VISIBLE_DEVICES."""
|
||
|
|
mode = (settings.gpus or "auto").strip().lower()
|
||
|
|
if mode in ("none", "off", "false", "-1", ""):
|
||
|
|
return []
|
||
|
|
physical = await detect_physical_gpus()
|
||
|
|
explicit = None if mode == "auto" else set(_parse_gpu_indices(settings.gpus))
|
||
|
|
cuda = _cuda_visible_mask()
|
||
|
|
visible: list[dict] = []
|
||
|
|
for g in physical:
|
||
|
|
idx = g["index"]
|
||
|
|
if explicit is not None and idx not in explicit:
|
||
|
|
continue
|
||
|
|
if cuda is not None and idx not in cuda:
|
||
|
|
continue
|
||
|
|
visible.append(g)
|
||
|
|
return visible
|
||
|
|
|
||
|
|
|
||
|
|
async def gpu_available() -> bool:
|
||
|
|
return len(await resolve_gpus()) > 0
|