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

fix: two firings of the same node must not overlap

Browse Source 
fire_once released the submit gate and then kept working:

    release_and_recheck();          // gate is now free
    if (stop_flag_) return;
    if (pending_) { ... }           // still reading node state
    on_input_ready();

The moment the gate is free another worker may enter fire_once for the same
node, so this invocation's reads of pending_ raced with the next one's writes
to pending_done_. ThreadSanitizer caught exactly that, between a firing
submitted by release_and_recheck and one submitted by try_submit.

The race is the visible half. The real damage is to the one-slot park, which
is sound only because "at most one fire_once runs per node at a time" — the
comment on pending_ says so explicitly. With two firings live, one can park a
value into the slot the other is about to overwrite, and the overwritten value
is gone with no drop recorded anywhere. That is silent data loss under
backpressure, from a node that reports itself healthy.

finish_firing() replaces release_and_recheck() at every exit: it evaluates the
follow-up decision — parked and waiting on output space, or drained and
waiting on input — while the claim is still held, and releases the gate as the
last thing the firing does. Nothing touches node state afterwards.

This also collapses three near-identical resubmit tails into one, which is
worth something on its own: the divergence between them is what 5628447 and
9c5ce5f were both picking at, and each fix had to be applied to every copy.

Pre-existing, not introduced by the gate rewrite: the old two-atomic version
cleared queued_ in the same place, with the same code after it.

