5
\$\begingroup\$

EDID: Thank you very much for your feedback. I updated the code and opened a new post for the updated version. See here.

This post is loosely based on A multi-threaded Producer Consumer with C++11.

I would like to implement producer consumer pattern, where a consumer consumes data from multiple producers. The idea is to share the data between each producer and the consumer via a buffer. The consumer has a list of these shared buffers.

The consumer is further encouraged to consume data as soon as it is available, no matter from which producer it comes. This is because in reality a producer might be delayed and it would not be ideal to wait for producer x, while producer y already produced something. The code below checks if this works by using a timer and delaying producer deliberately with different delays.

I would have liked to provide a ready-to-run example environment, but unfortunately, compiler-explorer does not allow multithreading. Please compile with -std=c++17 -pthread.

Code:

#include <atomic>
#include <chrono>
#include <iostream>
#include <math.h>
#include <memory>
#include <mutex>
#include <sstream>
#include <thread>
#include <vector>

/**
 * RAII-style timer. 
 * Used only in main to measure performance
 */
class MyTimer
{
public:
    MyTimer() : start(std::chrono::high_resolution_clock::now()) {}
    ~MyTimer()
    {
        std::cout << "elapsed time was " << std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::high_resolution_clock::now() - start).count() << " (us)\n";
    }

private:
    std::chrono::_V2::system_clock::time_point start;
};

class Buffer
{
public:
    Buffer(){};
    ~Buffer() = default;
    /**
     * Add an element to the buffer
     */
    void add(char c)
    {
        std::unique_lock<std::mutex> locker(mu);
        buffer_ << c;
        is_ready_ = true;
        return;
    }
    /**
     * pop/get an element from the buffer
     */
    char pop()
    {
        std::lock_guard<std::mutex> locker(mu);
        char c;
        buffer_ >> c;
        if (!production_ongoing_ && contains_input())
        {
            is_valid_.exchange(false);
            this->print("is valid to false in pop \n");
        }
        return c;
    }
    /**
     * getter for private is_valid_
     */
    bool is_valid()
    {
        return is_valid_.load();
    }
    /**
     * indicate to buffer that producer is finished/will not produce more data
     */
    void no_more_production()
    {
        std::lock_guard<std::mutex> locker(mu);
        production_ongoing_ = false;
        if (!contains_input())
        {
            is_valid_ = false;
        }
    }
    /**
     * helper for synced printing
     */
    void print(std::string msg)
    {
        std::lock_guard<std::mutex> lg(print_mu);
        std::cout << msg;
    }
    /**
     * getter for private is_ready_
     */
    bool is_ready()
    {
        std::lock_guard<std::mutex> locker(mu);
        return is_ready_;
    }
    /**
     * getter for private production_ongoing_
     */
    bool production_ongoing()
    {
        std::lock_guard<std::mutex> locker(mu);
        return production_ongoing_;
    }

private:
    std::mutex mu;       // sync all except print operation
    std::mutex print_mu; // sync print operations

    std::stringstream buffer_;         // a stream for sharing data
    bool production_ongoing_ = true;   // false if production is finished
    std::atomic_bool is_valid_ = true; // false, if producer is finished and buffer is empty
    bool is_ready_ = false;            // true after production initially began
    bool contains_input()              // check if there is input that can be retrieved from the buffer
    {
        // compare https://stackoverflow.com/questions/40608111/why-is-18446744073709551615-1-true
        int tmp = buffer_.peek(); 
        return tmp != -1 && tmp != std::pow(2, 64) - 1;
    }
};

class Producer
{
public:
    Producer(std::shared_ptr<Buffer> buffer, const int limit, const int id, const int delay) : buffer_(buffer), limit_(limit), id_(id), delay_(delay) {}
    /**
     * produces random data.
     */
    void run()
    {
        // for simulating delay of the producer
        std::this_thread::sleep_for(std::chrono::milliseconds(delay_));
        for (int count = 0; count < limit_; ++count)
        {
            char upper_case_char = (char)((random() % 26) + int('A'));
            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(50));
        }
        buffer_->no_more_production(); // indicate to buffer that production is done
    }

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
    const int delay_;                // start delay of producer
};

