Yaya48
c230971f07
Preserve formatted NAND boot and Wii Remote pairing, add DMA/SRAM conformance tests and a Wii comparison probe. Keep P1 profiling tools and native update A/B evidence; reject the unhelpful targeted SRAM shortcut.
257 lines
12 KiB
Python
257 lines
12 KiB
Python
"""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))
|