Single Producer - Multiple Consumers in C++11 using Linux Threads

Question

Hey everyone I was redirected here from Stack Overflow for this question.

I am completely new to thread synchronization using condition variables and pthreads in Linux.

Here is the code I wrote and I will explain the main idea behind it :

#include <pthread.h>
#include <iostream>

using namespace std;

pthread_cond_t bufferNotFull = PTHREAD_COND_INITIALIZER;
pthread_cond_t bufferNotEmpty = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

class Buffer {

private:
    string *buffer;
    int size;
    int total;
    int addIDX;
    int getIDX;

public:
    Buffer(int bufferSize) : size(bufferSize) {
        this->buffer = new string[this->size];
        this->total = 0;
        this->addIDX = 0;
        this->getIDX = 0;
    }

    virtual ~Buffer() {
        delete[] this->buffer;
    }

    void addItem(const string &item) {
        if (this->addIDX == this->size)
            this->addIDX = 0;
        this->buffer[this->addIDX] = item;
        this->total++;
        this->addIDX++;
    }

    string getItem() {
        if (this->getIDX == this->size)
            this->getIDX = 0;

        string item = this->buffer[this->getIDX];
        this->total--;
        this->getIDX++;
        return item;
    }

    bool isFull() { return (this->total == this->size); }

    bool isEmpty() { return (this->total == 0); }

};

class Resource {

private:
    int id;
    int sum;

public:
    Resource(int id) : id(id) { this->sum = 0; }

    void updateSum(int num) { this->sum += num; }

    void print() { cout << "Resource with ID " << this->id << " has sum : " << this->sum << endl; }
};

Buffer *buffer;
int totalResources = 6;
Resource **resourceManager = new Resource *[totalResources]{nullptr};

string randomMessage() {
    // Generate a random message to be added to the buffer in the following format : ResourceID#ValueForSum
    int randomResourceID = rand() % totalResources;
    int randomValue = (rand() % 10) + 1;
    return to_string(randomResourceID) + "#" + to_string(randomValue);
}

void decodeMessage(const string &message, string &resourceID, string &resourceValue) {

    // Decode a message by storing each component in the appropriate variable
    unsigned int j = 0;

    while (message[j] != '#') {
        resourceID.push_back(message[j]);
        j++;
    }

    j++;

    while (j != message.size()) {
        resourceValue.push_back(message[j]);
        j++;
    }

}

void *consumer(void *argPtr) {

    bool keepRunning = true;
    while (keepRunning) {
        pthread_mutex_lock(&mutex);

        while (buffer->isEmpty())
            pthread_cond_wait(&bufferNotEmpty, &mutex);

        string itemRead = buffer->getItem();
        string resourceID;
        string resourceValue;
        decodeMessage(itemRead, resourceID, resourceValue);

        if (resourceValue == "QUIT_STRING")
            keepRunning = false;
        else
            resourceManager[stoi(resourceID)]->updateSum(stoi(resourceValue));

        pthread_cond_signal(&bufferNotFull);

        pthread_mutex_unlock(&mutex);

    }

    pthread_exit(nullptr);
}

