c++ multiprocessing Barrier/CondVar/SharedMemory using SysV IPC

ipc.cc

#include "ipc.hpp"
#include <cassert>
#include <cstring>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/shm.h>

void *Shm::create(int key, size_t size, int shmflg) {
  flag = shmflg;
  shmid = shmget(key, size, shmflg);
  assert(shmid != -1);
  data = shmat(shmid, (void *)0, 0);
  assert(data != (void *)-1);
  return data;
}

void Shm::destroy() {
  shmdt(data);
  if (flag & IPC_CREAT) {
    shmctl(shmid, IPC_RMID, NULL);
  }
}

static int sem_op(int semid, unsigned short semnum, short op) {
  struct sembuf sb = {semnum, op, 0};
  return semop(semid, &sb, 1);
}

void CondVar::lock() { sem_op(semid, kMutx, -1); }

void CondVar::unlock() { sem_op(semid, kMutx, 1); }

void CondVar::create(Data *rhs, int semflag) {
  data = rhs;
  flag = semflag;
  semid = semget(data->key, kSize, semflag);
  if (semflag & IPC_CREAT) {
    semctl(semid, kMutx, SETVAL, 1);
    semctl(semid, kCond, SETVAL, 0);
  }
}

void CondVar::destroy() {
  if (flag & IPC_CREAT) {
    semctl(semid, 0, IPC_RMID);
  }
}

int CondVar::wait() {
  lock();
  data->wait_count++;
  unlock();
  if (sem_op(semid, kCond, -1)) {
    return -1;
  }
  lock();
  data->wait_count--;
  unlock();
  return 0;
}

void CondVar::notify_one() {
  lock();
  if (data->wait_count > 0) {
    sem_op(semid, kCond, 1);
  }
  unlock();
}

void CondVar::notify_all() {
  lock();
  auto waiting = data->wait_count;
  if (waiting > 0) {
    sem_op(semid, kCond, waiting);
  }
  unlock();
}

void Barrier::lock() { sem_op(semid, kMutx, -1); }

void Barrier::unlock() { sem_op(semid, kMutx, 1); }

void Barrier::create(Data *rhs, int bflag) {
  data = rhs;
  flag = bflag;
  const int key = data->key;
  if (flag & IPC_CREAT) {
    semid = semget(key, kSize, flag);
    semctl(semid, kCond0, SETVAL, 0);
    semctl(semid, kCond1, SETVAL, 0);
    semctl(semid, kMutx, SETVAL, 1);
  } else {
    semid = semget(key, kSize, flag);
  }
  lock();
  data->total++;
  unlock();
}

void Barrier::destroy() {
  if (flag & IPC_CREAT) {
    semctl(semid, 0, IPC_RMID);
  }
}

int Barrier::wait() {
  lock();
  data->counter++;
  const auto local_sense = data->flag ^ 1;
  if (data->counter == data->total) {
    data->counter = 0;
    data->flag = local_sense;
    unlock();
    // notify_all
    return sem_op(semid, local_sense, data->total - 1);
  }
  unlock();
  // wait
  return sem_op(semid, local_sense, -1);
}

ipc.hpp

#pragma once
#include <cstdint>
#include <sys/shm.h>

class Shm {
public:
  static constexpr const int kOpen = 0;
  static constexpr const int kCreate = 0600 | IPC_CREAT;

  template <typename T> T *create(int key, int shmflg = kOpen) {
    return static_cast<T *>(create(key, sizeof(T), shmflg));
  }
  void *create(int, size_t, int);
  void destroy();

private:
  int shmid, flag;
  void *data;
};

class CondVar {
public:
  static constexpr const int kOpen = 0;
  static constexpr const int kCreate = 0600 | IPC_CREAT;
  struct Data {
    int32_t key, wait_count;
  };

  void create(Data *, int);
  void destroy();
  int wait();
  void notify_one();
  void notify_all();

private:
  enum : unsigned short { kCond = 0x0, kMutx, kSize };
  int semid, flag;
  Data *data;

  void lock();
  void unlock();
};

class Barrier {
public:
  static constexpr const int kOpen = 0;
  static constexpr const int kCreate = 0600 | IPC_CREAT;
  struct Data {
    int32_t key, total, counter;
    unsigned short flag;
  };

  void create(Data *, int);
  void destroy();
  int wait();

private:
  enum : unsigned short { kCond0 = 0x0, kCond1, kMutx, kSize };
  int semid, flag;
  Data *data;

  void lock();
  void unlock();
};

main_barrier.cc

#include "ipc.hpp"
#include <cstdio>
#include <cstdlib>
#include <sys/wait.h>
#include <vector>

static void client(const int shm_key, int i) {
  Shm shm;
  Barrier barrier;
  auto data = shm.create<Barrier::Data>(shm_key);
  barrier.create(data, Barrier::kOpen);
  for (int j = 0; j < 4; ++j) {
    printf("Child %d: Wait %d.\n", i, j);
    sleep(rand() % 5);
    barrier.wait();
    printf("Child %d: Passed %d.\n", i, j);
  }
  shm.destroy();
}

int main() {
  printf("sizeof(Barrier::Data) = %zu\n", sizeof(Barrier::Data));
  Shm shm;
  Barrier barrier;
  const int shm_key = 0xcafe0;
  auto data = shm.create<Barrier::Data>(shm_key, Shm::kCreate);
  data->key = 0xcafe1;
  barrier.create(data, Barrier::kCreate);
  std::vector<pid_t> pids;
  for (size_t i = 1; i <= 5; ++i) {
    pids.push_back(fork());
    if (pids.back() == 0) {
      client(shm_key, i);
      exit(0);
    }
  }
  for (int j = 0; j < 4; ++j) {
    sleep(rand() % 5);
    printf("Parent: Wait %d.\n", j);
    barrier.wait();
    printf("Parent: Passed %d.\n", j);
  }
  // close
  barrier.destroy();
  shm.destroy();
  wait(0);
}

main_cv.cc

#include "ipc.hpp"
#include <cstdio>
#include <cstdlib>
#include <sys/wait.h>
#include <vector>

static void client(const int shm_key, int i) {
  Shm shm;
  CondVar cv;
  auto data = shm.create<CondVar::Data>(shm_key);
  cv.create(data, CondVar::kOpen);
  printf("Child %d: Waiting for signal...\n", i);
  cv.wait();
  printf("Child %d: Received signal!\n", i);
  shm.destroy();
}

int main() {
  printf("sizeof(CondVar::Data) = %zu\n", sizeof(CondVar::Data));
  Shm shm;
  CondVar cv;
  const int shm_key = 0xbeef0;
  auto data = shm.create<CondVar::Data>(shm_key, Shm::kCreate);
  cv.create(data, CondVar::kCreate);
  std::vector<pid_t> pids;
  for (size_t i = 0; i < 5; ++i) {
    pids.push_back(fork());
    if (pids.back() == 0) {
      client(shm_key, i);
      exit(0);
    }
  }
  sleep(3); // should be a barrier, but this example only wanted to demo CondVar
  printf("Parent: Broadcasting signal to all.\n");
  cv.notify_all();
  // close
  cv.destroy();
  shm.destroy();
  wait(0);
}