Interprocess sequencing system

main.cpp

/*

Write a simple sequence-number system through which three concurrent processes,
P1, P2, and P3, each obtain unique integers in the range [1, 500]. Use the
fork() call to create P, P2, and P3. Given a file, F, containing a single
number, each process must perform the following steps:

a. Open F.
b. Read the sequence number N from the file.
c. Close F.
d. Output N and the process' PID (either on screen or test file).
e. Increment N by 1
f. Open F.
g. Write N to F.
h. Flush F.
i. Close F.

*/

#include <assert.h>
#include <cstring>
#include <pthread.h>
#include <stdio.h>
#include <string>
#include <sys/shm.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <vector>

#define MAX_PROCS 3
#define FAIRNESS                                                               \
  1 // Makes each process sleep a bit to  make sure no one can hog the mutex to
    // completion.
#define MAX_SEQ 500

void die(int err) {
  perror(strerror(err));
  exit(1);
}

// Helper to lock a mutex with error handling.
void lockMtx(pthread_mutex_t *m) {
  int err;
  if ((err = pthread_mutex_lock(m) != 0)) {
    die(err);
  }
}

// Helper to unlock a mutex with error handling.
void unlockMtx(pthread_mutex_t *m) {
  int err;
  if ((err = pthread_mutex_unlock(m) != 0)) {
    die(err);
  }
}

std::vector<pid_t> startSequencing(pthread_mutex_t *m, char *filename) {
  std::vector<pid_t> pids;

  FILE *f;
  for (int i = 0; i < MAX_PROCS; i++) {
    pid_t pid = fork();
    if (pid == 0) {
      while (true) {
        lockMtx(m);

        // 1. Open file
        f = fopen(filename, "r");
        if (f == NULL) {
          printf("File does not exist?\n");
          unlockMtx(m);

          return std::vector<int>{-1};
        }

        // 2. Read seq num.
        int num;
        fscanf(f, "%d", &num);

        // 3. Close file
        fclose(f);

        // 4. Output N, and pid
        printf("pid: %d => seq number %d\n", getpid(), num);

        if (num >= MAX_SEQ) {
          unlockMtx(m);
          return std::vector<int>{-1};
        }

        // 5. Increment N by 1.
        num++;

        // 6. Open F.
        f = fopen(filename, "w+");
        assert(f != NULL);

        // 7. Write N to F.
        fprintf(f, "%d", num);

        // 8. Flush stdout
        fflush(stdout);

        // 9. Close F.
        fclose(f);

        unlockMtx(m);
        if (FAIRNESS) {
          sleep(1 / 10);
        }
      }
    } else {
      pids.push_back(pid);
    }
  }

  return pids;
}

pthread_mutex_t *initSharedMutex() {
  const int shmsz = sizeof(pthread_mutex_t);
  const int shmid = shmget(IPC_PRIVATE, shmsz, 0666);

  // Error case: Segment was not created.
  if (shmid < 0) {
    perror("shmget failed.");
    exit(1);
  }

  pthread_mutex_t *m = (pthread_mutex_t *)shmat(shmid, 0, 0);

  int err;
  pthread_mutexattr_t mutex_shared_attr;
  if ((err = pthread_mutexattr_init(&mutex_shared_attr))) {
    die(err);
  }

  if ((err = pthread_mutexattr_setpshared(&mutex_shared_attr,
                                          PTHREAD_PROCESS_SHARED)) != 0) {
    die(err);
  }

  if ((err = pthread_mutex_init(m, &mutex_shared_attr)) != 0) {
    die(err);
  }

  return m;
}

// Return shared memory to the system.
void clearSharedMemory(void *shmemPtr) {
  if (shmdt(shmemPtr) == -1) {
    perror("shmdt failed.");
    exit(1);
  }
}

int main(int argc, char *argv[]) {
  if (argc < 2) {
    printf("Please provide the filename as an argument: binary [filename]\n");

    return 1;
  }

  pthread_mutex_t *m = initSharedMutex();

  std::vector<pid_t> pids = startSequencing(m, argv[1]);
  for (auto pid : pids) {
    if (pid != -1) {
      waitpid(pid, NULL, 0);
    }
  }

  clearSharedMemory(m);

  return 0;
}

Makefile

# Compiler options
CXX = g++
CXXFLAGS = --std=c++11 -Wall -g -pthread

# Source code files, object files and the final binary name.
SRCS = $(wildcard *.cpp)
OBJS = $(SRCS:.cpp=.o)
BIN = binary

.PHONY: file clean

all: $(OBJS)
	$(CXX) $(CXXFLAGS) -o $(BIN) $(OBJS)

clean:
	$(RM) $(OBJS) $(BIN)
~