From 477886d10094184002cf49c8cae951f719024b94 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 18:50:38 +0300 Subject: [PATCH] fix(perception): keep cyclic GC outside realtime loop --- .../run_m48s_reference_graph_shadow_worker.py | 65 +++++++++++++++++-- tests/test_m48s_reference_graph_timing.py | 32 +++++++++ 2 files changed, 93 insertions(+), 4 deletions(-) diff --git a/experiments/perception/run_m48s_reference_graph_shadow_worker.py b/experiments/perception/run_m48s_reference_graph_shadow_worker.py index 602ec75..e3982bf 100644 --- a/experiments/perception/run_m48s_reference_graph_shadow_worker.py +++ b/experiments/perception/run_m48s_reference_graph_shadow_worker.py @@ -51,9 +51,10 @@ from k1link.perception.reference_graph_runtime import ReferenceGraphRuntimePaths from k1link.perception.rolling_map import RollingLocalObstacleMapProvider from k1link.perception.temporal import BoundedSpatialTemporalProvider -SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v1" +SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v2" FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1" PIPELINE_TIMING_SCHEMA: Final = "missioncore.m48s-frame-pipeline-timing/v0" +GC_POLICY_SCHEMA: Final = "missioncore.cyclic-gc-hot-loop-policy/v0" AUTHORITY: Final = { "ground_truth": False, "candidate_accepted": False, @@ -175,6 +176,51 @@ class GcPauseTelemetry: ) +class CyclicGcHotLoopPolicy: + """Keep cyclic GC outside the source-paced realtime envelope.""" + + def __init__(self) -> None: + self.enabled_before: bool | None = None + self.disabled_during_hot_loop = False + self.enabled_after: bool | None = None + self.pre_collected: int | None = None + self.post_collected: int | None = None + + def __enter__(self) -> CyclicGcHotLoopPolicy: + self.enabled_before = gc.isenabled() + self.pre_collected = gc.collect() + gc.disable() + self.disabled_during_hot_loop = not gc.isenabled() + if not self.disabled_during_hot_loop: + raise RuntimeError("cyclic GC remained enabled inside the hot loop") + return self + + def __exit__(self, *_args: object) -> None: + if self.enabled_before: + gc.enable() + self.post_collected = gc.collect() + self.enabled_after = gc.isenabled() + if self.enabled_after is not self.enabled_before: + raise RuntimeError("cyclic GC state was not restored after the hot loop") + + def to_dict(self) -> dict[str, object]: + if ( + self.enabled_before is None + or self.enabled_after is None + or self.pre_collected is None + or self.post_collected is None + ): + raise RuntimeError("cyclic GC hot-loop policy did not close") + return { + "schema_version": GC_POLICY_SCHEMA, + "enabled_before": self.enabled_before, + "disabled_during_hot_loop": self.disabled_during_hot_loop, + "enabled_after": self.enabled_after, + "pre_collected": self.pre_collected, + "post_collected": self.post_collected, + } + + class FrameTimingStore: """Join bounded decode, detector and provider timings by source sequence.""" @@ -397,9 +443,11 @@ def main() -> int: gc_telemetry, ), ) - loop_started_ns = time.monotonic_ns() - result = runtime.graph.run() - loop_completed_ns = time.monotonic_ns() + gc_policy = CyclicGcHotLoopPolicy() + with gc_policy: + loop_started_ns = time.monotonic_ns() + result = runtime.graph.run() + loop_completed_ns = time.monotonic_ns() detector_snapshot = cast( RfDetrShadowDetectorProvider, runtime.graph.detector, @@ -447,6 +495,7 @@ def main() -> int: completion_ages_ns=loop_completion_ages_ns, wall_seconds=loop_wall_seconds, setup_seconds=setup_seconds, + gc_policy=gc_policy.to_dict(), ) loop_documents.append(loop_document) completion_ages_ms.extend(value / 1_000_000.0 for value in loop_completion_ages_ns) @@ -523,6 +572,12 @@ def main() -> int: "distinct_class_family_policy": len(set(advisory_policy_matrix().values())) == len(AdvisoryFamily), "complete_pipeline_timing": len(all_pipeline_timings) == delivered, + "cyclic_gc_disabled_during_hot_loop": all( + cast(dict[str, object], loop["cyclic_gc_hot_loop"])["disabled_during_hot_loop"] is True + and cast(dict[str, object], loop["cyclic_gc_hot_loop"])["enabled_before"] + is cast(dict[str, object], loop["cyclic_gc_hot_loop"])["enabled_after"] + for loop in loop_documents + ), "authority_remains_false": all(value is False for value in AUTHORITY.values()), } integrated_runtime_gate_passed = all(checks.values()) @@ -634,6 +689,7 @@ def _loop_document( completion_ages_ns: list[int], wall_seconds: float, setup_seconds: float, + gc_policy: dict[str, object], ) -> dict[str, object]: outcomes = Counter(item.outcome.value for item in result.terminal_outcomes) outcome_stages = Counter( @@ -644,6 +700,7 @@ def _loop_document( "state": result.state.value, "wall_seconds": round(wall_seconds, 6), "setup_seconds": round(setup_seconds, 6), + "cyclic_gc_hot_loop": gc_policy, "admitted_count": result.admitted_count, "delivered_count": len(result.deliveries), "effective_world_state_fps": round(len(result.deliveries) / wall_seconds, 6), diff --git a/tests/test_m48s_reference_graph_timing.py b/tests/test_m48s_reference_graph_timing.py index 812335c..d19e806 100644 --- a/tests/test_m48s_reference_graph_timing.py +++ b/tests/test_m48s_reference_graph_timing.py @@ -3,6 +3,7 @@ from __future__ import annotations import importlib.util import time from pathlib import Path +from unittest.mock import patch from k1link.perception.detector import DetectorFrameTiming from k1link.perception.recorded_source import DecodedFrameTiming @@ -108,3 +109,34 @@ def test_gc_pause_telemetry_attributes_collection_to_active_stage() -> None: } ] assert summary["maximum_event"]["duration_ms"] >= 1.0 + + +def test_cyclic_gc_policy_collects_outside_hot_loop_and_restores_state() -> None: + state = {"enabled": True} + collections = iter((3, 0)) + + def disable() -> None: + state["enabled"] = False + + def enable() -> None: + state["enabled"] = True + + with ( + patch.object(RUNNER.gc, "isenabled", side_effect=lambda: state["enabled"]), + patch.object(RUNNER.gc, "disable", side_effect=disable), + patch.object(RUNNER.gc, "enable", side_effect=enable), + patch.object(RUNNER.gc, "collect", side_effect=lambda: next(collections)), + ): + policy = RUNNER.CyclicGcHotLoopPolicy() + with policy: + assert state["enabled"] is False + + assert state["enabled"] is True + assert policy.to_dict() == { + "schema_version": RUNNER.GC_POLICY_SCHEMA, + "enabled_before": True, + "disabled_during_hot_loop": True, + "enabled_after": True, + "pre_collected": 3, + "post_collected": 0, + }