1
\$\begingroup\$

This post is based on A multi-thread Producer Consumer, where a Consumer has multiple Producers (C++17). I am trying to build a Consumer that consumes data from multiple Producers in a thread-safe manner. I extended the code in such a way that it is now possible to have an n:m relationship (many Producers and many Consumers). I would appreciate your thoughts and criticism. I also want to note that I will probably use a boost version in the and, as suggested in the previous post. I still would like to know if I did this correctly.

Some notes:

A Producer will not live indefinitely. At some point, it is done and will signal this to the Buffer. If there is no more Producer producing, the Consumer will stop consuming and the program will exit. This synchronization is handled by the producer_sem.

I am assuming a buffer that can grow indefinitely. This is why I do not have an emptyCount sempathore (compare wiki).

I am using only a single Buffer this time, instead of one Buffer per Producer. I believe this scales better with an increasing number of Consumers and Producers.

The random delay in the threads is there to simulate delay in the real world and to see if I run into synchronization issues.

Some questions:

For the Semaphore I am not using atomics, but lock_guards, as advised in the previous post. Is this smart? Why should I not use atomics?

When calling Buffer::add and Buffer::pop, does it make a difference if I first do lock.unlock() and then cond_var.notify_all() vs. the other way around?

#include <memory>
#include <optional>
#include <atomic>
#include <chrono>
#include <cmath>
#include <cstdlib>
#include <iostream>
#include <memory>
#include <mutex>
#include <sstream>
#include <thread>
#include <vector>
#include <shared_mutex>

/**
 * RAII-style timer. 
 * Used only in main to measure performance
 */
class MyTimer
{
public:
    using clock = std::chrono::high_resolution_clock;

    MyTimer() : start(clock::now()) {}
    ~MyTimer()
    {
        auto duration = clock::now() - start;
        std::cout << "elapsed time was " << std::chrono::duration_cast<std::chrono::microseconds>(duration).count() << " (us)\n";
    }

private:
    clock::time_point start;
};

/**
 * Semaphore for coordination. Should I use lock_gards or atomics here?
 */
class Semaphore
{
public:
    Semaphore() = delete;
    Semaphore(int n) : m_(), n_(n) {}
    void up()
    {
        std::lock_guard<std::mutex> lg(m_);
        ++n_;
    }

    void down()
    {
        std::lock_guard<std::mutex> lg(m_);
        --n_;
    }

    bool greater_zero() const
    {
        std::lock_guard<std::mutex> lg(m_);
        return n_ > 0;
    }

private:
    mutable std::mutex m_;
    int n_;
};

class Buffer
{
public:
    Buffer(int producer_parallelism) : buff_sem(0), producer_sem(producer_parallelism), mu(), print_mu(), cond_var(), buffer_(){};
    Buffer() = delete;

    /**
     * Add an element to the buffer
     */
    void add(char c)
    {
        std::unique_lock<std::mutex> lock(mu);
        buffer_ << c;
        buff_sem.up();
        lock.unlock();
        cond_var.notify_all();
    }

    /**
     * Pop/get an element from the buffer. Return empty optional, if no value in queue
     */
    std::optional<char> pop()
    {
        std::unique_lock<std::mutex> lock(mu);
        // continue if there is data, or all producers are done
        cond_var.wait(lock, [this]() -> bool { return buff_sem.greater_zero() || !producer_sem.greater_zero(); });
        if (!producer_sem.greater_zero()) // return empty if all producers are done
        {
            return std::nullopt;
        }
        char c;
        buffer_ >> c;
        buff_sem.down();
        lock.unlock();
        cond_var.notify_all();
        return c;
    }

    /**
     * Indicate that one producer is finished 
     */
    void production_ended()
    {
        producer_sem.down();
        cond_var.notify_all(); // if we do not notify here, the consumer will get stuck
    }

    /**
     * Helper for synced printing
     */
    template <typename... Args>
    void print(Args... args) const
    {
        const std::lock_guard<std::mutex> lg(print_mu);
        (std::cout << ... << args);
    }

private:
    Semaphore buff_sem;
    Semaphore producer_sem;
    mutable std::mutex mu;                    // sync all except print operation
    mutable std::mutex print_mu;              // sync print operations
    mutable std::condition_variable cond_var; // sync access to underlying buffer

    std::stringstream buffer_; // a stream for sharing data
};

/**
 * A producer that produces a given number of items and shuts down afterwards.
 */
class Producer
{
public:
    Producer(std::shared_ptr<Buffer> buffer, const int limit, const int id) : buffer_(buffer), limit_(limit), id_(id) {}
    Producer() = delete;

