diff --git a/tests/ThreadPool_Tests.cpp b/tests/ThreadPool_Tests.cpp new file mode 100644 index 00000000..7e84d894 --- /dev/null +++ b/tests/ThreadPool_Tests.cpp @@ -0,0 +1,165 @@ +// SPDX-License-Identifier: GPL-2.0-only +/* + * Digital Voice Modem - Test Suite + * GPLv2 Open Source. Use is subject to license terms. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * Copyright (C) 2026 Bryan Biedenkapp, N2PLL + * + */ +#include "Defines.h" +#include "common/ThreadPool.h" + +#include + +#include +#include +#include +#include + +// --------------------------------------------------------------------------- +// Global Functions +// --------------------------------------------------------------------------- + +/** + * @brief Waits until the atomic value reaches the target or the timeout expires. + * @param value The atomic value to monitor. + * @param target The target value to wait for. + * @param timeoutMs The maximum time to wait in milliseconds. + * @return true if the target value was reached, false if the timeout expired. + */ +bool waitForCount(std::atomic& value, uint32_t target, uint32_t timeoutMs = 1500U) +{ + const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeoutMs); + while (std::chrono::steady_clock::now() < deadline) { + if (value.load() >= target) + return true; + + Thread::sleep(1U); + } + + return value.load() >= target; +} + +TEST_CASE("ThreadPool enforces minimum worker count", "[common][threadpool]") +{ + ThreadPool pool(1U, "tp-min"); + REQUIRE(pool.getMaxWorkerCnt() == 4U); +} + +TEST_CASE("ThreadPool runs enqueued tasks", "[common][threadpool]") +{ + ThreadPool pool(4U, "tp-run"); + std::atomic executed { 0U }; + + pool.start(); + + const uint32_t taskCount = 12U; + bool allAccepted = true; + for (uint32_t i = 0U; i < taskCount; i++) { + ThreadPoolTask* task = new_pooltask([&executed]() { + executed.fetch_add(1U); + }); + + if (!pool.enqueue(task)) { + delete task; + allAccepted = false; + break; + } + } + REQUIRE(allAccepted); + REQUIRE(waitForCount(executed, taskCount)); + + pool.stop(); + pool.wait(); + + REQUIRE(executed.load() == taskCount); +} + +TEST_CASE("ThreadPool rejects enqueue when stopped", "[common][threadpool]") +{ + ThreadPool pool(4U, "tp-stop"); + + ThreadPoolTask* beforeStart = new_pooltask([]() {}); + const bool beforeStartAccepted = pool.enqueue(beforeStart); + REQUIRE_FALSE(beforeStartAccepted); + if (!beforeStartAccepted) + delete beforeStart; + + pool.start(); + pool.stop(); + + ThreadPoolTask* afterStop = new_pooltask([]() {}); + const bool afterStopAccepted = pool.enqueue(afterStop); + REQUIRE_FALSE(afterStopAccepted); + if (!afterStopAccepted) + delete afterStop; + + pool.wait(); +} + +TEST_CASE("ThreadPool max queued task limit is enforced", "[common][threadpool]") +{ + ThreadPool pool(4U, "tp-queue"); + pool.setMaxQueuedTasks(0U); + + std::atomic executed { 0U }; + std::mutex gateMutex; + std::condition_variable gateCond; + bool releaseWorkers = false; + + pool.start(); + + // Block all workers so one extra task remains queued. + bool workerTasksAccepted = true; + for (uint32_t i = 0U; i < pool.getMaxWorkerCnt(); i++) { + ThreadPoolTask* task = new_pooltask([&]() { + std::unique_lock lock(gateMutex); + gateCond.wait(lock, [&]() { return releaseWorkers; }); + executed.fetch_add(1U); + }); + + if (!pool.enqueue(task)) { + delete task; + workerTasksAccepted = false; + break; + } + } + + if (workerTasksAccepted) + Thread::sleep(10U); + + // Now enforce a one-item queue while all workers are occupied. + pool.setMaxQueuedTasks(1U); + + ThreadPoolTask* acceptedQueuedTask = new_pooltask([&executed]() { + executed.fetch_add(1U); + }); + bool queuedAccepted = false; + if (workerTasksAccepted) + queuedAccepted = pool.enqueue(acceptedQueuedTask); + if (!queuedAccepted) + delete acceptedQueuedTask; + + ThreadPoolTask* overflowTask = new_pooltask([]() {}); + bool overflowAccepted = false; + if (workerTasksAccepted && queuedAccepted) + overflowAccepted = pool.enqueue(overflowTask); + if (!overflowAccepted) + delete overflowTask; + + // scope is intentional + { + std::lock_guard lock(gateMutex); + releaseWorkers = true; + } + gateCond.notify_all(); + + REQUIRE(workerTasksAccepted); + REQUIRE(queuedAccepted); + REQUIRE_FALSE(overflowAccepted); + REQUIRE(waitForCount(executed, pool.getMaxWorkerCnt())); + + pool.stop(); + pool.wait(); +} \ No newline at end of file diff --git a/tests/Thread_Tests.cpp b/tests/Thread_Tests.cpp new file mode 100644 index 00000000..f207d677 --- /dev/null +++ b/tests/Thread_Tests.cpp @@ -0,0 +1,206 @@ +// SPDX-License-Identifier: GPL-2.0-only +/* + * Digital Voice Modem - Test Suite + * GPLv2 Open Source. Use is subject to license terms. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * Copyright (C) 2026 Bryan Biedenkapp, N2PLL + * + */ +#include "Defines.h" +#include "common/Thread.h" + +#include + +#include +#include +#include +#include + +// --------------------------------------------------------------------------- +// Class Declaration +// --------------------------------------------------------------------------- + +/** + * @brief Test thread implementation for unit tests. + */ +class TestThread final : public Thread { +public: + /** + * @brief Constructs a TestThread instance. + * @param entryCount The atomic counter to track thread entry. + * @param mutex The mutex to protect shared state. + * @param cond The condition variable to signal thread completion. + * @param ran The flag indicating whether the thread has run. + */ + TestThread(std::atomic& entryCount, std::mutex& mutex, + std::condition_variable& cond, bool& ran) : + m_entryCount(entryCount), + m_mutex(mutex), + m_cond(cond), + m_ran(ran) + { + /* stub */ + } + + /** + * @brief Thread entry point. + */ + void entry() override + { + m_entryCount.fetch_add(1U); + + // scope is intentional + { + std::lock_guard lock(m_mutex); + m_ran = true; + } + + m_cond.notify_one(); + } + +private: + std::atomic& m_entryCount; + std::mutex& m_mutex; + std::condition_variable& m_cond; + bool& m_ran; +}; + +// --------------------------------------------------------------------------- +// Structure Declaration +// --------------------------------------------------------------------------- + +/** + * @brief Context structure for running a thread routine. + */ +struct RunAsThreadContext { + std::atomic callCount { 0U }; + std::mutex mutex; + std::condition_variable cond; + bool ran = false; +}; + +// --------------------------------------------------------------------------- +// Global Functions +// --------------------------------------------------------------------------- + +/** + * @brief Thread start routine that executes the thread's entry function. + * @param arg Pointer to the thread_t structure. + * @return void* Always returns nullptr. + */ +void* runAsThreadRoutine(void* arg) +{ + thread_t* thread = static_cast(arg); + if (thread == nullptr || thread->obj == nullptr) + return nullptr; + + RunAsThreadContext* ctx = static_cast(thread->obj); + ctx->callCount.fetch_add(1U); + + // scope is intentional + { + std::lock_guard lock(ctx->mutex); + ctx->ran = true; + } + + ctx->cond.notify_one(); + return nullptr; +} + +/** + * @brief Waits for a flag to become true with a timeout. + * @param mutex The mutex protecting the flag. + * @param cond The condition variable to wait on. + * @param flag The flag to wait for. + * @param timeoutMs The timeout in milliseconds. + * @return bool True if the flag became true within the timeout, false otherwise. + */ +bool waitForFlag(std::mutex& mutex, std::condition_variable& cond, bool& flag, uint32_t timeoutMs = 1000U) +{ + std::unique_lock lock(mutex); + return cond.wait_for(lock, std::chrono::milliseconds(timeoutMs), [&flag]() { return flag; }); +} + +TEST_CASE("Thread runs entry once and wait joins", "[common][thread]") +{ + std::atomic entryCount { 0U }; + std::mutex mutex; + std::condition_variable cond; + bool ran = false; + + TestThread thread(entryCount, mutex, cond, ran); + + REQUIRE_FALSE(thread.started()); + REQUIRE(thread.run()); + REQUIRE(thread.started()); + + REQUIRE(waitForFlag(mutex, cond, ran)); + + // A second run() call should not spawn another worker. + REQUIRE(thread.run()); + thread.wait(); + + REQUIRE(entryCount.load() == 1U); +} + +TEST_CASE("Thread wait and setName are safe before run", "[common][thread]") +{ + std::atomic entryCount { 0U }; + std::mutex mutex; + std::condition_variable cond; + bool ran = false; + + TestThread thread(entryCount, mutex, cond, ran); + + thread.setName("not-running"); + thread.wait(); + thread.detach(); + + REQUIRE(entryCount.load() == 0U); + REQUIRE_FALSE(thread.started()); +} + +TEST_CASE("Thread detach allows completion without wait", "[common][thread]") +{ + std::atomic entryCount { 0U }; + std::mutex mutex; + std::condition_variable cond; + bool ran = false; + + TestThread thread(entryCount, mutex, cond, ran); + + REQUIRE(thread.run()); + thread.detach(); + + REQUIRE(waitForFlag(mutex, cond, ran)); + REQUIRE(entryCount.load() == 1U); +} + +TEST_CASE("Thread runAsThread executes start routine", "[common][thread]") +{ + RunAsThreadContext ctx; + thread_t threadData; + + REQUIRE(Thread::runAsThread(&ctx, runAsThreadRoutine, &threadData)); + REQUIRE(waitForFlag(ctx.mutex, ctx.cond, ctx.ran)); + +#if defined(_WIN32) + ::WaitForSingleObject(threadData.thread, INFINITE); + ::CloseHandle(threadData.thread); +#else + ::pthread_join(threadData.thread, nullptr); +#endif // defined(_WIN32) + + REQUIRE(ctx.callCount.load() == 1U); +} + +TEST_CASE("Thread sleep delays execution", "[common][thread]") +{ + const auto begin = std::chrono::steady_clock::now(); + Thread::sleep(10U); + const auto end = std::chrono::steady_clock::now(); + + const auto elapsedMs = std::chrono::duration_cast(end - begin).count(); + REQUIRE(elapsedMs >= 5); +} \ No newline at end of file