parent
aea826fcbf
commit
8b7f5cee72
@ -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 <catch2/catch_test_macros.hpp>
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <mutex>
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 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<uint32_t>& 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<uint32_t> 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<uint32_t> 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<std::mutex> 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<std::mutex> lock(gateMutex);
|
||||
releaseWorkers = true;
|
||||
}
|
||||
gateCond.notify_all();
|
||||
|
||||
REQUIRE(workerTasksAccepted);
|
||||
REQUIRE(queuedAccepted);
|
||||
REQUIRE_FALSE(overflowAccepted);
|
||||
REQUIRE(waitForCount(executed, pool.getMaxWorkerCnt()));
|
||||
|
||||
pool.stop();
|
||||
pool.wait();
|
||||
}
|
||||
@ -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 <catch2/catch_test_macros.hpp>
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <mutex>
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 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<uint32_t>& 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<std::mutex> lock(m_mutex);
|
||||
m_ran = true;
|
||||
}
|
||||
|
||||
m_cond.notify_one();
|
||||
}
|
||||
|
||||
private:
|
||||
std::atomic<uint32_t>& 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<uint32_t> 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<thread_t*>(arg);
|
||||
if (thread == nullptr || thread->obj == nullptr)
|
||||
return nullptr;
|
||||
|
||||
RunAsThreadContext* ctx = static_cast<RunAsThreadContext*>(thread->obj);
|
||||
ctx->callCount.fetch_add(1U);
|
||||
|
||||
// scope is intentional
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<std::mutex> 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<uint32_t> 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<uint32_t> 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<uint32_t> 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<std::chrono::milliseconds>(end - begin).count();
|
||||
REQUIRE(elapsedMs >= 5);
|
||||
}
|
||||
Loading…
Reference in new issue