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()