Readers/Writers - CS 702 - Fall 2003

Code for readers and writers (rw.cpp)

#include <cstdlib>
#include <iostream>

#include <unistd.h>
#include <pthread.h>

#include "object.h"

const int RW_DELAY_MAX = 4;
const int MAIN_DELAY_MAX = 2;
const int RECIP_P_WRITER = 3;

struct rw_args
{
  int id;
  Object *obj;
};

void *reader(void *p)
{
  rw_args *args = static_cast(p);
      
  std::cout << "Reader " << args->id << " preparing to read" << std::endl;
  args->obj->beginRead();
  
  std::cout << "Reader " << args->id << " reading" << std::endl;
  int delay = random() % RW_DELAY_MAX;
  sleep(delay);
  
  std::cout << "Reader " << args->id << " finished reading" << std::endl;
  args->obj->endRead();
  
  delete args;

  return NULL;
}

void *writer(void *p)
{
  rw_args *args = static_cast(p);
 
  std::cout << "Writer " << args->id << " preparing to write" << std::endl;
  args->obj->beginWrite();
  
  std::cout << "Writer " << args->id << " writing" << std::endl;
  int delay = random() % RW_DELAY_MAX;
  sleep(delay);
  
  std::cout << "Writer " << args->id << " finished writing" << std::endl;
  args->obj->endWrite();

  delete args;

  return NULL;
}

int main(int argc, char **argv)
{
  int readers = 0;
  int writers = 0;

  Object *obj = new Object();

  while (1)
    {
      int rnd = random() % RECIP_P_WRITER;
      rw_args *args = new rw_args();
      args->obj = obj;
      pthread_t *thread = new pthread_t();
      
      if (rnd == 0)
	{
	  // make a writer
	  
	  args->id = writers++;

	  pthread_create(thread, NULL, writer, args);
	}
      else
	{
	  // make a reader

	  args->id = readers++;

	  pthread_create(thread, NULL, reader, args);
	}

      sleep(random() % MAIN_DELAY_MAX);
    }
}

object.h

#ifndef __OBJECT_H__
#define __OBJECT_H__

class Object
{
public:
  Object();

  void beginRead();
  void endRead();

  void beginWrite();
  void endWrite();

private:
  int readersReading;
  int readersWaiting;
  int writersWaiting;

  enum {IDLE, READING, WRITING};
  int state;

  static const pthread_mutex_t DEFAULT_MUTEX;
  static const pthread_cond_t DEFAULT_COND;

  pthread_mutex_t mutex;
  pthread_cond_t read;
  pthread_cond_t write;
};

#endif

object.cpp

#include <pthread.h>

#include "object.h"

const pthread_cond_t Object::DEFAULT_COND = PTHREAD_COND_INITIALIZER;
const pthread_mutex_t Object::DEFAULT_MUTEX = PTHREAD_MUTEX_INITIALIZER;

Object::Object()
  : mutex(DEFAULT_MUTEX),
    write(DEFAULT_COND),
    read(DEFAULT_COND)
{
  readersReading = 0;
  readersWaiting = 0;
  writersWaiting = 0;
  state = IDLE;
}

void Object::beginRead()
{
  pthread_mutex_lock(&mutex);

  if (state == IDLE || (state == READING && writersWaiting == 0))
      {
	state = READING;
	readersReading++;
      }
  else
    {
      readersWaiting++;
      pthread_cond_wait(&read, &mutex);
    }

  pthread_mutex_unlock(&mutex);
}

void Object::endRead()
{
  pthread_mutex_lock(&mutex);

  readersReading--;
  if (readersReading == 0 && writersWaiting > 0)
    {
      state = WRITING;
      writersWaiting--;
      pthread_cond_signal(&write);
    }
  else if (readersReading == 0)
    state = IDLE;

  pthread_mutex_unlock(&mutex);
}

void Object::beginWrite()
{
  pthread_mutex_lock(&mutex);

  if (state != IDLE)
    {
      writersWaiting++;
      pthread_cond_wait(&write, &mutex);
    }
  else
    state = WRITING;

  pthread_mutex_unlock(&mutex);
}

void Object::endWrite()
{
  pthread_mutex_lock(&mutex);

  if (readersWaiting > 0)
    {
      readersReading = readersWaiting;
      readersWaiting = 0;
      state = READING;
      pthread_cond_broadcast(&read);
    }
  else if (writersWaiting > 0)
    {
      writersWaiting--;
      pthread_cond_signal(&write);
    }
  else
    state = IDLE;

  pthread_mutex_unlock(&mutex);
}