lf-queue: Stick to C11 atomics primitives

Author: jserv

lf-queue/atomics.h

@@ -1,8 +1,10 @@
 #include <assert.h>
 #include <errno.h>
+#include <stdatomic.h>
 #include <stdbool.h>
+#include <stdlib.h>
+#include <string.h>
 
-#include "atomics.h"
 #include "lfq.h"
 
 #define MAX_FREE 150
@@ -20,14 +22,14 @@ static bool in_hp(struct lfq_ctx *ctx, struct lfq_node *node)
 static void insert_pool(struct lfq_ctx *ctx, struct lfq_node *node)
 {
     atomic_store(&node->free_next, NULL);
-    struct lfq_node *old_tail = XCHG(&ctx->fpt, node); /* seq_cst */
+    struct lfq_node *old_tail = atomic_exchange(&ctx->fpt, node); /* seq_cst */
     atomic_store(&old_tail->free_next, node);
 }
 
 static void free_pool(struct lfq_ctx *ctx, bool freeall)
 {
     bool old = 0;
-    if (!CAS(&ctx->is_freeing, &old, 1))
+    if (!atomic_compare_exchange_strong(&ctx->is_freeing, &old, 1))
         return;
 
     for (int i = 0; i < MAX_FREE || freeall; i++) {
@@ -39,20 +41,20 @@ static void free_pool(struct lfq_ctx *ctx, bool freeall)
         free(p);
     }
     atomic_store(&ctx->is_freeing, false);
-    smp_mb();
+    atomic_thread_fence(memory_order_seq_cst);
 }
 
 static void safe_free(struct lfq_ctx *ctx, struct lfq_node *node)
 {
     if (atomic_load(&node->can_free) && !in_hp(ctx, node)) {
         /* free is not thread-safe */
         bool old = 0;
-        if (CAS(&ctx->is_freeing, &old, 1)) {
+        if (atomic_compare_exchange_strong(&ctx->is_freeing, &old, 1)) {
             /* poison the pointer to detect use-after-free */
             node->next = (void *) -1;
             free(node); /* we got the lock; actually free */
             atomic_store(&ctx->is_freeing, false);
-            smp_mb();
+            atomic_thread_fence(memory_order_seq_cst);
         } else /* we did not get the lock; only add to a freelist */
             insert_pool(ctx, node);
     } else
@@ -65,7 +67,7 @@ static int alloc_tid(struct lfq_ctx *ctx)
     for (int i = 0; i < ctx->MAX_HP_SIZE; i++) {
         if (ctx->tid_map[i] == 0) {
             int old = 0;
-            if (CAS(&ctx->tid_map[i], &old, 1))
+            if (atomic_compare_exchange_strong(&ctx->tid_map[i], &old, 1))
                 return i;
         }
     }
@@ -141,7 +143,7 @@ int lfq_enqueue(struct lfq_ctx *ctx, void *data)
         return -errno;
 
     insert_node->data = data;
-    struct lfq_node *old_tail = XCHG(&ctx->tail, insert_node);
+    struct lfq_node *old_tail = atomic_exchange(&ctx->tail, insert_node);
     /* We have claimed our spot in the insertion order by modifying tail.
      * we are the only inserting thread with a pointer to the old tail.
      *
@@ -162,13 +164,13 @@ void *lfq_dequeue_tid(struct lfq_ctx *ctx, int tid)
     /* HP[tid] is necessary for deallocation. */
     do {
     retry:
-        /* continue jumps to the bottom of the loop, and would attempt a CAS
-         * with uninitialized new_head.
+        /* continue jumps to the bottom of the loop, and would attempt a
+         * atomic_compare_exchange_strong with uninitialized new_head.
          */
         old_head = atomic_load(&ctx->head);
 
         atomic_store(&ctx->HP[tid], old_head);
-        mb();
+        atomic_thread_fence(memory_order_seq_cst);
 
         /* another thread freed it before seeing our HP[tid] store */
         if (old_head != atomic_load(&ctx->head))
@@ -179,7 +181,7 @@ void *lfq_dequeue_tid(struct lfq_ctx *ctx, int tid)
             atomic_store(&ctx->HP[tid], 0);
             return NULL; /* never remove the last node */
         }
-    } while (!CAS(&ctx->head, &old_head, new_head));
+    } while (!atomic_compare_exchange_strong(&ctx->head, &old_head, new_head));
 
     /* We have atomically advanced head, and we are the thread that won the race
      * to claim a node. We return the data from the *new* head. The list starts
@@ -191,7 +193,7 @@ void *lfq_dequeue_tid(struct lfq_ctx *ctx, int tid)
     atomic_store(&new_head->can_free, true);
 
     /* we need to avoid freeing until other readers are definitely not going to
-     * load its ->next in the CAS loop
+     * load its ->next in the atomic_compare_exchange_strong loop
      */
     safe_free(ctx, (struct lfq_node *) old_head);
 
