// Contended stress tests for the lock-free SPSC Channel<T>.
//
// The other channel tests (test_channel.cpp) are single-threaded or use a
// single 20 ms sleep to order two threads — they never actually contend on the
// ring, so they exercise neither the memory-ordering pairing nor the
// spin/futex/lost-wakeup logic in pop().
//
// These tests are written to be run under ThreadSanitizer:
//
//   cmake -B build -DKPN_SANITIZER=thread -DKPN_BUILD_EXAMPLES=OFF -DKPN_BUILD_PYTHON=OFF
//   cmake --build build --target kpn_tests_tsan
//   ./build/tests/kpn_tests_tsan
//
// They are also valid (and meaningful) without a sanitizer: the value/sequence
// assertions catch lost or duplicated items regardless of build flags. TSan
// adds detection of the underlying data race even on runs where the race did
// not corrupt observable state.
//
// Channel<T> is SPSC: exactly one producer thread and one consumer thread per
// channel. Every scenario below honours that contract.

#include <string>
#include <catch2/catch_test_macros.hpp>
#include <atomic>
#include <chrono>
#include <kpn/channel.hpp>
#include <thread>
#include <vector>

using namespace kpn;
using namespace std::chrono_literals;

namespace {

// Repeat each scenario enough times that rare interleavings (spin window just
// missing / just catching the next push, disable landing inside the futex
// wait) actually occur across a run. Kept modest so a TSan run stays minutes,
// not hours.
constexpr int kReps = 200;

} // namespace

TEST_CASE("SPSC: every pushed item is popped exactly once, in order", "[channel][stress]") {
    // Small capacity forces frequent full/empty transitions, so both the
    // producer's overflow-retry and the consumer's spin->futex path are hit
    // many times. The producer retries on overflow rather than dropping, so
    // the consumer must observe a strictly contiguous 0..N-1 sequence.
    constexpr int N = 50'000;
    Channel<int> ch(/*capacity=*/4, /*spin_count=*/16);

    std::thread producer([&] {
        for (int i = 0; i < N; ++i) {
            for (;;) {
                try { ch.push(i); break; }
                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
            }
        }
    });

    int expected = 0;
    bool in_order = true;
    for (int i = 0; i < N; ++i) {
        int v = ch.pop();
        if (v != expected) in_order = false;
        ++expected;
    }
    producer.join();

    REQUIRE(in_order);
    REQUIRE(expected == N);
    REQUIRE(ch.size() == 0);
}

TEST_CASE("SPSC: tight empty<->non-empty transitions exercise spin/futex boundary",
          "[channel][stress]") {
    // spin_count=0 forces every empty pop() straight into atomic::wait, so this
    // hammers the lost-wakeup guard (snapshot wake_, re-check tail_, then wait).
    // The producer pushes one item then waits to go empty again, maximising the
    // number of empty->non-empty edges relative to item count.
    constexpr int N = 20'000;
    Channel<int> ch(/*capacity=*/2, /*spin_count=*/0);

    std::thread producer([&] {
        for (int i = 0; i < N; ++i) {
            for (;;) {
                try { ch.push(i); break; }
                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
            }
        }
    });

    long sum = 0;
    for (int i = 0; i < N; ++i) sum += ch.pop();
    producer.join();

    // Sum of 0..N-1 — detects any lost or duplicated item.
    REQUIRE(sum == static_cast<long>(N) * (N - 1) / 2);
}

