Multiple queue of images, multiple threads...how?

Question

3.22/5 (2 votes)

See more:

Hi,
I have a problem. I have to create a program that manage 4 queues, 3 threads, like in the diagram that I've linked.

THIS IS A DIAGRAM OF WHAT I HAVE TO DO:

http://img839.imageshack.us/img839/3196/schemah.jpg[^]

I have started to work on, but concurrency with "shared" queue works only with 2 threads and three don't.

I post here my code, wait for your response.Watch my code, give me an idea on what's wrong.

I HAVE UPDATE MY CODE NOW..Please help me. Thanks in advance.

C++

#include <opencv\cv.h>
#include <opencv\highgui.h>
#include <stdio.h>
#include <windows.h>
#include <process.h>
#include <queue>
#include <time.h>

using namespace std;
using namespace cv;


/*inizio della parte relativa alla coda thread safe*/

template<typename T>
class coda_concorr
{
private:
	std::queue<T> la_coda;
	HANDLE mutex;
public:
	bool complete;
	coda_concorr()
	{
		mutex = CreateMutex(NULL,FALSE,NULL);
		complete = false;
	}
	void push(T& data)
	{
		WaitForSingleObject(mutex,INFINITE);
		la_coda.push(data);
		ReleaseMutex(mutex);
	}
	bool vuota() const
	{
		bool RetCode;
		WaitForSingleObject(mutex,INFINITE);
		RetCode= la_coda.empty();
		ReleaseMutex(mutex);
		return RetCode;
	}

	bool try_pop(T& popped)
    {
        WaitForSingleObject(mutex,INFINITE);
        while (la_coda.empty())
        {
            if (complete)
            {
                ReleaseMutex(mutex);
                return false;
            }
            else
            {
                ReleaseMutex(mutex);
                Sleep(100); // FUTURE IMPROVEMENT
                WaitForSingleObject(mutex,INFINITE);
            }
        }
 
        popped = la_coda.front();
        la_coda.pop();
        ReleaseMutex(mutex);
        return true;
    }
};

//pacchetto per passaggio ai thread
struct Args
{
	coda_concorr<cv::Mat> in;
	coda_concorr<cv::Mat> *out; //puntatore a coda successiva
};

After declare the 3 functions that I've to launch:

C++

//funzione grey
void grey (void *param){
	Mat temp1,temp2;
	Args* arg = (Args*)param;
	if(!arg->in.vuota()){
	while(arg->in.try_pop(temp1)){
	cvtColor(temp1,temp2,CV_BGR2GRAY);
	arg->out->push(temp2);
	   }
	}
	//flag completato
	arg->out->complete=true;
	_endthread();
	
}

//funzione sogliatura
void soglia(void *param){
	Mat temp1a,temp2a;
	Args* arg = (Args*)param;
		while(arg->in.try_pop(temp1a)){
		threshold(temp1a,temp2a,128,255,THRESH_BINARY);
		arg->out->push(temp2a);
		}
		//flag completato
	 arg->out->complete=true;
	_endthread();
}


//funzione erosione/dilate
void finitura(void *param){
	Mat temp1b,temp2b,temp2c;
	Args* arg = (Args*)param;
		while(arg->in.try_pop(temp1b)){
		erode(temp1b,temp2b,cv::Mat());
		//dilate(temp2b,temp2c,Mat());
		arg->out->push(temp2b);
		}
		//flag completato
	 arg->out->complete=true;
	_endthread();
}

then this is main:

