Fix: rcu_barrier(): uninitialized futex field

[urcu.git] / urcu-call-rcu-impl.h

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index f7f0f71eeeb28fec5152a8a81194c141035cd550..f0751f538b4b66c829d4d988b0d5102192ad889b 100644 (file)
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -64,6 +64,16 @@ struct call_rcu_data {
        struct cds_list_head list;
 } __attribute__((aligned(CAA_CACHE_LINE_SIZE)));
 
+struct call_rcu_completion {
+       int barrier_count;
+       int32_t futex;
+};
+
+struct call_rcu_completion_work {
+       struct rcu_head head;
+       struct call_rcu_completion *completion;
+};
+
 /*
  * List of all call_rcu_data structures to keep valgrind happy.
  * Protected by call_rcu_mutex.
@@ -236,6 +246,26 @@ static void call_rcu_wake_up(struct call_rcu_data *crdp)
        }
 }
 
+static void call_rcu_completion_wait(struct call_rcu_completion *completion)
+{
+       /* Read completion barrier count before read futex */
+       cmm_smp_mb();
+       if (uatomic_read(&completion->futex) == -1)
+               futex_async(&completion->futex, FUTEX_WAIT, -1,
+                     NULL, NULL, 0);
+}
+
+static void call_rcu_completion_wake_up(struct call_rcu_completion *completion)
+{
+       /* Write to completion barrier count before reading/writing futex */
+       cmm_smp_mb();
+       if (caa_unlikely(uatomic_read(&completion->futex) == -1)) {
+               uatomic_set(&completion->futex, 0);
+               futex_async(&completion->futex, FUTEX_WAKE, 1,
+                     NULL, NULL, 0);
+       }
+}
+
 /* This is the code run by each call_rcu thread. */
 
 static void *call_rcu_thread(void *arg)
@@ -278,6 +308,8 @@ static void *call_rcu_thread(void *arg)
                        uatomic_or(&crdp->flags, URCU_CALL_RCU_PAUSED);
                        while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_PAUSE) != 0)
                                poll(NULL, 0, 1);
+                       uatomic_and(&crdp->flags, ~URCU_CALL_RCU_PAUSED);
+                       cmm_smp_mb__after_uatomic_and();
                        rcu_register_thread();
                }
 
@@ -604,6 +636,17 @@ static void wake_call_rcu_thread(struct call_rcu_data *crdp)
                call_rcu_wake_up(crdp);
 }
 
+static void _call_rcu(struct rcu_head *head,
+                     void (*func)(struct rcu_head *head),
+                     struct call_rcu_data *crdp)
+{
+       cds_wfcq_node_init(&head->next);
+       head->func = func;
+       cds_wfcq_enqueue(&crdp->cbs_head, &crdp->cbs_tail, &head->next);
+       uatomic_inc(&crdp->qlen);
+       wake_call_rcu_thread(crdp);
+}
+
 /*
  * Schedule a function to be invoked after a following grace period.
  * This is the only function that must be called -- the others are
@@ -618,20 +661,15 @@ static void wake_call_rcu_thread(struct call_rcu_data *crdp)
  *
  * call_rcu must be called by registered RCU read-side threads.
  */
-
 void call_rcu(struct rcu_head *head,
              void (*func)(struct rcu_head *head))
 {
        struct call_rcu_data *crdp;
 
-       cds_wfcq_node_init(&head->next);
-       head->func = func;
        /* Holding rcu read-side lock across use of per-cpu crdp */
        rcu_read_lock();
        crdp = get_call_rcu_data();
-       cds_wfcq_enqueue(&crdp->cbs_head, &crdp->cbs_tail, &head->next);
-       uatomic_inc(&crdp->qlen);
-       wake_call_rcu_thread(crdp);
+       _call_rcu(head, func, crdp);
        rcu_read_unlock();
 }
 
@@ -730,6 +768,79 @@ void free_all_cpu_call_rcu_data(void)
        free(crdp);
 }
 
+static
+void _rcu_barrier_complete(struct rcu_head *head)
+{
+       struct call_rcu_completion_work *work;
+       struct call_rcu_completion *completion;
+
+       work = caa_container_of(head, struct call_rcu_completion_work, head);
+       completion = work->completion;
+       uatomic_dec(&completion->barrier_count);
+       call_rcu_completion_wake_up(completion);
+       free(work);
+}
+
+/*
+ * Wait for all in-flight call_rcu callbacks to complete execution.
+ */
+void rcu_barrier(void)
+{
+       struct call_rcu_data *crdp;
+       struct call_rcu_completion completion;
+       int count = 0;
+       int was_online;
+
+       /* Put in offline state in QSBR. */
+       was_online = rcu_read_ongoing();
+       if (was_online)
+               rcu_thread_offline();
+       /*
+        * Calling a rcu_barrier() within a RCU read-side critical
+        * section is an error.
+        */
+       if (rcu_read_ongoing()) {
+               static int warned = 0;
+
+               if (!warned) {
+                       fprintf(stderr, "[error] liburcu: rcu_barrier() called from within RCU read-side critical section.\n");
+               }
+               warned = 1;
+               goto online;
+       }
+
+       call_rcu_lock(&call_rcu_mutex);
+       cds_list_for_each_entry(crdp, &call_rcu_data_list, list)
+               count++;
+
+       completion.barrier_count = count;
+       completion.futex = 0;
+
+       cds_list_for_each_entry(crdp, &call_rcu_data_list, list) {
+               struct call_rcu_completion_work *work;
+
+               work = calloc(sizeof(*work), 1);
+               if (!work)
+                       urcu_die(errno);
+               work->completion = &completion;
+               _call_rcu(&work->head, _rcu_barrier_complete, crdp);
+       }
+       call_rcu_unlock(&call_rcu_mutex);
+
+       /* Wait for them */
+       for (;;) {
+               uatomic_dec(&completion.futex);
+               /* Decrement futex before reading barrier_count */
+               cmm_smp_mb();
+               if (!uatomic_read(&completion.barrier_count))
+                       break;
+               call_rcu_completion_wait(&completion);
+       }
+online:
+       if (was_online)
+               rcu_thread_online();
+}
+
 /*
  * Acquire the call_rcu_mutex in order to ensure that the child sees
  * all of the call_rcu() data structures in a consistent state. Ensure
@@ -764,6 +875,10 @@ void call_rcu_after_fork_parent(void)
 
        cds_list_for_each_entry(crdp, &call_rcu_data_list, list)
                uatomic_and(&crdp->flags, ~URCU_CALL_RCU_PAUSE);
+       cds_list_for_each_entry(crdp, &call_rcu_data_list, list) {
+               while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_PAUSED) != 0)
+                       poll(NULL, 0, 1);
+       }
        call_rcu_unlock(&call_rcu_mutex);
 }