dtourolle/KPN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'fix/channel-push-callback-lost-wake'

Browse Source 
A push must wake its consumer even when the ring looked non-empty. The
edge-triggered push_callback_ was computed from a head_ sampled before
the item was published, so a concurrent pop could leave a PoolNode idle
holding work — permanently, since the edge never fired again.

Validated: 136/136 tests pass; the retargeted stress test fails at
1325/10000 against the old code. ~85k reproducer iterations and 10
consecutive full benchmark passes with no wedge, against 5/5 wedges
before the fix.
This commit is contained in:
dtourolle
2026-08-06 21:57:15 +02:00
parent 27f884496d 6802328e97
commit 433c3b3859
2 changed files with 55 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/kpn/channel.hpp
+41 -7
View File
@@ -136,7 +136,6 @@ public:
throw ChannelOverflowError(capacity_);
}
const bool was_empty = (t == h);
buf_[t & ring_mask_] = make_storage(std::move(value));
tail_.store(t + 1, std::memory_order_release);
stats_.record_push(t - h + 1, data_bytes);
@@ -144,7 +143,8 @@ public:
wake_.fetch_add(1, std::memory_order_release);
wake_.notify_one();
if (was_empty && push_callback_)
// Level-triggered, not edge-triggered — see set_push_callback.
if (push_callback_)
push_callback_();
}
@@ -187,13 +187,13 @@ public:
if (t - h >= capacity_) return PushResult::Full;
const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
const bool was_empty = (t == h);
buf_[t & ring_mask_] = make_storage(std::move(value));
tail_.store(t + 1, std::memory_order_release);
stats_.record_push(t - h + 1, data_bytes);
wake_.fetch_add(1, std::memory_order_release);
wake_.notify_one();
if (was_empty && push_callback_) push_callback_();
// Level-triggered, not edge-triggered — see set_push_callback.
if (push_callback_) push_callback_();
return PushResult::Taken;
}
@@ -212,13 +212,13 @@ public:
const std::size_t h = head_.load(std::memory_order_acquire);
if (t - h < capacity_) { // space available → normal push
const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
const bool was_empty = (t == h);
buf_[t & ring_mask_] = make_storage(std::move(value));
tail_.store(t + 1, std::memory_order_release);
stats_.record_push(t - h + 1, data_bytes);
wake_.fetch_add(1, std::memory_order_release);
wake_.notify_one();
if (was_empty && push_callback_) push_callback_();
// Level-triggered, not edge-triggered — see set_push_callback.
if (push_callback_) push_callback_();
return true;
}
// full: yield briefly and retry (consumer will drain)
@@ -388,7 +388,41 @@ public:
wake_.notify_all();
}
// Register a callback fired when the queue transitions empty→non-empty.
// Register a callback fired after every successful push.
//
// It fires on every push, not on the empty→non-empty transition, and that
// is a correctness requirement rather than a simplification.
//
// The edge version tested `was_empty = (t == h)` using an `h` sampled
// *before* the item was published. A PoolNode consumer decides whether to
// run again from the level (count_ready → approx_size), so the two sides
// could each read the other as stale and both stand 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.
// Worse, the failure is absorbing: every later push now sees a non-empty
// ring, so `was_empty` is false forever and the callback never fires again.
// The node sleeps while its backlog grows and its consumer waits on it.
//
// Re-reading head_ after the tail_ store does not fix it. 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 during a firing into the
// firing already in flight, so the cost is one CAS, not one extra run.
void set_push_callback(std::function<void()> cb) {
push_callback_ = std::move(cb);
}
tests/test_channel_stress.cpp
+14 -8
View File
@@ -153,11 +153,18 @@ TEST_CASE("SPSC: producer racing a disable() never throws and never hangs",
}
}
TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
TEST_CASE("SPSC: push_callback fires for every push, never missed",
"[channel][stress]") {
// The empty->non-empty callback ([channel.hpp] was_empty branch) is read by
// the consumer-side notification path. Run it under contention to make sure
// the was_empty detection isn't torn by a concurrent pop().
// Regression: this callback is the *only* thing that wakes a PoolNode, and
// it used to fire only on the empty->non-empty edge, computed from a head_
// sampled before the item was published. A concurrent pop() could drain the
// ring to empty in that window, so neither side saw the other: the item sat
// in the ring with the consumer idle, and because the trigger was an edge it
// never recovered. See set_push_callback in channel.hpp.
//
// The old version of this test asserted only `1 <= callbacks <= N`, which a
// *missed* callback satisfies — it named the hazard and could not detect it.
// One callback per successful push is the contract, so assert exactly that.
Channel<int> ch(/*capacity=*/4, /*spin_count=*/4);
std::atomic<int> callbacks{0};
ch.set_push_callback([&] { callbacks.fetch_add(1, std::memory_order_relaxed); });
@@ -175,10 +182,9 @@ TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
for (int i = 0; i < N; ++i) (void)ch.pop();
producer.join();
// At least one transition, at most one per item; mainly we assert the run
// completed without TSan flagging a race on push_callback_/was_empty.
REQUIRE(callbacks.load() >= 1);
REQUIRE(callbacks.load() <= N);
// Exactly one callback per successful push. Fewer means a wake was dropped,
// which is the bug; more would mean a spurious wake was manufactured.
REQUIRE(callbacks.load() == N);
}
// Ordering contract of the out-of-band sentinel under contention.