ThreadPool

Extends Algorithm

Overview
Features
Thread Safety
Class Interface
Signals
Usage Example
How It Works
Thread Management
Task Ownership
Stopping Tasks
Logger Connection
Performance Considerations
Best Practices

The ThreadPool class is a specialized Algorithm that manages a collection of Runnable tasks and executes them in parallel across multiple threads. It provides an efficient way to distribute workloads across multiple processor cores, maximizing resource utilization while maintaining a clean interface for monitoring execution progress and managing task lifecycles.

This thread-safe implementation ensures that parallel tasks can be safely executed, monitored, and controlled in multi-threaded applications, providing a robust foundation for high-performance concurrent programming.

Features

Parallel Execution: Runs multiple tasks concurrently across available CPU cores
Thread-Safe Task Management: Provides methods to safely add, create, and manage tasks
Concurrent Execution Control: Allows stopping all running tasks at once with proper synchronization
Dynamic Task Creation: Template method for creating and adding tasks in one step
Hardware Awareness: Automatically detects optimal thread count for the system
Thread-Safe Signal Forwarding: Centralizes logging and monitoring from all managed tasks
Execution Statistics: Reports information about execution time and task performance
Scalability: Efficiently distributes work across available processing resources

Thread Safety

The ThreadPool class provides comprehensive thread safety with the following guarantees:

Task Management Thread Safety: Adding and accessing tasks is protected from concurrent access
Concurrent Execution: Tasks run truly in parallel on separate threads with proper synchronization
Futures-Based Synchronization: Uses std::future for safe thread coordination and completion detection
Safe Task Stopping: All tasks can be safely requested to stop from any thread
Signal Thread Safety: Signal emissions follow thread-safe patterns using the thread-safe SignalSlot system
Progress Tracking: Thread-safe reporting of overall execution progress
Exception Safety: Exceptions in one task won't affect other tasks or the pool itself
Thread Ownership: Clear ownership and lifecycle management of created threads
Resource Cleanup: Guaranteed proper cleanup of all threads and resources

Class Interface

class ThreadPool : public Algorithm {
public:
    // Constructor & destructor
    explicit ThreadPool(bool verbose = true);
    ~ThreadPool() = default;
    
    // Task management
    void add(std::unique_ptr runnable);
    
    template 
    T* createAndAdd(Args&&... args);
    
    // Information methods
    unsigned int size() const;
    static unsigned int maxThreadCount();
    
    // Execute all tasks in the pool
    void exec(const ArgumentPack &args = {}) override;
    
    // Utility methods
    void connectLoggerToAll(Task* logger);
    void stopAll();
    
protected:
    // Access to managed runnables
    const std::vector>& runnables() const;
    
private:
    std::vector> m_runnables;
    bool m_verbose;
};

Signals

ThreadPool emits the following signals:

Signal	Type	Description	Arguments
`started`	Simple	Indicates pool execution has started	None
`finished`	Simple	Indicates pool execution has completed	None
`progress`	Data	Reports overall execution progress	float (0.0 to 1.0)
`stats`	Data	Reports execution statistics	int64_t (execution time in ms), unsigned int (number of tasks)
`log`, `warn`, `error`	Data	Forwarded log messages from tasks	std::string (message)

Usage Example

Basic Thread Pool Usage

// Create a thread pool
auto pool = std::make_shared();

// Create a logger and connect it to the pool
auto logger = std::make_shared("PoolLogger");
logger->connectAllSignalsTo(pool.get());

// Create and add tasks using the template method
auto task1 = pool->createAndAdd("Task1", 1000);
auto task2 = pool->createAndAdd("Task2", 2000);
auto task3 = pool->createAndAdd("Task3", 1500);

// Alternatively, create and add tasks separately
auto task4 = std::make_unique("Task4", 3000);
pool->add(std::move(task4));

// Connect to statistics signal
pool->connectData("stats", [](const ArgumentPack& args) {
    int64_t executionTime = args.get(0);
    unsigned int taskCount = args.get(1);
    
    std::cout << "Executed " << taskCount << " tasks in " 
              << executionTime << " ms ("
              << static_cast(executionTime) / taskCount 
              << " ms per task average)" << std::endl;
});

// Execute all tasks in parallel
pool->exec();

// Alternative: execute asynchronously
auto future = pool->run();

// Do other work while tasks are running

// Check if we need to stop tasks
if (userRequestedStop) {
    pool->stopAll();
}

// Wait for completion
future.wait();

Thread-Safe Multi-Pool Coordination

// Create multiple thread pools for different workload types
auto cpuPool = std::make_shared();
auto ioPool = std::make_shared();

