Coverage for orchestrant/benchmark/energy.py: 59%
234 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-18 00:48 +0000
1"""Measured energy per request, from the Windows Energy Meter Interface (EMI)."""
3from __future__ import annotations
5import sys
6import threading
7import time
8from typing import Any
11PWH_TO_J = 3.6e-9 # one picowatt-hour in joules
12EPOCH_1601_MS = 11_644_473_600_000 # 1601-01-01 -> 1970-01-01, in ms
13ENERGY_COUNTER = "\\Energy Meter(*)\\Energy"
14TIME_COUNTER = "\\Energy Meter(*)\\Time"
15SCOPE = (
16 "EMI CPU-cluster rails only. The NPU and GPU have no rail on this host, "
17 "so an NPU lane's own draw is NOT included."
18)
21class _Pdh:
22 """The few PDH calls needed, via ctypes. Windows only."""
24 PDH_FMT_LARGE = 0x00000400
25 PDH_MORE_DATA = 0x800007D2
26 # Declared here: a Linux-targeting checker sees __init__'s assignments as unreachable.
27 _dll: Any
28 _Item: Any
30 def __init__(self):
31 import ctypes
32 from ctypes import wintypes
34 self._ct = ctypes
35 self._wt = wintypes
36 if sys.platform != "win32": # also tells the type checker WinDLL exists
37 raise OSError("PDH is Windows-only")
38 self._dll = ctypes.WinDLL("pdh.dll")
40 class _Value(ctypes.Structure):
41 class _U(ctypes.Union):
42 _fields_ = [
43 ("longValue", ctypes.c_long),
44 ("doubleValue", ctypes.c_double),
45 ("largeValue", ctypes.c_longlong),
46 ]
48 _anonymous_ = ("u",)
49 _fields_ = [("CStatus", wintypes.DWORD), ("u", _U)]
51 class _Item(ctypes.Structure):
52 _fields_ = [("szName", wintypes.LPWSTR), ("FmtValue", _Value)]
54 self._Item = _Item
56 def open(self, *paths):
57 """One query holding every counter, so energy and its timestamp are read together."""
58 ct, wt = self._ct, self._wt
59 query = wt.HANDLE()
60 if self._dll.PdhOpenQueryW(None, 0, ct.byref(query)) != 0:
61 raise OSError("PdhOpenQueryW failed")
62 counters = []
63 for path in paths:
64 counter = wt.HANDLE()
65 # English names: a localized Windows calls the counter "Energiemessung".
66 rc = self._dll.PdhAddEnglishCounterW(query, path, 0, ct.byref(counter))
67 if rc != 0:
68 self._dll.PdhCloseQuery(query)
69 raise OSError(f"{path} is not available (PDH 0x{rc & 0xFFFFFFFF:08X})")
70 counters.append(counter)
71 return query, counters
73 def read(self, query, counters):
74 """[{instance: value}, ...], one dict per counter, from ONE collection."""
75 if self._dll.PdhCollectQueryData(query) != 0:
76 raise OSError("PdhCollectQueryData failed")
77 return [self._values(counter) for counter in counters]
79 def _values(self, counter):
80 ct, wt = self._ct, self._wt
81 size, count = wt.DWORD(0), wt.DWORD(0)
82 rc = self._dll.PdhGetFormattedCounterArrayW(
83 counter, self.PDH_FMT_LARGE, ct.byref(size), ct.byref(count), None
84 )
85 if (rc & 0xFFFFFFFF) != self.PDH_MORE_DATA:
86 raise OSError(
87 f"PdhGetFormattedCounterArrayW sizing: 0x{rc & 0xFFFFFFFF:08X}"
88 )
89 buf = (ct.c_byte * size.value)()
90 rc = self._dll.PdhGetFormattedCounterArrayW(
91 counter, self.PDH_FMT_LARGE, ct.byref(size), ct.byref(count), buf
92 )
93 if rc != 0:
94 raise OSError(f"PdhGetFormattedCounterArrayW: 0x{rc & 0xFFFFFFFF:08X}")
95 items = ct.cast(buf, ct.POINTER(self._Item))
96 return {
97 items[i].szName: items[i].FmtValue.largeValue for i in range(count.value)
98 }
100 def close(self, query):
101 self._dll.PdhCloseQuery(query)
104def _interpolate(samples, rail, t):
105 """Cumulative pWh of `rail` at unix time `t`, linear between samples."""
106 before = after = None
107 for ts, values in samples:
108 if ts <= t:
109 before = (ts, values[rail])
110 elif after is None:
111 after = (ts, values[rail])
112 break
113 if before is None or after is None:
114 return None
115 (t0, e0), (t1, e1) = before, after
116 return e0 + (e1 - e0) * ((t - t0) / (t1 - t0)) if t1 > t0 else e0
119class EnergyMeter:
120 """Background sampler over the EMI rails; wrap a run, then ask energy_between(t0, t1)."""
122 def __init__(self, poll_s=0.25, enabled=True):
123 self.poll_s = poll_s
124 self.available = False
125 self.reason = None
126 self.rails = []
127 self._samples = [] # (unix_s, {rail: pWh}), one per PUBLISHED update
128 self._lock = threading.Lock()
129 self._stop = threading.Event()
130 self._thread = None
131 self._pdh = None
132 self._query = None
133 self.baselines = [] # every idle_power() result, in order
134 self.resets = 0 # counter resets seen; a request spanning one is unmetered
135 if not enabled:
136 self.reason = "disabled (--no-energy)"
137 return
138 if sys.platform != "win32":
139 self.reason = "not Windows: no Energy Meter counters to read"
140 return
141 try:
142 self._pdh = _Pdh()
143 self._query = self._pdh.open(ENERGY_COUNTER, TIME_COUNTER)
144 energy, stamp = self._read()
145 except Exception as e: # counters absent or PDH refused
146 self.reason = f"Energy Meter unavailable: {e}"
147 return
148 # A rail that never accumulated anything is not metered here; _Total is the sum.
149 self.rails = sorted(r for r, v in energy.items() if v and r != "_Total")
150 if not self.rails:
151 self.reason = "Energy Meter present but no rail carries data"
152 return
153 self.available = True
154 self._append(energy, stamp)
156 def _read(self):
157 if self._pdh is None or self._query is None:
158 raise OSError("the meter is not open")
159 return self._pdh.read(*self._query)
161 def _append(self, energy, stamp):
162 ms = max(stamp.get(r, 0) for r in self.rails)
163 unix_s = (ms - EPOCH_1601_MS) / 1000.0
164 values = {r: energy.get(r, 0) for r in self.rails}
165 with self._lock:
166 if self._samples:
167 last_s, last = self._samples[-1]
168 if unix_s <= last_s:
169 return # the meter has not published since the last poll
170 # Rails that all stood still under a new stamp are a torn read.
171 if all(values[r] == last[r] for r in self.rails):
172 return
173 # A counter that went backwards was reset: restart the curve, keep later samples.
174 if any(values[r] < last[r] for r in self.rails):
175 self._samples.clear()
176 self.resets += 1
177 self._samples.append((unix_s, values))
179 def _run(self):
180 while not self._stop.wait(self.poll_s):
181 try:
182 self._append(*self._read())
183 except Exception: # nosec B112 -- a missed poll only widens the interpolation
184 continue
186 def start(self):
187 if self.available and self._thread is None:
188 self._thread = threading.Thread(target=self._run, name="emi", daemon=True)
189 self._thread.start()
190 return self
192 def stop(self):
193 if self._thread is not None:
194 self._stop.set()
195 self._thread.join(timeout=2)
196 self._thread = None
197 if self._query and self._pdh is not None:
198 self._pdh.close(self._query[0])
199 self._query = None
201 def __enter__(self):
202 return self.start()
204 def __exit__(self, *exc):
205 self.stop()
206 return False
208 def wait_until(self, t, timeout=3.0):
209 """Block until the meter has published a sample at or after `t`."""
210 deadline = time.monotonic() + timeout
211 while time.monotonic() < deadline:
212 with self._lock:
213 if self._samples and self._samples[-1][0] >= t:
214 return True
215 time.sleep(self.poll_s)
216 return False
218 def energy_between(self, t0, t1):
219 """Joules per rail over [t0, t1] (unix seconds), or None; waits for the covering sample."""
220 if not self.available or t1 <= t0:
221 return None
222 self.wait_until(t1)
223 with self._lock:
224 samples = list(self._samples)
225 joules = {}
226 for rail in self.rails:
227 e0, e1 = _interpolate(samples, rail, t0), _interpolate(samples, rail, t1)
228 if e0 is None or e1 is None:
229 return None
230 joules[rail] = round((e1 - e0) * PWH_TO_J, 3)
231 return joules
233 def idle_power(self, seconds=5.0):
234 """Mean CPU-rail watts over the next `seconds`, kept in `baselines` to show drift."""
235 if not self.available or seconds <= 0:
236 return None
237 t0 = time.time()
238 time.sleep(seconds)
239 joules = self.energy_between(t0, time.time())
240 if joules is None:
241 return None
242 watts = round(sum(joules.values()) / seconds, 3)
243 self.baselines.append(watts)
244 return watts
247# Baselines further apart than this make the run's net figures doubtful.
248DRIFT_LIMIT_W = 0.2
251def energy_block(meter):
252 """What a report says about the meter, with its idle baselines and their drift."""
253 block = {
254 "available": meter.available,
255 "reason": meter.reason,
256 "rails": meter.rails,
257 "scope": SCOPE,
258 }
259 if meter.baselines:
260 drift = round(max(meter.baselines) - min(meter.baselines), 3)
261 block["idle_w"] = meter.baselines
262 block["idle_w_used"] = round(sum(meter.baselines) / len(meter.baselines), 3)
263 block["idle_drift_w"] = drift
264 # One baseline cannot show drift: unknown, not reliable.
265 block["net_reliable"] = (
266 drift <= DRIFT_LIMIT_W if len(meter.baselines) > 1 else None
267 )
268 if getattr(meter, "resets", 0):
269 block["counter_resets"] = meter.resets
270 return block
273def energy_lines(meter):
274 """The idle-baseline line under the speed table: what net was taken against."""
275 block = energy_block(meter) if meter is not None else {}
276 if "idle_w" not in block:
277 return []
278 watts = " then ".join(f"{w:.2f}" for w in block["idle_w"])
279 line = f" Idle baseline: {watts} W (net uses {block['idle_w_used']:.2f})"
280 if block["net_reliable"] is False:
281 line += (
282 f" -- DRIFTED {block['idle_drift_w']:.2f} W: the net figures are not "
283 "reliable, read gross"
284 )
285 return [line]
288def renet(results, meter):
289 """Re-net every metered row against the MEAN of the run's idle baselines; return it."""
290 if meter is None or not meter.baselines:
291 return None
292 idle_w = round(sum(meter.baselines) / len(meter.baselines), 3)
293 for row in results:
294 gross, window = row.get("cpu_rail_energy_j"), row.get("cpu_rail_window_s")
295 if gross is None or not window:
296 continue
297 net = max(0.0, gross - idle_w * window)
298 row["cpu_rail_idle_w"] = idle_w
299 row["cpu_rail_net_energy_j"] = round(net, 3)
300 if row.get("completion_tokens"):
301 row["cpu_rail_net_j_per_token"] = round(net / row["completion_tokens"], 4)
302 return idle_w
305def request_energy(meter, t0, t1, completion_tokens, idle_w=None):
306 """The per-request energy fields a result row carries, or {} when unmetered."""
307 if meter is None or not meter.available:
308 return {}
309 joules = meter.energy_between(t0, t1)
310 if joules is None:
311 return {}
312 total = round(sum(joules.values()), 3)
313 fields = {
314 "cpu_rail_energy_j": total,
315 "cpu_rail_window_s": round(t1 - t0, 3),
316 "cpu_rail_avg_w": round(total / (t1 - t0), 3),
317 "cpu_rail_energy_by_rail_j": joules,
318 }
319 if completion_tokens:
320 fields["cpu_rail_j_per_token"] = round(total / completion_tokens, 4)
321 if idle_w is not None:
322 net = max(0.0, total - idle_w * (t1 - t0))
323 fields["cpu_rail_net_energy_j"] = round(net, 3)
324 if completion_tokens:
325 fields["cpu_rail_net_j_per_token"] = round(net / completion_tokens, 4)
326 return fields