"""Merge xperf's classic kernel events with native TDH events on their QPC clock. TDH may return ERROR_NOT_FOUND for classic kernel schemas on the analysis host. Xperf can still decode those MOF events from the same ETL. This is an offline adapter only: it never qualifies a native Wii update on its own. """ import argparse from collections import Counter import csv import json from pathlib import Path import re import statistics def parse_pid(label): match = re.search(r"\(\s*(\d+)\s*\)$", label.strip()) return int(match.group(1)) if match else None def convert(rows, native, pid): native_markers = {e["data"]["Name"]: e["qpc"] for e in native if e["native"] and e["pid"] == pid and e["event"] == "Marker"} native_payload_qpc = { (e["event"], int(e["data"]["QPC"])): e["qpc"] for e in native if e["native"] and e["pid"] == pid and e["event"] in ("Marker", "Snapshot", "Frame") and "QPC" in e["data"] } frequencies = {e["data"]["QPCFrequency"] for e in native if e["native"] and e["pid"] == pid and "QPCFrequency" in e["data"]} if len(frequencies) != 1: raise ValueError("Missing or inconsistent native QPC frequency") frequency = frequencies.pop() native_tids = {e["tid"] for e in native if e["native"] and e["pid"] == pid} if not native_tids: raise ValueError("No native threads for requested PID") # Xperf timestamps are trace-relative microseconds. Match the QPC payload # of periodic native events to their TDH event-header QPC. A real Dolphin # run need not emit two named markers, but normally emits many snapshots. anchors = [] raw_samples = [] raw_switches = [] invalid_events = 0 invalid_target_events = 0 for row in rows: if len(row) < 2: continue kind = row[0].strip() if kind.startswith("InvalidEvent/"): invalid_events += 1 if len(row) > 2 and parse_pid(row[2]) == pid: invalid_target_events += 1 if not row[1].strip().isdigit(): # xperf's per-event header line continue if kind in ("Dolphin.Performance/Marker/", "Dolphin.Performance/Snapshot/", "Dolphin.Performance/Frame/") and len(row) >= 12 and parse_pid(row[2]) == pid: event = kind.split("/")[1] payload_qpc = row[10].strip() if payload_qpc.isdigit() and (event, int(payload_qpc)) in native_payload_qpc: anchors.append((int(row[1]), native_payload_qpc[(event, int(payload_qpc))], f"{event}:{payload_qpc}")) elif event == "Marker" and len(row) >= 13: # Older synthetic captures also support named-marker matching. name = row[-1].strip() if name in native_markers: anchors.append((int(row[1]), native_markers[name], name)) elif kind == "SampledProfile" and len(row) >= 9 and parse_pid(row[2]) == pid: tid = int(row[3]) # Xperf does not CSV-quote C++ template commas in symbol names. # Count/type are the final two columns; rejoin the leaf symbol. # All threads of this PID matter: GPU/network threads may be hot # without ever emitting a native Dolphin.Performance event. raw_samples.append((int(row[1]), tid, int(row[4], 16), int(row[5]), int(row[-2]), ",".join(row[7:-2]).strip())) elif kind == "CSwitch" and len(row) >= 13: old_tid, new_tid = int(row[9]), int(row[3]) if (old_tid in native_tids or new_tid in native_tids or parse_pid(row[2]) == pid or parse_pid(row[8]) == pid): raw_switches.append((int(row[1]), old_tid, new_tid, row[12].strip())) if len(anchors) < 2: raise ValueError("Need at least two matching native/xperf QPC anchors") scale = frequency / 1_000_000 offset = statistics.median(qpc - us * scale for us, qpc, _ in anchors) max_anchor_error = max(abs(round(us * scale + offset) - qpc) for us, qpc, _ in anchors) if max_anchor_error > frequency / 100_000: # More than 10 microseconds. raise ValueError(f"Native/xperf clocks do not agree: {max_anchor_error} QPC ticks") def qpc(us): return round(us * scale + offset) system = [dict(native=False, event="SampledProfile", qpc=qpc(us), pid=pid, tid=tid, data=dict(ThreadId=tid, InstructionPointer=ip, Processor=processor, Count=count, Symbol=symbol)) for us, tid, ip, processor, count, symbol in raw_samples] system.extend(dict(native=False, event="CSwitch", qpc=qpc(us), pid=0, tid=0, data=dict(OldThreadId=old_tid, NewThreadId=new_tid, OldThreadState=5 if state == "Waiting" else -1, OldState=state)) for us, old_tid, new_tid, state in raw_switches) samples_by_processor = Counter() for _, _, _, processor, count, _ in raw_samples: samples_by_processor[processor] += count summary = dict(schema=1, pid=pid, qpc_frequency=frequency, anchors=len(anchors), max_anchor_error_qpc=max_anchor_error, cpu_sample_records=len(raw_samples), cpu_sample_count=sum(count for _, _, _, _, count, _ in raw_samples), cpu_sample_threads=len({tid for _, tid, _, _, _, _ in raw_samples}), cpu_samples_by_processor=dict(sorted(samples_by_processor.items())), context_switches=len(raw_switches), xperf_invalid_events=invalid_events, xperf_invalid_target_events=invalid_target_events, note="Classic kernel events decoded by xperf; invalid events and missing symbols remain separate quality gates.") return sorted(native + system, key=lambda e: e["qpc"]), summary if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("xperf_csv", type=Path) parser.add_argument("native_jsonl", type=Path) parser.add_argument("--pid", required=True, type=int) parser.add_argument("--output", required=True, type=Path) parser.add_argument("--summary", required=True, type=Path) args = parser.parse_args() with args.native_jsonl.open(encoding="utf-8") as source: native_events = [json.loads(line) for line in source] with args.xperf_csv.open(encoding="utf-8", errors="replace", newline="") as source: events, result = convert(csv.reader(source, skipinitialspace=True), native_events, args.pid) with args.output.open("x", encoding="utf-8") as target: for event in events: target.write(json.dumps(event, separators=(",", ":")) + "\n") with args.summary.open("x", encoding="utf-8") as target: json.dump(result, target, indent=2) print(json.dumps(result, indent=2))