push(), try_push() and push_blocking() fired push_callback_ only on the
empty->non-empty edge, and computed that edge from a head_ sampled before
the item was published. A PoolNode consumer decides whether to run again
from the level (count_ready -> approx_size), so a pop landing in that
window left both sides standing down:
producer (push) consumer (PoolNode firing)
------------------------ ----------------------------
samples t=782, h=781
-> was_empty = false, no wake
pops idx 781, head_ = 782
count_ready(): head_==tail_==782
-> not ready, gate released to Idle
tail_.store(783)
The item is in the ring, the node is idle, and no wake is outstanding.
The failure is absorbing: every later push then sees a non-empty ring, so
the edge never fires again and the node sleeps while its backlog grows.
Observed as a hang in bench_pipeline at (chain, depth=4, work_us=10,
shared pool): all pool workers asleep in worker_loop, the reader blocked
in pop(), and 218 items stranded in one channel with head_ stopped at
exactly the index where the edge was dropped.
Re-reading head_ after the tail_ store does not fix this. That is the
store-buffer pattern, and under acquire/release both sides may legally
read stale; forbidding it needs seq_cst on the producer's tail_ store and
head_ load *and* on the consumer's head_ store and tail_ load, a fence on
both hot paths. Firing unconditionally is correct by construction: the
callback runs after the publishing store, so a consumer that observes the
level at all observes the item. The redundant wakes are cheap --
on_input_ready re-checks the level and SubmitGate::claim() collapses a
wake arriving mid-firing into the firing already in flight.
The stress test named this exact hazard and could not detect it: it
asserted only 1 <= callbacks <= N, which a *missed* callback satisfies.
It now requires one callback per successful push, and fails at 1325/10000
against the old code.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
push_sentinel wrote eof_value_ unconditionally. Offering a second token
before the first was taken did two wrong things at once.
It lost the first silently, and a lost EOF is not a lost frame — it is the
token every downstream node is waiting for in order to shut down, so losing
it wedges the pipeline.
And it wrote the storage while the consumer could be moving the previous
value out of it. I expected that to be a stale read; ThreadSanitizer shows
it is worse. On the shared_ptr storage that non-trivial types use, the
racing write tears the refcount, and the stress case added here reports
heap-use-after-free in extract() alongside the data race.
try_push_sentinel now refuses when the slot is occupied, which turns the
slot into a correct SPSC handshake: the producer is the only writer of
eof_value_ and the only one that sets has_eof_, the consumer is the only one
that clears it, so observing it false is what licenses the write. Refusal is
recorded as a drop, and PoolNode reports it through the overflow event
callback, because a refused control token going unnoticed is the failure
this commit exists to stop.
Refusing rather than queueing is deliberate. Two control tokens on one
channel means the stream ended twice, which is a caller protocol error and
not backpressure; parking and retrying would spin against a slot only the
consumer can free, and there is no sensible second value to deliver after
the end of a stream. The non-consuming try_push_sentinel exists so a refused
token is still the caller's to report — the consuming push_sentinel cannot
offer that, since the value has already been moved into its parameter.
Single-shot EOF is what every current caller does, so this is latent for
them today. It stops being latent the moment a pipeline is reused for a
second input, which is what the persistent-pipeline work in 4b6e498 sets up.
Verified in both directions under -DKPN_SANITIZER=thread: the new contended
case reports three data races and a heap-use-after-free against the old
overwrite, and is clean with the handshake. Full suite 137/137, TSan clean
across unit and stress suites.
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>
dtourolleandClaude Opus 5