File size: 6,098 Bytes
45507bb | 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 | """Inference performance: latency percentiles, max throughput, paced-rate CPU/RAM cost.
Latency is end-to-end `decide()` (state encoding + tokenisation + model), the
number an embedding application actually pays. Max throughput is measured
unthrottled; the paced tests (1..1000 decisions/s) show what a given decision
rate costs in CPU, which is what matters when MiniJEV lives inside other software.
"""
from __future__ import annotations
import os
import time
from pathlib import Path
import numpy as np
import psutil
def _pct(xs: list[float]) -> dict:
a = np.array(xs) * 1000
return {"p50_ms": round(float(np.percentile(a, 50)), 4), "p95_ms": round(float(np.percentile(a, 95)), 4),
"p99_ms": round(float(np.percentile(a, 99)), 4), "mean_ms": round(float(a.mean()), 4)}
def latency(decider, states, question, choices, n: int = 2000, warmup: int = 100) -> dict:
for i in range(warmup):
decider.decide(states[i % len(states)], question, choices)
xs = []
for i in range(n):
t = time.perf_counter()
decider.decide(states[i % len(states)], question, choices)
xs.append(time.perf_counter() - t)
# model-only latency (pre-encoded input), to separate pre-processing cost
enc = [decider.encode([s]) for s in states[:64]]
ys = []
for i in range(n):
t = time.perf_counter()
decider.probs(None, question, choices, encoded=enc[i % len(enc)])
ys.append(time.perf_counter() - t)
return {"end_to_end": _pct(xs), "model_only": _pct(ys),
"max_decisions_per_sec_batch1": round(1.0 / float(np.mean(xs)), 1)}
def batched_throughput(decider, states, question, choices, batch: int = 64, seconds: float = 3.0) -> dict:
enc = decider.encode(states[:batch])
decider.probs(None, question, choices, encoded=enc)
n, t0 = 0, time.perf_counter()
while time.perf_counter() - t0 < seconds:
decider.probs(None, question, choices, encoded=enc)
n += len(enc)
return {"batch": len(enc), "decisions_per_sec": round(n / (time.perf_counter() - t0), 1)}
def paced(decider, states, question, choices, rate: float, seconds: float = 3.0) -> dict:
"""Issue decisions at a fixed rate; measure achieved rate, latency and process CPU."""
proc = psutil.Process(os.getpid())
period = 1.0 / rate
n_target = max(3, int(rate * seconds))
lat = []
proc.cpu_percent(None)
cpu_t0 = proc.cpu_times()
t0 = time.perf_counter()
next_t = t0
for i in range(n_target):
now = time.perf_counter()
if next_t > now:
time.sleep(next_t - now)
t = time.perf_counter()
decider.decide(states[i % len(states)], question, choices)
lat.append(time.perf_counter() - t)
next_t += period
wall = max(time.perf_counter() - t0, n_target * period) # a paced run lasts n_target periods
cpu_t1 = proc.cpu_times()
cpu_s = (cpu_t1.user - cpu_t0.user) + (cpu_t1.system - cpu_t0.system)
achieved = n_target / wall
return {"target_per_sec": rate, "achieved_per_sec": round(achieved, 1),
"sustained": achieved >= 0.95 * rate, **_pct(lat),
"cpu_percent_of_one_core": round(100 * cpu_s / wall, 1),
"rss_mb": round(proc.memory_info().rss / 2**20, 1)}
def benchmark_decider(path: str | Path, task=None, threads: int = 1, rates=(1, 10, 100, 500, 1000),
n_latency: int = 2000, quick: bool = False) -> dict:
from minijev.inference.runtime import MiniJEV
proc = psutil.Process(os.getpid())
rss0 = proc.memory_info().rss
dec = MiniJEV.load(path, threads=threads)
rss1 = proc.memory_info().rss
task = task or dec.task
states = task.example_states(256, seed=3)
q, ch = task.default_question, list(task.default_choices)
res = {"model": str(path), "file_size_mb": round(Path(path).stat().st_size / 2**20, 3), "threads": threads,
"model_load_rss_mb": round((rss1 - rss0) / 2**20, 1)}
if hasattr(dec, "model"):
res["params"] = dec.model.num_parameters()
elif hasattr(dec, "meta"):
res["params"] = dec.meta.get("params")
res["latency"] = latency(dec, states, q, ch, n=200 if quick else n_latency)
res["batched"] = batched_throughput(dec, states, q, ch, seconds=1.0 if quick else 3.0)
max_rate = res["latency"]["max_decisions_per_sec_batch1"]
res["paced"] = []
for r in rates:
secs = 2.0 if quick else (5.0 if r <= 10 else 3.0)
if r > 1.5 * max_rate:
res["paced"].append({"target_per_sec": r, "achieved_per_sec": None, "sustained": False,
"note": f"above measured max ({max_rate}/s)"})
continue
if r == 1 and quick:
secs = 3.0
res["paced"].append(paced(dec, states, q, ch, r, secs))
res["process_rss_mb"] = round(proc.memory_info().rss / 2**20, 1)
return res
def format_perf(res: dict) -> str:
L = res["latency"]
lines = [f"{res['model']} ({res['file_size_mb']} MB, params={res.get('params')}, threads={res['threads']})",
f" latency end-to-end p50 {L['end_to_end']['p50_ms']:.3f} ms p95 {L['end_to_end']['p95_ms']:.3f} ms "
f"p99 {L['end_to_end']['p99_ms']:.3f} ms",
f" latency model-only p50 {L['model_only']['p50_ms']:.3f} ms p95 {L['model_only']['p95_ms']:.3f} ms "
f"p99 {L['model_only']['p99_ms']:.3f} ms",
f" max decisions/s: batch=1 {L['max_decisions_per_sec_batch1']:,.0f} "
f"batch={res['batched']['batch']} {res['batched']['decisions_per_sec']:,.0f}",
" paced: target achieved p50ms p99ms cpu%(1 core) rss MB"]
for p in res["paced"]:
if p.get("achieved_per_sec") is None:
lines.append(f" {p['target_per_sec']:>6} -- {p['note']}")
else:
lines.append(f" {p['target_per_sec']:>6} {p['achieved_per_sec']:>8} {p['p50_ms']:>6.3f} "
f"{p['p99_ms']:>6.3f} {p['cpu_percent_of_one_core']:>12} {p['rss_mb']:>6}")
return "\n".join(lines)
|