@@ -208,4 +210,4 @@ void *lfq_dequeue(struct lfq_ctx *ctx)
     void *ret = lfq_dequeue_tid(ctx, tid);
     free_tid(ctx, tid);
     return ret;
-}
+}

lf-queue/lfq.c

@@ -1,14 +1,13 @@
 #define _GNU_SOURCE
 #include <inttypes.h>
 #include <pthread.h>
+#include <stdatomic.h>
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <time.h>
 #include <unistd.h>
 
-#include "atomics.h"
 #include "lfq.h"
 
 #ifndef MAX_PRODUCER
@@ -33,20 +32,20 @@ struct user_data {
 void *add_queue(void *data)
 {
     struct lfq_ctx *ctx = data;
-    int ret = 0;
     long added;
     for (added = 0; added < 500000; added++) {
         struct user_data *p = malloc(sizeof(struct user_data));
         p->data = SOME_ID;
+        int ret = 0;
         if ((ret = lfq_enqueue(ctx, p)) != 0) {
             printf("lfq_enqueue failed, reason:%s\n", strerror(-ret));
-            ATOMIC_ADD(&cnt_added, added);
-            ATOMIC_SUB(&cnt_producer, 1);
+            atomic_fetch_add(&cnt_added, added);
+            atomic_fetch_sub(&cnt_producer, 1);
             return 0;
         }
     }
-    ATOMIC_ADD(&cnt_added, added);
-    ATOMIC_SUB(&cnt_producer, 1);
+    atomic_fetch_add(&cnt_added, added);
+    atomic_fetch_sub(&cnt_producer, 1);
     printf("Producer thread [%lu] exited! Still %d running...\n",
            pthread_self(), atomic_load(&cnt_producer));
     return 0;
@@ -55,11 +54,10 @@ void *add_queue(void *data)
 void *remove_queue(void *data)
 {
     struct lfq_ctx *ctx = data;
-    struct user_data *p;
-    int tid = ATOMIC_ADD(&cnt_thread, 1);
+    int tid = atomic_fetch_add(&cnt_thread, 1);
     long deleted = 0;
     while (1) {
-        p = lfq_dequeue_tid(ctx, tid);
+        struct user_data *p = lfq_dequeue_tid(ctx, tid);
         if (p) {
             if (p->data != SOME_ID) {
                 printf("data wrong!!\n");
@@ -69,13 +67,12 @@ void *remove_queue(void *data)
             free(p);
             deleted++;
         } else {
-            if (ctx->count || atomic_load(&cnt_producer))
-                sched_yield(); /* queue is empty, release CPU slice */
-            else
-                break; /* queue is empty and no more producers */
+            if (!ctx->count && !atomic_load(&cnt_producer))
+                break;     /* queue is empty and no more producers */
+            sched_yield(); /* queue is empty, release CPU slice */
         }
     }
-    ATOMIC_ADD(&cnt_removed, deleted);
+    atomic_fetch_add(&cnt_removed, deleted);
 
     printf("Consumer thread [%lu] exited %d\n", pthread_self(), cnt_producer);
     return 0;
@@ -88,17 +85,17 @@ int main()
 
     pthread_t thread_cons[MAX_CONSUMER], thread_pros[MAX_PRODUCER];
 
-    ATOMIC_ADD(&cnt_producer, 1);
+    atomic_fetch_add(&cnt_producer, 1);
     for (int i = 0; i < MAX_CONSUMER; i++) {
         pthread_create(&thread_cons[i], NULL, remove_queue, (void *) &ctx);
     }
 
     for (int i = 0; i < MAX_PRODUCER; i++) {
-        ATOMIC_ADD(&cnt_producer, 1);
+        atomic_fetch_add(&cnt_producer, 1);
         pthread_create(&thread_pros[i], NULL, add_queue, (void *) &ctx);
     }
 
-    ATOMIC_SUB(&cnt_producer, 1);
+    atomic_fetch_sub(&cnt_producer, 1);
 
     for (int i = 0; i < MAX_PRODUCER; i++)
         pthread_join(thread_pros[i], NULL);