int main()
{
	//dichiarazione delle due code principali
	coda_concorr<cv::Mat> ingresso;
	coda_concorr<cv::Mat> uscita;

	//clock
	clock_t avvio;
	clock_t fine;

	
	//array locali di ingresso e uscita
	Mat inn[10];
	Mat out;

	//assegnazione 10 immagini di prova
	inn[0]=imread("C:/OPENCV/Test/imgtest/bird1.jpg",1);
	inn[1]=imread("C:/OPENCV/Test/imgtest/bird2.jpg",1);
	inn[2]=imread("C:/OPENCV/Test/imgtest/bird3.jpg",1);
	inn[3]=imread("C:/OPENCV/Test/imgtest/pig1.jpg",1);
	inn[4]=imread("C:/OPENCV/Test/imgtest/pig2.jpg",1);
	inn[5]=imread("C:/OPENCV/Test/imgtest/pig3.jpg",1);
	inn[6]=imread("C:/OPENCV/Test/imgtest/spider1.jpg",1);
	inn[7]=imread("C:/OPENCV/Test/imgtest/spider2.jpg",1);
	inn[8]=imread("C:/OPENCV/Test/imgtest/spider3.jpg",1);
	inn[9]=imread("C:/OPENCV/Test/imgtest/Nutella.jpg",1);

	
	//dichiarazione code
	Args dati,dati2,dati3,dati4;

	//inizio a contare tempo
	avvio = clock();

	//Popolamento coda ingresso
	for(int i=0;i<=9;i++){
		dati.in.push(inn[i]);
	}
	//punta all'output successivo
	dati.out=&dati2.in;
	dati2.out=&dati3.in;
	dati3.out=&uscita;
	
	//dichiarazione delle maniglie
	HANDLE handle1,handle2,handle3;

	//dichiarazione dei threads
	handle1 = (HANDLE) _beginthread(grey,0,&dati);
	handle2 = (HANDLE) _beginthread(soglia,0,&dati2);
	handle3 = (HANDLE) _beginthread(finitura,0,&dati3);

	//join
	WaitForSingleObject(handle1,INFINITE);
	WaitForSingleObject(handle2,INFINITE);
	WaitForSingleObject(handle3,INFINITE);
	
	//fine registrazione tempo
	fine = clock()-avvio;

	double tempoimp = fine / (double)CLOCKS_PER_SEC;

//output di verifica
	while (dati3.out->try_pop(out)){
		imshow("immagine",out);
		waitKey(100);
	}
	
	cout<<endl<<"Cicli di clock: "<<fine<<endl;
	cout<<"Secondi trascorsi: "<<tempoimp<<endl;

	system("PAUSE");
	return 0;

}

I don't have any output...Can someone help me?

Enviroment is: Visual C++ 2010, OpenCV 2.4.3 on WIndows 7 64bit.

Posted 5-Jun-13 5:57am

Domus1919

Updated 6-Jun-13 21:08pm

v12

Add a Solution

Comments

Steve44 5-Jun-13 15:25pm

What is exactly happening in the second thread? Is it immediately terminating and not doing any work? Or is it doing work, but the results are incorrect?

Domus1919 5-Jun-13 17:11pm

Not execute...program blocked.Watch diagram in the link that I've just posted...that it's what I have to do.

Sergey Alexandrovich Kryukov 5-Jun-13 15:36pm

This almost all the code dump. You really need to provide detail explanation, starting with the explanation of your goals.
—SA

Domus1919 5-Jun-13 17:11pm

Wow, nothing less is "dump" ... if I could put an outline of what I should do, I placed it. What did not you understand? I have posted a link with diagram.

Sergey Alexandrovich Kryukov 5-Jun-13 17:28pm

Thank you for clarification. However, I don't want anything specifically, I only want to help you, if possible. This is you who wants some help, if I understand your post right.
—SA

Domus1919 5-Jun-13 17:30pm

Exactly. Thanks for your attention. Have you see diagram? Tell me what you think...

Sergey Alexandrovich Kryukov 5-Jun-13 17:40pm

Nothing special. Looks quite reasonable, in fact, except the fixed queue sizes (in fact, it would be almost as no queue at all, almost as a single shared object; to make queues effective, you should allow a good number of items in it; this is a whole point of buffering). Implementation is also nothing special.

Code design looks clumsy, at least at first glance. A good approach would be this: thread synchronizations should be encapsulated inside queue object. This is a whole point of queues as inter-thread (or possibly IPC) communication. You should use a blocking queue; is is what you use? The dequeue operation should put a thread in a wait state, without using any spin wait. Is this how you are doing it?

Now, is it Windows? (Tag the platform.) If it is, and if this is inside the same process, you should use lighter CriticalSection instead of Mutex.

—SA

Domus1919 5-Jun-13 17:53pm

