I'm developing a producer/consumer system with three deque
s (each for one priority level, from max to min); first deque
has major priority, so it must be read before others: once empty, others deque
s 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 deque
s 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 deque
s 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?
[ ]
orat
): I have to deal with starvation too, and elements which were too long in thequeue
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.