Dynamic Thread Pool Sizing

The Strategy: Monitor, Decide, Act

Implementing dynamic sizing isn't just about adding and removing threads; it requires a control loop. The general strategy can be broken down into three phases:

1. Monitor: We need to collect metrics to understand the pool's current state. Is it overwhelmed, underutilized, or just right?
2. Decide: Based on these metrics, a policy or heuristic determines if a change in the number of threads is needed.
3. Act: The system executes the decision, either by creating new worker threads or by gracefully shutting down surplus ones.

To explore what kind of metrics are useful, let's turn to a research paper on high-performance thread pools.

As the paper suggests, key indicators for our scaling policy include:

• The number of pending tasks in our queue. A long queue is a sign of being under-provisioned.
• The number of idle threads. A large number of consistently idle threads suggests we are over-provisioned.

A simple policy could be:

• Scale-Up Condition: If tasks_in_queue > 0 and idle_threads == 0, we are at capacity and tasks are waiting. It's a good time to add a worker thread.
• Scale-Down Condition: If idle_threads > 1 for a sustained period, we have spare capacity and can probably remove a worker to save resources.

Now, let's get into the implementation details.

Part 1: Making Worker Threads Responsive

To remove a thread, we need a way to tell it to stop. As we've established, std::jthread and its std::stop_token are perfect for this. However, there's a problem: our worker threads are likely blocked on queue.wait_and_pop(). A stop request won't be noticed until a new task arrives and unblocks the thread.

We need to make our wait interruptible. C++20 provides a powerful tool for this: the wait overload on std::condition_variable_any that accepts a std::stop_token.

Let's modify our ThreadSafeQueue to include a new, interruptible pop function.

// In your ThreadSafeQueue class...
private:
    // We need std::condition_variable_any to use stop_tokens
    mutable std::condition_variable_any cv; 
    // The regular std::mutex is fine.

public:
    // Returns true if a value was popped, false if interrupted by the stop_token.
    bool wait_and_pop(T& value, std::stop_token stoken) {
        std::unique_lock lock(mut);
        
        // The magic happens here. The wait will be interrupted if either:
        // 1. The condition (!queue.empty()) is met.
        // 2. The stop_token (stoken) is requested.
        if (!cv.wait(lock, stoken, [this]{ return !queue.empty(); })) {
            // The wait was cancelled by the stop_token before the condition was met.
            return false;
        }

        value = std::move(queue.front());
        queue.pop();
        return true;
    }

Now, our worker thread's main loop can be updated to use this new function. The worker can be gracefully shut down even while it's waiting for a task.

// Inside the ThreadPool's worker function
void worker_loop(std::stop_token stoken) {
    while (!stoken.stop_requested()) {
        std::function<void()> task;
        // Use the new interruptible pop
        if (tasks.wait_and_pop(task, stoken)) {
            try {
                // We got a task, execute it
                task();
            } catch (...) {
                // In a future lesson, we'll handle exception propagation.
                // For now, just catch everything to prevent a thread from dying.
            }
        }
        // If wait_and_pop returned false, the stop was requested.
        // The loop condition will then terminate the thread.
    }
}

Part 2: Implementing the Resizing Logic

With our workers now responsive to stop requests, we can build the core resizing mechanism. A clean way to expose this is through a resize method in the ThreadPool class.

// In ThreadPool.h
class ThreadPool {
public:
    // ... constructor, submit, etc.
    void resize(size_t new_size);

private:
    std::vector<std::jthread> workers;
    size_t min_threads;
    size_t max_threads;
    // ... other members
};

// In ThreadPool.cpp
void ThreadPool::resize(size_t new_size) {
    // Ensure new_size is within configured bounds
    new_size = std::clamp(new_size, min_threads, max_threads);

    size_t current_size = workers.size();
    if (new_size == current_size) {
        return;
    }

    if (new_size > current_size) {
        // --- Scale Up ---
        workers.reserve(new_size);
        for (size_t i = 0; i < new_size - current_size; ++i) {
            workers.emplace_back([this](std::stop_token stoken){ this->worker_loop(stoken); });
        }
    } else {
        // --- Scale Down ---
        for (size_t i = 0; i < current_size - new_size; ++i) {
            // Signal the last thread in the vector to stop.
            // It will finish its current task (if any) and exit the loop.
            workers.back().request_stop();
            // Move the jthread to a temporary vector for joining.
            // This avoids blocking the resize operation for too long.
            threads_to_reap.push_back(std::move(workers.back()));
            workers.pop_back();
        }
    }
}