class Consumer
{
public:
    Consumer(std::vector<std::shared_ptr<Buffer>> &buffers, const int parallelism) : buffers_(buffers), parallelism_(parallelism){};
    void run()
    {
        // Consumer responsible for multiple producer. Is any of them still producing?
        bool any_valid = true;
        do
        {
            // if not all producers joined yet. This is in case the consumer is created earlier than the prod
            any_valid = buffers_.size() < parallelism_ ? true : false;

            // iterate over all available buffers
            for (size_t t = 0; t < buffers_.size(); ++t)
            {
                if (!buffers_.at(t)->is_ready())
                {
                    // will skip this producer. Helpful if producer is slow (network delay)
                    any_valid = true;
                    continue;
                }

                if (buffers_.at(t)->is_valid())
                {
                    // is_valid if we are expecting data from producer
                    any_valid = true;
                    char c = buffers_.at(t)->pop();
                    std::stringstream strs;
                    strs << "Consumed: " << c << '\n';
                    buffers_.at(t)->print(strs.str());
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        } while (any_valid);
        buffers_.at(0)->print("consumer finished\n");
    }

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

int main()
{
    {
        // all numbers are based on measurements on my machine in debug mode
        // Scenario 1: All producer threads have the same delay
        // if we do not start with ready thread, this will take about 0.3s
        // if we do start with ready thread, this will take about 0.25s
        MyTimer mt;
        const int parallelism = 3;
        std::vector<std::shared_ptr<Buffer>> buffVec;
        Consumer c{buffVec, parallelism};
        std::thread consumer_thread(&Consumer::run, &c);

        for (int i = 0; i < parallelism; ++i)
        {
            // each buffer is shared between a producer and the consumer
            std::shared_ptr<Buffer> b = std::make_shared<Buffer>();
            buffVec.push_back(b);
            Producer *p = new Producer(b, 3, i, 30);
            std::thread producer_thread(&Producer::run, &(*p));
            producer_thread.detach();
        }
        consumer_thread.join();
    }
    {
        // Scenario 2: First producer thread has long delay, others have none
        // Total delay is equal to Scenario 1
        // if we do not start with ready thread, this will take 0.5s
        // if we do start with ready thread, this will take about 0.3s
        MyTimer mt;
        const int parallelism = 3;
        std::vector<std::shared_ptr<Buffer>> buffVec;
        Consumer c{buffVec, parallelism};
        std::thread consumer_thread(&Consumer::run, &c);

        for (int i = 0; i < parallelism; ++i)
        {
            const int delay = i == 0 ? 90 : 0;
            // each buffer is shared between a producer and the consumer
            std::shared_ptr<Buffer> b = std::make_shared<Buffer>();
            buffVec.push_back(b);
            Producer *p = new Producer(b, 3, i, delay);
            std::thread producer_thread(&Producer::run, &(*p));
            producer_thread.detach(); // start producers independent from each other and from consumer
        }

        consumer_thread.join();
    }
    return 0;
}
\$\endgroup\$
  • \$\begingroup\$ Your code doesn't compile for me on wandbox. What compiler version are you using? \$\endgroup\$ – L. F. Jan 15 at 11:29
  • \$\begingroup\$ Builds fine for me (locally) with GCC 9.2. Just a couple of warnings. \$\endgroup\$ – Toby Speight Jan 15 at 11:47
  • \$\begingroup\$ Don't roll your own. This is too hard to get right. Use Boost's: boost.org/doc/libs/1_72_0/doc/html/boost/lockfree/queue.html Facebook's Folly library also has one. \$\endgroup\$ – Cody Gray Jan 15 at 22:03
4
\$\begingroup\$

std::this_thread::sleep_for(std::chrono::milliseconds(50));

Don't do this. Use a condition_variable instead. This will require some minor revising. 50ms might be a long time. Remember that to the OS, it means "context switch out the thread and keep it idle for at least 50ms". A std::mutex may have all manner of fancy implementation dependent optimizations. For example, if the consumer exhausts its work and waits on the condition variable, it may not need to be context switched at all if new work is very quickly produced.

Furthermore, this is wasting precious CPU resources. If production is stalled, it will context switch up to 20 times per second for no reason.


buffVec needs to be synchronized


Avoid adding artificial delays to your code. I believe they're hiding potential race conditions.


In my opinion, consider removing shared_ptr and making the client manage the memory. I believe should be implicit to the client that the memory used by the producer/consumer needs to outlive both of them to function properly. It can be more efficient in some cases, but require more code in other cases to move the shared state around. If this were a library, it could potentially be a templated type and the client could choose their desired storage strategy.


I am very adverse to seeing atomics in code that isn't building other low-level synchronization primitives. Using a mutex with RAII guards is much safer, and without any benchmarking to compare, I would argue that there's no reason to expect them to be underperformant. It can cause cache invalidation issues and out of order problems that are hard to reason about.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thank you very much. I think you are right. The artificial delays to hide race conditions \$\endgroup\$ – User12547645 Jan 16 at 11:01
6
\$\begingroup\$

I get a couple of warnings, which ought to be fixed:

g++ -std=c++2a -fPIC -g -Wall -Wextra -Wwrite-strings -Wno-parentheses -Wpedantic -Warray-bounds  -Weffc++    -pthread    235651.cpp    -o 235651
235651.cpp: In constructor ‘Buffer::Buffer()’:
235651.cpp:31:5: warning: ‘Buffer::mu’ should be initialized in the member initialization list [-Weffc++]
   31 |     Buffer(){};
      |     ^~~~~~
235651.cpp:31:5: warning: ‘Buffer::print_mu’ should be initialized in the member initialization list [-Weffc++]
235651.cpp:31:5: warning: ‘Buffer::buffer_’ should be initialized in the member initialization list [-Weffc++]
235651.cpp: In member function ‘void Consumer::run()’:
235651.cpp:159:41: warning: comparison of integer expressions of different signedness: ‘std::vector<std::shared_ptr<Buffer> >::size_type’ {aka ‘long unsigned int’} and ‘const int’ [-Wsign-compare]
  159 |             any_valid = buffers_.size() < parallelism_ ? true : false;
      |                         ~~~~~~~~~~~~~~~~^~~~~~~~~~~~~~

We include <math.h> but then use std::pow - we should be including <cmath> if we want the names to be in the std namespace (which we do).

The consumers don't block, but repeatedly get a null. That seems to be a failure of the whole purpose of the class:

Produced: N. Count at 0. Producer was 0
Produced: L. Count at 0. Producer was 2
Produced: W. Count at 0. Producer was 1
Consumed: N
Consumed: W
Consumed: L
Consumed: \0
Consumed: \0
Consumed: \0
Produced: B. Count at 1. Producer was 2
Produced: B. Count at 1. Producer was 0
Produced: R. Count at 1. Producer was 1
Consumed: \0
Consumed: \0
Consumed: \0
Produced: Q. Count at 2. Producer was 1
Produced: B. Count at 2. Producer was 2
Produced: M. Count at 2. Producer was 0
Consumed: \0

Other questionable bits:

  • buffers_.size() < parallelism_ ? true : false
    

    That should be written as just buffers_.size() < parallelism_.

  • char upper_case_char = (char)((random() % 26) + int('A'));
    

    We need <cstdlib> to define std::random(). And C++ doesn't guarantee that letters are contiguously encoded. Try

    static char const alphabet[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
    char upper_case_char = alphabet[(std::random() % (sizeof alphabet - 1))];
    
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thank you very much for the comment. Would you change anything regarding the synchronization mechanisms? \$\endgroup\$ – User12547645 Jan 15 at 14:36
  • 1
    \$\begingroup\$ Probably - we probably want a semaphore to fix the consumers to wait for input. Fix that bug first; I think you'll want a new review when you've done that. \$\endgroup\$ – Toby Speight Jan 15 at 15:16
  • 1
    \$\begingroup\$ Semaphores are definitely the right tool for the job, but standard C++ doesn't have them. You could use an OS implementation, or a cross platform library, or condition_variable. \$\endgroup\$ – butt Jan 15 at 20:45
5
\$\begingroup\$

Avoid mixing clock types

Why use std::chrono::_V2::system_clock::time_point start but initialize it with std::chrono::high_resolution_clock::now()? If there is a reason to use the non-standard _V2 clocks, then you should probably stick with it everywhere. But if there is no reason to, avoid the non-standard _V2 thing.

To make your code more consistent and to reduce the amount of code you have to type, define an alias for the clock namespace you want to use, like so:

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

Avoid useless definitions of default constructors and destructors

In class Buffer, the only constructor is not doing anything, and the destructor is set to the default. There is no need for this, just omit them completely.

There is no need to lock in print()

Single calls to member functions of iostreams are atomic, see this post. So there is no need for print_mu.

Avoid detaching threads

There is almost never a good reason to detach threads. Doing so means losing control over the threads and the resources it uses. Threads can be easily managed by STL containers. So in your main() you could write:

std::vector<std::thread> producer_threads;

for (int i = 0; i < parallelism; ++i)
{
    ...
    Producer *p = new Producer(b, 3, i, 30);
    producer_threads.emplace_back(&Producer::run, p);
}

consumer_thread.join();

for (auto &thread: producer_threads)
    thread.join();

Note that you are still leaking Producer objects, since you never delete them. You could put those in a std::vector as well, or you could change class Producer to start a thread in its own constructor, so you just need the vector holding Producers.

| improve this answer | |
\$\endgroup\$

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.