Simple C++ Deque Single Producer / Multiple Consumer Example

#include <cstdio>
#include <deque>
#include <condition_variable>
#include <mutex>
#include <memory>
#include <thread>
#include <algorithm>
#include <cassert>


template<typename T>
struct queue
{
    typedef std::deque<T> raw_queue_t;

    raw_queue_t m_queue;
    std::condition_variable m_cond;
    std::mutex m_mutex;

    void push(T const& obj)
    {
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_queue.push_back(obj);
        }

        m_cond.notify_one();
    }

    T pop()
    {
        T front;

        {
            std::unique_lock<std::mutex> lock(m_mutex);
            while (! m_queue.size()) m_cond.wait(lock);
            front = m_queue.front();
            m_queue.pop_front();
        }

        return front;
    }
};


typedef queue<unsigned int> string_queue_t;
#define CONSUMERS 5
#define N 1000


void producer(std::mutex& std_out_mtx, string_queue_t& queue)
{
    {
        std::unique_lock<std::mutex> lock(std_out_mtx);
        std::printf("producer::queue::(%p) - enter\n", (void*)&queue);
    }

    for (unsigned int i = 0; i < N; ++i)
    {
        queue.push(i + 1);
        std::this_thread::yield(); // just for some spice
    }

    for (unsigned int i = 0; i < CONSUMERS; ++i)
    {
        queue.push(0);
        std::this_thread::yield(); // just for some spice
    }

    {
        std::unique_lock<std::mutex> lock(std_out_mtx);
        std::printf("producer::queue::(%p) - exit\n", (void*)&queue);
    }
}


void consumer(unsigned int id, std::mutex& std_out_mtx, string_queue_t& queue)
{
    {
        std::unique_lock<std::mutex> lock(std_out_mtx);
        std::printf("consumer(%u)::queue::(%p) - enter\n", id, (void*)&queue);
    }

    unsigned int prev = 0;

    for (;;)
    {
        unsigned int msg = queue.pop();

        {
            std::unique_lock<std::mutex> lock(std_out_mtx);
            std::printf("consumer(%u)::msg::(%u)\n", id, msg);
        }

        if (msg == 0) break;

        assert(msg > prev); // validate fifo nature

        prev = msg;
    }

    {
        std::unique_lock<std::mutex> lock(std_out_mtx);
        std::printf("consumer::queue::(%p) - exit\n", (void*)&queue);
    }
}


int main(void)
{
    {
        string_queue_t queue;
        std::mutex std_out_mutex;

        std::thread consumers[CONSUMERS];

        for (unsigned int i = 0; i < CONSUMERS; ++i)
        {
            consumers[i] = std::thread(
                consumer,
                i,
                std::ref(std_out_mutex),
                std::ref(queue)
            );
        }

        std::thread producer_thread(
            producer,
            std::ref(std_out_mutex),
            std::ref(queue)
        );

        producer_thread.join();

        for (unsigned int i = 0; i < CONSUMERS; ++i)
        {
            consumers[i].join();
        }
    }

    std::printf("\nComplete, hit <ENTER> to exit...\n");
    std::fflush(stdout);
    std::getchar();

    return 0;
}