readers_writers.c

/*
 * Copyright 2021 Jeisson Hidalgo-Cespedes - Universidad de Costa Rica
 * Creates a secondary thread that greets in the standard output
 */

#include <ctype.h>
#include <inttypes.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

typedef struct shared_thread_data {
  size_t thread_count;
  pthread_mutex_t stdout_mutex;
  pthread_rwlock_t rwlock;
  size_t counter;
} shared_thread_data_t;

typedef struct private_thread_data {
  size_t thread_number;  // rank
  shared_thread_data_t* shared_thread_data;
} private_thread_data_t;

void* reader(void* data);
void* writer(void* data);
int create_threads(shared_thread_data_t* shared_thread_data);

int main(int argc, char* argv[]) {
  int error = 0;

  size_t thread_count = sysconf(_SC_NPROCESSORS_ONLN);
  if (argc >= 2) {
    thread_count = strtoull(argv[1], NULL, 10);
  }

  shared_thread_data_t* shared_thread_data = (shared_thread_data_t*)
    calloc(1, sizeof(shared_thread_data_t));

  if (shared_thread_data) {
    shared_thread_data->thread_count = thread_count;
    error = pthread_mutex_init(&shared_thread_data->stdout_mutex, /*attr*/NULL);
    error += pthread_rwlock_init(&shared_thread_data->rwlock, /*attr*/ NULL);

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

      error = create_threads(shared_thread_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_rwlock_destroy(&shared_thread_data->rwlock);
      pthread_mutex_destroy(&shared_thread_data->stdout_mutex);
    } else {
      fprintf(stderr, "error: could not init mutex\n");
      error = 11;
    }

    free(shared_thread_data);
  } else {
    fprintf(stderr, "error: could not allocated shared memory\n");
    error = 12;
  }

  return error;
}

int create_threads(shared_thread_data_t* shared_thread_data) {
  int error = 0;
  pthread_t* threads = (pthread_t*)
    malloc(shared_thread_data->thread_count * sizeof(pthread_t));

  private_thread_data_t* private_thread_data = (private_thread_data_t*)
    calloc(shared_thread_data->thread_count, sizeof(private_thread_data_t));

  if (threads && private_thread_data) {
    char ch = 0;
    size_t count = 0;
    while (scanf("%c", &ch) == 1 ) {
      private_thread_data[count].thread_number = count;
      private_thread_data[count].shared_thread_data = shared_thread_data;

      switch (tolower(ch)) {
        case 'r':
          error = pthread_create(&threads[count], NULL, reader
            , &private_thread_data[count]);
          break;

        case 'w':
          error = pthread_create(&threads[count], NULL, writer
            , &private_thread_data[count]);
          break;

        default:
          // ...
          break;
      }
      if (error) {
        fprintf(stderr, "error: could not create thread %zu\n", count);
        error = 21;
      }
      ++count;
    }

    pthread_mutex_lock(&shared_thread_data->stdout_mutex);
    printf("Hello from main thread\n");
    pthread_mutex_unlock(&shared_thread_data->stdout_mutex);

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

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

  return error;
}

void* reader(void* data) {
  private_thread_data_t* private_data = (private_thread_data_t*)data;
  shared_thread_data_t *shared_thread_data = private_data->shared_thread_data;

  pthread_rwlock_rdlock(&shared_thread_data->rwlock);
  size_t value = shared_thread_data->counter;
  pthread_rwlock_unlock(&shared_thread_data->rwlock);

  pthread_mutex_lock(&private_data->shared_thread_data->stdout_mutex);
  printf("Reader %zu got %zu\n"
    , (*private_data).thread_number, value);
  pthread_mutex_unlock(&shared_thread_data->stdout_mutex);

  return NULL;
}

void* writer(void* data) {
  private_thread_data_t* private_data = (private_thread_data_t*)data;
  shared_thread_data_t *shared_thread_data = private_data->shared_thread_data;

  pthread_rwlock_wrlock(&shared_thread_data->rwlock);
  size_t value = ++shared_thread_data->counter;
  pthread_rwlock_unlock(&shared_thread_data->rwlock);

  pthread_mutex_lock(&private_data->shared_thread_data->stdout_mutex);
  printf("Writer %zu increased to %zu\n"
    , (*private_data).thread_number, value);
  pthread_mutex_unlock(&shared_thread_data->stdout_mutex);

  return NULL;
}