A legacy project was upgraded from legacy code that used either SystemV FIFO device (or similar queue on Windows platform) which was deeply ingrained into every thread's implementation. Following implementation of locking queue with a wait was meant to emulate original FIFO use:
#include <queue>
// target platform doesn't have std::mutex or support >C++14
#include <QWaitCondition>
#include <QMutex>
template < typename _Msg, typename _Alloc = std::allocator<_Msg> >
class WaitQue
{
public:
typedef _Alloc Allocator;
typedef _Msg DataType;
// a thread calls this to place a copy of msg
void post(const DataType& msg)
{
QMutexLocker lock_(&mx);
que.push(msg);
cv.wakeOne();
}
// a thread calls this to to wait for a message
void wait(DataType& msg)
{
/// wait if que empty
QMutexLocker cvlock_(&mx);
while(que.empty()) // spurious exit from wait()
cv.wait(&mx);
msg = que.front();
que.pop();
}
unsigned long size()
{
QMutexLocker lock_(&mx);
return que.size();
}
private:
typedef std::deque<DataType, Allocator> Container;
std::queue<DataType, Container> que;
QWaitCondition cv;
QMutex mx;
};
A simplest use of such queue without any fluff:
#include <QCoreApplication>
#include <QtConcurrent>
#include <QFuture>
struct Message {
int value;
};
using Queue = WaitQue<Message>;
Queue east; // messages for east-side events
Queue west; // messages for west-side events
void westSide()
{
Message msg = {};
for(int i = 5; i-->0; )
{
qDebug() << "West side is working." << endl;
east.post(Message{i}); //
west.wait(msg); //
}
}
void eastSide()
{
Message msg = {};
do
{
east.wait(msg);
qDebug() << "East side received " << msg.value << endl;
west.post(msg);
}while(msg.value);
}
int main(int argc, char *argv[])
{
QCoreApplication a(argc,argv);
QFuture<void> future1 = QtConcurrent::run(westSide);
QFuture<void> future2 = QtConcurrent::run(eastSide);
future1.waitForFinished();
future2.waitForFinished();
return 0;
}
In reality most of workers are infinite state-machine loops operating only after receiving messages or generating such messages for other workers while some are actually doing ping-pong exchange as above.
Target platform got partial C++11 support and Qt < 5.4 support. std::queue
was used instead of Qt container for sake of custom allocator support and condition variable QWaitCondition
is used to lock queue. Custom allocator allows to avoid heap allocation. Is this correct implementation and if its not, what flaws are there? Is there a Qt analog for this maybe?
PS. Indefinite wait was a known problem that was solved by design of any scenario was ending with posting "stop thread" messages to all queues in past implementation of project.