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

[SimonHeybrock]

Summary

detector_data grows by roughly 50-65 MB per processing cycle and is eventually OOM-killed. On TBL, with the Timepix3 and ORCA detector views running, the service reached 29.9 GB resident on a 32 GB host and was killed 34 minutes after start.

The growth is entirely cyclic garbage: the live set stays constant. Every discarded object is reclaimable, but CPython's generation-2 collector never runs, because it is triggered by object counts and the wasted memory is a small number of very large arrays. Timepix3 makes it acute only because its arrays are the biggest (a 4096x4096 image is 134 MB); the mechanism applies to every instrument and every detector view.

This is not specific to saturation. In the minutes before the OOM the service was reporting batches=30, empty_batches=267-278 — i.e. not saturated — and still died four minutes later.

Evidence

Out of memory: Killed process ... (python) total-vm:34583736kB, anon-rss:31345108kB, ...
livedata-detector@tbl.service: Main process exited, code=killed, status=9/KILL
livedata-detector@tbl.service: Failed with result 'oom-kill'.
livedata-detector@tbl.service: Consumed 20min 9.441s CPU time.

errors=0 throughout; no exception, no traceback. Only detector_data died; monitor_data, data_reduction and timeseries were untouched.

Mechanism

Once per chunk, for every job:

1. ess/reduce/streaming.py:505 — StreamProcessor.accumulate does self._process_chunk_workflow[key] = value, inserting the incoming data into the task graph as a parameter. Same pattern at :525 in finalize and :442/:449 in set_context. Reached from src/ess/livedata/workflows/stream_processor_workflow.py:222 and :225.
2. sciline/data_graph.py:186 — DataGraph.__setitem__ forwards to self._cbgraph[key] = ....
3. cyclebane/graph.py:537 — Graph.__setitem__ rebuilds the graph and assigns self.graph = graph, dropping the previous networkx.DiGraph.
4. A networkx graph whose view attributes (edges, adj, degree, ...) have been materialised is self-referential: G.__dict__['edges'] = OutEdgeView(G), and the view stores self._graph = G. Such a graph is reclaimable only by the cyclic collector, and its node attribute dicts still transitively hold the previous cycle's arrays — measured payloads include a 134 MB DataArray {dim_0:4096, dim_1:4096}, a 134 MB DataArray {detector_number: 16777216} and a 67 MB Variable {4096,4096} int32.
5. Those graphs are live when generations 0 and 1 run, so they are promoted to generation 2 and only then become garbage. Over a 419-cycle run: 445 gen-0 collections, 40 gen-1, and 3 gen-2 — all three during startup, none in steady state. Nothing ever frees them.

Crucially, __setitem__ discards about 5.8 self-referential graphs per call, not one: _remove_ancestors, relabel_nodes and compose each build intermediates, and the validation loop's graph.pred[node] / graph.nodes[node] accesses materialise views on those intermediates.

There is currently no gc handling anywhere in src/ess/livedata/.

Reproduction

Everything below was measured against the versions pinned in requirements/base.txt: sciline==25.11.1, cyclebane==24.10.0.

python -m venv pinvenv
./pinvenv/bin/pip install "sciline==25.11.1" "cyclebane==24.10.0" numpy

Minimal — no ESSlivedata code at all

import gc, sys
from typing import NewType
import numpy as np
import sciline

Chunk = NewType('Chunk', np.ndarray)
Result = NewType('Result', float)


def total(chunk: Chunk) -> Result:
    return Result(float(chunk[0]))


def rss_mb() -> float:
    with open('/proc/self/statm') as f:
        return int(f.read().split()[1]) * 4096 / 1e6


do_collect = 'collect' in sys.argv
pipeline = sciline.Pipeline([total])
for i in range(60):
    # 4096x4096 float32 = 67 MB, the size of one Timepix3 detector image.
    pipeline[Chunk] = np.full(4096 * 4096, float(i), dtype=np.float32)
    pipeline.compute(Result)
    if do_collect:
        gc.collect()
    if i % 10 == 0:
        print(f'{i:>5} {rss_mb():>9.1f}')
print(f'end {rss_mb():.1f} MB')
gc.collect()
print(f'after final gc.collect(): {rss_mb():.1f} MB')

$ ./pinvenv/bin/python minimal_leak.py
   50       663.1
  end 1393.0 MB
after final gc.collect(): 117.8 MB

$ ./pinvenv/bin/python minimal_leak.py collect
   50       126.1
  end 126.1 MB

Unbounded growth; one gc.collect() recovers all of it; collecting every iteration keeps it flat. With gc.disable() the same loop reaches 4078 MB.

Full service

A harness driving the real pipeline in-process — real routing adapter, OrchestratingProcessor with AdaptiveMessageBatcher, JobManager/Job/StreamProcessorWorkflow, real da00/x5f2 sink serializer with the bytes discarded, only the Kafka transport faked (real ev44 flatbuffers via FakeConsumer on the production topic/source, ~45 msgs/s, ~135k events/cycle, real 4096x4096 grid, real tbl_detector_timepix3 workflow with ROI and device outputs):

420 cycles	RSS
no gc.collect()	1768 MB (c0) -> 5440 (c160) -> 9408 (c300) -> 9640 (c419); +65 MB/cycle, no plateau
gc.collect() each cycle	2018 (c20) -> 2163 (c259); slope -2.9 MB/cycle