    /**
     * produces random data.
     */
    void run()
    {
        // for simulating delay of the producer
        for (int count = 0; count < limit_; ++count)
        {
            static char const alphabet[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
            char upper_case_char = alphabet[(random() % (sizeof alphabet - 1))];
            buffer_->add(upper_case_char);
            std::stringstream strs;
            strs << "Produced: " << upper_case_char << ". Count at " << count << ". Producer was " << id_ << std::endl;
            buffer_->print(strs.str());
            std::this_thread::sleep_for(std::chrono::milliseconds(random() % 3));
        }
        buffer_->production_ended(); // signal to buffer that this producer is done
        return;
    }

private:
    std::shared_ptr<Buffer> buffer_; // buffer is shared between producer and consumer
    const int limit_;                // number of elements to produce
    const int id_;                   // id of producer
};

/**
 * A consumer that consumes as long as something is produced.
 */
class Consumer
{
public:
    Consumer(std::shared_ptr<Buffer> &buffer, const int parallelism, const int id) : buffer_(buffer), parallelism_(parallelism), id_(id){};
    Consumer() = delete;

    void run()
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(random() % 3));
        while (true)
        {
            auto c = buffer_->pop();
            if (!c)
            {
                break;
            }
            buffer_->print("Consumer ", id_, " consumed ", c.value(), '\n');
        }
    }

private:
    std::shared_ptr<Buffer> &buffer_; // a vector of shared buffers
    const unsigned int parallelism_;
    const int id_;
};

/**
 * A simple thread pool. You can add threads here and join the all. 
 */
class ThreadPool
{
public:
    ThreadPool() : threads_(new std::vector<std::thread *>()), is_finished_(false){};

    void add_thread(std::thread *t)
    {
        threads_->push_back(t);
    }

    void join_all()
    {
        for (auto it = threads_->begin(); it != threads_->end(); ++it)
        {
            (*it)->join();
        }
    }

private:
    std::vector<std::thread *> *threads_;
    bool is_finished_;
};

int main()
{
    {
        MyTimer mt;

        // constants for this "experiment"
        const int producer_parallelism = 5;
        const int consumer_parallelism = 3;
        const int produced_preaces_per_producer = 5;

        // one buffer and one threadPool for all threads
        std::shared_ptr<Buffer> buff = std::make_shared<Buffer>(producer_parallelism);
        ThreadPool tp;

        for (int i = 0; i < consumer_parallelism; ++i)
        {
            Consumer *c = new Consumer{buff, producer_parallelism, i};
            std::thread *consumer_thread = new std::thread(&Consumer::run, c);
            tp.add_thread(consumer_thread);
        }

        for (int i = 0; i < producer_parallelism; ++i)
        {
            Producer *p = new Producer{buff, produced_preaces_per_producer, i};
            std::thread *producer_thread = new std::thread(&Producer::run, p);
            tp.add_thread(producer_thread);
        }
        tp.join_all();
    }
    return 0;
}
\$\endgroup\$
  • \$\begingroup\$ I think I could remove a couple of the #include commands... \$\endgroup\$ – User12547645 Jan 18 at 14:14
  • \$\begingroup\$ Use of new in modern C++ is a definite code smell. \$\endgroup\$ – Martin York Jan 18 at 15:31
  • \$\begingroup\$ Thanks for the advice. Do you have a good article to read on that @MartinYork. If not, I will just google myself :D \$\endgroup\$ – User12547645 Jan 18 at 15:53
  • 1
    \$\begingroup\$ I will probably write a full review today. But for every call to new there must also be a matching call to delete. You have a lot of calls to new but zero calls to delete. Now you can wrap dynamic allocation in shared pointers. The one shared pointer you use (the buffer) does not even need to be allocated dynamically. \$\endgroup\$ – Martin York Jan 18 at 16:00
  • 1
    \$\begingroup\$ You should avoid dynamic allocation unless it is required. When it is required use a smart pointer (this avoids delete and new). Prefer automatic objects, pass by reference to allow other objects to use; pointers can be used but should be a last resort when non owning nullable references are needed (ie wait until you are an expert). \$\endgroup\$ – Martin York Jan 18 at 16:49
2
\$\begingroup\$

Overview

Normally a ThreadPool has a fixed number of threads. With a variable amount of work that is to be completed by the threads. You have gone the other way. You have as many threads as there is work. So your pool is not really a pool of workers it is more of a thread maintainer than a thread pool.

Threads are relatively expensive to create. The amount of a parallelism supported by the hardware is limited and fixed so there is no point in having more threads than the hardware physically supports.

So usually a thread pool is created is created with a fixed number of threads that matches the hardware limits. Then you add work (not threads) to the pool. Each thread then simply checks the work queue for work and executes that work. On completion of the work it checks a work queue to get more work.


Code Review

Nice comment. I usually complain about bad comments. But I actually don't mind this one so I though I should make a special effort to comment about it :-)

/**
 * RAII-style timer. 
 * Used only in main to measure performance
 */

Prefer to initialize one variable per line:

    Buffer(int producer_parallelism) : buff_sem(0), producer_sem(producer_parallelism), mu(), print_mu(), cond_var(), buffer_(){};

In all the rest of your code you only declare and initialize one variable per line. So why did you do all variables on one line here. The whole point os using a high level language is try and make it readable for humans. This si the opposite.