int main(int argc, char **argv) {

    srand(time(nullptr));

    int bufferSize = 3;
    buffer = new Buffer(bufferSize);

    for (int i = 0; i < totalResources; i++)
        resourceManager[i] = new Resource(i);

    int totalThreads = stoi(argv[1]);
    pthread_t *threadIDS = new pthread_t[totalThreads];
    for (int i = 0; i < totalThreads; i++)
        pthread_create(&threadIDS[i], nullptr, consumer, nullptr);


    volatile bool keepRunning = true;
    string userCommand;

    while (keepRunning) {

        /* Get user command */
        std::cout << "\nCommand : ";
        getline(cin, userCommand);

        /* Generate a random number of messages and insert them to bounded buffer */
        if (userCommand == "/fill") {
            int messagesToAdd = (rand() % 5) + 1;
            std::string messages[messagesToAdd];
            /* Print the messages generated to check for behavior afterwards */
            for (int i = 0; i < messagesToAdd; i++) {
                messages[i] = randomMessage();
                cout << messages[i] + " ";
            }
            cout << endl;

            for (unsigned int i = 0; i < messagesToAdd; i++) {
                pthread_mutex_lock(&mutex);
                while (buffer->isFull())
                    pthread_cond_wait(&bufferNotFull, &mutex);
                string itemToAdd = messages[i];
                buffer->addItem(itemToAdd);
                pthread_cond_signal(&bufferNotEmpty);
                pthread_mutex_unlock(&mutex);
            }
        }

        /* Insert the message 0#QUIT_STRING in bounded buffer totalThreads times so that threads know that it's time to exit*/
        else if (userCommand == "/exit") {

            for (unsigned int i = 0; i < totalThreads; i++) {
                pthread_mutex_lock(&mutex);
                while (buffer->isFull())
                    pthread_cond_wait(&bufferNotFull, &mutex);
                string itemToAdd = "0#QUIT_STRING";
                buffer->addItem(itemToAdd);
                pthread_cond_broadcast(&bufferNotEmpty);
                pthread_mutex_unlock(&mutex);
            }

            keepRunning = false;
        }

        else
            cout << "Unknown command" << endl;

    }


    for (int i = 0; i < totalThreads; i++)
        pthread_join(threadIDS[i], nullptr);

    delete buffer;
    delete[] threadIDS;
    for (int i = 0; i < totalResources; i++) {
        /* Print the outcome for each resource */
        resourceManager[i]->print();
        delete resourceManager[i];
    }

    delete[] resourceManager;

    pthread_cond_destroy(&bufferNotFull);
    pthread_cond_destroy(&bufferNotEmpty);
    pthread_mutex_destroy(&mutex);

}

The idea is to have a :

• A bounded/circular buffer
• The main thread that acts as a producer and adds items to the buffer
• Multiple "consumer" threads that work on and update shared resources after reading from buffer.

After multiple test cases the code above seems to work just fine with no race conditions and no threads left behind and therefore suspending the main thread forever.

However as I am new to this subject I was wondering if I am missing something and if there is a better way to do it.

Furthermore I was told on Stack Overflow that I should use things like std::condition_variable,std::thread and std::mutex but the thing is I am studying this subject for an upcoming assignment and these things or even simpler ones like std::vector are not allowed so any insight would be really useful.

Answer 1

Use the standard library where possible:

The standard library exists because it's... standard. It's code that lots of people need to use frequently, and if it didn't exist, people would be stuck implementing at least a subset of it themselves in every program (like you are).

While interesting and educational, that task isn't trivial, and it's not suitable as a "side-quest" for an assignment. So it's a shame that your course forbids using these parts of the standard library (and perhaps something to point out if you have a feedback form to submit at the end of the course).

You might find it easier (or at least more educational) to write a task like this using the standard library, and then figure out which parts you need to re-write to submit it.

---

Avoid global variables:

pthread_cond_t bufferNotFull = PTHREAD_COND_INITIALIZER;
pthread_cond_t bufferNotEmpty = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

Buffer *buffer;
int totalResources = 6;
Resource **resourceManager = new Resource *[totalResources]{nullptr};

It's best to avoid global variables. It makes code more fragile, harder to read, and less reusable. We can declare these as local variables, and pass them to the functions that need them by pointer or reference.

Note that we can pass multiple variables to the consumer function by wrapping them all in a struct, and passing a pointer to the struct instance as the void* argument to pthread_create(). We then cast it back to the correct type inside consumer().

---

Rule of three / five:

When writing a class that manages resources or memory, or enforces some condition on its member variables, we need to consider the following:

