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.
196 lines
11 KiB
Python
196 lines
11 KiB
Python
import unittest
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
from analyze_p1 import CodeMap, SampledCodeMap, analyze, percentile, read_relevant_events
|
|
from xperf_to_p1 import convert, parse_pid
|
|
|
|
|
|
class AnalysisTests(unittest.TestCase):
|
|
def test_temporal_address_reuse(self):
|
|
m = CodeMap()
|
|
def event(begin, end, kind, pc=0):
|
|
m.event(dict(Begin=begin, End=end, Kind=kind, GuestPC=pc, Domain="ARM"))
|
|
event(100, 200, "thumb", 4)
|
|
self.assertEqual(m.lookup(125), ("ARM", 4))
|
|
event(120, 180, "invalidate")
|
|
self.assertEqual(m.lookup(125)[0], "UnresolvedNative")
|
|
self.assertEqual(m.lookup(190), ("ARM", 4))
|
|
event(120, 180, "thumb", 8)
|
|
self.assertEqual(m.lookup(125), ("ARM", 8))
|
|
m.event(dict(Begin=100, End=200, Kind="block-near", GuestPC=0x8000, Domain="PPC"))
|
|
self.assertEqual(m.lookup(125), ("PPC", 0x8000))
|
|
self.assertEqual(m.lookup(190), ("PPC", 0x8000))
|
|
|
|
def test_invalid_range(self):
|
|
with self.assertRaises(ValueError):
|
|
CodeMap().event(dict(Begin=10, End=9))
|
|
|
|
def test_sampled_map_preserves_temporal_overlap_semantics(self):
|
|
sampled = SampledCodeMap([125, 190, 250])
|
|
sampled.event(dict(Begin=100, End=200, Kind="thumb", Domain="ARM", GuestPC=4))
|
|
self.assertEqual(sampled.lookup(125), ("ARM", 4))
|
|
sampled.event(dict(Begin=120, End=180, Kind="invalidate", Domain="ARM", GuestPC=0))
|
|
self.assertEqual(sampled.lookup(125), ("UnresolvedNative", None))
|
|
self.assertEqual(sampled.lookup(190), ("ARM", 4))
|
|
sampled.event(dict(Begin=120, End=180, Kind="thumb", Domain="ARM", GuestPC=8))
|
|
self.assertEqual(sampled.lookup(125), ("ARM", 8))
|
|
self.assertEqual(sampled.lookup(250), ("UnresolvedNative", None))
|
|
|
|
def test_percentiles_are_frame_intervals(self):
|
|
self.assertEqual(percentile([10, 20, 30], .5), 20)
|
|
self.assertIsNone(percentile([], .5))
|
|
|
|
def test_missing_telemetry_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
analyze([], 12)
|
|
|
|
def test_unknown_schema_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
analyze([dict(native=True, pid=12, tid=1, qpc=0, data=dict(Schema=99))], 12)
|
|
|
|
@staticmethod
|
|
def native(event, qpc, **data):
|
|
return dict(native=True, pid=12, tid=1, qpc=qpc, event=event, data=dict(Schema=1, **data))
|
|
|
|
def test_exclusive_accounting_does_not_sum_inclusive_regions(self):
|
|
rows = [self.native("Marker", 0, QPCFrequency=1000, Name="ProbeBegin")]
|
|
for region, inclusive, exclusive in ((0, 100, 20), (1, 80, 50), (6, 30, 30)):
|
|
rows.append(self.native("ScopeSample", 100, Region=region, Sample=1,
|
|
SamplingDenominator=1, Calls=1, AccountingErrors=0,
|
|
InclusiveQPC=inclusive, ExclusiveQPC=exclusive))
|
|
result = analyze(rows, 12)
|
|
self.assertEqual(result["accounting_errors"], 0)
|
|
rows[-1]["data"]["ExclusiveQPC"] = 31
|
|
self.assertEqual(analyze(rows, 12)["accounting_errors"], 1)
|
|
|
|
def test_frames_do_not_bridge_reset(self):
|
|
rows = [self.native("Marker", 0, QPCFrequency=1000, Name="ProbeBegin")]
|
|
for t in (10, 20):
|
|
rows.append(self.native("Frame", t, QPC=t, Kind="produced"))
|
|
rows.append(self.native("Marker", 100, Name="PerformanceMetricsReset"))
|
|
for t in (1000, 1010):
|
|
rows.append(self.native("Frame", t, QPC=t, Kind="produced"))
|
|
self.assertEqual(analyze(rows, 12)["frame_intervals"]["produced"],
|
|
dict(intervals=2, p50_ms=10, p95_ms=10, p99_ms=10))
|
|
|
|
def test_counter_regression_fails_even_with_good_scopes(self):
|
|
rows = [self.native("Marker", 0, QPCFrequency=1000, Name="ProbeBegin"),
|
|
self.native("Counter", 1, Counter=0, Value=5),
|
|
self.native("Counter", 2, Counter=0, Value=4)]
|
|
self.assertEqual(len(analyze(rows, 12)["counter_regressions"]), 1)
|
|
|
|
def test_kernel_sample_and_known_sleep_are_separate_evidence(self):
|
|
rows = [self.native("Marker", 0, QPCFrequency=1000, Name="ProbeBegin"),
|
|
self.native("CodeMap", 1, Begin=100, End=200, Domain="ARM",
|
|
GuestPC=4, Kind="thumb"),
|
|
dict(native=False, pid=0, tid=0, qpc=10, event="SampledProfile",
|
|
data=dict(ThreadId=1, InstructionPointer=125)),
|
|
self.native("Marker", 20, Name="KnownSleepBegin"),
|
|
dict(native=False, pid=0, tid=0, qpc=30, event="CSwitch",
|
|
data=dict(OldThreadId=1, NewThreadId=2, OldThreadState=5)),
|
|
self.native("Marker", 800, Name="KnownSleepEnd")]
|
|
result = analyze(rows, 12)
|
|
self.assertEqual(result["cpu_leaf_samples"], {"ARM": 1})
|
|
self.assertEqual(result["hot_generated_pcs"][0]["pc"], "0x4")
|
|
self.assertTrue(result["known_sleep"]["wait_evidence_present"])
|
|
self.assertEqual(result["known_sleep"]["wall_ms"], 780)
|
|
|
|
def test_jit_sample_after_invalidation_is_not_attributed_to_old_block(self):
|
|
rows = [self.native("Marker", 0, QPCFrequency=1000, Name="ProbeBegin"),
|
|
self.native("CodeMap", 1, Begin=100, End=200, Domain="ARM",
|
|
GuestPC=4, Kind="thumb"),
|
|
dict(native=False, pid=0, tid=1, qpc=2, event="SampledProfile",
|
|
data=dict(InstructionPointer=125)),
|
|
self.native("CodeMap", 3, Begin=100, End=200, Domain="ARM",
|
|
GuestPC=0, Kind="invalidate"),
|
|
dict(native=False, pid=0, tid=1, qpc=4, event="SampledProfile",
|
|
data=dict(InstructionPointer=125))]
|
|
self.assertEqual(analyze(rows, 12)["cpu_leaf_samples"],
|
|
{"ARM": 1, "UnresolvedNative": 1})
|
|
|
|
def test_static_module_is_not_mistaken_for_unmapped_jit(self):
|
|
rows = [self.native("Marker", 0, QPCFrequency=1000, Name="ProbeBegin"),
|
|
dict(native=False, pid=12, tid=1, qpc=1, event="SampledProfile",
|
|
data=dict(ThreadId=1, InstructionPointer=999, Count=3,
|
|
Symbol="TraceProbeP1.exe!IOS::LLE::ARMCore::Read32"))]
|
|
report = analyze(rows, 12)
|
|
self.assertEqual(report["cpu_leaf_samples"], {"HostC++": 3})
|
|
self.assertEqual(report["hot_static_symbols"][0]["samples"], 3)
|
|
|
|
def test_xperf_classic_events_align_to_native_qpc(self):
|
|
native = [self.native("Marker", 1001000, QPCFrequency=10000000, Name="ProbeBegin"),
|
|
self.native("CodeMap", 1001500, Begin=100, End=200, Domain="ARM",
|
|
GuestPC=4, Kind="thumb"),
|
|
self.native("Marker", 1011000, Name="KnownSleepBegin"),
|
|
self.native("Marker", 1111000, Name="KnownSleepEnd")]
|
|
rows = [["Dolphin.Performance/Marker/", "100000", "TraceProbeP1.exe (12)",
|
|
"1"] + [""] * 8 + ["ProbeBegin"],
|
|
["Dolphin.Performance/Marker/", "101000", "TraceProbeP1.exe (12)",
|
|
"1"] + [""] * 8 + ["KnownSleepBegin"],
|
|
["Dolphin.Performance/Marker/", "111000", "TraceProbeP1.exe (12)",
|
|
"1"] + [""] * 8 + ["KnownSleepEnd"],
|
|
["SampledProfile", "100100", "TraceProbeP1.exe (12)", "1",
|
|
"0x7d", "0", "", "", "2", "Unbatched"],
|
|
["CSwitch", "101100", "Idle (0)", "2", "", "", "", "",
|
|
"TraceProbeP1.exe (12)", "1", "", "", "Waiting"]]
|
|
merged, summary = convert(rows, native, 12)
|
|
report = analyze(merged, 12)
|
|
self.assertEqual(summary["max_anchor_error_qpc"], 0)
|
|
self.assertEqual(summary["cpu_sample_count"], 2)
|
|
self.assertEqual(summary["cpu_samples_by_processor"], {0: 2})
|
|
self.assertEqual(report["cpu_leaf_samples"], {"ARM": 2})
|
|
self.assertTrue(report["known_sleep"]["wait_evidence_present"])
|
|
self.assertEqual(parse_pid("Idle ( 0)"), 0)
|
|
|
|
def test_xperf_snapshots_align_without_named_markers(self):
|
|
native = [self.native("Snapshot", 1001000, QPC=1000999,
|
|
QPCFrequency=10000000, CoreId=1),
|
|
self.native("Snapshot", 2001000, QPC=2000999,
|
|
QPCFrequency=10000000, CoreId=1)]
|
|
def snapshot(us, payload_qpc):
|
|
row = ["Dolphin.Performance/Snapshot/", str(us), "DolphinNoGUI.exe (12)",
|
|
"1"] + [""] * 8
|
|
row[10] = str(payload_qpc)
|
|
return row
|
|
rows = [snapshot(100000, 1000999), snapshot(200000, 2000999),
|
|
["SampledProfile", "150000", "DolphinNoGUI.exe (12)", "1",
|
|
"0x7d", "0", "", "", "2", "DolphinNoGUI.exe!ARMCore::Run"]]
|
|
merged, summary = convert(rows, native, 12)
|
|
self.assertEqual(summary["anchors"], 2)
|
|
self.assertEqual(summary["max_anchor_error_qpc"], 0)
|
|
self.assertEqual(summary["cpu_sample_count"], 2)
|
|
self.assertEqual(merged[1]["qpc"], 1501000)
|
|
|
|
def test_uninstrumented_process_thread_is_counted(self):
|
|
native = [self.native("Snapshot", 1001000, QPC=1000999,
|
|
QPCFrequency=10000000, CoreId=1, Instructions=1,
|
|
NativeInstructions=1, SlowReads=0, SlowWrites=0,
|
|
Compiles=0, Fallbacks=0),
|
|
self.native("Snapshot", 2001000, QPC=2000999,
|
|
QPCFrequency=10000000, CoreId=1, Instructions=2,
|
|
NativeInstructions=2, SlowReads=0, SlowWrites=0,
|
|
Compiles=0, Fallbacks=0)]
|
|
rows = []
|
|
for us, payload in ((100000, 1000999), (200000, 2000999)):
|
|
row = ["Dolphin.Performance/Snapshot/", str(us), "DolphinNoGUI.exe (12)",
|
|
"1"] + [""] * 8
|
|
row[10] = str(payload)
|
|
rows.append(row)
|
|
rows.append(["SampledProfile", "150000", "DolphinNoGUI.exe (12)", "9",
|
|
"0x7d", "0", "", "DolphinNoGUI.exe!Render", "3", "Unbatched"])
|
|
merged, summary = convert(rows, native, 12)
|
|
self.assertEqual(summary["cpu_sample_threads"], 1)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "events.jsonl"
|
|
path.write_text("".join(json.dumps(event) + "\n" for event in merged),
|
|
encoding="utf-8")
|
|
report = analyze(read_relevant_events(path, 12), 12)
|
|
self.assertEqual(report["native_threads"], [1])
|
|
self.assertEqual(report["sampled_process_threads"], [9])
|
|
self.assertEqual(report["hot_static_symbols"][0]["samples"], 3)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|