Producer-Consumer in C++

Question

This is my first experimentation with the producer-consumer problem (and thread manipulation in general), so any feedback on taboos and fixes is greatly appreciated!

#include <condition_variable>
std::condition_variable cv;

#include <iostream>
#include <random>

#include <mutex>
std::mutex mtx;

#include <thread>

// standard in all of my programs
void panic(const std::string &msg) {
    throw std::runtime_error("Error: " + msg + "\n");
}

int count = 0, buff_size = 0;
char* buff;

int random_int(int lower_bound) {
    std::random_device seed;
    std::mt19937 generator(seed());
    std::uniform_int_distribution<int> dist(lower_bound, std::nextafter(26, DBL_MAX));

    return dist(generator);
}

char random_char(int lower_bound) {
    return 'A' + (random_int(lower_bound) % 26);
}

/* Consumer

Reads characters from the buffer and prints them.

*/
void consume(int job) {
    std::unique_lock<std::mutex> lck(mtx);

    while (count == 0) {
        cv.wait(lck);
    }

    /*
    job + 1 = Running
    job = Ready
    */
    for (int i = 0; i < buff_size; i++) {
        std::cout << buff[i] << std::endl;
    }

    count--;
}

/* Producer

Randomly generates letters at (pos > buff_size & pos <= 26),
inserts them at the next available position in the buffer,
and then prints out the lowercase form of the inputted letter.

*/
void produce(int job) {
    std::unique_lock<std::mutex> lck(mtx);

    for (int i = 0; i < buff_size; i++) {
        buff[i] = random_char(buff_size);
        std::cout << static_cast<char>(tolower(buff[i])) << std::endl;
    }

    count++;
    cv.notify_one();
}

int main() {
    int buf_size = 0;

    std::cout << "The Producer-Consumer Problem (in C++11!)" << std::endl << "Enter the buffer size: ";
    std::cin >> buf_size;

    if (buf_size > 0 && buf_size <= 26) {
        // set the buffer size
        buff_size = buf_size;
        buff = new char[buff_size];
    }
    else {
        // rage quit
        panic("The range must be > 0 and <= 26.");
    }

    std::thread production[10], processed[10];

    /* Initialize the arrays */
    for (int order = 0; order < buff_size; order++) {
        production[order] = std::thread(produce, order);
        processed[order] = std::thread(consume, order);
    }

    /* Join the threads to the main threads */
    for (int order = 0; order < buff_size; order++) {
        processed[order].join();
        production[order].join();
    }
}

Answer 1

#include <condition_variable>
std::condition_variable cv;

#include <iostream>
#include <random>

#include <mutex>
std::mutex mtx;

#include <thread>

Don't mix includes and code, makes it easy to miss those globals you're defining in the middle. Keep the headers by themselves sorted in some order or grouped logically/by function as you prefer.

throw std::runtime_error(...);

You're missing <stdexcept>, your code could fail to compile.

int count = 0, buff_size = 0;
char* buff;

What are these? Why are they global? Why initialize two explicitly and not the third?

int random_int(int lower_bound) {
    std::random_device seed;
    std::mt19937 generator(seed());
    std::uniform_int_distribution<int> dist(lower_bound, std::nextafter(26, DBL_MAX));

    return dist(generator);
}

This isn't good. You're seeding a generator each time you generate a new random number - that's not how you're supposed to do it. You're supposed to seed a generator once, then use it repeatedly.

std::nextafter(26, DBL_MAX)

You're missing <cfloat> for DBL_MAX, and I really don't understand what this is for, or why 26 appears magically here with no warning (no indication in the function name, no comment). nextafter/before are really meant for floating point numbers (or template code where you don't know if you'll get ints or floats). If it's a plain int constant, just code it in. Actually, don't put that magic constant at all and allow the user to pass the lower and upper bound.

char random_char(int lower_bound) {
    return 'A' + (random_int(lower_bound) % 26);
}

