KPN/tests/test_shared_resource.cpp

#include <catch2/catch_test_macros.hpp>
#include <catch2/catch_approx.hpp>
#include <kpn/shared_resource.hpp>
#include <kpn/channel.hpp>

#include <atomic>
#include <chrono>
#include <thread>
#include <vector>

using namespace kpn;
using namespace std::chrono_literals;

// ── Basic acquire / release ───────────────────────────────────────────────────

TEST_CASE("acquire returns guard that accesses the resource", "[shared_resource]") {
    SharedResource<int> res(42);
    {
        auto g = res.acquire();
        REQUIRE(*g == 42);
        *g = 99;
    } // g released here — second acquire must not overlap in the same thread
    {
        auto g2 = res.acquire();
        REQUIRE(*g2 == 99);
    }
}

TEST_CASE("guard operator-> reaches resource members", "[shared_resource]") {
    struct Pair { int x{1}; int y{2}; };
    SharedResource<Pair> res;
    auto g = res.acquire();
    REQUIRE(g->x == 1);
    REQUIRE(g->y == 2);
    g->x = 10;
    REQUIRE(g.get().x == 10);
}

TEST_CASE("guard releases on scope exit", "[shared_resource]") {
    SharedResource<int> res(0);
    {
        auto g = res.acquire();
        REQUIRE(res.snapshot("r").held);
    }
    // After guard destroyed, resource is free
    REQUIRE(!res.snapshot("r").held);
}

// ── Mutual exclusion ──────────────────────────────────────────────────────────

TEST_CASE("only one thread holds the resource at a time", "[shared_resource]") {
    SharedResource<int> res(0);
    std::atomic<int> concurrent_holders{0};
    std::atomic<int> violations{0};
    std::atomic<bool> go{false};

    auto worker = [&] {
        while (!go.load()) std::this_thread::yield();
        for (int i = 0; i < 20; ++i) {
            auto g = res.acquire();
            int h = concurrent_holders.fetch_add(1) + 1;
            if (h > 1) violations.fetch_add(1);
            std::this_thread::sleep_for(100us);
            concurrent_holders.fetch_sub(1);
        }
    };

    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i) threads.emplace_back(worker);
    go.store(true);
    for (auto& t : threads) t.join();

    REQUIRE(violations.load() == 0);
}

// ── Priority ordering ─────────────────────────────────────────────────────────

TEST_CASE("higher priority waiter is served before lower priority waiter", "[shared_resource]") {
    SharedResource<int> res(0);

    // Hold the resource so threads have to queue.
    auto holder = res.acquire();

    std::vector<int> order;
    std::mutex order_mtx;

    // Launch two waiters: low priority first, then high priority.
    std::thread low([&] {
        auto g = res.acquire([] { return 0.1f; });
        std::lock_guard lk(order_mtx);
        order.push_back(1);
    });
    std::this_thread::sleep_for(5ms); // ensure low is queued first

    std::thread high([&] {
        auto g = res.acquire([] { return 0.9f; });
        std::lock_guard lk(order_mtx);
        order.push_back(2);
    });
    std::this_thread::sleep_for(5ms); // ensure high is also queued

    // Release — high priority should win even though low arrived first.
    { auto drop = std::move(holder); }

    low.join();
    high.join();

    REQUIRE(order.size() == 2);
    REQUIRE(order[0] == 2); // high priority served first
    REQUIRE(order[1] == 1);
}

// ── acquire_balanced uses channel fills ───────────────────────────────────────

TEST_CASE("acquire_balanced: full input + empty output gives score ~1.0", "[shared_resource]") {
    // We test the priority function indirectly via ordering.
    // Node A: in=full, out=empty  → score ≈ 1.0  (high)
    // Node B: in=empty, out=full  → score ≈ 0.0  (low)

    Channel<int> in_a(4);   // fill it
    Channel<int> out_a(4);  // leave empty
    Channel<int> in_b(4);   // leave empty
    Channel<int> out_b(4);  // fill it

    in_a.enable();  out_a.enable();
    in_b.enable();  out_b.enable();

    for (int i = 0; i < 4; ++i) { in_a.push(i); out_b.push(i); }

    SharedResource<int> res(0);
    auto holder = res.acquire(); // block others

    std::vector<int> order;
    std::mutex mtx;

    // Node B (low priority) waits first
    std::thread tb([&] {
        auto g = res.acquire_balanced(in_b, out_b);
        std::lock_guard lk(mtx);
        order.push_back(2);
    });
    std::this_thread::sleep_for(5ms);

    // Node A (high priority) waits second
    std::thread ta([&] {
        auto g = res.acquire_balanced(in_a, out_a);
        std::lock_guard lk(mtx);
        order.push_back(1);
    });
    std::this_thread::sleep_for(5ms);

    { auto drop = std::move(holder); } // release

    ta.join();
    tb.join();

    REQUIRE(order.size() == 2);
    REQUIRE(order[0] == 1); // node A served first despite arriving second
}

