2
\$\begingroup\$

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.

\$\endgroup\$
0
2
\$\begingroup\$
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?

\$\endgroup\$
4
  • \$\begingroup\$ Are you sure about post()? all examples of use show opposite (usually with direct lock() for Qt examples). It is possible that two threads would wake one thread once while posting two items in queue because one would slip in-between .unlock() and .wakeOne()? \$\endgroup\$ – Swift Feb 18 at 7:36
  • 1
    \$\begingroup\$ That's the advice I've seen for the Standard Library condition variables; it's hard to see why it wouldn't also be good advice for the Qt types too. I can't quite see a path where wait() could find the queue empty and miss being woken (if we enter the mutex on the reader side between unlock and wake, then the queue is not empty, because releasing the mutex on the writer side is a barrier). OTOH, it is possible that two readers arrive, one takes the item and the other gets woken with an empty queue, but that's already handled well. \$\endgroup\$ – Toby Speight Feb 18 at 8:04
  • \$\begingroup\$ I was afraid f getting problem with that part - std::queue::empty() isn't said to be thread-safe or atomic. \$\endgroup\$ – Swift Feb 18 at 8:40
  • \$\begingroup\$ We hold the mutex when calling empty() so I'm satisfied about that. I wonder if the differing advice to wake whilst holding the lock comes from Java programmers. I vaguely remember that's a thing to do with Java CVs. \$\endgroup\$ – Toby Speight Feb 18 at 11:45

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.