Visual C++和併發線程......爲什麼發生這種情況？

Question

我必須編寫一個程序，用於併發處理線程，用於處理OpenCV Mat圖像。每個線程從隊列中選擇一個圖像，處理它並將結果放到另一個隊列中。我使用Mat圖像的線程安全模板隊列（如代碼中所示）。

但是線程的奇怪行爲是：如果我多次啓動程序，每次獲得不同結果的單個線程的詳細數目（我插入的計數器「添加」，以監視圖像處理數量由單線程）。

第一個線程（零）總是做所有它的詳細說明（在本例中爲10），但其餘的線程不做。有時每個線程都會執行10次精化，有時3次，有時5次... 2..新的更改（CONDITION VARIABLES和CRITICAL SECTION）線程只執行1次操作。

我不知道問題在哪裏......爲什麼發生這種情況。

我過去在這裏我的代碼，並要求你檢查它，並告訴我你的意見，什麼問題......我絕望。

這是代碼：

#include <opencv\cv.h> 
#include <opencv\highgui.h> 
#include <opencv2\highgui\highgui.hpp> 
#include <stdio.h> 
#include <stdlib.h> 
#include <windows.h> 
#include <process.h> 
#include <queue> 

using namespace std; 
using namespace cv; 

/*thread safe queue*/ 

template<typename T> 
class coda_concorr 
{ 
private: 
    std::queue<T> la_coda; 
    HANDLE mutex; 

public: 
    bool elemento; 
    coda_concorr() 
    { 
     mutex = CreateMutex(NULL,FALSE,NULL); 
     } 
    ~coda_concorr() 
    {} 
    void push(T& data) 
    { 
     WaitForSingleObject(mutex,INFINITE); 
     la_coda.push(data); 
     ReleaseMutex(mutex); 
     } 
    bool vuota() const 
    { 
     WaitForSingleObject(mutex,INFINITE); 
     bool RetCode = la_coda.empty(); 
     ReleaseMutex(mutex); 
     return RetCode; 
    } 
    bool try_pop(T& popped) 
    { 
     WaitForSingleObject(mutex,INFINITE); 
     while (la_coda.empty()){ 
      ReleaseMutex(mutex); 
      return false; 
     } 
     WaitForSingleObject(mutex,INFINITE); 
     popped = la_coda.front(); 
     la_coda.pop(); 
     ReleaseMutex(mutex); 
     return true; 
    } 
}; 


struct Args 
{ 
    coda_concorr<cv::Mat> in; 
    coda_concorr<cv::Mat> *out; //puntatore a coda successiva 
}; 


CONDITION_VARIABLE NonVuoto1; 
CONDITION_VARIABLE NonVuoto2; 
CONDITION_VARIABLE NonVuoto3; 
CONDITION_VARIABLE NonVuoto4; 
CRITICAL_SECTION Lock1; 
CRITICAL_SECTION Lock2; 
CRITICAL_SECTION Lock3; 
CRITICAL_SECTION Lock4; 

bool stop; 

//initial populating queue 
void puts (void* param){ 
    Args* arg = (Args*)param; 
    int i=0; 
    Mat image; 

    while(!arg->in.vuota()){ 
     arg->in.try_pop(image); 
     arg->out->push(image); 
     i++;   
     WakeConditionVariable(&NonVuoto1); 
     } 
    //fine 
    cout<<endl<<"Thread (PUSH) terminato con "<<i<<" elaborazioni."<<endl; 
    WakeConditionVariable(&NonVuoto1); 
    _endthread(); 
} 

//grey funct 
void grey (void *param){ 
    Mat temp1,temp2; 
    int add = 0; 
    Args* arg = (Args*)param; 
    while(true){ 
     EnterCriticalSection(&Lock1); 
     //se vuoto 
     while(arg->in.vuota() && !stop){ 
      SleepConditionVariableCS(&NonVuoto1,&Lock1,INFINITE); 
      } 
      if(stop==true){ 
      LeaveCriticalSection(&Lock1); 
      break; 
      } 
     arg->in.try_pop(temp1); 
     cvtColor(temp1,temp2,CV_BGR2GRAY); 
     arg->out->push(temp2); 
     add++; 
     cout<<endl<<"grey ha fatto: "<<add<<endl; 
     LeaveCriticalSection(&Lock1); 
     WakeConditionVariable(&NonVuoto2); 
     } 
    //fine 
    cout<<endl<<"Thread (GREY) terminato con "<<add<<" elaborazioni."<<endl; 
    _endthread(); 
} 

