diff --git a/.gitignore b/.gitignore index 483232f..d3fb3e0 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,5 @@ Thumbs.db # Claude Code local settings .claude/settings.local.json include/kpn/ort_cache/ +build-tsan/ +build-*/ diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index 7c9f73c..bff47b3 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -5,6 +5,7 @@ #include "inode.hpp" #include "port.hpp" #include "scheduler.hpp" +#include "submit_gate.hpp" #include "traits.hpp" #include @@ -31,7 +32,7 @@ namespace kpn { // Reactive alternative to Node<>. Instead of owning a blocked thread, the node // is submitted to a shared IScheduler whenever all its input channels become // non-empty. A single fire_once() call pops all inputs, executes the function, -// and pushes outputs. At most one fire_once() runs at a time (queued_ flag). +// and pushes outputs. At most one fire_once() runs at a time (see SubmitGate). // // Source nodes (input_count == 0) submit themselves immediately on start() and // resubmit after each fire_once(). @@ -83,7 +84,7 @@ public: void start() override { enable_inputs(std::make_index_sequence{}); stop_flag_.store(false, std::memory_order_relaxed); - queued_.store(false, std::memory_order_relaxed); + gate_.force_idle(); register_callbacks(std::make_index_sequence{}); if constexpr (input_count == 0) try_submit(0.5f); @@ -145,8 +146,8 @@ public: total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0, qwait_ms, stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0, - queued_.load(std::memory_order_relaxed), - wake_pending_.load(std::memory_order_relaxed), + gate_.queued(), + gate_.wake_pending(), }; } @@ -258,7 +259,7 @@ private: 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); + gate_.force_idle(); stop_flag_.store(true, std::memory_order_relaxed); } @@ -346,39 +347,35 @@ private: : 0.5f), ...); } - /// Submit unless already queued. A wake that arrives while this node is - /// queued or running is *recorded*, never dropped. + /// Submit unless a firing is already in flight. A wake that arrives while + /// one is is *recorded* against it, 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 + /// transition, once. A dropped one never returns, so a node could 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. + /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same + /// variable, so the two cannot both be true — see submit_gate.hpp. void try_submit(float priority) { - bool expected = false; - if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + if (gate_.claim()) 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. + /// End this firing, honouring any wake recorded during it. Every path that + /// 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 + /// claimed and is handed to the next firing, so the node is never + /// momentarily idle with work outstanding. 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); + if (gate_.release()) + scheduler_->submit([this] { fire_once(); }, priority); } // ── Execution ───────────────────────────────────────────────────────────── void fire_once() { if (stop_flag_.load(std::memory_order_relaxed)) { - queued_.store(false, std::memory_order_release); + gate_.force_idle(); return; } @@ -397,7 +394,7 @@ private: release_and_recheck(); if (pending_) { // Close the lost-wakeup race: a space_callback that fired - // between the failed push and clearing queued_ was + // between the failed push and releasing the gate was // swallowed, and nothing else will wake this node. Re-check // now that the flag is down. if (outputs_have_space(std::make_index_sequence{})) @@ -475,8 +472,8 @@ private: // 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 + // 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 @@ -499,7 +496,7 @@ private: } // Pop all inputs — safe because we're the sole consumer and fire_once - // is guarded by queued_ (only one fire_once runs at a time). + // is guarded by the submit gate (only one fire_once runs at a time). template args_tuple pop_inputs(std::index_sequence) { return {pop_one()...}; @@ -584,9 +581,9 @@ private: input_channels_t input_channels_; 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}; + /// Serialises firings and records wakes that arrive during one. See + /// submit_gate.hpp for why this cannot be two separate flags. + SubmitGate gate_; /// 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 @@ -652,7 +649,7 @@ public: void start() override { enable_inputs(std::make_index_sequence{}); stop_flag_.store(false, std::memory_order_relaxed); - queued_.store(false, std::memory_order_relaxed); + gate_.force_idle(); register_callbacks(std::make_index_sequence{}); if constexpr (input_count == 0) try_submit(0.5f); @@ -694,8 +691,8 @@ public: total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0, qwait_ms, stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0, - queued_.load(std::memory_order_relaxed), - wake_pending_.load(std::memory_order_relaxed), + gate_.queued(), + gate_.wake_pending(), }; } @@ -774,7 +771,7 @@ private: 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); + gate_.force_idle(); stop_flag_.store(true, std::memory_order_relaxed); } @@ -855,37 +852,33 @@ private: : 0.5f), ...); } - /// Submit unless already queued. A wake that arrives while this node is - /// queued or running is *recorded*, never dropped. + /// Submit unless a firing is already in flight. A wake that arrives while + /// one is is *recorded* against it, 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 + /// transition, once. A dropped one never returns, so a node could 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. + /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same + /// variable, so the two cannot both be true — see submit_gate.hpp. void try_submit(float priority) { - bool expected = false; - if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + if (gate_.claim()) 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. + /// End this firing, honouring any wake recorded during it. Every path that + /// 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 + /// claimed and is handed to the next firing, so the node is never + /// momentarily idle with work outstanding. 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); + if (gate_.release()) + scheduler_->submit([this] { fire_once(); }, priority); } void fire_once() { if (stop_flag_.load(std::memory_order_relaxed)) { - queued_.store(false, std::memory_order_release); + gate_.force_idle(); return; } auto t0 = clock_t::now(); @@ -902,7 +895,7 @@ private: release_and_recheck(); if (pending_) { // Close the lost-wakeup race: a space_callback that fired - // between the failed push and clearing queued_ was + // between the failed push and releasing the gate was // swallowed, and nothing else will wake this node. Re-check // now that the flag is down. if (outputs_have_space(std::make_index_sequence{})) @@ -974,8 +967,8 @@ private: // 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 + // 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 @@ -1054,9 +1047,9 @@ private: input_channels_t input_channels_; 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}; + /// Serialises firings and records wakes that arrive during one. See + /// submit_gate.hpp for why this cannot be two separate flags. + SubmitGate gate_; /// 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/include/kpn/submit_gate.hpp b/include/kpn/submit_gate.hpp new file mode 100644 index 0000000..85cbce8 --- /dev/null +++ b/include/kpn/submit_gate.hpp @@ -0,0 +1,101 @@ +#pragma once +#include + +namespace kpn { + +// ── SubmitGate ──────────────────────────────────────────────────────────────── +// +// Decides, for one node, whether a wake must turn into a scheduler submission. +// Exactly one firing of a node may be in flight at a time, and a wake that +// arrives while one is already in flight must not be lost — it has to be +// honoured when that firing finishes, or the node sleeps holding work. +// +// 9c5ce5f wrote this as two independent atomics: queued_ said a firing was in +// flight, wake_pending_ recorded a wake that arrived during one. That cannot be +// made correct, because the release side has to read and write both, and a wake +// can land between the two operations: +// +// producer (try_submit) worker (release_and_recheck) +// ------------------------ ---------------------------- +// CAS reads queued_ == true, fails +// queued_.store(false) +// wake_pending_.exchange(false) -> false +// wake_pending_.store(true) +// +// End state: queued_ false, wake_pending_ true, nothing running and nothing +// scheduled. The node sleeps with a wake outstanding, which is precisely the +// invariant that commit set out to establish. It is not a memory-ordering +// subtlety — the interleaving above holds under seq_cst. +// +// It survived because every caller happened to follow release_and_recheck() +// with a level re-check (on_input_ready(), or outputs_have_space() on the +// parked path), which rediscovers the state a lost wake would have signalled. +// That is a property of the call sites, not of the mechanism, and any new early +// return that forgets the re-check turns it back into a hang. +// +// One atomic with three states makes the race unrepresentable: "idle" and "wake +// outstanding" are the same variable, so no interleaving can produce both. +// +// Idle nothing in flight +// Queued a firing is in flight or queued; no wake since it was claimed +// QueuedWake a firing is in flight or queued, and a wake arrived meanwhile +// +class SubmitGate { +public: + /// Register a wake. Returns true when the caller must submit the node; + /// false when a firing is already in flight and the wake has been recorded + /// against it instead. + bool claim() noexcept { + int cur = state_.load(std::memory_order_acquire); + for (;;) { + if (cur == kIdle) { + if (state_.compare_exchange_weak(cur, kQueued, + std::memory_order_acq_rel, std::memory_order_acquire)) + return true; + } else if (cur == kQueued) { + if (state_.compare_exchange_weak(cur, kQueuedWake, + std::memory_order_acq_rel, std::memory_order_acquire)) + return false; + } else { + return false; // a wake is already recorded + } + } + } + + /// End the in-flight firing. Returns true when a wake arrived during it and + /// the caller must submit again — in which case the gate stays claimed, so + /// the node is handed straight from one firing to the next and is never + /// momentarily idle with work outstanding. Returns false when the node is + /// now idle. + bool release() noexcept { + int cur = state_.load(std::memory_order_acquire); + for (;;) { + if (cur == kQueuedWake) { + if (state_.compare_exchange_weak(cur, kQueued, + std::memory_order_acq_rel, std::memory_order_acquire)) + return true; + } else { + // kQueued, or kIdle if a stop already forced the gate down. + if (state_.compare_exchange_weak(cur, kIdle, + std::memory_order_acq_rel, std::memory_order_acquire)) + return false; + } + } + } + + /// Drop the claim and any recorded wake. For stop paths only: honouring a + /// wake there would resubmit a dead node. + void force_idle() noexcept { state_.store(kIdle, std::memory_order_release); } + + bool queued() const noexcept { return state_.load(std::memory_order_relaxed) != kIdle; } + bool wake_pending() const noexcept { return state_.load(std::memory_order_relaxed) == kQueuedWake; } + +private: + static constexpr int kIdle = 0; + static constexpr int kQueued = 1; + static constexpr int kQueuedWake = 2; + + std::atomic state_{kIdle}; +}; + +} // namespace kpn diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 9e43788..f567afc 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -36,6 +36,7 @@ add_executable(kpn_tests test_pool_node.cpp test_backpressure_deadlock.cpp test_scheduler.cpp + test_submit_gate.cpp ) target_link_libraries(kpn_tests PRIVATE diff --git a/tests/test_submit_gate.cpp b/tests/test_submit_gate.cpp new file mode 100644 index 0000000..b84ac90 --- /dev/null +++ b/tests/test_submit_gate.cpp @@ -0,0 +1,173 @@ +// Regression: a node must never end up idle with a wake outstanding. +// +// 9c5ce5f established that invariant and implemented it as two independent +// atomics — queued_ for "a firing is in flight", wake_pending_ for "a wake +// arrived during one". Two variables cannot express it, because the release +// side has to read and write both and a wake can land in between: +// +// producer (try_submit) worker (release_and_recheck) +// ------------------------ ---------------------------- +// CAS reads queued_ == true, fails +// queued_.store(false) +// wake_pending_.exchange(false) -> false +// wake_pending_.store(true) +// +// queued_ false, wake_pending_ true, nothing running and nothing scheduled. +// Not a memory-ordering subtlety: the interleaving holds under seq_cst. +// +// LegacyGate below is that protocol verbatim, with a hook between the failed +// CAS and the wake_pending_ store so the interleaving can be forced rather than +// waited for. That makes the loss deterministic and the test non-flaky, and it +// keeps the defect on record now that the code implementing it is gone. +// +// A note on what is NOT tested here, because it would be misleading to imply +// otherwise: there is no black-box, node-level test that fails before this fix +// and passes after. Every call site of release_and_recheck() happens to follow +// it with a level re-check — on_input_ready(), or outputs_have_space() on the +// parked path — which rediscovers the state a lost wake would have signalled. +// That masking is a property of the call sites, not of the mechanism, and the +// point of the fix is that a future early return that forgets the re-check no +// longer reintroduces a hang. The value is structural, so the tests are +// structural: the state machine is pinned by contract, and the defect it +// replaces is pinned by demonstration. +#include +#include + +#include +#include +#include + +using namespace kpn; + +namespace { + +// The pre-fix protocol, with a seam at the point where the race lives. +class LegacyGate { +public: + std::function before_recording_wake; + + bool claim() noexcept { + bool expected = false; + if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + return true; + if (before_recording_wake) before_recording_wake(); + wake_pending_.store(true, std::memory_order_release); + return false; + } + bool release() noexcept { + queued_.store(false, std::memory_order_release); + if (wake_pending_.exchange(false, std::memory_order_acq_rel)) { + bool expected = false; + if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + return true; + } + return false; + } + bool queued() const noexcept { return queued_.load(std::memory_order_relaxed); } + bool wake_pending() const noexcept { return wake_pending_.load(std::memory_order_relaxed); } + +private: + std::atomic queued_{false}; + std::atomic wake_pending_{false}; +}; + +} // namespace + +TEST_CASE("the two-atomic gate loses a wake, deterministically", "[submit_gate]") { + LegacyGate gate; + bool resubmitted = true; + + REQUIRE(gate.claim()); // a firing is now in flight + + // Force the interleaving: the firing completes in the window between the + // second wake's failed CAS and its record of that wake. + gate.before_recording_wake = [&] { resubmitted = gate.release(); }; + + const bool submitted = gate.claim(); + + // The wake was neither submitted by the producer nor honoured by the + // release. Nothing is scheduled, and nothing else will re-trigger it. + CHECK_FALSE(submitted); + CHECK_FALSE(resubmitted); + CHECK_FALSE(gate.queued()); + CHECK(gate.wake_pending()); // recorded, and never to be consumed +} + +TEST_CASE("submit gate: a wake during a firing is honoured", "[submit_gate]") { + SubmitGate gate; + + REQUIRE(gate.claim()); // idle -> queued, caller submits + REQUIRE(gate.queued()); + REQUIRE_FALSE(gate.wake_pending()); + + REQUIRE_FALSE(gate.claim()); // second wake is recorded, not submitted + REQUIRE(gate.wake_pending()); + + REQUIRE(gate.release()); // and honoured when the firing ends + // The gate stays claimed across the handover, so the node is never + // momentarily idle while a submission for it is in flight. This is the + // state the legacy gate could not represent. + REQUIRE(gate.queued()); + REQUIRE_FALSE(gate.wake_pending()); + + REQUIRE_FALSE(gate.release()); // no further wake: now idle + REQUIRE_FALSE(gate.queued()); +} + +TEST_CASE("submit gate: repeated wakes collapse to one resubmission", "[submit_gate]") { + // Collapsing is deliberate. A firing consumes one item and its caller then + // re-checks the input level, so the gate only has to guarantee that at + // least one more firing follows a wake, not one per wake. + SubmitGate gate; + REQUIRE(gate.claim()); + for (int i = 0; i < 10; ++i) REQUIRE_FALSE(gate.claim()); + REQUIRE(gate.release()); + REQUIRE_FALSE(gate.release()); +} + +TEST_CASE("submit gate: force_idle drops a recorded wake", "[submit_gate]") { + // Stop paths use this deliberately — honouring a wake there would resubmit + // a node that has already been told to stop. + SubmitGate gate; + REQUIRE(gate.claim()); + REQUIRE_FALSE(gate.claim()); + REQUIRE(gate.wake_pending()); + + gate.force_idle(); + REQUIRE_FALSE(gate.queued()); + REQUIRE_FALSE(gate.wake_pending()); + REQUIRE(gate.claim()); // and the gate is reusable afterwards +} + +TEST_CASE("submit gate: concurrent claim and release stay consistent", "[submit_gate]") { + // Not a lost-wake test — see the header note. This is a TSan target and a + // check that the CAS loops always terminate and always leave the gate in a + // reachable state: exactly one party may hold the claim at a time, so the + // count of claims granted must equal the count of releases that ended idle. + SubmitGate gate; + std::atomic granted{0}, ended_idle{0}; + std::atomic stop{false}; + + std::thread waker([&] { + while (!stop.load(std::memory_order_relaxed)) + if (gate.claim()) granted.fetch_add(1, std::memory_order_relaxed); + }); + std::thread worker([&] { + while (!stop.load(std::memory_order_relaxed)) + if (gate.queued() && !gate.release()) + ended_idle.fetch_add(1, std::memory_order_relaxed); + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + stop.store(true, std::memory_order_relaxed); + waker.join(); + worker.join(); + + // Drain whatever claim is outstanding so the two counts can be compared. + while (gate.queued()) + if (!gate.release()) ended_idle.fetch_add(1, std::memory_order_relaxed); + + INFO("granted " << granted.load() << " ended idle " << ended_idle.load()); + REQUIRE(granted.load() > 0); + CHECK(granted.load() == ended_idle.load()); +}