2
\$\begingroup\$

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;
}

\$\endgroup\$
  • 1
    \$\begingroup\$ Welcome to Code Review! Your edit places the question on the edge of what is considered on-topic here, since it seems like the code is not working as intended. Against this background, pleace check if your code is ready to be reviewed in full here. \$\endgroup\$ – AlexV Apr 5 at 12:41
  • 1
    \$\begingroup\$ @Alex Hello, actually my code is working for the primary job I stated. You can execute the main and it is working as intended. I thought it would work for a larger scope but it is not. This point is "on topic" because: Questions about improving scalability are allowable, as long as your code works for small inputs. \$\endgroup\$ – Romain Laneuville Apr 5 at 14:04

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.