"""Offline P1 event analysis. Unknowns fail qualification rather than becoming zero cost.""" import argparse from bisect import bisect_left from collections import Counter, defaultdict import json from pathlib import Path import statistics REGIONS = ("Slice", "ARM", "Jit", "Compile", "Devices", "NAND", "Journal", "JournalBuild", "JournalWrite", "JournalFlush", "NAT", "DMARead", "DMAWrite", "AES", "SHA") class CodeMap: def __init__(self): self.ranges = [] def event(self, data): begin, end = int(data["Begin"]), int(data["End"]) if not begin or end <= begin: raise ValueError("Invalid generated-code interval") # Reuse replaces overlapping mappings at this timestamp. Never use the final # mapping to interpret samples collected before an invalidation/recompilation. kept = [] for lo, hi, domain, pc in self.ranges: if hi <= begin or lo >= end: kept.append((lo, hi, domain, pc)) else: if lo < begin: kept.append((lo, begin, domain, pc)) if hi > end: kept.append((end, hi, domain, pc)) if data["Kind"] != "invalidate": kept.append((begin, end, data["Domain"], data["GuestPC"])) self.ranges = kept def lookup(self, ip): for lo, hi, domain, pc in reversed(self.ranges): if lo <= ip < hi: return domain, pc return "UnresolvedNative", None class SampledCodeMap: """Time-ordered mappings only for host PCs actually sampled by ETW. Native updates can emit hundreds of thousands of JIT map events. Keeping split intervals and scanning them for every sample makes offline analysis quadratic. The report only needs the finite set of sampled host addresses. """ def __init__(self, sampled_addresses): self.addresses = sorted(set(sampled_addresses)) self.mappings = {} def event(self, data): begin, end = int(data["Begin"]), int(data["End"]) if not begin or end <= begin: raise ValueError("Invalid generated-code interval") lo = bisect_left(self.addresses, begin) hi = bisect_left(self.addresses, end) if data["Kind"] == "invalidate": for address in self.addresses[lo:hi]: self.mappings.pop(address, None) else: value = (data["Domain"], data["GuestPC"]) for address in self.addresses[lo:hi]: self.mappings[address] = value def lookup(self, ip): return self.mappings.get(ip, ("UnresolvedNative", None)) def percentile(values, fraction): if not values: return None values = sorted(values) position = (len(values) - 1) * fraction low = int(position) high = min(low + 1, len(values) - 1) return values[low] + (values[high] - values[low]) * (position - low) def read_relevant_events(path, pid): # ETW is host-wide. Keep this process's telemetry and all its CPU samples; # rendering/network threads can be hot without emitting native telemetry. native = [] with path.open(encoding="utf-8") as source: for line in source: e = json.loads(line) if e["native"] and e["pid"] == pid: native.append(e) if not native: raise ValueError("No native events for requested process") native_tids = {e["tid"] for e in native} begin, end = min(e["qpc"] for e in native), max(e["qpc"] for e in native) rows = native with path.open(encoding="utf-8") as source: for line in source: e = json.loads(line) if e["native"] or not begin <= e["qpc"] <= end: continue if e["event"] == "SampledProfile" and e["pid"] == pid: rows.append(e) sampled_tids = {e["data"].get("ThreadId", e["tid"]) for e in rows if not e["native"] and e["event"] == "SampledProfile"} process_tids = native_tids | sampled_tids with path.open(encoding="utf-8") as source: for line in source: e = json.loads(line) if e["native"] or e["event"] != "CSwitch" or not begin <= e["qpc"] <= end: continue d = e["data"] if d.get("OldThreadId") in process_tids or d.get("NewThreadId") in process_tids: rows.append(e) return rows def analyze(rows, pid): rows = sorted(rows, key=lambda event: event["qpc"]) native = [e for e in rows if e["native"] and e["pid"] == pid] if not native: raise ValueError("No native events for requested process") if any(e["data"].get("Schema") != 1 for e in native): raise ValueError("Unknown or missing native schema") native_tids = {e["tid"] for e in native} sampled_tids = {e["data"].get("ThreadId", e["tid"]) for e in rows if not e["native"] and e["event"] == "SampledProfile" and e["pid"] == pid} tids = native_tids | sampled_tids frequencies = {e["data"]["QPCFrequency"] for e in native if "QPCFrequency" in e["data"]} if len(frequencies) != 1 or not next(iter(frequencies)): raise ValueError("Missing or inconsistent QPC frequency") frequency = next(iter(frequencies)) mappings = SampledCodeMap(e["data"]["InstructionPointer"] for e in rows if not e["native"] and "InstructionPointer" in e["data"]) samples = Counter() hot_pc = Counter() hot_static = Counter() phases = {} scope_samples = defaultdict(list) frames = defaultdict(list) counter_last = {} counter_errors = [] snapshots = [] cswitches = [] queues = [] accounting_errors = 0 for e in rows: data = e["data"] if e["native"] and e["pid"] == pid: if e["event"] == "CodeMap": mappings.event(data) elif e["event"] == "Marker": phases[data["Name"]] = e["qpc"] if data["Name"] == "PerformanceMetricsReset": # Keep intervals per generation, never bridge a reset/reboot. for events in frames.values(): events.append(None) elif e["event"] == "ScopeSample": accounting_errors += data["AccountingErrors"] scope_samples[(e["tid"], data["Sample"])].append(data) elif e["event"] == "Frame": frames[data["Kind"]].append((data["QPC"], data)) elif e["event"] == "Counter": key = (e["tid"], data["Counter"]) if data["Value"] < counter_last.get(key, 0): counter_errors.append({"tid": e["tid"], "counter": data["Counter"], "qpc": e["qpc"]}) counter_last[key] = data["Value"] elif e["event"] == "Snapshot": snapshots.append(data) elif e["event"] == "USBQueue": queues.append(data) elif not e["native"]: tid = data.get("ThreadId", data.get("ThreadID", e["tid"])) if "InstructionPointer" in data and tid in tids: domain, pc = mappings.lookup(data["InstructionPointer"]) weight = data.get("Count", 1) if domain == "UnresolvedNative" and "!" in data.get("Symbol", ""): symbol = data["Symbol"] module = symbol.split("!", 1)[0].strip('"').lower() if module in ("traceprobep1.exe", "dolphinnogui.exe", "dolphin.exe"): domain = "HostC++" elif module and module != "unknown": domain = "SystemOrLibrary" if domain != "UnresolvedNative": hot_static[symbol] += weight samples[domain] += weight if pc is not None: hot_pc[(domain, pc)] += weight if "OldThreadId" in data and (data["OldThreadId"] in tids or data.get("NewThreadId") in tids): cswitches.append(e) timing = defaultdict(lambda: {"calls": 0, "inclusive_ms": 0., "exclusive_ms": 0.}) for group in scope_samples.values(): roots = [d for d in group if d["Region"] == 0] if len(roots) != 1 or sum(d["ExclusiveQPC"] for d in group) != roots[0]["InclusiveQPC"]: accounting_errors += 1 for data in group: region = REGIONS[data["Region"]] + "/sample-1-in-" + str(data["SamplingDenominator"]) # Actual sampled time only. Do not present extrapolated duration as exact CPU. stat = timing[region] stat["calls"] += data["Calls"] stat["inclusive_ms"] += 1000 * data["InclusiveQPC"] / frequency stat["exclusive_ms"] += 1000 * data["ExclusiveQPC"] / frequency snapshot_errors = [] previous = {} for data in snapshots: key = data["CoreId"] if key in previous: for field in ("Instructions", "NativeInstructions", "SlowReads", "SlowWrites", "Compiles", "Fallbacks"): if data[field] < previous[key][field]: snapshot_errors.append({"generation": key, "field": field}) previous[key] = data durations = {} for kind, events in frames.items(): deltas = [(b[0] - a[0]) * 1000 / frequency for a, b in zip(events, events[1:]) if a and b] durations[kind] = {"intervals": len(deltas), "p50_ms": percentile(deltas, .5), "p95_ms": percentile(deltas, .95), "p99_ms": percentile(deltas, .99)} sleep = None if "KnownSleepBegin" in phases and "KnownSleepEnd" in phases: begin, end = phases["KnownSleepBegin"], phases["KnownSleepEnd"] changes = [e for e in cswitches if begin <= e["qpc"] <= end] waits = [e for e in changes if e["data"]["OldThreadId"] in tids and e["data"].get("OldThreadState") == 5] sleep = {"wall_ms": (end - begin) * 1000 / frequency, "cswitch_events": len(changes), "waiting_switches": len(waits), "wait_evidence_present": bool(waits)} return {"schema": 1, "pid": pid, "qpc_frequency": frequency, "native_events": len(native), "native_threads": sorted(native_tids), "sampled_process_threads": sorted(sampled_tids), "cpu_leaf_samples": dict(samples), "hot_generated_pcs": [{"domain": domain, "pc": hex(pc), "samples": count} for (domain, pc), count in hot_pc.most_common(20)], "hot_static_symbols": [{"symbol": symbol, "samples": count} for symbol, count in hot_static.most_common(20)], "sampled_scope_wall_times_not_cpu": dict(timing), "scope_samples": len(scope_samples), "accounting_errors": accounting_errors, "counter_regressions": counter_errors, "snapshot_regressions": snapshot_errors, "core_generations": sorted(previous), "last_counters_by_thread": [{"tid": tid, "counter": counter, "value": value} for (tid, counter), value in sorted(counter_last.items())], "usb_queue_events": len(queues), "usb_queue_peak": max((q["PeakFrames"] for q in queues), default=None), "frame_intervals": durations, "known_sleep": sleep, "markers": phases, "native_update_qualified": False, "remaining_gates": ["Resolve static C++/system stacks and critical-path waits in ETL", "Qualify PPC and check local NAT counters", "Check ETW losses and full collector overhead"], "note": "Partial offline qualification, not a native update bottleneck report. CPU leaf samples include all sampled threads of the target process within native telemetry bounds. Leaf sample counts are not exact CPU durations. Scopes are sampled wall times, not CPU. Never compare differently sampled totals."} if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("events", type=Path) parser.add_argument("--pid", required=True, type=int) parser.add_argument("--output", required=True, type=Path) args = parser.parse_args() result = analyze(read_relevant_events(args.events, args.pid), args.pid) with args.output.open("x", encoding="utf-8") as output: json.dump(result, output, indent=2) print(json.dumps(result, indent=2))