+static
+void free_completion(struct urcu_ref *ref)
+{
+ struct call_rcu_completion *completion;
+
+ completion = caa_container_of(ref, struct call_rcu_completion, ref);
+ free(completion);
+}
+
+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;
+ if (!uatomic_sub_return(&completion->barrier_count, 1))
+ call_rcu_completion_wake_up(completion);
+ urcu_ref_put(&completion->ref, free_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;
+ }
+
+ completion = calloc(sizeof(*completion), 1);
+ if (!completion)
+ urcu_die(errno);
+
+ call_rcu_lock(&call_rcu_mutex);
+ cds_list_for_each_entry(crdp, &call_rcu_data_list, list)
+ count++;
+
+ /* Referenced by rcu_barrier() and each call_rcu thread. */
+ urcu_ref_set(&completion->ref, count + 1);
+ completion->barrier_count = count;
+
+ 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);
+ }
+
+ urcu_ref_put(&completion->ref, free_completion);
+
+online:
+ if (was_online)
+ rcu_thread_online();
+}
+