rcu_list.c

#define _POSIX_C_SOURCE 200809L
#define _GNU_SOURCE

#include <stdatomic.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/wait.h>

typedef struct rcu_node {
    struct rcu_node *next;
    int value;
} rcu_node_t;

typedef struct {
    _Atomic(rcu_node_t *) head;
    _Atomic(int) quiescent[64];
    int nthreads;
    pthread_mutex_t write_lock;
} rcu_list_t;

typedef struct {
    rcu_list_t *list;
    int tid;
} rarg_t;

static void rcu_list_init(rcu_list_t *l, int nthreads)
{
    atomic_store_explicit(&l->head, NULL, memory_order_relaxed);
    l->nthreads = nthreads;
    for (int i = 0; i < 64; i++) {
        atomic_store_explicit(&l->quiescent[i], 0, memory_order_relaxed);
    }
    pthread_mutex_init(&l->write_lock, NULL);
}

static rcu_node_t *rcu_read_begin(rcu_list_t *l, int tid)
{
    atomic_store_explicit(&l->quiescent[tid], 0, memory_order_release);
    return atomic_load_explicit(&l->head, memory_order_acquire);
}

static void rcu_read_end(rcu_list_t *l, int tid)
{
    atomic_store_explicit(&l->quiescent[tid], 1, memory_order_release);
}

static void rcu_sync(rcu_list_t *l)
{
    for (;;) {
        int all = 1;
        for (int i = 0; i < l->nthreads; i++) {
            int q = atomic_load_explicit(&l->quiescent[i], memory_order_acquire);
            if (!q) {
                all = 0;
                break;
            }
        }
        if (all) break;
        usleep(1000);
    }
}

static void rcu_insert_front(rcu_list_t *l, int value)
{
    pthread_mutex_lock(&l->write_lock);
    rcu_node_t *n = malloc(sizeof(rcu_node_t));
    n->value = value;
    rcu_node_t *old;
    do {
        old = atomic_load_explicit(&l->head, memory_order_acquire);
        n->next = old;
    } while (!atomic_compare_exchange_weak_explicit(&l->head, &old, n, memory_order_release, memory_order_acquire));
    pthread_mutex_unlock(&l->write_lock);
}

static void rcu_remove_value(rcu_list_t *l, int value)
{
    pthread_mutex_lock(&l->write_lock);
    rcu_node_t *head = atomic_load_explicit(&l->head, memory_order_acquire);
    rcu_node_t dummy;
    dummy.next = head;
    rcu_node_t *prev = &dummy;
    rcu_node_t *cur = head;
    rcu_node_t *removed = NULL;
    while (cur) {
        if (cur->value == value) {
            prev->next = cur->next;
            removed = cur;
            break;
        }
        prev = cur;
        cur = cur->next;
    }
    if (removed) {
        atomic_store_explicit(&l->head, dummy.next, memory_order_release);
        rcu_sync(l);
        free(removed);
    }
    pthread_mutex_unlock(&l->write_lock);
}

static void *reader(void *arg)
{
    rarg_t *a = arg;
    rcu_list_t *l = a->list;
    int tid = a->tid;
    for (int i = 0; i < 1000; i++) {
        rcu_node_t *h = rcu_read_begin(l, tid);
        int sum = 0;
        for (rcu_node_t *p = h; p; p = p->next) {
            sum += p->value;
        }
        rcu_read_end(l, tid);
        if ((i % 100) == 0) {
            printf("reader %d sum=%d\n", tid, sum);
        }
        usleep(1000);
    }
    return NULL;
}

static void *writer(void *arg)
{
    rarg_t *a = arg;
    rcu_list_t *l = a->list;
    for (int i = 0; i < 20; i++) {
        rcu_insert_front(l, i);
        usleep(5000);
    }
    for (int i = 0; i < 20; i++) {
        rcu_remove_value(l, i);
        usleep(5000);
    }
    return NULL;
}

int main(void)
{
    rcu_list_t l;
    int nthreads = 4;
    rcu_list_init(&l, nthreads);
    pthread_t r[3];
    pthread_t w;
    rarg_t ar[3];
    rarg_t aw;
    aw.list = &l;
    aw.tid = 3;
    pthread_create(&w, NULL, writer, &aw);
    for (int i = 0; i < 3; i++) {
        ar[i].list = &l;
        ar[i].tid = i;
        pthread_create(&r[i], NULL, reader, &ar[i]);
    }
    for (int i = 0; i < 3; i++) {
        pthread_join(r[i], NULL);
    }
    pthread_join(w, NULL);
    rcu_node_t *p = atomic_load_explicit(&l.head, memory_order_acquire);
    while (p) {
        rcu_node_t *n = p->next;
        free(p);
        p = n;
    }
    return 0;
}