Add thread-based RCU

linD026 · linD026 · commit bd2a65933f6c · 2022-04-25T00:56:26.000+08:00
Add the Linux Kernel style thread-based simple RCU. It supports sparse checking[1]. With sparse, it will report: main.c: note: in included file: thrd_rcu.h:95:42: warning: Using plain integer as NULL pointer thrd_rcu.h:95:42: warning: Using plain integer as NULL pointer It is from the pthread_mutex_t initializer, we can ignore it. thread-rcu target to use Linux Kernel Memory Model (LKMM). C11 memory model may no compatible to LKMM. We need to be careful about ower architecture. you can check the paper[2] to see more detail. [1] https://www.kernel.org/doc/html/latest/dev-tools/sparse.html [2] http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2018/p0124r6.html
diff --git a/README.md b/README.md
@@ -30,6 +30,7 @@ purpose of these programs is to be illustrative and educational.
     - [qsbr](qsbr/): An implementation of Quiescent state based reclamation (QSBR).
     - [list-move](list-move/): Evaluation of two concurrent linked lists: QSBR and lock-based.
     - [rcu\_queue](rcu_queue/): An efficient concurrent queue based on QSBR.
+    - [thread-rcu](thread-rcu/): A Linux Kernel style thread-based simple RCU.
 * Applications
     - [httpd](httpd/): A multi-threaded web server.
     - [map-reduce](map-reduce/): word counting using MapReduce.
