| |
| |
| """Stage breakdown of warm forwards: the numbers OPT_BASELINE.md / OPT_REPORT.md / the card quote. |
| |
| bin/devrun -t 900 -- python code/scripts/bench.py --iters 100 --json logs/baseline_bench.json |
| bin/devrun -t 900 -- python code/scripts/bench.py --input a.npz --input b.npz --dispatch worker --cqs 2 |
| |
| For every ``--input`` cloud (default: the shipped sample ``samples/test_pcd.npz``), batch 1, after ``--warmup`` |
| calls: ``--iters`` iterations of the served path cut into stages, each synchronised on its own (p50 / p99 / mean / |
| min in ms, so the rows do not add up to ``e2e`` exactly), then ``--iters`` plain ``model(points)`` calls, then |
| back-to-back replays: |
| |
| ===================== ================================================================================================ |
| ``pre`` host pre-processing as served (``model._prepare``): decode + densification + sampling + |
| deterministic voxelization (first 20 points per pillar, N_MAX 65,536) + the 11 decorated |
| features + the canvas index |
| ``pre.load`` the file decode alone (``io.load_points``) |
| ``pre.densify`` ``densified_points``: the (N, 5) x, y, z, intensity (uint8 semantics), time_lag cloud |
| ``pre.voxelize`` ``voxelize``: sampling mode, strict range filter, ``assign_pillars`` |
| ``pre.decorate`` ``decorate_pillars``: the (65536, 20, 11) float32 encoder input |
| ``pre.index`` ``canvas_gather_index``: cell -> pillar row (the C19 gather index) |
| ``host_in`` ``stage_inputs`` (C24 ``FrameStaging.write``): the kept pillars -> bf16 in the persistent buffers |
| ``h2d`` upload of both trace inputs + device synchronise (``TraceRunner.upload``) |
| ``h2d.<input>`` the upload split per input (``features`` 64 MiB bf16 rows, ``canvas_index`` 2.25 MiB uint32) |
| ``trace`` one blocking replay of the whole-graph trace |
| ``d2h`` ``TraceRunner.read`` as served: the one packed fp32 output row -> ``{heads, query_heat, indices}`` |
| ``d2h.copy`` the readback split: the device-to-host copy into a preallocated host tensor ... |
| ``d2h.to_numpy`` ... and its host conversion to numpy |
| ``unpack`` ``TtTransFusion.assemble``: the graph outputs ``cls_score0`` / ``bbox_pred0`` / ``dir_cls_pred0`` |
| ``post`` host post-processing (decode, score threshold, circle NMS, ROS conversion, IoU-BEV NMS, remapper) |
| ``e2e`` ``model(points)`` as served (its ``timing_ms`` medians are reported too) |
| ``b2b`` back-to-back replays, one synchronise per round of ``--b2b-iters``: device time per forward |
| ===================== ================================================================================================ |
| |
| AICLK / power / temperature are sampled from sysfs during the run (``ttaw.profiling.AiclkSampler``), and the host |
| load average is recorded (the host stages move with it). ``--check`` (default) compares the served output of each |
| input with a stored fp32 CPU-reference ``/predict`` body (``--reference PATH`` per input, ``-`` for none; default: |
| ``<stem>.reference.json`` next to the input) with the container-smoke gates (``server/smoke_test.py`` |
| ``REFERENCE_GATES``). Always quote the configuration line (dispatch / CQs / grid) with the numbers. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import platform |
| import statistics |
| import time |
| from pathlib import Path |
| from typing import Any, Dict, List, Optional |
|
|
| import numpy as np |
|
|
| from tt_transfusion import TransFusion |
| from tt_transfusion.ttaw.profiling import AiclkSampler, StageBench, time_b2b |
|
|
| CODE = Path(__file__).resolve().parents[1] |
| SAMPLE = CODE / "tt_transfusion" / "samples" / "test_pcd.npz" |
|
|
|
|
| def check_output(model, src: str, reference: Optional[Path]) -> Dict[str, Any]: |
| """Served output vs the stored fp32 CPU-reference ``/predict`` body (the container-smoke comparison).""" |
| from tt_transfusion.server import smoke_test as st |
| from tt_transfusion.ttaw.server import smoke |
|
|
| out = model(src).to_dict() |
| res: Dict[str, Any] = {"num_detections": out["num_detections"], "counts": out["meta"].get("counts"), |
| "num_pillars": out["meta"].get("num_pillars")} |
| if reference is not None and reference.is_file(): |
| metrics, fails = smoke.compare_with_reference(out, smoke.load_json(reference), gates=st.REFERENCE_GATES) |
| res["reference"] = {"file": str(reference), "metrics": metrics, "failures": fails} |
| return res |
|
|
|
|
| def bench_input(model, src: str, *, iters: int, warmup: int, b2b_iters: int, b2b_rounds: int, |
| name: str) -> Dict[str, Any]: |
| import ttnn |
|
|
| from tt_transfusion import io as tio |
| from tt_transfusion.host.preprocess import decorate_pillars |
| from tt_transfusion.tt.model import stage_inputs |
| from tt_transfusion.ttaw.tensors import to_numpy |
| from tt_transfusion.ttaw.voxelize import canvas_gather_index |
|
|
| tr = model.runner |
| dev, pre = model.device, model.pre |
| params = model.validate_params({}) |
| bench = StageBench(name) |
| for _ in range(warmup): |
| model(src) |
| |
| variant = getattr(model.staging, "variant", "default") |
| feat_name = getattr(model.staging, "input_name", "features") |
| out_dev = tr._trace_for(variant, 0).leaves[0][1] |
| host_out = ttnn.allocate_tensor_on_host(out_dev.spec, dev) |
| slots = {k: tr._slot(k, "input") for k in (feat_name, "canvas_index")} |
| for _ in range(iters): |
| t0 = time.perf_counter() |
| prepared = model._prepare(points=src) |
| t1 = time.perf_counter() |
| inputs = stage_inputs(model.staging, prepared) |
| t2 = time.perf_counter() |
| tr.upload(inputs) |
| ttnn.synchronize_device(dev) |
| t3 = time.perf_counter() |
| tr.replay(variant) |
| ttnn.synchronize_device(dev) |
| t4 = time.perf_counter() |
| packed = tr.read(variant) |
| t5 = time.perf_counter() |
| raw = model.graph.assemble(packed) |
| t6 = time.perf_counter() |
| model._postprocess(raw, prepared, params) |
| t7 = time.perf_counter() |
| for stage, a, b in (("pre", t0, t1), ("host_in", t1, t2), ("h2d", t2, t3), ("trace", t3, t4), |
| ("d2h", t4, t5), ("unpack", t5, t6), ("post", t6, t7)): |
| bench.add(stage, (b - a) * 1e3) |
| |
| with bench.stage("pre.load"): |
| tio.load_points(src) |
| with bench.stage("pre.densify"): |
| pts, _, _, _ = pre.densified_points(src) |
| with bench.stage("pre.voxelize"): |
| pillars, _ = pre.voxelize(pts, None) |
| with bench.stage("pre.decorate"): |
| decorate_pillars(pillars, pre.scale_zyx, pre.offset_zyx) |
| with bench.stage("pre.index"): |
| canvas_gather_index(pillars.coords, pillars.num_pillars, pre.spec, sentinel=pre.spec.max_pillars) |
| for k, slot in slots.items(): |
| with bench.stage(f"h2d.{k}"): |
| ttnn.copy_host_to_device_tensor(inputs[k], slot.buffers[0], cq_id=0) |
| ttnn.synchronize_device(dev) |
| with bench.stage("d2h.copy"): |
| ttnn.copy_device_to_host_tensor(out_dev, host_out, blocking=True) |
| with bench.stage("d2h.to_numpy"): |
| to_numpy(host_out) |
| timing: List[Dict[str, float]] = [] |
| for _ in range(iters): |
| t0 = time.perf_counter() |
| out = model(src) |
| bench.add("e2e", (time.perf_counter() - t0) * 1e3) |
| timing.append(dict(out.timing_ms)) |
| rounds = [time_b2b(lambda: tr.replay(variant), lambda: ttnn.synchronize_device(dev), n=b2b_iters) |
| for _ in range(b2b_rounds)] |
| for r in rounds: |
| bench.add("b2b", r) |
| b2b = statistics.median(rounds) |
| stats = prepared.stats() |
| return {"stages_ms": bench.summary(), "table": bench.table(), |
| "timing_ms_median": {k: statistics.median(t[k] for t in timing) for k in timing[0]}, |
| "b2b_rounds_ms": rounds, "b2b_iters": b2b_iters, "device_fps_b2b": 1e3 / b2b, |
| "e2e_fps_sync": 1e3 / bench.summary()["e2e"]["p50"], |
| "frame": {k: stats.get(k) for k in ("num_points", "num_points_in_range", "num_pillars", |
| "num_pillars_total", "pillar_overflow")}, |
| "variant": variant, |
| "bytes": {"features": int(np.prod(slots[feat_name].shape)) * 2, |
| "canvas_index": int(np.prod(slots["canvas_index"].shape)) * 4, |
| "output": int(np.prod(out_dev.shape)) * 4}} |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) |
| ap.add_argument("--input", action="append", default=None, |
| help="cloud .npz / .pcd (repeatable; default: the shipped sample)") |
| ap.add_argument("--reference", action="append", default=None, |
| help="fp32 CPU-reference /predict body per --input ('-' for none; default <stem>.reference.json" |
| " next to the input)") |
| ap.add_argument("--iters", type=int, default=100) |
| ap.add_argument("--warmup", type=int, default=10) |
| ap.add_argument("--b2b-iters", type=int, default=50) |
| ap.add_argument("--b2b-rounds", type=int, default=5) |
| ap.add_argument("--dispatch", default=None, choices=["eth", "worker"], help="default: TRANSFUSION_DISPATCH or eth") |
| ap.add_argument("--cqs", type=int, default=None, choices=[1, 2], |
| help="command queues (default: TRANSFUSION_NUM_CQS or 1)") |
| ap.add_argument("--weights-dir", default=None, help="holds transfusion.onnx + yaml (default: the pinned HF snapshot)") |
| ap.add_argument("--allow-fallback", action="store_true", help="accept a WORKER fallback of an ETH open") |
| ap.add_argument("--no-check", dest="check", action="store_false") |
| ap.add_argument("--tag", default="", help="label of this configuration in the output") |
| ap.add_argument("--json") |
| a = ap.parse_args() |
| inputs = [Path(p) for p in (a.input or [str(SAMPLE)])] |
| refs = [None if r in (None, "-") else Path(r) for r in (a.reference or [])] + [None] * len(inputs) |
| load0 = os.getloadavg() |
| t0 = time.perf_counter() |
| with TransFusion.from_pretrained(dispatch=a.dispatch, num_command_queues=a.cqs, |
| weights_dir=a.weights_dir) as model: |
| load_s = time.perf_counter() - t0 |
| info = dict(model.device_info) |
| want = a.dispatch or os.environ.get("TRANSFUSION_DISPATCH", "eth") |
| if info.get("dispatch") != want and not a.allow_fallback: |
| raise SystemExit(f"device opened with dispatch {info.get('dispatch')!r}, asked for {want!r}") |
| tag = a.tag or f"{info.get('dispatch')}-{info.get('num_command_queues')}cq {info.get('grid')}" |
| res: Dict[str, Any] = {"tag": tag, "config": info, "variant": model.variant, |
| "precision": model.graph.policy.describe(), "knobs": model.graph.knobs.as_dict(), |
| "trace": model.runner.describe(), "load_s": load_s, |
| "warmup_ms": dict(model.warmup_ms), "iters": a.iters, |
| "host": {"cpus": os.cpu_count(), "machine": platform.processor() or platform.machine(), |
| "loadavg_start": load0}, "inputs": {}} |
| with AiclkSampler(interval_s=0.05) as clk: |
| for k, (path, ref) in enumerate(zip(inputs, refs)): |
| src = str(path) |
| if ref is None and a.reference is None: |
| cand = path.parent / f"{path.stem}.reference.json" |
| ref = cand if cand.is_file() else None |
| r = bench_input(model, src, iters=a.iters, warmup=a.warmup, b2b_iters=a.b2b_iters, |
| b2b_rounds=a.b2b_rounds, name=f"{path.name} | {tag}") |
| r["input"] = src |
| if a.check: |
| r["check"] = check_output(model, src, ref) |
| res["inputs"][f"{k}:{path.name}"] = r |
| print(r["table"], flush=True) |
| print(f"b2b rounds ms: {[round(x, 3) for x in r['b2b_rounds_ms']]} " |
| f"timing_ms medians: { {k2: round(v, 2) for k2, v in r['timing_ms_median'].items()} } " |
| f"frame: {r['frame']}", flush=True) |
| if a.check: |
| print("check:", json.dumps(r["check"], default=str), flush=True) |
| res["aiclk"] = clk.summary() |
| res["trace_after"] = model.runner.describe() |
| res["host"]["loadavg_end"] = os.getloadavg() |
| print(json.dumps({"tag": tag, "config": info, "aiclk": res["aiclk"], "load_s": round(load_s, 2), |
| "program_cache_entries": res["trace_after"].get("program_cache_entries"), |
| "loadavg": [load0, res["host"]["loadavg_end"]]}, default=str), flush=True) |
| 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") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
