Multithreaded data processing pipeline in Qt

Question

What would be a good way to solve the following problem in Qt:

I have a sensor class, which continuously produces data. On this data, several operations have to be performed after another, which may take quite long. For this I have some additional classes. Basically, every time a new data item is recorded, the first class should get the data, process it, pass it to the next and so on.

sensor --> class 1 --> ... --> last class

I want to put the individual classes of the pipeline into their own threads, so that class 1 may already work on sample n+1 when class 2 is processing sample n...

Also, as the individual steps may differ greatly in their performance (e.g. the sensor is way faster than the rest) and I'm not interested in outdated data, I want class 1 (and everything after it) to always get the newest data from their predecessor, discarding old data. So, no big buffer between the steps of the pipeline.

First I thought about using Qt::QueuedConnections for signals/slots, but I guess that this would introduce a queue full of outdated samples waiting to be processed by the slower parts of the pipeline?

Answer 1

Just build your own one-element "queue" class. It should have:

A piece of data (or pointer to data)
A Boolean "dataReady"
A mutex
A condition variable

The "enqueue" function is just:

lock mutex
Replace data with new data
dataReady = true
signal condition variable

The "dequeue" function is just:

lock mutex
while (!dataReady) cond_wait(condition, mutex)
tmpData = data
data = NULL (or zero)
dataReady = false
unlock mutext
return tmpData

The type of the data can be a template parameter.

Answer 2

What you are dealing with is a Producer Consumer Pattern. You can find a general overview of that here. http://en.wikipedia.org/wiki/Producer-consumer_problem

You want to use a QMutex to limit access to the data to one thread at a time. Use the QMutexLocker to lock it.

For a VERY simplified example:

QList<quint32> data;
QMutex mutex;

//  Consumer Thread calls this
int GetData()
{
   quint32 result(-1); // if =1 is a valid value, you may have to return a bool and 
                       // get the value through a reference to an int 
                       // in the parameter list.

   QMutexLocker lock(&mutex);

   if (data.size())
   {
      result = data.front();  // or back
      data.clear();
   }

   return result;
}

// Producer Thread calls this
void SetData(quint32 value)
{
    QMutexLocker lock(&mutex);

    data.push_back(value);
}