Yes, I'm on Visual C++ 2010. I have put synchronization in queue, then you advice of use "Critical Section" instead mutex? Directly substituition? Tomorrow I'll try, an let you know about. For now, thanks for your time. Regards.

Sergey Alexandrovich Kryukov 5-Jun-13 18:04pm

Well, nearly one-to-one substitution, method-by-method. The logic is 100% the same. Named mutex is for IPC, unnamed is good for inter-thread of the same process, and CriricalSection is functionally equivalent to nameless mutex, only lighter.
—SA

Sergey Alexandrovich Kryukov 5-Jun-13 18:06pm

Add the tag "Windows": click here.
It will help other members qualified in Windows determine that they can load this page by just looking at the question list.
—SA

Domus1919 6-Jun-13 5:55am

I added tag windows :) Thanks for your time. I HAVE JUST UPDATE MY POSTED CODE...see it. Second thread always see input queue empty...what's wrong? I have try also with CriticalSection, but the result it's the same...

2 solutions

Add a Solution

Add your solution here

Treat my content as plain text, not as HTML

Preview 0

…

Existing Members

Sign in to your account

...or Join us

Download, Vote, Comment, Publish.

Your Email
Password
Forgot your password?

Your Email
This email is in use. Do you need your password?
Optional Password

I have read and agree to the Terms of Service and Privacy Policy
Please subscribe me to the CodeProject newsletters

When answering a question please:

Read the question carefully.
Understand that English isn't everyone's first language so be lenient of bad spelling and grammar.
If a question is poorly phrased then either ask for clarification, ignore it, or edit the question and fix the problem. Insults are not welcome.
Don't tell someone to read the manual. Chances are they have and don't get it. Provide an answer or move on to the next question.

Let's work to help developers, not make them feel stupid.

This content, along with any associated source code and files, is licensed under The Code Project Open License (CPOL)

Steve44 · Accepted Answer · 2013-06-05T12:28:00

I think I see that the blocking is caused by some bugs within your queue code:

C++

//queue code
template<typename T>
class coda_concorr
{
private:
    ...
    // incorrect code:
    bool vuota() const
    {
        WaitForSingleObject(mutex,INFINITE);
        return la_coda.empty();
        ReleaseMutex(mutex);  // The mutex is never released
                              // as the function return in the previous line and
                              // ReleaseMutex is never called.
    }
    // fixed code:
    bool vuota() const
    {
        bool RetCode;
        WaitForSingleObject(mutex,INFINITE);
        RetCode = la_coda.empty();
        ReleaseMutex(mutex);
        return RetCode;
    }

 
    // incorrect code:
    bool try_pop(T& popped)
    {
        WaitForSingleObject(mutex,INFINITE);
        if (la_coda.empty()){
            return false; // same here, the mutex is not released
        }
        WaitForSingleObject(mutex,INFINITE); // This takes the mutex a second time
                                             // It would require a second ReleaseMutex()
                                             // for proper operation
        popped = la_coda.front();
        la_coda.pop();
        ReleaseMutex(mutex);
        return true;
    }
    // fixed code:
    bool try_pop(T& popped)
    {
        WaitForSingleObject(mutex,INFINITE);
        if (la_coda.empty()){
            ReleaseMutex(mutex);
            return false;
        }
        popped = la_coda.front();
        la_coda.pop();
        ReleaseMutex(mutex);
        return true;
    }
};

<V2> Another bug:
Here the incorrect code with explanation, see underlined lines:

C++

int main()
{
	coda_concorr<cv::Mat> ingresso;
	coda_concorr<cv::Mat> coda1;
	coda_concorr<cv::Mat> coda2;
	coda_concorr<cv::Mat> uscita;
 
	
//assing second queue	
dati.out=coda1; // This creates a copy of coda1	
...
	//share part that don't WORK
	dati2.in=coda1; // This creates a second copy of coda1,
                         // it does not reference the same queue
	dati2.out=coda2;
...
 
}

So both threads are seeing different copies of coda1 while you would like to have them see the same instance, coda1.
Here a proposal to fix the issue, make the out queue a pointer so you can link Args structures:

C++