// CPU-bound tasks
for (int i = 0; i < 4; i++) {
    cpuPool->createAndAdd("Compute-" + std::to_string(i), data[i]);
}

// I/O-bound tasks
for (int i = 0; i < 8; i++) {
    ioPool->createAndAdd("IO-" + std::to_string(i), files[i]);
}

// Start both pools asynchronously from the main thread
auto cpuFuture = cpuPool->run();
auto ioFuture = ioPool->run();

// Monitor progress from another thread
std::thread monitorThread([cpuPool, ioPool]() {
    while (cpuPool->isRunning() || ioPool->isRunning()) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        
        // Thread-safely check status and update UI
        std::cout << "CPU tasks running: " << (cpuPool->isRunning() ? "Yes" : "No") << std::endl;
        std::cout << "I/O tasks running: " << (ioPool->isRunning() ? "Yes" : "No") << std::endl;
    }
});

// Wait for both pools to complete their work
cpuFuture.wait();
ioFuture.wait();

// Join the monitor thread
monitorThread.join();

std::cout << "All work completed" << std::endl;

How It Works

The ThreadPool operates through the following thread-safe mechanisms:

Task Collection:
- Tasks are added to the pool via add() or createAndAdd<T>()
- All tasks must inherit from Runnable to provide a consistent interface
- Tasks are stored in a synchronized data structure for thread safety
Parallel Execution:
- When exec() is called, the pool starts all tasks using std::async
- Each task runs in its own thread, with the system managing thread distribution
- The pool uses futures to track task completion and handle exceptions
- Thread communication happens through futures and atomic operations
Progress Tracking:
- Overall progress is calculated as tasks complete
- Progress updates are synchronized to prevent data races
- Each completed task contributes equally to overall progress
Signal Forwarding:
- Log messages from tasks are forwarded to pool listeners through thread-safe signal mechanism
- This allows centralized logging and monitoring across all threads
Task Management:
- The pool maintains ownership of tasks via unique pointers with clear ownership semantics
- Tasks remain valid until the pool is destroyed
- Pool controls task execution and can halt all tasks if needed

Thread-Safe Execution Implementation

void ThreadPool::exec(const ArgumentPack &args) {
    if (m_runnables.empty()) {
        emitString("warn", "ThreadPool is empty, nothing to execute");
        return;
    }

    std::vector> futures;
    auto start = std::chrono::steady_clock::now();

    unsigned int taskCount = m_runnables.size();

    // Report starting information
    std::stringstream ss;
    ss << "Starting execution of " << taskCount << " tasks";
    emitString("log", ss.str());

    // Start progress reporting
    float progressStep = 1.0f / static_cast(taskCount);
    reportProgress(0.0f);

    // Launch tasks in parallel with async - each task gets its own thread
    for (const auto &runnable : m_runnables) {
        futures.push_back(
            std::async(std::launch::async, &Runnable::run, runnable.get()));
    }

    // Wait for all tasks to complete
    for (size_t i = 0; i < futures.size(); ++i) {
        futures[i].get();
        // Update progress after each task completes
        reportProgress((i + 1) * progressStep);
    }

    auto end = std::chrono::steady_clock::now();
    auto diffMs =
        std::chrono::duration_cast(end - start)
            .count();

    // Create stats information
    ArgumentPack statsArgs;
    statsArgs.add(diffMs);         // Execution time in ms
    statsArgs.add(taskCount); // Number of executed tasks
    emit("stats", statsArgs);

    if (m_verbose) {
        std::stringstream ss;
        ss << "Executed " << taskCount << " tasks in " << diffMs << " ms ("
           << static_cast(diffMs) / taskCount
           << " ms per task average)";
        emitString("log", ss.str());
    }
}

Thread Management

The ThreadPool handles thread creation and management automatically with these key features:

Thread Count Detection: The maxThreadCount() static method returns the number of available hardware threads
Thread Creation: Uses std::async with std::launch::async policy for automatic thread management
Thread Safety: All tasks are automatically thread-confined when executing
Resource Management: Threads are properly joined when tasks complete or when stopped
Work Distribution: Tasks are evenly distributed across available processor cores
Thread Isolation: Exceptions in one thread don't affect other threads

Automatic Hardware Detection: The ThreadPool automatically adapts to the hardware it's running on by using std::thread::hardware_concurrency() to determine the optimal number of threads. This ensures efficient execution across different systems.

Task Ownership

The ThreadPool takes ownership of tasks with clear ownership semantics:

Tasks are stored as std::unique_ptr<Runnable> in the pool
The createAndAdd<T>() method returns a raw pointer for convenience
Tasks are automatically destroyed when the pool is destroyed
Do not delete raw pointers returned by createAndAdd<T>()
Thread safety is maintained during task addition and execution

