Learn Thread-Safe Queues | Synchronization and Communication

Thread-safe data structures are vital in concurrent programming because multiple threads may try to access or modify shared data simultaneously. Without proper synchronization, this can cause corrupted data, unpredictable results, or even crashes.

A thread-safe queue allows threads to safely add and remove items without interfering with each other. To achieve this, the underlying std::queue must be protected using synchronization tools like std::mutex for exclusive access and std::condition_variable to coordinate waiting and notifications between threads.

main.cpp


              12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
            
#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <optional>

template <typename T>
class ThreadSafeQueue {
public:
    void push(const T& value)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        queue_.push(value);
        cv_.notify_one();
    }

    std::optional<T> pop()
    {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_.wait(lock, [this]{ return !queue_.empty(); });
        T value = queue_.front();
        queue_.pop();
        return value;
    }

    bool empty() const
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return queue_.empty();
    }
private:
    mutable std::mutex mtx_;
    std::queue<T> queue_;
    std::condition_variable cv_;
};

void producer(ThreadSafeQueue<int>& q)
{
    for (int i = 0; i < 5; ++i)
    {
        q.push(i);
        std::cout << "Produced: " << i << std::endl;
    }
}

void consumer(ThreadSafeQueue<int>& q)
{
    for (int i = 0; i < 5; ++i)
    {
        auto value = q.pop();
        if (value) 
            std::cout << "Consumed: " << *value << std::endl;
    }
}

int main()
{
    ThreadSafeQueue<int> queue;
    std::thread t1(producer, std::ref(queue));
    std::thread t2(consumer, std::ref(queue));
    t1.join();
    t2.join();
}

Everything was clear?

Thanks for your feedback!

Section 2. Chapter 3

Ask AI

Ask anything or try one of the suggested questions to begin our chat

Swipe to show menu

main.cpp


              12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
            
#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <optional>

template <typename T>
class ThreadSafeQueue {
public:
    void push(const T& value)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        queue_.push(value);
        cv_.notify_one();
    }

    std::optional<T> pop()
    {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_.wait(lock, [this]{ return !queue_.empty(); });
        T value = queue_.front();
        queue_.pop();
        return value;
    }

    bool empty() const
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return queue_.empty();
    }
private:
    mutable std::mutex mtx_;
    std::queue<T> queue_;
    std::condition_variable cv_;
};

void producer(ThreadSafeQueue<int>& q)
{
    for (int i = 0; i < 5; ++i)
    {
        q.push(i);
        std::cout << "Produced: " << i << std::endl;
    }
}

void consumer(ThreadSafeQueue<int>& q)
{
    for (int i = 0; i < 5; ++i)
    {
        auto value = q.pop();
        if (value) 
            std::cout << "Consumed: " << *value << std::endl;
    }
}

int main()
{
    ThreadSafeQueue<int> queue;
    std::thread t1(producer, std::ref(queue));
    std::thread t2(consumer, std::ref(queue));
    t1.join();
    t2.join();
}

Everything was clear?

Thanks for your feedback!

Section 2. Chapter 3