• Construction (resource allocation): Buffer(int size).
• Destruction (freeing resources): ~Buffer().
• Copying (copy construction and copy assignment): Buffer(Buffer const&), Buffer& operator=(Buffer const&).
• Moving (move construction and move assignment): Buffer(Buffer&&), Buffer& operator=(Buffer&&).

To avoid leaking resources, or double deletion, we have to implement these in certain combinations. These are referred to as the rules of three, five and zero.

class Buffer {

private:
    string *buffer;

public:
    Buffer(int bufferSize) : size(bufferSize) {
        this->buffer = new string[this->size];
        this->total = 0;
        this->addIDX = 0;
        this->getIDX = 0;
    }

    virtual ~Buffer() {
        delete[] this->buffer;
    }
};

Here we allocate and free resources. But we don't consider copy-construction or assignment:

Buffer a(5);
Buffer b(a); // copy construction
Buffer c(3);
c = b; // copy assignment

These lines compile because the compiler generates the operators for us. The compiler implementations copy each member across to the new class instance by value. But this isn't what we want! (we'll get a seg-fault).

Each class instance "owns" the resources pointed to by the buffer member. We only want to delete this data once, and we don't want to share it between class instances.

So we either have to implement our own copy operations, or we need to explicitly prevent the compiler from generating its own. I'd suggest the latter for simplicity. We can do this like so:

    Buffer(Buffer const&) = delete;
    Buffer& operator=(Buffer const&) = delete;

If we wanted to support moveing the class, we would also need to define our own move operators to properly manage the buffer resource. Since defining copy operators (or deleting them as above) also prevents the compiler generating move operators automatically, we can now ignore that too.

The Resource class is ok. Since it holds plain integer values, the compiler generated operators will do the right thing.

---

Don't do using namespace std;:

It's a bad habit to get into, even for small programs, as it can lead to name collisions and other problems.

We should explicitly qualify names where necessary. i.e. type the full std::string, std::cout, etc.

---

Use <random> instead of rand():

C++11 introduced much better facilities for random number generation in the <random> header. We should use those instead of rand(), something like:

std::string randomMessage(std::mt19937& rng, int totalResources) {
    std::uniform_int_distribution<int> idDist(0, totalResources - 1);
    std::uniform_int_distribution<int> valueDist(1, 10);
    return std::to_string(idDist(rng)) + "#" + std::to_string(valueDist(rng));
}

int main()
{
    std::mt19937 rng{ std::random_device()() }; // seed the random number generator
    ...
    std::string message = randomMessage(rng, 6);
}

---

Use algorithms instead of plain loops where practical:

void decodeMessage(const string &message, string &resourceID, string &resourceValue) {

    // Decode a message by storing each component in the appropriate variable
    unsigned int j = 0;

    while (message[j] != '#') {
        resourceID.push_back(message[j]);
        j++;
    }

    j++;

    while (j != message.size()) {
        resourceValue.push_back(message[j]);
        j++;
    }

}

We can make this more meaningful by using existing functions from std::string. We should also try to make sure that the '#' actually exists in the string.

void decodeMessage(const std::string &message, std::string &resourceID, std::string &resourceValue) {

auto hash = message.find('#'); // get the index of the first '#' character in message (or npos if not found)
assert(hash != std::string::npos); // ensure that '#' is present
assert(hash != message.size() - 1); // ensure that there are characters after the '#'

resourceID = message.substr(0, hash); // get substring of "hash" characters, starting at zero
resourceValue = message.substr(hash + 1); // get substring from index "hash" to the end

}

---

Variable length arrays:

        int messagesToAdd = (rand() % 5) + 1;
        std::string messages[messagesToAdd];

Although some compilers will let you get away with this (with some combinations of settings), variable length arrays are not actually valid C++. messagesToAdd must be a compile time constant to use as an array size.

(I suppose we could create an array of size 6 here, or allocate it on the heap).

---

Stopping execution:

It seems very hacky (and it won't always be practical) to encode a special message to tell the consumer threads to quit. We should probably use a separate boolean variable, also protected by the mutex to do this.

---