//threshold funct 
void soglia(void *param){ 
    Mat temp1a,temp2a; 
    int add=0; 
    Args* arg = (Args*)param; 
    while(true){ 
     EnterCriticalSection(&Lock2); 
     while(arg->in.vuota() && stop == false){ 
      SleepConditionVariableCS(&NonVuoto2,&Lock2,INFINITE); 
      } 
     if(stop==true){ 
      LeaveCriticalSection(&Lock2); 
      break; 
      } 
     arg->in.try_pop(temp1a); 
     threshold(temp1a,temp2a,128,255,THRESH_BINARY); 
     arg->out->push(temp2a); 
     add++; 
     LeaveCriticalSection(&Lock2); 
     WakeConditionVariable(&NonVuoto3); 
     cout<<endl<<"soglia ha fatto: "<<add<<endl; 
     } 
     //fine 
    cout<<endl<<"Thread (SOGLIA) terminato con "<<add<<" elaborazioni."<<endl; 
    _endthread(); 
} 

//erode/dilate funct 
void finitura(void *param){ 
    Mat temp1b,temp2b,temp2c; 
    int add = 0; 
    Args* arg = (Args*)param; 
    //come consumatore 
    while(true){ 
     EnterCriticalSection(&Lock3); 
     while(arg->in.vuota() && stop == false){ 
      SleepConditionVariableCS(&NonVuoto3,&Lock3,INFINITE); 
      } 
     if(stop==TRUE){ 
      LeaveCriticalSection(&Lock3); 
      break; 
      } 
     arg->in.try_pop(temp1b); 
     erode(temp1b,temp2b,cv::Mat()); 
     dilate(temp2b,temp2c,Mat()); 
     arg->out->push(temp2c); 
     add++; 
     LeaveCriticalSection(&Lock3); 
     WakeConditionVariable(&NonVuoto4); 
     cout<<endl<<"erode ha fatto: "<<add<<endl; 
     } 
    //fine 
    cout<<endl<<"Thread (ERODE) terminato con "<<add<<" elaborazioni."<<endl; 
    _endthread(); 
} 

//contour funct 
void contorno (void *param){ 
    Mat temp; 
    int add=0; 
    Args* arg = (Args*)param; 
    //come consumatore 
    while(true){ 
     EnterCriticalSection(&Lock4); 
     while(arg->in.vuota() && stop == false){ 
      SleepConditionVariableCS(&NonVuoto4,&Lock4,INFINITE); 
      } 
     if(stop==TRUE){ 
      LeaveCriticalSection(&Lock4); 
      break; 
      } 
    //esegue pop 
    arg->in.try_pop(temp); 
    //trova i contorni 
    vector<vector<Point>> contorni; 
    findContours(temp,contorni,CV_RETR_LIST, CV_CHAIN_APPROX_SIMPLE); 
    //disegna i contoni in un'immagine 
    Mat dst(temp.size(), CV_8UC3, Scalar(0,0,0)); 
    Scalar colors[3]; 
    colors[0] = Scalar(255,0,0); 
    colors[1] = Scalar(0,255,0); 
    colors[2] = Scalar(0,0,255); 
    for (size_t idx = 0; idx < contorni.size(); idx++){ 
     drawContours(dst,contorni,idx,colors[idx %3]); 
     } 

    //come produttore 
    arg->out->push(dst); 
    add++; 
    cout<<endl<<"cont ha fatto: "<<add<<endl; 
    LeaveCriticalSection(&Lock4); 
    } 
    cout<<endl<<"Thread (CONTOUR) terminato con "<<add<<" elaborazioni."<<endl; 
    _endthread(); 
} 

