KPN/tests/test_submit_gate.cpp

// 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 <catch2/catch_test_macros.hpp>
#include <kpn/submit_gate.hpp>

#include <atomic>
#include <functional>
#include <thread>

using namespace kpn;

namespace {

// The pre-fix protocol, with a seam at the point where the race lives.
class LegacyGate {
public:
    std::function<void()> 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<bool> queued_{false};
    std::atomic<bool> 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<long> granted{0}, ended_idle{0};
    std::atomic<bool> 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());
}