1

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?

  • An LMax Disruptor might work well here. There is supposedly a C++ version; it might be worth a look: sourceforge.net/projects/cxxdisruptor – Robert Harvey Jul 28 '16 at 14:41
  • You could use a single priority queue. Each item has a priority, and you insert items in the queue, keeping the queue in priority order. – Mike Dunlavey Jul 28 '16 at 17:39
  • @RobertHarvey sorry I can only write code of my own, without external libraries. @MikeDunlavey I need a random access ([ ] or at): I have to deal with starvation too, and elements which were too long in the queue should be popped() from where they are and pushed() to the top; this elements might be everywere in the queue, so I need a random access. – elmazzun Jul 29 '16 at 6:37
4

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.

  • This is how I did it in the beginning but then I found out that the priority is something "absolute": I have to remove high prio elements, all of them, then go to medium prio deque, begin to remove; if newly high prio elements are pushed in high prio deque, stop removing from second deque, remove from first deque and then come back to remove from second deque. This is the trick that makes me go mad. – elmazzun Jul 29 '16 at 11:34
  • @elmazzun`: see my update. – Bart van Ingen Schenau Jul 29 '16 at 11:49
  • 1
    One nitpick: DequeEntity::remove will need to let the caller choose the length of timeout. Otherwise, DequeEntity::remove can be waiting for a whole second on a lower priority queue, during which a new arrived item in a higher priority queue will be ignored. – rwong Jul 29 '16 at 16:52

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.