Periodic gc.collect() + malloc_trim probes inside the leaking run show the live set is constant — 1689 MB at c30, 1897 MB at c419 — and each probe reclaimed exactly the accumulated garbage (7342 MB at c419).

I can attach this harness on request; it is ~400 lines and depends on repo internals.

Ruled out

• glibc fragmentation. MALLOC_MMAP_THRESHOLD_/MALLOC_TRIM_THRESHOLD_=128K changed nothing (8.4 GB at cycle 140). A single gc.collect() took 8370 MB to 1337 MB. Allocation-size variation does cause real arena growth (+13.0 MB/update with fluctuating event counts vs +5.1 constant) but it plateaus around 3.1 GB.
• scipp. group_event_data over the 16.7M-bin grid, sc.values, and bins.concat were each run 300 times on fixed-size inputs; all plateau, none leak.
• librdkafka producer buffering. sink.py calls flush(timeout=3) after every batch and it never timed out.
• ToNXevent_data peak-retaining buffers. Real (_ensure_capacity doubles and never shrinks; clear() and release_buffers() free nothing), but it needs growing batch sizes to fire and SimpleMessageBatcher windows by data time, so a backlog cannot inflate batches. It shows up as a decelerating creep in the live set (1.1 MB/cycle at c30-80 falling to 0.22 MB/cycle at c300-419) — two orders of magnitude below the primary leak. Worth its own issue.

Mitigation, and its cost

Measured in the minimal reproduction on the pinned versions:

	RSS	gc.collect() cost
per-cycle gc.collect()	flat	mean 9.94 ms, max 10.68 ms
gc.freeze() at startup + per-cycle gc.collect()	flat	mean 0.41 ms, max 0.48 ms

In the full service the same comparison was 84 ms without freeze and 0.6 ms mean / 2.1 ms max with it — the difference matters, because a Timepix3 view update already costs ~840 ms against a ~1 s batch cadence.

gc.freeze() after service construction moves everything alive at that point into the permanent generation, so the per-cycle collect only walks objects created since.

Caveats, if we ship this:

• gc.freeze() permanently exempts everything alive at call time. Safe immediately after construction, but anything frozen that later becomes garbage leaks silently.
• Full-collect cost scales with the number of tracked objects, so it will drift upward as jobs accumulate. It should be measured, not assumed.
• It does not remove the CPU cost of building ~6 throwaway graphs per chunk.

This is a symptom fix. It makes the collector run often enough to reclaim garbage that should not be created. It should be labelled as such and not close this issue.

Potential root-cause fixes

In increasing order of depth and payoff.

Break the cycles in cyclebane. Have Graph.__setitem__ leave no self-referential graphs behind. Note that the obvious version of this does not work: patching __setitem__ to drop the cached view attributes from the graph it replaces has no effect (4078 MB both patched and unpatched, with the cyclic collector disabled), because the intermediates built inside the call are self-referential too. A correct fix has to deal with all of them, or avoid materialising views on graphs that are about to be discarded.

Stop rebuilding the graph per chunk. StreamProcessor.accumulate reassigning a graph parameter for every incoming chunk means a full graph rebuild per chunk, with 134 MB payloads living in graph node attribute dicts. A task graph is being used as a data container. If per-chunk data were passed as an argument rather than written into the graph structure, neither the cycles nor the count-based GC heuristic would matter. This is the fix that makes the problem go away permanently, and it is an ess.reduce/sciline design question rather than something we can do here.

Not fixable by us: CPython triggers generation-2 collection on object counts, not bytes, so a handful of 100 MB arrays trips nothing. This is the reason the other two layers become fatal rather than merely wasteful.

Related findings

Separate defects found while investigating; each probably warrants its own issue rather than being folded in here.

• Preprocessing is not gated on active jobs. MessagePreprocessor.preprocess_messages (core/orchestrating_processor.py:122) runs for every stream that arrives regardless of whether any job consumes it, and _accumulators has no eviction path. Once Timepix3 has been seen, GroupByPixel.get() keeps grouping into 16.7M bins every batch forever — about 100 ms and 440 MB of transient allocation per batch, for nothing. Stopping a workflow does not stop paying for it.
• queue.buffering.max.kbytes: 104857600 in config/defaults/kafka_dev.yaml and kafka_docker.yaml.jinja is in kilobytes, so it is 100 GB, not the 100 MB the adjacent comment claims. It sits directly under message.max.bytes, which is in bytes. At 100 GB the BufferError backpressure path in kafka/sink.py is unreachable — the host dies first. Should be 102400.
• num_bins is bounded without reference to pixel count. parameter_models.py:87 allows le=10000. Timepix3 view memory scales at a measured 4.24 MB per TOA bin (two 512x512xN float64 histograms; theory says 4.19), so the maximum is roughly 42 GB for a single job, settable from the dashboard. The bound should be on pixels * bins, not bins alone.
• Timepix3 view cost. ~840 ms per update against a ~1 s batch cadence, so one view consumes ~84% of the budget and two saturate the service (observed in production: batches 30 -> 21 per 30 s). Attribution per update: _concat_bins 432 ms, sc.values 210 ms, group_event_data 96 ms — all set by the declared 4096x4096 grid size. Overlaps with the discussion in the Timepix3 resolution PR.

[SimonHeybrock]

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

[SimonHeybrock]

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.