TEST_CASE("SPSC: disable() while consumer is blocked in pop() unblocks cleanly",
          "[channel][stress]") {
    // The data race of record: consumer blocked in pop() (spinning or parked in
    // the futex) while the owner thread calls disable(). pop() must observe the
    // close and throw ChannelClosedError — it must not hang and must not read
    // past the ring. Repeated so disable() lands at many points in pop()'s loop.
    for (int rep = 0; rep < kReps; ++rep) {
        Channel<int> ch(/*capacity=*/4, /*spin_count=*/8);
        std::atomic<bool> threw{false};
        std::atomic<bool> finished{false};

        std::thread consumer([&] {
            try {
                ch.pop();              // empty channel: will block
            } catch (const ChannelClosedError&) {
                threw.store(true, std::memory_order_relaxed);
            }
            finished.store(true, std::memory_order_relaxed);
        });

        // Give the consumer a chance to reach the wait, then close.
        std::this_thread::sleep_for(50us);
        ch.disable();

        consumer.join();
        REQUIRE(finished.load());
        REQUIRE(threw.load());
    }
}

TEST_CASE("SPSC: producer racing a disable() never throws and never hangs",
          "[channel][stress]") {
    // Mirror of the above from the producer side: push() racing disable() must
    // either enqueue or silently drop, never throw ChannelClosedError and never
    // wedge. Overflow is still a legal outcome (full accepting channel) and is
    // tolerated here.
    for (int rep = 0; rep < kReps; ++rep) {
        Channel<int> ch(/*capacity=*/8, /*spin_count=*/8);
        std::atomic<bool> bad{false};

        std::thread producer([&] {
            for (int i = 0; i < 1000; ++i) {
                try { ch.push(i); }
                catch (const ChannelOverflowError&) { /* legal: full */ }
                catch (...) { bad.store(true, std::memory_order_relaxed); break; }
            }
        });

        std::this_thread::sleep_for(20us);
        ch.disable();           // owner closes mid-stream
        producer.join();

        REQUIRE_FALSE(bad.load());
    }
}

