detector_data leaks ~60 MB per cycle in uncollected reference cycles from sciline graph updates, OOM-killing the service #1264

Activity

1. SimonHeybrock

   added

   bug

   area:backendServices, Kafka, message pipeline, preprocessors, job handlingServices, Kafka, message pipeline, preprocessors, job handling

   upstreamBlocked on Panel, Bokeh, HoloViews or another dependencyBlocked on Panel, Bokeh, HoloViews or another dependency

   on Aug 28, 2026
2. SimonHeybrock commented on Aug 28, 2026

   

   SimonHeybrock

   on Aug 28, 2026

   MemberAuthor

   More actions

   Full-service reproduction harness

   Attaching the harness referenced above, since it is otherwise unpreserved. This is a debugging snapshot, not maintained code — it is pinned to the state of the tree on 2026-08-28 and will drift. The minimal reproduction in the issue body is the one to hand to upstream; this one exists to show the leak manifests through the real service wiring, and to reproduce the +65 MB/cycle figure.

   Save both files into a checkout root and run with PYTHONPATH=src:

   # leak, no collection
   PYTHONPATH=src python leak_pipeline.py --cycles 420 --sample-every 20
   
   # flat, with per-cycle collection
   PYTHONPATH=src python leak_pipeline.py --cycles 260 --collect-every 1
   
   # live-set probes inside a leaking run (gc.collect + malloc_trim, then continue)
   PYTHONPATH=src python leak_pipeline.py --cycles 420 --probe 30,80,160,300,419
   
   # what the cyclic collector is holding, at cycle 30
   PYTHONPATH=src python leak_pipeline.py --saveall 30
   
   # mitigation cost
   PYTHONPATH=src python leak_pipeline.py --cycles 200 --collect-every 1 --freeze

   Two environment caveats that matter for interpreting results:

   • --no-eager-free is on by default. The sciline checkout used here was an editable dev version whose NaiveScheduler frees intermediates as soon as their dependent count reaches zero; that raises KeyError: EmptyDetector[SampleRun] on the Timepix3 graph, because a requested target key is also consumed as an argument by another provider. Released sciline does not have that optimisation, so it is disabled to match production. Pass --eager-free to restore it.
   • Not exercised: librdkafka itself, live ROI updates from the dashboard, f144 context streams, and multiple concurrent jobs.

   scipp_sizes.py is only needed for --saveall; put it next to leak_pipeline.py.

   leak_pipeline.py

   """Long-running, faithful reproduction of the production `detector_data` service
   for TBL/Timepix3, driven entirely in-process with fakes.
   
   Mirrors production wiring:
     make_detector_service_builder(instrument='tbl') ->
     AdaptingMessageSource(RoutingAdapter) -> DeviceSynthesizer ->
     OrchestratingProcessor(job_threads=5,
                            AdaptiveMessageBatcher(RateAwareMessageBatcher)) ->
     UnrollingSinkAdapter(sink with the real da00/x5f2 serializers)
   
   The only production component replaced is the Kafka transport itself: the
   consumer is `tests.helpers.livedata_app.FakeConsumer` and the sink serializes
   every message with the real serializer and then drops the bytes (producer-side
   buffering was already ruled out separately).
   """
   
   from __future__ import annotations
   
   import argparse
   import gc
   import resource
   import sys
   import time
   import uuid
   
   import numpy as np
   from streaming_data_types import eventdata_ev44
   
   from ess.livedata.config import instrument_registry, workflow_spec
   from ess.livedata.config.streams import stream_kind_to_topic
   from ess.livedata.config.workflow_spec import JobId
   from ess.livedata.core.message import StreamKind
   from ess.livedata.core.message_batcher import AdaptiveMessageBatcher
   from ess.livedata.core.rate_aware_batcher import RateAwareMessageBatcher
   from ess.livedata.core.sciline_scheduler import configure_sciline_scheduler
   from ess.livedata.kafka.message_adapter import FakeKafkaMessage
   from ess.livedata.kafka.sink import UnrollingSinkAdapter
   from ess.livedata.kafka.sink_serializers import make_default_sink_serializer
   from ess.livedata.services.detector_data import make_detector_service_builder
   
   sys.path.insert(0, '.')
   from tests.helpers.livedata_app import FakeConsumer  # noqa: E402
   
   
   def _unpatch_sciline_eager_free() -> None:
       """Restore released-sciline NaiveScheduler semantics.
   
       The sciline checkout on this machine (editable, 26.8.0, PR #231
       "naive-scheduler-reduce-memory-use") deletes intermediate results as soon
       as their dependent count reaches zero. That crashes with ``KeyError`` when
       a requested target key is also consumed as an argument by another
       provider, which is exactly what the Timepix3 detector-view graph does
       (``EmptyDetector[SampleRun]``). Production runs a released sciline without
       that optimisation, so we disable it to reproduce production behaviour.
       """
       import sciline.scheduler as _sched
   
       _sched._consume_arguments = lambda provider, counts, results: None
   
   
   def rss_mb() -> float:
       with open('/proc/self/statm') as f:
           return int(f.read().split()[1]) * 4096 / 1e6
   
   
   def peak_mb() -> float:
       return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1e3
   
   
   def _malloc_trim() -> None:
       import ctypes
   
       try:
           ctypes.CDLL('libc.so.6').malloc_trim(0)
       except Exception as e:  # noqa: BLE001
           print(f'  malloc_trim unavailable: {e}')
   
   
   class SerializingDiscardSink:
       """Sink that runs the real serializer and drops the bytes."""
   
       def __init__(self, serializer) -> None:
           self._serializer = serializer
           self.n_serialized = 0
           self.n_failed = 0
           self.total_bytes = 0
   
       def publish_messages(self, messages) -> None:
           for msg in messages:
               try:
                   out = self._serializer.serialize(msg)
               except Exception as e:  # noqa: BLE001
                   self.n_failed += 1
                   if self.n_failed < 5:
                       print(f'  [sink] serialize failed: {type(e).__name__}: {e}')
                   continue
               self.n_serialized += 1
               self.total_bytes += len(out.value)
               del out
   
   
   class NullSink:
       """Sink that drops everything without serializing (bisection control)."""
   
       def __init__(self) -> None:
           self.n_serialized = 0
           self.n_failed = 0
           self.total_bytes = 0
   
       def publish_messages(self, messages) -> None:
           self.n_serialized += len(messages)
   
   
   def report_saveall() -> None:
       """Collect with DEBUG_SAVEALL and report what the cyclic collector found."""
       from collections import Counter
   
       sys.path.insert(0, '.')
       import scipp as sc
   
       from scipp_sizes import obj_bytes
   
       scipp_types = (sc.Variable, sc.DataArray, sc.Dataset, sc.DataGroup)
   
       rss0 = rss_mb()
       print(f'RSS before collect: {rss0:.1f} MB')
       gc.set_debug(gc.DEBUG_SAVEALL)
       n = gc.collect()
       gc.set_debug(0)
       garbage = gc.garbage
       print(f'collected {n} objects into gc.garbage ({len(garbage)}), '
             f'RSS still {rss_mb():.1f} MB')
       print('  by type:', Counter(type(o).__name__ for o in garbage).most_common(20))
   
       # scipp objects are pybind11 types the cyclic collector does not track, so
       # they never appear in gc.garbage themselves. BFS out of the garbage roots
       # via gc.get_referents to find the data they keep alive.
       import numpy as _np
   
       seen_ids: set[int] = set(id(o) for o in garbage)
       seen_buf: set[int] = set()
       found: list[tuple[int, object]] = []
       frontier = list(garbage)
       depth = 0
       while frontier and depth < 10:
           nxt = []
           for o in frontier:
               for r in gc.get_referents(o):
                   if id(r) in seen_ids:
                       continue
                   seen_ids.add(id(r))
                   if isinstance(r, scipp_types):
                       b = obj_bytes(r, seen_buf)
                       if b > 1_000_000:
                           found.append((b, r))
                       continue
                   if isinstance(r, _np.ndarray):
                       if r.nbytes > 1_000_000 and id(r) not in seen_buf:
                           seen_buf.add(id(r))
                           found.append((r.nbytes, r))
                       continue
                   if isinstance(r, str | bytes | int | float | type):
                       continue
                   nxt.append(r)
           frontier = nxt
           depth += 1
   
       found.sort(key=lambda t: -t[0])
       print(
           f'data reachable from cyclic garbage, >1 MB: {len(found)} objects, '
           f'total {sum(b for b, _ in found) / 1e6:.1f} MB'
       )
       agg: Counter[str] = Counter()
       aggn: Counter[str] = Counter()
       for b, obj in found:
           try:
               key = f'{type(obj).__name__} sizes={dict(obj.sizes)} dtype={obj.dtype}'
           except Exception:
               key = f'{type(obj).__name__} {getattr(obj, "shape", "")} '
               f'{getattr(obj, "dtype", "")}'
           agg[key] += b
           aggn[key] += 1
       for key, b in agg.most_common(15):
           print(f'  {b / 1e6:9.1f} MB  x{aggn[key]:<4} {key}')
   
       del garbage[:]
       gc.collect()
       rss2 = rss_mb()
       print(f'RSS after dropping garbage: {rss2:.1f} MB  '
             f'(freed {rss0 - rss2:.1f} MB)')
   
   
   def build(args):
       if args.no_eager_free:
           _unpatch_sciline_eager_free()
       configure_sciline_scheduler(args.scheduler)
       builder = make_detector_service_builder(instrument='tbl', dev=args.dev)
       builder.job_threads = args.job_threads
       builder.message_batcher = AdaptiveMessageBatcher(
           inner_factory=RateAwareMessageBatcher
       )
       if args.sink == 'null':
           inner = NullSink()
       else:
           inner = SerializingDiscardSink(make_default_sink_serializer(instrument='tbl'))
       consumer = FakeConsumer()
       service = builder.from_consumer(
           consumer=consumer,
           sink=UnrollingSinkAdapter(inner),
           raise_on_adapter_error=True,
           use_background_source=False,
       )
       return service, consumer, inner
   
   
   def workflow_id_for(name: str):
       reg = instrument_registry['tbl'].workflow_factory
       for wid, spec in reg.items():
           if spec.name == name:
               return wid
       raise SystemExit(
           f'workflow {name!r} not found; have: '
           f'{sorted(s.name for s in reg.values())}'
       )
   
   
   def main() -> None:
       p = argparse.ArgumentParser()
       p.add_argument('--cycles', type=int, default=400)
       p.add_argument('--msgs-per-cycle', type=int, default=45)
       p.add_argument('--events-mean', type=int, default=5000)
       p.add_argument(
           '--events-jitter',
           type=float,
           default=0.5,
           help='relative fluctuation of per-message event count',
       )
       p.add_argument('--sample-every', type=int, default=10)
       p.add_argument('--job-threads', type=int, default=5)
       p.add_argument('--sink', choices=['real', 'null'], default='real')
       p.add_argument('--workflow', default='tbl_detector_timepix3')
       p.add_argument('--source', default='timepix3_detector')
       p.add_argument('--dev', action='store_true', default=False)
       p.add_argument('--scheduler', default='naive')
       p.add_argument(
           '--eager-free',
           dest='no_eager_free',
           action='store_false',
           default=True,
           help='keep this sciline checkout\'s eager intermediate freeing',
       )
       p.add_argument('--gc-types', action='store_true')
       p.add_argument(
           '--collect-every',
           type=int,
           default=0,
           help='call gc.collect() every N cycles (0 = never)',
       )
       p.add_argument('--gc-stats', action='store_true')
       p.add_argument(
           '--freeze',
           action='store_true',
           help='gc.freeze() after the workflow is built, to shrink the gen-2 scan',
       )
       p.add_argument(
           '--freeze-at-start',
           action='store_true',
           help='gc.freeze() right after service construction, before any job',
       )
       p.add_argument(
           '--probe',
           default='',
           help='comma-separated cycles at which to gc.collect()+malloc_trim() '
           'and report the live set',
       )
       p.add_argument(
           '--saveall',
           type=int,
           default=0,
           help='at cycle N: gc.set_debug(DEBUG_SAVEALL), collect, report garbage',
       )
       p.add_argument('--tracemalloc', action='store_true')
       p.add_argument(
           '--stop-after',
           type=int,
           default=0,
           help='stop the job after N cycles and keep stepping',
       )
       args = p.parse_args()
   
       print(f'args: {vars(args)}')
       print(f'RSS before build: {rss_mb():.1f} MB')
       service, consumer, sink = build(args)
       processor = service._processor
       print(f'RSS after build:  {rss_mb():.1f} MB')
       if args.freeze_at_start:
           gc.collect()
           gc.freeze()
           print(f'gc.freeze() at service start: {gc.get_freeze_count()} objects')
   
       from ess.livedata.config.streams import get_stream_mapping
   
       mapping = get_stream_mapping(instrument='tbl', dev=args.dev)
       key = next(k for k, v in mapping.detectors.items() if v == args.source)
       topic, ev44_source = key.topic, key.source_name
       print(f'detector stream: topic={topic!r} ev44 source_name={ev44_source!r}')
   
       wid = workflow_id_for(args.workflow)
       cfg = workflow_spec.WorkflowConfig(
           identifier=wid,
           job_id=JobId(source_name=args.source, job_number=uuid.uuid4()),
       )
       consumer.add_message(
           FakeKafkaMessage(
               key=None,
               value=cfg.model_dump_json().encode('utf-8'),
               topic=stream_kind_to_topic(
                   instrument='tbl', kind=StreamKind.LIVEDATA_COMMANDS
               ),
               timestamp=0,
           )
       )
       service.step()
       print(f'RSS after workflow start: {rss_mb():.1f} MB')
   
       if args.freeze:
           gc.collect()
           gc.freeze()
           print(f'gc.freeze(): {gc.get_freeze_count()} objects moved to permanent gen')
   
       rng = np.random.default_rng(42)
       npix = 4096 * 4096
       t_ns = 1_000_000_000
       step_ns = 1_000_000_000 // args.msgs_per_cycle
   
       probe_cycles = {int(c) for c in args.probe.split(',') if c.strip()}
       collect_times: list[float] = []
       snap0 = None
       base_counts = None
       if args.tracemalloc:
           import tracemalloc
   
           tracemalloc.start(20)
   
       print()
       hdr = (
           f'{"cycle":>6} {"steps":>6} {"events":>10} {"cyc_s":>7} '
           f'{"RSS MB":>9} {"dRSS":>8} {"peak":>9} {"msgs_out":>9}'
       )
       print(hdr)
       t_start = time.perf_counter()
       rss_prev = rss_mb()
       rows = []
       stopped = False
       for cycle in range(args.cycles):
           if args.stop_after and cycle == args.stop_after and not stopped:
               stopped = True
               from ess.livedata.core.job_manager import JobAction, JobCommand
   
               cmd = JobCommand(action=JobAction.stop)
               consumer.add_message(
                   FakeKafkaMessage(
                       key=None,
                       value=cmd.model_dump_json().encode('utf-8'),
                       topic=stream_kind_to_topic(
                           instrument='tbl', kind=StreamKind.LIVEDATA_COMMANDS
                       ),
                       timestamp=0,
                   )
               )
               print(f'--- stopping job at cycle {cycle} ---')
   
           n_ev_total = 0
           for i in range(args.msgs_per_cycle):
               n = max(
                   1,
                   int(
                       args.events_mean
                       * (1.0 + args.events_jitter * rng.standard_normal())
                   ),
               )
               n_ev_total += n
               toa = rng.uniform(0, 70_000_000, n).astype(np.int32)
               pid = rng.integers(1, npix + 1, n, dtype=np.int32)
               consumer.add_message(
                   FakeKafkaMessage(
                       value=eventdata_ev44.serialise_ev44(
                           source_name=ev44_source,
                           message_id=0,
                           reference_time=[],
                           reference_time_index=0,
                           time_of_flight=toa,
                           pixel_id=pid,
                       ),
                       topic=topic,
                       timestamp=t_ns + i * step_ns,
                   )
               )
           t_ns += 1_000_000_000
   
           t0 = time.perf_counter()
           steps = 0
           # Drain the fake consumer: source takes <=100 messages per step.
           while consumer._messages:
               service.step()
               steps += 1
               if steps > 50:
                   break
           dt = time.perf_counter() - t0
   
           if args.collect_every and cycle % args.collect_every == 0:
               _t = time.perf_counter()
               _n = gc.collect()
               collect_times.append(time.perf_counter() - _t)
   
           if cycle in probe_cycles:
               r_before = rss_mb()
               n_coll = gc.collect()
               r_gc = rss_mb()
               _malloc_trim()
               r_trim = rss_mb()
               print(
                   f'PROBE cycle {cycle}: RSS {r_before:.1f} -> gc {r_gc:.1f} '
                   f'(-{r_before - r_gc:.1f}, {n_coll} objs) -> trim {r_trim:.1f} '
                   f'(-{r_gc - r_trim:.1f})  LIVE={r_trim:.1f} MB',
                   flush=True,
               )
   
           if args.saveall and cycle == args.saveall:
               report_saveall()
               return
   
           if cycle % args.sample_every == 0 or cycle == args.cycles - 1:
               r = rss_mb()
               print(
                   f'{cycle:>6} {steps:>6} {n_ev_total:>10} {dt:>7.2f} '
                   f'{r:>9.1f} {r - rss_prev:>+8.1f} {peak_mb():>9.1f} '
                   f'{sink.n_serialized:>9}',
                   flush=True,
               )
               rows.append((cycle, r))
               rss_prev = r
   
               if args.gc_types:
                   from collections import Counter
   
                   c = Counter(type(o).__name__ for o in gc.get_objects())
                   if base_counts is None:
                       base_counts = c
                   else:
                       delta = {
                           k: c[k] - base_counts.get(k, 0)
                           for k in c
                           if c[k] - base_counts.get(k, 0) > 50
                       }
                       top = sorted(delta.items(), key=lambda kv: -kv[1])[:10]
                       print(f'         gc-delta: {top}', flush=True)
   
               if args.gc_stats:
                   print(
                       f'         gc: count={gc.get_count()} '
                       f'stats={[(s["collections"], s["collected"], s["uncollectable"]) for s in gc.get_stats()]}',
                       flush=True,
                   )
   
               if args.tracemalloc:
                   import tracemalloc
   
                   snap = tracemalloc.take_snapshot()
                   if snap0 is not None:
                       stats = snap.compare_to(snap0, 'lineno')[:8]
                       for s in stats:
                           print(f'         tm: {s}', flush=True)
                   snap0 = snap
   
       print()
       print(f'total wall: {time.perf_counter() - t_start:.1f} s')
       if collect_times:
           ct = sorted(collect_times)
           print(
               f'gc.collect(): n={len(ct)} mean={1e3 * sum(ct) / len(ct):.1f} ms '
               f'median={1e3 * ct[len(ct) // 2]:.1f} ms max={1e3 * ct[-1]:.1f} ms'
           )
       gc.collect()
       print(f'final RSS after gc: {rss_mb():.1f} MB, peak {peak_mb():.1f} MB')
       print(
           f'sink: serialized={sink.n_serialized} failed={sink.n_failed} '
           f'bytes={sink.total_bytes / 1e6:.1f} MB'
       )
       if len(rows) > 3:
           (c0, r0), (c1, r1) = rows[len(rows) // 2], rows[-1]
           print(
               f'second-half slope: {(r1 - r0) / max(1, c1 - c0):.2f} MB/cycle '
               f'(cycles {c0}..{c1})'
           )
       _ = processor
   
   
   if __name__ == '__main__':
       main()

   scipp_sizes.py

   """Walk live scipp objects and attribute bytes, for leak localisation.
   
   Python-level object *counts* stay flat in the detector_data leak, so the growth
   must sit inside a fixed set of scipp containers. This module measures the
   underlying buffer bytes of every live scipp object and tracks them by ``id`` so
   a growing container can be identified across cycles.
   """
   
   from __future__ import annotations
   
   import gc
   
   import scipp as sc
   
   _SCIPP_TYPES = (sc.Variable, sc.DataArray, sc.Dataset, sc.DataGroup)
   
   
   def _var_bytes(v: sc.Variable, seen: set[int]) -> int:
       total = 0
       try:
           const = v.bins.constituents if v.bins is not None else None
       except Exception:
           const = None
       if const is not None:
           # begin/end index variable
           idx = const.get('begin')
           if idx is not None:
               total += _plain_bytes(idx, seen)
           idx = const.get('end')
           if idx is not None:
               total += _plain_bytes(idx, seen)
           data = const.get('data')
           if data is not None:
               total += obj_bytes(data, seen)
           return total
       return _plain_bytes(v, seen)
   
   
   def _plain_bytes(v: sc.Variable, seen: set[int]) -> int:
       total = 0
       for attr in ('values', 'variances'):
           try:
               arr = getattr(v, attr)
           except Exception:
               continue
           if arr is None:
               continue
           nbytes = getattr(arr, 'nbytes', None)
           if nbytes is None:
               continue
           base = arr
           while getattr(base, 'base', None) is not None:
               base = base.base
           key = base.__array_interface__['data'][0] if hasattr(
               base, '__array_interface__'
           ) else id(base)
           if key in seen:
               continue
           seen.add(key)
           total += nbytes
       return total
   
   
   def obj_bytes(obj, seen: set[int] | None = None) -> int:
       if seen is None:
           seen = set()
       if isinstance(obj, sc.Variable):
           return _var_bytes(obj, seen)
       if isinstance(obj, sc.DataArray):
           total = _var_bytes(obj.data, seen)
           for c in obj.coords.values():
               total += _var_bytes(c, seen)
           for m in obj.masks.values():
               total += _var_bytes(m, seen)
           return total
       if isinstance(obj, sc.Dataset):
           return sum(obj_bytes(da, seen) for da in obj.values())
       if isinstance(obj, sc.DataGroup):
           return sum(
               obj_bytes(v, seen)
               for v in obj.values()
               if isinstance(v, _SCIPP_TYPES)
           )
       return 0
   
   
   def snapshot() -> tuple[dict[int, int], dict[int, str], int]:
       """Return (id -> bytes, id -> repr-ish label, total bytes) of live scipp objects."""
       gc.collect()
       sizes: dict[int, int] = {}
       labels: dict[int, str] = {}
       seen: set[int] = set()
       objs = [o for o in gc.get_objects() if isinstance(o, _SCIPP_TYPES)]
       # Largest-first so shared buffers are attributed to the biggest container.
       for o in objs:
           n = obj_bytes(o, seen)
           if n < 1_000_000:
               continue
           sizes[id(o)] = n
           try:
               labels[id(o)] = f'{type(o).__name__} sizes={dict(o.sizes)}'
           except Exception:
               labels[id(o)] = type(o).__name__
       return sizes, labels, sum(sizes.values())
3. SimonHeybrock

   self-assigned this

   on Aug 29, 2026
4. SimonHeybrock

   added a commit that references this issue on Aug 31, 2026

   Avoid reference cycles in graphs

   ...

   f3fd0e4
5. SimonHeybrock commented on Sep 22, 2026

   

   SimonHeybrock

   on Sep 22, 2026

   MemberAuthor

   More actions

   Closing: the cyclic garbage described here is removed at its source.

   • Avoid reference cycles in graphs cyclebane#33 stops cyclebane from leaving self-referential networkx graphs behind (cyclebane 26.9.0).
   • Feed streaming values into workflows instead of assigning them ess#732 feeds streaming values into the workflow instead of assigning them with workflow[key] = value, so the graph is no longer rebuilt per chunk (essreduce 26.9.0).
   • Require essreduce 26.9.0 and essspectroscopy 26.9.0 #1317 requires both versions; released in 26.9.2.

   The minimal reproduction above now stays flat: RSS is constant over all 60 cycles with cyclebane 26.9.0, and a final gc.collect() frees nothing.

   One correction to the description: "the live set stays constant" held only while the service kept up. The staging deploy of #1265 still OOMed because the live set was growing. The Timepix3 view took ~11 s per update, longer than the 8 s maximum batch, so unprocessed messages accumulated in the batcher's unbounded _overflow. #1270 removed that per-update cost, and #1271 bounds _overflow so a workflow slower than the maximum batch length drops the oldest data instead of running out of memory.

   Not yet confirmed in production: a long detector_data run on TBL with Timepix3 active on 26.9.2. Reopen if it still grows.
6. SimonHeybrock

   closed this as completedon Sep 22, 2026