test

// once.

class ThreadPool {

public:

    explicit ThreadPool(size_t num_threads) {

    // max_queue_size: 0 (default) keeps the original unbounded-queue behavior for every

    // existing caller (quicDispatchPool() and any other pre-existing user of this class). A

    // positive value makes submit() reject (return false) once that many tasks are already

    // queued and not yet picked up by a worker -- lets a caller degrade gracefully (e.g.

    // answer with a fast error) instead of piling work up behind a pool that may be fully

    // stuck, where an unbounded queue means new work waits forever with no way to notice.

    explicit ThreadPool(size_t num_threads, size_t max_queue_size = 0)

        : _max_queue_size(max_queue_size) {

        _workers.reserve(num_threads);

        for (size_t i = 0; i < num_threads; ++i) {

            _workers.emplace_back([this] { workerLoop(); });

    ThreadPool(const ThreadPool&) = delete;

    ThreadPool& operator=(const ThreadPool&) = delete;

    // Queues a task for execution on some worker thread. Safe to call from

    // any thread, including from within a task running on this same pool.

    void submit(std::function<void()> task) {

    // Queues a task for execution on some worker thread. Safe to call from any thread,

    // including from within a task running on this same pool. Returns false without queuing

    // the task if max_queue_size was set and the queue is already full -- a caller that needs

    // to recover state captured in the rejected task (e.g. to respond with an overload error

    // instead) should keep its own reference to that state alongside the lambda rather than

    // relying on getting `task` back, since it's discarded here on rejection either way.

    bool submit(std::function<void()> task) {

        {

            std::lock_guard<std::mutex> lk(_mtx);

            if (_max_queue_size > 0 && _tasks.size() >= _max_queue_size) return false;

            _tasks.push_back(std::move(task));

        }

        _cv.notify_one();

        return true;

    }

private:

    std::vector<std::thread> _workers;

    std::deque<std::function<void()>> _tasks;

    size_t _max_queue_size;

    std::mutex _mtx;

    std::condition_variable _cv;

    bool _stopping = false;

    target_link_libraries(idle_reaper_test netplus-static)

endif()

add_test(NAME idle_reaper_test COMMAND idle_reaper_test)

add_executable(threadpool_bounded_queue_test threadpool_bounded_queue_test.cpp)

if(WIN32)

    target_link_libraries(threadpool_bounded_queue_test netplus-static ws2_32)

else()

    target_link_libraries(threadpool_bounded_queue_test netplus-static)

endif()

add_test(NAME threadpool_bounded_queue_test COMMAND threadpool_bounded_queue_test)

// Correctness test for ThreadPool's opt-in bounded queue (src/threadpool.h),

// added after a production proxyplus incident where an unbounded queue let

// requests pile up forever behind a stuck pool with no way to notice or

// recover -- see proxyplus's HttpProxyServer / libhttppp's

// HttpEvent::_dispatchH2Stream for the real caller this exists for.

//

// Three things are checked:

//   1. Default (max_queue_size == 0) behaves exactly like before this

//      existed: submit() always returns true, queue grows unbounded.

//   2. A bounded pool accepts up to its limit and then rejects (returns

//      false) further submissions without blocking the caller or touching

//      the rejected task.

//   3. Once workers drain the backlog below the limit, submit() succeeds

//      again -- rejection is a transient "try something else right now"

//      signal, not a permanent trip.

#include <atomic>

#include <chrono>

#include <condition_variable>

#include <iostream>

#include <mutex>

#include <thread>

#include "../src/threadpool.h"

namespace {

bool testUnboundedDefaultAcceptsEverything() {

    netplus::ThreadPool pool(2);

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

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

        if (!pool.submit([&completed] { ++completed; })) {

            std::cerr << "FAIL: default-constructed pool rejected a submit (should be unbounded)\n";

            return false;

        }

    }

    for (int i = 0; i < 1000 && completed.load() < 500; ++i)

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

    if (completed.load() != 500) {

        std::cerr << "FAIL: expected 500 completed tasks, got " << completed.load() << "\n";

        return false;

    }

    return true;

}

bool testBoundedQueueRejectsPastLimit() {

    std::mutex mtx;

    std::condition_variable cv;

    bool release = false;

    // 1 worker, queue capacity 3: the worker immediately picks up task #1 and

    // blocks on `release`, leaving exactly 3 queue slots for everything else.

    netplus::ThreadPool pool(1, /*max_queue_size=*/3);

    if (!pool.submit([&] {

            std::unique_lock<std::mutex> lk(mtx);

            cv.wait(lk, [&] { return release; });

        })) {

        std::cerr << "FAIL: first submit (into an empty queue) was rejected\n";

        return false;

    }

    // Give the worker a moment to actually dequeue task #1 so the 3 slots

    // below land in the queue, not competing for the worker itself.

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

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

        if (!pool.submit([] {})) {

            std::cerr << "FAIL: submit " << i << " of 3 (within capacity) was rejected\n";

            return false;

        }

    }

    // Queue is now full (3/3) with the worker still blocked -- the next

    // submit must be rejected rather than queued or blocking this thread.

    bool rejected = !pool.submit([] {});

    if (!rejected) {

        std::cerr << "FAIL: submit past max_queue_size was accepted, expected rejection\n";

        return false;

    }

    {

        std::lock_guard<std::mutex> lk(mtx);

        release = true;

    }

    cv.notify_all();

    return true;

}

bool testCapacityFreesUpAfterDraining() {

    netplus::ThreadPool pool(2, /*max_queue_size=*/2);

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

    // Saturate: 2 workers pick up 2 tasks immediately, 2 more fill the queue.

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

        if (!pool.submit([&completed] {

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

                ++completed;

            })) {

            std::cerr << "FAIL: initial saturating submit " << i << " was rejected unexpectedly\n";

            return false;

        }

    }

    for (int i = 0; i < 200 && completed.load() < 4; ++i)

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

    if (completed.load() != 4) {

        std::cerr << "FAIL: backlog never drained, completed=" << completed.load() << "\n";

        return false;

    }

    // Pool is idle again -- submit() must accept new work, not stay stuck

    // rejecting because of the earlier saturation.

    if (!pool.submit([&completed] { ++completed; })) {

        std::cerr << "FAIL: submit after backlog drained was still rejected\n";

        return false;

    }

    for (int i = 0; i < 200 && completed.load() < 5; ++i)

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

    if (completed.load() != 5) {

        std::cerr << "FAIL: post-drain task never ran, completed=" << completed.load() << "\n";

        return false;

    }

    return true;

}

} // namespace

int main() {

    bool ok = true;

    ok &= testUnboundedDefaultAcceptsEverything();

    ok &= testBoundedQueueRejectsPastLimit();

    ok &= testCapacityFreesUpAfterDraining();

    if (ok) {

        std::cout << "threadpool_bounded_queue_test: all checks passed\n";

        return 0;

    }

    return 1;

}