File size: 5,650 Bytes
9438a57 | 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 | """Gerçek zamanlı GPU durumu: nvidia-smi (NVIDIA) + Ollama /api/ps (modelin GPU payı).
nvidia-smi, uygulama doğrudan bilgisayarda çalışırken PATH'ten; Docker'da ise NVIDIA konteyner çalışma
zamanının (docker-compose.gpu.yml) konteynere eklediği ikiliden okunur. Bulunamazsa yalnızca Ollama bilgisi
döner. Sonuç kısa süre önbelleğe alınır; böylece birden çok sekme sorgulasa da nvidia-smi saniyede en fazla
bir kez çalışır.
"""
from __future__ import annotations
import shutil
import subprocess
import threading
import time
from typing import Any, Optional
FIELDS = [
("index", int), ("name", str), ("driver_version", str), ("utilization.gpu", float),
("utilization.memory", float), ("memory.used", float), ("memory.total", float), ("temperature.gpu", float),
("power.draw", float), ("power.limit", float), ("clocks.sm", float), ("clocks.max.sm", float),
("fan.speed", float), ("pstate", str),
]
KEYS = {"utilization.gpu": "util", "utilization.memory": "mem_util", "memory.used": "mem_used_mib",
"memory.total": "mem_total_mib", "temperature.gpu": "temp_c", "power.draw": "power_w",
"power.limit": "power_limit_w", "clocks.sm": "clock_mhz", "clocks.max.sm": "clock_max_mhz",
"fan.speed": "fan_pct", "driver_version": "driver"}
_lock = threading.Lock()
_cache: dict[str, Any] = {"ts": 0.0, "data": None}
def _num(v: str, cast):
v = v.strip()
if not v or v.startswith("[") or v.lower() in ("n/a", "not supported"):
return None
try:
return cast(v)
except ValueError:
return None
def parse_nvidia_smi(text: str) -> list[dict[str, Any]]:
out = []
for line in text.strip().splitlines():
parts = [p.strip() for p in line.split(",")]
if len(parts) < len(FIELDS):
continue
g: dict[str, Any] = {}
for (field, cast), raw in zip(FIELDS, parts):
g[KEYS.get(field, field)] = _num(raw, cast) if cast is not str else (raw or None)
out.append(g)
return out
def query_nvidia(binary: Optional[str] = None, timeout: float = 3.0) -> tuple[list[dict[str, Any]], Optional[str]]:
exe = binary or shutil.which("nvidia-smi")
if not exe:
return [], "nvidia-smi bulunamadı"
try:
p = subprocess.run([exe, "--query-gpu=" + ",".join(f for f, _ in FIELDS), "--format=csv,noheader,nounits"],
capture_output=True, text=True, timeout=timeout)
except (OSError, subprocess.TimeoutExpired) as e:
return [], str(e)
if p.returncode != 0:
return [], (p.stderr or p.stdout).strip()[:300] or f"nvidia-smi çıkış kodu {p.returncode}"
return parse_nvidia_smi(p.stdout), None
def status(ollama_running: Optional[list[dict[str, Any]]] = None, max_age: float = 1.0) -> dict[str, Any]:
with _lock:
now = time.time()
if _cache["data"] is not None and now - _cache["ts"] < max_age:
gpus, err = _cache["data"]
else:
gpus, err = query_nvidia()
_cache.update(ts=now, data=(gpus, err))
return {"ts": time.time(), "available": bool(gpus), "source": "nvidia-smi" if gpus else None,
"gpus": gpus, "error": None if gpus else err, "models": ollama_running or []}
class EnergyMeter:
"""İş süresince GPU güç çekişini örnekleyip enerjiyi (Wh) yamuk kuralıyla toplar.
Ölçülen değer kartın toplam güç çekişidir (o sırada çalışan başka işler de dahil); deneyler tek tek
koşulduğundan iş başına enerji için iyi bir yaklaşımdır. nvidia-smi yoksa ölçüm yapılmaz.
"""
def __init__(self, interval: float = 1.0, reader=None) -> None:
self.interval = interval
self._read = reader or (lambda: status(None, max_age=interval * 0.9)["gpus"])
self._stop = threading.Event()
self._thread: Optional[threading.Thread] = None
self.wh = 0.0
self.samples = 0
self.util_sum = 0.0
self.peak_mem_mib = 0.0
self._last: Optional[tuple[float, float]] = None
self.available = False
def _sample(self) -> None:
try:
gpus = self._read()
except Exception: # noqa: BLE001
gpus = []
watts = [g.get("power_w") for g in gpus if g.get("power_w") is not None]
if not watts:
return
self.available = True
now, p = time.time(), float(sum(watts))
if self._last:
t0, p0 = self._last
self.wh += (p0 + p) / 2 * (now - t0) / 3600.0
self._last = (now, p)
self.samples += 1
self.util_sum += sum(g.get("util") or 0 for g in gpus) / max(1, len(gpus))
self.peak_mem_mib = max(self.peak_mem_mib, float(sum(g.get("mem_used_mib") or 0 for g in gpus)))
def _loop(self) -> None:
while not self._stop.is_set():
self._sample()
self._stop.wait(self.interval)
def start(self) -> "EnergyMeter":
self._sample()
if not self.available: # nvidia-smi yok: iş parçacığı açma
return self
self._thread = threading.Thread(target=self._loop, daemon=True)
self._thread.start()
return self
def stop(self) -> dict[str, Any]:
self._stop.set()
if self._thread:
self._thread.join(timeout=3)
self._sample()
if not self.available or self.samples < 2:
return {}
return {"energy_wh": round(self.wh, 3), "gpu_util_avg": round(self.util_sum / self.samples, 1),
"gpu_mem_peak_gb": round(self.peak_mem_mib / 1024, 2), "energy_samples": self.samples}
|