futex_latch.c

#define _GNU_SOURCE
#define _POSIX_C_SOURCE 200809L
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <errno.h>
#include <time.h>
#include <unistd.h>
#include <limits.h>
#if defined(__linux__)
#include <sys/syscall.h>
#include <linux/futex.h>
#include <sys/time.h>

static int futex_wait(atomic_int *addr, int expected)
{
    return syscall(SYS_futex, (int *)addr, FUTEX_WAIT, expected, NULL, NULL, 0);
}

static int futex_wake(atomic_int *addr)
{
    return syscall(SYS_futex, (int *)addr, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
}
#endif

typedef struct {
    atomic_int state;
#if !defined(__linux__)
    pthread_mutex_t m;
    pthread_cond_t c;
#endif
} latch_t;

static void latch_init(latch_t *l)
{
    atomic_init(&l->state, 0);
#if !defined(__linux__)
    pthread_mutex_init(&l->m, NULL);
    pthread_cond_init(&l->c, NULL);
#endif
}

static void latch_release(latch_t *l)
{
    int prev = atomic_exchange_explicit(&l->state, 1, memory_order_release);
#if defined(__linux__)
    if (prev == 0) futex_wake(&l->state);
#else
    if (prev == 0) {
        pthread_mutex_lock(&l->m);
        pthread_cond_broadcast(&l->c);
        pthread_mutex_unlock(&l->m);
    }
#endif
}

static void latch_wait(latch_t *l)
{
    for (;;) {
        int s = atomic_load_explicit(&l->state, memory_order_acquire);
        if (s == 1) return;
#if defined(__linux__)
        futex_wait(&l->state, 0);
#else
        pthread_mutex_lock(&l->m);
        while (atomic_load_explicit(&l->state, memory_order_acquire) == 0) {
            pthread_cond_wait(&l->c, &l->m);
        }
        pthread_mutex_unlock(&l->m);
#endif
    }
}

typedef struct {
    latch_t *l;
    int id;
} arg_t;

static void *worker(void *p)
{
    arg_t *a = p;
    latch_wait(a->l);
    printf("worker %d passed latch\n", a->id);
    return NULL;
}

int main(void)
{
    latch_t l;
    latch_init(&l);
    pthread_t th[4];
    arg_t a[4];
    for (int i = 0; i < 4; i++) {
        a[i].l = &l;
        a[i].id = i;
        pthread_create(&th[i], NULL, worker, &a[i]);
    }
    struct timespec ts = { .tv_sec = 1, .tv_nsec = 0 };
    nanosleep(&ts, NULL);
    printf("releasing latch\n");
    latch_release(&l);
    for (int i = 0; i < 4; i++) {
        pthread_join(th[i], NULL);
    }
    return 0;
}