struct Args
{
	coda_concorr<cv::Mat> in;
	coda_concorr<cv::Mat> *out; // this is a pointer now, make it point to the
                                     // next queue in sequence
};


void grey (void *param){
...
            arg->out->push(temp2);
        }
        arg->out->complete=true;
...
}
//threshold funct
void soglia(void *param){
...
            arg->out->push(temp2a);
        }
    }
    arg->out->complete=true;
    _endthread();
}


int main()
{
...
	coda_concorr<cv::Mat> coda1; // Not needed
...
//assing second queue	
dati.out=&dati2.in; // This stores a pointer to the next queue	
...

	dati2.in=&coda1; // Not needed as the queues are linked, see above
	dati2.out=&coda2;
...
	//output
	while (dati2.out->try_pop(out)){
		imshow("immagine",out);
		waitKey(100);
	}
...
}

Also there is a startup issue: When the second thread starts and its input queue is empty, it immediately exits. The same happens when the input queue becomes empty during execution (i.e. the second thread executes faster than the first one).
You have to change the design of your queuing in way that you can distinguish a "queue empty due to all work done" from a "queue empty due to previous step not yet finished".
One option would be to have a queue state added that is set when the producer is done providing data, so your thread only terminates when the queue is empty and the producer has indicated that it will not provide further data, otherwise the thread waits until the queue becomes not-empty.

<V3> Add the waiting in the queue try_pop method, this should get rid of the immediate termination at startup issue:

C++

bool try_pop(T& popped)
{
    WaitForSingleObject(mutex,INFINITE);
    while (la_coda.empty())
    {
        if (complete)
        {
            ReleaseMutex(mutex);
            return false;
        }
        else
        {
            ReleaseMutex(mutex);
            Sleep(100); // FUTURE IMPROVEMENT: This can be improved by creating
                        // an event signaling that the queue is not empty.
                        // I'll keep it like this for simplicity for now.
                        // I try to avoid mixing functionality re-work with
                        // performance improvements.
            WaitForSingleObject(mutex,INFINITE);
        }
    }

    popped = la_coda.front();
    la_coda.pop();
    ReleaseMutex(mutex);
    return true;
}

Domus1919 · Accepted Answer · 2013-06-06T22:26:00

After more try, this is the solution:

/*programma di prova gestione thread e OpenCV con timer di evalutazione tempo 

impiegato v.1.0 */

#include <opencv\cv.h>
#include <opencv\highgui.h>
#include <stdio.h>
#include <windows.h>
#include <process.h>
#include <queue>
#include <time.h>

using namespace std;
using namespace cv;


/*inizio della parte relativa alla coda thread safe*/

template<typename T>
class coda_concorr
{
private:
	std::queue<T> la_coda;
	HANDLE mutex;
public:
	bool complete;
	coda_concorr()
	{
		mutex = CreateMutex(NULL,FALSE,NULL);
		complete = false;
	}
	void push(T& data)
	{
		WaitForSingleObject(mutex,INFINITE);
		la_coda.push(data);
		ReleaseMutex(mutex);
	}
	bool vuota() const
	{
		bool RetCode;
		WaitForSingleObject(mutex,INFINITE);
		RetCode= la_coda.empty();
		ReleaseMutex(mutex);
		return RetCode;
	}
	bool try_pop(T& popped)
    {
        WaitForSingleObject(mutex,INFINITE);
        while (la_coda.empty()){
            //ReleaseMutex(mutex);
            //Sleep(100);
			//WaitForSingleObject(mutex,INFINITE);
			return false;
        }
        popped = la_coda.front();
        la_coda.pop();
        ReleaseMutex(mutex);
        return true;
    }
};

//pacchetto per passaggio ai thread
struct Args
{
	coda_concorr<cv::Mat> in;
	coda_concorr<cv::Mat> *out; //puntatore a coda successiva
};

//funzione grey
void grey (void *param){
	Mat temp1,temp2;
	int add = 0;
	Args* arg = (Args*)param;
	while(arg->in.try_pop(temp1)){
	cvtColor(temp1,temp2,CV_BGR2GRAY);
	arg->out->push(temp2);
	add++;
	   }
	//flag completato
	arg->out->complete=true;
	cout<<"Thread 1 terminato con "<<add<<" elaborazioni."<<endl;
	_endthread();
	
}

