Concurrency & Thread-Safe Collections

Relevant source files

The following files were used as context for generating this wiki page:

• include/HXLibs/container/ThreadPool.hpp
• include/HXLibs/container/SafeQueue.hpp
• tests/container/03_thread_pool.cpp
• tests/container/04_FutureResult.cpp
• tests/execution/demo.cpp
• include/HXLibs/container/FutureResult.hpp (Implicitly referenced via ThreadPool implementation)

HXLibs provides a set of high-performance concurrency primitives and thread-safe containers built for modern C++20 environments. The system facilitates asynchronous task execution through a managed thread pool and provides thread-safe data structures like SafeQueue to handle communication between execution contexts.

These components integrate with the project's coroutine system, allowing asynchronous results to be awaited in an EventLoop or chained using functional continuations.
Sources: include/HXLibs/container/ThreadPool.hpp, tests/execution/demo.cpp:20-40

Thread Pool Architecture

The ThreadPool class manages a collection of worker threads to execute tasks asynchronously. It supports two primary execution models: a fixed-size pool and a dynamic pool that scales based on task demand.

Execution Models

The pool operates in one of two modes defined by the ThreadPool::Model enum:

• FixedSizeAndNoCheck: Initializes a constant number of worker threads. It does not perform dynamic scaling.
• AsyncCheck: Spawns a dedicated administrator thread (_opThread) to periodically monitor load and adjust the number of workers based on a strategy.
  Sources: include/HXLibs/container/ThreadPool.hpp:82-85, include/HXLibs/container/ThreadPool.hpp:163-173

Dynamic Scaling Logic

In AsyncCheck mode, the administrator thread evaluates the ThreadPoolData using a Strategy function. The default strategy increases threads when the task queue grows and decreases them when threads remain idle.

The diagram shows the lifecycle of the administrator thread managing dynamic scaling while worker threads process tasks or exit based on signals.
Sources: include/HXLibs/container/ThreadPool.hpp:51-70, include/HXLibs/container/ThreadPool.hpp:185-212

Thread-Safe Task Management

Tasks are submitted to the pool via the addTask method, which returns a FutureResult<Res>. This object allows the caller to retrieve the result later or chain further processing.

Task Submission Flow

1. Wrap the function and arguments into a MoveOnlyFunctionAny.
2. Store the result placeholder in a FutureResult.
3. Push the task into the SafeQueue.
4. Notify a worker thread via condition_variable.
   Sources: include/HXLibs/container/ThreadPool.hpp:123-148

Result Chaining (thenTry)

FutureResult supports functional chaining via thenTry. This allows developers to define a pipeline of asynchronous operations where the output of one task serves as the input for the next, even if an exception occurs.
Sources: tests/container/03_thread_pool.cpp:80-120, include/HXLibs/container/ThreadPool.hpp:303-315

Component	Description
SafeQueue<T>	A thread-safe wrapper around std::queue using std::shared_mutex for concurrent access.
ThreadPoolData	A struct containing metrics like taskNum, nowThreadNum, and sleepThreadNum.
ThreadPool::run	Starts the pool with specific timers and strategies.
FutureResult<T>	A handle for asynchronous results, supporting get(), thenTry(), and coroutine co_await.
Sources: include/HXLibs/container/SafeQueue.hpp, include/HXLibs/container/ThreadPool.hpp:40-50, tests/container/04_FutureResult.cpp:21-34

Safe Containers

SafeQueue

SafeQueue is the primary synchronization container for the ThreadPool. It uses a std::shared_mutex to allow multiple threads to query the size or empty status (std::shared_lock) while ensuring exclusive access during modifications (std::unique_lock).

• frontAndPop(): Atomically retrieves and removes the front element using move semantics.
• emplace(Args&&... args): Forwards arguments to the internal queue while holding a unique lock.
  Sources: include/HXLibs/container/SafeQueue.hpp:39-65

This sequence illustrates the locking mechanism during a task retrieval operation in the thread pool.
Sources: include/HXLibs/container/SafeQueue.hpp:44-50

Implementation Details

Atomic Operations

The ThreadPool utilizes atomic CAS (Compare-And-Swap) in tryDecrementIfPositive to safely manage the destruction of worker threads without holding heavy locks during the count update.

// include/HXLibs/container/ThreadPool.hpp:265-279
bool tryDecrementIfPositive(std::atomic_uint32_t& val) {
    uint32_t now = val.load(std::memory_order_relaxed);
    while (now > 0) {
        if (val.compare_exchange_weak(
            now, now - 1,
            std::memory_order_acquire,
            std::memory_order_relaxed
        )) {
            return true;
        }
    }
    return false;
}

Sources: include/HXLibs/container/ThreadPool.hpp:265-282

Worker Lifecycle

Workers are stored in a std::map keyed by std::thread::id. When a worker thread is signaled to stop, it pushes its own ID into _delIdQueue. The administrator thread then joins and erases these threads from the map to prevent resource leaks.
Sources: include/HXLibs/container/ThreadPool.hpp:236-248, include/HXLibs/container/ThreadPool.hpp:253-263

Summary

The "Concurrency & Thread-Safe Collections" module provides the foundation for asynchronous execution in HXLibs. Through the use of a dynamically scaling ThreadPool, thread-safe SafeQueue, and the FutureResult abstraction, the system ensures efficient task distribution and safe data sharing across multiple threads. Its native support for C++20 coroutines and functional chaining makes it suitable for complex network and I/O-bound applications.
Sources: include/HXLibs/container/ThreadPool.hpp, tests/execution/demo.cpp:24-38