Proper way of empting deques with higher priorities with a threaded removal

Question

I'm developing a producer/consumer system with three deques (each for one priority level, from max to min); first deque has major priority, so it must be read before others: once empty, others deques will be served.

My code is something like this:

// this is the class associated to a deque
class DequeEntity {
    public:
        bool remove(Elem& elem); // simple RAII methods for inserting
        void insert(Elem info);     // and removing in this deque
        size_t size() const { return q.size(); }
        bool isEmpty() { return q.empty(); }
    private:
        std::mutex m;
        std::condition_variable write, read;
        std::deque<Elem> q;
};

// this class contains the three deques
class DequeManager {
    public:
        void insert(Elem e);
        void remove(void); // <--- this is my real struggle
        void printQueues(void);
        void startRemoverThread(void); // start removerThread; called once

    private:
        std::thread removerThread;
        std::array<DequeEntity, 3> deques;
};

This is the code of bool DequeEntity::remove(Elem& elem)

bool DequeEntity::remove(Elem& elem) {
    std::unique_lock<std::mutex> lck(m);
    if (read.wait_for(lck, std::chrono::milliseconds(1000), 
        [this]() { return q.size() > 0; })) {
        try {
            elem = std::move(q.front());
            q.pop_front();
        } catch (const std::exception& e) {
            debug_red(e.what());
            return false;
        }
        write.notify_one();
        return true;
    }
    else {
        // timeout expired, failed to retrieve element from deque
        return false;
    }
}

This is the (pseudo)code of void DequeManager::remove(void)

void DequeManager::remove(void) {
    Elem elem;

    while (true) {
        numQueue = 0;
        for (auto &q : deques) {
            if (q.remove(elem)) {
                // do stuff with elem
                // if there's more to read, go ahead
                while (q.size() > 0) {
                    Elem elem;
                    if (q.remove(elem)) {
                        // do stuff with elem
                    } else {
                        // failed to get another element;
                        // go on next deque
                        break;                      
                    }
                }
            }
        }
    }
}

DequeManager::remove(void) loops over three deques, read all elements (if there are any) from first deque, then it goes on empting second deque, and same happens for the third.

I thought this was a decent implementation of priority, but I was wrong: I was told that priority is not only empting deques with higher priority, but always remove them first, regardless of the others.

For example, I just read all elements from first deque, so let's move to the second one; while empting it, new elements are inserted in the first deque: my current DequeManager::remove() method loops over all deques but will first empty the second deque (which is were we still was), the third one and finally begin the loop again, removing the newly inserted elements.

But this is not what it's supposed to do: even if it's removing from second deque and new elements are inserted in the first one, the remover thread should stop removing from deque 2, read newly inserted elements in deque 1 and (if deque 1 is actually empty), recommence to empty deque 2.

This is something I've never thought, so I have to rethink the removing strategy (thus my void DequeManager::remove(void)): how can a thread stop reading deque 2, start reading deque 1 and once finished go back to deque 2? I can't think of a way to do this with just a mutex and two condition_variable for each DequeEntity: DequeManager::remove has a normal for loop, and I can't go from deque 2 to deque 1 and then back deque2 (or deque 3 -> deque 2 -> deque 3, or deque 3 -> deque 1 -> deque3). I've been struggling on this for days, would you share with me some advices on how to deal with this kind of priority?

Answer 1

You are overthinking the problem a bit.

What you actually want is to retrieve an element from the queue with the highest priority that has an element available. Once you have processed that one element, you go back to the beginning and again look in each of the queues which one has a next element for you.

void DequeManager::remove(void) {
    Elem elem;

    while (true) {
        numQueue = 0;
        for (auto &q : deques) {
            if (q.remove(elem)) {
                // do stuff with elem

                break; /* we have processed our element, so start 
                          looking in queues again in priority order */
            }
        }
    }
}

The break in the if statement will ensure that as long as there are elements in the first (highest-prio) queue, only those elements are processed.
Once the highest-prio queue is empty, an element from the second queue can be taken (and so on for lower priorities).
Once an element has been processed, all the higher-priority queues are checked again (in priority order) if something came in in the mean time to ensure that the elements from the highest-priority queue get processed first.