futex.cpp

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <atomic>
#include <cstdio>
#include <linux/futex.h>
#include <sys/syscall.h>
#include <sys/types.h>
#include <thread>
#include <unistd.h>
#include <vector>

#define CMP
const unsigned int producer = 1;
const unsigned int consumer = 1000;
const unsigned int num_test = 50;

unsigned int wakeup = 0;
int cnd = 0;
int mtx = 0;
int go = 0;

int futex(int *uaddr, int op, int val, void *val2, int *uaddr2, int val3) {
  return syscall(SYS_futex, uaddr, op, val, val2, uaddr2, val3);
}

void producer_func(void) {
  while (wakeup < consumer) {
    int ret;
    cnd = 1;
#ifndef CMP
    ret = futex(&cnd, FUTEX_WAKE_PRIVATE, consumer, NULL, NULL, 0);
    break;
#else
    int old = __atomic_load_n(&cnd, __ATOMIC_SEQ_CST);
    ret = syscall(SYS_futex, &cnd, FUTEX_CMP_REQUEUE_PRIVATE, 1, INT32_MAX,
                  &mtx, old);
    break;
#endif
  }

  while (wakeup < (consumer * 2 - 1)) {
    int ret;
    // wakeup remaing waiters
    ret = futex(&mtx, FUTEX_WAKE_PRIVATE, 1, NULL, NULL, 0);
  }
}

void consumer_func(void) {
#ifdef CMP
  int ret, old;

  do {
    old = __atomic_load_n(&cnd, __ATOMIC_SEQ_CST);
    while (cnd != 1) {
      ret = futex(&cnd, FUTEX_WAIT_PRIVATE, old, NULL, NULL, 0);
      __atomic_fetch_add(&wakeup, 1, __ATOMIC_SEQ_CST);
    }
  } while (ret != 0);

  // notify woken up
  __atomic_fetch_add(&wakeup, 1, __ATOMIC_SEQ_CST);
#else
  int ret, old;

  do {
    old = __atomic_load_n(&cnd, __ATOMIC_SEQ_CST);
    while (cnd != 1) {
      ret = futex(&cnd, FUTEX_WAIT_PRIVATE, old, NULL, NULL, 0);
    }
  } while (ret != 0);
  // notify woken up
  old = __atomic_fetch_add(&wakeup, 1, __ATOMIC_SEQ_CST);

  int zero = 0;
  int ten = 10;
  if (!__atomic_compare_exchange(&mtx, &zero, &ten, false, __ATOMIC_SEQ_CST,
                                 __ATOMIC_SEQ_CST)) {
    old = __atomic_load_n(&mtx, __ATOMIC_SEQ_CST);
    ret = futex(&mtx, FUTEX_WAIT_PRIVATE, old, NULL, NULL, 0);
    old = __atomic_fetch_add(&wakeup, 1, __ATOMIC_SEQ_CST);
  }
#endif
}

double test(void) {
  std::vector<std::thread> pro;
  std::vector<std::thread> con;
  std::chrono::system_clock::time_point start, end;

  // cleanup global resources
  wakeup = 0;
  cnd = 0;
  mtx = 0;

  for (auto i = 0; i < consumer; ++i) {
    con.push_back(std::thread(consumer_func));
  }
  usleep(10);

  start = std::chrono::system_clock::now();
  for (auto i = 0; i < producer; ++i) {
    pro.push_back(std::thread(producer_func));
  }
  for (auto &p : pro) {
    p.join();
  }
  for (auto &c : con) {
    c.join();
  }
  end = std::chrono::system_clock::now();
  return std::chrono::duration_cast<std::chrono::milliseconds>(end - start)
      .count();
}

int main(void) {
  double elapsed;

  for (auto i = 0; i < num_test; ++i) {
    elapsed += test();
  }
  std::printf("%lf[ms]\n", elapsed / num_test);
}