Or: How to create back-pressure?
First of all, I realise that resource limiting (especially for memory) is in the phlex road map,
but since I ran into this today I wanted to open a discussion about changes in phlex or on the user-side
to deal with this, as I could not find any open issues/discussions about this topic.
Current state from what I can gather from the code/experimentation: phlex places
no limit on the number of events in flight. The driver source
(tbb::flow::input_node) produces a new event whenever TBB activates it, and
every downstream algorithm node is a tbb::flow::function_node (or
multifunction_node) constructed with TBB's default queueing policy, which is an
unbounded internal buffer. When any node is registered with a concurrency
limit lower than the effective production rate allows, messages accumulate in
that node's queue without bound. If I understand things correctly,
each queued message holds a std::shared_ptr to the event's full product store,
so the pinned memory is the entire event payload, and RSS grows linearly with the
number of events processed until the run ends.
For a concrete example, in SHiP's aegir simulation (~900k objects ≈ 9 MB/event),
an RNTuple writer registered with concurrency{4} on a 56-core host (-j defaulting
to 56) accumulated a backlog growing at ~9 MB/event: 100 events
peaked at ~5 GB RSS, 500 events at ~8.7 GB, with no bound in sight. A control
run with writing disabled stayed flat, and the writer's own buffers are
capped at ~0.5 GB, so the growth is entirely queued messages ahead of the
concurrency-limited node. Production batch jobs (~3,200 events) were evicted
for memory (ShipSoft/aegir#77).
As a workaround, I now register the writer with concurrency::unlimited
and let excess calls block inside the node on an internal resource pool.
The blocked calls occupy TBB worker threads, which starves the source and
throttles production. This bounds in-flight events at roughly the thread count.
This works but couples backpressure to thread blocking, wastes the blocked threads,
and every module with a bounded internal resource has to reimplement it.
A framework-level bound on in-flight events could solve this. Maybe this could
be implemented using the TBB pattern of a limiter_node (or token bank) between
the driver source and the graph, with the decrement edge fed by end-of-event flush.
A configurable max-events-in-flight (defaulting to a small multiple of -j) would
make memory predictable for any workload where a sink is slower than the
producers, without requiring modules to block threads.
Happy to try creating a PR with a concrete implementation for further discussion.
Or: How to create back-pressure?
First of all, I realise that resource limiting (especially for memory) is in the phlex road map,
but since I ran into this today I wanted to open a discussion about changes in phlex or on the user-side
to deal with this, as I could not find any open issues/discussions about this topic.
Current state from what I can gather from the code/experimentation: phlex places
no limit on the number of events in flight. The driver source
(
tbb::flow::input_node) produces a new event whenever TBB activates it, andevery downstream algorithm node is a
tbb::flow::function_node(ormultifunction_node) constructed with TBB's default queueing policy, which is anunbounded internal buffer. When any node is registered with a concurrency
limit lower than the effective production rate allows, messages accumulate in
that node's queue without bound. If I understand things correctly,
each queued message holds a
std::shared_ptrto the event's full product store,so the pinned memory is the entire event payload, and RSS grows linearly with the
number of events processed until the run ends.
For a concrete example, in SHiP's aegir simulation (~900k objects ≈ 9 MB/event),
an RNTuple writer registered with
concurrency{4}on a 56-core host (-jdefaultingto 56) accumulated a backlog growing at ~9 MB/event: 100 events
peaked at ~5 GB RSS, 500 events at ~8.7 GB, with no bound in sight. A control
run with writing disabled stayed flat, and the writer's own buffers are
capped at ~0.5 GB, so the growth is entirely queued messages ahead of the
concurrency-limited node. Production batch jobs (~3,200 events) were evicted
for memory (ShipSoft/aegir#77).
As a workaround, I now register the writer with
concurrency::unlimitedand let excess calls block inside the node on an internal resource pool.
The blocked calls occupy TBB worker threads, which starves the source and
throttles production. This bounds in-flight events at roughly the thread count.
This works but couples backpressure to thread blocking, wastes the blocked threads,
and every module with a bounded internal resource has to reimplement it.
A framework-level bound on in-flight events could solve this. Maybe this could
be implemented using the TBB pattern of a
limiter_node(or token bank) betweenthe driver source and the graph, with the decrement edge fed by end-of-event flush.
A configurable
max-events-in-flight(defaulting to a small multiple of-j) wouldmake memory predictable for any workload where a sink is slower than the
producers, without requiring modules to block threads.
Happy to try creating a PR with a concrete implementation for further discussion.