The "Reaping" Problem:
When scaling down, we signal a thread to stop. Its std::jthread object must eventually be joined to release its resources. Simply calling join() inside resize would block the function until the worker finishes its current task, which could be long.

A better approach is to move the jthreads marked for termination into a separate container (threads_to_reap in the example). The destructors of these jthread objects will automatically request_stop() and join(). We can periodically clean this "reaping" container in a low-priority manner. For now, our simple resize function demonstrates the principle of signaling and removing threads from the active pool.

Part 3: The Manager Thread

Manually calling resize isn't very dynamic. The final piece is to create a manager thread that automates the "Monitor, Decide, Act" loop. This thread will run separately from the workers and will be responsible for adjusting the pool size.

Here is a skeleton for the manager's logic:

// In ThreadPool.h, add a jthread for the manager
private:
    std::jthread manager_thread;
    void manager_loop(std::stop_token stoken);

// In the ThreadPool constructor, start the manager
ThreadPool::ThreadPool(size_t min_thr, size_t max_thr)
    : min_threads(min_thr), max_threads(max_thr) 
{
    // Start initial threads...
    for(size_t i = 0; i < min_threads; ++i) { /* ... */ }

    // Start the manager thread
    manager_thread = std::jthread([this](std::stop_token stoken){ this->manager_loop(stoken); });
}

// And the manager's logic
void ThreadPool::manager_loop(std::stop_token stoken) {
    while (!stoken.stop_requested()) {
        // 1. Wait for a management interval or until stopped
        std::this_thread::sleep_for(std::chrono::seconds(1));
        if(stoken.stop_requested()) break;

        // For thread-safety, lock the pool while inspecting/resizing
        std::unique_lock lock(pool_mutex); // Assuming you have a mutex for pool state
        
        // 2. Monitor
        const size_t task_count = tasks.size();
        const size_t current_threads = workers.size();
        // NOTE: Determining `idle_threads` is non-trivial. It requires tracking
        // thread state, perhaps with an atomic flag each worker sets/unsets.
        // For now, let's use a simplified heuristic based on task count.

        // 3. Decide & 4. Act
        if (task_count > current_threads && current_threads < max_threads) {
            // If tasks are piling up, scale up by one.
            resize(current_threads + 1);
        } else if (task_count == 0 && current_threads > min_threads) {
            // If the queue is empty, we may be over-provisioned. Scale down.
            resize(current_threads - 1);
        }
        
        // Also, clean up any reaped threads...
    }
}

This manager provides autonomous, intelligent control over your thread pool's resources. The policy here is very simple; real-world pools often use more sophisticated algorithms (e.g., averaging metrics over time to avoid frantic scaling).

Conclusion

You have now designed and sketched out the implementation for one of the most advanced features of a modern thread pool. This demonstrates a strong grasp of concurrent architecture, resource management, and the practical application of C++20 concurrency features.

Key Takeaways:

• Dynamic thread pool sizing is a control loop problem: Monitor, Decide, Act.
• Effective monitoring relies on metrics like queue length and thread utilization.
• Scaling down requires cooperative cancellation. This can be implemented by making the workers' wait on the task queue interruptible using std::condition_variable_any and std::stop_token.
• A dedicated manager thread is a clean architectural pattern to automate the scaling logic, decoupling it from the main task-submission pathway.
• The resize method must handle both scaling up (adding jthreads) and scaling down (signaling and removing jthreads).

Next Lesson Preview:

Now that our thread pool's structure can change dynamically, we must ensure that other features remain robust. One of the most critical aspects of a tasking system is error handling. In the next lesson, we will tackle this by learning how to implement robust exception propagation from tasks back to the submitting thread, ensuring that a std::future can correctly rethrow any exception that occurred in the worker.