线程.c

该程序运用线程的mutex控制数据流的锁定

/* Inter-Thread Communication Using POSIX Standard Threads, Global Variables and Mutexes  
(EMARO Real Time Systems Project)

By: Boris Takac
Date: December 2008
Place: Genova, Liguria, Italy
E-mail: [e-mail]
*/

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <limits.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <math.h>
#include <signal.h>
#include <pthread.h>


/*   Definitions*/

#define BUFFER_SIZE 100			//data buffer size
#define OBJECT_SIZE 30			//maximal new object size


/*   Global variables   */

pthread_mutex_t buff_lock = PTHREAD_MUTEX_INITIALIZER;          //static mutex initialization
int buff_el;							//global counter for number of elements in data buffer
int seek_pos[BUFFER_SIZE]={0};					//global array with writing start position for every object in buffer
int seek_len[BUFFER_SIZE]={0};					//global array with writing length for every object in buffer
char buffer[BUFFER_SIZE]={0};					//main data buffer
unsigned int pseed;						//random seed for producer thread 
unsigned int cseed;						//random seed for consumer thread
int producer_stop=1;
int consumer_stop=1;


/*  Prototypes   */
void * stopper (void *arg);
void * producer (void *arg);
void * consumer (void *arg);
int make_rand_object(char *object);


/* Main function   */

int main (void) {
pthread_t t1, t2, t3;

pthread_create (&t1, NULL, stopper, NULL);
pthread_create (&t2, NULL, producer, NULL);
pthread_create (&t3, NULL, consumer, NULL);

pthread_join (t1, NULL);
pthread_join (t2, NULL);
pthread_join (t3, NULL);

printf("Successfull exit!\n");	

return 0;
}



/*  Thread for stopping producer and consumer threads    */

void * stopper (void *arg) {
printf("Press q to exit program:\n"); 				
int ch;			
while (1){
	ch=getchar();	
	printf("You typed %d\n",ch);
	if (ch==113) {
		consumer_stop=0;
		producer_stop=0;
		break;	
	}		
       }
return NULL;
}



/*  Producer  thread    */

void * producer (void *arg) {

	int buff_set;				//write position of the last element
	int obj_len;				//lenght of newly created object
	int prod_count=0;			//counter for number of writing failures
	int prod_count_old=0;		
	char object[OBJECT_SIZE];		//array for generating nwe objects
	char write_print[OBJECT_SIZE+1];	//help memory location for printing out new objects as 0 concluded strings 	

	time_t seconds;				//system time in seconds as the seed for rand() function
	double r=0;
	int M=4;				//max value for waiting in seconds for random function 
	double x;
	int rand_secs;
	
	time(&seconds);				//system time in seconds as the seed for rand_r() function	
	pseed= (unsigned int) (getpid() * seconds * 2); 	//generate random number seed for this thread

	printf ("Producer ONLINE!\n");		
	
	while(producer_stop) {
	/* begin critical section */
		pthread_mutex_lock (&buff_lock);		//use mutex to lock the resources for this thread
		
		/* produce random new object  */
		if (prod_count==prod_count_old){				//produce new object only if the last iteration writing was success
			obj_len=make_rand_object(object); 			//call function  for generating new object of random length
			memcpy(write_print, object, obj_len);			//prepare new object for printout
			write_print[obj_len]=0;			
			printf("New object of lenght %d generated: %s\n",obj_len, write_print);					
		}
		
		buff_set=seek_pos[buff_el]+seek_len[buff_el];			//use last object position and length info to calculate starting position for new object
					
		if (buff_set<=(BUFFER_SIZE-obj_len)) {					//if there is enough unused space in buffer --> write object
			memcpy(&buffer[buff_set], object, obj_len);			//PUT object to writing postition

			buff_el++;							//increment counter for number of elements in data buffer	
			seek_pos[buff_el]=buff_set;					//write info for newest object writing start position
			seek_len[buff_el]=obj_len;					//write info for newest object lenght
	
			printf("Data Buffer Elements: %d; WRITE Last Element--> Position: %d, Length: %d\n", buff_el, buff_set, obj_len);
			prod_count_old=prod_count;
		} else { 							// if there is not enough space in data buffer 
			prod_count++;						//increment producer counter for writing faulures
			printf("Write buffer full! Producer miss count: %d\n", prod_count);
		}
		pthread_mutex_unlock (&buff_lock);				//unlock resources
	/* end critical section */
	/* random timer section	*/
		r = ((double)rand_r(&pseed) / ((double)(RAND_MAX)+(double)(1)));
		x=(r*M);
		rand_secs=(int) x;
	
		printf("Producer waits %d secs\n", rand_secs);		
		sleep(rand_secs);	
	}

return NULL;
}