//main 
int main() 
{ 

    coda_concorr<cv::Mat> ingresso; 
    coda_concorr<cv::Mat> uscita; 

    InitializeConditionVariable(&NonVuoto1); 
    InitializeConditionVariable(&NonVuoto2); 
    InitializeConditionVariable(&NonVuoto3); 
    InitializeConditionVariable(&NonVuoto4); 
    InitializeCriticalSection(&Lock1); 
    InitializeCriticalSection(&Lock2); 
    InitializeCriticalSection(&Lock3); 
    InitializeCriticalSection(&Lock4); 


    LARGE_INTEGER count1, count2, freq; 
    double elapsed; 


    Mat temp[10]; 
    Mat out; 

    //dichiarazione code 
    Args dati0,dati1,dati2,dati3,dati4; 


    //avvio contatori 
    QueryPerformanceFrequency(&freq); 
    QueryPerformanceCounter (&count1); 

    for(int i=0;i<10;i++){ 
     temp[i] = imread("C:/OPENCV/Test/imgtest/bird1.jpg",1); 
     ingresso.push(temp[i]); 
    } 

    //next queue pointer 
    dati0.in=ingresso; 
    dati0.out=&dati1.in; 
    dati1.out=&dati2.in; 
    dati2.out=&dati3.in; 
    dati3.out=&dati4.in; 
    dati4.out=&uscita; 



    //handle 
    HANDLE handle0,handle1,handle2,handle3,handle4; 

    //start threads 
    handle0 = (HANDLE) _beginthread(puts,0,&dati0); 
    handle1 = (HANDLE) _beginthread(grey,0,&dati1); 
    handle2 = (HANDLE) _beginthread(soglia,0,&dati2); 
    handle3 = (HANDLE) _beginthread(finitura,0,&dati3); 
    handle4 = (HANDLE) _beginthread(contorno,0,&dati4); 

    cout<<endl<<"..Join dei threads..."<<endl; 

    //join 
    WaitForSingleObject(handle0,INFINITE); 
    WaitForSingleObject(handle1,INFINITE); 
    WaitForSingleObject(handle2,INFINITE); 
    WaitForSingleObject(handle3,INFINITE); 
    WaitForSingleObject(handle4,INFINITE); 



    //chiusura contatori 
    QueryPerformanceCounter (&count2); 

    CloseHandle(handle0); 
    CloseHandle(handle1); 
    CloseHandle(handle2); 
    CloseHandle(handle3); 
    CloseHandle(handle4); 

    elapsed = (count2.QuadPart - count1.QuadPart) * 1000.0/freq.QuadPart; 


    cout <<endl<<"Tempo di esecuzione approssimativo: " <<elapsed<<" ms."<<endl; 
     system("PAUSE"); 
    return 0; 
} 

如果previuos線程把所有隊列它的圖像，爲什麼一個線程不一樣嗎？

我在64位Windows 7用Visual C++ 2010的OpenCV 2.4.4

請幫我看看哪裏出了問題......

Answer 1

看來要實現的是類似的在一個工廠裏進行流水線工作，在這個工廠裏，每個人都做好自己的工作，然後把它交給下一個工人，直到完成所有工作。如果我錯了，請糾正我。你的每個線程函數的成語是

void dowork(){ 
    while(noinput()){ 
     sleep(0.01); 
    } 
    while(getSomeInput()){ 
     processInput(); 
     queueResult(); 
    } 
    displayAmountOfWorkDone(); 
} 

你成功地提供互斥與你的互斥。你的設計的問題是，一旦一個線程觀察到他的輸入隊列是非空的，他就會消耗所有的工作，然後退出。由於線程調度和處理時間processInput()，工人可以以比他之前可以生產的工人更高的速度消耗他的輸入。例如，在init和datit之間進行排隊，這是可能的：

datit: see 0, sleep 
init : see 10 - add 1 
datit: see 1, process 
datit: see 0, exit and outputs 1 
init : add 2 
init : add 3 
init : add 4 
.... 
init : add 10 

您需要更改設計。 應該有一個不同的機制來表明工作已經結束了，其中有。現在您正在使用線程可以觀察的輸入量。一個快速和骯髒的解決將是給每個線程的他有望來處理輸入量，並重寫算法

void dowork(){ 
    while(workIsnotDone()){//local test, no lock 
     if(getSomeInput()){ 
      processInput(); 
      queueResult(); 
      updateWorkProgress();//local operation 
     } 
     else{ 
      sleep(0.01); 
     } 
    } 
    displayAmountOfWorkDone(); 
} 

一個更好的替代方法是設置類coda_concorr作爲生產者 - 消費者機制。爲此你可以添加一個條件變量。每個線程都是一個隊列上的消費者，另一個隊列上是一個製作者你也可以添加一個字段來顯式指定類似沒有更多的輸入。看看this other question on SO