我需要解决生产者-消费者问题，需要在进程中使用线程的单进程解决方案。单个进程的主程序应该在全局内存中设置和初始化所需的环形缓冲区和控制变量，以便进程的所有其他线程可以看到它们。然后主例程将开始创建一些生产者线程和一些消费者线程。
我在互联网上做了很多研究，并且能够编写其中的一些代码。但是，它似乎无法正常工作，并且所有值都陷入僵局。有人可以帮我吗？
我的头文件：
#include <sys/signal.h>
#include <sys/time.h>
#include <pthread.h>
#include <sys/signal.h>
#include <string.h>
#include <stdio.h>
#define NUMFLAVORS 4
#define NUMSLOTS 400
#define NUMCONSUMERS 50
#define NUMPRODUCERS 30
#define NoOfDozens 200
struct DONUT_SHOP {
int flavor [NUMFLAVORS] [NUMSLOTS];
int outptr [NUMFLAVORS];
int in_ptr [NUMFLAVORS];
int serial [NUMFLAVORS];
};
我的程序：
void *sig_waiter ( void *arg );
void *producer ( void *arg );
void *consumer ( void *arg) ;
void sig_handler ( int );
/****************************/
/* GLOBAL VARIABLES */
/****************************/
#include "ml_pc.h"
struct DONUT_SHOP shared_ring;
int space_count [NUMFLAVORS];
int donut_count [NUMFLAVORS];
int serial [NUMFLAVORS];
pthread_mutex_t prod [NUMFLAVORS];
pthread_mutex_t cons [NUMFLAVORS];
pthread_cond_t prod_cond [NUMFLAVORS];
pthread_cond_t cons_cond [NUMFLAVORS];
pthread_t thread_id [NUMCONSUMERS+1], sig_wait_id;
int main ( int argc, char *argv[] )
{
int i, j, k, nsigs;
struct timeval randtime, first_time, last_time;
struct sigaction new_act;
int arg_array[NUMCONSUMERS];
sigset_t all_signals;
int sigs[] = { SIGBUS, SIGSEGV, SIGFPE };
pthread_attr_t thread_attr;
struct sched_param sched_struct;
/**********************************************************************/
/* INITIAL TIMESTAMP VALUE FOR PERFORMANCE MEASURE */
/**********************************************************************/
gettimeofday (&first_time, (struct timezone *) 0 );
for ( i = 0; i < NUMCONSUMERS + 1 ; i++ ) {
arg_array [i] = i+1; /** SET ARRAY OF ARGUMENT VALUES **/
}
/**********************************************************************/
/* GENERAL PTHREAD MUTEX AND CONDITION INIT AND GLOBAL INIT */
/**********************************************************************/
for ( i = 0; i < NUMFLAVORS; i++ ) {
pthread_mutex_init ( &prod [i], NULL );
pthread_mutex_init ( &cons [i], NULL );
pthread_cond_init ( &prod_cond [i], NULL );
pthread_cond_init ( &cons_cond [i], NULL );
shared_ring.outptr [i] = 0;
shared_ring.in_ptr [i] = 0;
shared_ring.serial [i] = 0;
space_count [i] = NUMSLOTS;
donut_count [i] = 0;
}
/**********************************************************************/
/* SETUP FOR MANAGING THE SIGTERM SIGNAL, BLOCK ALL SIGNALS */
/**********************************************************************/
sigfillset (&all_signals );
nsigs = sizeof ( sigs ) / sizeof ( int );
for ( i = 0; i < nsigs; i++ ){
sigdelset ( &all_signals, sigs [i] );
}
sigprocmask ( SIG_BLOCK, &all_signals, NULL );
sigfillset (&all_signals );
for( i = 0; i < nsigs; i++ ) {
new_act.sa_handler = sig_handler;
new_act.sa_mask = all_signals;
new_act.sa_flags = 0;
if ( sigaction ( sigs[i], &new_act, NULL ) == -1 ){
perror("can't set signals: ");
exit(1);
}
}
printf ( "just before threads created\n" );
/*********************************************************************/
/* CREATE SIGNAL HANDLER THREAD, PRODUCER AND CONSUMERS */
/*********************************************************************/
if ( pthread_create (&sig_wait_id, NULL,
sig_waiter, NULL) != 0 ){
printf ( "pthread_create failed " );
exit ( 3 );
}
pthread_attr_init ( &thread_attr );
pthread_attr_setinheritsched ( &thread_attr,
PTHREAD_INHERIT_SCHED );
#ifdef GLOBAL
sched_struct.sched_priority = sched_get_priority_max(SCHED_OTHER);
pthread_attr_setinheritsched ( &thread_attr,
PTHREAD_EXPLICIT_SCHED );
pthread_attr_setschedpolicy ( &thread_attr, SCHED_OTHER );
pthread_attr_setschedparam ( &thread_attr, &sched_struct );
pthread_attr_setscope ( &thread_attr,
PTHREAD_SCOPE_SYSTEM );
#endif
if ( pthread_create (&thread_id[0], &thread_attr,
producer, NULL ) != 0 ) {
printf ( "pthread_create failed " );
exit ( 3 );
}
for ( i = NUMPRODUCERS; i < NUMCONSUMERS + 1; i++ ) {
if ( pthread_create ( &thread_id [i], &thread_attr,
consumer, ( void * )&arg_array [i]) != 0 ){
printf ( "pthread_create failed" );
exit ( 3 );
}
}
printf ( "just after threads created\n" );
/*********************************************************************/
/* WAIT FOR ALL CONSUMERS TO FINISH, SIGNAL WAITER WILL */
/* NOT FINISH UNLESS A SIGTERM ARRIVES AND WILL THEN EXIT */
/* THE ENTIRE PROCESS....OTHERWISE MAIN THREAD WILL EXIT */
/* THE PROCESS WHEN ALL CONSUMERS ARE FINISHED */
/*********************************************************************/
for ( i = 1; i < NUMCONSUMERS + 1; i++ ) {
pthread_join ( thread_id [i], NULL );
}
/*****************************************************************/
/* GET FINAL TIMESTAMP, CALCULATE ELAPSED SEC AND USEC */
/*****************************************************************/
gettimeofday (&last_time, ( struct timezone * ) 0 );
if ( ( i = last_time.tv_sec - first_time.tv_sec) == 0 )
j = last_time.tv_usec - first_time.tv_usec;
else{
if ( last_time.tv_usec - first_time.tv_usec < 0 ) {
i--;
j = 1000000 +
( last_time.tv_usec - first_time.tv_usec );
} else {
j = last_time.tv_usec - first_time.tv_usec; }
}
printf ( "Elapsed consumer time is %d sec and %d usec\n", i, j );
printf ( "\n\n ALL CONSUMERS FINISHED, KILLING PROCESS\n\n" );
exit ( 0 );
}
/*********************************************/
/* INITIAL PART OF PRODUCER..... */
/*********************************************/
void *producer ( void *arg )
{
int i, j, k;
int rand_donuts;
unsigned short xsub1 [3];
struct timeval randtime;
int threadid = pthread_self ( );
FILE *prod_file ;
char prod_id[10];
char prod_filename[20]="prod_";
gettimeofday ( &randtime, ( struct timezone * ) 0 );
xsub1 [0] = ( unsigned short )randtime.tv_usec;
xsub1 [1] = ( unsigned short ) ( randtime.tv_usec >> 16 );
xsub1 [2] = ( unsigned short ) ( pthread_self );
while ( 1 )
{
rand_donuts = nrand48 ( xsub1 ) & 3;
pthread_mutex_lock ( &prod [rand_donuts] );
/** check if there is space available to produce donuts **/
while ( space_count [rand_donuts] == 0 )
{
pthread_cond_wait ( &prod_cond [rand_donuts], &prod [rand_donuts] );
}
serial[rand_donuts] ++;
fprintf(prod_file,"thread %d is producing donuts of type :%d , serial number of donut: %d \n",threadid,rand_donuts,serial[rand_donuts]);
shared_ring.flavor[rand_donuts][shared_ring.in_ptr[rand_donuts]]=serial[rand_donuts];
/** Increment in_ptr modulo NUMSLOTS **/
shared_ring.in_ptr[rand_donuts] = (shared_ring.in_ptr[rand_donuts]+1) % NUMSLOTS ;
/** Decrement the space count **/
space_count [rand_donuts] --;
pthread_mutex_unlock ( &prod [rand_donuts] );
/* get cons mutex */
/* inc donut count */
/* unlock cons mutex */
/* signal cons_condx_var */
pthread_mutex_lock(&cons [rand_donuts]);
donut_count [rand_donuts]++;
pthread_mutex_unlock(&cons [rand_donuts]);
pthread_cond_signal(&cons_cond[rand_donuts]);
usleep(10000);
}
fclose(prod_file);
return NULL;
}
/*********************************************/
/* Consumer Code ,starts here */
/*********************************************/
void *consumer ( void *arg )
{
int i, j, k, m, id ,rand_donuts,outptr,numofdonut;
unsigned short xsub [3];
struct timeval randtime;
int printresult;
int serial [NUMFLAVORS];
int collection [NUMFLAVORS][12];
FILE *cons_file;
char fileid[10];
char cons_filename[10]="cons_";
time_t current_time;
id = *( int * ) arg;
sprintf(fileid, "%d", id) ;
strcat(cons_filename,fileid);
gettimeofday ( &randtime, ( struct timezone * ) 0 );
xsub [0] = ( unsigned short )randtime.tv_usec;
xsub [1] = ( unsigned short ) ( randtime.tv_usec >> 16 );
xsub [2] = ( unsigned short ) ( pthread_self );
if ((cons_file = fopen(cons_filename, "a+")) == NULL)
fprintf(stderr, "Cannot open %s\n", "cons_file");
for( i = 0; i < 10; i++ )
{
for( k = 0; k < NUMFLAVORS; k++)
{
for( m = 0; m < 12; m++)
{
collection[k][m] = -1;
}
}
for(k = 0; k < NUMFLAVORS; k++)
{
serial [k] = 0;
}
for( m = 0; m < 12; m++ )
{
rand_donuts = nrand48( xsub ) & 3;
/*******************************************************
get cons mutex
check donut count
loop:
if donut count == 0
wait cons_condx_var
*******************************************************/
pthread_mutex_lock ( &cons [rand_donuts] );
while ( donut_count [rand_donuts] == 0 )
{
pthread_cond_wait ( &cons_cond [rand_donuts], &cons [rand_donuts] );
}
/************************************************
Consume donut
************************************************/
outptr = shared_ring.outptr[rand_donuts];
numofdonut = shared_ring.flavor[rand_donuts][outptr];
collection[rand_donuts][serial[rand_donuts]]=numofdonut ;
serial[rand_donuts] ++;
outptr ++;
shared_ring.outptr[rand_donuts] = outptr % NUMSLOTS;
donut_count [rand_donuts] --;
pthread_mutex_unlock ( &cons [rand_donuts] );
/* lock producer mutex */
pthread_mutex_lock(&prod [rand_donuts]);
/*increment the space count*/
space_count [rand_donuts] ++;
/* unlock producer mutex */
pthread_mutex_unlock(&prod [rand_donuts]);
//signal the producer
pthread_cond_signal(&prod_cond[rand_donuts]);
}
current_time=time(NULL);
fprintf(cons_file,"Consumer %d finished collecting dozen#: %d \n",id,i);
fprintf(cons_file,"Consumer thread#: %d , time: %s, Dozen#: %d",id,asctime(localtime(¤t_time)),i);
fprintf(cons_file,"\t plain \t jelly \t coconut \t honey-dip \t");
fprintf(cons_file,"\n");
printresult=0;
for (i = 1; i < NoOfDozens; i++)
{
for( m = 0; m < 12 && printresult <=3 ; m++)
{
printresult = 0;
for(k = 0; k < NUMFLAVORS ; k++)
{
if(collection[k][m] >= 0)
{
fprintf(cons_file,"\t %d \t",collection[k][m]);
printresult = 0;
}
else
{
fprintf(cons_file,"\t \t");
printresult =printresult+1;
}
}
fprintf(cons_file,"\n");
}
}
/*Sleeping*/
fflush(cons_file);
usleep(1000); /* sleep 1 ms */
}
fprintf(cons_file,"Consumer %d finished collecting all 10 dozen donuts \n",id);
fclose(cons_file);
return NULL;
}
/***********************************************************/
/* PTHREAD ASYNCH SIGNAL HANDLER ROUTINE... */
/***********************************************************/
void *sig_waiter ( void *arg )
{
sigset_t sigterm_signal;
int signo;
sigemptyset ( &sigterm_signal );
sigaddset ( &sigterm_signal, SIGTERM );
sigaddset ( &sigterm_signal, SIGINT );
if (sigwait ( &sigterm_signal, & signo) != 0 ) {
printf ( "\n sigwait ( ) failed, exiting \n");
exit(2);
}
printf ( "process going down on SIGNAL (number %d)\n\n", signo );
exit ( 1 );
return NULL;
}
/**********************************************************/
/* PTHREAD SYNCH SIGNAL HANDLER ROUTINE... */
/**********************************************************/
void sig_handler ( int sig )
{
pthread_t signaled_thread_id;
int i, thread_index;
signaled_thread_id = pthread_self ( );
for ( i = 0; i < (NUMCONSUMERS + NUMPRODUCERS ); i++) {
if ( signaled_thread_id == thread_id [i] ) {
thread_index = i;
break;
}
}
printf ( "\nThread %d took signal # %d, PROCESS HALT\n",
thread_index, sig );
exit ( 1 );
}
有人可以告诉我哪里出了问题以及我需要做些什么改变吗？提前致谢！！