title: Producer - Consumer using ring buffers and pthreads
body: <p>I need to solve the producer-consumer problem requiring a single process solution using threads within a process. The main routine of the single process should setup and initialize the required ring buffers and control variables in global memory where they can be seen by all other threads of the process. The main routine will then begin creating some number of producer threads and some number of consumer threads. </p>
  
  <p>I did a lot of research on the internet and was able to code some of it. However, it doesn't seem to be working right and I'm getting deadlocks for all the values. Can someone please help me out?</p>
  
  <p>My header file:</p>
  
  <pre><code>    #include &lt;sys/signal.h&gt;
  #include &lt;sys/time.h&gt;
  #include &lt;pthread.h&gt;
  #include &lt;sys/signal.h&gt;
  #include &lt;string.h&gt;
  #include &lt;stdio.h&gt;
  
  #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];
  };
  </code></pre>
  
  <p>My program:</p>
  
  <pre><code>    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 (&amp;first_time, (struct timezone *) 0 );
  
  
          for ( i = 0; i &lt; 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 &lt; NUMFLAVORS; i++ ) {
          pthread_mutex_init ( &amp;prod [i], NULL );
          pthread_mutex_init ( &amp;cons [i], NULL );
          pthread_cond_init ( &amp;prod_cond [i],  NULL );
          pthread_cond_init ( &amp;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 (&amp;all_signals );
      nsigs = sizeof ( sigs ) / sizeof ( int );
      for ( i = 0; i &lt; nsigs; i++ ){
              sigdelset ( &amp;all_signals, sigs [i] );
          }
      sigprocmask ( SIG_BLOCK, &amp;all_signals, NULL );
      sigfillset (&amp;all_signals );
      for( i = 0; i &lt;  nsigs; i++ ) {
              new_act.sa_handler  = sig_handler;
              new_act.sa_mask     = all_signals;
              new_act.sa_flags    = 0;
              if ( sigaction ( sigs[i], &amp;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 (&amp;sig_wait_id, NULL,
                      sig_waiter, NULL) != 0 ){
  
                  printf ( "pthread_create failed " );
                  exit ( 3 );
          }
  
         pthread_attr_init ( &amp;thread_attr );
         pthread_attr_setinheritsched ( &amp;thread_attr,
          PTHREAD_INHERIT_SCHED );
  
          #ifdef  GLOBAL
  
          sched_struct.sched_priority = sched_get_priority_max(SCHED_OTHER);
          pthread_attr_setinheritsched ( &amp;thread_attr,
          PTHREAD_EXPLICIT_SCHED );
          pthread_attr_setschedpolicy ( &amp;thread_attr, SCHED_OTHER );
          pthread_attr_setschedparam ( &amp;thread_attr, &amp;sched_struct );  
          pthread_attr_setscope ( &amp;thread_attr,
          PTHREAD_SCOPE_SYSTEM );
           #endif
  
      if ( pthread_create (&amp;thread_id[0], &amp;thread_attr,
                          producer, NULL ) != 0 ) {
          printf ( "pthread_create failed " );
          exit ( 3 );
      }
  
      for ( i = NUMPRODUCERS; i &lt; NUMCONSUMERS + 1; i++ ) {
          if ( pthread_create ( &amp;thread_id [i], &amp;thread_attr,
                  consumer, ( void * )&amp;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 &lt; NUMCONSUMERS + 1; i++ ) {
                      pthread_join ( thread_id [i], NULL ); 
      }
  
  /*****************************************************************/
  /* GET FINAL TIMESTAMP, CALCULATE ELAPSED SEC AND USEC           */
  /*****************************************************************/
  
  
      gettimeofday (&amp;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 &lt; 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 ( &amp;randtime, ( struct timezone * ) 0 );
      xsub1 [0] = ( unsigned short     )randtime.tv_usec;
      xsub1 [1] = ( unsigned short     ) ( randtime.tv_usec &gt;&gt; 16 );
      xsub1 [2] = ( unsigned short     ) ( pthread_self );    
  
      while ( 1 ) 
      {
        rand_donuts = nrand48 ( xsub1 ) &amp; 3;
  
        pthread_mutex_lock ( &amp;prod [rand_donuts] );
  
        /** check if there is space available to produce donuts **/
  
      while ( space_count [rand_donuts] == 0 ) 
      {
                  pthread_cond_wait ( &amp;prod_cond [rand_donuts], &amp;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 ( &amp;prod [rand_donuts] );
  
      /* get cons mutex */                
      /* inc donut count */               
      /* unlock cons mutex */         
      /* signal cons_condx_var */
  
      pthread_mutex_lock(&amp;cons [rand_donuts]);
  
      donut_count [rand_donuts]++;
  
      pthread_mutex_unlock(&amp;cons [rand_donuts]);
  
      pthread_cond_signal(&amp;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 ( &amp;randtime, ( struct timezone * ) 0 );
      xsub [0] = ( unsigned short )randtime.tv_usec;
      xsub [1] = ( unsigned short ) ( randtime.tv_usec &gt;&gt; 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 &lt; 10; i++ ) 
      {
  
  
          for( k = 0; k &lt; NUMFLAVORS; k++)
          {
                      for( m = 0; m &lt; 12; m++)
              {
                      collection[k][m] = -1;
  }
  
          }
  
  
  
          for(k = 0; k &lt; NUMFLAVORS; k++)
          {
              serial [k] = 0;
          }
  
  
          for( m = 0; m &lt; 12; m++ ) 
          {
  
              rand_donuts = nrand48( xsub ) &amp; 3;
  
  
          /*******************************************************        
          get cons mutex 
          check donut count
          loop:
          if donut count == 0
          wait cons_condx_var
          *******************************************************/
  
              pthread_mutex_lock ( &amp;cons [rand_donuts] );
              while ( donut_count [rand_donuts] == 0 ) 
              {
                  pthread_cond_wait ( &amp;cons_cond [rand_donuts], &amp;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 ( &amp;cons [rand_donuts] );
  
  
  
              /* lock producer mutex */
              pthread_mutex_lock(&amp;prod [rand_donuts]);
  
              /*increment the space count*/   
              space_count [rand_donuts] ++;
  
              /* unlock producer mutex */
              pthread_mutex_unlock(&amp;prod [rand_donuts]);
  
              //signal the producer
              pthread_cond_signal(&amp;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(&amp;current_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 &lt; NoOfDozens; i++)
  {
              for( m = 0; m &lt; 12  &amp;&amp; printresult &lt;=3 ; m++)
              {
                  printresult = 0;
                  for(k = 0; k &lt; NUMFLAVORS ; k++)
                  {               
                      if(collection[k][m] &gt;= 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 ( &amp;sigterm_signal );
      sigaddset ( &amp;sigterm_signal, SIGTERM );
      sigaddset ( &amp;sigterm_signal, SIGINT );
  
      if (sigwait ( &amp;sigterm_signal, &amp; 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 &lt; (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 );
  }
  </code></pre>
  
  <p>Can someone please tell me where I'm going wrong and what changes I need to make?? Thanks in advance!!</p>