Warning: Raw pointers returned by createAndAdd<T>() are non-owning pointers. They're valid as long as the ThreadPool exists, but should never be deleted manually.

Thread-Safe Task Stopping

The ThreadPool provides a coordinated way to safely stop all running tasks:

Stopping All Tasks

// Stop all tasks in the pool
pool->stopAll();

// Implementation of stopAll() - Thread-safe cancellation
void ThreadPool::stopAll() {
    for (const auto &runnable : m_runnables) {
        if (runnable->isRunning()) {
            runnable->requestStop();
        }
    }
    emitString("log", "Stop requested for all running tasks");
}

This calls requestStop() on all runnables in the pool. Tasks should check stopRequested() regularly and exit gracefully when requested.

            Thread Safety Guarantee: The stopAll() method can be safely called from any
            thread, even while tasks are running in parallel threads. The stop requests are made using atomic operations
            that allow for safe communication between threads.
        

Thread-Safe Logger Connection

The connectLoggerToAll() method provides a convenient way to connect a logger to all tasks in the pool with thread safety:

Thread-Safe Logger Connection

// Connect a logger to all tasks
auto logger = std::make_shared();
pool->connectLoggerToAll(logger.get());

// Implementation
void ThreadPool::connectLoggerToAll(Task *logger) {
    if (!logger)
        return;

    // Use lambda functions to properly forward signals to the logger
    // Thread-safely connect the logger to the ThreadPool
    connectData("log", [logger](const ArgumentPack &args) {
        logger->emit("log", args);
    });
    connectData("warn", [logger](const ArgumentPack &args) {
        logger->emit("warn", args);
    });
    connectData("error", [logger](const ArgumentPack &args) {
        logger->emit("error", args);
    });

    // Thread-safely connect the logger to each Runnable
    for (const auto &runnable : m_runnables) {
        runnable->connectData("log", [logger](const ArgumentPack &args) {
            logger->emit("log", args);
        });
        runnable->connectData("warn", [logger](const ArgumentPack &args) {
            logger->emit("warn", args);
        });
        runnable->connectData("error", [logger](const ArgumentPack &args) {
            logger->emit("error", args);
        });
    }
}

This ensures that log messages from all threads and tasks are properly captured and processed in a thread-safe manner.

Performance Considerations

Task Granularity: For optimal performance, tasks should be sufficiently large to offset the overhead of thread creation
CPU vs. I/O: The pool works best with CPU-bound tasks; I/O-bound tasks might benefit from a larger pool size
Thread Count: Using more threads than CPU cores typically does not improve performance for CPU-bound tasks
Task Dependencies: Tasks should be independent; dependent tasks should be managed separately
Synchronization Overhead: The thread-safe implementation adds minimal overhead thanks to careful use of atomic operations and futures
Memory Requirements: Each thread requires its own stack space, which should be considered for systems with limited memory
Scalability: The pool's performance scales well with the number of available processor cores

Thread Safety Implementation

            Key Thread Safety Features
            Future-Based Synchronization: Uses std::future for thread coordination and
                    exception handling
Thread-Safe Signal System: All signal emissions leverage the thread-safe SignalSlot
                    implementation
Atomic Progress Tracking: Progress updates are thread-safe
Task Ownership Model: Clear ownership semantics with unique pointers
Exception Containment: Exceptions in one task don't affect other tasks
Thread Confinement: Each task runs in its own confined thread
Properly Synchronized Access: Task collection access is properly synchronized
Lambda Capture Safety: Careful use of lambdas with properly captured state for
                    thread safety

        

Thread-Safe Best Practices

Task Independence: Design tasks to be independent and avoid shared state to minimize synchronization needs
Error Handling: Implement proper error handling in tasks to prevent pool execution failures
Resource Management: Ensure tasks properly manage resources even when stopped
Progress Reporting: Implement progress reporting in tasks for accurate pool progress tracking
Appropriate Sizing: Create an appropriate number of tasks based on workload characteristics and hardware capabilities
Task Granularity: Balance task size to minimize threading overhead while maximizing parallelism
Handler Thread Safety: Ensure signal handlers are thread-safe as they may be invoked from multiple task threads
Cancel Points: Implement regular cancellation checking in long-running tasks
Thread-Local Resources: Use thread-local storage for resources that should not be shared across tasks
Deterministic Resource Cleanup: Implement RAII for resources to ensure cleanup on early termination

Advanced Thread Safety Patterns

Thread-Safe Task Communication

If tasks need to communicate results, use thread-safe mechanisms such as:

Thread-Safe Result Aggregation

