simulation.c

/*
 * Copyright 2021 Jeisson Hidalgo-Cespedes - Universidad de Costa Rica
 */

#include <assert.h>
#include <inttypes.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>

#include "simulation.h"

int simulate_producer_consumer(shared_thread_data_t* shared_data) {
  assert(shared_data);
  int error = 0;
  if (shared_data) {
    error += queue_init(&shared_data->queue);
    error += sem_init(&shared_data->can_consume, /*pshared*/0, 0);
    error += pthread_mutex_init(&shared_data->stdout_mutex, /*attr*/NULL);
    error += pthread_mutex_init(&shared_data->next_product_mutex
      , /*attr*/NULL);

    if (error == 0) {
      struct timespec start_time, finish_time;
      clock_gettime(/*clk_id*/CLOCK_MONOTONIC, &start_time);

      error = create_threads(shared_data);

      clock_gettime(/*clk_id*/CLOCK_MONOTONIC, &finish_time);
      double elapsed_time = finish_time.tv_sec - start_time.tv_sec +
        (finish_time.tv_nsec - start_time.tv_nsec) * 1e-9;
      printf("execution time: %.9lfs\n", elapsed_time);

      pthread_mutex_destroy(&shared_data->next_product_mutex);
      pthread_mutex_destroy(&shared_data->stdout_mutex);
    } else {
      fprintf(stderr, "error: could not init mutex\n");
      error = 11;
    }

    queue_free(&shared_data->queue);
    queue_destroy(&shared_data->queue);
    free(shared_data);
  } else {
    fprintf(stderr, "error: could not allocated shared memory\n");
    error = 12;
  }

  return error;
}

int create_threads(shared_thread_data_t* shared_data) {
  assert(shared_data);
  int error = 0;

  const size_t thread_count = shared_data->producer_count
    + shared_data->consumer_count;

  pthread_t* threads = (pthread_t*) malloc(thread_count * sizeof(pthread_t));

  private_thread_data_t* private_data = (private_thread_data_t*)
    calloc(thread_count, sizeof(private_thread_data_t));

  if (threads && private_data) {
    for (size_t index = 0; index < shared_data->producer_count; ++index) {
      private_data[index].thread_number = index;
      private_data[index].shared_data = shared_data;

      error = pthread_create(&threads[index], NULL, produce
        , &private_data[index]);

      if (error) {
        fprintf(stderr, "error: could not create thread %zu\n", index);
        error = 21;
        break;
      }
    }

    for (size_t index = 0; index < shared_data->consumer_count; ++index) {
      const size_t array_index = shared_data->producer_count + index;
      private_data[array_index].thread_number = index;
      private_data[array_index].shared_data = shared_data;

      error = pthread_create(&threads[array_index], NULL, consume
        , &private_data[array_index]);

      if (error) {
        fprintf(stderr, "error: could not create thread %zu\n", array_index);
        error = 21;
        break;
      }
    }

    for (size_t index = 0; index < thread_count; ++index) {
      pthread_join(threads[index], NULL);
    }

    free(private_data);
    free(threads);
  } else {
    fprintf(stderr, "error: could not allocate memory for %zu threads\n"
      , thread_count);
    error = 22;
  }

  return error;
}

void* produce(void* data) {
  assert(data);
  private_thread_data_t* private_data = (private_thread_data_t*)data;
  shared_thread_data_t *shared_data = private_data->shared_data;

  struct timespec time;
  clock_gettime(/*clk_id*/CLOCK_MONOTONIC, &time);
  unsigned int seed = time.tv_nsec;

  while (true) {
    pthread_mutex_lock(&shared_data->next_product_mutex);
    const size_t product_index = shared_data->next_product_index++;
    pthread_mutex_unlock(&shared_data->next_product_mutex);

    if (product_index >= shared_data->product_count) {
      // Give all consumers an opportunity to leave the semaphor
      if (product_index >= shared_data->product_count
        + shared_data->producer_count - 1) {
        printf("All producers finished\n");
        for (size_t index = 0; index < shared_data->consumer_count; ++index) {
          sem_post(&shared_data->can_consume);
        }
      }
      break;
    }

    random_delay(shared_data->min_producer_delay
      , shared_data->max_producer_delay, &seed);

    pthread_mutex_lock(&shared_data->stdout_mutex);
    printf("%zu produced %zu\n", private_data->thread_number
      , product_index + 1);
    pthread_mutex_unlock(&shared_data->stdout_mutex);

    queue_append(&shared_data->queue, product_index);
    sem_post(&shared_data->can_consume);
  }

  return NULL;
}

void* consume(void* data) {
  assert(data);

  private_thread_data_t* private_data = (private_thread_data_t*)data;
  shared_thread_data_t *shared_data = private_data->shared_data;

  struct timespec time;
  clock_gettime(/*clk_id*/CLOCK_MONOTONIC, &time);
  unsigned int seed = time.tv_nsec;

  while (true) {
    sem_wait(&shared_data->can_consume);

    if (queue_is_empty(&shared_data->queue)) {
      break;
    }

    size_t product_index = queue_dequeue(&shared_data->queue);

    pthread_mutex_lock(&shared_data->stdout_mutex);
    printf("\t\t%zu consuming %zu\n", private_data->thread_number
      , product_index + 1);
    pthread_mutex_unlock(&shared_data->stdout_mutex);

    random_delay(shared_data->min_consumer_delay
      , shared_data->max_consumer_delay, &seed);

    // pthread_mutex_lock(&shared_data->stdout_mutex);
    // printf("\t\t%zu consumed %zu\n", private_data->thread_number
      // , product_index + 1);
    // pthread_mutex_unlock(&shared_data->stdout_mutex);
  }

  return NULL;
}

void random_delay(useconds_t min, useconds_t max, unsigned* seedp) {
  assert(min <= max);
  useconds_t milliseconds = min;
  if (max > min) {
    milliseconds += rand_r(seedp) % (max - min);
  }
  usleep(milliseconds * 1000);
}