import argparse
import os
import statistics
import subprocess
import sys
import xml.etree.ElementTree as ET
from collections import defaultdict
from typing import Dict, List, Optional, Tuple
XCTRACE = "/Applications/Xcode.app/Contents/Developer/usr/bin/xctrace"
def export_table(trace_path: str, schema: str) -> str:
if not os.path.isdir(trace_path):
raise FileNotFoundError(f"trace bundle not found: {trace_path}")
xpath = f'/trace-toc/run/data/table[@schema="{schema}"]'
proc = subprocess.run(
[XCTRACE, "export", "--input", trace_path, "--xpath", xpath],
check=True,
capture_output=True,
text=True,
)
return proc.stdout
def export_toc(trace_path: str) -> str:
proc = subprocess.run(
[XCTRACE, "export", "--input", trace_path, "--toc"],
check=True,
capture_output=True,
text=True,
)
return proc.stdout
def resolve(elem: ET.Element, table: dict) -> str:
rid = elem.get("id")
ref = elem.get("ref")
if rid is not None:
val = elem.get("fmt") or (elem.text or "")
table[rid] = val
return val
if ref is not None:
return table.get(ref, "")
return elem.get("fmt") or (elem.text or "")
def text_int(elem: ET.Element, table: dict) -> Optional[int]:
rid = elem.get("id")
ref = elem.get("ref")
if rid is not None:
t = (elem.text or "").strip()
try:
v = int(t) if t else int(elem.get("fmt") or "0")
table[rid] = v
return v
except ValueError:
return None
if ref is not None:
return table.get(ref)
t = (elem.text or "").strip()
try:
return int(t) if t else int(elem.get("fmt") or "0")
except ValueError:
return None
def parse_gpu_execution_points(xml_text: str) -> List[dict]:
root = ET.fromstring(xml_text)
rows = []
table = {}
for row in root.iter("row"):
children = list(row)
if len(children) < 5:
continue
t_str = resolve(children[0], table)
t_text = (children[0].text or "").strip()
try:
t_ns = int(t_text) if t_text else int(t_str)
except ValueError:
try:
t_ns = int(t_str)
except ValueError:
continue
chan = resolve(children[1], table)
try:
fn = int(resolve(children[2], table))
except ValueError:
continue
slot = resolve(children[3], table)
sub = resolve(children[4], table)
rows.append(
dict(t_ns=t_ns, channel=chan, fn=fn, slot=slot, sub_id=sub)
)
return rows
def pair_dispatches(rows: List[dict]) -> Tuple[List[dict], int, int]:
starts = {}
paired = []
unpaired_ends = 0
for r in rows:
key = r["sub_id"]
if r["fn"] == 1:
starts[key] = r
elif r["fn"] == 2:
s = starts.pop(key, None)
if s is None:
unpaired_ends += 1
continue
paired.append(dict(
sub_id=key,
channel=s["channel"],
start_ns=s["t_ns"],
end_ns=r["t_ns"],
duration_ns=r["t_ns"] - s["t_ns"],
))
return paired, unpaired_ends, len(starts)
def parse_encoders_list(xml_text: str) -> List[dict]:
root = ET.fromstring(xml_text)
rows = []
table = {}
for row in root.iter("row"):
children = list(row)
if len(children) < 13:
continue
t_text = (children[0].text or "").strip()
try:
t_ns = int(t_text)
except ValueError:
try:
t_ns = int(resolve(children[0], table))
except ValueError:
continue
d_text = (children[1].text or "").strip()
try:
dur_ns = int(d_text) if d_text else int(resolve(children[1], table))
except ValueError:
continue
encoder_label = resolve(children[8], table)
cmdbuffer_label = resolve(children[6], table)
event_type = resolve(children[10], table)
rows.append(dict(
start_ns=t_ns,
duration_ns=dur_ns,
encoder_label=encoder_label,
cmdbuffer_label=cmdbuffer_label,
event_type=event_type,
))
return rows
def parse_shader_list(xml_text: str, target_process_prefix: str) -> List[dict]:
root = ET.fromstring(xml_text)
rows = []
table = {}
for row in root.iter("row"):
children = list(row)
if len(children) < 10:
continue
name = resolve(children[1], table)
pso_name = resolve(children[3], table)
cache_id = resolve(children[4], table)
pc_start = resolve(children[5], table)
pc_end = resolve(children[6], table)
shader_type = resolve(children[7], table)
proc = resolve(children[8], table)
if not proc.startswith(target_process_prefix):
continue
if not name:
continue
rows.append(dict(
name=name,
pso_name=pso_name,
cache_id=cache_id,
pc_start=pc_start,
pc_end=pc_end,
shader_type=shader_type,
process=proc,
))
return rows
def kernel_family(label: str) -> str:
base = label.split("|", 1)[0].split(" (", 1)[0].strip()
if not base.startswith("kernel_"):
return base or "(unknown)"
body = base[len("kernel_"):]
families = [
("mul_mm_id_map0", "moe_map0"),
("mul_mm_id_q4_0_tensor", "moe_q4_0_mm"),
("mul_mm_id_q5_K_tensor", "moe_q5_K_mm"),
("mul_mm_id_q6_K_tensor", "moe_q6_K_mm"),
("mul_mm_id", "moe_mm"),
("mul_mv_id_q4_0", "moe_q4_0_mv"),
("mul_mv_id_q5_K", "moe_q5_K_mv"),
("mul_mv_id_q6_K", "moe_q6_K_mv"),
("mul_mv_id_q8_0", "moe_q8_0_mv"),
("mul_mv_id", "moe_mv"),
("mul_mm_q4_0_tensor", "dense_q4_0_mm"),
("mul_mm_q5_K_tensor", "dense_q5_K_mm"),
("mul_mm_q6_K_tensor", "dense_q6_K_mm"),
("mul_mm_q8_0_tensor", "dense_q8_0_mm"),
("mul_mm", "dense_mm"),
("mul_mv_q4_0", "dense_q4_0_mv"),
("mul_mv_q5_K", "dense_q5_K_mv"),
("mul_mv_q6_K", "dense_q6_K_mv"),
("mul_mv_q8_0", "dense_q8_0_mv"),
("mul_mv", "dense_mv"),
("flash_attn", "flash_attn"),
("rms_norm", "rms_norm"),
("rope", "rope"),
("silu", "silu"),
("swiglu", "swiglu"),
("kv_cache_copy", "kv_cache"),
("argmax", "argmax"),
("permute", "permute"),
("cast", "cast"),
("residual_add", "residual_add"),
("fused_norm", "fused_norm"),
]
for prefix, fam in families:
if body.startswith(prefix):
return fam
return "other_" + body.split("_", 1)[0]
def shader_list_summary(rows: List[dict]) -> dict:
by_family = defaultdict(set)
for r in rows:
fam = kernel_family(r["name"])
by_family[fam].add(r.get("pso_name", "") or r["name"])
out = {}
for fam, names in by_family.items():
out[fam] = sorted(n for n in names if n)
return out
def parse_encoders_list_with_ids(xml_text: str, target_process_prefix: str = "") -> List[dict]:
root = ET.fromstring(xml_text)
rows = []
table: Dict[str, str] = {}
def _register_subtree(elem):
rid = elem.get("id")
if rid is not None:
fmt = elem.get("fmt")
if fmt is not None:
table[rid] = fmt
else:
t = (elem.text or "").strip()
if t:
table[rid] = t
for child in elem:
_register_subtree(child)
text_table: Dict[str, int] = {}
def _register_text(elem):
rid = elem.get("id")
if rid is not None:
t = (elem.text or "").strip()
if t:
try:
text_table[rid] = int(t)
except ValueError:
pass
for child in elem:
_register_text(child)
_register_text(root)
for row in root.iter("row"):
_register_subtree(row)
children = list(row)
if len(children) < 13:
continue
t_text = (children[0].text or "").strip()
try:
if t_text:
t_ns = int(t_text)
else:
ref = children[0].get("ref")
if ref is not None and ref in text_table:
t_ns = text_table[ref]
else:
t_ns = int(resolve(children[0], table))
except ValueError:
continue
d_text = (children[1].text or "").strip()
try:
if d_text:
dur_ns = int(d_text)
else:
ref = children[1].get("ref")
if ref is not None and ref in text_table:
dur_ns = text_table[ref]
else:
dur_ns = int(resolve(children[1], table))
except ValueError:
continue
proc_elem = children[3]
proc_str = resolve(proc_elem, table) or ""
if not proc_str:
ref = proc_elem.get("ref")
if ref is not None:
proc_str = table.get(ref, "") or ""
encoder_label = resolve(children[8], table)
cmdbuffer_label = resolve(children[6], table)
event_type = resolve(children[10], table)
cmdbuf_id = None
enc_id = None
if len(children) > 11:
cmdbuf_id = resolve(children[11], table)
if len(children) > 12:
enc_id = resolve(children[12], table)
if target_process_prefix and not proc_str.startswith(target_process_prefix):
continue
rows.append(dict(
start_ns=t_ns,
duration_ns=dur_ns,
encoder_label=encoder_label,
cmdbuffer_label=cmdbuffer_label,
event_type=event_type,
cmdbuffer_id=cmdbuf_id,
encoder_id=enc_id,
process=proc_str,
))
return rows
def encoder_family(enc_label: str) -> str:
if not enc_label:
return "(unknown)"
if enc_label.startswith("Compute Command"):
return "compute"
if enc_label.startswith("Blit Command"):
return "blit"
if enc_label.startswith("Render Command"):
return "render"
if enc_label.startswith("Acceleration"):
return "accel"
return enc_label.split(" ")[0].lower() or "(unknown)"
def parse_submission_to_encoder_map(xml_text: str, target_process_prefix: str = "") -> Dict[str, str]:
root = ET.fromstring(xml_text)
out: Dict[str, str] = {}
table: Dict[str, str] = {}
def _register_subtree(elem):
rid = elem.get("id")
if rid is not None:
fmt = elem.get("fmt")
if fmt is not None:
table[rid] = fmt
else:
t = (elem.text or "").strip()
if t:
table[rid] = t
for child in elem:
_register_subtree(child)
for row in root.iter("row"):
_register_subtree(row)
children = list(row)
if len(children) < 13:
continue
sub_id = resolve(children[2], table) or ""
enc_id = resolve(children[5], table) or ""
proc_elem = children[12]
proc_str = resolve(proc_elem, table) or ""
if not proc_str:
ref = proc_elem.get("ref")
if ref is not None:
proc_str = table.get(ref, "") or ""
if target_process_prefix and not proc_str.startswith(target_process_prefix):
continue
if sub_id and enc_id:
out[sub_id] = enc_id
return out
def encoder_gpu_summary(
encoders: List[dict],
paired_dispatches: List[dict],
sub_to_enc: Dict[str, str],
) -> Dict[str, dict]:
enc_by_id: Dict[str, str] = {}
enc_count_by_family: Dict[str, int] = defaultdict(int)
enc_host_sum_by_family: Dict[str, int] = defaultdict(int)
seen_ids = set()
for e in encoders:
eid = e.get("encoder_id")
if not eid or eid in seen_ids:
continue
seen_ids.add(eid)
fam = encoder_family(e.get("encoder_label", ""))
enc_by_id[eid] = fam
enc_count_by_family[fam] += 1
enc_host_sum_by_family[fam] += e.get("duration_ns", 0)
enc_gpu_sum_by_family: Dict[str, int] = defaultdict(int)
enc_disp_count_by_family: Dict[str, int] = defaultdict(int)
matched = 0
unmatched = 0
unmapped = 0
for p in paired_dispatches:
sid = p.get("sub_id")
eid = sub_to_enc.get(sid)
if eid is None:
unmapped += 1
continue
fam = enc_by_id.get(eid)
if fam is None:
unmatched += 1
continue
matched += 1
enc_gpu_sum_by_family[fam] += p.get("duration_ns", 0)
enc_disp_count_by_family[fam] += 1
out: Dict[str, dict] = {}
families = set(enc_by_id.values())
families.update(enc_gpu_sum_by_family.keys())
for fam in sorted(families):
out[fam] = dict(
count_encoders=enc_count_by_family.get(fam, 0),
count_dispatches_in_encoder_family=enc_disp_count_by_family.get(fam, 0),
host_sum_ns=enc_host_sum_by_family.get(fam, 0),
gpu_sum_ns=enc_gpu_sum_by_family.get(fam, 0),
)
out["_meta"] = dict(
matched_dispatches=matched,
unmatched_dispatches=unmatched, unmapped_dispatches=unmapped, total_encoders=len(seen_ids),
)
return out
def median_encoder_summaries(per_trial: List[Dict[str, dict]], n_tokens_list: List[int]) -> Dict[str, dict]:
if not per_trial:
return {}
families = set()
for s in per_trial:
families.update(k for k in s.keys() if not k.startswith("_"))
out: Dict[str, dict] = {}
for fam in sorted(families):
gpu_us_per_tok = []
host_us_per_tok = []
n_enc_per_tok = []
for s, n_tok in zip(per_trial, n_tokens_list):
n_tok = max(n_tok, 1)
b = s.get(fam) or {}
gpu_us_per_tok.append(b.get("gpu_sum_ns", 0) / 1000.0 / n_tok)
host_us_per_tok.append(b.get("host_sum_ns", 0) / 1000.0 / n_tok)
n_enc_per_tok.append(b.get("count_encoders", 0) / n_tok)
out[fam] = dict(
median_gpu_us_per_token=statistics.median(gpu_us_per_tok),
median_host_us_per_token=statistics.median(host_us_per_tok),
median_encoders_per_token=statistics.median(n_enc_per_tok),
gpu_us_per_token_per_trial=gpu_us_per_tok,
)
return out
def normalize_phase_label(label: str) -> str:
if not label:
return "(unknown)"
if label.startswith("[") and "] " in label:
label = label.split("] ", 1)[1]
parts = label.split(".")
parts = [p for p in parts if not p.isdigit()]
return ".".join(parts) if parts else label
def cb_label_gpu_summary(
encoders: List[dict],
paired_dispatches: List[dict],
sub_to_enc: Dict[str, str],
) -> Dict[str, dict]:
enc_to_phase: Dict[str, str] = {}
enc_to_host_ns: Dict[str, int] = {}
seen_enc = set()
cb_count_by_phase: Dict[str, int] = defaultdict(int)
enc_host_sum_by_phase: Dict[str, int] = defaultdict(int)
for e in encoders:
eid = e.get("encoder_id")
if not eid or eid in seen_enc:
continue
seen_enc.add(eid)
phase = normalize_phase_label(e.get("cmdbuffer_label", "") or "")
enc_to_phase[eid] = phase
enc_to_host_ns[eid] = e.get("duration_ns", 0)
cb_count_by_phase[phase] += 1
enc_host_sum_by_phase[phase] += e.get("duration_ns", 0)
enc_gpu_sum_by_phase: Dict[str, int] = defaultdict(int)
enc_disp_count_by_phase: Dict[str, int] = defaultdict(int)
matched = 0
unmapped = 0
unmatched_phase = 0
for p in paired_dispatches:
sid = p.get("sub_id")
eid = sub_to_enc.get(sid)
if eid is None:
unmapped += 1
continue
phase = enc_to_phase.get(eid)
if phase is None:
unmatched_phase += 1
continue
matched += 1
enc_gpu_sum_by_phase[phase] += p.get("duration_ns", 0)
enc_disp_count_by_phase[phase] += 1
out: Dict[str, dict] = {}
phases = set(enc_to_phase.values())
phases.update(enc_gpu_sum_by_phase.keys())
for phase in sorted(phases):
out[phase] = dict(
count_cbs=cb_count_by_phase.get(phase, 0),
count_dispatches=enc_disp_count_by_phase.get(phase, 0),
host_sum_ns=enc_host_sum_by_phase.get(phase, 0),
gpu_sum_ns=enc_gpu_sum_by_phase.get(phase, 0),
)
out["_meta"] = dict(
matched_dispatches=matched,
unmapped_dispatches=unmapped,
unmatched_phase_dispatches=unmatched_phase,
total_encoders=len(seen_enc),
labelled_encoders=sum(
1 for p in enc_to_phase.values()
if p and not p.startswith("Command Buffer")
and not p.startswith("Compute Command")
),
)
return out
def median_cb_label_summaries(
per_trial: List[Dict[str, dict]], n_tokens_list: List[int]
) -> Dict[str, dict]:
if not per_trial:
return {}
phases = set()
for s in per_trial:
phases.update(k for k in s.keys() if not k.startswith("_"))
out: Dict[str, dict] = {}
for phase in sorted(phases):
gpu_us_per_tok = []
host_us_per_tok = []
cbs_per_tok = []
disps_per_tok = []
for s, n_tok in zip(per_trial, n_tokens_list):
n_tok = max(n_tok, 1)
b = s.get(phase) or {}
gpu_us_per_tok.append(b.get("gpu_sum_ns", 0) / 1000.0 / n_tok)
host_us_per_tok.append(b.get("host_sum_ns", 0) / 1000.0 / n_tok)
cbs_per_tok.append(b.get("count_cbs", 0) / n_tok)
disps_per_tok.append(b.get("count_dispatches", 0) / n_tok)
out[phase] = dict(
median_gpu_us_per_token=statistics.median(gpu_us_per_tok),
median_host_us_per_token=statistics.median(host_us_per_tok),
median_cbs_per_token=statistics.median(cbs_per_tok),
median_dispatches_per_token=statistics.median(disps_per_tok),
mean_us_per_cb=(
statistics.mean(gpu_us_per_tok) / max(statistics.mean(cbs_per_tok), 1e-9)
if cbs_per_tok and any(c > 0 for c in cbs_per_tok)
else 0.0
),
gpu_us_per_token_per_trial=gpu_us_per_tok,
)
return out
BUCKETS = [
("xs_<2us", 0, 2_000),
("sm_2_8us", 2_000, 8_000),
("md_8_32us", 8_000, 32_000),
("lg_32_80us", 32_000, 80_000),
("xl_>=80us", 80_000, None),
]
def bucket_of(dur_ns: int) -> str:
for name, lo, hi in BUCKETS:
if dur_ns >= lo and (hi is None or dur_ns < hi):
return name
return "unknown"
def bucket_summary(paired: List[dict]) -> Dict[str, dict]:
by_bucket = defaultdict(list)
for p in paired:
by_bucket[bucket_of(p["duration_ns"])].append(p["duration_ns"])
out = {}
for name, _, _ in BUCKETS:
durs = by_bucket.get(name, [])
out[name] = dict(
count=len(durs),
sum_ns=sum(durs),
p50_ns=int(statistics.median(durs)) if durs else 0,
p95_ns=int(durs[int(0.95 * (len(durs) - 1))]) if len(durs) >= 2 else (durs[0] if durs else 0),
mean_ns=int(sum(durs) / len(durs)) if durs else 0,
)
out["_total"] = dict(
count=len(paired),
sum_ns=sum(p["duration_ns"] for p in paired),
)
return out
def summarize_trace(trace_path: str, n_tokens: int, target_process_prefix: str = "") -> dict:
xml = export_table(trace_path, "metal-gpu-execution-points")
rows = parse_gpu_execution_points(xml)
paired, unpaired_ends, leftover = pair_dispatches(rows)
enc_xml = export_table(trace_path, "metal-application-encoders-list")
encoders = parse_encoders_list(enc_xml)
encoders_filtered = parse_encoders_list_with_ids(
enc_xml, target_process_prefix=target_process_prefix or ""
)
sub_to_enc: Dict[str, str] = {}
try:
map_xml = export_table(trace_path, "metal-gpu-submission-to-command-buffer-id")
sub_to_enc = parse_submission_to_encoder_map(
map_xml, target_process_prefix=target_process_prefix or ""
)
except Exception:
pass
encoder_gpu = encoder_gpu_summary(encoders_filtered, paired, sub_to_enc)
cb_label_gpu = cb_label_gpu_summary(encoders_filtered, paired, sub_to_enc)
shader_registry: Dict[str, List[str]] = {}
shader_count = 0
try:
sl_xml = export_table(trace_path, "metal-shader-profiler-shader-list")
sl_rows = parse_shader_list(sl_xml, target_process_prefix or "")
shader_count = len(sl_rows)
shader_registry = shader_list_summary(sl_rows)
except Exception:
pass
shader_timeline_rows = 0
try:
st_xml = export_table(trace_path, "metal-shader-profiler-intervals")
shader_timeline_rows = st_xml.count("<row")
except Exception:
pass
buckets = bucket_summary(paired)
total_gpu_ns = buckets["_total"]["sum_ns"]
total_dispatches = buckets["_total"]["count"]
return dict(
path=trace_path,
rows=len(rows),
paired=len(paired),
unpaired_ends=unpaired_ends,
leftover_starts=leftover,
encoders=len(encoders),
encoder_total_ns=sum(e["duration_ns"] for e in encoders),
n_tokens=n_tokens,
buckets=buckets,
dispatches_per_token=total_dispatches / max(n_tokens, 1),
gpu_us_per_token=total_gpu_ns / 1000.0 / max(n_tokens, 1),
shader_registry=shader_registry,
shader_count=shader_count,
shader_timeline_rows=shader_timeline_rows,
encoder_filtered_count=len(encoders_filtered),
encoder_gpu=encoder_gpu,
cb_label_gpu=cb_label_gpu,
)
def median_summaries(summaries: List[dict]) -> dict:
if not summaries:
return {}
registry_union: Dict[str, set] = defaultdict(set)
for s in summaries:
for fam, names in (s.get("shader_registry") or {}).items():
registry_union[fam].update(names)
registry_out = {fam: sorted(names) for fam, names in registry_union.items()}
shader_timeline_rows_max = max(
(s.get("shader_timeline_rows", 0) for s in summaries), default=0
)
encoder_gpu_per_trial = [s.get("encoder_gpu", {}) for s in summaries]
n_tokens_per_trial = [s["n_tokens"] for s in summaries]
encoder_gpu_med = median_encoder_summaries(encoder_gpu_per_trial, n_tokens_per_trial)
cb_label_gpu_per_trial = [s.get("cb_label_gpu", {}) for s in summaries]
cb_label_gpu_med = median_cb_label_summaries(cb_label_gpu_per_trial, n_tokens_per_trial)
out = {
"n_trials": len(summaries),
"n_tokens_per_trial": n_tokens_per_trial,
"paired_per_trial": [s["paired"] for s in summaries],
"gpu_us_per_token_per_trial": [s["gpu_us_per_token"] for s in summaries],
"median_dispatches_per_token": statistics.median(
[s["dispatches_per_token"] for s in summaries]
),
"median_gpu_us_per_token": statistics.median(
[s["gpu_us_per_token"] for s in summaries]
),
"buckets": {},
"shader_registry": registry_out,
"shader_timeline_rows": shader_timeline_rows_max,
"encoder_gpu": encoder_gpu_med,
"encoder_gpu_per_trial": encoder_gpu_per_trial,
"cb_label_gpu": cb_label_gpu_med,
"cb_label_gpu_per_trial": cb_label_gpu_per_trial,
}
for name, _, _ in BUCKETS:
counts_per_tok = []
sums_per_tok_us = []
p50_us = []
for s in summaries:
n_tok = max(s["n_tokens"], 1)
b = s["buckets"].get(name, {})
counts_per_tok.append(b.get("count", 0) / n_tok)
sums_per_tok_us.append(b.get("sum_ns", 0) / 1000.0 / n_tok)
p50_us.append(b.get("p50_ns", 0) / 1000.0)
out["buckets"][name] = dict(
median_dispatches_per_token=statistics.median(counts_per_tok),
median_us_per_token=statistics.median(sums_per_tok_us),
median_p50_us_per_dispatch=statistics.median(p50_us),
)
return out
def fmt_int(v) -> str:
if isinstance(v, float):
if v >= 100:
return f"{v:>10.1f}"
return f"{v:>10.3f}"
return f"{v:>10}"
def write_report(out_path: str, hf2q: dict, llama: dict, hf2q_trials: List[dict], llama_trials: List[dict]):
lines = []
lines.append("=" * 110)
lines.append("ADR-015 iter9/iter11 — Q4_0 dispatch attribution (xctrace MST)")
lines.append("=" * 110)
lines.append("")
lines.append("Methodology:")
lines.append(" - canonical frame: metal-gpu-execution-points fn=1/2 paired by sub_id (per AC2)")
lines.append(" - encoders sidecar: metal-application-encoders-list (informational; not summed)")
lines.append(" - iter11 status: kernel REGISTRY surfaced via metal-shader-profiler-shader-list")
lines.append(" (now populated post-iter9b labels at mlx-native@a7d2b95). Per-dispatch")
lines.append(" PSO→duration JOIN STILL BLOCKED: no per-dispatch table carries pso-id, and")
lines.append(" Shader Timeline (the metal-shader-profiler-intervals row source) cannot be")
lines.append(" enabled from xctrace CLI. iter11 verified 4 incantations:")
lines.append(" (a) default `Metal System Trace`")
lines.append(" (b) MST + --instrument 'Metal GPU Counters' / 'Metal Performance Overview'")
lines.append(" + --instrument 'Advanced Graphics Statistics'")
lines.append(" (c) MST + (b) + --instrument 'Metal Application' + --instrument 'GPU'")
lines.append(" (d) `Game Performance` template")
lines.append(" All produce the kernel-name registry but ZERO Shader Timeline samples.")
lines.append(" Recommended pivot: iter11b enabler = mlx-native pushDebugGroup(label) +")
lines.append(" popDebugGroup() around each kernel dispatch in src/encoder.rs.")
lines.append(" - bucketing strategy (best-available CLI signal): per-dispatch duration")
lines.append(" histogram into 5 bands, where each band cleanly maps to a kernel class on")
lines.append(" the dwq46 decode workload:")
lines.append(" xs_<2us : rms_norm, scalar mul, reshape")
lines.append(" sm_2_8us : rope, soft-cap, small mat-vec")
lines.append(" md_8_32us : Q4_0 MoE mat-vec_id (gate/up/down), dense Q4_0 mat-vec")
lines.append(" lg_32_80us : flash_attn, pooled mul_mm_id")
lines.append(" xl_>=80us : prefill mul_mm_id, lm_head, large blits")
lines.append("")
lines.append("Inputs:")
lines.append(f" hf2q trials: {len(hf2q_trials)}")
for s in hf2q_trials:
lines.append(f" - {os.path.basename(s['path'])}: paired={s['paired']:>6d} dispatches "
f"({s['dispatches_per_token']:.1f}/tok), gpu={s['gpu_us_per_token']:.1f} µs/tok")
lines.append(f" llama trials: {len(llama_trials)}")
for s in llama_trials:
lines.append(f" - {os.path.basename(s['path'])}: paired={s['paired']:>6d} dispatches "
f"({s['dispatches_per_token']:.1f}/tok), gpu={s['gpu_us_per_token']:.1f} µs/tok")
lines.append("")
if hf2q and llama:
lines.append("=" * 110)
lines.append("Side-by-side bucketed attribution (medians across trials)")
lines.append("=" * 110)
header = (f"{'BUCKET':<14s} "
f"{'hf2q disp/tok':>14s} {'hf2q µs/disp':>14s} {'hf2q µs/tok':>14s} "
f"{'llama disp/tok':>15s} {'llama µs/disp':>15s} {'llama µs/tok':>14s} "
f"{'Δµs/tok':>10s} {'Δ%':>8s}")
lines.append(header)
lines.append("-" * len(header))
for name, _, _ in BUCKETS:
hb = hf2q["buckets"][name]
lb = llama["buckets"][name]
d_us = hb["median_us_per_token"] - lb["median_us_per_token"]
d_pct = (d_us / lb["median_us_per_token"] * 100) if lb["median_us_per_token"] > 0 else 0.0
lines.append(
f"{name:<14s} "
f"{hb['median_dispatches_per_token']:>14.1f} "
f"{hb['median_p50_us_per_dispatch']:>14.2f} "
f"{hb['median_us_per_token']:>14.1f} "
f"{lb['median_dispatches_per_token']:>15.1f} "
f"{lb['median_p50_us_per_dispatch']:>15.2f} "
f"{lb['median_us_per_token']:>14.1f} "
f"{d_us:>+10.1f} "
f"{d_pct:>+7.1f}%"
)
lines.append("-" * len(header))
lines.append(
f"{'TOTAL':<14s} "
f"{hf2q['median_dispatches_per_token']:>14.1f} "
f"{'-':>14s} "
f"{hf2q['median_gpu_us_per_token']:>14.1f} "
f"{llama['median_dispatches_per_token']:>15.1f} "
f"{'-':>15s} "
f"{llama['median_gpu_us_per_token']:>14.1f} "
f"{(hf2q['median_gpu_us_per_token'] - llama['median_gpu_us_per_token']):>+10.1f} "
f"{((hf2q['median_gpu_us_per_token'] - llama['median_gpu_us_per_token']) / llama['median_gpu_us_per_token'] * 100):>+7.1f}%"
)
lines.append("")
lines.append("Q4_0-attributable summary (md_8_32us bucket — Q4_0 MoE mat-vec_id territory):")
hb = hf2q["buckets"]["md_8_32us"]
lb = llama["buckets"]["md_8_32us"]
lines.append(f" hf2q : {hb['median_dispatches_per_token']:.1f} disp/tok × {hb['median_p50_us_per_dispatch']:.2f} µs/disp = {hb['median_us_per_token']:.1f} µs/tok")
lines.append(f" llama: {lb['median_dispatches_per_token']:.1f} disp/tok × {lb['median_p50_us_per_dispatch']:.2f} µs/disp = {lb['median_us_per_token']:.1f} µs/tok")
d_us = hb["median_us_per_token"] - lb["median_us_per_token"]
d_pct_of_total = d_us / max(llama["median_gpu_us_per_token"], 1) * 100
lines.append(f" delta: {d_us:+.1f} µs/tok ({d_pct_of_total:+.2f}% of llama wall)")
lines.append("")
lines.append("Iter10 attack target (largest positive Δµs/tok bucket):")
target = max(BUCKETS, key=lambda b: hf2q["buckets"][b[0]]["median_us_per_token"] - llama["buckets"][b[0]]["median_us_per_token"])
tname = target[0]
d_us = hf2q["buckets"][tname]["median_us_per_token"] - llama["buckets"][tname]["median_us_per_token"]
lines.append(f" bucket: {tname}")
lines.append(f" Δµs/tok: {d_us:+.1f}")
lines.append(f" Likely kernel class: {bucket_kernel_hint(tname)}")
lines.append("")
elif hf2q:
lines.append("=" * 110)
lines.append("hf2q-only partial attribution (llama traces not yet available)")
lines.append("=" * 110)
header = f"{'BUCKET':<14s} {'disp/tok':>10s} {'µs/disp p50':>14s} {'µs/tok':>10s}"
lines.append(header)
lines.append("-" * len(header))
for name, _, _ in BUCKETS:
hb = hf2q["buckets"][name]
lines.append(
f"{name:<14s} "
f"{hb['median_dispatches_per_token']:>10.1f} "
f"{hb['median_p50_us_per_dispatch']:>14.2f} "
f"{hb['median_us_per_token']:>10.1f}"
)
lines.append(
f"{'TOTAL':<14s} "
f"{hf2q['median_dispatches_per_token']:>10.1f} "
f"{'-':>14s} "
f"{hf2q['median_gpu_us_per_token']:>10.1f}"
)
lines.append("")
else:
lines.append("(no traces summarised)")
if hf2q and llama and (hf2q.get("encoder_gpu") or llama.get("encoder_gpu")):
lines.append("")
lines.append("=" * 110)
lines.append("iter12 — Per-encoder GPU-time attribution (xctrace MST CLI-only)")
lines.append("=" * 110)
lines.append("Methodology: encoders bucketed by family (compute / blit / render / accel),")
lines.append(" filtered to target binary's process. GPU time = sum of paired dispatch")
lines.append(" durations joined to the encoder by THREADING the join through the")
lines.append(" metal-gpu-submission-to-command-buffer-id table (sub_id -> encoder_id),")
lines.append(" because sub_id and encoder_id live in different id namespaces (sub_id is")
lines.append(" a 32-bit GPU submission counter; encoder_id is a 40-bit MTLObject id).")
lines.append(" Host time = encoder lifetime from metal-application-encoders-list")
lines.append(" (encoding wall-clock, not GPU wall-clock — kept for reference).")
lines.append("")
lines.append("Granularity caveat: both llama.cpp and mlx-native emit only generic encoder")
lines.append(" labels ('Compute Command N' / 'Blit Command N') — neither pushes debug")
lines.append(" groups nor sets MTLObject labels. Per-encoder attribution is therefore")
lines.append(" COARSER than per-kernel; on a typical decode token both binaries emit a")
lines.append(" single Compute encoder containing many dispatches, so per-encoder GPU sum")
lines.append(" is roughly per-CB GPU sum.")
lines.append("")
header = (f"{'family':<10s} "
f"{'hf2q enc/tok':>13s} {'hf2q gpu_µs/tok':>15s} "
f"{'llama enc/tok':>14s} {'llama gpu_µs/tok':>16s} "
f"{'Δgpu_µs/tok':>12s} {'Δ%':>8s}")
lines.append(header)
lines.append("-" * len(header))
all_families = sorted(
set((hf2q.get("encoder_gpu") or {}).keys())
| set((llama.get("encoder_gpu") or {}).keys())
)
rows_sorted = []
for fam in all_families:
hb = (hf2q.get("encoder_gpu") or {}).get(fam) or {}
lb = (llama.get("encoder_gpu") or {}).get(fam) or {}
h_gpu = hb.get("median_gpu_us_per_token", 0.0)
l_gpu = lb.get("median_gpu_us_per_token", 0.0)
d_us = h_gpu - l_gpu
d_pct = (d_us / l_gpu * 100) if l_gpu > 0 else 0.0
rows_sorted.append((fam, hb, lb, d_us, d_pct))
rows_sorted.sort(key=lambda r: -r[3])
for fam, hb, lb, d_us, d_pct in rows_sorted:
lines.append(
f"{fam:<10s} "
f"{hb.get('median_encoders_per_token', 0):>13.2f} "
f"{hb.get('median_gpu_us_per_token', 0):>15.1f} "
f"{lb.get('median_encoders_per_token', 0):>14.2f} "
f"{lb.get('median_gpu_us_per_token', 0):>16.1f} "
f"{d_us:>+12.1f} "
f"{d_pct:>+7.1f}%"
)
h_total = sum(b.get("median_gpu_us_per_token", 0)
for b in (hf2q.get("encoder_gpu") or {}).values())
l_total = sum(b.get("median_gpu_us_per_token", 0)
for b in (llama.get("encoder_gpu") or {}).values())
lines.append("-" * len(header))
lines.append(
f"{'TOTAL':<10s} "
f"{'-':>13s} "
f"{h_total:>15.1f} "
f"{'-':>14s} "
f"{l_total:>16.1f} "
f"{(h_total - l_total):>+12.1f} "
f"{((h_total - l_total) / l_total * 100 if l_total else 0):>+7.1f}%"
)
lines.append("")
lines.append("Per-trial encoder gpu µs/tok (compute family only) for stat visibility:")
for fam in ["compute", "blit"]:
for trace_label, side in [("hf2q ", hf2q), ("llama", llama)]:
if not side:
continue
rows_pt = side.get("encoder_gpu_per_trial") or []
vals = []
for s, n_tok in zip(rows_pt, side.get("n_tokens_per_trial") or []):
if not s:
continue
b = s.get(fam) or {}
vals.append(b.get("gpu_sum_ns", 0) / 1000.0 / max(n_tok, 1))
if vals:
lines.append(f" {trace_label} {fam:<7s}: "
f"{', '.join(f'{x:.1f}' for x in vals)}")
lines.append("")
if hf2q and (hf2q.get("cb_label_gpu") or (llama or {}).get("cb_label_gpu")):
lines.append("")
lines.append("=" * 110)
lines.append("iter16 — Per-CB semantic-phase attribution (xctrace MST CLI-only)")
lines.append("=" * 110)
lines.append("Methodology: hf2q's `mlx_native::CommandEncoder::commit_*labeled(label)`")
lines.append(" now propagates the semantic phase string to MTLCommandBuffer.label and")
lines.append(" the active MTLComputeCommandEncoder.label, populating xctrace's")
lines.append(" `metal-application-encoders-list.cmdbuffer-label` column. Phases")
lines.append(" are joined to per-dispatch GPU duration via")
lines.append(" metal-gpu-submission-to-command-buffer-id (sub_id -> encoder_id) ->")
lines.append(" metal-application-encoders-list (encoder_id -> cmdbuffer_label).")
lines.append("")
lines.append("Comparable-axis caveat: llama.cpp does NOT setLabel on its CBs (verified")
lines.append(" iter15 Phase 0; iter16 §A.2 Phase 0 probe re-confirmed). llama rows here")
lines.append(" bucket under generic 'Command Buffer N' phase names — so this table")
lines.append(" shows hf2q's INTERNAL distribution across phases (the actionable signal")
lines.append(" for ranking iter17 hypotheses) and llama's TOTAL as a single anchor.")
lines.append("")
try:
sample_per_trial = hf2q.get("cb_label_gpu_per_trial") or []
if sample_per_trial:
meta = sample_per_trial[0].get("_meta", {})
lines.append(
f"hf2q label coverage (trial 0): {meta.get('labelled_encoders', 0)}"
f" / {meta.get('total_encoders', 0)} encoders carry a semantic label"
)
lines.append("")
except Exception:
pass
header = (f"{'phase':<48s} "
f"{'hf2q cbs/tok':>13s} {'hf2q disp/tok':>14s} {'hf2q gpu_µs/tok':>15s} "
f"{'llama cbs/tok':>14s} {'llama gpu_µs/tok':>16s} "
f"{'Δgpu_µs/tok':>12s}")
lines.append(header)
lines.append("-" * len(header))
all_phases = sorted(
set((hf2q.get("cb_label_gpu") or {}).keys())
| set((llama or {}).get("cb_label_gpu", {}).keys() if llama else [])
)
rows_sorted = []
for phase in all_phases:
hb = (hf2q.get("cb_label_gpu") or {}).get(phase) or {}
lb = ((llama or {}).get("cb_label_gpu") or {}).get(phase) or {}
h_gpu = hb.get("median_gpu_us_per_token", 0.0)
l_gpu = lb.get("median_gpu_us_per_token", 0.0)
d_us = h_gpu - l_gpu
rows_sorted.append((phase, hb, lb, d_us))
rows_sorted.sort(key=lambda r: -r[1].get("median_gpu_us_per_token", 0.0))
for phase, hb, lb, d_us in rows_sorted:
lines.append(
f"{phase[:48]:<48s} "
f"{hb.get('median_cbs_per_token', 0):>13.2f} "
f"{hb.get('median_dispatches_per_token', 0):>14.2f} "
f"{hb.get('median_gpu_us_per_token', 0):>15.1f} "
f"{lb.get('median_cbs_per_token', 0):>14.2f} "
f"{lb.get('median_gpu_us_per_token', 0):>16.1f} "
f"{d_us:>+12.1f}"
)
h_total = sum(b.get("median_gpu_us_per_token", 0)
for b in (hf2q.get("cb_label_gpu") or {}).values())
l_total = sum(b.get("median_gpu_us_per_token", 0)
for b in ((llama or {}).get("cb_label_gpu") or {}).values())
lines.append("-" * len(header))
lines.append(
f"{'TOTAL':<48s} "
f"{'-':>13s} "
f"{'-':>14s} "
f"{h_total:>15.1f} "
f"{'-':>14s} "
f"{l_total:>16.1f} "
f"{(h_total - l_total):>+12.1f}"
)
lines.append("")
labelled_rows = [
r for r in rows_sorted
if r[0] and not r[0].startswith("Command Buffer")
and not r[0].startswith("Compute Command")
and r[0] != "(unknown)"
]
if labelled_rows:
lines.append("iter17 candidate ranking — top-3 hf2q phases by gpu_µs/token:")
for phase, hb, _lb, _d in labelled_rows[:3]:
cbs = hb.get("median_cbs_per_token", 0.0)
disp = hb.get("median_dispatches_per_token", 0.0)
gpu = hb.get("median_gpu_us_per_token", 0.0)
mean_per_cb = hb.get("mean_us_per_cb", 0.0)
lines.append(
f" {phase}: {cbs:.2f} cbs/tok × ~{mean_per_cb:.1f} µs/cb"
f" = {gpu:.1f} gpu_µs/tok ({disp:.1f} dispatches/tok)"
)
lines.append("")
lines.append("")
lines.append("=" * 110)
lines.append("iter11 — Kernel registry (metal-shader-profiler-shader-list, post-iter9b labels)")
lines.append("=" * 110)
if hf2q and hf2q.get("shader_registry"):
lines.append("")
lines.append("hf2q registry (PSO labels by family, deduped):")
for fam in sorted(hf2q["shader_registry"].keys()):
names = hf2q["shader_registry"][fam]
lines.append(f" {fam:>20s} ({len(names):>2d}): {', '.join(names[:5])}"
f"{' …' if len(names) > 5 else ''}")
lines.append(f" Shader Timeline samples (metal-shader-profiler-intervals): "
f"{hf2q.get('shader_timeline_rows', 0)} rows "
f"({'EMPTY (CLI cannot toggle)' if hf2q.get('shader_timeline_rows', 0) == 0 else 'populated'})")
if llama and llama.get("shader_registry"):
lines.append("")
lines.append("llama-cli registry (PSO labels by family, deduped):")
for fam in sorted(llama["shader_registry"].keys()):
names = llama["shader_registry"][fam]
lines.append(f" {fam:>20s} ({len(names):>2d}): {', '.join(names[:5])}"
f"{' …' if len(names) > 5 else ''}")
lines.append(f" Shader Timeline samples: "
f"{llama.get('shader_timeline_rows', 0)} rows")
lines.append("")
lines.append("Verdict: kernel-NAME attribution per dispatch is BLOCKED on Shader Timeline")
lines.append("toggle which xctrace CLI cannot enable. iter11b enabler (mlx-native")
lines.append("pushDebugGroup) is the recommended unblock; expected to populate")
lines.append("metal-application-event-interval with per-dispatch labeled intervals")
lines.append("joinable to GPU duration via canonical fn=1/2 sub_id pairs.")
lines.append("")
if hf2q and hf2q.get("gpu_us_per_token_per_trial"):
lines.append(f"hf2q per-trial gpu µs/tok: "
f"{', '.join(f'{x:.1f}' for x in hf2q['gpu_us_per_token_per_trial'])}")
if llama and llama.get("gpu_us_per_token_per_trial"):
lines.append(f"llama per-trial gpu µs/tok: "
f"{', '.join(f'{x:.1f}' for x in llama['gpu_us_per_token_per_trial'])}")
lines.append("")
text = "\n".join(lines) + "\n"
with open(out_path, "w") as f:
f.write(text)
sys.stdout.write(text)
def bucket_kernel_hint(name: str) -> str:
return {
"xs_<2us": "rms_norm / reshape / scalar",
"sm_2_8us": "rope / soft-cap / small mat-vec",
"md_8_32us": "Q4_0 MoE mat-vec_id (gate/up/down) — primary Q4_0 attack surface",
"lg_32_80us": "flash_attn / pooled mul_mm_id",
"xl_>=80us": "prefill mul_mm_id / lm_head / large blits",
}.get(name, "unknown")
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--hf2q-trace", action="append", default=[], help="hf2q .trace bundle (repeatable)")
ap.add_argument("--llama-trace", action="append", default=[], help="llama .trace bundle (repeatable)")
ap.add_argument("--n-tokens", type=int, default=64, help="decode tokens per trial (default 64)")
ap.add_argument("--output", default="/tmp/adr015-iter9/aggregate-q4_0.txt")
ap.add_argument("--toc-dump", default=None, help="if set, dump xctrace --toc to this path for one trace")
args = ap.parse_args()
if args.toc_dump and args.hf2q_trace:
with open(args.toc_dump, "w") as f:
f.write(export_toc(args.hf2q_trace[0]))
print(f"toc dumped: {args.toc_dump}", file=sys.stderr)
hf2q_trials = []
for t in args.hf2q_trace:
try:
s = summarize_trace(t, args.n_tokens, target_process_prefix="hf2q")
hf2q_trials.append(s)
print(
f"ok: hf2q {t}: {s['paired']} paired, "
f"{s.get('shader_count', 0)} shaders, "
f"{s.get('shader_timeline_rows', 0)} timeline samples",
file=sys.stderr,
)
except Exception as e:
print(f"WARN: hf2q {t}: {e}", file=sys.stderr)
llama_trials = []
for t in args.llama_trace:
try:
s = summarize_trace(t, args.n_tokens, target_process_prefix="llama")
llama_trials.append(s)
print(
f"ok: llama {t}: {s['paired']} paired, "
f"{s.get('shader_count', 0)} shaders, "
f"{s.get('shader_timeline_rows', 0)} timeline samples",
file=sys.stderr,
)
except Exception as e:
print(f"WARN: llama {t}: {e}", file=sys.stderr)
hf2q_med = median_summaries(hf2q_trials)
llama_med = median_summaries(llama_trials)
os.makedirs(os.path.dirname(args.output), exist_ok=True)
write_report(args.output, hf2q_med, llama_med, hf2q_trials, llama_trials)
print(f"\nwrote: {args.output}", file=sys.stderr)
if __name__ == "__main__":
main()