"""Micro-benchmark helpers: open chip (ETH dispatch, 12x10), time an op closure via trace replay.""" import sys, time sys.path.insert(0, "/home/ttuser/experiments/tt-models/tools") import torch import ttnn def open_dev(trace=256 << 20, l1_small=0): import os from open_device_12x10 import open_device kw = {} extra = int(os.environ.get("MICRO_KCFG_EXTRA_KB", "0")) # round 12: the model's larger kernel-config ring (model.py) if extra: kw["worker_l1_size"] = 1461376 - extra * 1024 return open_device(grid="12x10", dispatch=os.environ.get("MICRO_DISPATCH", "eth"), trace_region_size=trace, l1_small_size=l1_small, **kw) def pcc(a, b): a = a.reshape(-1).float(); b = b.reshape(-1).float() a = a - a.mean(); b = b - b.mean() return float((a @ b) / (a.norm() * b.norm()).clamp_min(1e-12)) def time_op(dev, fn, reps=20, iters=10): """fn() -> output tensor(s). Returns device us per call (trace of `reps` calls, replayed `iters` times).""" out = fn(); ttnn.synchronize_device(dev) _free(out) tid = ttnn.begin_trace_capture(dev, cq_id=0) outs = [] try: for _ in range(reps): o = fn(); outs.append(o) finally: ttnn.end_trace_capture(dev, tid, cq_id=0) ttnn.execute_trace(dev, tid, cq_id=0, blocking=True) t0 = time.perf_counter() for _ in range(iters): ttnn.execute_trace(dev, tid, cq_id=0, blocking=False) ttnn.synchronize_device(dev) us = (time.perf_counter() - t0) / (iters * reps) * 1e6 ttnn.release_trace(dev, tid) for o in outs: _free(o) return us def _free(o): if isinstance(o, (list, tuple)): for x in o: _free(x) elif o is not None: try: ttnn.deallocate(o) except Exception: pass _PROF_LABELS = [] def prof_run(dev, label, fn, n=6): """Run fn() n times (sync after each) for the device profiler (TT_METAL_DEVICE_PROFILER=1); the spans are read by prof_report() AFTER close_device (the csv is written then). Round 7.""" for _ in range(n): o = fn() ttnn.synchronize_device(dev) _free(o) _PROF_LABELS.append(label) def prof_report(): """Per label: kernel spans (us, first kernel start .. last kernel end of one program run over all cores).""" import os, csv from collections import defaultdict f = os.environ["TT_METAL_PROFILER_DIR"] + "/.logs/profile_log_device.csv" rows = list(csv.reader(open(f)))[2:] lo, hi = {}, {} for r in rows: try: cyc, run = int(r[5]), r[7].strip() except Exception: continue lo[run] = min(lo.get(run, cyc), cyc) hi[run] = max(hi.get(run, cyc), cyc) runs = sorted(lo, key=lambda v: int(v) if v.isdigit() else 0) spans = [(hi[r] - lo[r]) / 1.35e3 for r in runs] spans = spans[-len(_PROF_LABELS):] by = defaultdict(list) for lab, sp in zip(_PROF_LABELS[-len(spans):], spans): by[lab].append(sp) for lab, v in by.items(): v2 = sorted(v[1:]) or v print(f"[prof] {lab}: median {v2[len(v2) // 2]:.2f} us min {v2[0]:.2f} (n={len(v)})", flush=True) return by def tl_report(risks=("TRISC_1", "BRISC", "NCRISC"), skip=(), core_filter=None): """Round 10: zone timeline (start / end medians and maxima over cores, us from the first kernel start) of the LAST program run in the device-profiler log (call after close_device or ReadDeviceProfiler). core_filter(x, y) on the physical core coords selects cores.""" import os, csv from collections import defaultdict f = os.environ["TT_METAL_PROFILER_DIR"] + "/.logs/profile_log_device.csv" rows = list(csv.reader(open(f)))[2:] ev = defaultdict(list) for r in rows: try: cx, cy, risc, cyc, name, typ, run = int(r[1]), int(r[2]), r[3].strip(), int(r[5]), r[10].strip(), r[11].strip(), r[7].strip() except Exception: continue ev[(run, cx, cy)].append((cyc, risc, name, typ)) last = sorted({k[0] for k in ev}, key=lambda v: int(v) if v.isdigit() else 0)[-1] t00 = min(c for (run, _, _), v in ev.items() if run == last for c, rk, n, t in v if "KERNEL" in n) tl = defaultdict(lambda: defaultdict(list)) for (run, cx, cy), v in ev.items(): if run != last or (core_filter is not None and not core_filter(cx, cy)): continue st = {} for c, rk, n, t in sorted(v): if "START" in t: st[(rk, n)] = c elif "END" in t and (rk, n) in st: tl[rk][n].append(((st.pop((rk, n)) - t00) / 1.35e3, (c - t00) / 1.35e3)) for rk in risks: for n, v in sorted(tl[rk].items(), key=lambda kv: sorted(kv[1])[len(kv[1]) // 2][0]): if n in skip: continue s_ = sorted(a_ for a_, b_ in v); e_ = sorted(b_ for a_, b_ in v) d_ = sorted(b_ - a_ for a_, b_ in v) print(f"{rk:8s} {n:22s} n={len(v):4d} start med {s_[len(v) // 2]:6.2f} max {s_[-1]:6.2f} end med {e_[len(v) // 2]:6.2f} max {e_[-1]:6.2f} dur sum/core-med {sum(d_):7.1f}", flush=True) def tl_seq(names, risk="NCRISC", core_filter=None): """Round 10: per-occurrence timeline of zones ``names`` on ``risk`` (k-th occurrence on each core), medians over cores of start / end (us from the first kernel start) of the LAST program run in the device-profiler log.""" import os, csv from collections import defaultdict f = os.environ["TT_METAL_PROFILER_DIR"] + "/.logs/profile_log_device.csv" rows = list(csv.reader(open(f)))[2:] ev = defaultdict(list) for r in rows: try: cx, cy, rk, cyc, name, typ, run = int(r[1]), int(r[2]), r[3].strip(), int(r[5]), r[10].strip(), r[11].strip(), r[7].strip() except Exception: continue ev[(run, cx, cy)].append((cyc, rk, name, typ)) last = sorted({k[0] for k in ev}, key=lambda v: int(v) if v.isdigit() else 0)[-1] t00 = min(c for (run, _, _), v in ev.items() if run == last for c, rk, n, t in v if "KERNEL" in n) for nm in names: per = defaultdict(list) for (run, cx, cy), v in ev.items(): if run != last or (core_filter is not None and not core_filter(cx, cy)): continue st, k = None, 0 for c, rk, n, t in sorted(v): if rk != risk or n != nm: continue if "START" in t: st = c elif "END" in t and st is not None: per[k].append(((st - t00) / 1.35e3, (c - t00) / 1.35e3)) k += 1 st = None line = [] for k in sorted(per): v = per[k] s_ = sorted(a for a, b in v)[len(v) // 2] d_ = sorted(b - a for a, b in v)[len(v) // 2] line.append(f"{s_:5.1f}+{d_:.2f}") print(f"{risk:7s} {nm:6s} (n={len(per.get(0, []))}): " + " ".join(line), flush=True)