Task Queues in C++: Implementation, Application, and Optimization

Working with Task Queues in C++: From Simple to Complex

In modern C++ development, asynchronous data processing is rarely avoided. Task queues are one of the key tools for organizing parallel code execution. They allow distributing work between threads, avoiding locks, and efficiently utilizing CPU resources. In this article, we will analyze how to implement a task queue in C++, examine the producer-consumer pattern, and learn how to build a simple thread pool.

1. Basic Queue Based on std::queue

The simplest task queue is a container where elements (tasks) can be added and retrieved in FIFO order. In C++, std::queue from the standard library is perfectly suited for this. However, when working with multithreading, a regular queue is unsafe: simultaneous access from different threads leads to data races.

Let's look at a basic example of a synchronized queue using a mutex and a condition variable:

#include <queue>#include <mutex>#include <condition_variable>#include <functional>

class TaskQueue {public:    using Task = std::function<void()>;

    void push(Task task) {        std::lock_guard<std::mutex> lock(mutex_);        queue_.push(std::move(task));        cond_var_.notify_one();    }

    Task pop() {        std::unique_lock<std::mutex> lock(mutex_);        cond_var_.wait(lock, [this]() { return !queue_.empty(); });        Task task = std::move(queue_.front());        queue_.pop();        return task;    }

    bool empty() const {        std::lock_guard<std::mutex> lock(mutex_);        return queue_.empty();    }

private:    std::queue<Task> queue_;    mutable std::mutex mutex_;    std::condition_variable cond_var_;};

In this code, push() adds a task and notifies one waiting thread. pop() blocks the thread until the queue becomes non-empty. This approach solves the synchronization problem but does not scale to multiple consumers without modifications.

2. Producer-Consumer Pattern

The classic producer-consumer problem fits perfectly with a task queue. One or more producer threads generate tasks, while consumer threads execute them. It is important to properly manage the queue size to avoid overflow or idle time.

Example with multiple consumers:

#include <thread>#include <vector>#include <iostream>

void worker(TaskQueue& queue, int id) {    while (true) {        auto task = queue.pop();        if (!task) break;  // termination signal        std::cout << "Worker " << id << " executing task";        task();    }}

int main() {    TaskQueue queue;    std::vector<std::thread> workers;

    // Launch 4 consumer threads    for (int i = 0; i < 4; ++i) {        workers.emplace_back(worker, std::ref(queue), i);    }

    // Producer adds tasks    for (int i = 0; i < 10; ++i) {        queue.push([i]() {            std::this_thread::sleep_for(std::chrono::milliseconds(100));            std::cout << "Task " << i << " done";        });    }

    // Termination signal (nullptr task)    for (auto& w : workers) {        queue.push(nullptr);    }

    for (auto& w : workers) {        w.join();    }    return 0;}

Here, each consumer receives a task, executes it, and waits for the next one. The termination signal (nullptr) stops the loop. This pattern is often used in server applications, event handlers, and game engines.

3. Thread Pool — A Pool of Threads for the Task Queue

A thread pool is an extension built on top of a task queue that manages