// Thread-safe result collector
class ResultCollector : public Task {
public:
    void addResult(const Result& result) {
        std::lock_guard lock(m_mutex);
        m_results.push_back(result);
        
        // Emit notification about the new result
        ArgumentPack args;
        args.add(m_results.size());
        emit("resultAdded", args);
    }
    
    std::vector getResults() const {
        std::lock_guard lock(m_mutex);
        return m_results;
    }
    
private:
    mutable std::mutex m_mutex;
    std::vector m_results;
};

// Using with ThreadPool
auto collector = std::make_shared();

// Create tasks that use the collector
for (int i = 0; i < 10; i++) {
    auto task = pool->createAndAdd(data[i], collector);
}

Optimal Thread Count Management

For systems with varying workloads, dynamically adjusting the thread count can improve efficiency:

Adaptive Thread Count

// Create task batches based on system capabilities
unsigned int optimalTaskCount = ThreadPool::maxThreadCount();

// For CPU-intensive tasks, match the number of cores
if (taskType == TaskType::CPU_INTENSIVE) {
    optimalTaskCount = ThreadPool::maxThreadCount();
} 
// For I/O-bound tasks, use more threads as they'll often be waiting
else if (taskType == TaskType::IO_BOUND) {
    optimalTaskCount = ThreadPool::maxThreadCount() * 2;
}

// Create the appropriate number of tasks
for (int i = 0; i < optimalTaskCount; i++) {
    pool->createAndAdd(workBatches[i]);
}

Work Stealing Pattern

For more complex workloads, implement a work stealing pattern for better load balancing:

Thread-Safe Work Queue

// Create a thread-safe work queue
class WorkQueue : public Task {
public:
    void addWork(const WorkItem& item) {
        std::lock_guard lock(m_mutex);
        m_queue.push(item);
        
        // Signal new work available
        emit("workAdded");
    }
    
    std::optional getNextWork() {
        std::lock_guard lock(m_mutex);
        if (m_queue.empty()) {
            return std::nullopt;
        }
        
        WorkItem item = m_queue.front();
        m_queue.pop();
        return item;
    }
    
    bool hasWork() const {
        std::lock_guard lock(m_mutex);
        return !m_queue.empty();
    }
    
private:
    mutable std::mutex m_mutex;
    std::queue m_queue;
};

// Worker task that processes items until queue is empty
class WorkerTask : public Runnable {
public:
    WorkerTask(std::shared_ptr queue) : m_queue(queue) {}
    
protected:
    void runImpl() override {
        while (!stopRequested()) {
            // Try to get work
            auto workItem = m_queue->getNextWork();
            if (!workItem) {
                // No more work
                break;
            }
            
            // Process the work item
            processItem(*workItem);
        }
    }
    
private:
    std::shared_ptr m_queue;
};

// Using the work queue with ThreadPool
auto workQueue = std::make_shared();

// Fill the queue with work
for (const auto& item : workItems) {
    workQueue->addWork(item);
}

// Create worker tasks in the pool
unsigned int workerCount = ThreadPool::maxThreadCount();
for (unsigned int i = 0; i < workerCount; i++) {
    pool->createAndAdd(workQueue);
}

// Execute all workers in parallel
pool->exec();

Integration with Other Components

ThreadPool works well with other task library components in a thread-safe manner:

Chronometer: Measure detailed execution time of the pool and individual tasks
Logger: Centralize logging from all tasks in the pool with thread-safe message handling
TaskObserver: Monitor performance metrics across the pool with thread-safe statistics
ProgressMonitor: Track overall progress of complex operations across multiple threads
ParallelAlgorithm: Combine with ThreadPool for nested parallelism with hierarchical task organization
Counter: Track task completions or other metrics in a thread-safe manner

Summary

The ThreadPool class provides a robust, thread-safe foundation for parallel task execution in high-performance applications. Its careful design ensures safe execution, cancellation, and monitoring across multiple threads, without imposing excessive synchronization overhead on task implementations.

By leveraging modern C++ concurrency primitives like futures, atomic operations, and proper thread management, the ThreadPool enables developers to harness the full power of multi-core processors while maintaining code clarity and safety. The integration with the SignalSlot system provides comprehensive monitoring capabilities to track execution progress and performance across all parallel tasks.

ThreadPool

Contents

Features

Thread Safety

Class Interface

Signals

Usage Example

How It Works

Thread Management

Task Ownership

Thread-Safe Task Stopping

Thread-Safe Logger Connection

Performance Considerations

Thread Safety Implementation

Key Thread Safety Features

Thread-Safe Best Practices

Advanced Thread Safety Patterns

Thread-Safe Task Communication

Optimal Thread Count Management

Work Stealing Pattern

Integration with Other Components

Summary