Add periodic cyclic GC to the service loop to stop unbounded memory growth

src/ess/livedata/core/cyclic_gc.py

@@ -0,0 +1,130 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2026 Scipp contributors (https://github.com/scipp)
+"""Deterministic cyclic garbage collection for long-running service loops.
+
+CPython triggers generation-2 collection on object *counts*, not bytes. The
+per-chunk task-graph updates in sciline/cyclebane leave a handful of
+self-referential ``networkx`` graphs behind for every processed chunk, each
+transitively holding that chunk's detector arrays (O(100 MB) for large
+detectors). Such graphs are reclaimable only by the cyclic collector, and in
+steady state the generation-2 collector never runs because a few huge arrays
+do not trip any count threshold. The result is unbounded growth of purely
+reclaimable garbage until the service is OOM-killed.
+
+This module makes collection deterministic instead of count-triggered:
+
+- :meth:`PeriodicGarbageCollector.freeze` moves everything alive at loop start
+  into the permanent generation, so a full collection only walks objects
+  allocated since (measured: ~0.5 ms per collection instead of ~84 ms).
+- :meth:`PeriodicGarbageCollector.maybe_collect` runs a full collection at a
+  fixed time interval from the service loop, bounding retained garbage to
+  roughly one interval's worth.
+
+This is a mitigation, not a fix: the garbage should not be created in the
+first place. See https://github.com/scipp/esslivedata/issues/1264 for the
+root-cause analysis and the upstream work it depends on.
+"""
+
+from __future__ import annotations
+
+import gc
+import time
+from collections.abc import Callable
+
+import structlog
+
+
+class PeriodicGarbageCollector:
+    """Runs full cyclic garbage collections on a fixed time interval.
+
+    Collection cost scales with the number of tracked objects allocated after
+    :meth:`freeze`, so it can drift upward as jobs accumulate; every
+    ``log_interval`` seconds a summary of collection durations is logged so
+    that drift is observed rather than assumed.
+
+    Durations and reclaimed counts are logged together because their
+    *combination* is what diagnoses. A full collection walks everything
+    reachable, so duration tracks the live tracked population, while the
+    reclaimed count tracks only the garbage. Rising duration at a flat
+    reclaimed count therefore means live objects are accumulating -- a
+    retained buffer rather than a cycle leak -- and no amount of collection
+    will help. At roughly 0.045 ms per 1000 live tracked objects the logged
+    duration converts back to an object count without a heap dump. Do not
+    read a rising duration as the benign drift described above without
+    checking the reclaimed count beside it: that pair is how the unbounded
+    batcher backlog was found after this mitigation was already deployed.
+
+    Parameters
+    ----------
+    interval:
+        Minimum seconds between collections. Chosen to match the ~1 s batch
+        cadence of the services: garbage from at most one batch cycle is
+        retained, at a per-collection cost that is negligible against it.
+    log_interval:
+        Seconds between summary log lines reporting collection statistics.
+    clock:
+        Monotonic time source, injectable for testing.
+    """
+
+    def __init__(
+        self,
+        interval: float = 1.0,
+        log_interval: float = 600.0,
+        clock: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self._interval = interval
+        self._log_interval = log_interval
+        self._clock = clock
+        self._logger = structlog.get_logger()
+        self._last_collect = clock()
+        self._last_log = clock()
+        self._count = 0
+        self._collected = 0
+        self._total_duration = 0.0
+        self._max_duration = 0.0
+
+    def freeze(self) -> None:
+        """Move all currently live objects into the permanent generation.
+
+        Call once, after service construction and before the processing loop.
+        Frozen objects are exempt from collection forever, which is safe for
+        process-lifetime service state but would silently leak anything
+        shorter-lived; per-job state is allocated later and stays collectable.
+        """
+        gc.collect()
+        gc.freeze()
+        self._logger.info("gc_freeze", frozen_objects=gc.get_freeze_count())
+
+    def maybe_collect(self) -> bool:
+        """Run a full collection if ``interval`` has elapsed since the last one.
+
+        Returns
+        -------
+        :
+            True if a collection ran.
+        """
+        now = self._clock()
+        if now - self._last_collect < self._interval:
+            return False
+        start = time.perf_counter()
+        collected = gc.collect()
+        duration = time.perf_counter() - start
+        self._last_collect = now
+        self._count += 1
+        self._collected += collected
+        self._total_duration += duration
+        self._max_duration = max(self._max_duration, duration)
+        if now - self._last_log >= self._log_interval:
+            self._logger.info(
+                "gc_collect_stats",
+                collections=self._count,
+                collected_objects=self._collected,
+                mean_ms=round(1e3 * self._total_duration / self._count, 2),
+                max_ms=round(1e3 * self._max_duration, 2),
+            )
+            self._last_log = now
+            self._count = 0
+            self._collected = 0
+            self._total_duration = 0.0
+            self._max_duration = 0.0
+        return True

src/ess/livedata/core/service.py

@@ -16,6 +16,7 @@
 import structlog
 
 from ..config.instruments import available_instruments
+from .cyclic_gc import PeriodicGarbageCollector
 from .processor import Processor
 
 
@@ -147,6 +148,12 @@ class Service(ServiceBase):
 
     Calls the injected processor in a loop with a configurable poll interval.
     If resources were passed, this class should be used as a context manager.
+
+    The loop periodically runs the cyclic garbage collector: per-chunk task
+    graph updates create self-referential garbage holding large arrays that
+    CPython's count-based collection never reclaims in steady state (see
+    https://github.com/scipp/esslivedata/issues/1264). Pass
+    ``garbage_collector=None`` to opt out.
     """
 
     def __init__(
@@ -157,13 +164,18 @@ def __init__(
         log_level: int = logging.INFO,
         poll_interval: float = 0.01,
         resources: ExitStack | None = None,
+        garbage_collector: PeriodicGarbageCollector | None = None,
+        collect_garbage: bool = True,
     ):
         super().__init__(name=name, log_level=log_level)
         self._poll_interval = poll_interval
         self._processor = processor
         self._thread: threading.Thread | None = None
         self._resources = resources
         self._worker_error: str | None = None
+        self._garbage_collector = garbage_collector or (
+            PeriodicGarbageCollector() if collect_garbage else None
+        )
 
     def __enter__(self) -> Self:
         """Enter the context manager protocol."""
@@ -196,9 +208,15 @@ def step(self) -> None:
     def _run_loop(self) -> None:
         """Main service loop"""
         try:
+            if self._garbage_collector is not None:
+                # Everything alive now is process-lifetime service state; per-job
+                # state is allocated later and stays collectable.
+                self._garbage_collector.freeze()
             while self.is_running:
                 start_time = time.monotonic()
                 self._processor.process()
+                if self._garbage_collector is not None:
+                    self._garbage_collector.maybe_collect()
                 elapsed = time.monotonic() - start_time
                 remaining = max(0.0, self._poll_interval - elapsed)
                 if remaining > 0:

tests/core/cyclic_gc_test.py

@@ -0,0 +1,140 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2026 Scipp contributors (https://github.com/scipp)
+import gc
+import threading
+import weakref
+
+import pytest
+
+from ess.livedata.core.cyclic_gc import PeriodicGarbageCollector
+from ess.livedata.core.service import Service
+
+
+class _SelfReferential:
+    """Object reclaimable only by the cyclic collector, like a materialised
+    networkx graph view (see issue #1264)."""
+
+    def __init__(self) -> None:
+        self.cycle = self
+
+
+@pytest.fixture
+def gc_disabled():
+    """Disable automatic collection so only explicit collects can reclaim."""
+    was_enabled = gc.isenabled()
+    gc.disable()
+    yield
+    if was_enabled:
+        gc.enable()
+
+
+class FakeClock:
+    def __init__(self) -> None:
+        self.now = 0.0
+
+    def __call__(self) -> float:
+        return self.now
+
+
+class TestPeriodicGarbageCollector:
+    def test_collects_cyclic_garbage(self, gc_disabled) -> None:
+        clock = FakeClock()
+        collector = PeriodicGarbageCollector(interval=1.0, clock=clock)
+        obj = _SelfReferential()
+        ref = weakref.ref(obj)
+        del obj
+        # Refcounting alone cannot reclaim the cycle.
+        assert ref() is not None
+        clock.now = 1.0
+        assert collector.maybe_collect() is True
+        assert ref() is None
+
+    def test_respects_interval(self, gc_disabled) -> None:
+        clock = FakeClock()
+        collector = PeriodicGarbageCollector(interval=10.0, clock=clock)
+        obj = _SelfReferential()
+        ref = weakref.ref(obj)
+        del obj
+        clock.now = 9.9
+        assert collector.maybe_collect() is False
+        assert ref() is not None
+        clock.now = 10.0
+        assert collector.maybe_collect() is True
+        assert ref() is None
+
+    def test_interval_measured_from_last_collection(self, gc_disabled) -> None:
+        clock = FakeClock()
+        collector = PeriodicGarbageCollector(interval=10.0, clock=clock)
+        clock.now = 25.0
+        assert collector.maybe_collect() is True
+        clock.now = 34.9
+        assert collector.maybe_collect() is False
+        clock.now = 35.0
+        assert collector.maybe_collect() is True
+
+    def test_freeze_exempts_preexisting_objects(self, gc_disabled) -> None:
+        collector = PeriodicGarbageCollector(interval=0.0, clock=FakeClock())
+        frozen_before = gc.get_freeze_count()
+        obj = _SelfReferential()
+        ref = weakref.ref(obj)
+        try:
+            collector.freeze()
+            assert gc.get_freeze_count() > frozen_before
+            # The cycle was alive at freeze time, so collection ignores it
+            # even after it becomes garbage.
+            del obj
+            collector.maybe_collect()
+            assert ref() is not None
+        finally:
+            gc.unfreeze()
+        gc.collect()
+        assert ref() is None
+
+
+class _CyclicGarbageProcessor:
+    """Processor that leaves one cyclic garbage object per process() call."""
+
+    def __init__(self) -> None:
+        self.refs: list[weakref.ref] = []
+        self.processed = threading.Event()
+
+    def process(self) -> None:
+        obj = _SelfReferential()
+        self.refs.append(weakref.ref(obj))
+        del obj
+        self.processed.set()
+
+    def finalize(self, *, error: str | None = None) -> None:
+        pass
+
+
+class TestServiceGarbageCollection:
+    def test_service_loop_collects_cyclic_garbage(self, gc_disabled) -> None:
+        processor = _CyclicGarbageProcessor()
+        service = Service(
+            processor=processor,
+            garbage_collector=PeriodicGarbageCollector(interval=0.0),
+        )
+        try:
+            service.start(blocking=False)
+            assert processor.processed.wait(timeout=5.0)
+        finally:
+            service.stop()
+            gc.unfreeze()
+        assert len(processor.refs) > 0
+        # Every cycle except possibly the last (created after the loop's final
+        # collect) was reclaimed without the automatic collector running.
+        assert sum(ref() is not None for ref in processor.refs) <= 1
+
+    def test_service_without_collector_leaves_cyclic_garbage(self, gc_disabled) -> None:
+        processor = _CyclicGarbageProcessor()
+        service = Service(processor=processor, collect_garbage=False)
+        try:
+            service.start(blocking=False)
+            assert processor.processed.wait(timeout=5.0)
+        finally:
+            service.stop()
+        assert len(processor.refs) > 0
+        assert all(ref() is not None for ref in processor.refs)
+        gc.collect()
+        assert all(ref() is None for ref in processor.refs)