28e0667 stopped nodes blocking a worker on a full output, but replaced an
intermittent hang with a quieter one: the pipeline still wedged about 2
runs in 30, now with every worker idle in pthread_cond_wait rather than
asleep in a push. Nothing was blocked; nothing had been woken.
try_submit discarded any wake arriving while queued_ was up:
if (queued_.compare_exchange_strong(expected, true, ...))
scheduler_->submit(...);
// else: silently gone
Wakes are edge-triggered — a channel fires its space callback once, on the
transition — so a dropped one never returns. A node could park a value,
release its worker, and sleep forever holding exactly the output its
consumer was waiting for, while its producer parked on an input channel
that would never drain.
try_submit now records the drop in wake_pending_, and release_and_recheck()
consumes it at every site that releases a node, giving one invariant: a
node never sleeps with a wake outstanding. This subsumes the two ad-hoc
re-checks added for the parked-retry and normal push paths, which only
moved the stall (1211 items to 6472) because each new early return was a
fresh chance to drop a wake. self_stop keeps a plain store — honouring a
pending wake there would resubmit a dead node.
Adds "a saturated chain never stalls", which reproduces this in about a
second where the pipeline needed ~30 runs. It asserts *progress does not
freeze* rather than a completion total: capacity-1 channels are slow, and
slow must never be reported as wedged. Verified in both directions — it
stalls after 1211 items on 28e0667 and passes here.
The existing chain test could not catch it: 40 items drain before any
strand occurs, and its producer emits forever, so fresh input keeps
re-triggering on_input_ready() and flushing the stranded value.
Tests: 122/122.
push_blocking parked a scheduler worker inside the push. Nodes own a
private single-thread pool, so the parked thread was the only one that
could drain that node's own input — hold-and-wait, and under sustained
backpressure four nodes of a five-node chain slept in nanosleep at once.
channel.hpp already warned about this for sentinels; it applies just as
much to data pushes.
The scheduler was purely input-driven: on_input_ready() wakes a node when
input arrives, with no counterpart for "my output has room". Lacking that
signal, blocking the thread was the only way to handle a full output.
This adds the missing half.
- Channel::try_push + has_space + set_space_callback; the callback fires
from both pop() and try_pop_now().
- PoolNode/PoolObjectNode keep a one-slot pending_ buffer with per-element
done flags, so a retry cannot duplicate an already-accepted element. One
slot suffices because queued_ admits at most one fire_once per node.
- The re-check after clearing queued_ closes the lost-wakeup race where a
space callback fires while the flag is still up and is swallowed.
Two bugs surfaced once nodes actually parked, both fixed here:
- pop_one reports an *empty* channel as ChannelClosedError, which is also
the node's "upstream finished, self-stop" signal. A node woken by output
space with empty inputs therefore killed itself. fire_once now releases
the worker when its inputs are not ready rather than falling through.
- The drained-park path resubmitted unconditionally instead of via
on_input_ready(), firing nodes with nothing to read.
compute_priority is now output-aware: mean output fill is deducted from
mean input fill, mapped as 0.5·(1 + in - out). Input fill alone asks only
"how much work is waiting for me"; a node whose outputs are already full
cannot deliver, so running it just parks it again and wastes the slot
while the node that would drain that channel waits behind it. The
scheduler now favours whoever is furthest downstream of a bottleneck.
Also adds a network-level error listener. A node's exception was discarded
at the node boundary and survived only as a Closed event, which reports
that a node stopped but not why — that missing detail is what made the
above slow to diagnose. INode::set_network_error_callback plus
StaticNetwork::set_error_handler forward it to the application.
Tests: 121/121. test_backpressure_deadlock drives a five-node chain with
capacity-2 channels against a slow sink and fails on the old code. The
four test_pool_node overflow tests now assert parking rather than the
removed drop-and-report behaviour.
Known-incomplete: a rare hang remains, roughly 1 run in 20 against a 300s
timeout, down from every run failing. Committed because the fix is a large
strict improvement and the residual case needs its own reproduction.
Channel<T>::pop() surfaced the out-of-band sentinel from its empty branch
using the tail_ snapshot taken at the top of the loop. Under contention the
producer can push more values *and* the sentinel in the window between that
snapshot and take_sentinel(), so pop() could return the sentinel while real
values still sat in the ring — the sentinel jumping ahead of values pushed
before it. No value was lost (a consumer that keeps draining still receives
them, and approx_size() keeps counting them so a PoolNode reschedules), but a
consumer treating the sentinel as a hard "last message" barrier would act on
EOF early.
Re-confirm emptiness against a fresh tail_ load before taking the sentinel.
Costs one acquire-load on the empty-ring path only; never runs in steady
state. The spin and post-spin takes already reload tail_ on the line above
them; try_pop_now() already reads tail_ fresh in the same branch — both were
correct and are unchanged.
The two sentinel stress cases now assert the strict "sentinel is last, after
every value" ordering (previously relaxed to avoid the flake this fixes).
Verified TSan-clean (2606 assertions, no data races) over repeated runs.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add a contended SPSC stress suite (tests/test_channel_stress.cpp) that
actually exercises the ring's memory-ordering pairing and spin/futex/
lost-wakeup logic, plus the CMake and CI plumbing to run it under TSan:
- KPN_SANITIZER cache var + kpn_sanitizer_flags() helper (no-op when unset)
- kpn_tests_stress executable, labelled "stress" for CTest
- reusable tsan.yaml workflow (gcc:14 builder image, already ships libtsan)
- ci.yaml gains a tsan job on the same code/dockerfile triggers as test
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Channel::push() drops values on overflow (the intended backpressure
policy for data), and PoolNode swallows the resulting ChannelOverflowError.
For a control sentinel like EOF this is fatal: a single dropped EOF under
backpressure wedges every downstream pop() forever, so the pipeline never
tears down.
Deliver sentinels out-of-band instead. Channel::push_sentinel() stores the
token in a dedicated slot that does not consume ring capacity, so it can
never overflow and — crucially — never blocks the caller. That non-blocking
property is essential: each KPN node has a single worker thread, so a
*blocking* push would park that thread and stop it draining its own input,
cascading into a hold-and-wait deadlock under backpressure. The consumer's
pop()/try_pop_now() drain the ring first, then deliver the sentinel, so it
always arrives after every value pushed before it.
approx_size() (which node readiness checks call) counts a pending sentinel
as consumable work, so a channel carrying only a sentinel still schedules
its consumer's next fire — without this the token would sit undelivered and
the pipeline would still deadlock at teardown.
PoolNode/PoolObjectNode route values carrying an eof flag (direct .eof or
nested .source.eof) through push_sentinel via a SFINAE-safe is_sentinel_value
trait; all other values keep the existing lossy throwing push. The trait
compiles to false for types without an eof convention, so this is a no-op
for pipelines that don't use one.
Verified end-to-end: scene_analyze now reaches EOF, flushes its output, and
exits cleanly instead of hanging.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
dtourolle
