Thread: How to implement a POSIX threaded program to solve producer/consumer problem

Originally Posted by C_ntua

You have a class that icontains a vector, called m_vctQueue. This is where the elements are stored. The type of the elements of the vector is Message, which as you are instructed can be changed through message.h. You use the functions pop, push to insert or extract elements from the queue (vector). When you pop you decrease m_size (initialized at 0). When you push you increase it. The point is for you not to pop() when there are no elements int the queue. So you there is the condition if m_size == 0 to wait for a signal. When m_size == 1 again, which means a push() was performed, then you can use pop() again. The pthread_mutex_lock/unlock is to make sure no other thread will use pop(), push() at the same time a pop() or push() is performed (for obvious reasons).
You also have functions getsize() to get the size of the queue.

So you create one object SynchronizedQueue and all the threads will pop()/push() elements there. The class will make sure it is properly syncrhonized. Meaning you won't be able to extract elements with pop() if the queue is empty. That is all. If you want more explanation we ll be welcome to help with possibly an example

Yesterday, 05:29 PM #4
C_ntua Thank you. If you have got time. Could you please provide me an example program. It would be much helpful for me to understand the SynchronizedQueue class. Thank you in advance

My application can be simplified as the following problem.

In the main function, the program constantly receives data (a struct data type, containing two integer type elements), whenever a struct of data is received, the struct are pushed into a queue; On the other hand, once there are idle thread (totally, n threads are created when the program is started), one data unit are popped out, and are processed by one certain thread. ArkM, could you please provide a COMPLETE program to implement the above task using SynchronizedQueue class. Thanks, I really need this.

the data struct is defined as
struct DataIn{
int ItemA;
int ItemB;
}

QUestions:
(1) How to create n threads.
(2) Peled told me to use something like ProcessData procD(SynQ); I see this means that I should use a SynchronizedQueue type object as an argument to instantiate a ProcessData type object , but what should be done with SynQ within the ProcessData class.


I tried my best to make the following code based on my understanding of the SynchronizedQueue class and poor knowledge of C++/OO design. The code is erroreous and incomplete even looks stupid. my apologies for this

Code:

#include "SynchronizedQueue.h"
struct TWO_INT
{
	int iA;
	int iB;
};

vector<int> SumofTwoNumbers;
class ProcessData{
private:
	SynchronizedQueue* jobs
		bool destroy;
public:
	ProcessData(SynchronizedQueue* jobs)
	~ProcessData()
	void destroy()
	void* run(void* args)
}

ProcessData::run()
{
	while (!destroy){
		struct TWO_INT* data  = jobs->pop();
		//process data
		Sum0fTwoNumbers.push(data->iA + data->iB);
	}
}

SynchronizedQueue SynQ;
SynQ.destroy = 0;
main()
{
	int i;
                ProcessData ProcD;
	for(i=0;  i<1000; ++i)
	{
		struct TWO_INT* twoint = new TWO_INT;
		twoint->iA = i;
		twoint->iB = i + 1;
		SynQ.push(twoint);
                                
                                 ProcD.run();
	}
                //ProcD.run();// Or put the line here
	SynQ.destroy = 1;

                ProcD.destroy();//But what the code in the destroy function body should look like?
}#include "SynchronizedQueue.h"
struct TWO_INT
{
	int iA;
	int iB;
};

vector<int> SumofTwoNumbers;
class ProcessData{
private:
	SynchronizedQueue* jobs
		bool destroy;
public:
	ProcessData(SynchronizedQueue* jobs)
	~ProcessData()
	void destroy()
	void* run(void* args)
}

ProcessData::run()
{
	while (!destroy){
		struct TWO_INT* data  = jobs->pop();
		//process data
		Sum0fTwoNumbers.push(data->iA + data->iB);
	}
}

SynchronizedQueue SynQ;
SynQ.destroy = 0;
main()
{
	int i;
                ProcessData ProcD;
	for(i=0;  i<1000; ++i)
	{
		struct TWO_INT* twoint = new TWO_INT;
		twoint->iA = i;
		twoint->iB = i + 1;
		SynQ.push(twoint);
                                
                                 ProcD.run();
	}
                //ProcD.run();// Or put the line here
	SynQ.destroy = 1;

                ProcD.destroy();//But what the code in the destroy function body should look like?
}