TEST_CASE("SPSC: push_callback fires for every push, never missed",
          "[channel][stress]") {
    // 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); });

    constexpr int N = 10'000;
    std::thread producer([&] {
        for (int i = 0; i < N; ++i) {
            for (;;) {
                try { ch.push(i); break; }
                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
            }
        }
    });

    for (int i = 0; i < N; ++i) (void)ch.pop();
    producer.join();

    // 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.
//
// push_sentinel() publishes has_eof_ (release) after the producer's N ring
// pushes; a consumer that observes has_eof_ (acquire) therefore also observes
// every value pushed before it. Both pop() and try_pop_now() only surface the
// sentinel once the ring is *freshly* observed empty, so the sentinel is the
// strictly last item received — it never jumps ahead of a ring value pushed
// before it. These tests treat the sentinel as a hard "last message" barrier
// (the consumer stops draining the moment it sees it) and assert that all N
// values arrived, in a contiguous 0..N-1 sequence, before it.
//
// Regression guard: an earlier version of pop() checked emptiness against a
// stale tail_ snapshot from the top of its loop, so under load the sentinel
// could surface with a few real values still queued — breaking in_order /
// values==N here. Under TSan these also cover the has_eof_/eof_value_
// acquire/release handshake and the spin/futex wakeup on push_sentinel().

TEST_CASE("SPSC: sentinel is strictly last, after every value (blocking pop)",
          "[channel][stress]") {
    constexpr int N        = 20'000;
    constexpr int SENTINEL = -1;

    for (int rep = 0; rep < kReps; ++rep) {
        // Small ring + tiny spin window so the ring is frequently empty exactly
        // when the sentinel is published — the interleaving under test.
        Channel<int> ch(/*capacity=*/4, /*spin_count=*/8);

        std::thread producer([&] {
            for (int i = 0; i < N; ++i) {
                for (;;) {
                    try { ch.push(i); break; }
                    catch (const ChannelOverflowError&) { std::this_thread::yield(); }
                }
            }
            ch.push_sentinel(SENTINEL);   // must-deliver, never overflows/blocks
        });

        int  expected = 0;
        bool in_order = true;
        bool saw_sentinel = false;
        // Treat the sentinel as EOF: stop draining the instant it appears.
        for (;;) {
            int v = ch.pop();
            if (v == SENTINEL) { saw_sentinel = true; break; }
            if (v != expected) in_order = false;
            ++expected;
        }
        producer.join();

        REQUIRE(saw_sentinel);
        REQUIRE(in_order);
        REQUIRE(expected == N);        // all N values received before the sentinel
        REQUIRE(ch.size() == 0);
        REQUIRE(ch.approx_size() == 0);
    }
}

TEST_CASE("SPSC: sentinel is strictly last, after every value (try_pop_now)",
          "[channel][stress]") {
    // The pool-node consume path is try_pop_now(), not pop(): it must surface
    // the out-of-band sentinel only once the ring is freshly observed empty.
    // The consumer spins with no sleeps, racing the producer at full tilt
    // across the empty-ring boundary where take_sentinel() is reached.
    constexpr int N        = 20'000;
    constexpr int SENTINEL = -1;

    for (int rep = 0; rep < kReps; ++rep) {
        Channel<int> ch(/*capacity=*/4, /*spin_count=*/0);

        std::thread producer([&] {
            for (int i = 0; i < N; ++i) {
                for (;;) {
                    try { ch.push(i); break; }
                    catch (const ChannelOverflowError&) { std::this_thread::yield(); }
                }
            }
            ch.push_sentinel(SENTINEL);
        });

        int  expected = 0;
        bool in_order = true;
        bool saw_sentinel = false;
        int  v;
        for (;;) {
            if (!ch.try_pop_now(v)) { std::this_thread::yield(); continue; }
            if (v == SENTINEL) { saw_sentinel = true; break; }
            if (v != expected) in_order = false;
            ++expected;
        }
        producer.join();

        REQUIRE(saw_sentinel);
        REQUIRE(in_order);
        REQUIRE(expected == N);
        // Sentinel held no ring slot; once taken the channel is fully empty.
        REQUIRE(ch.size() == 0);
        REQUIRE(ch.approx_size() == 0);
    }
}

// Contended: a producer offering sentinels while the consumer takes them.
//
// The old push_sentinel wrote eof_value_ with no regard for whether the
// consumer was reading it, so a second offer racing a take was a data race on
// the storage — for the shared_ptr form used by non-trivial types, on the
// refcount. Under TSan the old code reports it; the handshake added alongside
// this test makes the producer's write conditional on observing the slot free,
// which is what serialises the two.
//
// Payload is a std::string so the storage is the shared_ptr path rather than
// the trivially-copyable one, and each token carries its own identity so a torn
// value shows up as a mismatch rather than as a plausible-looking result.
TEST_CASE("SPSC: offering sentinels concurrently with takes is race-free",
          "[channel][stress][sentinel]") {
    constexpr int kRounds = 20000;
    Channel<std::string> ch(4);

    std::atomic<int> taken{0};
    std::atomic<bool> torn{false};
    std::atomic<bool> done{false};

    std::thread consumer([&] {
        std::string out;
        while (!done.load(std::memory_order_acquire) || ch.approx_size() > 0) {
            if (ch.try_pop_now(out)) {
                if (out.rfind("eof-", 0) != 0) torn.store(true, std::memory_order_relaxed);
                taken.fetch_add(1, std::memory_order_relaxed);
            }
        }
    });

    int accepted = 0;
    for (int i = 0; i < kRounds; ++i) {
        std::string tok = "eof-" + std::to_string(i);
        if (ch.try_push_sentinel(tok) == Channel<std::string>::SentinelResult::Taken)
            ++accepted;
    }
    done.store(true, std::memory_order_release);
    consumer.join();

    INFO("accepted " << accepted << " taken " << taken.load());
    CHECK_FALSE(torn.load(std::memory_order_relaxed));
    // Every accepted token must be delivered: the slot is refused while full,
    // so acceptance and delivery are one-to-one.
    CHECK(taken.load(std::memory_order_relaxed) == accepted);
    CHECK(accepted > 0);
}