cmap: Fix data race reported by ThreadSanitizer

cmap/Makefile

@@ -1,5 +1,11 @@
+CFLAGS = -Wall -O2
+
+ifeq ("$(TSAN)", "1")
+    CFLAGS += -g -fsanitize=thread
+endif
+
 all:
-	gcc -Wall -O2 -o test-cmap \
+	gcc $(CFLAGS) -o test-cmap \
 		cmap.c  random.c  rcu.c  test-cmap.c \
 		-lpthread
 

cmap/cmap.c

@@ -35,8 +35,8 @@ static void cmap_insert__(struct cmap_impl *impl, struct cmap_node *node)
     size_t i = node->hash & impl->max;
     node->next = impl->arr[i].first;
     if (!impl->arr[i].first)
-        impl->utilization++;
-    impl->arr[i].first = node;
+        atomic_fetch_add(&impl->utilization, 1);
+    atomic_store(&impl->arr[i].first, node);
 }
 
 static void cmap_destroy_callback(void *args)
@@ -142,7 +142,7 @@ double cmap_utilization(const struct cmap *cmap)
 {
     struct rcu *impl_rcu = rcu_acquire(cmap->impl->p);
     struct cmap_impl *impl = rcu_get(impl_rcu, struct cmap_impl *);
-    double res = (double) impl->utilization / (impl->max + 1);
+    double res = (double) atomic_load(&impl->utilization) / (impl->max + 1);
     rcu_release(impl_rcu);
     return res;
 }
@@ -182,7 +182,7 @@ size_t cmap_remove(struct cmap *cmap, struct cmap_node *node)
     struct cmap_node **node_p = &cmap_entry->first;
     while (*node_p) {
         if (*node_p == node) {
-            *node_p = node->next;
+            atomic_store(node_p, node->next);
             count--;
             break;
         }
@@ -214,12 +214,13 @@ struct cmap_cursor cmap_find__(struct cmap_state state, uint32_t hash)
 
     struct cmap_cursor cursor = {
         .entry_idx = hash & impl->max,
-        .node = impl->arr[hash & impl->max].first,
+        .node = atomic_load(&impl->arr[hash & impl->max].first),
         .next = NULL,
         .accross_entries = false,
     };
     if (cursor.node)
         cursor.next = cursor.node->next;
+
     return cursor;
 }
 

cmap/locks.h

@@ -102,7 +102,7 @@ static inline void mutex_lock_at(struct mutex *mutex, const char *where)
         return;
     if (pthread_mutex_lock(&mutex->lock))
         abort_msg("pthread_mutex_lock fail");
-    mutex->where = where;
+    atomic_store_explicit(&mutex->where, where, memory_order_relaxed);
 }
 
 static inline void mutex_unlock(struct mutex *mutex)
@@ -111,7 +111,7 @@ static inline void mutex_unlock(struct mutex *mutex)
         return;
     if (pthread_mutex_unlock(&mutex->lock))
         abort_msg("pthread_mutex_unlock fail");
-    mutex->where = NULL;
+    atomic_store_explicit(&mutex->where, NULL, memory_order_relaxed);
 }
 
 static inline void cond_init(struct cond *cond)

cmap/random.c

@@ -6,7 +6,8 @@
 #include "random.h"
 #include "util.h"
 
-static uint32_t seed = 0;
+/* Maintain thread-independent random seed to prevent race. */
+static __thread uint32_t seed = 0;
 
 static uint32_t random_next(void);
 

cmap/test-cmap.c

@@ -17,8 +17,10 @@
 struct elem {
     struct cmap_node node;
     uint32_t value;
+    struct elem *next;
 };
 
+static struct elem *freelist = NULL;
 static size_t num_values;
 static uint32_t max_value;
 static uint32_t *values;
@@ -67,6 +69,13 @@ static void destroy_values()
     struct elem *elem;
     free(values);
 
+    elem = freelist;
+    while (elem) {
+        struct elem *next = elem->next;
+        free(elem);
+        elem = next;
+    }
+
     cmap_state = cmap_state_acquire(&cmap_values);
     MAP_FOREACH (elem, node, cmap_state) {
         cmap_remove(&cmap_values, &elem->node);
@@ -107,11 +116,11 @@ static void *update_cmap(void *args)
     struct elem *elem;
     struct cmap_state cmap_state;
 
-    while (running) {
+    while (atomic_load_explicit(&running, memory_order_relaxed)) {
         /* Insert */
         uint32_t value = random_uint32() + max_value + 1;
         insert_value(value);
-        inserts++;
+        atomic_fetch_add_explicit(&inserts, 1, memory_order_relaxed);
         wait();
 
         /* Remove */
@@ -120,8 +129,16 @@ static void *update_cmap(void *args)
         MAP_FOREACH_WITH_HASH (elem, node, hash, cmap_state) {
             if (elem->value > max_value) {
                 cmap_remove(&cmap_values, &elem->node);
-                free(elem);
-                removes++;
+                /* FIXME: If we free 'elem' directly, it may lead to use-after-free
+                 * when the reader thread obtains it. To deal with the issue, here
+                 * we just simply keep it in a list and release the memory at the end
+                 * of program.
+                 *
+                 * We should consider better strategy like reference counting or hazard pointer,
+                 * which allow us to free each chunk of memory at the correct time */
+                elem->next = freelist;
+                freelist = elem;
+                atomic_fetch_add_explicit(&removes, 1, memory_order_relaxed);
                 break;
             }
         }
@@ -134,18 +151,18 @@ static void *update_cmap(void *args)
 /* Constantly check whever values in cmap can be composed */
 static void *read_cmap(void *args)
 {
-    while (running) {
+    while (atomic_load_explicit(&running, memory_order_relaxed)) {
         uint32_t index = random_uint32() % num_values;
         if (!can_compose_value(values[index])) {
-            running = false;
-            error = true;
+            atomic_store_explicit(&running, false, memory_order_relaxed);
+            atomic_store_explicit(&error, true, memory_order_relaxed);
             break;
         }
         atomic_fetch_add(&checks, 1);
         wait();
         if (can_compose_value(values[index] + max_value + 1)) {
-            running = false;
-            error = true;
+            atomic_store_explicit(&running, false, memory_order_relaxed);
+            atomic_store_explicit(&error, true, memory_order_relaxed);
             break;
         }
         atomic_fetch_add(&checks, 1);
@@ -187,13 +204,15 @@ int main(int argc, char **argv)
         printf(
             "#checks: %u, #inserts: %u, #removes: %u, "
             "cmap elements: %u, utilization: %.2lf \n",
-            (uint32_t) current_checks, inserts, removes,
+            (uint32_t) current_checks,
+            atomic_load_explicit(&inserts, memory_order_relaxed),
+            atomic_load_explicit(&removes, memory_order_relaxed),
             (uint32_t) cmap_size(&cmap_values), cmap_utilization(&cmap_values));
         usleep(250e3);
     }
 
     /* Stop threads */
-    running = false;
+    atomic_store_explicit(&running, false, memory_order_relaxed);
     for (int i = 0; i <= readers; ++i)
         pthread_join(threads[i], NULL);