C++ Producer / Consumer multi threads pattern with maximum container size

Question

I've just written a Producer / Consumer pattern templatized that should work in multi threads environment (IE producer(s) and consumer(s) in separate threads).

The original use case is one producer and one consumer in separate thread with a limit of container size.

It should work for a multi producers / consumers case though.

[EDIT 1]: it appears to not working at all with multi consumers

Standard used is C++ 14

My question is about the correctness of the std::unique_lock<std::mutex> instructions and synchronizer.wait ones.

I have also a question about the forward declaration of templates, is there a way to avoid it without having a compiler error in the linking step ?

And last question about the handle of the container (the list), it must become the owner of the elements and handle its life time so what's the method signature I have to use (const T &e, T e, T &&e, ...) ? Or even std::move and implementation of perfect forwarding, I don't know.

ProducerConsumer.hpp

#ifndef CUDA_HARDWARE_VIDEO_READER_PRODUCERCONSUMER_HPP
#define CUDA_HARDWARE_VIDEO_READER_PRODUCERCONSUMER_HPP

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

template<typename T> class ProducerConsumer
{
    public:
        explicit ProducerConsumer(ulong maxSizeInBytes);

        void produce(const T &element);
        T    consume();
        void producerIsOver();
        bool isFinished();

    private:
        std::queue<T>           queue;
        std::condition_variable synchronizer;
        std::mutex              mutex;
        bool                    producerOver;
        ulong                   maxSizeInBytes;
};

#endif //CUDA_HARDWARE_VIDEO_READER_PRODUCERCONSUMER_HPP

ProducerConsumer.cpp

#include "ProducerConsumer.hpp"

/**
 * @brief Constructor
 *
 * @tparam T - The type of element's container
 * @param maxSizeInMB - The maximum size the container can have in bytes
 */
template<typename T> ProducerConsumer<T>::ProducerConsumer(ulong maxSizeInBytes): maxSizeInBytes(maxSizeInBytes), producerOver(false) { }

/**
 * @brief Producer - add element in the list
 *
 * @tparam T - The type of the element
 * @param element - The element to add
 */
template<typename T> void ProducerConsumer<T>::produce(const T &element)
{
    std::unique_lock<std::mutex> lock(mutex);

    // Block the thread until the size of the queue allows to add more elements in it
    if (sizeof(T) * (queue.size() + 1) > maxSizeInBytes) {
        synchronizer.wait(lock, [&]() {
            return sizeof(T) * (queue.size() + 1) <= maxSizeInBytes;
        });
    }

    queue.push(element);
    synchronizer.notify_one();
}

/**
 * @brief Consumer - Get an element of the list and wait for it if the list is empty
 *
 * @tparam T - The type of the element
 *
 * @return The element
 */
template<typename T> T ProducerConsumer<T>::consume()
{
    std::unique_lock<std::mutex> lock(mutex);

    // Block the thread until the queue is empty and the producer has finished to fill the queue
    if (queue.empty() && !producerOver) {
        synchronizer.wait(lock, [&]() {
            return !queue.empty() || producerOver;
        });
    }

    T element = queue.front();

    queue.pop();
    // Wake up the producer if he's waiting
    synchronizer.notify_all();

    return element;
}

/**
 * @brief Tells the producer has finished to fill the container
 *
 * @tparam T - The type of element's container
 */
template<typename T> void ProducerConsumer<T>::producerIsOver()
{
    // @todo Useless ?
    std::unique_lock<std::mutex> lock(mutex);

    producerOver = true;
}

/**
 * @brief Tells if the whole process is over (IE producer has finished to fill the container and the container is empty)
 *
 * @tparam T - The type of element's container
 *
 * @return True if the whole process is over, false otherwise
 */
template<typename T> bool ProducerConsumer<T>::isFinished()
{
    std::unique_lock<std::mutex> lock(mutex);

    return producerOver && queue.empty();
}

// @todo Annoying forward declaration there ...
template class ProducerConsumer<long double>;

main.cpp

#include <iostream>
#include <random>
#include <ctime>
#include <unistd.h>
#include "ProducerConsumer.hpp"

/**
 * @brief Test the producer / consumer pattern that preserves order and is limited in size
 *        Required output is :
 *        Consume 1,
 *        Consume 2,
 *        Consume 3,
 *        ...,
 *        Consume 20
 *
 * @return EXIT_SUCCESS if no error occurred
 */
int main()
{
    ProducerConsumer<long double>           container(sizeof(long double));
    std::random_device                      randomDevice;
    std::mt19937                            randomGenerator(randomDevice());
    std::uniform_int_distribution<uint32_t> distribution(10000, 1000000);

    randomGenerator.seed(std::time(nullptr));

    std::thread producer([&]() {
        for (int i = 1; i < 21; ++i) {
            usleep(distribution(randomGenerator));
            container.produce(i);
        }

        container.producerIsOver();
    });

    std::thread consumer([&]() {
        do {
            usleep(distribution(randomGenerator));
            std::cout << "Consume " << container.consume() << std::endl;
        } while (!container.isFinished());
    });

    producer.join();
    consumer.join();

    return EXIT_SUCCESS;
}