A thread-safe blocking FIFO queue

Question

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.

Answer 1

template < typename _Msg, typename _Alloc = std::allocator<_Msg> >

Those names are dangerous - underscore followed by capital is reserved to the implementation for any purpose. That means they could be macros, which would give you some tricky parse errors to diagnose! Stick to names you're allowed to use (in this case, I'd just drop the underscores).

post()

{
    QMutexLocker lock_(&mx);
    que.push(msg);
    cv.wakeOne();
}

There's an inefficiency here, because waking another thread while we still hold the mutex will cause that thread to immediately block. Better to release the mutex before waking:

{
    {
        QMutexLocker lock_(&mx);
        que.push(msg);
    }
    cv.wakeOne();
}

A design choice here: should a queue be allowed to expand until it throws std::bad_alloc? Or do we want a maximum capacity at which the write end blocks?

If you have enough C++11 support, prefer to accept the argument by value, then std::move() it to the container. That avoids copying rvalue arguments.

wait()

void wait(DataType& msg)

That's a clunky interface. Prefer to just return the DataType.

I'm not a fan of the truncated "Que" name, and seem to stutter on it every time I read it. What's wrong with the full word?