Producer/consumer problem with four priority levels

Question

Here you can see a mediocre code of mine where I have fun with four deques, each with a different priority.

Here is how it works: in a while loop I generate a random number between 1 (highest priority) and 4 (lowest priority), and this number tells me in which deque I will insert the new element (by insert(int num, int fd)).

Before the while loop, a remover thread is detached with remove(void) method: this thread will remove from the highest priority maxQ deque (if there are any elements), then from medQ and so on to the lowest one, `unknownQ'.

Apparently it works: I know there is a matter of starvation since it will always remove elements with higher priorities; what if deque maxQ is always full of elements and minQ has few elements? Yep, deques with lower priorities will never be updated. For now the elements to insert are randomly generated and this assures me that, sooner or later, all elements from all deques will be removed.

But this is not my problem: I just hate the way insert and remove methods are designed. They are (to me) ugly and redundant: the code is repeated, almost identical, for each deque. I come from C and I still don't know the true power of C++: what suggestions do you provide to improve insert and remove?

Compiled with:

g++ -std=c++11 -o funwithmultideque funwithmultideque.cpp -pthread

The code:

#include <iostream>
#include <string>
#include <mutex>
#include <condition_variable>
#include <deque>
#include <chrono>
#include <thread>
#include <random>

#define MAX_QUEUE_SIZE      100

#define DEFAULTCOLOR        "\033[0m"
#define RED                 "\033[22;31m"
#define YELLOW              "\033[1;33m"
#define GREEN               "\033[0;0;32m"

#define debug_default(...) std::cout << __VA_ARGS__ << DEFAULTCOLOR << '\n' << std::flush;
#define debug_red(...) std::cout << RED << __VA_ARGS__ << DEFAULTCOLOR << '\n' << std::flush;
#define debug_yellow(...) std::cout << YELLOW << __VA_ARGS__ << DEFAULTCOLOR << '\n' << std::flush;
#define debug_green(...) std::cout << GREEN << __VA_ARGS__ << DEFAULTCOLOR << '\n' << std::flush;

// this is the element the deques will contain
typedef struct info_connection {
    int fd;
    std::chrono::time_point<std::chrono::system_clock> start;
} info_conn;

class QueuesManager {
    public:
        void initWorkerThread(void);
        void remove(void);
        void insert(int num, int fd);
        // for debug
        void printQueues(int flag);
    private:
        std::thread threadRead; // remover thread
        std::mutex m1, m2, m3, m4;
        // read and write conditions for each deque
        std::condition_variable w1, r1, w2, r2, w3, r3, w4, r4;
        std::deque<info_conn> maxQ, medQ, minQ, unknownQ;
};

void QueuesManager::printQueues(int flag) {
    // show deques after inserting
    if (flag == 1) {
        debug_green(maxQ.size() << ' ' << medQ.size() << ' ' <<
            minQ.size() << ' ' << unknownQ.size());
    }
    // show deques after removing
    else {
        debug_yellow(maxQ.size() << ' ' << medQ.size() << ' ' <<
            minQ.size() << ' ' << unknownQ.size());     
    }
}

void QueuesManager::insert(int num, int fd) {
    info_conn ic;
    switch(num) {
        case 1: {
            std::unique_lock<std::mutex> locker(m1);
            w1.wait(locker, [this] () { return maxQ.size() < MAX_QUEUE_SIZE; });
            ic.start = std::chrono::system_clock::now();
            ic.fd = fd;
            maxQ.push_back(ic);
            printQueues(1);
            r1.notify_one();
            break;
        }
        case 2: {
            std::unique_lock<std::mutex> locker(m2);
            w2.wait(locker, [this] () { return medQ.size() < MAX_QUEUE_SIZE; });
            ic.start = std::chrono::system_clock::now();
            ic.fd = fd;
            medQ.push_back(ic);
            printQueues(1);
            r2.notify_one();
            break;
        }
        case 3: {
            std::unique_lock<std::mutex> locker(m3);
            w3.wait(locker, [this] () { return minQ.size() < MAX_QUEUE_SIZE; });
            ic.start = std::chrono::system_clock::now();
            ic.fd = fd;
            minQ.push_back(ic);
            printQueues(1);
            r3.notify_one();
            break;      
        }
        case 4: {
            std::unique_lock<std::mutex> locker(m4);
            w4.wait(locker, [this] () { return unknownQ.size() < MAX_QUEUE_SIZE; });
            ic.start = std::chrono::system_clock::now();
            ic.fd = fd;
            unknownQ.push_back(ic);
            printQueues(1);
            r4.notify_one();
            break; 
        }
        default: {
            std::cout << "You shouldn't be here\n" << std::flush;
            break;
        }
    }
}

void QueuesManager::remove(void) {
    while(true) {
        info_conn ic;
        if (maxQ.size() > 0) {
            std::unique_lock<std::mutex> lck(m1);
            r1.wait(lck, [this] () { return maxQ.size() > 0; });
            ic = maxQ.front();
            maxQ.pop_front();
            printQueues(0);
            w1.notify_one();
            continue;
        }
        if (medQ.size() > 0) {
            std::unique_lock<std::mutex> lck(m2);
            r2.wait(lck, [this] () { return medQ.size() > 0; });
            ic = medQ.front();
            medQ.pop_front();
            printQueues(0);
            w2.notify_one();
            continue;
        }
        if (minQ.size() > 0) {
            std::unique_lock<std::mutex> lck(m3);
            r3.wait(lck, [this] () { return minQ.size() > 0; });
            ic = minQ.front();
            minQ.pop_front();
            printQueues(0);
            w3.notify_one();
            continue;
        }
        if (unknownQ.size() > 0) {
            std::unique_lock<std::mutex> lck(m4);
            r4.wait(lck, [this] () { return unknownQ.size() > 0; });
            ic = unknownQ.front();
            unknownQ.pop_front();
            printQueues(0);
            w4.notify_one();
            continue;
        }
    }
}

void QueuesManager::initWorkerThread(void) {
    threadRead = std::thread(&QueuesManager::remove, this);
    threadRead.detach();
}

int main(void) {

    int randomNum = 0;
    int fd = 0;
    QueuesManager qm;

    std::default_random_engine eng((std::random_device())());
    std::uniform_int_distribution<int> randomPrio(1, 4);

    qm.initWorkerThread();

    while (true) {
        fd++;
        randomNum = randomPrio(eng);
        qm.insert(randomNum, fd);
    }

    return 0;
}

Answer 1

Encapsulate individual queues

There's a lot of identical code shared between your queues, which is an indication to encapsulate them. Since they're stateful, a class is appropriate.

class QueueEntry {
public:

    info_connection remove();

    void insert(int fd);

    size_t size() const { return q.size(); }

private:
    std::mutex m;
    std::condition_variable w, r;
    std::deque<info_conn> q;
};

info_connection QueueEntry::remove(){
    std::unique_lock<std::mutex> lck(m);
    r.wait(lck, [this]() { return q.size() > 0; });
    auto i = std::move(q.front());
    q.pop_front();
    w.notify_one();
    return i;
}

void QueueEntry::insert(int fd){
    std::unique_lock<std::mutex> locker(m);
    w.wait(locker, [this]() { return q.size() < MAX_QUEUE_SIZE; });
    info_connection i { fd, std::chrono::system_clock::now() };
    q.push_back(i);
    r.notify_one();
}

This let's us significantly simplify the remaining code in the queue manager:

class QueuesManager {
public:
    void initWorkerThread(void);

    void remove(void);

    void insert(int num, int fd);

    // for debug
    void printQueues(bool flag);

private:
    std::thread threadRead; // remover thread
    QueueEntry maxQ, medQ, minQ, unknownQ;
};

void QueuesManager::printQueues(bool flag) {
    // show deques after inserting
    if (flag) {
        debug_green(maxQ.size() << ' ' << medQ.size() << ' ' <<
                    minQ.size() << ' ' << unknownQ.size());
    }
        // show deques after removing
    else {
        debug_yellow(maxQ.size() << ' ' << medQ.size() << ' ' <<
                     minQ.size() << ' ' << unknownQ.size());
    }
}

void QueuesManager::insert(int num, int fd) {
    switch (num) {
        case 1: {
            maxQ.insert(fd);
            break;
        }
        case 2: {
            medQ.insert(fd);
            break;
        }
        case 3: {
            minQ.insert(fd);
            break;
        }
        case 4: {
            unknownQ.insert(fd);
            break;
        }
        default: {
            std::cout << "You shouldn't be here" << std::endl;
            break;
        }
    }
    printQueues(true);
}

void QueuesManager::remove(void) {
    while (true) {
        printQueues(false);
        info_conn ic;
        if (maxQ.size() > 0) {
            ic = maxQ.remove();
            continue;
        }
        if (medQ.size() > 0) {
            ic = medQ.remove();
            continue;
        }
        if (minQ.size() > 0) {
            ic = minQ.remove();
            continue;
        }
        if (unknownQ.size() > 0) {
            ic = unknownQ.remove();
            continue;
        }
    }
}

Note that we've rewritten printQueues to take a bool rather than a bool disguised as an int. If the specific queues are not that interesting, and merely the indexing we can simplify our methods further:

class QueuesManager {
    ...
    std::array<QueueEntry, 4> qs;
};

void QueuesManager::insert(int num, int fd) {
    QueueEntry &q = qs.at(num);
    q.insert(fd);
    printQueues(true);
}

void QueuesManager::remove(void) {
    while (true) {
        printQueues(false);
        for(auto &q : qs){
            if(q.size() > 0){
                info_conn ic = q.remove();
                break;
            }
        }
    }
}

Rewriting printQueues is left as an exercise.

Answer 2

Don't silently fail

This should probably throw an exception, swallowing errors like this leads to hidden bugs:

default: {
    std::cout << "You shouldn't be here\n" << std::flush;
    break;
}

'remove' naming

From the class header it is not obvious that this is going to be a never exit thread worker method. I would be concerned that a future client might try to call it to remove one item from a queue. Consider renaming it and making it private.

Worker thread

Your worker thread has no termination condition. You should provide a mechanism to signal it that processing is complete and it is time to terminate.

If there are no items in any of the queues (such as when the thread is initially created) then the thread function will spin, draining resources that could be used on other things such as inserting items into a queue. One approach to improve this would be to count items managed and wait while it is zero, or use a single flag that is set when items are inserted and have the thread wait on it if it fails to read from any of the queues.

initWorkedThread

If this is called twice you will start multiple worked threads. Is this expected/desired behaviour? I would assume not, in which case you should add protection to stop a second thread being started. One approach might be to make the initWorkerThread private and have it called automatically from the constructor. With the current approach if two threads are started, there is a race condition between checking the queues size and removing an item.

void QueuesManager::remove(void) {
while(true) {
    info_conn ic;
    if (maxQ.size() > 0) {  // <-- If two threads check this at the same 
                            // time and pass, but there is only 1 item
                            // in the queue
        std::unique_lock<std::mutex> lck(m1);
                            // One thread will make it through the wait
                            // condition, whilst the other will be stuck
                            // waiting for another item to be inserted
                            // into the queue
        r1.wait(lck, [this] () { return maxQ.size() > 0; });

size

I don't like that you are calling size outside of the guard methods. Is it documented that it is thread safe to do so? It probably won't cause an issue, since you check again within a guard, but it feels like you are making assumptions about how it is implemented.

r1 to 4

These feel wrong. If you only have one worker removing items from the queue and you are confident about size being thread safe then they are redundant. The if statement ensures the condition you are going to wait on has already been met. The mutex can simply be locked.

if (maxQ.size() > 0) {  // <- This
    std::unique_lock<std::mutex> lck(m1);
    r1.wait(lck, [this] () { return maxQ.size() > 0; }); // <- And this
                                                         // are the same check

If either of these isn't true and the remove thread actually blocks on these variables, it is because the thread is blocked on an empty queue that it expected to have an item (which seems like a bug). It will wait an indeterminate amount of time during which it isn't going to service other queues that might have items in them.

Answer 3

struct definition

Although the typedef struct tag { /* body */ } name; style is common in C, it's unnecessary in C++. You can just use: class name { /* body */ }; and get essentially the same effect.

Consider using a priority_queue

Most of the time, it'll be quite a bit simpler and cleaner to use a single priority queue instead of a separate queue for each priority level. This does require that each task store its priority and know how to compare by priority, so when you're inserting a new task you know where to insert it relative to the others:

enum priority { unknown, min, med, max };

struct info_conn {
    int fd;
    std::chrono::time_point<std::chrono::system_clock> start;
    priority p;

    bool operator<(info_conn const &other) { 
        return p < other.p;
    }
};

Since we have only one queue, so we only need one mutex and one condition variable.

Consider adding a ctor for info_conn

Since creating an info_conn means initializing its start_time, that should probably be handled in a ctor:

info_conn(int fd, priority p) :
    fd(fd),
    p(p),
    start(std::chrono::system_clock::now())
{}

Relieve the user of initializing the worker thread

One possibility would be to initialize the worker thread in the queue's constructor instead of leaving it to the client to do that.

If initWorkerThread were doing much work, it might be worth defining a separate WorkerThread class with that initialization done in its ctor. In this case, initializing the worker thread is pretty simple, so I doubt that would really gain a lot though.

Naming

Your classes should be named from the perspective of a user. The main class you're defining isn't a queue manager, it's a queue. Yes, it may own a more primitive (non-concurrent) queue that it uses as part of its implementation, but from the user's perspective, it's just a queue--the user doesn't pass it a queue to manage, it pushes tasks to be executed.

Likewise, Queue::printQueue is redundant. Queue::print seems perfectly adequate (at least to me).

Consider parameterizing the maximum size

Right now, you're using a #define to specify the maximum size for your queue. That's certainly a common way to do things in C. Different uses of your queue, however, might favor different queue depths, so I'd prefer to pass the maximum depth to the constructor, so each queue can be the right depth for its job (and if you want 100 as the default, add it as a default argument.

Prefer to emplace

std::priority_queue has an emplace member that allows you to specify arguments to the ctor, and have the actual object constructed in place instead of creating an object, then copying that to the queue.

Avoid bool parameters

Your printQueues currently takes an int parameter that's pretty much a stand-in for a Boolean (only allows two values). Unfortunately, it's not immediately obvious what printQueues(0) means or how it differs from printQueues(1). Changing those to false and true doesn't really help either.

This parameter really controls the color, so it might make sense to use something like:

enum colors { green, red };
// ...

printQueues(red);

It could also make sense to make this a little more about the program logic and less about how that's displayed:

enum class display { post_insert, post_remove };
// ...

printQueues(display::post_remove);

Either way the call does a lot more to describe what's really happening and why it's being done. It might be worth considering having the queue itself keep track of what was done more recently, and react appropriately:

void Queue::push(...) { 

     last_action = push;
}

void Queue::print() { 
    if (last_action == push)
        print(red);
    else
        print(green);
}

Separation of Concerns

Right now, this is called a queue, but it also includes the code for the consumer that gets things from the queue and executes them. Worse, the code that knows about how to execute a task from the queue is in the queue itself--it seems to me as if that would make more sense as part of the task. I can see two reasonable ways to go here:

1. Make the queue just a queue--all it has are insert, remove, and possibly a few other queue-like things such as getting the current size and possibly printing out the current contents.

2. Rename it, and make this into more of a generalized parallel execution engine.

Either way, however, the code to execute for the task should probably be defined in something like operator() in the task itself:

class task_conn {
    // stuff from before

    void operator()() { 
        // do stuff using `fd`
    }
};

Then the parallel execution engine has one or more threads that look something like this:

void Executer::run() { 
    for (;;) 
        pop()();
}

I'd also add (for example) a special task at the lowest priority to tell the execution thread to exit, so this would become more like:

enum priority { exit, unknown, min, med, max };

task t;
while ((t = pop()) > exit) // remember: comparison goes by priority
    t();

This way, once we've finished giving our execution engine jobs to do, we can push a dummy task with its priority set to exit for each thread, and they'll all exit. Since those are the lowest priority, they won't be consumed until all the other tasks have been.

To keep life simple for the user, we might consider adding a join to our task executer that will work about like join does on a single thread--except that in this case, it'll know it needs to tell the threads they're finished, then join all the threads it's spawned:

void join() { 
    for (int i=0; i<thread_count; i++)
        insert(-1, exit);
    for (int i=0; i<thread_count; i++)
        join(threads[i]);
}