Concurrency issues with QThreads. Threads receiving the same signal are blocking each other

Question

So I'm working on a program that processes video in real-time and I'm having some trouble with threads "blocking" each other.

My system is pretty much set up like this:

         DataSourceThread
             /      \
            /        \
           /          \
     Receiver       Receiver
         /              \
        /                \ 
       /                  \
 Processor1            Processor2

(All these classes are extending QThread.)

So DataSourceThread fetches frames from a videostream and emits a signal containing the frame to the receivers. Connection type: Qt::DirectConnection

The receivers basically receive the the frames sent out by DataSourceThread and if the Processor is done processing the previous frame, it will emit a signal containing the frame to the Processor. Connection type: Qt::QueuedConnection. If the processor is not done processing the previous frame, it will just return without emitting a signal (skipping frames).

To test if this works, all I've done is have Processor1 just print out a message when it receives a frame and Processor2 does QThread::sleep(3); and print out a message.

(The receivers will also do a deep copy of the frame before passing it on to the processors.)

Expected result:

Processor1 should be constantly printing messages. Processor2 should be printing a message every 3 seconds.

The problem:

Both processors print their messages at the same time (every 3 seconds). Processor1 waits until Processor2 is done before printing its message. So the output is pretty much like this:

"Message from processor1"
"Message from processor2"
"Message from processor1"
"Message from processor2"
"Message from processor1"
"Message from processor2"

and so on.

I'm running out of ideas here, so any help would be greatly appreciated!

EDIT: Here's some of the code:

main.cpp:

int main(int argc, char *argv[])
{
    QApplication app(argc, argv);

    DataSourceThread dataSourceThread;
    dataSourceThread.start();

    GUIThread *guiProcessor = new GUIThread();
    FrameReceiver *guiReceiver = new FrameReceiver(guiProcessor, 0);

    QObject::connect(
        &dataSourceThread, SIGNAL(frameReceived(Frame*)),
        guiReceiver, SLOT(receive(Frame*)),
        Qt::DirectConnection
    );

    DetectorThread *detectorProcessor = new DetectorThread();
    FrameReceiver *detectorReceiver = new FrameReceiver(detectorProcessor, 0);

    QObject::connect(
        &dataSourceThread, SIGNAL(frameReceived(Frame*)),
        detectorReceiver, SLOT(receive(Frame*)),
        Qt::DirectConnection
    );

    return app.exec();
}  

From DataSourceThread.cpp:

void DataSourceThread::run()
{
    ... stuff ...

    while (true) {
        image = cvQueryFrame(capture);

        if (!image) { 
            qDebug() << QString("Could not capture frame"); 
            continue;
        }

        cvReleaseImage(&temp_image);
        temp_image = cvCreateImage(cvSize(640, 480), image->depth, 3);

        cvResize(image, temp_image, 1);

        frame->lock();
        frame->setImage(temp_image);
        frame->unlock();

        emit frameReceived(frame);

        msleep(1); 
    }
} 

FrameReceiver.cpp:

FrameReceiver::FrameReceiver(FrameProcessor* processor, QObject *parent) : QThread(parent) {
    m_ready = true;

    m_processor = processor;
    m_processor->start();

    QObject::connect(
        (QObject*)this, SIGNAL(frameReceived(Frame*)),
        m_processor, SLOT(receive(Frame*)), 
        Qt::QueuedConnection
    );

    QObject::connect(
        m_processor, SIGNAL(ready()),
        (QObject*)this, SLOT(processCompleted()),
        Qt::DirectConnection
    ); }

void FrameReceiver::processCompleted() {
    m_ready = true; }

void FrameReceiver::receive(Frame *frame) {
    if (m_ready == true) {
        m_ready = false;
        frame->lock();
        Frame *f = new Frame(*frame);
        frame->unlock();
        emit frameReceived(f);
    } else {
        // SKIPPED THIS FRAME
    }
}

GUIThread.cpp: (Processor1)

GUIThread::GUIThread(QObject *parent) : FrameProcessor(parent)
{
    m_frame = new Frame();
}

void GUIThread::setFrame(Frame *frame)
{ 
    qDebug() << QString("Guithread received frame");
}    

FrameProcessor.cpp

// (The processors extend this class)
void FrameProcessor::receive(Frame *frame)
 {
     setFrame(frame);
     delete frame;
     emit ready();
 }

DetectorThread (Processor2) does the same as guithread, but with a 3 sec sleep in setFrame.

Answer 1

I think part of the problem is that all of your QObjects are owned by the main application thread. This means they all share a one event loop for delivering asynchronous signals, effectively serializing your whole processing chain.

I think the proper way to set this up would be something like:

GUIProcessor *guiProcessor = new GUIProcessor();
QThread guiProcessorThread;
guiProcessor.moveToThread(&guiProcessorThread);

FrameReceiver *guiReceiver = new FrameReceiver(guiProcessor, 0);
QThread guiReceiverThread;
guiReceiver.moveToThread(&guiReceiverThread);

guiProcessorThread.start();
guiReceiverThread.start();

If you do it this way I would suggest not using DirectConnection between threads, but rather BlockingQueuedConnection if you want to ensure that the current frame is processed before you capture the next one.

See this: http://labs.qt.nokia.com/2010/06/17/youre-doing-it-wrong/

And this: http://labs.qt.nokia.com/2006/12/04/threading-without-the-headache/

Hope this helps!

EDIT: To be clear, with my suggestion your classes would inherit QObject instead of QThread.