Verified with -DKPN_SANITIZER=thread. The race is intermittent — roughly one
run in three before the fix — so five consecutive clean runs of the unit suite
plus the contended channel stress suite, all zero. Full suite 132/132.
This commit is contained in:
dtourolle
2026-08-05 13:48:13 +02:00
parent a5c016833d
commit 15e993f6ca
1 changed files with 114 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/kpn/pool_node.hpp
+113 -110
View File
@@ -264,7 +264,7 @@ private:
disable_inputs(std::make_index_sequence<input_count>{}); disable_inputs(std::make_index_sequence<input_count>{});
disable_outputs(std::make_index_sequence<output_count>{}); disable_outputs(std::make_index_sequence<output_count>{});
stats_.exec_start_us.store(0, std::memory_order_relaxed); stats_.exec_start_us.store(0, std::memory_order_relaxed);
// Plain store, not release_and_recheck(): this node is stopping, and // force_idle, not finish_firing(): this node is stopping, and
// honouring a pending wake here would resubmit a dead node. // honouring a pending wake here would resubmit a dead node.
gate_.force_idle(); gate_.force_idle();
stop_flag_.store(true, std::memory_order_relaxed); stop_flag_.store(true, std::memory_order_relaxed);
@@ -368,17 +368,49 @@ private:
scheduler_->submit([this] { fire_once(); }, priority); scheduler_->submit([this] { fire_once(); }, priority);
} }
/// End this firing, honouring any wake recorded during it. Every path that // ── Execution ─────────────────────────────────────────────────────────────
/// finishes or parks a firing must release the node through here rather
/// than touching the gate directly. When a wake was recorded the gate stays /// Decide whether this node should run again, then release the gate — in
/// claimed and is handed to the next firing, so the node is never /// that order, always.
/// momentarily idle with work outstanding. ///
void release_and_recheck(float priority = 0.5f) { /// Releasing first is what let two firings of the same node overlap: the
if (gate_.release()) /// moment the gate is free another worker may enter fire_once, while this
scheduler_->submit([this] { fire_once(); }, priority); /// invocation is still reading pending_ and writing pending_done_. TSan
/// caught it as a race on pending_done_ between a firing submitted by the
/// old release_and_recheck and one submitted by try_submit. It also quietly
/// broke the one-slot park, which is sound only because "at most one
/// fire_once runs per node at a time" — with two, a value can be parked by
/// one firing and overwritten by the other.
///
/// Everything this reads belongs to the firing that holds the claim, so it
/// is all evaluated first and the release is the last thing the firing does.
void finish_firing() {
bool want_more = false;
float prio = 0.5f;
if (!stop_flag_.load(std::memory_order_relaxed)) {
bool parked = false;
if constexpr (!std::is_void_v<return_raw>)
parked = pending_.has_value();
if (parked) {
// Still holding output: only worth running again once the
// consumer has made room.
want_more = outputs_have_space(std::make_index_sequence<output_count>{});
} else {
if constexpr (input_count == 0) {
want_more = true; // sources always run again
} else {
want_more = count_ready(std::make_index_sequence<input_count>{})
== input_count;
if (want_more) prio = compute_priority();
}
}
} }
// ── Execution ───────────────────────────────────────────────────────────── if (gate_.release()) scheduler_->submit([this] { fire_once(); }, prio);
else if (want_more) try_submit(prio);
}
void fire_once() { void fire_once() {
if (stop_flag_.load(std::memory_order_relaxed)) { if (stop_flag_.load(std::memory_order_relaxed)) {
@@ -398,25 +430,13 @@ private:
if constexpr (!std::is_void_v<return_raw>) { if constexpr (!std::is_void_v<return_raw>) {
if (pending_) { if (pending_) {
push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{}); push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
release_and_recheck(); // Whether the value went out or is still parked, finish_firing
if (pending_) { // reads pending_ and picks the right follow-up: output space if
// Close the lost-wakeup race: a space_callback that fired // still holding, input readiness if drained. Resubmitting
// between the failed push and releasing the gate was // unconditionally would fire a node whose inputs are empty, and
// swallowed, and nothing else will wake this node. Re-check // pop_one reports an empty channel as ChannelClosedError — which
// now that the flag is down. // this node treats as "upstream finished" and self-stops on.
if (outputs_have_space(std::make_index_sequence<output_count>{})) finish_firing();
try_submit(0.5f);
return; // parked
}
// Drained: resume normal firing, resubmitting exactly the way
// the normal tail below does. An unconditional try_submit here
// would fire a node whose inputs are empty, and pop_inputs
// reports an empty channel as ChannelClosedError — which this
// node treats as "upstream finished" and self-stops on. That
// is a live node killing itself purely because it was woken by
// *output* space rather than by input arrival.
if constexpr (input_count == 0) try_submit(0.5f);
else on_input_ready();
return; return;
} }
} }
@@ -428,8 +448,9 @@ private:
// on_input_ready() resubmits when data actually lands. // on_input_ready() resubmits when data actually lands.
if constexpr (input_count > 0) { if constexpr (input_count > 0) {
if (count_ready(std::make_index_sequence<input_count>{}) != input_count) { if (count_ready(std::make_index_sequence<input_count>{}) != input_count) {
release_and_recheck(); // finish_firing re-checks readiness after the work above, so
on_input_ready(); // data may have arrived while we checked // data that landed while we looked is not missed.
finish_firing();
return; return;
} }
} }
@@ -473,33 +494,11 @@ private:
} }
stats_.exec_start_us.store(0, std::memory_order_relaxed); stats_.exec_start_us.store(0, std::memory_order_relaxed);
release_and_recheck(); // If the push above parked, finish_firing waits on output space rather
// than input arrival: this firing consumed its input, so an input-level
if (stop_flag_.load(std::memory_order_relaxed)) return; // check would not resubmit and the node would hold its value forever
// while its consumer waits for exactly that value.
// Parked by the push above. Same situation as the retry path at the top finish_firing();
// of fire_once — and the same lost-wakeup race, which that path closes
// and this one did not. A space callback that fired while the gate was
// still claimed is recorded there, and the resubmit below
// cannot cover it: this firing consumed its input, so inputs are empty
// and on_input_ready() will not resubmit. The node would then hold its
// value forever while its consumer waits for exactly that value and its
// producer parks on an input channel that never drains. Re-check now
// that the flag is down.
if constexpr (!std::is_void_v<return_raw>) {
if (pending_) {
if (outputs_have_space(std::make_index_sequence<output_count>{}))
try_submit(0.5f);
return; // parked
}
}
// Source nodes always resubmit; others resubmit only if inputs are ready.
if constexpr (input_count == 0) {
try_submit(0.5f);
} else {
on_input_ready();
}
} }
// Pop all inputs — safe because we're the sole consumer and fire_once // Pop all inputs — safe because we're the sole consumer and fire_once
@@ -787,7 +786,7 @@ private:
disable_inputs(std::make_index_sequence<input_count>{}); disable_inputs(std::make_index_sequence<input_count>{});
disable_outputs(std::make_index_sequence<output_count>{}); disable_outputs(std::make_index_sequence<output_count>{});
stats_.exec_start_us.store(0, std::memory_order_relaxed); stats_.exec_start_us.store(0, std::memory_order_relaxed);
// Plain store, not release_and_recheck(): this node is stopping, and // force_idle, not finish_firing(): this node is stopping, and
// honouring a pending wake here would resubmit a dead node. // honouring a pending wake here would resubmit a dead node.
gate_.force_idle(); gate_.force_idle();
stop_flag_.store(true, std::memory_order_relaxed); stop_flag_.store(true, std::memory_order_relaxed);
@@ -884,14 +883,46 @@ private:
scheduler_->submit([this] { fire_once(); }, priority); scheduler_->submit([this] { fire_once(); }, priority);
} }
/// End this firing, honouring any wake recorded during it. Every path that /// Decide whether this node should run again, then release the gate — in
/// finishes or parks a firing must release the node through here rather /// that order, always.
/// than touching the gate directly. When a wake was recorded the gate stays ///
/// claimed and is handed to the next firing, so the node is never /// Releasing first is what let two firings of the same node overlap: the
/// momentarily idle with work outstanding. /// moment the gate is free another worker may enter fire_once, while this
void release_and_recheck(float priority = 0.5f) { /// invocation is still reading pending_ and writing pending_done_. TSan
if (gate_.release()) /// caught it as a race on pending_done_ between a firing submitted by the
scheduler_->submit([this] { fire_once(); }, priority); /// old release_and_recheck and one submitted by try_submit. It also quietly
/// broke the one-slot park, which is sound only because "at most one
/// fire_once runs per node at a time" — with two, a value can be parked by
/// one firing and overwritten by the other.
///
/// Everything this reads belongs to the firing that holds the claim, so it
/// is all evaluated first and the release is the last thing the firing does.
void finish_firing() {
bool want_more = false;
float prio = 0.5f;
if (!stop_flag_.load(std::memory_order_relaxed)) {
bool parked = false;
if constexpr (!std::is_void_v<return_raw>)
parked = pending_.has_value();
if (parked) {
// Still holding output: only worth running again once the
// consumer has made room.
want_more = outputs_have_space(std::make_index_sequence<output_count>{});
} else {
if constexpr (input_count == 0) {
want_more = true; // sources always run again
} else {
want_more = count_ready(std::make_index_sequence<input_count>{})
== input_count;
if (want_more) prio = compute_priority();
}
}
}
if (gate_.release()) scheduler_->submit([this] { fire_once(); }, prio);
else if (want_more) try_submit(prio);
} }
void fire_once() { void fire_once() {
@@ -910,25 +941,13 @@ private:
if constexpr (!std::is_void_v<return_raw>) { if constexpr (!std::is_void_v<return_raw>) {
if (pending_) { if (pending_) {
push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{}); push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
release_and_recheck(); // Whether the value went out or is still parked, finish_firing
if (pending_) { // reads pending_ and picks the right follow-up: output space if
// Close the lost-wakeup race: a space_callback that fired // still holding, input readiness if drained. Resubmitting
// between the failed push and releasing the gate was // unconditionally would fire a node whose inputs are empty, and
// swallowed, and nothing else will wake this node. Re-check // pop_one reports an empty channel as ChannelClosedError — which
// now that the flag is down. // this node treats as "upstream finished" and self-stops on.
if (outputs_have_space(std::make_index_sequence<output_count>{})) finish_firing();
try_submit(0.5f);
return; // parked
}
// Drained: resume normal firing, resubmitting exactly the way
// the normal tail below does. An unconditional try_submit here
// would fire a node whose inputs are empty, and pop_inputs
// reports an empty channel as ChannelClosedError — which this
// node treats as "upstream finished" and self-stops on. That
// is a live node killing itself purely because it was woken by
// *output* space rather than by input arrival.
if constexpr (input_count == 0) try_submit(0.5f);
else on_input_ready();
return; return;
} }
} }
@@ -938,8 +957,9 @@ private:
// into pop_inputs on an empty channel. // into pop_inputs on an empty channel.
if constexpr (input_count > 0) { if constexpr (input_count > 0) {
if (count_ready(std::make_index_sequence<input_count>{}) != input_count) { if (count_ready(std::make_index_sequence<input_count>{}) != input_count) {
release_and_recheck(); // finish_firing re-checks readiness after the work above, so
on_input_ready(); // data that landed while we looked is not missed.
finish_firing();
return; return;
} }
} }
@@ -980,28 +1000,11 @@ private:
} }
stats_.exec_start_us.store(0, std::memory_order_relaxed); stats_.exec_start_us.store(0, std::memory_order_relaxed);
release_and_recheck(); // If the push above parked, finish_firing waits on output space rather
if (stop_flag_.load(std::memory_order_relaxed)) return; // than input arrival: this firing consumed its input, so an input-level
// check would not resubmit and the node would hold its value forever
// Parked by the push above. Same situation as the retry path at the top // while its consumer waits for exactly that value.
// of fire_once — and the same lost-wakeup race, which that path closes finish_firing();
// and this one did not. A space callback that fired while the gate was
// still claimed is recorded there, and the resubmit below
// cannot cover it: this firing consumed its input, so inputs are empty
// and on_input_ready() will not resubmit. The node would then hold its
// value forever while its consumer waits for exactly that value and its
// producer parks on an input channel that never drains. Re-check now
// that the flag is down.
if constexpr (!std::is_void_v<return_raw>) {
if (pending_) {
if (outputs_have_space(std::make_index_sequence<output_count>{}))
try_submit(0.5f);
return; // parked
}
}
if constexpr (input_count == 0) try_submit(0.5f);
else on_input_ready();
} }
template<std::size_t... Is> template<std::size_t... Is>