consumer producer problem with threads

[nabi]

I am facing a synchronisation or scheduling problem with threads. I have simplified my problem to the following one.

I have a producer thread that increases an integer value by one and a consumer thread that decreases the integer value by one too. Now because the integer is shared i need to use pthread mutex as shown in program below. If no signaling is used, that is no pthread_cond_t is used, I found out that the times the process stays in one thread, consumer or producer is not equal at all.

So, I thought that each time i produce something the consumer is called by a signal. However this method which is shown below, still writes in console values of 1600 by the producer. Clearly not a fair by the threads.

Question: how can i make this program fair, so that the value is "often" in a range of "0"? I have also tried to extend to extend the waiting time up to 1000. No difference.

Thanks

#include <stdio.h>
#include <pthread.h>
#include <signal.h>

#define MAXBUFFER 10
typedef struct share share_t;
struct share
{
	pthread_mutex_t mutex;
	int buff [MAXBUFFER];
	int nput;
	int nvalue;
};
share_t shared;
int runcond;
pthread_cond_t token_cond;

void sigfun(int sig)
{
	runcond=0;
	(void) signal (SIGINT, SIG_DFL);
}

void initialize_shared(void)
{
	runcond=1;
	pthread_mutex_init(&shared.mutex,NULL);
	pthread_mutex_lock(&shared.mutex);
	int i;
	shared.nput=0;
	shared.nvalue=0;
	for (i=0;i<MAXBUFFER;i++)	
	{	
		shared.buff[i]=0;
	}
	pthread_mutex_unlock(&shared.mutex);

	return;
}

void * produce ()
{
	int i=0;

	while (runcond){
		pthread_mutex_lock(&shared.mutex);
		shared.nput++;
		if ((shared.nput % 10)==0)printf("Nput Producer %d\n",shared.nput);
		pthread_cond_signal(&token_cond);
		//pthread_cond_wait(&token_cond,&shared.mutex);
		pthread_mutex_unlock(&shared.mutex);
	}
	pthread_exit(NULL);
}

void * consume () 
{
	int i;
	while (runcond){
		pthread_mutex_lock(&shared.mutex);
		pthread_cond_wait(&token_cond,&shared.mutex);
		shared.nput--;
		if ((shared.nput % 10)==0)printf("Nput Consumer %d\n",shared.nput);
		//pthread_cond_signal(&token_cond);
		pthread_mutex_unlock(&shared.mutex);
	}

	pthread_exit(NULL);
}


int main (void)
{
	(void) signal (SIGINT, sigfun);
	initialize_shared();
	pthread_t producera;
	pthread_t consumer;
	int i;
		pthread_mutex_lock(&shared.mutex);
		pthread_create(&consumer,NULL,consume,NULL);
		pthread_create(&producera,NULL,produce,NULL);
		pthread_mutex_unlock(&shared.mutex);
		wait(10);
		pthread_mutex_lock(&shared.mutex);
			runcond=0;
		pthread_mutex_unlock(&shared.mutex);
		pthread_join(consumer,NULL);
		pthread_join(producera,NULL);

	
	return 0;
}

[Codeplug]

Using signal() in a multi-thread application is "unspecified" by Posix.

If you are going to assign a global inside a signal handler, it must be of type "volatile sig_atomic_t".

There is unsynchronized access to "runcond". main() writes to it with the lock acquired, but the threads read from it without the lock acquired.

No need to lock inside initialize_shared(), as it is only called on one thread.

Conditions variables should always be used in a "predicate loop".

If you want your threads to ping-pong back and forth - you'll need two condition variables (and predicates). Example code in this thread: Mutex, Cond and Thread questions (pthread, linux)

gg

[nabi]

Excellent example gg.

I have changed the program by modifying "runcond" variable and adding a second condition variable. Now it works as expected, each time one produces, one is consumed.

I ve got a question here, how can i control the while without using runcond? the problem is that if I simply forget about the treads in my original program (as it is in the example, where only is waited the producer by pthread_join ), there are many variables left without been freed. I am using the signal handler for ctrl+c to keep it simple but i would appreciate if anybody has another hint here.

Below the program that works:

#include <stdio.h>
#include <pthread.h>
#include <signal.h>

#define MAXBUFFER 10
typedef struct share share_t;
struct share
{
	pthread_mutex_t mutex;
	int buff [MAXBUFFER];
	int nput;
	int nvalue;
};
share_t shared;
volatile sig_atomic_t runcond;
//int runcond;
pthread_cond_t ConsumedData=PTHREAD_COND_INITIALIZER;;
int bconsumed=0;
pthread_cond_t ProducedData=PTHREAD_COND_INITIALIZER;;
int bproduced=0;

void sigfun(int sig)
{
	runcond=0;
	(void) signal (SIGINT, SIG_DFL);
}

void initialize_shared(void)
{
	runcond=1;
	pthread_mutex_init(&shared.mutex,NULL);
	int i;
	shared.nput=0;
	shared.nvalue=0;
	for (i=0;i<MAXBUFFER;i++)	
	{	
		shared.buff[i]=0;
	}
	return;
}

void * produce ()
{
	int i=0;
	while (runcond){
		pthread_mutex_lock(&shared.mutex);
		shared.nput++;
		/*if ((shared.nput % 10)==0)*/printf("Nput Producer %d\n",shared.nput);
		bconsumed=1;
		pthread_cond_signal(&ConsumedData);

		bproduced=0;		
		while (!bproduced)
			pthread_cond_wait(&ProducedData,&shared.mutex);
	
		pthread_mutex_unlock(&shared.mutex);
	}
	pthread_exit(NULL);
}

void * consume () 
{
	int i;
	while (runcond){
		pthread_mutex_lock(&shared.mutex);
		while (!bconsumed)
			pthread_cond_wait(&ConsumedData,&shared.mutex);
		shared.nput--;
		/*if ((shared.nput % 10)==0)*/printf("Nput Consumer %d\n",shared.nput);
		bproduced=1;
		pthread_cond_signal(&ProducedData);
		
		bconsumed=0;
		pthread_mutex_unlock(&shared.mutex);
	}

	pthread_exit(NULL);
}


int main (void)
{
	(void) signal (SIGINT, sigfun);
	initialize_shared();
	pthread_t producera;
	pthread_t consumer;
	int i;
		pthread_create(&consumer,NULL,consume,NULL);
		pthread_create(&producera,NULL,produce,NULL);
		pthread_join(consumer,NULL);
		pthread_join(producera,NULL);

	
	return 0;
}

Output is a constant
Nput Consumer 0
Nput Producer 1

until you press 'Ctrl+c'

Thanks

[Codeplug]

>> how can i control the while without using runcond?
Thread cancellation is one method: http://www.opengroup.org/onlinepubs/...l#tag_15_09_05
You'll have to use pthread_cleanup_push() to ensure locks and resources are released: http://www.opengroup.org/onlinepubs/...anup_push.html

The safest way to handle signals in a multi-threaded app is to create a dedicated thread that processes signals directly via sigwait: http://www.opengroup.org/onlinepubs/...d_sigmask.html

gg