gistfile1.c

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <semaphore.h>
#include "list.h"

List *fifo;

int pro_count;
int con_count;
int buffer_size;
int product_count;
int produced_count = 0; // How many have we produced?
int consumed_count = 0; // How many have we consumed?

sem_t empty;
sem_t full;

void *producer(void *param);
void *consumer(void *param);
void my_sleep(float wait_time_ms);

int main(int argc, char* argv[])
{
  // Our fifo buffer
  fifo = list_new();

  // Collect input from args
  pro_count = atoi(argv[1]);
  con_count = atoi(argv[2]);
  buffer_size = atoi(argv[3]);
  product_count = atoi(argv[4]);

  // Initialize semaphores
  sem_init(&empty, 0, buffer_size);
  sem_init(&full, 0, 0);

  // Seed the random number generator
  struct timeval tv;
  gettimeofday(&tv, NULL);
  srand(tv.tv_usec);

  // Create thread id arrays
  pthread_t pro_thread_ids[pro_count];
  pthread_t con_thread_ids[con_count];

  // Create producer threads
  int i;
  for(i = 0; i < pro_count; i++) {
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_create(&pro_thread_ids[i],&attr,producer,(void *) i);
  }

  // Create consumer threads
  for(i = 0; i < con_count; i++) {
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_create(&con_thread_ids[i],&attr,consumer,(void *) i);
  }

  // Join threads
  for(i = 0; i < pro_count; i++) pthread_join(pro_thread_ids[i], NULL);
  for(i = 0; i < con_count; i++) pthread_join(con_thread_ids[i], NULL);

  return 0;
}

void *producer(void *param)
{
  int producer_number = (int) param;
  while(produced_count < product_count) {
    //printf("Pro count: %i\n", produced_count);
    sem_wait(&empty); // Wait if there are no empty spots

    // Produce item
    char buffer[8];
    sprintf(buffer, "Item_%i", produced_count++);
    list_add(fifo, node_new_str(buffer));

    printf("Producer %i produced %s. Items in buffer: %i (out of %i).\n", producer_number, buffer, produced_count-consumed_count, product_count);
    my_sleep(1000);
    sem_post(&full); // Signal that a spot has been filled
  }

}

void *consumer(void *param)
{
  int consumer_number = (int) param;
  while(consumed_count < product_count) {
    //printf("Con count: %i\n", consumed_count);
    sem_wait(&full); // Wait if there are no items

    // Consume item
    char *item = (char *)list_remove(fifo)->elm;
    consumed_count++;

    printf("Consumer %i consumed %s. Items in buffer: %i (out of %i).\n", consumer_number, item, produced_count-consumed_count, product_count);
    my_sleep(1000);
    sem_post(&empty); // Signal that an item has been consumed
  }
}

/* Random sleep function */
void my_sleep(float wait_time_ms)
{
  wait_time_ms = ((float)rand())*wait_time_ms / (float)RAND_MAX;
  usleep((int) (wait_time_ms * 1e3f)); // convert from ms to us
}