| |
| |
| """Stage bench of warm forwards: 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/bevdet/bench_eth_1cq.json |
| bin/devrun -t 3600 -- 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 (every frame pushes the BEV history ring, as in |
| steady operation) after 6 warm frames: |
| |
| - ``synthetic[=preset]``: six 1600x900 camera frames made by ``tt_bevdet.tt.warmup.synthetic_images`` (data |
| generated by us, the node's native size: no resize) with a calibration preset (default ``synthetic_6cam``, the |
| shipped sample's rig; OPT_BASELINE.md of 2026-10-08 used ``pandaset_019``, which now waits in the git-ignored |
| ``staging_samples_pandaset/calib``), three seeds, Autoware semantics (identity ego pose: the AlignBEV grid never |
| changes, so it is never re-uploaded); |
| - ``pandaset[=dir]``: ``pandaset_019_f{38,39,40}.json`` from ``dir``, ``samples/`` or the git-ignored |
| ``staging_samples_pandaset/samples`` (1920x1080: the node's ``cv::resize`` to 1600x900 runs on the host), Autoware |
| semantics; |
| - ``sample=<manifest>[+<manifest>...]``: any sample manifest(s) of ``load_sample`` (e.g. the shipped sample); |
| - ``nuscenes=<manifest>``: key-frames of a ``research/bevdet/public_data`` manifest (nuScenes v1.0-mini, CC |
| BY-NC-SA 4.0: local measurements only, never shipped), 1600x900, with the manifest's ego poses |
| (``stream.T_world_from_ego``: a new AlignBEV grid is uploaded every frame). |
| |
| Images are decoded once before timing (Autoware hands the node decoded images), so ``host_pre`` is the node |
| transport + Preprocess gather + normalisation + NHWC rows + the per-calibration cache lookups + the temporal 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``; ``host_in.encode`` (numpy -> ttnn host tensors) and ``h2d.encode`` (enqueue + |
| synchronize); ``trace.encode`` (one replay + synchronize); ``d2h.encode`` (device -> host copy) and |
| ``host_out.encode`` (TILE -> numpy conversion); ``bevpool`` (host BEVPool + history grid + plan); the same four |
| for ``decode``; ``trace.push`` / ``trace.fill`` (ring update); ``host_post`` (decode + scale-NMS + node filter); |
| ``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.<variant>`` one blocking replay; ``b2b.<variant>`` |
| back-to-back replays per replay; ``b2b.frame`` = encode + decode + push back to back per frame (device-bound |
| throughput). AICLK / power / temperature are sampled during the loops (``AiclkSampler``), and an agreement check of |
| PandaSet f38-40 against the stored CPU-reference boxes (``tests/goldens``, ``host.agreement``) is run first, so every |
| dispatch / CQ configuration's numbers come with its accuracy. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| 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_bevdet" / "samples" |
| GOLDENS = CODE / "tt_bevdet" / "tests" / "goldens" |
| ROOT = Path(os.environ.get("TT_MODELS_ROOT", "/home/ubuntu/experiments/tt-models")) |
| DEFAULT_NUSCENES = ROOT / "research/bevdet/public_data/nuscenes/manifests/port2hz__scene-0103.json" |
|
|
|
|
| |
| def synthetic_frames(cfg, n: int = 3, preset: str = "") -> List[Dict[str, Any]]: |
| from tt_bevdet.host.inputs import CALIB_DIR |
| from tt_bevdet.tt.warmup import WARMUP_PRESET, synthetic_images |
| from tt_bevdet.ttaw import io as tio |
|
|
| preset = preset or WARMUP_PRESET |
| calib_dir = CALIB_DIR |
| if not (CALIB_DIR / f"{preset}.json").is_file() and (CODE.parent / "staging_samples_pandaset" / "calib" |
| / f"{preset}.json").is_file(): |
| calib_dir = CODE.parent / "staging_samples_pandaset" / "calib" |
| calib = tio.resolve_calibration({"preset": preset}, calib_dir) |
| return [{"images": synthetic_images(cfg, seed=s + 1), "calibration": calib} for s in range(n)] |
|
|
|
|
| def pandaset_dir(override: str = "") -> Path: |
| """Where the PandaSet 019 f38-40 manifests are: ``override``, else ``samples/``, else the git-ignored |
| ``staging_samples_pandaset/samples`` of the bundle (where they wait while they may not ship).""" |
| 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 (38, 39, 40)): |
| return c |
| return cands[0] |
|
|
|
|
| def pandaset_frames(override: str = "") -> List[Dict[str, Any]]: |
| from tt_bevdet import load_sample |
|
|
| d = pandaset_dir(override) |
| paths = [d / f"pandaset_019_f{n}.json" for n in (38, 39, 40)] |
| if not all(p.is_file() for p in paths): |
| return [] |
| return [load_sample(p) for p in paths] |
|
|
|
|
| def nuscenes_frames(manifest: Path, n: int = 3) -> List[Dict[str, Any]]: |
| """The first ``n`` key-frames of a public-data manifest as decoded requests with the ego pose (no reset flag: |
| the cycle is a steady stream).""" |
| sys.path.insert(0, str(HERE.parent)) |
| from run_public_tt import frame_request |
| from tt_bevdet.ttaw import io as tio |
|
|
| m = json.loads(manifest.read_text()) |
| out = [] |
| for f in [f for f in m["frames"] if f.get("key_frame", True)][:n]: |
| req = frame_request(m, f, "bench") |
| req["images"] = {k: tio.load_image(v) for k, v in req["images"].items()} |
| stream = {k: v for k, v in req.pop("stream").items() if k == "T_world_from_ego"} |
| req["_stream"] = stream |
| out.append(req) |
| return out |
|
|
|
|
| def describe_frames(frames: List[Dict[str, Any]]) -> Dict[str, Any]: |
| shapes = set() |
| for fr in frames: |
| imgs = fr["images"] |
| vals = imgs.values() if isinstance(imgs, dict) else imgs |
| for v in vals: |
| a = getattr(v, "image", v) |
| shapes.add(tuple(int(s) for s in np.asarray(a).shape)) |
| return {"frames": len(frames), "image_shapes": sorted(shapes)} |
|
|
|
|
| |
| class Probe: |
| """Splits one ``model(...)`` call into synchronized stages (see the 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_bevdet.ttaw.trace as trace_mod |
|
|
| self.model, self.bench, self.active = model, bench, False |
| self._seg = "encode" |
| 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 |
| orig_to_numpy = trace_mod.to_numpy |
| self._restore = [(r, "_collect_uploads"), (r, "_enqueue_uploads"), (r, "_execute"), (r, "read"), |
| (model, "_prepare"), (model, "_postprocess")] |
| self._trace_mod, self._orig_to_numpy = trace_mod, orig_to_numpy |
| self._conv_ms = 0.0 |
|
|
| 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 collect(inputs, params): |
| self._seg = "encode" if inputs and ("img" in inputs or "crop" in inputs) else "decode" |
| return timed(f"host_in.{self._seg}", orig_collect, inputs, params) |
|
|
| def enqueue(uploads): |
| def run(): |
| orig_enqueue(uploads) |
| sync() |
| return timed(f"h2d.{self._seg}", run) |
|
|
| def execute(trace): |
| def run(): |
| orig_execute(trace) |
| sync() |
| return timed(f"trace.{trace.variant}", 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(f"d2h.{variant}", total - self._conv_ms) |
| self.bench.add(f"host_out.{variant}", self._conv_ms) |
| return out |
|
|
| r._collect_uploads, r._enqueue_uploads, r._execute, r.read = collect, 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 |
|
|
| def remove(self) -> None: |
| for obj, name in self._restore: |
| obj.__dict__.pop(name, None) |
| self._trace_mod.to_numpy = self._orig_to_numpy |
|
|
|
|
| |
| 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")} |
| 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" |
| try: |
| env["tt_metal_commit"] = subprocess.run(["git", "-C", str(tm), "rev-parse", "HEAD"], capture_output=True, |
| text=True, env=dict(os.environ, GIT_OPTIONAL_LOCKS="0")).stdout.strip() |
| except OSError: |
| pass |
| 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_bevdet.ttaw import __version__ as ttaw_version |
|
|
| env["ttaw"] = ttaw_version |
| env["ttaw_vendored"] = json.loads((CODE / "tt_bevdet/ttaw/VENDORED.json").read_text()).get("source_commit") |
| except (ImportError, OSError, ValueError): |
| pass |
| try: |
| env["bundle_commit"] = subprocess.run(["git", "-C", str(CODE.parent), "rev-parse", "HEAD"], |
| capture_output=True, text=True, |
| env=dict(os.environ, GIT_OPTIONAL_LOCKS="0")).stdout.strip() |
| env["bundle_dirty"] = bool(subprocess.run(["git", "-C", str(CODE.parent), "status", "--porcelain", "code"], |
| capture_output=True, text=True, |
| env=dict(os.environ, GIT_OPTIONAL_LOCKS="0")).stdout.strip()) |
| except OSError: |
| pass |
| return env |
|
|
|
|
| |
| def agreement_check(model) -> Dict[str, Any]: |
| """PandaSet f38-40 through a fresh stream vs the stored CPU-reference boxes (the e2e gate's definition).""" |
| from tt_bevdet.host.agreement import agreement, vendor_boxes |
|
|
| from tt_bevdet.tests import paths |
|
|
| gold = paths.pandaset_golden() or GOLDENS / paths.PANDASET_GOLDEN |
| frames = pandaset_frames() |
| if not frames or not gold.is_file(): |
| return {"skipped": "PandaSet sample or its goldens not in the bundle"} |
| rec = json.loads(gold.read_text()) |
| res = {} |
| for n, req in zip((38, 39, 40), frames): |
| out = model(**req, stream={"id": "bench-agreement"}, score_threshold=0.0) |
| a = agreement(vendor_boxes(out), rec["frames"][f"f{n}"]["boxes"]) |
| res[f"f{n}"] = {k: (round(v, 5) if isinstance(v, float) else v) for k, v in a.items() if k != "score_errs"} |
| return res |
|
|
|
|
| def bench_input(model, name: str, frames: List[Dict[str, Any]], iters: int, warm: int) -> Dict[str, Any]: |
| from tt_bevdet.ttaw.profiling import AiclkSampler, StageBench |
|
|
| sid = f"bench-{name}" |
|
|
| def call(i: int): |
| fr = dict(frames[i % len(frames)]) |
| stream = dict(fr.pop("_stream", {}), id=sid) |
| return model(**fr, stream=stream) |
|
|
| for i in range(warm): |
| call(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): |
| t0 = time.perf_counter() |
| call(warm + i) |
| 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): |
| t0 = time.perf_counter() |
| out = call(warm + iters + i) |
| 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, back-to-back replays, and the |
| frame sequence encode + decode + push back to back. The AlignBEV grid is set to the identity of a full history |
| (Autoware mode, every slot a bit-exact copy); when the last input uploaded another grid (the nuScenes ego-pose |
| frames), ``decode`` and the frame are measured again with that grid (``*_moving``): grid_sample's cost depends |
| on the sampling positions.""" |
| import ttnn |
|
|
| from tt_bevdet.tt.bev_fusion import RING_SLOTS, history_grid |
| from tt_bevdet.ttaw.profiling import AiclkSampler, StageBench, time_b2b |
|
|
| r, dev = model.runner, model.device |
| sync = lambda: ttnn.synchronize_device(dev) |
| slot = r._slot("grid", "param") |
| last = None if slot.last_value is None else np.frombuffer(slot.last_value, np.float32).reshape(slot.shape).copy() |
| identity = history_grid(np.tile(np.array([1, 0, 0, 0, 1, 0], np.float32), (RING_SLOTS, 1)), |
| list(range(RING_SLOTS)), 1, model.cfg.bev_hw) |
| moving = None if last is None or np.array_equal(last, identity) else last |
| sb = StageBench("device only") |
| mv = "decode_move" if getattr(model.tt, "tile_ring", None) is not None else "decode" |
|
|
| after = getattr(model.tt, "after_variant", lambda v, d: v) |
|
|
| def frame(dec="decode"): |
| r.replay("encode") |
| r.replay(dec) |
| r.replay(after("push", dec)) |
|
|
| with AiclkSampler(interval_s=0.05) as clk: |
| r.upload(params={"grid": identity}) |
| sync() |
| for v in ("encode", "decode", "push", "fill") + (("push_x",) if getattr(model.tt, "push_x", False) else ()): |
| r.replay(v) |
| sync() |
| for _ in range(iters): |
| t0 = time.perf_counter() |
| r.replay(v) |
| sync() |
| sb.add(f"replay.{v}", (time.perf_counter() - t0) * 1e3) |
| for _ in range(3): |
| sb.add(f"b2b.{v}", time_b2b(lambda: r.replay(v), sync, n=b2b, warmup=2)) |
| for _ in range(3): |
| sb.add("b2b.frame", time_b2b(frame, sync, n=b2b, warmup=2)) |
| if moving is not None: |
| r.upload(params={"grid": moving}) |
| sync() |
| for _ in range(iters): |
| t0 = time.perf_counter() |
| r.replay(mv) |
| sync() |
| sb.add("replay.decode_moving", (time.perf_counter() - t0) * 1e3) |
| for _ in range(3): |
| sb.add("b2b.decode_moving", time_b2b(lambda: r.replay(mv), sync, n=b2b, warmup=2)) |
| sb.add("b2b.frame_moving", time_b2b(lambda: frame(mv), sync, n=b2b, warmup=2)) |
| return {"stages_ms": sb.summary(), "aiclk": clk.summary(), "moving_grid": moving is not None} |
|
|
|
|
| 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=50, help="back-to-back replays per measurement") |
| ap.add_argument("--inputs", default="synthetic,pandaset", |
| help="comma list of synthetic, pandaset[=dir], nuscenes[=manifest], sample=<manifest>[+<manifest>...] " |
| f"(default nuScenes manifest: {DEFAULT_NUSCENES})") |
| 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_bevdet import BEVDet |
|
|
| t0 = time.perf_counter() |
| model = BEVDet.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) |
| res["conv_placement"] = model.tt.encoder.placements() |
| res["bottleneck_tails"] = model.tt.encoder.tails() |
| if res["conv_placement"]: |
| from collections import Counter |
| print("conv placement:", dict(Counter(res["conv_placement"].values())), |
| {k: v for k, v in res["conv_placement"].items() if v != "l1"}, flush=True) |
| if not a.no_agreement: |
| res["agreement_pandaset_f38_40"] = agreement_check(model) |
| print("agreement:", json.dumps(res["agreement_pandaset_f38_40"]), flush=True) |
| inputs: List[Tuple[str, List[Dict[str, Any]]]] = [] |
| for spec in [s.strip() for s in a.inputs.split(",") if s.strip()]: |
| if spec.split("=", 1)[0] == "synthetic": |
| inputs.append(("synthetic", synthetic_frames(model.cfg, preset=spec.partition("=")[2]))) |
| elif spec.split("=", 1)[0] == "pandaset": |
| fr = pandaset_frames(spec.split("=", 1)[1] if "=" in spec else "") |
| if fr: |
| inputs.append(("pandaset", fr)) |
| else: |
| print("skip pandaset: pandaset_019_f38-40.json not found", flush=True) |
| elif spec.startswith("sample="): |
| from tt_bevdet import load_sample |
|
|
| paths = [Path(x) for x in spec.split("=", 1)[1].split("+")] |
| inputs.append((paths[0].stem, [load_sample(x) for x in paths])) |
| elif spec.startswith("nuscenes"): |
| man = Path(spec.split("=", 1)[1]) if "=" in spec else DEFAULT_NUSCENES |
| if man.is_file(): |
| inputs.append(("nuscenes", nuscenes_frames(man))) |
| else: |
| print(f"skip nuscenes: {man} not found", flush=True) |
| 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() |
| 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()) |
|
|