why not like this?

    Buffer(int producer_parallelism)
        : buff_sem(0)
        , producer_sem(producer_parallelism)
        , mu()
        , print_mu()
        , cond_var()
        , buffer_()
    {};

Now that it is easier to read. I would not bother being explicit with the ones that use a default constructor. So I would simplify to:

    Buffer(int producer_parallelism)
        : buff_sem(0)
        , producer_sem(producer_parallelism)
    {};

There is no need to delete the default constructor.

    Buffer() = delete;

If any other constructor is defined then the compiler will not generate a default constructor.


OK. This comment is a bit usless.

    /**
     * Add an element to the buffer
     */
    void add(char c)

The self documeting nature of the function already tells me this. Don't need a comment to tell me the exact same thing.


Think I may have just learned something new.

        const std::lock_guard<std::mutex> lg(print_mu);
        (std::cout << ... << args);

Don't recognize this format. Will need to look up what it means.



ThreadPool

You don't need to dynamically allocate the vector!

    std::vector<std::thread *> *threads_;  // Also the * should move left.
                                           // The * is part of the type info
                                           // so should be with the type
                                           // not the member name.

This can simply be:

    std::vector<std::thread*>   threads_;

Don't dynamically allocate something if it is not required.


Why are you keeping pointers to the threads?
Why does the thread pool not own the threads? You can create the thread then move the thread into the pool. Or simply pass the function to the thread pool and allow it to assign the function to a thread.

    // This is what I would do.
    std::vector<std::thread>   threads_;

    template<typename F>
    void add_action(F&& action)
    {
        threads.emplace_back(std::move(action));

}

The member is_finished_ is never used.

You should turn on your compiler warnings and fix all warnings. A warning is a an error in your logical thinking. The compiler lets it go because it is technically valid but the warning is there for a reason (you have messed up in some way).


You have a method join_all() which is fine. But would you not want to force this call from the destructor (if they had all already not been joined?

That way you can never accidentally go out of scope and leave threads running.

If I look at your main.

{
    // STUFF
    ThreadPool tp;

    // STUFF

    tp.join_all();
}

Yes it looks like that should simply be called from the destructor of the ThreadPool. That way if there is a problem you don't leave threads accidentally hanging.


Looking at main.
Does not look like buff needs to be dynamically allocated.

    {
        std::shared_ptr<Buffer> buff = std::make_shared<Buffer>(producer_parallelism);
        ThreadPool tp;

        for (/*LOOP*/)
        {
            Consumer *c = new Consumer{buff, producer_parallelism, i};
            // STUFF but `c` is added to the `tp` as a thread
        }

        for (/*LOOP*/)
        {
            Producer *p = new Producer{buff, produced_preaces_per_producer, i};
            // STUFF but `p` is added to the `tp`  as a thread
        }

        // Make sure all threads finish.
        tp.join_all();
    }

Here it created. Used only in the threads. You make sure all the threads terminate before you exit the scope. So all threads have access to the object for their lifetimes any only after all threads have finished do you exit scope and destroy the buffer. So easier to make this a local variable.

    {
        Buffer buff(producer_parallelism);
        ThreadPool tp;

        for (/*LOOP*/)
        {
            // Note this does not need to change.
            // Simply pass the buffer by reference and keep the
            // reference in the consumer.
            Consumer *c = new Consumer{buff, producer_parallelism, i};
            // STUFF but `c` is added to the `tp` as a thread
        }

        for (/*LOOP*/)
        {
            // Note this does not need to change.
            // Simply pass the buffer by reference and keep the
            // reference in the producer.
            Producer *p = new Producer{buff, produced_preaces_per_producer, i};
            // STUFF but `p` is added to the `tp`  as a thread
        }

        // Make sure all threads finish.
        tp.join_all();
    }

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thank you very much for your comment. Check out this link, if you want to learn more about folding expression (used in the print function) en.cppreference.com/w/cpp/language/fold \$\endgroup\$ – User12547645 Jan 22 at 10:46
  • \$\begingroup\$ Which parameter would you use to detect unused private member variables? g++ -Wunused-parameter does not do the trick. \$\endgroup\$ – User12547645 Jan 22 at 10:51
  • \$\begingroup\$ I really like the idea with the join_all in the destructor. Thanks for that! \$\endgroup\$ – User12547645 Jan 22 at 10:54
  • \$\begingroup\$ I use -Wall -Wextra -Wstrict-aliasing -Wunreachable-code -Werror -pedantic The extra and all turn on a whole bunch of parameters gcc.gnu.org/onlinedocs/gcc/Warning-Options.html the important one to me is the -Werror which makes me fix all my logical errors before it compiles. \$\endgroup\$ – Martin York Jan 22 at 18:59
  • \$\begingroup\$ A more standard looking thread Pool: codereview.stackexchange.com/q/47122/507 \$\endgroup\$ – Martin York Jan 22 at 19:06

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.