Coverage for src/lilbee/providers/fleet/gpu_backends/intel.py: 100%
142 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Intel GPU utilization from the first available source.
3Intel exposes GPU activity three different ways depending on the GPU class,
4kernel, and privileges. We try them in order and take the first that reports:
61. ``xpu-smi`` (Intel XPU Manager) -- Data Center GPU Max/Flex and Arc only.
7 Reads a ``device_level`` array of {metrics_type, value} objects keyed by the
8 xpum_stats_type_enum names. One device per call, no root.
92. DRM ``fdinfo`` -- the kernel's per-client engine-busy counters. Covers
10 consumer iGPUs with no root and no extra tool, but only on kernels new enough
11 to publish i915 engine stats (5.19+, or 6.5+ under GuC submission).
123. ``intel_gpu_top`` (Intel GPU Tools) -- reads the i915 PMU, so it covers
13 essentially every consumer iGPU on kernel 4.16+, but needs CAP_PERFMON (or
14 root); it falls through cleanly when the permission isn't granted.
16The fdinfo and intel_gpu_top sources report one device (the consumer case is a
17single iGPU); their reading is keyed to the lowest requested index. VRAM is left
18as the 0/0 structural sentinel since none of these report an iGPU's total memory.
19"""
21from __future__ import annotations
23import functools
24import json
25import shutil
26import subprocess
27from dataclasses import dataclass
28from enum import StrEnum
30from lilbee.providers.fleet.gpu_backends import fdinfo
31from lilbee.providers.fleet.gpu_backends.base import UtilSample, extract_int, run_smi
33_TOOL_XPU_SMI = "xpu-smi"
34_TOOL_IGT = "intel_gpu_top"
35_TIMEOUT_S = 5.0
37# xpu-smi device_level metrics_type names (xpum_stats_type_enum).
38_METRIC_UTIL = "XPUM_STATS_GPU_UTILIZATION"
39_METRIC_TEMP = "XPUM_STATS_GPU_CORE_TEMPERATURE"
41# intel_gpu_top streams JSON samples forever; run it for one short window and
42# read the partial output. A couple of 200ms periods lands a fresh per-interval
43# reading while keeping the (synchronous) probe from stalling the caller long.
44_IGT_SAMPLE_MS = 200
45_IGT_CAPTURE_S = 0.6
47# The i915 DRM driver name, for the fdinfo reader.
48_I915 = "i915"
50# intel_gpu_top needs CAP_PERFMON to read the i915 PMU. When it is installed but
51# unprivileged, the util reads back empty. A working PMU streams, so the check
52# hits the timeout and reports usable.
53_IGT_PERM_CHECK_S = 1.0
56class IntelHintKind(StrEnum):
57 """Which fix would make an Intel GPU's utilization readable."""
59 GRANT = "grant" # intel_gpu_top is installed but the i915 PMU is permission-blocked
60 INSTALL = "install" # intel_gpu_top is not installed at all
63@dataclass(frozen=True)
64class IntelUtilHint:
65 """An actionable fix for an unreadable Intel GPU utilization reading."""
67 kind: IntelHintKind
68 binary: str | None # the intel_gpu_top path when installed, else None
71class IntelBackend:
72 """Intel util from xpu-smi, then DRM fdinfo, then intel_gpu_top."""
74 def sample(self, indices: frozenset[int]) -> dict[int, UtilSample]:
75 for source in (_xpu_smi_samples, _fdinfo_samples, _intel_gpu_top_samples):
76 result = source(indices)
77 if result:
78 return result
79 return {}
82def _xpu_smi_output(index: int) -> str:
83 """xpu-smi stats JSON stdout for one device, or "" when it can't run."""
84 return run_smi(_TOOL_XPU_SMI, ["stats", "-d", str(index), "-j"], _TIMEOUT_S)
87def _device_level(data: object) -> list[object]:
88 """Return the device_level metric array from parsed stats JSON, or []."""
89 # stats -d <id> -j emits a single device object with a "device_level" array.
90 # Tolerate a bare list or a {"device_list": [...]} wrapper defensively.
91 if isinstance(data, dict):
92 top_level = data.get("device_level")
93 if isinstance(top_level, list):
94 return top_level
95 wrapped = data.get("device_list")
96 if isinstance(wrapped, list) and wrapped and isinstance(wrapped[0], dict):
97 inner = wrapped[0].get("device_level")
98 return inner if isinstance(inner, list) else []
99 if isinstance(data, list) and data and isinstance(data[0], dict):
100 inner = data[0].get("device_level")
101 return inner if isinstance(inner, list) else []
102 return []
105def _metric_int(entries: list[object], metrics_type: str) -> int | None:
106 """First device_level entry with this metrics_type, coerced to int, or None."""
107 for entry in entries:
108 if isinstance(entry, dict) and entry.get("metrics_type") == metrics_type:
109 return extract_int(entry, ("value",))
110 return None
113def _parse_xpu_smi(raw: str, index: int) -> UtilSample | None:
114 """Parse one device's xpu-smi stats JSON into a UtilSample, or None on failure."""
115 if not raw:
116 return None
117 try:
118 data = json.loads(raw)
119 except json.JSONDecodeError:
120 return None
121 entries = _device_level(data)
122 if not entries:
123 return None
124 return UtilSample(
125 index=index,
126 utilization_pct=_metric_int(entries, _METRIC_UTIL),
127 temperature_c=_metric_int(entries, _METRIC_TEMP),
128 # stats reports memory-used but not total; leave the 0/0 sentinel so the
129 # orchestrator keeps structural VRAM.
130 free_bytes=0,
131 total_bytes=0,
132 )
135def _xpu_smi_samples(indices: frozenset[int]) -> dict[int, UtilSample]:
136 samples: dict[int, UtilSample] = {}
137 for index in sorted(indices):
138 sample = _parse_xpu_smi(_xpu_smi_output(index), index)
139 if sample is not None:
140 samples[index] = sample
141 return samples
144def _fdinfo_samples(indices: frozenset[int]) -> dict[int, UtilSample]:
145 if not indices:
146 return {}
147 util = fdinfo.read_drm_util(_I915)
148 if util is None:
149 return {}
150 return _single(indices, util)
153def _intel_gpu_top_output() -> str:
154 """intel_gpu_top -J partial stdout over one short window, or "" on failure.
156 intel_gpu_top streams until killed, so it always hits the timeout on success;
157 a permission failure exits fast with empty stdout. Both yield the right thing.
158 """
159 binary = shutil.which(_TOOL_IGT)
160 if binary is None:
161 return ""
162 try:
163 proc = subprocess.run( # noqa: S603 - fixed args, resolved binary
164 [binary, "-J", "-s", str(_IGT_SAMPLE_MS)],
165 capture_output=True,
166 text=True,
167 encoding="utf-8",
168 errors="replace",
169 timeout=_IGT_CAPTURE_S,
170 check=False,
171 )
172 except subprocess.TimeoutExpired as exc:
173 out = exc.stdout
174 if out is None:
175 return ""
176 return out if isinstance(out, str) else out.decode(errors="replace")
177 except (OSError, subprocess.SubprocessError):
178 return ""
179 return proc.stdout
182def _last_json_object(raw: str) -> dict[str, object] | None:
183 """Parse the last complete sample from intel_gpu_top's streamed JSON array."""
184 raw = raw.strip()
185 if not raw:
186 return None
187 # The stream is an unclosed '[ {..}, {..},' -- close it and take the last item.
188 if raw.startswith("[") and not raw.endswith("]"):
189 raw = raw.rstrip().rstrip(",") + "]"
190 try:
191 data = json.loads(raw)
192 except json.JSONDecodeError:
193 return None
194 if isinstance(data, list):
195 last = data[-1] if data else None
196 return last if isinstance(last, dict) else None
197 return data if isinstance(data, dict) else None
200def _igt_max_busy(raw: str) -> int | None:
201 """Busiest engine's busy percent from an intel_gpu_top sample, or None."""
202 obj = _last_json_object(raw)
203 if obj is None:
204 return None
205 engines = obj.get("engines")
206 if not isinstance(engines, dict):
207 return None
208 busies = [
209 float(eng["busy"])
210 for eng in engines.values()
211 if isinstance(eng, dict) and isinstance(eng.get("busy"), (int, float))
212 ]
213 if not busies:
214 return None
215 return round(max(busies))
218def _intel_gpu_top_samples(indices: frozenset[int]) -> dict[int, UtilSample]:
219 if not indices:
220 return {}
221 util = _igt_max_busy(_intel_gpu_top_output())
222 if util is None:
223 return {}
224 return _single(indices, util)
227def _single(indices: frozenset[int], util: int) -> dict[int, UtilSample]:
228 """One device-level reading keyed to the lowest requested index (iGPU case)."""
229 idx = min(indices)
230 return {
231 idx: UtilSample(idx, utilization_pct=util, temperature_c=None, free_bytes=0, total_bytes=0)
232 }
235@functools.cache
236def intel_util_hint() -> IntelUtilHint | None:
237 """The fix that would make Intel GPU util readable, or None when nothing is blocked.
239 INSTALL when intel_gpu_top is absent (igt-gpu-tools reads the i915 PMU on
240 kernels too old to publish fdinfo engine counters), GRANT when it is
241 installed but permission-blocked, None when the tool already works. Cached:
242 the install/permission state does not change within a run.
243 """
244 binary = shutil.which(_TOOL_IGT)
245 if binary is None:
246 return IntelUtilHint(IntelHintKind.INSTALL, None)
247 if _igt_permission_denied(binary):
248 return IntelUtilHint(IntelHintKind.GRANT, binary)
249 return None
252def _igt_permission_denied(binary: str) -> bool:
253 """True when intel_gpu_top reports the i915 PMU is permission-blocked."""
254 try:
255 proc = subprocess.run( # noqa: S603 - resolved binary, fixed args
256 [binary, "-J", "-s", str(_IGT_SAMPLE_MS)],
257 capture_output=True,
258 text=True,
259 encoding="utf-8",
260 errors="replace",
261 timeout=_IGT_PERM_CHECK_S,
262 check=False,
263 )
264 except subprocess.TimeoutExpired:
265 return False # it streamed rather than erroring out, so the PMU is usable
266 except (OSError, subprocess.SubprocessError):
267 return False
268 return "CAP_PERFMON" in proc.stderr or "Permission denied" in proc.stderr