//funzione sogliatura
void soglia(void *param){
	Mat temp1a,temp2a;
	int add=0;
	Args* arg = (Args*)param;
	while(arg->in.vuota()){
	Sleep(50);
	}
	while(arg->in.try_pop(temp1a)){
		threshold(temp1a,temp2a,128,255,THRESH_BINARY);
		arg->out->push(temp2a);
		add++;
		}
		//flag completato
	 arg->out->complete=true;
	 cout<<"Thread 2 terminato con "<<add<<" elaborazioni."<<endl;
	_endthread();
}


//funzione erosione/dilate
void finitura(void *param){
	Mat temp1b,temp2b,temp2c;
	int add = 0;
	Args* arg = (Args*)param;
	while(arg->in.vuota()){
	Sleep(40);
	}
	while(arg->in.try_pop(temp1b)){
		erode(temp1b,temp2b,cv::Mat());
		dilate(temp2b,temp2c,Mat());
		arg->out->push(temp2c);
		add++;
		}
		//flag completato
	 arg->out->complete=true;
	 cout<<"Thread 3 terminato con "<<add<<" elaborazioni."<<endl;
	_endthread();
}



//main
int main()
{
	//dichiarazione delle due code principali
	coda_concorr<cv::Mat> ingresso;
	coda_concorr<cv::Mat> uscita;

	//clock
	clock_t avvio;
	clock_t fine;

	
	//array locali di ingresso e uscita
	Mat inn[10];
	Mat out;
	
	//inizio a contare tempo
	avvio = clock();

	//assegnazione 10 immagini di prova
	inn[0]=imread("C:/OPENCV/Test/imgtest/bird1.jpg",1);
	inn[1]=imread("C:/OPENCV/Test/imgtest/bird2.jpg",1);
	inn[2]=imread("C:/OPENCV/Test/imgtest/bird3.jpg",1);
	inn[3]=imread("C:/OPENCV/Test/imgtest/pig1.jpg",1);
	inn[4]=imread("C:/OPENCV/Test/imgtest/pig2.jpg",1);
	inn[5]=imread("C:/OPENCV/Test/imgtest/pig3.jpg",1);
	inn[6]=imread("C:/OPENCV/Test/imgtest/spider1.jpg",1);
	inn[7]=imread("C:/OPENCV/Test/imgtest/spider2.jpg",1);
	inn[8]=imread("C:/OPENCV/Test/imgtest/spider3.jpg",1);
	inn[9]=imread("C:/OPENCV/Test/imgtest/Nutella.jpg",1);

	
	//dichiarazione code
	Args dati,dati2,dati3;

	

	//Popolamento coda ingresso
	for(int i=0;i<=9;i++){
		dati.in.push(inn[i]);
	}
	//punta all'output successivo
	dati.out=&dati2.in;
	dati2.out=&dati3.in;
	dati3.out=&uscita;
	//dichiarazione delle maniglie
	HANDLE handle1,handle2,handle3;

	//dichiarazione dei threads
	handle1 = (HANDLE) _beginthread(grey,0,&dati);
	handle2 = (HANDLE) _beginthread(soglia,0,&dati2);
	handle3 = (HANDLE) _beginthread(finitura,0,&dati3);

	//join
	WaitForSingleObject(handle1,INFINITE);
	WaitForSingleObject(handle2,INFINITE);
	WaitForSingleObject(handle3,INFINITE);
	
	
	//fine registrazione tempo
	fine = clock()-avvio;

	double tempoimp = fine / (double)CLOCKS_PER_SEC;

//output di verifica
	/*
	while(dati3.out->try_pop(out)){
		imshow("immagine",out);
		waitKey(100);
	}*/
	
	cout<<endl<<"Cicli di clock: "<<fine<<endl;
	cout<<"Secondi trascorsi: "<<tempoimp<<endl;
	system("PAUSE");
	
	
	return 0;

}

This manage 3 threads that works on shared elements, in 4 queues.

Multiple queue of images, multiple threads...how?

2 solutions

Solution 1

Solution 2

Add your solution here

Preview 0