diff --git a/thread-rcu/Makefile b/thread-rcu/Makefile
@@ -0,0 +1,20 @@
+CFLAGS = -Wall
+CFLAGS += -g
+CFLAGS += -std=c11
+CFLAGS += -D'N_READERS=10'
+CFLAGS += -fsanitize=thread
+LDFLAGS += -lpthread
+
+all:
+	$(CC) -o main main.c $(CFLAGS) $(LDFLAGS)
+
+# Semantic Checker
+# https://www.kernel.org/doc/html/latest/dev-tools/sparse.html
+sparse:
+	sparse main.c $(CFLAGS) $(LDFLAGS)
+
+indent:
+	clang-format -i *.[ch]
+
+clean:
+	rm -f main
diff --git a/thread-rcu/main.c b/thread-rcu/main.c
@@ -0,0 +1,74 @@
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "rcu.h"
+
+struct test {
+    unsigned int count;
+};
+
+static struct test __rcu *dut;
+
+static void *reader_side(void *argv)
+{
+    struct test *tmp;
+
+    if (rcu_init())
+        abort();
+
+    rcu_read_lock();
+
+    tmp = rcu_dereference(dut);
+
+    printf("[reader %u] %u\n", current_tid(), tmp->count);
+
+    rcu_read_unlock();
+
+    pthread_exit(NULL);
+}
+
+static void *updater_side(void *argv)
+{
+    struct test *oldp;
+    struct test *newval = malloc(sizeof(struct test));
+
+    newval->count = current_tid();
+
+    printf("[updater %u]\n", newval->count);
+
+    oldp = rcu_assign_pointer(dut, newval);
+
+    synchronize_rcu();
+    free(oldp);
+
+    pthread_exit(NULL);
+}
+
+int main(int argc, char *argv[])
+{
+    pthread_t reader[N_READERS];
+    pthread_t updater;
+    int i;
+
+    dut = (struct test __rcu *) malloc(sizeof(struct test));
+    rcu_uncheck(dut)->count = 0;
+
+    for (i = 0; i < N_READERS / 2; i++)
+        pthread_create(&reader[i], NULL, reader_side, NULL);
+
+    pthread_create(&updater, NULL, updater_side, NULL);
+
+    for (i = N_READERS / 2; i < N_READERS; i++)
+        pthread_create(&reader[i], NULL, reader_side, NULL);
+
+    for (i = 0; i < N_READERS; i++)
+        pthread_join(reader[i], NULL);
+
+    pthread_join(updater, NULL);
+
+    free(rcu_uncheck(dut));
+    rcu_clean();
+
+    return 0;
+}
diff --git a/thread-rcu/rcu.h b/thread-rcu/rcu.h
@@ -0,0 +1,250 @@
+/* Target to use Linux Kernel Memory Model (LKMM) for thread-rcu,
+ * C11 memory model may no compatible to LKMM.
+ * Be careful about the architecture you use.
+ * you can check the paper to see more detail:
+ *
+ * http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2018/p0124r6.html
+ */
+
+#ifndef __RCU_H__
+#define __RCU_H__
+
+/* lock primitives from pthread and compiler primitives */
+
+#include <pthread.h>
+#include <stdatomic.h>
+
+typedef pthread_mutex_t spinlock_t;
+
+#define SPINLOCK_INIT PTHREAD_MUTEX_INITIALIZER
+static inline void spin_lock(spinlock_t *sp)
+{
+    int ret;
+
+    ret = pthread_mutex_lock(sp);
+    if (ret != 0) {
+        fprintf(stderr, "spin_lock:pthread_mutex_lock %d\n", ret);
+        abort();
+    }
+}
+
+static inline void spin_unlock(spinlock_t *sp)
+{
+    int ret;
+
+    ret = pthread_mutex_unlock(sp);
+    if (ret != 0) {
+        fprintf(stderr, "spin_unlock:pthread_mutex_unlock %d\n", ret);
+        abort();
+    }
+}
+
+#define current_tid() (unsigned int) pthread_self()
+
+/* Be careful here, since the C11 terms do no have the same sequential
+ * consistency  for the smp_mb(). Here we use the closely C11 terms,
+ * memory_order_seq_cst.
+ */
+#define smp_mb() atomic_thread_fence(memory_order_seq_cst)
+
+/* Compiler barrier, prevent the compiler to move the memory access from side
+ * to the other side.
+ */
+#define barrier() __asm__ __volatile__("" : : : "memory")
+
+/* Don't use it directly. Use READ_ONCE() or WRITE_ONCE(). */
+#define ACCESS_ONCE(x) (*(volatile __typeof__(x) *) &(x))
+
+/* READ_ONCE() close to those of a C11 volatile memory_order_relaxed atomic read
+ * . However, for address, data, or control dependency chain, it is more like
+ * memory_order_consume. But, presently most of implement promote those kind of
+ * thing to memory_order_aquire.
+ */
+#define READ_ONCE(x)                                                     \
+    ({                                                                   \
+        barrier();                                                       \
+        __typeof__(x) ___x =                                             \
+            atomic_load_explicit(&ACCESS_ONCE(x), memory_order_consume); \
+        barrier();                                                       \
+        ___x;                                                            \
+    })
+
+/* WRITE_ONCE() quite close to C11 volatile memory_order_relaxed atomic store */
+#define WRITE_ONCE(x, val)                                                   \
+    do {                                                                     \
+        atomic_store_explicit(&ACCESS_ONCE(x), (val), memory_order_relaxed); \
+    } while (0)
+
+#include <errno.h>
+#include <stdatomic.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+#ifdef __CHECKER__
+#define __rcu __attribute__((noderef, address_space(__rcu)))
+#define rcu_check_sparse(p, space) ((void) (((typeof(*p) space *) p) == p))
+#define __force __attribute__((force))
+#define rcu_uncheck(p) ((__typeof__(*p) __force *) p)
+#define rcu_check(p) ((__typeof__(*p) __force __rcu *) p)
+#else
+#define __rcu
+#define rcu_check_sparse(p, space)
+#define __force
+#define rcu_uncheck(p) p
+#define rcu_check(p) p
+#endif /* __CHECKER__ */
+
+/* Avoid false sharing */
+#define __rcu_aligned __attribute__((aligned(128)))
+
+struct rcu_node {
+    unsigned int tid;
+    int rcu_nesting[2];
+    struct rcu_node *next;
+} __rcu_aligned;
+
+struct rcu_data {
+    unsigned int nr_thread;
+    struct rcu_node *head;
+    unsigned int rcu_thread_nesting_idx;
+    spinlock_t sp;
+};
+
+/* Easy to use */
+#define __rcu_thread_idx rcu_data.rcu_thread_nesting_idx
+#define __rcu_thread_nesting(ptr) \
+    ptr->rcu_nesting[READ_ONCE(__rcu_thread_idx) & 0x01]
+#define rcu_thread_nesting __rcu_thread_nesting(__rcu_per_thread_ptr)
+
+static struct rcu_data rcu_data = {
+    .nr_thread = 0,
+    .head = NULL,
+    .rcu_thread_nesting_idx = 0,
+    .sp = SPINLOCK_INIT,
+};
+static __thread struct rcu_node *__rcu_per_thread_ptr;
+
+static inline struct rcu_node *__rcu_node_add(unsigned int tid)
+{
+    struct rcu_node **indirect = &rcu_data.head;
+    struct rcu_node *node = malloc(sizeof(struct rcu_node));
+
+    if (!node) {
+        fprintf(stderr, "__rcu_node_add: malloc failed\n");
+        abort();
+    }
+
+    node->tid = tid;
+    node->rcu_nesting[0] = 0;
+    node->rcu_nesting[1] = 0;
+    node->next = NULL;
+
+    spin_lock(&rcu_data.sp);
+
+    while (*indirect) {
+        if ((*indirect)->tid == node->tid) {
+            spin_unlock(&rcu_data.sp);
+            free(node);
+            return NULL;
+        }
+        indirect = &(*indirect)->next;
+    }
+
+    *indirect = node;
+    rcu_data.nr_thread++;
+
+    spin_unlock(&rcu_data.sp);
+
+    smp_mb();
+
+    return node;
+}
+
+static inline int rcu_init(void)
+{
+    unsigned int tid = current_tid();
+
+    __rcu_per_thread_ptr = __rcu_node_add(tid);
+
+    return (__rcu_per_thread_ptr == NULL) ? -ENOMEM : 0;
+}
+
+static inline void rcu_clean(void)
+{
+    struct rcu_node *node, *tmp;
+
+    spin_lock(&rcu_data.sp);
+
+    for (node = rcu_data.head; node; node = tmp) {
+        tmp = node->next;
+        free(node);
+    }
+
+    rcu_data.head = NULL;
+    rcu_data.nr_thread = 0;
+
+    spin_unlock(&rcu_data.sp);
+}
+
+/* The per-thread reference count will only modified by their owner
+ * thread but will read by other threads. So here we use WRITE_ONCE().
+ */
+static inline void rcu_read_lock(void)
+{
+    WRITE_ONCE(rcu_thread_nesting, 1);
+}
+
+static inline void rcu_read_unlock(void)
+{
+    WRITE_ONCE(rcu_thread_nesting, 0);
+}
+
+static inline void synchronize_rcu(void)
+{
+    struct rcu_node *node;
+
+    smp_mb();
+
+    spin_lock(&rcu_data.sp);
+
+    /* When the rcu_thread_nesting is odd, which means that the LSB set 1,
+     * that thread is in the read-side critical section. Also, we need
+     * to skip the read side when it is in the new grace period.
+     */
+    for (node = rcu_data.head; node; node = node->next) {
+        while (READ_ONCE(__rcu_thread_nesting(node)) & 0x1) {
+            barrier();
+        }
+    }
+
+    /* Going to next grace period
+     *
+     * It would be great to put smp_mb() (here implemented by memory_seq_cst)
+     * after and before of C11 atomics.
+     */
+    atomic_fetch_add_explicit(&__rcu_thread_idx, 1, memory_order_release);
+
+    spin_unlock(&rcu_data.sp);
+
+    smp_mb();
+}
+
+#define rcu_dereference(p)                                                 \
+    ({                                                                     \
+        __typeof__(*p) *__r_d_p = (__typeof__(*p) __force *) READ_ONCE(p); \
+        rcu_check_sparse(p, __rcu);                                        \
+        __r_d_p;                                                           \
+    })
+
+#define rcu_assign_pointer(p, v)                                 \
+    ({                                                           \
+        __typeof__(*p) *__r_a_p =                                \
+            (__typeof__(*p) __force *) atomic_exchange_explicit( \
+                &(p), (__typeof__(*(p)) __force __rcu *) v,      \
+                memory_order_release);                           \
+        rcu_check_sparse(p, __rcu);                              \
+        __r_a_p;                                                 \
+    })
+
+#endif /* __RCU_H__ */
