Coverage for orchestrant/benchmark/hostload.py: 94%
276 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 11:22 +0000
1"""Who burned CPU during a request, measured over the request itself."""
3from __future__ import annotations
5import time
6from urllib.parse import urlsplit
8from orchestrant.benchmark import winhost
11LOOPBACK = {"127.0.0.1", "localhost", "::1"}
14def _psutil():
15 try:
16 import psutil
18 return psutil
19 except Exception:
20 return None
23def _port(base_url):
24 parts = urlsplit(base_url or "")
25 if parts.hostname not in LOOPBACK:
26 return None
27 return parts.port or (443 if parts.scheme == "https" else 80)
30class LaneProcess:
31 """The local process serving `base_url`, if it runs on this host.
33 `listens_here` is None when the lookup never ran; a hidden-pid listener is still local.
34 """
36 def __init__(self, base_url):
37 self.base_url = base_url
38 self.proc = None
39 self.reason = None
40 self.listens_here = None
41 ps = _psutil()
42 port = _port(base_url)
43 if ps is None:
44 self.reason = "psutil not installed"
45 return
46 if port is None:
47 self.reason = "endpoint is not on this host"
48 return
49 try:
50 self.listens_here, self.proc = self._find_listener(ps, port)
51 except Exception as e: # access denied on some hosts
52 self.listens_here = None
53 self.reason = f"listener lookup failed: {type(e).__name__}"
54 return
55 if self.listens_here and self.proc is None:
56 self.reason = (
57 f"a process listens on port {port} here, but its pid is hidden"
58 )
59 elif self.proc is None:
60 self.reason = f"no local process listens on port {port} (WSL2 cannot see the Windows host's)"
62 @staticmethod
63 def _find_listener(ps, port):
64 listens = False
65 for conn in ps.net_connections(kind="tcp"):
66 listening = conn.status == ps.CONN_LISTEN and conn.laddr
67 if listening and conn.laddr.port == port:
68 listens = True
69 if conn.pid:
70 return True, ps.Process(conn.pid)
71 return listens, None
73 @property
74 def available(self):
75 return self.proc is not None
77 def info(self):
78 """pid, exe and command line: what exactly is serving, with which flags."""
79 if self.proc is None:
80 return None
81 try:
82 with self.proc.oneshot():
83 return {
84 "pid": self.proc.pid,
85 "exe": self.proc.exe(),
86 "cmdline": self.proc.cmdline(),
87 "started": self.proc.create_time(),
88 }
89 except Exception as e:
90 return {"pid": self.proc.pid, "error": f"{type(e).__name__}: {e}"[:160]}
92 def cpu_seconds(self):
93 """user+system CPU-seconds of the lane and its children; None once the lane is gone."""
94 if self.proc is None:
95 return None
96 try:
97 root = self.proc.cpu_times()
98 children = self.proc.children(recursive=True)
99 except Exception: # the lane exited or was restarted
100 return None
101 total = root.user + root.system
102 for p in children:
103 try:
104 t = p.cpu_times()
105 total += t.user + t.system
106 except Exception: # nosec B112 -- a child that exited mid-read
107 continue
108 return total
111def open_meters(base_url, energy=True):
112 """The lane handle and (unless disabled) a started energy meter, announced."""
113 from orchestrant.benchmark.energy import EnergyMeter
115 lane = LaneProcess(base_url)
116 # This host's rails measure this host, not a lane elsewhere.
117 meter = EnergyMeter(enabled=energy and lane.available).start()
118 if energy and not lane.available:
119 meter.reason = f"not metered: {lane.reason}"
120 print(
121 f" Lane process: {'pid ' + str(lane.proc.pid) if lane.proc is not None else lane.reason}"
122 )
123 rails = f"EMI rails {', '.join(meter.rails)}" if meter.available else meter.reason
124 print(f" Energy: {rails}\n")
125 return lane, meter
128_GPU_KEYS = (
129 ("gpu_utilization_percent", 1),
130 ("gpu_memory_used_gb", 2),
131 ("gpu_power_watts", 1),
132)
135def avg_optional(before, after, key):
136 """Average a sampler metric that may be absent entirely (no GPU): None, not 0."""
137 if key not in before or key not in after:
138 return None
139 return (before[key] + after[key]) / 2
142class RequestBracket:
143 """Everything measured AROUND one request: start(), stop(), then fields(completion_tokens)."""
145 def __init__(self, sample, top, lane=None, meter=None, idle_w=None):
146 self._sample, self._top = sample, top
147 self._lane, self._meter, self._idle_w = lane, meter, idle_w
148 ps = _psutil()
149 self._ps_cores = ps.cpu_count() if ps else None
151 def start(self):
152 self._before = self._sample()
153 self._top() # prime the per-process counters
154 self._window = Window(self._lane).start()
155 self._wall0 = time.time()
156 return self
158 def stop(self):
159 self._wall1 = time.time()
160 self._load = self._window.stop()
161 self._after = self._sample()
162 self._busiest = self._top() # who actually did the work
163 return self
165 def fields(self, completion_tokens):
166 from orchestrant.benchmark.energy import request_energy
168 before, after, load = self._before, self._after, self._load
169 # Integrate only when the lane runs here; otherwise the counters describe another machine.
170 windowed = self._lane is not None and self._lane.available
171 if windowed and "cpu_busy_percent_window" in load:
172 cpu, method = load["cpu_busy_percent_window"], "window"
173 else:
174 cpu = (before["cpu_percent"] + after["cpu_percent"]) / 2
175 method = "before/after snapshots"
176 out = {
177 "cpu_percent": round(cpu, 1),
178 "cpu_percent_method": method,
179 "ram_used_gb": round((before["ram_used_gb"] + after["ram_used_gb"]) / 2, 2),
180 "top_processes": self._busiest,
181 }
182 # GPU fields only when the local probe answered for both samples.
183 for key, digits in _GPU_KEYS:
184 value = avg_optional(before, after, key)
185 if value is not None:
186 out[key] = round(value, digits)
187 out.update({k: v for k, v in load.items() if k != "cpu_busy_percent_window"})
188 # Everything else the machine did: llama.cpp spreads over every core, so each one costs.
189 if method == "window" and "lane_cores" in load and self._ps_cores:
190 other = cpu / 100.0 * self._ps_cores - load["lane_cores"]
191 out["other_cores"] = round(max(0.0, other), 2)
192 out.update(
193 request_energy(
194 self._meter, self._wall0, self._wall1, completion_tokens, self._idle_w
195 )
196 )
197 if self._idle_w is not None:
198 out["cpu_rail_idle_w"] = self._idle_w
199 return out
202def _window_s(row):
203 """Seconds a row's joules cover; older rows fall back to the request's latency."""
204 return row.get("cpu_rail_window_s") or row.get("latency_s") or 0
207def _mean(values):
208 return sum(values) / len(values)
211def summary_lines(results):
212 """The who-burned-what lines of the speed table's summary."""
213 lines = []
214 # The serving process and the inference worker can be different PIDs.
215 busiest = {}
216 for r in results:
217 for proc in r.get("top_processes") or []:
218 key = f"{proc['name']} (pid {proc['pid']})"
219 busiest[key] = max(busiest.get(key, 0), proc["cpu_percent"])
220 if busiest:
221 top = sorted(busiest.items(), key=lambda kv: kv[1], reverse=True)[:2]
222 lines.append(
223 " Busiest proc: " + ", ".join(f"{n} {p:.0f}%" for n, p in top)
224 )
225 cores = [r["lane_cores"] for r in results if r.get("lane_cores") is not None]
226 if cores:
227 lines.append(
228 f" Lane CPU: {_mean(cores):.2f} cores avg (the serving process tree, over each request)"
229 )
230 other = [r["other_cores"] for r in results if r.get("other_cores") is not None]
231 if other:
232 lines.append(
233 f" Other load: {_mean(other):.2f} cores avg, max {max(other):.2f} "
234 f"(not the lane -- a CPU lane loses throughput to every one)"
235 )
236 # A ratio of sums: a short answer's meter noise must not outweigh a long one's.
237 metered = [
238 r
239 for r in results
240 if r.get("cpu_rail_energy_j") is not None and r.get("completion_tokens")
241 ]
242 if metered:
243 tokens = sum(r["completion_tokens"] for r in metered)
244 joules = sum(r["cpu_rail_energy_j"] for r in metered)
245 seconds = sum(map(_window_s, metered))
246 line = f" CPU-rail energy: {joules / tokens:.3f} J/token gross"
247 if all("cpu_rail_net_energy_j" in r for r in metered):
248 net = sum(r["cpu_rail_net_energy_j"] for r in metered)
249 line += f", {net / tokens:.3f} net of idle"
250 line += f" ({joules / seconds:.1f} W avg)" if seconds else ""
251 lines.append(line + " -- CPU clusters only; NPU/GPU draw is not metered")
252 return lines
255class Window:
256 """CPU accounting over one request: start() before sending, stop() after."""
258 def __init__(self, lane=None):
259 self.lane = lane
260 self._ps = _psutil()
261 self._t0 = 0.0
262 self._sys0 = self._lane0 = None
263 self.wall = 0.0
264 self.result = {}
266 def start(self):
267 self._t0 = time.monotonic()
268 self._sys0 = self._ps.cpu_times() if self._ps else None
269 self._lane0 = self.lane.cpu_seconds() if self.lane else None
270 return self
272 def elapsed(self):
273 return time.monotonic() - self._t0
275 def stop(self):
276 wall = time.monotonic() - self._t0
277 self.wall = wall
278 out = {}
279 if self._ps is not None and self._sys0 is not None:
280 sys1 = self._ps.cpu_times()
281 busy = sum(sys1) - sys1.idle - (sum(self._sys0) - self._sys0.idle)
282 total = sum(sys1) - sum(self._sys0)
283 if total > 0:
284 out["cpu_busy_percent_window"] = round(100.0 * busy / total, 1)
285 if self._lane0 is not None and self.lane is not None:
286 lane1 = self.lane.cpu_seconds()
287 if lane1 is not None and wall > 0:
288 used = max(0.0, lane1 - self._lane0)
289 out["lane_cpu_s"] = round(used, 2)
290 out["lane_cores"] = round(used / wall, 2)
291 self.result = out
292 return out
295# Past one other core a CPU-lane number is about half its quiet rate.
296BUSY_HOST_CORES = 1.0
298# The spread of two quiet runs; further apart, the load differed, not just the moment.
299LOAD_DIFF_CORES = 0.3
302_LANE_UNREADABLE = "the lane's CPU time could not be read (it exited or restarted)"
304# `via` in a snapshot: which host its busy_cores describe.
305VIA_LOCAL = "local"
306VIA_WSL_INTEROP = "wsl-interop"
309def _busy_cores(percent, cpus):
310 return round(percent / 100.0 * cpus, 2) if percent is not None and cpus else None
313def _snapshot_note(lane, busy, lane_cores, ps):
314 """Why `other_cores` is what it is, or why it is missing."""
315 if ps is None:
316 return "psutil not installed"
317 if busy is None:
318 return "no CPU time elapsed in the window"
319 if lane is None:
320 return "no lane named: every busy core counts as other load"
321 if not lane.available:
322 # From WSL2 the counters are the VM's; a remote lane shares no cores with this host.
323 return f"the lane's host is not visible from here: {lane.reason}"
324 if lane_cores is None:
325 return _LANE_UNREADABLE
326 return None
329def _interop_port(lane):
330 """The lane's port when only the Windows host can be serving it, else None."""
331 if (
332 lane is None
333 or lane.available
334 or getattr(lane, "listens_here", None) is not False
335 ):
336 return None
337 port = _port(getattr(lane, "base_url", None))
338 return port if port is not None and winhost.in_wsl() else None
341def _windows_host(seconds, lane):
342 """winhost.measure() for a lane only Windows sees; {"error"} if it failed; else None."""
343 port = _interop_port(lane)
344 if port is None:
345 return None
346 try:
347 return winhost.measure(port, seconds)
348 except winhost.InteropError as e:
349 return {"error": str(e)}
350 except Exception as e: # a load reading must never cost a run
351 return {"error": f"{type(e).__name__}: {e}"[:160]}
354def _interop_snapshot(reading):
355 """A Windows-host reading on the scale of a local one; no listener leaves cores None."""
356 load, cpus = reading["load"], reading["cpus"]
357 busy = _busy_cores(load.get("cpu_busy_percent_window"), cpus)
358 lane_cores = load.get("lane_cores")
359 note = None
360 if reading["pid"] is None:
361 note = (
362 f"no process listens on port {reading['port']} on the Windows host either"
363 )
364 elif lane_cores is None:
365 note = _LANE_UNREADABLE
366 return {
367 "busy_cores": busy,
368 "lane_cores": lane_cores,
369 "other_cores": None if note else round(max(0.0, busy - lane_cores), 2),
370 "seconds": round(reading["wall"], 2),
371 "cpus": cpus,
372 "note": note,
373 "via": VIA_WSL_INTEROP,
374 }
377def load_snapshot(seconds=3, lane=None):
378 """How busy the lane's host was just before a run, net of the lane; never raises.
380 `lane` is a LaneProcess, a base URL, or None.
381 """
382 if isinstance(lane, str):
383 lane = LaneProcess(lane)
384 ps = _psutil()
385 window = Window(lane if lane is not None and lane.available else None).start()
386 host = _windows_host(seconds, lane)
387 # A Windows reading that failed early still leaves the local fallback its full window.
388 time.sleep(max(0.0, seconds - window.elapsed()))
389 load = window.stop()
390 if host is not None and "error" not in host:
391 return _interop_snapshot(host)
392 cpus = ps.cpu_count() if ps else None
393 busy = _busy_cores(load.get("cpu_busy_percent_window"), cpus)
394 lane_cores = load.get("lane_cores")
395 note = _snapshot_note(lane, busy, lane_cores, ps)
396 if host is not None:
397 note = f"{note}; the Windows host through WSL interop: {host['error']}"
398 other = None
399 if note is None:
400 other = round(max(0.0, busy - lane_cores), 2)
401 elif lane is None and busy is not None:
402 other = busy
403 return {
404 "busy_cores": busy,
405 "lane_cores": lane_cores,
406 "other_cores": other,
407 "seconds": round(window.wall, 2),
408 "cpus": cpus,
409 "note": note,
410 "via": VIA_LOCAL,
411 }
414def load_line(snapshot):
415 """The console line announcing a run-start snapshot, with a warning when busy."""
416 snapshot = snapshot or {}
417 other = snapshot.get("other_cores")
418 where = " on the Windows host" if snapshot.get("via") == VIA_WSL_INTEROP else ""
419 if other is None:
420 busy = snapshot.get("busy_cores")
421 seen = f"{busy:.2f} busy cores{where or ' here'}; " if busy is not None else ""
422 return f" Host load: {seen}other load unknown -- {snapshot.get('note')}"
423 line = (
424 f" Host load: {other:.2f} other cores{where} over the "
425 f"{snapshot.get('seconds', 0):.0f} s before the first request"
426 )
427 if snapshot.get("lane_cores") is not None:
428 line += f" (the lane itself: {snapshot['lane_cores']:.2f})"
429 if other > BUSY_HOST_CORES:
430 line += (
431 "\n WARNING: over one core busy with something else. A CPU lane "
432 "measured now\n reads about half its quiet rate; close what you "
433 "can (the IDE, a scan) first."
434 )
435 return line