Проблемы параллелизма с QThreads. Нити, получающие один и тот же сигнал, блокируют друг друга

Question

Так что я работаю над программой, обрабатывающей видео в режиме реального времени, и у меня возникают проблемы с «блокировкой» потоков.

Моя система в значительной степени настроить так:

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

(Все эти классы расширяют QThread.)

Так DataSourceThread извлекает кадры из видеопотока и излучает сигнал, содержащий раму к приемников. Тип соединения: Qt :: DirectConnection

Приемники основном получить кадры, посылаемые от DataSourceThread, и если процессор выполняется обработка предыдущего кадра, он будет излучать сигнал, содержащий раму с процессором. Тип подключения: Qt :: QueuedConnection. Если процессор не завершил обработку предыдущего кадра, он просто вернется без излучения сигнала (пропускает кадры).

Чтобы проверить, работает ли это, все, что у меня было, это заставить Processor1 просто распечатать сообщение, когда он получает кадр, а Processor2 - QThread :: sleep (3); и распечатать сообщение.

(Приемники будут также делать глубокую копию кадра перед передачей его на процессорах.)

Ожидаемый результат:

процессора1 должны постоянно печать сообщений. Процессор2 должен печатать сообщение каждые 3 секунды.

Проблема:

Оба процессора печатать свои сообщения одновременно (каждые 3 секунды). Процессор1 ожидает завершения процесса обработки процессора до его сообщения. Так что выход довольно много, как это:

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

и так далее.

У меня заканчиваются идеи здесь, , поэтому любая помощь будет принята с благодарностью!

EDIT: Вот некоторые из кода:

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(); 
} 

От 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: (процессора1)

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) делает то же самое, как guithread, но с 3 сек сна в setFrame.

Answer 1

Я думаю, что часть проблемы в том, что все ваши объекты QObject принадлежат основному потоку приложения. Это означает, что все они используют один цикл событий для доставки асинхронных сигналов, эффективно сериализуя всю цепочку обработки.

Я думаю, что правильный способ установить это будет что-то вроде:

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

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

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

Если вы делаете это таким образом, я предложил бы не использовать DirectConnection между потоками, а BlockingQueuedConnection, если вы хотите, чтобы убедиться, что текущий кадр обрабатывается перед захватом следующего.

Смотреть это: http://labs.qt.nokia.com/2010/06/17/youre-doing-it-wrong/

И это: http://labs.qt.nokia.com/2006/12/04/threading-without-the-headache/

Надеется, что это помогает!

EDIT: Чтобы было ясно, с моим предложением ваши классы наследуют QObject вместо QThread.