/*  Consumer  thread    */

void * consumer (void *arg) {

	int buff_set;				//starting position for reading  the last object in the buffer
	int obj_len;				//lenght of last object that is being read from data buffer
	int consum_count=0;			//counter for number of reading failures
	char read_object[OBJECT_SIZE];		//array for reading objects from buffer
	char read_print[OBJECT_SIZE+1];	 	//help memory location for printing out newly  read objects as 0 concluded strings 	
 
	time_t seconds;				//system time in seconds as the seed for rand() function
	double r=0;
	int M=4;				//max value for waiting in seconds for random function 
	double x;
	int rand_secs;

	sleep(1);
		
	time(&seconds);						//system time in seconds as the seed for rand_r() function	
	cseed= (unsigned int) (getpid() * seconds * 3);   	//generate random number seed for this thread

	printf ("Consumer ONLINE!\n");	

	while(consumer_stop) {	
	/* begin critical section */
		pthread_mutex_lock (&buff_lock);		//use mutex to lock the resources for this thread

		buff_set=seek_pos[buff_el];
		obj_len=seek_len[buff_el];

		if (buff_el > 0) {									//if there is something in buffer do reading 
       	      	        printf("Data Buffer Elements: %d; READ Last Element--> Position: %d, Length: %d\n", buff_el, buff_set, obj_len);
			memcpy(read_object,&buffer[buff_set], obj_len);					//GET object from shared buffer
			memcpy(read_print, read_object, obj_len);					//copy object for printout
			read_print[obj_len]=0;						
			
			printf("Consumer reads: %s\n",read_print);		
			
			buff_el--;									//decrement counter of number of elements in data buffer	
				
		} else { 										//if shared memory buffer is empty
			consum_count++;									//increment  counter for counting failed readings
			printf("Read buffer empty! Consumer read failure count: %d\n", consum_count);
		}
		pthread_mutex_unlock (&buff_lock);						//unlock resources
	/* end critical section */
	/* random timer section*/
		r = ((double)rand_r(&cseed) / ((double)(RAND_MAX)+(double)(1)));
		x=(r*M);
		rand_secs=(int) x;
	
		printf("Consumer waits %d secs\n", rand_secs);		
		sleep(rand_secs);	
	}
	
return NULL;
}



/* Make object as random sized string filled with random numbers between 0-9 */

int make_rand_object(char *object){
	
int i=0;	
double r=0;
int max_size=OBJECT_SIZE+1;			//max value for object lenght 
double x;
int obj_len;					//random value of object lenght
int max_reach=10;				//max value for generating numbers from 0 to 9	
int rand_value;
char ch;					

/* calculate random object length  */
r = ((double)rand_r(&pseed) / ((double)(RAND_MAX)+(double)(1)));
x=(r*max_size);			
obj_len=(int) x;	//obj_len value goes between 0 and (max_size-1)

/* fill string with random numbers  */	
for (i=0;i<obj_len;i++){
	r = ((double)rand_r(&pseed) / ((double)(RAND_MAX)+(double)(1)));
	x=(r*max_reach);		
	rand_value=(int) x;		//rand_value goes between 0 and (max_reach-1)
	ch=(char)(48+rand_value);		
	object[i]=ch;
}

return obj_len;
}