[lttng-dev] [RFC PATCH urcu] Implement rcu_barrier()
Mathieu Desnoyers
mathieu.desnoyers at efficios.com
Fri May 31 11:35:17 EDT 2013
Awaits for all in-flight call_rcu handlers to complete execution before
returning.
Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
---
diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index f7f0f71..fb3568f 100644
--- 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)
@@ -604,6 +634,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 +659,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 +766,89 @@ 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, work_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;
+
+ cds_list_for_each_entry(crdp, &call_rcu_data_list, list) {
+ struct call_rcu_completion_work *work;
+
+ work = calloc(sizeof(*work), 1);
+ if (!work) {
+ static int warned = 0;
+
+ if (!warned) {
+ fprintf(stderr, "[error] liburcu: unable to allocate memory for rcu_barrier()\n");
+ }
+ warned = 1;
+ break;
+ }
+ work->completion = &completion;
+ _call_rcu(&work->head, _rcu_barrier_complete, crdp);
+ work_count++;
+ }
+ call_rcu_unlock(&call_rcu_mutex);
+
+ if (work_count != count)
+ uatomic_sub(&completion.barrier_count, count - work_count);
+
+ /* 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
diff --git a/urcu-call-rcu.h b/urcu-call-rcu.h
index 997bb2f..30388c5 100644
--- a/urcu-call-rcu.h
+++ b/urcu-call-rcu.h
@@ -92,6 +92,8 @@ void call_rcu_before_fork(void);
void call_rcu_after_fork_parent(void);
void call_rcu_after_fork_child(void);
+void rcu_barrier(void);
+
#ifdef __cplusplus
}
#endif
diff --git a/urcu-flavor.h b/urcu-flavor.h
index c04f1db..5e7f292 100644
--- a/urcu-flavor.h
+++ b/urcu-flavor.h
@@ -41,6 +41,8 @@ struct rcu_flavor_struct {
void (*thread_online)(void);
void (*register_thread)(void);
void (*unregister_thread)(void);
+
+ void (*barrier)(void);
};
#define DEFINE_RCU_FLAVOR(x) \
@@ -56,6 +58,7 @@ const struct rcu_flavor_struct x = { \
.thread_online = rcu_thread_online, \
.register_thread = rcu_register_thread, \
.unregister_thread = rcu_unregister_thread,\
+ .barrier = rcu_barrier, \
}
extern const struct rcu_flavor_struct rcu_flavor;
diff --git a/urcu/map/urcu-bp.h b/urcu/map/urcu-bp.h
index 92863fe..67ba5c3 100644
--- a/urcu/map/urcu-bp.h
+++ b/urcu/map/urcu-bp.h
@@ -63,6 +63,7 @@
#define call_rcu_before_fork call_rcu_before_fork_bp
#define call_rcu_after_fork_parent call_rcu_after_fork_parent_bp
#define call_rcu_after_fork_child call_rcu_after_fork_child_bp
+#define rcu_barrier rcu_barrier_bp
#define defer_rcu defer_rcu_bp
#define rcu_defer_register_thread rcu_defer_register_thread_bp
diff --git a/urcu/map/urcu-qsbr.h b/urcu/map/urcu-qsbr.h
index e3261ff..b89dd24 100644
--- a/urcu/map/urcu-qsbr.h
+++ b/urcu/map/urcu-qsbr.h
@@ -65,6 +65,7 @@
#define call_rcu_before_fork call_rcu_before_fork_qsbr
#define call_rcu_after_fork_parent call_rcu_after_fork_parent_qsbr
#define call_rcu_after_fork_child call_rcu_after_fork_child_qsbr
+#define rcu_barrier rcu_barrier_qsbr
#define defer_rcu defer_rcu_qsbr
#define rcu_defer_register_thread rcu_defer_register_thread_qsbr
diff --git a/urcu/map/urcu.h b/urcu/map/urcu.h
index 77b3721..8f04caf 100644
--- a/urcu/map/urcu.h
+++ b/urcu/map/urcu.h
@@ -69,6 +69,7 @@
#define call_rcu_before_fork call_rcu_before_fork_memb
#define call_rcu_after_fork_parent call_rcu_after_fork_parent_memb
#define call_rcu_after_fork_child call_rcu_after_fork_child_memb
+#define rcu_barrier rcu_barrier_memb
#define defer_rcu defer_rcu_memb
#define rcu_defer_register_thread rcu_defer_register_thread_memb
@@ -116,6 +117,7 @@
#define call_rcu_before_fork call_rcu_before_fork_sig
#define call_rcu_after_fork_parent call_rcu_after_fork_parent_sig
#define call_rcu_after_fork_child call_rcu_after_fork_child_sig
+#define rcu_barrier rcu_barrier_sig
#define defer_rcu defer_rcu_sig
#define rcu_defer_register_thread rcu_defer_register_thread_sig
@@ -160,6 +162,7 @@
#define call_rcu_before_fork call_rcu_before_fork_mb
#define call_rcu_after_fork_parent call_rcu_after_fork_parent_mb
#define call_rcu_after_fork_child call_rcu_after_fork_child_mb
+#define rcu_barrier rcu_barrier_mb
#define defer_rcu defer_rcu_mb
#define rcu_defer_register_thread rcu_defer_register_thread_mb
--
Mathieu Desnoyers
EfficiOS Inc.
http://www.efficios.com
More information about the lttng-dev
mailing list