Feed streaming values into workflows instead of assigning them - #732
Feed streaming values into workflows instead of assigning them#732SimonHeybrock wants to merge 2 commits into
Conversation
github-actions
Bot
changed the title
7676f52 to
c8f6aee
Compare
SimonHeybrock
changed the title
c8f6aee to
e1a5f34
Compare
e1a5f34 to
6569aea
Compare
StreamProcessor used its internal pipelines as data containers: every chunk, every accumulator value and every context update was handed over with `pipeline[key] = value`. Each such assignment rebuilds the underlying graph, although the graph structure does not depend on the values, and the value stays reachable from the long-lived pipeline until the next assignment overwrites it. Both costs are paid at batch cadence, for the largest objects in the process: the discarded graphs hold the arrays in node attribute dicts and are reclaimable only by the cyclic collector, whose count-based heuristic does not trigger for a few huge objects. Feed the values through a mapping that the workflow reads from instead. The first value for a key is still assigned, once, so the branch it replaces is pruned as before and every later cycle computes on the same small graph. Later values are dict writes, and the graph is left alone. Chunks are released once the accumulators have been pushed, and accumulator values once the targets have been computed. The latter is scipp/esslivedata#1259: an accumulator that drops its value in on_finalize(), such as a sliding window, previously had that array pinned by the finalize pipeline for a full cycle. Context values are not released, being meant for cycles that do not feed them again, and neither are the dynamic keys that finalize() reads directly. The latter is a per-key property rather than a property of allow_bypass: a key whose only path to a target runs through an accumulator is not read at finalize, so it is released with its chunk even when the flag is set. Which keys those are is determined once, from the graph the finalize workflow computes on. Context goes through the same path. It only looks like a rare one: a third of the context keys of esslivedata's workflows are f144 device logs, which arrive at the same cadence as the chunks, and one such update assigned twice per key plus once per cached node derived from it -- ten graph rebuilds for a detector view, more than the chunk path itself costs. Discussed in scipp/esslivedata#1267. Measured on a minimal cumulative-histogram processor with 16 MB chunks, over 20 accumulate+finalize cycles with the cyclic collector disabled: streaming cyclebane ceedc52 cyclebane 5cf257e assignment 138 -> 747 MB 138 MB, flat feeding 106 MB, flat 106 MB, flat Keeping the graph static removes the unbounded growth independently of the cyclebane fix for the reference cycles, and additionally drops the steady-state footprint by one chunk plus one accumulator value.
6569aea to
7020f11
Compare
| self._target_keys = target_keys | ||
| self._allow_bypass = allow_bypass | ||
|
|
||
| # Chunks, accumulator values, and context are fed to the workflows rather than |
There was a problem hiding this comment.
__init__ is too long and complex. This is not new but this addition makes it worse. Can you refactor it, ideally to simply set attributes instead of containing procedural code?
This may just be a symptom of this class being too large. It has a lot of attributes. It might help to reorganise them.
There was a problem hiding this comment.
Agreed — and it was your comment that made me look at it properly rather than adding to it. The five analyses it was doing inline are now module-level functions (_validate_streaming_keys, _build_streaming_workflow, _map_context_to_cached_nodes, _make_accumulators, _find_keys_read_at_finalize), so __init__ is assignments plus those calls plus the wiring of the three workflows.
On the attribute count: encapsulating the pipelines removed three, and _allow_bypass turned out to have been dead since the previous commit, so that went too.
| def __setitem__(self, key: sciline.typing.Key, value: Any) -> None: | ||
| if key not in self._values: | ||
| self._workflow[key] = value # prunes the branch this value replaces | ||
| self._insert_provider(key) |
There was a problem hiding this comment.
Why isn't this done by Sciline? Why does inserting a parameter prune but inserting a provider does not?
Also, instead of abusing __setitem__ here, how about adding a prune method to sciline.Pipeline?
There was a problem hiding this comment.
Good question, and it pushed me somewhere useful. It isn't parameter vs provider: __setitem__ goes through cyclebane's _remove_ancestors and drops the ancestors, whereas insert goes through _get_clean_node, which cuts only the node's incoming edges and leaves them orphaned. The orphans matter because to_task_graph starts from to_networkx(), which copies the whole data graph before subgraphing to the targets' ancestors — so a dead branch is copied on every get(). That's now in the class docstring rather than the one-line comment you were reading.
Your prune suggestion is the right shape and I've opened scipp/sciline#242. It would clean up more than this class: workflow[key] = None # hack to prune branches predates this PR in the same file.
Checking the assignment against your question also showed it didn't need to happen at feed time at all. Pruning doesn't depend on the value, and which keys each workflow is fed follows from the graph, so both are known at construction. Inputs are now wired once, there, which removes the first-value special case entirely: after construction the processor rebuilds no graph at all, where the previous version still did one rebuild per key. The resulting graphs are identical to before, node for node and edge for edge.
| self._values[key] = None | ||
|
|
||
| def _insert_provider(self, key: sciline.typing.Key) -> None: | ||
| values = self._values |
There was a problem hiding this comment.
Is this about preventing a reference cycle from the pipeline to itself? If so, can you add a comment?
There was a problem hiding this comment.
Yes, exactly that, and it deserved a comment. Capturing self would close the cycle pipeline → provider → closure → _FedWorkflow → pipeline, which only the cyclic collector can break — which is precisely what this PR is trying to stop depending on. Comment added.
| # Seeding the mapping here would make the next feed skip the assignment | ||
| # that wires the key up, so the value would never reach the workflow. | ||
| raise KeyError(f"Cannot release '{key}': no value was fed for it") | ||
| self._values[key] = None |
There was a problem hiding this comment.
What is the purpose of this? After releasing, it still produces a value but not the one the consumer expected. So won't this raise a TypeError in the provider?
There was a problem hiding this comment.
Fair, and you've found the weak spot: it was safe only by convention — every release sits immediately after the compute that consumed the value, every read is preceded by a feed — and nothing enforced that.
release now drops the entry outright, and the provider raises if a key is read with no value, naming the key. That also removed the "seeding the mapping here…" comment, which existed only because one dict was doing double duty as both "current values" and "keys already wired up"; those are now separate concerns.
One visible consequence: the never-set-context test used to pass on a TypeError from arithmetic on the None the branch was pruned with, and now gets ValueError: No value was fed for 'Context'.
| # assigned to them, see :class:`_Inputs`. | ||
| self._context_inputs = _Inputs(self._context_workflow) | ||
| self._chunk_inputs = _Inputs(self._process_chunk_workflow) | ||
| self._finalize_inputs = _Inputs(self._finalize_workflow) |
There was a problem hiding this comment.
I think _Inputs has a risky API. It shares a workflow with the StreamProcessor and it modifies that workflow. But then you access the workflow directly through, e.g., self._context_workflow. So you rely on a hidden dependency which prevents local reasoning.
How about you instead encapsulate the workflow in a new type and only access it through that type?
There was a problem hiding this comment.
Agreed — two paths to the same pipeline, and neither end readable on its own. It's now _FedWorkflow, which owns its pipeline and exposes compute(). The three _context_workflow / _chunk_workflow / _finalize_workflow attributes are the only handles, and the parallel _..._inputs ones are gone.
|
|
||
| # The accumulator dropped the value in on_finalize, so nothing should keep it | ||
| # alive. Note that this must hold without running the cyclic garbage collector. | ||
| assert ref() is None |
There was a problem hiding this comment.
Just to be clear: This ties livedata to CPython because, e.g., PyPy and GraalPy use garbage collection instead of ref counting. That is probably fine because we compile Scipp agains CPython, I just wanted to make sure you are aware.
There was a problem hiding this comment.
Yes — thanks for spelling it out. Scipp is a CPython extension so these tests can't run under PyPy or GraalPy regardless, but the assumption was implicit in the assertion; the test comments now say it relies on reference counting.
Related: the warm-up cycle those tests needed is gone, since with the wiring moved to construction the retention holds from the first chunk.
Feeding values instead of assigning them made the first value for each key a special case: it was assigned, to prune the branch it replaces, and only later values went through the mapping. The pruning does not depend on the value, so it can happen when the processor is built. Which keys each of the three workflows is fed follows from the graph and is likewise known then. Wiring every input up front removes the special case, and with it the last graph rebuild: after construction the processor calls Pipeline.__setitem__ zero times, so the retention now holds from the first chunk rather than from the second. The resulting graphs are identical, node for node and edge for edge, to those the previous version arrives at after a cycle. Reading an input that holds no value now raises, naming the key, rather than returning a released None for a provider downstream to trip over. _Inputs mutated a pipeline that StreamProcessor also used directly, so neither could be reasoned about locally; it now owns its pipeline and exposes compute(). The graph analysis that had accumulated in __init__ moves to module-level functions. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2 participants
StreamProcessorused its internal pipelines as per-chunk data containers: every chunk and every accumulator value was handed over withpipeline[key] = value. That is wrong on two counts. Each assignment rebuilds the underlying graph, although the graph structure never changes between chunks — only the values do. And the assigned value stays reachable from the long-lived pipeline until the next assignment overwrites it, so a payload that is dead the momentcompute()returns is kept alive for a full cycle. Both costs are paid per chunk and per finalize, and the values in question are the largest objects in the process. This OOM-killed a live-data backend service (scipp/esslivedata#1264) and is the subject of scipp/esslivedata#1267 and scipp/esslivedata#1259.Feeding instead of assigning
Each workflow gets a mapping it reads its inputs from: a key is provided by a small provider that returns whatever the mapping currently holds, so feeding a value is a dict write and the graph, whose structure does not depend on the values, is left alone. This is the "keep the graph static, pass values at compute time" direction of scipp/esslivedata#1267, expressed with the API sciline already has:
insert()mutates the node in place, whereas__setitem__goes through a full cyclebane rebuild.The inputs are wired up once, when the processor is constructed. Wiring assigns to the pipeline before inserting the provider, because only assignment prunes the branch the input replaces — the whole dynamic branch is superseded by the accumulator value, taking the finalize graph from 14 nodes to 2 in a small example — whereas
insert()only cuts the key's incoming edges and leaves the orphaned branch forPipeline.get()to walk on every later call, which cost more than the rebuilds saved when I measured it. Which keys each workflow is fed follows from the graph, so it is known at construction and needs no bookkeeping at feed time. The resulting graphs are identical, node for node and edge for edge, to whatmainarrives at after a cycle.After construction the processor rebuilds no graph at all:
Pipeline.__setitem__goes from one call per dynamic key per chunk plus one per accumulator per finalize, to zero. So the retention holds from the very first chunk, and does not depend on scipp/cyclebane#33 — there is no discarded graph left to collect.Reading an input that has not been fed raises, naming the key. A workflow whose context never arrived therefore still fails rather than silently computing from a stand-in value, and now says which key is missing instead of surfacing as a
TypeErrorfrom arithmetic on theNonethe branch was pruned with. There is a test for that.StreamProcessor.__init__did all of the graph analysis inline and had grown hard to follow; the analyses are now module-level functions, leaving__init__as assignments plus the wiring above.Releasing values, and the two places we must not
Feeding only removes the rebuild; the retention goes away because the value is dropped once it has been consumed — chunks after the accumulators have been pushed, accumulator values after the targets have been computed. The latter is scipp/esslivedata#1259: an accumulator that drops its value in
on_finalize(), such as a sliding window, previously had that array pinned by the finalize pipeline until the next cycle overwrote it, costing one extra buffer per view at peak.Two kinds of value are deliberately kept. A dynamic key that finalize reads directly must stay readable across finalize calls that have no preceding
accumulate— esslivedata's wavelength-LUT workflow depends on exactly this, its chopper trigger arriving only when the choppers change. Context, likewise, is a cache by construction: its whole purpose is to be reused until it changes. Both are the pre-existing semantics, and both fall out of the same mechanism, since a fed value stays readable until it is overwritten or released."Read directly at finalize" is a per-key property, not a property of
allow_bypass. The flag is set for the whole processor, but a dynamic key whose only path to a target runs through an accumulator is not read at finalize, and is released with its chunk like any other. The set is determined once, from the graph with the accumulator inputs cut, which is whatfinalizecomputes on; feeding context cuts further edges, so the set may name a key finalize turns out not to need, never miss one it does. A test pins both halves on a bypass-plus-accumulator configuration: the terminated key released, the bypassed key retained and still the value the target is computed from.Context is not the rare path it looks like
set_contextis easy to dismiss as occasional, and an earlier version of this PR did exactly that. It is wrong. Of the 21 esslivedata workflows that pass context keys to aStreamProcessor, the keys fall into three populations:ROIRectangleRequestDetectorARotationLog,DetectorCarriageLogRotationSpeedSetpoint_*The middle row is the problem. Motors and sensors publish about once per second and event data is batched at roughly the same rate, so those keys are fed as often as the chunks are — and one such update assigned twice per context key plus once per cached node derived from it. For
magic/detector_projection/1that is 10 rebuilds at 0.90 ms, about 9 ms per rotation update; forloki/detector_xy_projection/1, 10 at 0.48 ms. The chunk path in those same two workflows costs 2.52 ms and 1.99 ms per cycle, so context was the larger of the two all along, because a device log fans out to eight cached nodes where a dynamic key is one.Retention is a different matter and does not change: context is a cache and is kept, as described above.
Why this lives here and not in sciline
The mechanism is not streaming-specific and would be better placed upstream; scipp/sciline#241 tracks that, with the requirements this PR turned up. Two shapes are worth distinguishing. A
compute(targets, params={...})-style API removes the rebuild and gives values a call-scoped lifetime, which is roughly what this PR already achieves, the gain being that the mechanism lives upstream and the runtime__annotations__patch disappears. A reusable task graph — built once, values supplied per call — would be the larger win, sincePipeline.get()is roughly 80% of acompute()call under the naive scheduler and neither this PR nor a bareparams=touches it.What stays here in either case is the policy: which values are transient and which are cached. That is streaming semantics, not sciline's business, so this code gets thinner rather than disappearing.
The reason to land this first is not urgency — scipp/cyclebane#33 already stops the unbounded growth, and this PR's distinct contributions are the steady-state buffers and the rebuild CPU. It is that the requirements for a good upstream API are not obvious from the armchair. The culling requirement in particular was found by measurement here, after an earlier version of this change that only inserted providers, never assigning, turned out to be slower than the code it replaced.
Why not build the task graph once instead?
A fair question is why we do not call
Pipeline.get()once and then set values on the resultingTaskGraph, which is close to a plain dict. That is worth doing, but it is a follow-up rather than an alternative, and it needs this change first.It does not address the retention on its own:
Graphisdict[Key, Provider], so a value in a prebuilt task graph is aProvider.parameter(value)sitting in a long-lived dict, which is the same container problem one layer down and still needs an explicit release. Injecting one means constructing asciline._provider.Providerand writing it intoTaskGraph._graph, two private layers deep. The task graph is also target-dependent:Pipeline.get()culls to the ancestors of its targets, whereasaccumulatecomputes a varying subset that depends on which dynamic keys arrived in the chunk. Passingtargets=to a prebuilt superset graph is not a substitute, becauseNaiveScheduler.gettopologically sorts and executes every task in the graph it is handed, regardless of which keys were requested — and esslivedata pins the naive scheduler process-wide. Reuse therefore means one cached graph per subset of dynamic keys, invalidated wheneverset_contextbakes new context values into providers.The relevant point for this PR is that feeding is what makes such caching possible. A provider captured in a task graph reads the current value when it is called, so a graph built once stays valid across chunks, and the value is still released when it is dropped. With assignment the value is baked into the graph and every chunk invalidates it. The prize is real — under the naive scheduler
Pipeline.get()accounts for roughly 80% of acompute()call, 0.52 ms out of 0.64 ms for a 30-node chain — but it is a separate change on top of this one.Relation to scipp/cyclebane#33
The two overlap but neither subsumes the other. scipp/cyclebane#33 makes the discarded graphs refcount-reclaimable, which stops the unbounded growth for every cyclebane user. It does not remove the per-chunk rebuild, and it cannot help with the retained value, which sits in the live graph by design. This change removes the growth independently, and is the only one of the two that recovers the steady-state buffers.
Measurements
Minimal cumulative-histogram processor, 16 MB chunks, 20 accumulate+finalize cycles with the cyclic collector disabled:
ceedc525cf257e(scipp/cyclebane#33)The 32 MB difference in the flat rows is one chunk plus one accumulator value.
On the 41 real esslivedata workflows that wrap a
StreamProcessor, the removed work is one graph rebuild per dynamic key per chunk, one per accumulator per finalize, and — for the 21 with context — two per context key plus one per derived node per update. It depends only on graph structure, so it can be measured directly on the real graphs. Chunk and finalize paths below (representative rows; ms per accumulate+finalize cycle), with the context path quantified in the section above.The finalize column has to be taken after the first assignment rather than before it, since that assignment prunes the dynamic branch and every later cycle rebuilds the smaller graph that is left. The
finalize nodescolumn gives that before and after: the pruning is real but partial on real workflows, because targets and context branches downstream of the accumulators survive. Numbers are the minimum over repeated rounds and over two independent runs, this machine being contended.loki/i_of_q/1magic/detector_projection/1loki/detector_xy_projection/1tbl/multiblade_detector_view/1bifrost/unified_detector_view/1dream/mantle_front_layer/1loki/tube_view/1tbl/tbl_detector_timepix3/1odin/odin_detector_xy/1dream/powder_reduction/1dream/monitor_histogram/1estia/reflectometry_reduction/1loki/monitor_histogram/1loki/wavelength_lut/1dummy/total_counts/1Across all 41 the median is 0.61 ms and the maximum 3.54 ms.
loki/i_of_qis the outlier because the rebuild is paid per key, and it has three dynamic keys and three accumulators on a 66-node graph. Feeding the same values costs about 0.1 us, so the row is effectively the saving.Against a full cycle, using cycle times measured end to end on a real monitor workflow (2.3 ms) and a 32x32 detector view (4.5 ms) under the naive scheduler that services default to, the removed work above is roughly 20% of the monitor cycle and 20% of the detector-view cycle. Under dask, where execution dominates, the same absolute saving is 5 to 8%. The naive-scheduler numbers need scipp/sciline#240, without which these workflows cannot run under that scheduler at all.
Tests
Two tests assert that the chunk and the accumulator value are unreachable once consumed, via a weak reference, without invoking the cyclic collector — they fail on
main. They measure the first cycle, which is meaningful now that construction is the only time a graph is built. A third pins the bypass exception, and a fourth pins that a workflow whose context never arrived raises an error naming the missing key rather than computing from a stand-in value. Context retention and the caching of derived nodes were already covered by the existing suite.Test plan: essreduce's suite passes (927 tests). esslivedata's suite was run against this branch as well; it has 5 failures that reproduce on unmodified
mainand stem from unrelated signature drift between essreducemainand the released 26.6.3 it normally resolves against.