From 1d706dd169ea4e5085a712c051a976bda5c93864 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 18:44:57 +0300 Subject: [PATCH] fix(perception): correlate GC with active stages --- .../run_m48s_reference_graph_shadow_worker.py | 75 +++++++++++++------ tests/test_m48s_reference_graph_timing.py | 11 ++- 2 files changed, 63 insertions(+), 23 deletions(-) diff --git a/experiments/perception/run_m48s_reference_graph_shadow_worker.py b/experiments/perception/run_m48s_reference_graph_shadow_worker.py index 8987c59..602ec75 100644 --- a/experiments/perception/run_m48s_reference_graph_shadow_worker.py +++ b/experiments/perception/run_m48s_reference_graph_shadow_worker.py @@ -112,8 +112,17 @@ class GcPauseTelemetry: def __init__(self) -> None: self.events: list[dict[str, object]] = [] - self._starts: dict[int, tuple[int, int, str | None, int | None]] = {} - self._stage = threading.local() + self._starts: dict[ + int, + tuple[ + int, + int, + str | None, + int | None, + tuple[tuple[str, int, str], ...], + ], + ] = {} + self._active_stages: dict[int, tuple[str, int, str]] = {} def __enter__(self) -> GcPauseTelemetry: gc.callbacks.append(self._observe) @@ -123,21 +132,27 @@ class GcPauseTelemetry: gc.callbacks.remove(self._observe) def enter_stage(self, stage_id: str, sequence: int) -> None: - self._stage.value = (stage_id, sequence) + thread_id = threading.get_ident() + self._active_stages[thread_id] = ( + stage_id, + sequence, + threading.current_thread().name, + ) def exit_stage(self) -> None: - self._stage.value = None + self._active_stages.pop(threading.get_ident(), None) def _observe(self, phase: str, info: dict[str, int]) -> None: thread_id = threading.get_ident() if phase == "start": - stage = getattr(self._stage, "value", None) - stage_id, sequence = stage if stage is not None else (None, None) + stage = self._active_stages.get(thread_id) + stage_id, sequence, _thread_name = stage if stage is not None else (None, None, None) self._starts[thread_id] = ( time.perf_counter_ns(), info["generation"], stage_id, sequence, + tuple(sorted(self._active_stages.values())), ) return if phase != "stop": @@ -145,7 +160,7 @@ class GcPauseTelemetry: started = self._starts.pop(thread_id, None) if started is None: return - started_ns, generation, stage_id, sequence = started + started_ns, generation, stage_id, sequence, active_stages = started self.events.append( { "duration_ns": max(0, time.perf_counter_ns() - started_ns), @@ -155,6 +170,7 @@ class GcPauseTelemetry: "thread_name": threading.current_thread().name, "stage_id": stage_id, "sequence": sequence, + "active_stages": active_stages, } ) @@ -805,8 +821,36 @@ def _telemetry_summary(samples: list[dict[str, float]]) -> dict[str, Any]: def _gc_telemetry_summary(events: list[dict[str, object]]) -> dict[str, object]: durations_ms = [cast(int, event["duration_ns"]) / 1_000_000.0 for event in events] generations = Counter(cast(int, event["generation"]) for event in events) - attributed = [event for event in events if event["stage_id"] is not None] + attributed = [event for event in events if event["active_stages"]] maximum = max(events, key=lambda event: cast(int, event["duration_ns"]), default=None) + significant = sorted( + events, + key=lambda event: cast(int, event["duration_ns"]), + reverse=True, + )[:20] + + def summarize(event: dict[str, object]) -> dict[str, object]: + return { + "duration_ms": round(cast(int, event["duration_ns"]) / 1_000_000.0, 6), + "generation": event["generation"], + "collected": event["collected"], + "uncollectable": event["uncollectable"], + "thread_name": event["thread_name"], + "trigger_stage_id": event["stage_id"], + "trigger_sequence": event["sequence"], + "active_stages": [ + { + "stage_id": stage_id, + "sequence": sequence, + "thread_name": thread_name, + } + for stage_id, sequence, thread_name in cast( + tuple[tuple[str, int, str], ...], + event["active_stages"], + ) + ], + } + return { "event_count": len(events), "generation_counts": { @@ -814,19 +858,8 @@ def _gc_telemetry_summary(events: list[dict[str, object]]) -> dict[str, object]: }, "duration_ms": _distribution(durations_ms), "pipeline_attributed_event_count": len(attributed), - "maximum_event": ( - { - "duration_ms": round(cast(int, maximum["duration_ns"]) / 1_000_000.0, 6), - "generation": maximum["generation"], - "collected": maximum["collected"], - "uncollectable": maximum["uncollectable"], - "thread_name": maximum["thread_name"], - "stage_id": maximum["stage_id"], - "sequence": maximum["sequence"], - } - if maximum is not None - else None - ), + "maximum_event": summarize(maximum) if maximum is not None else None, + "significant_events": [summarize(event) for event in significant], } diff --git a/tests/test_m48s_reference_graph_timing.py b/tests/test_m48s_reference_graph_timing.py index ead3ba7..812335c 100644 --- a/tests/test_m48s_reference_graph_timing.py +++ b/tests/test_m48s_reference_graph_timing.py @@ -98,6 +98,13 @@ def test_gc_pause_telemetry_attributes_collection_to_active_stage() -> None: assert summary["event_count"] == 1 assert summary["pipeline_attributed_event_count"] == 1 assert summary["maximum_event"]["generation"] == 2 - assert summary["maximum_event"]["stage_id"] == "rolling" - assert summary["maximum_event"]["sequence"] == 3928 + assert summary["maximum_event"]["trigger_stage_id"] == "rolling" + assert summary["maximum_event"]["trigger_sequence"] == 3928 + assert summary["maximum_event"]["active_stages"] == [ + { + "stage_id": "rolling", + "sequence": 3928, + "thread_name": "MainThread", + } + ] assert summary["maximum_event"]["duration_ms"] >= 1.0