// ── Statistics ────────────────────────────────────────────────────────────────

TEST_CASE("stats: acquisitions counted correctly", "[shared_resource]") {
    SharedResource<int> res(0);
    {
        auto g1 = res.acquire();
    }
    {
        auto g2 = res.acquire();
    }
    REQUIRE(res.snapshot("r").acquisitions == 2);
}

TEST_CASE("stats: peak_waiters reflects maximum concurrent queue depth", "[shared_resource]") {
    SharedResource<int> res(0);
    auto holder = res.acquire();

    std::atomic<int> ready{0};

    auto waiter = [&] {
        ready.fetch_add(1);
        auto g = res.acquire();
    };

    std::thread t1(waiter), t2(waiter), t3(waiter);

    // Wait until all three are queued
    while (ready.load() < 3) std::this_thread::sleep_for(1ms);
    std::this_thread::sleep_for(5ms); // give them time to block on acquire

    { auto drop = std::move(holder); } // release

    t1.join(); t2.join(); t3.join();

    REQUIRE(res.snapshot("r").peak_waiters >= 2); // at least 2 queued simultaneously
}

TEST_CASE("stats: current_waiters returns to 0 after all served", "[shared_resource]") {
    SharedResource<int> res(0);
    auto holder = res.acquire();

    std::thread t1([&] { auto g = res.acquire(); });
    std::thread t2([&] { auto g = res.acquire(); });
    std::this_thread::sleep_for(10ms);

    { auto drop = std::move(holder); }
    t1.join(); t2.join();

    REQUIRE(res.snapshot("r").current_waiters == 0);
}

TEST_CASE("stats: avg_wait_ms is positive when contention occurred", "[shared_resource]") {
    SharedResource<int> res(0);
    {
        auto holder = res.acquire();
        std::thread t([&] { auto g = res.acquire(); });
        std::this_thread::sleep_for(10ms);
        { auto drop = std::move(holder); }
        t.join();
    }
    REQUIRE(res.snapshot("r").avg_wait_ms > 0.0);
}

// ── No-arg acquire ────────────────────────────────────────────────────────────

TEST_CASE("no-arg acquire works and releases correctly", "[shared_resource]") {
    SharedResource<int> res(7);
    auto g = res.acquire();
    REQUIRE(*g == 7);
    REQUIRE(res.snapshot("r").held);
}

// ── make_shared_resource factory ──────────────────────────────────────────────

TEST_CASE("make_shared_resource constructs with forwarded args", "[shared_resource]") {
    auto res = make_shared_resource<std::string>("hello");
    auto g = res.acquire();
    REQUIRE(*g == "hello");
}

// Regression: a waiter must be releasable, or teardown waits on it forever.
//
// acquire() blocks on a condition variable whose predicate only becomes true
// when release() hands over ownership. There was no timeout and no stop
// condition, so a node parked there ignored teardown entirely: its worker never
// returned, the pool's join never completed, and shutdown hung waiting for a
// resource nobody was going to release — which is exactly the case when the
// holder is being stopped too.
//
// close() turns that into an exception the node's existing error path already
// handles, and networks now call it on registered resources before stopping any
// node, for the same reason.
TEST_CASE("closing a shared resource releases its waiters", "[shared_resource]") {
    SharedResource<int> res(42);

    auto holder = res.acquire();          // resource is now held

    std::atomic<bool> threw{false}, returned{false};
    std::thread waiter([&] {
        try {
            auto g = res.acquire();       // blocks: someone else holds it
            (void)g;
        } catch (const ResourceClosedError&) {
            threw.store(true, std::memory_order_release);
        }
        returned.store(true, std::memory_order_release);
    });

    // Let it park, then tear down without ever releasing the holder.
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    REQUIRE_FALSE(returned.load(std::memory_order_acquire));

    res.close();
    waiter.join();

    CHECK(threw.load(std::memory_order_acquire));
}

TEST_CASE("acquiring a closed resource fails immediately", "[shared_resource]") {
    SharedResource<int> res(7);
    res.close();
    CHECK_THROWS_AS(res.acquire(), ResourceClosedError);

    // Reusable across runs once reopened.
    res.reopen();
    CHECK_NOTHROW(res.acquire());
}