From 9c5ce5f34a93945e19c357feef6adbd0e9ed8f4c Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Sun, 2 Aug 2026 18:33:17 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20never=20drop=20a=20wake=20=E2=80=94=20a?= =?UTF-8?q?=20node=20must=20not=20sleep=20with=20one=20outstanding?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- include/kpn/pool_node.hpp | 101 +++++++++++++++++++++++++-- tests/test_backpressure_deadlock.cpp | 86 +++++++++++++++++++++++ 2 files changed, 180 insertions(+), 7 deletions(-) diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index 85532da..2d12cbe 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -265,6 +265,8 @@ private: disable_inputs(std::make_index_sequence{}); disable_outputs(std::make_index_sequence{}); stats_.exec_start_us.store(0, std::memory_order_relaxed); + // Plain store, not release_and_recheck(): this node is stopping, and + // honouring a pending wake here would resubmit a dead node. queued_.store(false, std::memory_order_release); stop_flag_.store(true, std::memory_order_relaxed); } @@ -353,10 +355,32 @@ private: : 0.5f), ...); } + /// Submit unless already queued. A wake that arrives while this node is + /// queued or running is *recorded*, never dropped. + /// + /// Wakes are edge-triggered: a channel fires its space callback on the + /// transition, once. If that lands while queued_ is up, the CAS below fails + /// and — before wake_pending_ — the wake was gone. A node could then park a + /// value, release its worker, and sleep forever holding output its consumer + /// was waiting for, with every worker idle in cond_wait and nothing left to + /// re-trigger it. Recording the drop turns the signal level-triggered: the + /// invariant is that a node never sleeps with a wake outstanding, enforced + /// by release_and_recheck() at every point that releases the node. void try_submit(float priority) { bool expected = false; if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) scheduler_->submit([this] { fire_once(); }, priority); + else + wake_pending_.store(true, std::memory_order_release); + } + + /// Clear queued_, then honour any wake that was dropped while it was up. + /// Every path that finishes or parks a firing must release the node through + /// here rather than storing queued_ directly. + void release_and_recheck(float priority = 0.5f) { + queued_.store(false, std::memory_order_release); + if (wake_pending_.exchange(false, std::memory_order_acq_rel)) + try_submit(priority); } // ── Execution ───────────────────────────────────────────────────────────── @@ -379,7 +403,7 @@ private: if constexpr (!std::is_void_v) { if (pending_) { push_outputs(std::move(*pending_), std::make_index_sequence{}); - queued_.store(false, std::memory_order_release); + release_and_recheck(); if (pending_) { // Close the lost-wakeup race: a space_callback that fired // between the failed push and clearing queued_ was @@ -408,7 +432,7 @@ private: if constexpr (!std::is_void_v) { if (pending_) { push_outputs(std::move(*pending_), std::make_index_sequence{}); - queued_.store(false, std::memory_order_release); + release_and_recheck(); if (pending_) { // Close the lost-wakeup race: a space_callback that fired // between the failed push and clearing queued_ was @@ -438,7 +462,7 @@ private: // on_input_ready() resubmits when data actually lands. if constexpr (input_count > 0) { if (count_ready(std::make_index_sequence{}) != input_count) { - queued_.store(false, std::memory_order_release); + release_and_recheck(); on_input_ready(); // data may have arrived while we checked return; } @@ -483,10 +507,27 @@ private: } stats_.exec_start_us.store(0, std::memory_order_relaxed); - queued_.store(false, std::memory_order_release); + release_and_recheck(); if (stop_flag_.load(std::memory_order_relaxed)) return; + // Parked by the push above. Same situation as the retry path at the top + // of fire_once — and the same lost-wakeup race, which that path closes + // and this one did not. A space callback that fired while queued_ was + // still up got swallowed by try_submit's CAS, 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) { + if (pending_) { + if (outputs_have_space(std::make_index_sequence{})) + 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); @@ -574,6 +615,8 @@ private: output_channels_t output_channels_{}; std::atomic stop_flag_{true}; std::atomic queued_{false}; + /// A wake that arrived while queued_ was up. See try_submit. + std::atomic wake_pending_{false}; /// The hidden one-slot output buffer (see push_outputs). Holding the value /// here is what lets a node stop running without dropping it or occupying a @@ -753,6 +796,8 @@ private: disable_inputs(std::make_index_sequence{}); disable_outputs(std::make_index_sequence{}); stats_.exec_start_us.store(0, std::memory_order_relaxed); + // Plain store, not release_and_recheck(): this node is stopping, and + // honouring a pending wake here would resubmit a dead node. queued_.store(false, std::memory_order_release); stop_flag_.store(true, std::memory_order_relaxed); } @@ -834,10 +879,32 @@ private: : 0.5f), ...); } + /// Submit unless already queued. A wake that arrives while this node is + /// queued or running is *recorded*, never dropped. + /// + /// Wakes are edge-triggered: a channel fires its space callback on the + /// transition, once. If that lands while queued_ is up, the CAS below fails + /// and — before wake_pending_ — the wake was gone. A node could then park a + /// value, release its worker, and sleep forever holding output its consumer + /// was waiting for, with every worker idle in cond_wait and nothing left to + /// re-trigger it. Recording the drop turns the signal level-triggered: the + /// invariant is that a node never sleeps with a wake outstanding, enforced + /// by release_and_recheck() at every point that releases the node. void try_submit(float priority) { bool expected = false; if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) scheduler_->submit([this] { fire_once(); }, priority); + else + wake_pending_.store(true, std::memory_order_release); + } + + /// Clear queued_, then honour any wake that was dropped while it was up. + /// Every path that finishes or parks a firing must release the node through + /// here rather than storing queued_ directly. + void release_and_recheck(float priority = 0.5f) { + queued_.store(false, std::memory_order_release); + if (wake_pending_.exchange(false, std::memory_order_acq_rel)) + try_submit(priority); } void fire_once() { @@ -856,7 +923,7 @@ private: if constexpr (!std::is_void_v) { if (pending_) { push_outputs(std::move(*pending_), std::make_index_sequence{}); - queued_.store(false, std::memory_order_release); + release_and_recheck(); if (pending_) { // Close the lost-wakeup race: a space_callback that fired // between the failed push and clearing queued_ was @@ -884,7 +951,7 @@ private: // into pop_inputs on an empty channel. if constexpr (input_count > 0) { if (count_ready(std::make_index_sequence{}) != input_count) { - queued_.store(false, std::memory_order_release); + release_and_recheck(); on_input_ready(); return; } @@ -926,8 +993,26 @@ private: } stats_.exec_start_us.store(0, std::memory_order_relaxed); - queued_.store(false, std::memory_order_release); + release_and_recheck(); if (stop_flag_.load(std::memory_order_relaxed)) return; + + // Parked by the push above. Same situation as the retry path at the top + // of fire_once — and the same lost-wakeup race, which that path closes + // and this one did not. A space callback that fired while queued_ was + // still up got swallowed by try_submit's CAS, 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) { + if (pending_) { + if (outputs_have_space(std::make_index_sequence{})) + try_submit(0.5f); + return; // parked + } + } + if constexpr (input_count == 0) try_submit(0.5f); else on_input_ready(); } @@ -986,6 +1071,8 @@ private: output_channels_t output_channels_{}; std::atomic stop_flag_{true}; std::atomic queued_{false}; + /// A wake that arrived while queued_ was up. See try_submit. + std::atomic wake_pending_{false}; /// The hidden one-slot output buffer (see push_outputs). Holding the value /// here is what lets a node stop running without dropping it or occupying a diff --git a/tests/test_backpressure_deadlock.cpp b/tests/test_backpressure_deadlock.cpp index 4679439..17835ec 100644 --- a/tests/test_backpressure_deadlock.cpp +++ b/tests/test_backpressure_deadlock.cpp @@ -83,3 +83,89 @@ TEST_CASE("a saturated chain keeps draining", "[backpressure][deadlock]") { INFO("items drained: " << got << " of 40"); CHECK(got >= 40); } + +// Regression: a saturated chain must never stall. +// +// push_outputs parks from two places: the retry at the top of fire_once, and +// the ordinary push after the node function returns. Both release the worker, +// so both face the same lost wakeup — a space callback firing while queued_ is +// still up is swallowed by try_submit's CAS. Only the retry path re-checked for +// space afterwards. The normal path fell through to on_input_ready(), which +// resubmits only if inputs are ready — and the firing that just parked had +// consumed its input, so they are not. +// +// The strand is permanent under saturation: the node holds its value, its +// consumer waits for exactly that value, and its producer fills the node's +// input channel and parks too. Nothing moves again. +// +// The test above cannot catch it — 40 items drain before any strand occurs. +// This one runs the chain saturated and watches for progress to *freeze*, which +// is the signature of the deadlock. It deliberately does not assert a total: +// capacity-1 channels are slow, and "slow" must never be reported as "wedged". +namespace { + +struct FreeRun { + static constexpr std::string_view label() { return "free_run"; } + int n{0}; + int operator()() { return n++; } +}; + +struct CountingSink { + static constexpr std::string_view label() { return "counting_sink"; } + std::atomic* seen; + void operator()(int) { seen->fetch_add(1, std::memory_order_relaxed); } +}; + +} // namespace + +TEST_CASE("a saturated chain never stalls", "[backpressure][deadlock]") { + std::atomic seen{0}; + + FreeRun p_fn; + Relay r1_fn, r2_fn; + CountingSink s_fn{&seen}; + + // Capacity 1 everywhere: every push contends, so the park path is taken + // constantly and the race window is sampled millions of times. + kpn::ObjectNode, kpn::out<"a">, "free_run", 0> p (p_fn, 1); + kpn::ObjectNode, kpn::out<"b">, "relay1", 0> r1(r1_fn, 1); + kpn::ObjectNode, kpn::out<"c">, "relay2", 0> r2(r2_fn, 1); + kpn::ObjectNode, kpn::out<>, "sink", 0> s (s_fn, 1); + + auto net = kpn::make_network( + kpn::edge(p.output<"a">(), r1.input<"a">()), + kpn::edge(r1.output<"b">(), r2.input<"b">()), + kpn::edge(r2.output<"c">(), s.input<"c">()) + ); + net.start(); + + // A live chain moves thousands of items a second, so 3 s with no movement + // at all is a wedge, not a slow patch. Sampling for 25 s gives the race + // ample opportunity: the pipeline hit it roughly twice in 30 runs. + const auto giveup = std::chrono::steady_clock::now() + std::chrono::seconds(25); + int last = 0; + auto last_move = std::chrono::steady_clock::now(); + bool stalled = false; + int stall_at = 0; + + while (std::chrono::steady_clock::now() < giveup) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + const int now_seen = seen.load(std::memory_order_relaxed); + if (now_seen != last) { + last = now_seen; + last_move = std::chrono::steady_clock::now(); + } else if (std::chrono::steady_clock::now() - last_move > + std::chrono::seconds(3)) { + stalled = true; + stall_at = now_seen; + break; + } + } + + net.stop(); + + INFO("chain stalled after " << stall_at << " items"); + CHECK_FALSE(stalled); + // Guard against the test passing because nothing ever ran. + CHECK(last > 1000); +}