A Simple BlockingQueue implementation in C++

Question

I'm just dusting off my C++ knowledge in area of multithreading. I started with implementing a producer-consumer pattern inspired by https://jenkov.com/tutorials/java-util-concurrent/blockingqueue.html.

Here's the code:

#include <atomic>
#include <chrono>
#include <cstdint>
#include <deque>
#include <mutex>
#include <thread>
#include <iostream>
#include <condition_variable>
#include <random>

using namespace std::chrono_literals;
using engine = std::mt19937;

template <typename Task>
class BlockingQueue
{
public:
    BlockingQueue() = default;

    void put(Task t) {
        {
            std::unique_lock lock{mtx};
            canPut.wait(lock, [this]{return tasks.size() < sizeLimit; });
            tasks.push_front(t);
            std::cout << "Tasks: " << tasks.size() << "\n";
        }
        canTake.notify_one();
    }

    Task take() {
        std::unique_lock lock{mtx};
        canTake.wait(lock, [this]{ return !tasks.empty(); });
        Task taken {std::move(tasks.back())};
        tasks.pop_back();
        // std::cout << "Task: " << taken << " taken by thread: " << std::this_thread::get_id() << "\n";
        std::cout << "Tasks: " << tasks.size() << "\n";
        return taken;
    }

private:
    std::mutex mtx;
    std::condition_variable canPut;
    std::condition_variable canTake;
    int sizeLimit {50};
    std::deque<Task> tasks;
};


int main() {

    std::random_device os_seed;
    const auto seed = os_seed();
    engine generator{seed};
    std::uniform_int_distribution<uint32_t> distribute(0, 10);

    BlockingQueue<int> q;

    std::atomic_int i = 0;
    std::jthread prod {
        [&]{
            while(true){
                q.put(i++);
                auto sleepTime = 
                    std::chrono::seconds(distribute(generator));
                std::this_thread::sleep_for(sleepTime);
            }
        }
    };
    std::jthread prod2 {
        [&]{
            while(true){
                q.put(i++);
                auto sleepTime = 
                    std::chrono::seconds(distribute(generator));
                std::this_thread::sleep_for(sleepTime);
            }
        }
    };
    std::jthread consumer {
        [&]{
            while(true){
                auto task = q.take();
                auto sleepTime = 
                    std::chrono::seconds(distribute(generator));
                std::this_thread::sleep_for(sleepTime);
            }
    }
    };
    return 0;
}

Answer 1

This is quite a decent implementation, you did the locking part correctly, have already thought a bit about efficiency by moving tasks in take(), you used C++20's std::jthread, and you used C++'s random number generator facilities. There are still some issues though:

Tasks are copied unnecessarily in put()

Up to two copies are made of a task in put(): the first time possibly when the function is called, since t is passed by value. The second time is when the task is pushed to tasks.

There are various ways the caller might want to pass the task. It could be by value, by r-value or by (const) l-value reference. Or they might even want to "emplace" it. While you could try to handle all cases individually, consider creating a single, templated function that perfectly forwards all the arguments:

template<typename... Args>
void put(Args&&... args) {
    …
    tasks.emplace_front(std::forward<Args>(args)...);
    …
}

Make the size limit configurable

You have hardcoded the size limit to 50. Make the size limit a parameter that is passed to the constructor.

Only one condition variable is needed

It is extremely unlikely that you have both threads waiting for tasks to become non-full and threads waiting for it to become non-empty, so there is no need to have two condition variables to notify different kinds of waiters.

Add a way to signal consumers that no more items will be produced

While in your code both producers and consumers will work forever, in many actual use cases the producers will only produce a certain number of items, and you want the consumer threads to exit when all items have been consumed. Ideally you add a way to signal that no more items will be produced, and ensure take() has a way to indicate in some way that no more items are forthcoming, for example by using a std::optional<Task> as the return value.