bugfix

//

// This test exercises netplus::WriterPreferringSharedMutex directly (not a

// full QUIC connection) since the fix is a lock-fairness property of the

// mutex itself: with several threads continuously cycling a shared lock,

// confirm a concurrent writer still acquires the exclusive lock within a

// small bounded time instead of waiting for the readers to stop.

// mutex itself. Confirmed empirically while writing this test: with too few

// reader threads (e.g. 4, well under this machine's core count), even a

// plain, reader-preferring std::shared_mutex lets the writer in almost

// immediately -- there's always a gap between reader iterations for the

// writer to slip into. Starvation only reproduces once reader thread count

// is at or above the core count, so every core is kept continuously busy

// re-acquiring the shared lock with no free core left to service a queued

// writer's wakeup. So this test deliberately scales reader count to

// hardware_concurrency() (floored at 16) rather than using a small fixed

// number, and bounds its wait so a real regression fails fast instead of

// hanging CI.

#include <iostream>

#include <thread>

#include <vector>

#include <mutex>

#include <shared_mutex>

#include <algorithm>

#include "rwlock.h"

    else    { std::cout << "  FAIL: " << name << std::endl; g_failed++; }

}

int main() {

    std::cout << "=== rwlock_writer_starvation_test ===" << std::endl;

// Returns milliseconds to acquire the writer lock, or -1 if it didn't

// acquire within maxWait (bounded so a real regression fails fast rather

// than hanging).

template <typename Mutex>

static long measureWriterAcquire(int numReaders, std::chrono::milliseconds maxWait) {

    Mutex mtx;

    std::atomic<bool> stopReaders{false};

    WriterPreferringSharedMutex mtx;

    std::atomic<bool> stop{false};

    std::atomic<size_t> readerIterations{0};

    // Simulate pumpIncomingLocked()'s hot loop: several threads continuously

    // taking and releasing the shared lock with no gaps, exactly the access

    // pattern sendStreamData()'s congestion-wait spin produces on a real,

    // busy connection.

    static constexpr int NUM_READERS = 4;

    std::vector<std::thread> readers;

    for (int i = 0; i < NUM_READERS; ++i) {

    readers.reserve(numReaders);

    for (int i = 0; i < numReaders; ++i) {

        readers.emplace_back([&] {

            while (!stop.load(std::memory_order_relaxed)) {

                std::shared_lock<WriterPreferringSharedMutex> lock(mtx);

                readerIterations.fetch_add(1, std::memory_order_relaxed);

            while (!stopReaders.load(std::memory_order_relaxed)) {

                std::shared_lock<Mutex> lock(mtx);

            }

        });

    }

    // Let readers get into steady-state cycling before the writer shows up --

    // mirrors a real connection's transfer already being in progress when a

    // brand-new peer tries to register.

    std::this_thread::sleep_for(std::chrono::milliseconds(20));

    // Give the readers a head start so they're mid-cycle, same as a real

    // transfer already in progress when a new connection tries to register.

    std::this_thread::sleep_for(std::chrono::milliseconds(50));

    check(readerIterations.load() > 0, "readers are actively cycling the shared lock");

    // Writer, simulating accept() registering a brand-new connection.

    std::atomic<bool> writerDone{false};

    long elapsedMs = -1;

    std::thread writer([&] {

        auto start = std::chrono::steady_clock::now();

    {

        std::unique_lock<WriterPreferringSharedMutex> lock(mtx);

        std::unique_lock<Mutex> lock(mtx);

        elapsedMs = std::chrono::duration_cast<std::chrono::milliseconds>(

            std::chrono::steady_clock::now() - start).count();

        writerDone.store(true, std::memory_order_relaxed);

    });

    auto deadline = std::chrono::steady_clock::now() + maxWait;

    while (!writerDone.load(std::memory_order_relaxed) && std::chrono::steady_clock::now() < deadline) {

        std::this_thread::sleep_for(std::chrono::milliseconds(5));

    }

    auto elapsed = std::chrono::steady_clock::now() - start;

    auto elapsedMs = std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();

    stop.store(true, std::memory_order_relaxed);

    stopReaders.store(true, std::memory_order_relaxed);

    for (auto& t : readers) t.join();

    writer.join();

    return writerDone.load(std::memory_order_relaxed) ? elapsedMs : -1;

}

int main() {

    std::cout << "=== rwlock_writer_starvation_test ===" << std::endl;

    const unsigned int hwThreads = std::max(16u, std::thread::hardware_concurrency());

    static constexpr auto MAX_WAIT = std::chrono::milliseconds(2000);

    long fixedMs = measureWriterAcquire<WriterPreferringSharedMutex>(hwThreads, MAX_WAIT);

    std::cout << "  WriterPreferringSharedMutex (" << hwThreads << " readers): "

              << (fixedMs < 0 ? "STARVED" : std::to_string(fixedMs) + "ms") << std::endl;

    check(fixedMs >= 0 && fixedMs < 500,

          "writer-preferring mutex acquires promptly despite continuous reader load");

    std::cout << "  writer acquired exclusive lock after " << elapsedMs << "ms" << std::endl;

    // Generous bound (the confirmed prod incident saw multi-second/minutes

    // starvation) -- this just needs to prove the writer isn't stuck behind

    // an unbounded stream of new readers, not pin down an exact latency.

    check(elapsedMs < 500, "writer acquired the lock promptly despite continuous reader load");

#if defined(__GLIBC__)

    // On non-glibc platforms WriterPreferringSharedMutex is just an alias

    // for std::shared_mutex (rwlock.h), so this comparison would be

    // measuring the exact same type against itself -- only meaningful here.

    long plainMs = measureWriterAcquire<std::shared_mutex>(hwThreads, MAX_WAIT);

    std::cout << "  plain std::shared_mutex (" << hwThreads << " readers): "

              << (plainMs < 0 ? "STARVED" : std::to_string(plainMs) + "ms") << std::endl;

    check(plainMs < 0 || plainMs >= 500,

          "sanity check: plain shared_mutex actually starves under the same load "

          "(proves this test has real discriminating power, not a false pass)");

#endif

    std::cout << (g_failed == 0 ? "ALL TESTS PASSED" : "SOME TESTS FAILED")

              << " (" << g_passed << " passed, " << g_failed << " failed)" << std::endl;