/******************************************************************************
* FILE: experiment-x2.c
* DESCRIPTION:
*   Demontrating consumer producer with threads which are synchronized
*   One mutex-variable is used to support thread synchronisation
*
* LAST REVISED: 25/08/03 Florian Weizenegger
******************************************************************************/

#include <pthread.h>
#include <sys/time.h>		/* for gettimeofday() */
#include <unistd.h>		/* for gettimeofday() */
#include <stdio.h>
#define NUM_THREADS	45
#define SIZEOFBUFFER    8000
#define NO_OF_ITERATIONS 10000

pthread_t consumers[NUM_THREADS];
pthread_t producers[NUM_THREADS];

pthread_mutex_t mutex_p, mutex_c;

int buffer[SIZEOFBUFFER];
int *start;
int *rp;
int *wp;

int myadd(int toAdd) {
  //buffer full
  if ((rp != (wp+1)) && (wp != rp + SIZEOFBUFFER - 1)) {
    *wp = toAdd;
    wp++;
    //buffer at its end
    if (wp == (start + SIZEOFBUFFER)) {
      wp = start;
    }
    return 1;
  } else return 0;
}

int myremove() {
  //buffer empty
  if (wp != rp) {
    int retValue = *rp;
    rp++;
    if (rp == (start + SIZEOFBUFFER)) {
      rp = start;
    }
    return retValue;
  } else return 0;
}


//Producer adds NO_OF_ITERATIONS times the value 10
//to the buffer
void *produce(void *threadid)
{
  int i = 0;
  int sum = 0;
  int ret = 0;

  printf("Producer #%d started...\n", threadid);

  while (i < NO_OF_ITERATIONS) {
    pthread_mutex_lock (&mutex_p);
    ret = myadd(10);
    if (ret) {
      i++;
      sum += 10;
    }
    pthread_mutex_unlock (&mutex_p);
  }
  printf("Producer #%d sum : %d\n", threadid, sum);
  pthread_exit(NULL);
}

//Consumer removes
void *consume(void *threadid)
{
  int i = 0;
  int sum = 0;
  int ret = 0;

  printf("Consumer #%d started...\n", threadid);

  while (i < NO_OF_ITERATIONS) {
    pthread_mutex_lock (&mutex_c);
    ret = myremove();
    if (ret != 0) {
      i++;
      sum += ret;
    }
    pthread_mutex_unlock (&mutex_c);
  }
  printf("Consumer #%d sum : %d\n", threadid, sum);
  pthread_exit(NULL);
}

int main(int argc, char *argv[])
{
  int tc, tp, status, i, rc;
  float delta;
  struct timeval start_time;
  struct timeval stop_time;

  gettimeofday( &start_time, NULL );

  start = &buffer[0];
  wp = start;
  rp = start;

  //init mutex
  pthread_mutex_init(&mutex_p, NULL);
  pthread_mutex_init(&mutex_c, NULL);
  
  for (i=0;i<NUM_THREADS;i++) {
    tc = pthread_create(&consumers[i], NULL, consume, (void *)i+1);
    if (tc) {
      printf("ERROR: couldn't create consumer thread\n");
      exit(-1);
    }
    tp = pthread_create(&producers[i], NULL, produce, (void *)i+1);
    if (tp) {
      printf("ERROR: couldn't create producer thread\n");
      exit(-1);
    }
  }

  for(i=0;i < NUM_THREADS;i++) {
    rc = pthread_join(consumers[i], (void **)&status);
    if (rc)
      {
	printf("ERROR; return code from pthread_join() is %d\n", rc);
	exit(-1);
      }
  }  
  
  pthread_mutex_destroy(&mutex_p);
  pthread_mutex_destroy(&mutex_c);

  gettimeofday( &stop_time, NULL );
  delta = (float)(stop_time.tv_sec  - start_time.tv_sec);
  delta += (stop_time.tv_usec - start_time.tv_usec)/(float)1000000;
  printf("\nTotal time : %.10f sec\n\n", delta);

  pthread_exit(NULL);
}