| |
| |
| """Stage bench of warm frames: the numbers OPT_BASELINE.md, OPT_REPORT.md and the card quote. |
| |
| bin/devrun -t 1800 -- python code/scripts/bench.py --iters 100 --json logs/streampetr/baseline/bench_eth-1cq.json |
| bin/devrun -t 1800 -- python code/scripts/bench.py --dispatch worker --num-cqs 2 --json ... # the D14 matrix |
| |
| Each input is a short frame sequence cycled through ONE stream (the temporal memory is carried from frame to frame, |
| as in steady operation) after ``--warm`` warm frames; the time stamps advance monotonically across cycles: |
| |
| - ``synthetic``: five 1920x1080 camera frames generated by us (seeded smooth colour fields + noise, three seeds) with |
| the calibration preset ``--synthetic-preset`` (default ``pandaset_019``: 1920x1080, from the git-ignored |
| ``staging_samples_pandaset/calib`` while it does not ship; the OPT_BASELINE input), 10 Hz |
| stamps, no ego pose (identity: a static ego); |
| - ``pandaset[=dir]``: ``pandaset_019_f{40..43}.json`` (``samples/``, else the git-ignored |
| ``staging_samples_pandaset/samples``, else ``dir``): PandaSet 019 (CC BY 4.0), 1920x1080 JPEGs, real stamps and |
| ego poses (each cycle shifts the stamps by 0.4 s; the pose jumps back once per cycle); |
| - ``sample=<manifest>[+<manifest>...]``: any sample manifests of ``load_sample``. |
| |
| Images are decoded before timing (Autoware hands the node decoded images), so ``host_pre`` is the request parsing + |
| the C14 triangle resize / normalisation of the five cameras + the per-calibration cache lookup + the stream plan. |
| |
| Two loops per input (``ttaw.profiling.StageBench``; p50 / p99 / mean / min / max in ms): |
| |
| 1. **instrumented**: one ``model(...)`` call split into synchronized stages by wrapping the model's host hooks and |
| the ``TraceRunner`` I/O: ``host_pre``; ``img_rows`` (NCHW float32 -> NHWC rows); ``host_in`` (numpy -> ttnn host |
| tensors: the bf16 image rows, the ages, the changed params); ``h2d`` (enqueue + synchronize); ``trace`` (one |
| replay + synchronize); ``d2h`` (device -> host copy) and ``host_out`` (TILE -> numpy); ``host_post`` (decode of |
| the 6 layers + IoU-BEV NMS + Detections3D); ``e2e_instrumented``. |
| 2. **plain**: ``model(...)`` exactly as a client calls it (no extra synchronization): ``e2e`` and the API's own |
| ``timing_ms`` (``api.preprocess`` ... ``api.total``). |
| |
| Device only, once per process (input independent): ``replay.frame`` one blocking replay; ``b2b.frame`` back-to-back |
| replays per replay (device-bound throughput). AICLK / power / temperature are sampled during the loops |
| (``AiclkSampler``), and the agreement checks (the shipped synthetic sample and the staged PandaSet frame 40, cold |
| starts, against their stored CPU-reference bodies with the container smoke's ``REFERENCE_GATES``) run first, so every dispatch / CQ configuration's numbers come |
| with its accuracy. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import copy |
| import json |
| import os |
| import platform |
| import subprocess |
| import sys |
| import time |
| from pathlib import Path |
| from typing import Any, Dict, List, Tuple |
|
|
| import numpy as np |
|
|
| HERE = Path(__file__).resolve() |
| CODE = HERE.parents[1] |
| sys.path.insert(0, str(CODE)) |
| SAMPLES = CODE / "tt_streampetr" / "samples" |
| ROOT = Path(os.environ.get("TT_MODELS_ROOT", "/home/ubuntu/experiments/tt-models")) |
| CAMS = ("CAM_FRONT", "CAM_FRONT_LEFT", "CAM_BACK_LEFT", "CAM_FRONT_RIGHT", "CAM_BACK_RIGHT") |
|
|
|
|
| |
| def synthetic_image(seed: int, cam: int, hw: Tuple[int, int] = (1080, 1920)) -> np.ndarray: |
| """A deterministic 8-bit RGB camera frame made by us: a smooth colour field (low-frequency sines) + noise.""" |
| rng = np.random.default_rng(1000 * seed + cam) |
| h, w = hw |
| y = np.linspace(0, 1, h, dtype=np.float32)[:, None] |
| x = np.linspace(0, 1, w, dtype=np.float32)[None, :] |
| planes = [] |
| for c in range(3): |
| fx, fy, ph = rng.uniform(1, 6), rng.uniform(1, 6), rng.uniform(0, 6.28) |
| planes.append(127 + 90 * np.sin(6.28 * fx * x + ph) * np.cos(6.28 * fy * y + ph / 2)) |
| img = np.stack(planes, -1) + rng.normal(0, 12, (h, w, 3)).astype(np.float32) |
| return np.clip(img, 0, 255).astype(np.uint8) |
|
|
|
|
| def synthetic_frames(preset: str, n: int = 3) -> List[Dict[str, Any]]: |
| """Seeded colour-field frames at the image size of the calibration preset ``preset`` (the package's ``calib/``, |
| else the git-ignored ``staging_samples_pandaset/calib``; sent inline).""" |
| from tt_streampetr.host.inputs import CALIB_DIR |
|
|
| path = CALIB_DIR / f"{preset}.json" |
| if not path.is_file(): |
| path = CODE.parent / "staging_samples_pandaset" / "calib" / f"{preset}.json" |
| cal = json.loads(path.read_text()) |
| calib = {k: v for k, v in cal.items() if k != "_doc"} |
| w, h = cal.get("image_size", [1920, 1080]) |
| frames = [] |
| for s in range(n): |
| frames.append({"images": [{"camera": c, "image": synthetic_image(s + 1, i, (h, w))} |
| for i, c in enumerate(CAMS)], |
| "calibration": calib, "_dt": 0.1}) |
| return frames |
|
|
|
|
| def pandaset_dir(override: str = "") -> Path: |
| cands = [Path(override)] if override else [] |
| cands += [SAMPLES, CODE.parent / "staging_samples_pandaset" / "samples"] |
| for c in cands: |
| if all((c / f"pandaset_019_f{n}.json").is_file() for n in (40, 41, 42, 43)): |
| return c |
| return cands[0] |
|
|
|
|
| def decoded(req: Dict[str, Any]) -> Dict[str, Any]: |
| from tt_streampetr.ttaw import io as tio |
|
|
| out = dict(req) |
| out["images"] = [{**{k: v for k, v in e.items() if k != "path"}, "image": tio.load_image(e["path"])} |
| for e in req["images"]] |
| return out |
|
|
|
|
| def manifest_frames(paths: List[Path]) -> List[Dict[str, Any]]: |
| from tt_streampetr import load_sample |
|
|
| return [decoded(load_sample(p)) for p in paths] |
|
|
|
|
| def describe_frames(frames: List[Dict[str, Any]]) -> Dict[str, Any]: |
| shapes = {tuple(int(s) for s in np.asarray(e["image"]).shape) for fr in frames for e in fr["images"]} |
| return {"frames": len(frames), "image_shapes": sorted(shapes)} |
|
|
|
|
| class Sequencer: |
| """Turns a short frame list into an endless stream: stamps shifted per cycle so time never goes backwards.""" |
|
|
| def __init__(self, frames: List[Dict[str, Any]], sid: str, t0: float = 1.0e9): |
| self.frames, self.sid, self.t0 = frames, sid, t0 |
| st = [f.get("stream", {}).get("timestamp_s") for f in frames] |
| self.real = all(s is not None for s in st) |
| self.period = (max(st) - min(st) + 0.1) if self.real else 0.1 * len(frames) |
| self.base = min(st) if self.real else 0.0 |
|
|
| def __call__(self, i: int) -> Dict[str, Any]: |
| fr = self.frames[i % len(self.frames)] |
| cyc = i // len(self.frames) |
| req = {k: v for k, v in fr.items() if not k.startswith("_")} |
| stream = dict(fr.get("stream") or {}) |
| stream.pop("reset", None) |
| if self.real: |
| shift = cyc * self.period |
| stream["timestamp_s"] = float(stream["timestamp_s"]) + shift |
| req["images"] = [dict(e, timestamp_s=float(e["timestamp_s"]) + shift) if "timestamp_s" in e else e |
| for e in fr["images"]] |
| else: |
| stream["timestamp_s"] = self.t0 + 0.1 * i |
| stream.setdefault("T_world_from_ego", np.eye(4).tolist()) |
| stream["id"] = self.sid |
| if i == 0: |
| stream["reset"] = True |
| req["stream"] = stream |
| return req |
|
|
|
|
| |
| class Probe: |
| """Splits one ``model(...)`` call into synchronized stages (module docstring). Installed on the model instance |
| and its runner (instance attributes shadow the methods); ``active`` switches the timing on.""" |
|
|
| def __init__(self, model, bench): |
| import ttnn |
|
|
| import tt_streampetr.tt.model as tt_model |
| import tt_streampetr.ttaw.trace as trace_mod |
|
|
| self.model, self.bench, self.active = model, bench, False |
| r, dev = model.runner, model.device |
| sync = lambda: ttnn.synchronize_device(dev) |
| orig_collect, orig_enqueue, orig_execute, orig_read = (r._collect_uploads, r._enqueue_uploads, r._execute, |
| r.read) |
| orig_prepare, orig_post = model._prepare, model._postprocess |
| self._trace_mod, self._orig_to_numpy = trace_mod, trace_mod.to_numpy |
| self._tt_model, self._orig_img_rows = tt_model, tt_model.img_rows |
| self._restore = [(r, "_collect_uploads"), (r, "_enqueue_uploads"), (r, "_execute"), (r, "read"), |
| (model, "_prepare"), (model, "_postprocess")] |
| self._conv_ms = 0.0 |
| orig_to_numpy, orig_img_rows = self._orig_to_numpy, self._orig_img_rows |
|
|
| def timed(name, fn, *a, **k): |
| if not self.active: |
| return fn(*a, **k) |
| t0 = time.perf_counter() |
| out = fn(*a, **k) |
| self.bench.add(name, (time.perf_counter() - t0) * 1e3) |
| return out |
|
|
| def enqueue(uploads): |
| def run(): |
| orig_enqueue(uploads) |
| sync() |
| return timed("h2d", run) |
|
|
| def execute(trace): |
| def run(): |
| orig_execute(trace) |
| sync() |
| return timed("trace", run) |
|
|
| def to_numpy(*a, **k): |
| t0 = time.perf_counter() |
| out = orig_to_numpy(*a, **k) |
| self._conv_ms += (time.perf_counter() - t0) * 1e3 |
| return out |
|
|
| def read(variant=None, **kw): |
| if not self.active: |
| return orig_read(variant, **kw) |
| self._conv_ms = 0.0 |
| t0 = time.perf_counter() |
| out = orig_read(variant, **kw) |
| total = (time.perf_counter() - t0) * 1e3 |
| self.bench.add("d2h", total - self._conv_ms) |
| self.bench.add("host_out", self._conv_ms) |
| return out |
|
|
| r._collect_uploads = lambda inputs, params: timed("host_in", orig_collect, inputs, params) |
| r._enqueue_uploads, r._execute, r.read = enqueue, execute, read |
| model._prepare = lambda *a, **k: timed("host_pre", orig_prepare, *a, **k) |
| model._postprocess = lambda *a, **k: timed("host_post", orig_post, *a, **k) |
| trace_mod.to_numpy = to_numpy |
| tt_model.img_rows = lambda *a, **k: timed("img_rows", orig_img_rows, *a, **k) |
|
|
| def remove(self) -> None: |
| for obj, name in self._restore: |
| obj.__dict__.pop(name, None) |
| self._trace_mod.to_numpy = self._orig_to_numpy |
| self._tt_model.img_rows = self._orig_img_rows |
|
|
|
|
| |
| def _git(*args: str) -> str: |
| return subprocess.run(["git", *args], capture_output=True, text=True, |
| env=dict(os.environ, GIT_OPTIONAL_LOCKS="0")).stdout.strip() |
|
|
|
|
| def environment() -> Dict[str, Any]: |
| env: Dict[str, Any] = {"host": platform.node(), "python": platform.python_version(), |
| "loadavg_start": os.getloadavg(), "cpus": os.cpu_count(), |
| "omp_num_threads": os.environ.get("OMP_NUM_THREADS"), |
| "trace_alloc_tracking": os.environ.get("TT_METAL_TRACE_ALLOC_TRACKING")} |
| try: |
| env["cpu_model"] = next(line.split(":", 1)[1].strip() for line in open("/proc/cpuinfo") |
| if line.startswith("model name")) |
| except (OSError, StopIteration): |
| pass |
| gov = Path("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor") |
| env["cpu_governor"] = gov.read_text().strip() if gov.is_file() else None |
| tm = ROOT / "tt-metal" |
| env["tt_metal_commit"] = _git("-C", str(tm), "rev-parse", "HEAD") |
| topo = tm / "tt_metal/impl/dispatch/topology.cpp" |
| env["eth_patch_marker"] = topo.is_file() and "single_chip_arch_1cq_no_dispatch_s" in topo.read_text() |
| try: |
| from tt_streampetr.ttaw import __version__ as ttaw_version |
|
|
| env["ttaw"] = ttaw_version |
| env["ttaw_vendored"] = json.loads((CODE / "tt_streampetr/ttaw/VENDORED.json").read_text()).get("source_commit") |
| except (ImportError, OSError, ValueError): |
| pass |
| env["bundle_commit"] = _git("-C", str(CODE.parent), "rev-parse", "HEAD") |
| env["bundle_dirty"] = bool(_git("-C", str(CODE.parent), "status", "--porcelain", "code")) |
| return env |
|
|
|
|
| |
| def agreement_check(model, sample: Path = None) -> Dict[str, Any]: |
| """A cold-start sample frame vs its stored CPU-reference body (``tests/test_e2e_device.py``): the shipped synthetic |
| sample by default, or the staged PandaSet frame 40 (``pandaset_dir() / pandaset_019_f40.json``).""" |
| from tt_streampetr import load_sample |
| from tt_streampetr.server.smoke_test import REFERENCE_GATES |
| from tt_streampetr.ttaw.server import smoke |
|
|
| sample = Path(sample) if sample is not None else SAMPLES / "synthetic_5cam.json" |
| ref_path = sample.with_suffix(".reference.json") |
| if not ref_path.is_file(): |
| return {"skipped": f"{ref_path} not found"} |
| kw = load_sample(sample) |
| kw["stream"] = dict(kw.get("stream") or {}, id="bench-agreement", reset=True) |
| out = model(**kw) |
| body = json.loads(json.dumps(out.to_dict())) |
| metrics, fails = smoke.compare_with_reference(body, json.loads(ref_path.read_text()), gates=REFERENCE_GATES) |
| d = metrics["detections"] |
| return {"recall": d.get("recall"), "precision": d.get("precision"), "score_err_max": d.get("score_err_max"), |
| "num": body.get("num_detections"), "failures": fails} |
|
|
|
|
| def bench_input(model, name: str, frames: List[Dict[str, Any]], iters: int, warm: int) -> Dict[str, Any]: |
| from tt_streampetr.ttaw.profiling import AiclkSampler, StageBench |
|
|
| seq = Sequencer(frames, f"bench-{name}") |
| for i in range(warm): |
| model(**seq(i)) |
| inst = StageBench(f"{name} instrumented") |
| probe = Probe(model, inst) |
| try: |
| probe.active = True |
| with AiclkSampler(interval_s=0.05) as clk_i: |
| for i in range(iters): |
| req = seq(warm + i) |
| t0 = time.perf_counter() |
| model(**req) |
| inst.add("e2e_instrumented", (time.perf_counter() - t0) * 1e3) |
| probe.active = False |
| finally: |
| probe.remove() |
| plain = StageBench(f"{name} plain") |
| ndet = [] |
| with AiclkSampler(interval_s=0.05) as clk_p: |
| for i in range(iters): |
| req = seq(warm + iters + i) |
| t0 = time.perf_counter() |
| out = model(**req) |
| plain.add("e2e", (time.perf_counter() - t0) * 1e3) |
| for k, v in out.timing_ms.items(): |
| plain.add(f"api.{k}", v) |
| ndet.append(len(out.scores)) |
| return {"input": name, **describe_frames(frames), "iters": iters, "warm": warm, |
| "instrumented": inst.summary(), "plain": plain.summary(), |
| "detections_per_frame": {"min": int(min(ndet)), "max": int(max(ndet)), "mean": float(np.mean(ndet))}, |
| "aiclk_instrumented": clk_i.summary(), "aiclk_plain": clk_p.summary(), "loadavg": os.getloadavg()} |
|
|
|
|
| def device_only(model, iters: int, b2b: int) -> Dict[str, Any]: |
| """Replays only (no host work between them, no uploads): one blocking replay and back-to-back replays of the |
| ``frame`` trace on the inputs / state the last frame left (the shapes are fixed: the time does not depend on |
| the values).""" |
| import ttnn |
|
|
| from tt_streampetr.ttaw.profiling import AiclkSampler, StageBench, time_b2b |
|
|
| r, dev = model.runner, model.device |
| sync = lambda: ttnn.synchronize_device(dev) |
| sb = StageBench("device only") |
| with AiclkSampler(interval_s=0.05) as clk: |
| r.replay("frame") |
| sync() |
| for _ in range(iters): |
| t0 = time.perf_counter() |
| r.replay("frame") |
| sync() |
| sb.add("replay.frame", (time.perf_counter() - t0) * 1e3) |
| for _ in range(3): |
| sb.add("b2b.frame", time_b2b(lambda: r.replay("frame"), sync, n=b2b, warmup=2)) |
| return {"stages_ms": sb.summary(), "aiclk": clk.summary()} |
|
|
|
|
| def main() -> int: |
| ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) |
| ap.add_argument("--iters", type=int, default=100, help="timed frames per loop and input (>= 50)") |
| ap.add_argument("--warm", type=int, default=6) |
| ap.add_argument("--b2b", type=int, default=30, help="back-to-back replays per measurement") |
| ap.add_argument("--inputs", default="synthetic,pandaset", |
| help="comma list of synthetic, pandaset[=dir], sample=<manifest>[+<manifest>...]") |
| ap.add_argument("--synthetic-preset", default="pandaset_019") |
| ap.add_argument("--dispatch", default=None, choices=[None, "eth", "worker"]) |
| ap.add_argument("--num-cqs", type=int, default=None) |
| ap.add_argument("--weights-dir", default=None) |
| ap.add_argument("--no-agreement", action="store_true") |
| ap.add_argument("--json") |
| a = ap.parse_args() |
| env = environment() |
| from tt_streampetr import StreamPETR |
|
|
| t0 = time.perf_counter() |
| model = StreamPETR.from_pretrained(dispatch=a.dispatch, num_command_queues=a.num_cqs, weights_dir=a.weights_dir) |
| res: Dict[str, Any] = {"env": env} |
| try: |
| res.update({"config": model.device_info, "knobs": model.knobs.as_dict(), |
| "load_s": round(time.perf_counter() - t0, 2), |
| "warmup_ms": {k: round(v, 1) for k, v in model.warmup_ms.items()}, |
| "program_cache_entries": getattr(model.device, "num_program_cache_entries", lambda: None)()}) |
| print("config:", json.dumps(res["config"]), "| knobs:", json.dumps(res["knobs"]), flush=True) |
| if not a.no_agreement: |
| res["agreement_shipped"] = agreement_check(model) |
| print("agreement (shipped synthetic sample):", json.dumps(res["agreement_shipped"]), flush=True) |
| res["agreement_sample_f40"] = agreement_check(model, pandaset_dir() / "pandaset_019_f40.json") |
| print("agreement (staged PandaSet f40):", json.dumps(res["agreement_sample_f40"]), flush=True) |
| inputs: List[Tuple[str, List[Dict[str, Any]]]] = [] |
| for spec in [s.strip() for s in a.inputs.split(",") if s.strip()]: |
| kind, _, arg = spec.partition("=") |
| if kind == "synthetic": |
| inputs.append(("synthetic", synthetic_frames(arg or a.synthetic_preset))) |
| elif kind == "pandaset": |
| d = pandaset_dir(arg) |
| paths = [d / f"pandaset_019_f{n}.json" for n in (40, 41, 42, 43)] |
| if all(p.is_file() for p in paths): |
| inputs.append(("pandaset", manifest_frames(paths))) |
| else: |
| print("skip pandaset: pandaset_019_f40-43.json not found", flush=True) |
| elif kind == "sample": |
| paths = [Path(x) for x in arg.split("+")] |
| inputs.append((paths[0].stem, manifest_frames(paths))) |
| else: |
| raise SystemExit(f"unknown input {spec!r}") |
| res["inputs"] = {} |
| for name, frames in inputs: |
| r = bench_input(model, name, frames, a.iters, a.warm) |
| res["inputs"][name] = r |
| p, i = r["plain"], r["instrumented"] |
| print(f"[{name}] e2e p50 {p['e2e']['p50']:.1f} / p99 {p['e2e']['p99']:.1f} ms | " |
| + " | ".join(f"{k} {v['p50']:.2f}" for k, v in i.items()), flush=True) |
| res["device_only"] = device_only(model, a.iters, a.b2b) |
| print("device only:", " | ".join(f"{k} p50 {v['p50']:.3f}" for k, v in |
| res["device_only"]["stages_ms"].items()), flush=True) |
| res["trace"] = model.runner.describe() |
| bb = getattr(getattr(model, "tt", None), "backbone", None) |
| if bb is not None and hasattr(bb, "placements"): |
| pl = bb.placements() |
| res["conv_placements"] = pl |
| if pl: |
| from collections import Counter |
| print("conv placements (CONV_L1):", dict(Counter(pl.values())), |
| "| not on L1:", [k for k, v in pl.items() if not v.startswith("l1")], flush=True) |
| res["env"]["loadavg_end"] = os.getloadavg() |
| finally: |
| model.close() |
| if a.json: |
| Path(a.json).parent.mkdir(parents=True, exist_ok=True) |
| Path(a.json).write_text(json.dumps(res, indent=1, default=str) + "\n") |
| print("wrote", a.json) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