(This only works for encodings where the letters are consecutive. This isn't guaranteed, but chances that your code will run on something with e.g. EBCDIC are small.)

Why %26 if your RNG already gives you values in the right range? The % trick is likely to ruin your distribution's statistical properties. So again, just pass the min and max to your random generating function (e.g. return myrand('A','Z');).

Here's one way to fix this:

// bounds are inclusive
char rand_char(char lower_bound, char upper_bound)
{
  thread_local std::random_device rd;
  thread_local std::mt19937 gen(rd());
  thread_local std::uniform_int_distribution<int> dist(lower_bound,
                                                        upper_bound);
  return static_cast<char>(dist(gen));
}

Then you can call:

rand_char('A', 'Z');

thread_local gives each thread its own generator and distribution, initialized only once per thread on first call of the function.

/* Producer

Randomly generates letters at (pos > buff_size & pos <= 26),
inserts them at the next available position in the buffer,
and then prints out the lowercase form of the inputted letter.

*/

This comment does not match the code, the code doesn't contain a pos at all. There's a job parameter, but it is not used. There does not appear to be any "next available" concept in the code. There is no "inputted letter" either. And if pos > buff_size, generating something at buff[pos] would be an overflow. So fix that comment (and fix the code to match it) or remove it. Bad comments are worse than no comment at all.

int buff_size = 0;
// ...
int main() {
  int buf_size = 0;

That's just evil! Are you running in an underhanded C++ contest? :-) Really that's a terrible choice of variable names.

std::cout << "The Producer-Consumer Problem (in C++11!)" << std::endl << "Enter the buffer size: ";
std::cin >> buf_size;
if (buf_size > 0 && buf_size <= 26) {
  // ...
} else {
  // rage quit
  panic("The range must be > 0 and <= 26.");
}

Bad user experience: inform about the limits before asking for the input.

buff = new char[buff_size];

Where's the delete[]? If your comment above produce was what you intended, you'd need to initialize too, otherwise your producers could read uninitialized data (which leads to undefined behavior). Use a std::vector instead to avoid the leaks and initialize all values to a sane default (e.g. a space).

As for the whole producer/consumer thing: your code is completely serial. There can only be one active thread, whether it is a producer or a consumer. Multiple produce threads could run one after the other without a consumer running in between, so you're "losing products" and it is likely that you'll see the same "product" output multiple times.

You should change your code so that the producers produce something predictable (e.g. "produce" and print out their job number) so that you can actually visually verify whether things are working. Then test your code with starting all the consumers first (and waiting a tiny bit), then the consumers. Then do the opposite. Run it several times, always checking that the output is correct. Increase the concurrency. Use sanitizers if you can (-fsanitize option for recent-enough GCC and clang for instance).

Answer 2

My revised code from the answer of @Mat:

#include <condition_variable>
#include <iostream>
#include <random>
#include <mutex>
#include <thread>
#include <vector>

// global variables
std::condition_variable cv;
std::mutex mtx;
std::vector<char> data;

int count = 0, buff_size = 0;

char random_char() {
    thread_local std::random_device seed;
    thread_local std::mt19937 generator(seed());
    thread_local std::uniform_int_distribution<int> dist('A', 'Z');

    return static_cast<char>(dist(generator));
}

/* Consumer

Prints out the contents of the shared buffer.

*/
void consume() {
    std::unique_lock<std::mutex> lck(mtx);

    while (count == 0) {
        cv.wait(lck);
    }

    for (const auto& it : data) {
        std::cout << it << std::endl;
    }

    count--;
}

/* Producer

Randomly generates capital letters in the range of A to Z,
prints out those letters in lowercase, and then
inserts them into the shared buffer.

*/
void produce() {
    std::unique_lock<std::mutex> lck(mtx);

    char c = random_char();
    std::cout << " " << static_cast<char>(tolower(c)) << std::endl;
    data.push_back(c);

    count++;
    cv.notify_one();
}

int main() {
    std::cout << "The Producer-Consumer Problem (in C++11!)" << std::endl << "Enter the buffer size: ";
    std::cin >> buff_size;

    // keep the buffer in-range of the alphabet
    if (buff_size < 0) {
        buff_size = 0;
    }
    else if (buff_size > 26) {
        buff_size = 26;
    }

    std::thread production[26], processed[26];

    // initialize the arrays
    for (int order = 0; order < buff_size; order++) {
        production[order] = std::thread(produce);
        processed[order] = std::thread(consume);
    }

    // join the threads to the main threads
    for (int order = 0; order < buff_size; order++) {
        processed[order].join();
        production[order].join();
    }

    std::cout << "Succeeded with a shared buffer of " << data.size() << " letters!";
}