Task Queues for C++: Implementation and Usage

Task Queues for C++: From Theory to Practice

A task queue is one of the key patterns in C++ multithreaded programming. It allows efficient distribution of work among threads, avoiding data races and excessive resource consumption. In this article, we will explore how to create and use task queues in C++ with a focus on performance and safety.

Why Do You Need a Task Queue?

In modern applications, there is often a need to execute many small tasks in parallel. Directly creating threads for each task incurs overhead: thread creation takes time and memory. A task queue solves this problem through a thread pool:

• Thread reuse — threads are not destroyed after completing a task but are returned to the pool.
• Load balancing — tasks are distributed evenly.
• Resource management — limiting the number of simultaneously running threads.

Basic Implementation of a Task Queue in C++

Let's consider a simple but thread-safe task queue using std::queue and mutexes. For this, we use the C++11 standard library and later.

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

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

    explicit TaskQueue(size_t numThreads) : stop(false) {        for (size_t i = 0; i < numThreads; ++i) {            workers.emplace_back([this] { workerLoop(); });        }    }

    ~TaskQueue() {        {            std::unique_lock<std::mutex> lock(mutex);            stop = true;        }        condition.notify_all();        for (auto& worker : workers) {            if (worker.joinable()) {                worker.join();            }        }    }

    void enqueue(Task task) {        {            std::unique_lock<std::mutex> lock(mutex);            tasks.push(std::move(task));        }        condition.notify_one();    }

private:    void workerLoop() {        while (true) {            Task task;            {                std::unique_lock<std::mutex> lock(mutex);                condition.wait(lock, [this] { return stop || !tasks.empty(); });                if (stop && tasks.empty()) {                    return;                }                task = std::move(tasks.front());                tasks.pop();            }            task(); // Execute the task        }    }

    std::queue<Task> tasks;    std::mutex mutex;    std::condition_variable condition;    bool stop;    std::vector<std::thread> workers;};

Usage Example

Now let's show how to use our TaskQueue class for parallel execution of simple computations:

#include <iostream>

int main() {    TaskQueue tq(4); // Pool of 4 threads

    for (int i = 0; i < 10; ++i) {        tq.enqueue([i] {            std::cout << "Task " << i << " executed by thread "                      << std::this_thread::get_id() << std::endl;        });    }

    // The destructor of tq will wait for all tasks to complete    return 0;}

Advanced Features: Priority Queue

Sometimes tasks need to be processed in order of importance. For this, you can use std::priority_queue instead of a regular queue. Let's add priority support:

#include <queue>#include <functional>

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

    struct PrioritizedTask {        int priority;        Task t