“Mat”的线程和线程安全队列图像...为什么线程不处理所有图像？ [英] Threads and thread-safe queue of "Mat" images...why threads don't process all images?

问题描述

我已经编写了这段代码，用于处理带有线程和并发队列的图像......我创建了一个线程安全的队列模板，使用mutex来管理队列的读/写。



现在，我已经为信令工作引入了CRITICAL SECTION和CONDITION VARIABLES ...但情况是所有线程（除了第一个明确的原因）只做1次操作。



这是代码：

Hi, I have wrote this code, for processing images with threads and concurrent queues...I have create a thread-safe queue template, with mutex for manage reading/writing of queue.

Now, I have introduced CRITICAL SECTION and CONDITION VARIABLES for signaling work...but the situation is that all thread (except the first for clear reason) doing only 1 operation.

This is the code:

#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;
        }

并发不起作用：此程序是多个生产者/多个消费者，因为每个线程都是生产者和消费者（队列的消费者，另一个队列的生产者）......在我看来，问题不在队列代码中，而是在线程的同时...但是我不知道如何解决这个问题...引入另一个检查？怎么样？



请帮帮我...我很绝望......

The concurrency don't work: This program is multiple producer/multiple consumer, because every threads is both a producer and consumer (consumer for a queue, producer for another queue)...in my opinion the problem is not in the queue code, but in the while cicle of the threads...but I don't know how fix this problem...introducing another check? How?

Please help me...I'm desperate...

推荐答案

我有很难理解你的代码以及问题描述。我认为你有一个多线程的应用程序很不高兴，很多时候你运行它，所有的工作只由一个线程完成。



请记住线程创建是一种资源密集型活动，如果要完成的工作相当简短，那么在创建剩余线程之前，所有工作都可以由其中一个线程完成。换句话说，程序中的所有测试代码都可以在上面的第一个beginthread（）调用之后有效地完成。



要对此类活动进行基准测试，您应该：

- 首先创建所有主题（在开始任何主题之前）

- 使活动/活动的持续时间更长

- 多次运行循环并丢弃（或折扣）前几次运行。

I'm having a hard time both understanding your code as well as the problem description. I think you are unhappy that you have a multithreaded app and many times when you run it, all the work is being done by only one thread.

Keep in mind that thread creation is a resource intensive activity, and if the work to be done is fairly brief, all of the work can be completed by one of the threads before the remaining threads are created. In other words, all of the test code in your program can be effectively completed following the first beginthread() call above.

To benchmark this kind of activity, you should:
- create all threads first (before starting the work in any of them)
- make the activity/activities longer in duration
- run the loop(s) multiple times and discard (or discount the value of) the first several runs.

这篇关于“Mat”的线程和线程安全队列图像...为什么线程不处理所有图像？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！