[ltt-dev] [PATCH userspace-rcu] Allow multiple flavors of RCU in same executable

[ltt-dev] [PATCH userspace-rcu] Allow multiple flavors of RCU in same executable

[Paul E. McKenney]

Hello!

This series of patches permits a single executable to use multiple
flavors of userspace RCU.  This is important, as it allows userspace
RCU to be used freely by multiple libraries without requiring that
the multiple authors of this library code to come to agreement about
which flavor of userspace RCU they will all use.

This is updated from the version posted on May 31st to add documentation
and to make the child atfork interface use unlocking rather than
re-initialization.

							Thanx, Paul

------------------------------------------------------------------------

 Makefile.am            |    4 
 README                 |   36 ++
 a/urcu-call-rcu.c      |  617 -------------------------------------------------
 b/API.txt              |  120 +++++++++
 b/Makefile.am          |    3 
 b/README               |    4 
 b/tests/Makefile.am    |   33 +-
 b/tests/rcutorture.h   |    1 
 b/tests/urcutorture.c  |   13 -
 b/urcu-bp-map.h        |   61 ++++
 b/urcu-bp.c            |    4 
 b/urcu-bp.h            |    8 
 b/urcu-call-rcu-impl.h |  615 ++++++++++++++++++++++++++++++++++++++++++++++++
 b/urcu-call-rcu.c      |   26 +-
 b/urcu-defer-impl.h    |  450 +++++++++++++++++++++++++++++++++++
 b/urcu-map.h           |  117 +++++++++
 b/urcu-qsbr-map.h      |   63 +++++
 b/urcu-qsbr.c          |   36 +-
 b/urcu-qsbr.h          |   18 -
 b/urcu.c               |    5 
 b/urcu.h               |   20 +
 tests/Makefile.am      |    2 
 urcu-bp-map.h          |    6 
 urcu-bp.c              |    1 
 urcu-bp.h              |    5 
 urcu-defer-static.h    |  108 --------
 urcu-defer.c           |  385 ------------------------------
 urcu-map.h             |   26 ++
 urcu-qsbr-map.h        |    6 
 urcu-qsbr.c            |    1 
 urcu-qsbr.h            |   11 
 urcu.c                 |    1 
 urcu.h                 |   13 -
 33 files changed, 1627 insertions(+), 1192 deletions(-)

[ltt-dev] [PATCH urcu 1/5] Provide pthread_atfork-friendly interfaces

[Paul E. McKenney]

Provides call_rcu_before_fork() and call_rcu_after_fork_parent() to
go with the existing call_rcu_after_fork_child().

Signed-off-by: Paul E. McKenney <paulmck at linux.vnet.ibm.com>
---
 README          |    4 ++++
 urcu-call-rcu.c |   26 +++++++++++++++++++++++++-
 2 files changed, 29 insertions(+), 1 deletions(-)

diff --git a/README b/README
index f7f0dec..56e98d7 100644
--- a/README
+++ b/README
@@ -204,3 +204,7 @@ Interaction with fork()
 	liburcu-bp, which is designed to handle fork() by calling
 	rcu_bp_before_fork, rcu_bp_after_fork_parent and
 	rcu_bp_after_fork_child.
+
+	Applications that use call_rcu() are required to invoke
+	call_rcu_after_fork_child() from the child process after a
+	successful fork() system call that is not followed by exec().
diff --git a/urcu-call-rcu.c b/urcu-call-rcu.c
index bb56dbb..9cc56c5 100644
--- a/urcu-call-rcu.c
+++ b/urcu-call-rcu.c
@@ -566,13 +566,37 @@ void free_all_cpu_call_rcu_data(void)
 }
 
 /*
+ * Acquire the call_rcu_mutex in order to ensure that the child sees
+ * all of the call_rcu() data structures in a consistent state.
+ * Suitable for pthread_atfork() and friends.
+ */
+void call_rcu_before_fork(void)
+{
+	call_rcu_lock(&call_rcu_mutex);
+}
+
+/*
+ * Clean up call_rcu data structures in the parent of a successful fork()
+ * that is not followed by exec() in the child.  Suitable for
+ * pthread_atfork() and friends.
+ */
+void call_rcu_after_fork_parent(void)
+{
+	call_rcu_unlock(&call_rcu_mutex);
+}
+
+/*
  * Clean up call_rcu data structures in the child of a successful fork()
- * that is not followed by exec().
+ * that is not followed by exec().  Suitable for pthread_atfork() and
+ * friends.
  */
 void call_rcu_after_fork_child(void)
 {
 	struct call_rcu_data *crdp;
 
+	/* Release the mutex. */
+	call_rcu_unlock(&call_rcu_mutex);
+
 	/*
 	 * Allocate a new default call_rcu_data structure in order
 	 * to get a working call_rcu thread to go with it.
-- 
1.7.3.2

[ltt-dev] [PATCH urcu 2/5] Map symbols to allow multiple RCU flavors to be used in one binary

[Paul E. McKenney]

Probably need similar mapping for rcu_defer().  Definitely need
backwards-compatibility mapping for programs compiled against
old versions of the library.

Signed-off-by: Paul E. McKenney <paulmck at linux.vnet.ibm.com>
---
 Makefile.am          |    3 +-
 tests/Makefile.am    |   33 ++--
 tests/rcutorture.h   |    1 -
 tests/urcutorture.c  |   13 +-
 urcu-bp-map.h        |   61 +++++
 urcu-bp.c            |    4 +
 urcu-bp.h            |    8 +-
 urcu-call-rcu-impl.h |  615 +++++++++++++++++++++++++++++++++++++++++++++++++
 urcu-call-rcu.c      |  617 --------------------------------------------------
 urcu-map.h           |  117 ++++++++++
 urcu-qsbr-map.h      |   63 +++++
 urcu-qsbr.c          |   36 ++--
 urcu-qsbr.h          |   18 +-
 urcu.c               |    5 +
 urcu.h               |   20 ++-
 15 files changed, 942 insertions(+), 672 deletions(-)
 create mode 100644 urcu-bp-map.h
 create mode 100644 urcu-call-rcu-impl.h
 delete mode 100644 urcu-call-rcu.c
 create mode 100644 urcu-map.h
 create mode 100644 urcu-qsbr-map.h

diff --git a/Makefile.am b/Makefile.am
index 776fa2b..4b70644 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -30,7 +30,7 @@ COMPAT+=compat_futex.c
 endif
 
 lib_LTLIBRARIES = liburcu.la liburcu-qsbr.la liburcu-mb.la liburcu-signal.la \
-		  liburcu-bp.la liburcu-defer.la liburcu-call.la \
+		  liburcu-bp.la liburcu-defer.la \
 		  libwfqueue.la libwfstack.la librculfqueue.la librculfstack.la
 
 liburcu_la_SOURCES = urcu.c urcu-pointer.c $(COMPAT)
@@ -45,7 +45,6 @@ liburcu_signal_la_CFLAGS = -DRCU_SIGNAL
 
 liburcu_bp_la_SOURCES = urcu-bp.c urcu-pointer.c $(COMPAT)
 
-liburcu_call_la_SOURCES = urcu-call-rcu.c $(COMPAT)
 liburcu_defer_la_SOURCES = urcu-defer.c $(COMPAT)
 
 libwfqueue_la_SOURCES = wfqueue.c $(COMPAT)
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 3c025a4..8dacb11 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -28,21 +28,20 @@ if COMPAT_FUTEX
 COMPAT+=$(top_srcdir)/compat_futex.c
 endif
 
-URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/urcu-call-rcu.c $(top_srcdir)/wfqueue.c $(COMPAT)
-URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/urcu-call-rcu.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
 # URCU_MB uses urcu.c but -DRCU_MB must be defined
-URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/urcu-call-rcu.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
 # URCU_SIGNAL uses urcu.c but -DRCU_SIGNAL must be defined
-URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/urcu-call-rcu.c $(top_srcdir)/wfqueue.c $(COMPAT)
-URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/urcu-call-rcu.c $(top_srcdir)/wfqueue.c $(COMPAT)
-URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-defer.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/urcu-call-rcu.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-defer.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
 
 URCU_LIB=$(top_builddir)/liburcu.la
 URCU_QSBR_LIB=$(top_builddir)/liburcu-qsbr.la
 URCU_MB_LIB=$(top_builddir)/liburcu-mb.la
 URCU_SIGNAL_LIB=$(top_builddir)/liburcu-signal.la
 URCU_BP_LIB=$(top_builddir)/liburcu-bp.la
-URCU_CALL_LIB=$(top_builddir)/liburcu-call.la
 WFQUEUE_LIB=$(top_builddir)/libwfqueue.la
 WFSTACK_LIB=$(top_builddir)/libwfstack.la
 RCULFQUEUE_LIB=$(top_builddir)/librculfqueue.la
@@ -95,24 +94,24 @@ test_perthreadlock_SOURCES = test_perthreadlock.c $(URCU_SIGNAL)
 
 
 rcutorture_urcu_SOURCES = urcutorture.c
-rcutorture_urcu_CFLAGS = -DTORTURE_URCU $(AM_CFLAGS)
-rcutorture_urcu_LDADD = $(URCU) $(URCU_CALL_LIB) $(WFQUEUE_LIB)
+rcutorture_urcu_CFLAGS = -DRCU_MEMBARRIER $(AM_CFLAGS)
+rcutorture_urcu_LDADD = $(URCU) $(WFQUEUE_LIB)
 
 rcutorture_urcu_mb_SOURCES = urcutorture.c
-rcutorture_urcu_mb_CFLAGS = -DTORTURE_URCU_MB $(AM_CFLAGS)
-rcutorture_urcu_mb_LDADD = $(URCU_MB_LIB) $(URCU_CALL_LIB) $(WFQUEUE_LIB)
+rcutorture_urcu_mb_CFLAGS = -DRCU_MB $(AM_CFLAGS)
+rcutorture_urcu_mb_LDADD = $(URCU_MB_LIB) $(WFQUEUE_LIB)
 
 rcutorture_qsbr_SOURCES = urcutorture.c
-rcutorture_qsbr_CFLAGS = -DTORTURE_QSBR $(AM_CFLAGS)
-rcutorture_qsbr_LDADD = $(URCU_QSBR_LIB) $(URCU_CALL_LIB) $(WFQUEUE_LIB)
+rcutorture_qsbr_CFLAGS = -DRCU_QSBR $(AM_CFLAGS)
+rcutorture_qsbr_LDADD = $(URCU_QSBR_LIB) $(WFQUEUE_LIB)
 
 rcutorture_urcu_signal_SOURCES = urcutorture.c
-rcutorture_urcu_signal_CFLAGS = -DTORTURE_URCU_SIGNAL $(AM_CFLAGS)
-rcutorture_urcu_signal_LDADD = $(URCU_SIGNAL_LIB) $(URCU_CALL_LIB) $(WFQUEUE_LIB)
+rcutorture_urcu_signal_CFLAGS = -DRCU_SIGNAL $(AM_CFLAGS)
+rcutorture_urcu_signal_LDADD = $(URCU_SIGNAL_LIB) $(WFQUEUE_LIB)
 
 rcutorture_urcu_bp_SOURCES = urcutorture.c
-rcutorture_urcu_bp_CFLAGS = -DTORTURE_URCU_BP $(AM_CFLAGS)
-rcutorture_urcu_bp_LDADD = $(URCU_BP_LIB) $(URCU_CALL_LIB) $(WFQUEUE_LIB)
+rcutorture_urcu_bp_CFLAGS = -DRCU_BP $(AM_CFLAGS)
+rcutorture_urcu_bp_LDADD = $(URCU_BP_LIB) $(WFQUEUE_LIB)
 
 test_mutex_SOURCES = test_mutex.c $(URCU)
 
diff --git a/tests/rcutorture.h b/tests/rcutorture.h
index 66fdd7f..aba74b0 100644
--- a/tests/rcutorture.h
+++ b/tests/rcutorture.h
@@ -66,7 +66,6 @@
  */
 
 #include <stdlib.h>
-#include "../urcu-call-rcu.h"
 
 DEFINE_PER_THREAD(long long, n_reads_pt);
 DEFINE_PER_THREAD(long long, n_updates_pt);
diff --git a/tests/urcutorture.c b/tests/urcutorture.c
index 63fa386..a098d87 100644
--- a/tests/urcutorture.c
+++ b/tests/urcutorture.c
@@ -8,22 +8,19 @@
 #include "api.h"
 #define _LGPL_SOURCE
 
-#ifdef TORTURE_RCU_MEMBARRIER
-#define RCU_MEMBARRIER
+#ifdef RCU_MEMBARRIER
 #include <urcu.h>
 #endif
-#ifdef TORTURE_URCU_SIGNAL
-#define RCU_SIGNAL
+#ifdef RCU_SIGNAL
 #include <urcu.h>
 #endif
-#ifdef TORTURE_URCU_MB
-#define RCU_MB
+#ifdef RCU_MB
 #include <urcu.h>
 #endif
-#ifdef TORTURE_QSBR
+#ifdef RCU_QSBR
 #include <urcu-qsbr.h>
 #endif
-#ifdef TORTURE_URCU_BP
+#ifdef RCU_BP
 #include <urcu-bp.h>
 #endif
 
diff --git a/urcu-bp-map.h b/urcu-bp-map.h
new file mode 100644
index 0000000..6321802
--- /dev/null
+++ b/urcu-bp-map.h
@@ -0,0 +1,61 @@
+#ifndef _URCU_BP_MAP_H
+#define _URCU_BP_MAP_H
+
+/*
+ * urcu-map.h
+ *
+ * Userspace RCU header -- name mapping to allow multiple flavors to be
+ * used in the same executable.
+ *
+ * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
+ * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
+ *
+ * LGPL-compatible code should include this header with :
+ *
+ * #define _LGPL_SOURCE
+ * #include <urcu.h>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ *
+ * IBM's contributions to this file may be relicensed under LGPLv2 or later.
+ */
+
+/* Mapping macros to allow multiple flavors in a single binary. */
+
+#define rcu_read_lock			rcu_read_lock_bp
+#define _rcu_read_lock			_rcu_read_lock_bp
+#define rcu_read_unlock			rcu_read_unlock_bp
+#define _rcu_read_unlock		_rcu_read_unlock_bp
+#define rcu_register_thread		rcu_register_thread_bp
+#define rcu_unregister_thread		rcu_unregister_thread_bp
+#define rcu_init			rcu_init_bp
+#define rcu_exit			rcu_exit_bp
+#define synchronize_rcu			synchronize_rcu_bp
+#define rcu_reader			rcu_reader_bp
+#define rcu_gp_ctr			rcu_gp_ctr_bp
+
+#define get_cpu_call_rcu_data		get_cpu_call_rcu_data_bp
+#define get_call_rcu_thread		get_call_rcu_thread_bp
+#define create_call_rcu_data		create_call_rcu_data_bp
+#define set_cpu_call_rcu_data		set_cpu_call_rcu_data_bp
+#define get_default_call_rcu_data	get_default_call_rcu_data_bp
+#define get_call_rcu_data		get_call_rcu_data_bp
+#define get_thread_call_rcu_data	get_thread_call_rcu_data_bp
+#define set_thread_call_rcu_data	set_thread_call_rcu_data_bp
+#define create_all_cpu_call_rcu_data	create_all_cpu_call_rcu_data_bp
+#define free_all_cpu_call_rcu_data	free_all_cpu_call_rcu_data_bp
+#define call_rcu			call_rcu_bp
+
+#endif /* _URCU_BP_MAP_H */
diff --git a/urcu-bp.c b/urcu-bp.c
index 04bb675..5474f9f 100644
--- a/urcu-bp.c
+++ b/urcu-bp.c
@@ -35,6 +35,8 @@
 #include <unistd.h>
 #include <sys/mman.h>
 
+#include "urcu-bp-map.h"
+
 #include "urcu-bp-static.h"
 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
 #include "urcu-bp.h"
@@ -375,3 +377,5 @@ void rcu_bp_after_fork_child(void)
 	ret = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
 	assert(!ret);
 }
+
+#include "urcu-call-rcu-impl.h"
diff --git a/urcu-bp.h b/urcu-bp.h
index d92fbd1..fdf885c 100644
--- a/urcu-bp.h
+++ b/urcu-bp.h
@@ -46,6 +46,8 @@
 extern "C" {
 #endif
 
+#include "urcu-bp-map.h"
+
 /*
  * Important !
  *
@@ -69,8 +71,8 @@ extern "C" {
  *
  * Mark the beginning and end of a read-side critical section.
  */
-#define rcu_read_lock()		_rcu_read_lock()
-#define rcu_read_unlock()	_rcu_read_unlock()
+#define rcu_read_lock_bp()		_rcu_read_lock()
+#define rcu_read_unlock_bp()		_rcu_read_unlock()
 
 #else /* !_LGPL_SOURCE */
 
@@ -115,4 +117,6 @@ static inline void rcu_init(void)
 }
 #endif
 
+#include "urcu-call-rcu.h"
+
 #endif /* _URCU_BP_H */
diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
new file mode 100644
index 0000000..9fa6aa6
--- /dev/null
+++ b/urcu-call-rcu-impl.h
@@ -0,0 +1,615 @@
+/*
+ * urcu-call-rcu.c
+ *
+ * Userspace RCU library - batch memory reclamation with kernel API
+ *
+ * Copyright (c) 2010 Paul E. McKenney <paulmck at linux.vnet.ibm.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include <stdio.h>
+#include <pthread.h>
+#include <signal.h>
+#include <assert.h>
+#include <stdlib.h>
+#include <string.h>
+#include <errno.h>
+#include <poll.h>
+#include <sys/time.h>
+#include <syscall.h>
+#include <unistd.h>
+
+#include "config.h"
+#include "urcu/wfqueue.h"
+#include "urcu-call-rcu.h"
+#include "urcu-pointer.h"
+#include "urcu/list.h"
+
+/* Data structure that identifies a call_rcu thread. */
+
+struct call_rcu_data {
+	struct cds_wfq_queue cbs;
+	unsigned long flags;
+	pthread_mutex_t mtx;
+	pthread_cond_t cond;
+	unsigned long qlen;
+	pthread_t tid;
+	struct cds_list_head list;
+} __attribute__((aligned(CAA_CACHE_LINE_SIZE)));
+
+/*
+ * List of all call_rcu_data structures to keep valgrind happy.
+ * Protected by call_rcu_mutex.
+ */
+
+CDS_LIST_HEAD(call_rcu_data_list);
+
+/* Link a thread using call_rcu() to its call_rcu thread. */
+
+static __thread struct call_rcu_data *thread_call_rcu_data;
+
+/* Guard call_rcu thread creation. */
+
+static pthread_mutex_t call_rcu_mutex = PTHREAD_MUTEX_INITIALIZER;
+
+/* If a given thread does not have its own call_rcu thread, this is default. */
+
+static struct call_rcu_data *default_call_rcu_data;
+
+/*
+ * If the sched_getcpu() and sysconf(_SC_NPROCESSORS_CONF) calls are
+ * available, then we can have call_rcu threads assigned to individual
+ * CPUs rather than only to specific threads.
+ */
+
+#if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF)
+
+/*
+ * Pointer to array of pointers to per-CPU call_rcu_data structures
+ * and # CPUs.
+ */
+
+static struct call_rcu_data **per_cpu_call_rcu_data;
+static long maxcpus;
+
+/* Allocate the array if it has not already been allocated. */
+
+static void alloc_cpu_call_rcu_data(void)
+{
+	struct call_rcu_data **p;
+	static int warned = 0;
+
+	if (maxcpus != 0)
+		return;
+	maxcpus = sysconf(_SC_NPROCESSORS_CONF);
+	if (maxcpus <= 0) {
+		return;
+	}
+	p = malloc(maxcpus * sizeof(*per_cpu_call_rcu_data));
+	if (p != NULL) {
+		memset(p, '\0', maxcpus * sizeof(*per_cpu_call_rcu_data));
+		per_cpu_call_rcu_data = p;
+	} else {
+		if (!warned) {
+			fprintf(stderr, "[error] liburcu: unable to allocate per-CPU pointer array\n");
+		}
+		warned = 1;
+	}
+}
+
+#else /* #if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF) */
+
+static const struct call_rcu_data **per_cpu_call_rcu_data = NULL;
+static const long maxcpus = -1;
+
+static void alloc_cpu_call_rcu_data(void)
+{
+}
+
+static int sched_getcpu(void)
+{
+	return -1;
+}
+
+#endif /* #else #if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF) */
+
+/* Acquire the specified pthread mutex. */
+
+static void call_rcu_lock(pthread_mutex_t *pmp)
+{
+	if (pthread_mutex_lock(pmp) != 0) {
+		perror("pthread_mutex_lock");
+		exit(-1);
+	}
+}
+
+/* Release the specified pthread mutex. */
+
+static void call_rcu_unlock(pthread_mutex_t *pmp)
+{
+	if (pthread_mutex_unlock(pmp) != 0) {
+		perror("pthread_mutex_unlock");
+		exit(-1);
+	}
+}
+
+/* This is the code run by each call_rcu thread. */
+
+static void *call_rcu_thread(void *arg)
+{
+	unsigned long cbcount;
+	struct cds_wfq_node *cbs;
+	struct cds_wfq_node **cbs_tail;
+	struct call_rcu_data *crdp = (struct call_rcu_data *)arg;
+	struct rcu_head *rhp;
+
+	thread_call_rcu_data = crdp;
+	for (;;) {
+		if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
+			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
+				poll(NULL, 0, 1);
+			_CMM_STORE_SHARED(crdp->cbs.head, NULL);
+			cbs_tail = (struct cds_wfq_node **)
+				uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
+			synchronize_rcu();
+			cbcount = 0;
+			do {
+				while (cbs->next == NULL &&
+				       &cbs->next != cbs_tail)
+				       	poll(NULL, 0, 1);
+				if (cbs == &crdp->cbs.dummy) {
+					cbs = cbs->next;
+					continue;
+				}
+				rhp = (struct rcu_head *)cbs;
+				cbs = cbs->next;
+				rhp->func(rhp);
+				cbcount++;
+			} while (cbs != NULL);
+			uatomic_sub(&crdp->qlen, cbcount);
+		}
+		if (crdp->flags & URCU_CALL_RCU_STOP)
+			break;
+		if (crdp->flags & URCU_CALL_RCU_RT)
+			poll(NULL, 0, 10);
+		else {
+			call_rcu_lock(&crdp->mtx);
+			_CMM_STORE_SHARED(crdp->flags,
+				     crdp->flags & ~URCU_CALL_RCU_RUNNING);
+			if (&crdp->cbs.head ==
+			    _CMM_LOAD_SHARED(crdp->cbs.tail) &&
+			    pthread_cond_wait(&crdp->cond, &crdp->mtx) != 0) {
+				perror("pthread_cond_wait");
+				exit(-1);
+			}
+			_CMM_STORE_SHARED(crdp->flags,
+				     crdp->flags | URCU_CALL_RCU_RUNNING);
+			poll(NULL, 0, 10);
+			call_rcu_unlock(&crdp->mtx);
+		}
+	}
+	call_rcu_lock(&crdp->mtx);
+	crdp->flags |= URCU_CALL_RCU_STOPPED;
+	call_rcu_unlock(&crdp->mtx);
+	return NULL;
+}
+
+/*
+ * Create both a call_rcu thread and the corresponding call_rcu_data
+ * structure, linking the structure in as specified.  Caller must hold
+ * call_rcu_mutex.
+ */
+
+static void call_rcu_data_init(struct call_rcu_data **crdpp,
+			       unsigned long flags)
+{
+	struct call_rcu_data *crdp;
+
+	crdp = malloc(sizeof(*crdp));
+	if (crdp == NULL) {
+		fprintf(stderr, "Out of memory.\n");
+		exit(-1);
+	}
+	memset(crdp, '\0', sizeof(*crdp));
+	cds_wfq_init(&crdp->cbs);
+	crdp->qlen = 0;
+	if (pthread_mutex_init(&crdp->mtx, NULL) != 0) {
+		perror("pthread_mutex_init");
+		exit(-1);
+	}
+	if (pthread_cond_init(&crdp->cond, NULL) != 0) {
+		perror("pthread_cond_init");
+		exit(-1);
+	}
+	crdp->flags = flags | URCU_CALL_RCU_RUNNING;
+	cds_list_add(&crdp->list, &call_rcu_data_list);
+	cmm_smp_mb();  /* Structure initialized before pointer is planted. */
+	*crdpp = crdp;
+	if (pthread_create(&crdp->tid, NULL, call_rcu_thread, crdp) != 0) {
+		perror("pthread_create");
+		exit(-1);
+	}
+}
+
+/*
+ * Return a pointer to the call_rcu_data structure for the specified
+ * CPU, returning NULL if there is none.  We cannot automatically
+ * created it because the platform we are running on might not define
+ * sched_getcpu().
+ */
+
+struct call_rcu_data *get_cpu_call_rcu_data(int cpu)
+{
+	static int warned = 0;
+
+	if (per_cpu_call_rcu_data == NULL)
+		return NULL;
+	if (!warned && maxcpus > 0 && (cpu < 0 || maxcpus <= cpu)) {
+		fprintf(stderr, "[error] liburcu: get CPU # out of range\n");
+		warned = 1;
+	}
+	if (cpu < 0 || maxcpus <= cpu)
+		return NULL;
+	return per_cpu_call_rcu_data[cpu];
+}
+
+/*
+ * Return the tid corresponding to the call_rcu thread whose
+ * call_rcu_data structure is specified.
+ */
+
+pthread_t get_call_rcu_thread(struct call_rcu_data *crdp)
+{
+	return crdp->tid;
+}
+
+/*
+ * Create a call_rcu_data structure (with thread) and return a pointer.
+ */
+
+static struct call_rcu_data *__create_call_rcu_data(unsigned long flags)
+{
+	struct call_rcu_data *crdp;
+
+	call_rcu_data_init(&crdp, flags);
+	return crdp;
+}
+
+struct call_rcu_data *create_call_rcu_data(unsigned long flags)
+{
+	struct call_rcu_data *crdp;
+
+	call_rcu_lock(&call_rcu_mutex);
+	crdp = __create_call_rcu_data(flags);
+	call_rcu_unlock(&call_rcu_mutex);
+	return crdp;
+}
+
+/*
+ * Set the specified CPU to use the specified call_rcu_data structure.
+ *
+ * Use NULL to remove a CPU's call_rcu_data structure, but it is
+ * the caller's responsibility to dispose of the removed structure.
+ * Use get_cpu_call_rcu_data() to obtain a pointer to the old structure
+ * (prior to NULLing it out, of course).
+ */
+
+int set_cpu_call_rcu_data(int cpu, struct call_rcu_data *crdp)
+{
+	int warned = 0;
+
+	call_rcu_lock(&call_rcu_mutex);
+	if (cpu < 0 || maxcpus <= cpu) {
+		if (!warned) {
+			fprintf(stderr, "[error] liburcu: set CPU # out of range\n");
+			warned = 1;
+		}
+		call_rcu_unlock(&call_rcu_mutex);
+		errno = EINVAL;
+		return -EINVAL;
+	}
+	alloc_cpu_call_rcu_data();
+	call_rcu_unlock(&call_rcu_mutex);
+	if (per_cpu_call_rcu_data == NULL) {
+		errno = ENOMEM;
+		return -ENOMEM;
+	}
+	per_cpu_call_rcu_data[cpu] = crdp;
+	return 0;
+}
+
+/*
+ * Return a pointer to the default call_rcu_data structure, creating
+ * one if need be.  Because we never free call_rcu_data structures,
+ * we don't need to be in an RCU read-side critical section.
+ */
+
+struct call_rcu_data *get_default_call_rcu_data(void)
+{
+	if (default_call_rcu_data != NULL)
+		return rcu_dereference(default_call_rcu_data);
+	call_rcu_lock(&call_rcu_mutex);
+	if (default_call_rcu_data != NULL) {
+		call_rcu_unlock(&call_rcu_mutex);
+		return default_call_rcu_data;
+	}
+	call_rcu_data_init(&default_call_rcu_data, 0);
+	call_rcu_unlock(&call_rcu_mutex);
+	return default_call_rcu_data;
+}
+
+/*
+ * Return the call_rcu_data structure that applies to the currently
+ * running thread.  Any call_rcu_data structure assigned specifically
+ * to this thread has first priority, followed by any call_rcu_data
+ * structure assigned to the CPU on which the thread is running,
+ * followed by the default call_rcu_data structure.  If there is not
+ * yet a default call_rcu_data structure, one will be created.
+ */
+struct call_rcu_data *get_call_rcu_data(void)
+{
+	int curcpu;
+	static int warned = 0;
+
+	if (thread_call_rcu_data != NULL)
+		return thread_call_rcu_data;
+	if (maxcpus <= 0)
+		return get_default_call_rcu_data();
+	curcpu = sched_getcpu();
+	if (!warned && (curcpu < 0 || maxcpus <= curcpu)) {
+		fprintf(stderr, "[error] liburcu: gcrd CPU # out of range\n");
+		warned = 1;
+	}
+	if (curcpu >= 0 && maxcpus > curcpu &&
+	    per_cpu_call_rcu_data != NULL &&
+	    per_cpu_call_rcu_data[curcpu] != NULL)
+	    	return per_cpu_call_rcu_data[curcpu];
+	return get_default_call_rcu_data();
+}
+
+/*
+ * Return a pointer to this task's call_rcu_data if there is one.
+ */
+
+struct call_rcu_data *get_thread_call_rcu_data(void)
+{
+	return thread_call_rcu_data;
+}
+
+/*
+ * Set this task's call_rcu_data structure as specified, regardless
+ * of whether or not this task already had one.  (This allows switching
+ * to and from real-time call_rcu threads, for example.)
+ *
+ * Use NULL to remove a thread's call_rcu_data structure, but it is
+ * the caller's responsibility to dispose of the removed structure.
+ * Use get_thread_call_rcu_data() to obtain a pointer to the old structure
+ * (prior to NULLing it out, of course).
+ */
+
+void set_thread_call_rcu_data(struct call_rcu_data *crdp)
+{
+	thread_call_rcu_data = crdp;
+}
+
+/*
+ * Create a separate call_rcu thread for each CPU.  This does not
+ * replace a pre-existing call_rcu thread -- use the set_cpu_call_rcu_data()
+ * function if you want that behavior.
+ */
+
+int create_all_cpu_call_rcu_data(unsigned long flags)
+{
+	int i;
+	struct call_rcu_data *crdp;
+	int ret;
+
+	call_rcu_lock(&call_rcu_mutex);
+	alloc_cpu_call_rcu_data();
+	call_rcu_unlock(&call_rcu_mutex);
+	if (maxcpus <= 0) {
+		errno = EINVAL;
+		return -EINVAL;
+	}
+	if (per_cpu_call_rcu_data == NULL) {
+		errno = ENOMEM;
+		return -ENOMEM;
+	}
+	for (i = 0; i < maxcpus; i++) {
+		call_rcu_lock(&call_rcu_mutex);
+		if (get_cpu_call_rcu_data(i)) {
+			call_rcu_unlock(&call_rcu_mutex);
+			continue;
+		}
+		crdp = __create_call_rcu_data(flags);
+		if (crdp == NULL) {
+			call_rcu_unlock(&call_rcu_mutex);
+			errno = ENOMEM;
+			return -ENOMEM;
+		}
+		call_rcu_unlock(&call_rcu_mutex);
+		if ((ret = set_cpu_call_rcu_data(i, crdp)) != 0) {
+			/* FIXME: Leaks crdp for now. */
+			return ret; /* Can happen on race. */
+		}
+	}
+	return 0;
+}
+
+/*
+ * Wake up the call_rcu thread corresponding to the specified
+ * call_rcu_data structure.
+ */
+static void wake_call_rcu_thread(struct call_rcu_data *crdp)
+{
+	if (!(_CMM_LOAD_SHARED(crdp->flags) & URCU_CALL_RCU_RT)) {
+		call_rcu_lock(&crdp->mtx);
+		if (!(_CMM_LOAD_SHARED(crdp->flags) & URCU_CALL_RCU_RUNNING)) {
+			if (pthread_cond_signal(&crdp->cond) != 0) {
+				perror("pthread_cond_signal");
+				exit(-1);
+			}
+		}
+		call_rcu_unlock(&crdp->mtx);
+	}
+}
+
+/*
+ * Schedule a function to be invoked after a following grace period.
+ * This is the only function that must be called -- the others are
+ * only present to allow applications to tune their use of RCU for
+ * maximum performance.
+ *
+ * Note that unless a call_rcu thread has not already been created,
+ * the first invocation of call_rcu() will create one.  So, if you
+ * need the first invocation of call_rcu() to be fast, make sure
+ * to create a call_rcu thread first.  One way to accomplish this is
+ * "get_call_rcu_data();", and another is create_all_cpu_call_rcu_data().
+ */
+
+void call_rcu(struct rcu_head *head,
+	      void (*func)(struct rcu_head *head))
+{
+	struct call_rcu_data *crdp;
+
+	cds_wfq_node_init(&head->next);
+	head->func = func;
+	crdp = get_call_rcu_data();
+	cds_wfq_enqueue(&crdp->cbs, &head->next);
+	uatomic_inc(&crdp->qlen);
+	wake_call_rcu_thread(crdp);
+}
+
+/*
+ * Free up the specified call_rcu_data structure, terminating the
+ * associated call_rcu thread.  The caller must have previously
+ * removed the call_rcu_data structure from per-thread or per-CPU
+ * usage.  For example, set_cpu_call_rcu_data(cpu, NULL) for per-CPU
+ * call_rcu_data structures or set_thread_call_rcu_data(NULL) for
+ * per-thread call_rcu_data structures.
+ *
+ * We silently refuse to free up the default call_rcu_data structure
+ * because that is where we put any leftover callbacks.  Note that
+ * the possibility of self-spawning callbacks makes it impossible
+ * to execute all the callbacks in finite time without putting any
+ * newly spawned callbacks somewhere else.  The "somewhere else" of
+ * last resort is the default call_rcu_data structure.
+ *
+ * We also silently refuse to free NULL pointers.  This simplifies
+ * the calling code.
+ */
+void call_rcu_data_free(struct call_rcu_data *crdp)
+{
+	struct cds_wfq_node *cbs;
+	struct cds_wfq_node **cbs_tail;
+	struct cds_wfq_node **cbs_endprev;
+
+	if (crdp == NULL || crdp == default_call_rcu_data) {
+		return;
+	}
+	if ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0) {
+		call_rcu_lock(&crdp->mtx);
+		crdp->flags |= URCU_CALL_RCU_STOP;
+		call_rcu_unlock(&crdp->mtx);
+		wake_call_rcu_thread(crdp);
+		while ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0)
+			poll(NULL, 0, 1);
+	}
+	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
+		while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
+			poll(NULL, 0, 1);
+		_CMM_STORE_SHARED(crdp->cbs.head, NULL);
+		cbs_tail = (struct cds_wfq_node **)
+			uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
+		cbs_endprev = (struct cds_wfq_node **)
+			uatomic_xchg(&default_call_rcu_data, cbs_tail);
+		*cbs_endprev = cbs;
+		uatomic_add(&default_call_rcu_data->qlen,
+			    uatomic_read(&crdp->qlen));
+		cds_list_del(&crdp->list);
+		free(crdp);
+	}
+}
+
+/*
+ * Clean up all the per-CPU call_rcu threads.
+ */
+void free_all_cpu_call_rcu_data(void)
+{
+	int cpu;
+	struct call_rcu_data *crdp;
+
+	if (maxcpus <= 0)
+		return;
+	for (cpu = 0; cpu < maxcpus; cpu++) {
+		crdp = get_cpu_call_rcu_data(cpu);
+		if (crdp == NULL)
+			continue;
+		set_cpu_call_rcu_data(cpu, NULL);
+		call_rcu_data_free(crdp);
+	}
+}
+
+/*
+ * Acquire the call_rcu_mutex in order to ensure that the child sees
+ * all of the call_rcu() data structures in a consistent state.
+ * Suitable for pthread_atfork() and friends.
+ */
+void call_rcu_before_fork(void)
+{
+	call_rcu_lock(&call_rcu_mutex);
+}
+
+/*
+ * Clean up call_rcu data structures in the parent of a successful fork()
+ * that is not followed by exec() in the child.  Suitable for
+ * pthread_atfork() and friends.
+ */
+void call_rcu_after_fork_parent(void)
+{
+	call_rcu_unlock(&call_rcu_mutex);
+}
+
+/*
+ * Clean up call_rcu data structures in the child of a successful fork()
+ * that is not followed by exec().  Suitable for pthread_atfork() and
+ * friends.
+ */
+void call_rcu_after_fork_child(void)
+{
+	struct call_rcu_data *crdp;
+
+	/* Release the mutex. */
+	call_rcu_unlock(&call_rcu_mutex);
+
+	/*
+	 * Allocate a new default call_rcu_data structure in order
+	 * to get a working call_rcu thread to go with it.
+	 */
+	default_call_rcu_data = NULL;
+	(void)get_default_call_rcu_data();
+
+	/* Dispose of all of the rest of the call_rcu_data structures. */
+	while (call_rcu_data_list.next != call_rcu_data_list.prev) {
+		crdp = cds_list_entry(call_rcu_data_list.prev,
+				      struct call_rcu_data, list);
+		if (crdp == default_call_rcu_data)
+			crdp = cds_list_entry(crdp->list.prev,
+					      struct call_rcu_data, list);
+		crdp->flags = URCU_CALL_RCU_STOPPED;
+		call_rcu_data_free(crdp);
+	}
+}
diff --git a/urcu-call-rcu.c b/urcu-call-rcu.c
deleted file mode 100644
index 9cc56c5..0000000
--- a/urcu-call-rcu.c
+++ /dev/null
@@ -1,617 +0,0 @@
-/*
- * urcu-call-rcu.c
- *
- * Userspace RCU library - batch memory reclamation with kernel API
- *
- * Copyright (c) 2010 Paul E. McKenney <paulmck at linux.vnet.ibm.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
- */
-
-#include <stdio.h>
-#include <pthread.h>
-#include <signal.h>
-#include <assert.h>
-#include <stdlib.h>
-#include <string.h>
-#include <errno.h>
-#include <poll.h>
-#include <sys/time.h>
-#include <syscall.h>
-#include <unistd.h>
-
-#include "config.h"
-#include "urcu/wfqueue.h"
-#include "urcu-call-rcu.h"
-#include "urcu-pointer.h"
-#include "urcu/list.h"
-
-/* Data structure that identifies a call_rcu thread. */
-
-struct call_rcu_data {
-	struct cds_wfq_queue cbs;
-	unsigned long flags;
-	pthread_mutex_t mtx;
-	pthread_cond_t cond;
-	unsigned long qlen;
-	pthread_t tid;
-	struct cds_list_head list;
-} __attribute__((aligned(CAA_CACHE_LINE_SIZE)));
-
-/*
- * List of all call_rcu_data structures to keep valgrind happy.
- * Protected by call_rcu_mutex.
- */
-
-CDS_LIST_HEAD(call_rcu_data_list);
-
-/* Link a thread using call_rcu() to its call_rcu thread. */
-
-static __thread struct call_rcu_data *thread_call_rcu_data;
-
-/* Guard call_rcu thread creation. */
-
-static pthread_mutex_t call_rcu_mutex = PTHREAD_MUTEX_INITIALIZER;
-
-/* If a given thread does not have its own call_rcu thread, this is default. */
-
-static struct call_rcu_data *default_call_rcu_data;
-
-extern void synchronize_rcu(void);
-
-/*
- * If the sched_getcpu() and sysconf(_SC_NPROCESSORS_CONF) calls are
- * available, then we can have call_rcu threads assigned to individual
- * CPUs rather than only to specific threads.
- */
-
-#if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF)
-
-/*
- * Pointer to array of pointers to per-CPU call_rcu_data structures
- * and # CPUs.
- */
-
-static struct call_rcu_data **per_cpu_call_rcu_data;
-static long maxcpus;
-
-/* Allocate the array if it has not already been allocated. */
-
-static void alloc_cpu_call_rcu_data(void)
-{
-	struct call_rcu_data **p;
-	static int warned = 0;
-
-	if (maxcpus != 0)
-		return;
-	maxcpus = sysconf(_SC_NPROCESSORS_CONF);
-	if (maxcpus <= 0) {
-		return;
-	}
-	p = malloc(maxcpus * sizeof(*per_cpu_call_rcu_data));
-	if (p != NULL) {
-		memset(p, '\0', maxcpus * sizeof(*per_cpu_call_rcu_data));
-		per_cpu_call_rcu_data = p;
-	} else {
-		if (!warned) {
-			fprintf(stderr, "[error] liburcu: unable to allocate per-CPU pointer array\n");
-		}
-		warned = 1;
-	}
-}
-
-#else /* #if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF) */
-
-static const struct call_rcu_data **per_cpu_call_rcu_data = NULL;
-static const long maxcpus = -1;
-
-static void alloc_cpu_call_rcu_data(void)
-{
-}
-
-static int sched_getcpu(void)
-{
-	return -1;
-}
-
-#endif /* #else #if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF) */
-
-/* Acquire the specified pthread mutex. */
-
-static void call_rcu_lock(pthread_mutex_t *pmp)
-{
-	if (pthread_mutex_lock(pmp) != 0) {
-		perror("pthread_mutex_lock");
-		exit(-1);
-	}
-}
-
-/* Release the specified pthread mutex. */
-
-static void call_rcu_unlock(pthread_mutex_t *pmp)
-{
-	if (pthread_mutex_unlock(pmp) != 0) {
-		perror("pthread_mutex_unlock");
-		exit(-1);
-	}
-}
-
-/* This is the code run by each call_rcu thread. */
-
-static void *call_rcu_thread(void *arg)
-{
-	unsigned long cbcount;
-	struct cds_wfq_node *cbs;
-	struct cds_wfq_node **cbs_tail;
-	struct call_rcu_data *crdp = (struct call_rcu_data *)arg;
-	struct rcu_head *rhp;
-
-	thread_call_rcu_data = crdp;
-	for (;;) {
-		if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
-			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
-				poll(NULL, 0, 1);
-			_CMM_STORE_SHARED(crdp->cbs.head, NULL);
-			cbs_tail = (struct cds_wfq_node **)
-				uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
-			synchronize_rcu();
-			cbcount = 0;
-			do {
-				while (cbs->next == NULL &&
-				       &cbs->next != cbs_tail)
-				       	poll(NULL, 0, 1);
-				if (cbs == &crdp->cbs.dummy) {
-					cbs = cbs->next;
-					continue;
-				}
-				rhp = (struct rcu_head *)cbs;
-				cbs = cbs->next;
-				rhp->func(rhp);
-				cbcount++;
-			} while (cbs != NULL);
-			uatomic_sub(&crdp->qlen, cbcount);
-		}
-		if (crdp->flags & URCU_CALL_RCU_STOP)
-			break;
-		if (crdp->flags & URCU_CALL_RCU_RT)
-			poll(NULL, 0, 10);
-		else {
-			call_rcu_lock(&crdp->mtx);
-			_CMM_STORE_SHARED(crdp->flags,
-				     crdp->flags & ~URCU_CALL_RCU_RUNNING);
-			if (&crdp->cbs.head ==
-			    _CMM_LOAD_SHARED(crdp->cbs.tail) &&
-			    pthread_cond_wait(&crdp->cond, &crdp->mtx) != 0) {
-				perror("pthread_cond_wait");
-				exit(-1);
-			}
-			_CMM_STORE_SHARED(crdp->flags,
-				     crdp->flags | URCU_CALL_RCU_RUNNING);
-			poll(NULL, 0, 10);
-			call_rcu_unlock(&crdp->mtx);
-		}
-	}
-	call_rcu_lock(&crdp->mtx);
-	crdp->flags |= URCU_CALL_RCU_STOPPED;
-	call_rcu_unlock(&crdp->mtx);
-	return NULL;
-}
-
-/*
- * Create both a call_rcu thread and the corresponding call_rcu_data
- * structure, linking the structure in as specified.  Caller must hold
- * call_rcu_mutex.
- */
-
-static void call_rcu_data_init(struct call_rcu_data **crdpp,
-			       unsigned long flags)
-{
-	struct call_rcu_data *crdp;
-
-	crdp = malloc(sizeof(*crdp));
-	if (crdp == NULL) {
-		fprintf(stderr, "Out of memory.\n");
-		exit(-1);
-	}
-	memset(crdp, '\0', sizeof(*crdp));
-	cds_wfq_init(&crdp->cbs);
-	crdp->qlen = 0;
-	if (pthread_mutex_init(&crdp->mtx, NULL) != 0) {
-		perror("pthread_mutex_init");
-		exit(-1);
-	}
-	if (pthread_cond_init(&crdp->cond, NULL) != 0) {
-		perror("pthread_cond_init");
-		exit(-1);
-	}
-	crdp->flags = flags | URCU_CALL_RCU_RUNNING;
-	cds_list_add(&crdp->list, &call_rcu_data_list);
-	cmm_smp_mb();  /* Structure initialized before pointer is planted. */
-	*crdpp = crdp;
-	if (pthread_create(&crdp->tid, NULL, call_rcu_thread, crdp) != 0) {
-		perror("pthread_create");
-		exit(-1);
-	}
-}
-
-/*
- * Return a pointer to the call_rcu_data structure for the specified
- * CPU, returning NULL if there is none.  We cannot automatically
- * created it because the platform we are running on might not define
- * sched_getcpu().
- */
-
-struct call_rcu_data *get_cpu_call_rcu_data(int cpu)
-{
-	static int warned = 0;
-
-	if (per_cpu_call_rcu_data == NULL)
-		return NULL;
-	if (!warned && maxcpus > 0 && (cpu < 0 || maxcpus <= cpu)) {
-		fprintf(stderr, "[error] liburcu: get CPU # out of range\n");
-		warned = 1;
-	}
-	if (cpu < 0 || maxcpus <= cpu)
-		return NULL;
-	return per_cpu_call_rcu_data[cpu];
-}
-
-/*
- * Return the tid corresponding to the call_rcu thread whose
- * call_rcu_data structure is specified.
- */
-
-pthread_t get_call_rcu_thread(struct call_rcu_data *crdp)
-{
-	return crdp->tid;
-}
-
-/*
- * Create a call_rcu_data structure (with thread) and return a pointer.
- */
-
-static struct call_rcu_data *__create_call_rcu_data(unsigned long flags)
-{
-	struct call_rcu_data *crdp;
-
-	call_rcu_data_init(&crdp, flags);
-	return crdp;
-}
-
-struct call_rcu_data *create_call_rcu_data(unsigned long flags)
-{
-	struct call_rcu_data *crdp;
-
-	call_rcu_lock(&call_rcu_mutex);
-	crdp = __create_call_rcu_data(flags);
-	call_rcu_unlock(&call_rcu_mutex);
-	return crdp;
-}
-
-/*
- * Set the specified CPU to use the specified call_rcu_data structure.
- *
- * Use NULL to remove a CPU's call_rcu_data structure, but it is
- * the caller's responsibility to dispose of the removed structure.
- * Use get_cpu_call_rcu_data() to obtain a pointer to the old structure
- * (prior to NULLing it out, of course).
- */
-
-int set_cpu_call_rcu_data(int cpu, struct call_rcu_data *crdp)
-{
-	int warned = 0;
-
-	call_rcu_lock(&call_rcu_mutex);
-	if (cpu < 0 || maxcpus <= cpu) {
-		if (!warned) {
-			fprintf(stderr, "[error] liburcu: set CPU # out of range\n");
-			warned = 1;
-		}
-		call_rcu_unlock(&call_rcu_mutex);
-		errno = EINVAL;
-		return -EINVAL;
-	}
-	alloc_cpu_call_rcu_data();
-	call_rcu_unlock(&call_rcu_mutex);
-	if (per_cpu_call_rcu_data == NULL) {
-		errno = ENOMEM;
-		return -ENOMEM;
-	}
-	per_cpu_call_rcu_data[cpu] = crdp;
-	return 0;
-}
-
-/*
- * Return a pointer to the default call_rcu_data structure, creating
- * one if need be.  Because we never free call_rcu_data structures,
- * we don't need to be in an RCU read-side critical section.
- */
-
-struct call_rcu_data *get_default_call_rcu_data(void)
-{
-	if (default_call_rcu_data != NULL)
-		return rcu_dereference(default_call_rcu_data);
-	call_rcu_lock(&call_rcu_mutex);
-	if (default_call_rcu_data != NULL) {
-		call_rcu_unlock(&call_rcu_mutex);
-		return default_call_rcu_data;
-	}
-	call_rcu_data_init(&default_call_rcu_data, 0);
-	call_rcu_unlock(&call_rcu_mutex);
-	return default_call_rcu_data;
-}
-
-/*
- * Return the call_rcu_data structure that applies to the currently
- * running thread.  Any call_rcu_data structure assigned specifically
- * to this thread has first priority, followed by any call_rcu_data
- * structure assigned to the CPU on which the thread is running,
- * followed by the default call_rcu_data structure.  If there is not
- * yet a default call_rcu_data structure, one will be created.
- */
-struct call_rcu_data *get_call_rcu_data(void)
-{
-	int curcpu;
-	static int warned = 0;
-
-	if (thread_call_rcu_data != NULL)
-		return thread_call_rcu_data;
-	if (maxcpus <= 0)
-		return get_default_call_rcu_data();
-	curcpu = sched_getcpu();
-	if (!warned && (curcpu < 0 || maxcpus <= curcpu)) {
-		fprintf(stderr, "[error] liburcu: gcrd CPU # out of range\n");
-		warned = 1;
-	}
-	if (curcpu >= 0 && maxcpus > curcpu &&
-	    per_cpu_call_rcu_data != NULL &&
-	    per_cpu_call_rcu_data[curcpu] != NULL)
-	    	return per_cpu_call_rcu_data[curcpu];
-	return get_default_call_rcu_data();
-}
-
-/*
- * Return a pointer to this task's call_rcu_data if there is one.
- */
-
-struct call_rcu_data *get_thread_call_rcu_data(void)
-{
-	return thread_call_rcu_data;
-}
-
-/*
- * Set this task's call_rcu_data structure as specified, regardless
- * of whether or not this task already had one.  (This allows switching
- * to and from real-time call_rcu threads, for example.)
- *
- * Use NULL to remove a thread's call_rcu_data structure, but it is
- * the caller's responsibility to dispose of the removed structure.
- * Use get_thread_call_rcu_data() to obtain a pointer to the old structure
- * (prior to NULLing it out, of course).
- */
-
-void set_thread_call_rcu_data(struct call_rcu_data *crdp)
-{
-	thread_call_rcu_data = crdp;
-}
-
-/*
- * Create a separate call_rcu thread for each CPU.  This does not
- * replace a pre-existing call_rcu thread -- use the set_cpu_call_rcu_data()
- * function if you want that behavior.
- */
-
-int create_all_cpu_call_rcu_data(unsigned long flags)
-{
-	int i;
-	struct call_rcu_data *crdp;
-	int ret;
-
-	call_rcu_lock(&call_rcu_mutex);
-	alloc_cpu_call_rcu_data();
-	call_rcu_unlock(&call_rcu_mutex);
-	if (maxcpus <= 0) {
-		errno = EINVAL;
-		return -EINVAL;
-	}
-	if (per_cpu_call_rcu_data == NULL) {
-		errno = ENOMEM;
-		return -ENOMEM;
-	}
-	for (i = 0; i < maxcpus; i++) {
-		call_rcu_lock(&call_rcu_mutex);
-		if (get_cpu_call_rcu_data(i)) {
-			call_rcu_unlock(&call_rcu_mutex);
-			continue;
-		}
-		crdp = __create_call_rcu_data(flags);
-		if (crdp == NULL) {
-			call_rcu_unlock(&call_rcu_mutex);
-			errno = ENOMEM;
-			return -ENOMEM;
-		}
-		call_rcu_unlock(&call_rcu_mutex);
-		if ((ret = set_cpu_call_rcu_data(i, crdp)) != 0) {
-			/* FIXME: Leaks crdp for now. */
-			return ret; /* Can happen on race. */
-		}
-	}
-	return 0;
-}
-
-/*
- * Wake up the call_rcu thread corresponding to the specified
- * call_rcu_data structure.
- */
-static void wake_call_rcu_thread(struct call_rcu_data *crdp)
-{
-	if (!(_CMM_LOAD_SHARED(crdp->flags) & URCU_CALL_RCU_RT)) {
-		call_rcu_lock(&crdp->mtx);
-		if (!(_CMM_LOAD_SHARED(crdp->flags) & URCU_CALL_RCU_RUNNING)) {
-			if (pthread_cond_signal(&crdp->cond) != 0) {
-				perror("pthread_cond_signal");
-				exit(-1);
-			}
-		}
-		call_rcu_unlock(&crdp->mtx);
-	}
-}
-
-/*
- * Schedule a function to be invoked after a following grace period.
- * This is the only function that must be called -- the others are
- * only present to allow applications to tune their use of RCU for
- * maximum performance.
- *
- * Note that unless a call_rcu thread has not already been created,
- * the first invocation of call_rcu() will create one.  So, if you
- * need the first invocation of call_rcu() to be fast, make sure
- * to create a call_rcu thread first.  One way to accomplish this is
- * "get_call_rcu_data();", and another is create_all_cpu_call_rcu_data().
- */
-
-void call_rcu(struct rcu_head *head,
-	      void (*func)(struct rcu_head *head))
-{
-	struct call_rcu_data *crdp;
-
-	cds_wfq_node_init(&head->next);
-	head->func = func;
-	crdp = get_call_rcu_data();
-	cds_wfq_enqueue(&crdp->cbs, &head->next);
-	uatomic_inc(&crdp->qlen);
-	wake_call_rcu_thread(crdp);
-}
-
-/*
- * Free up the specified call_rcu_data structure, terminating the
- * associated call_rcu thread.  The caller must have previously
- * removed the call_rcu_data structure from per-thread or per-CPU
- * usage.  For example, set_cpu_call_rcu_data(cpu, NULL) for per-CPU
- * call_rcu_data structures or set_thread_call_rcu_data(NULL) for
- * per-thread call_rcu_data structures.
- *
- * We silently refuse to free up the default call_rcu_data structure
- * because that is where we put any leftover callbacks.  Note that
- * the possibility of self-spawning callbacks makes it impossible
- * to execute all the callbacks in finite time without putting any
- * newly spawned callbacks somewhere else.  The "somewhere else" of
- * last resort is the default call_rcu_data structure.
- *
- * We also silently refuse to free NULL pointers.  This simplifies
- * the calling code.
- */
-void call_rcu_data_free(struct call_rcu_data *crdp)
-{
-	struct cds_wfq_node *cbs;
-	struct cds_wfq_node **cbs_tail;
-	struct cds_wfq_node **cbs_endprev;
-
-	if (crdp == NULL || crdp == default_call_rcu_data) {
-		return;
-	}
-	if ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0) {
-		call_rcu_lock(&crdp->mtx);
-		crdp->flags |= URCU_CALL_RCU_STOP;
-		call_rcu_unlock(&crdp->mtx);
-		wake_call_rcu_thread(crdp);
-		while ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0)
-			poll(NULL, 0, 1);
-	}
-	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
-		while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
-			poll(NULL, 0, 1);
-		_CMM_STORE_SHARED(crdp->cbs.head, NULL);
-		cbs_tail = (struct cds_wfq_node **)
-			uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
-		cbs_endprev = (struct cds_wfq_node **)
-			uatomic_xchg(&default_call_rcu_data, cbs_tail);
-		*cbs_endprev = cbs;
-		uatomic_add(&default_call_rcu_data->qlen,
-			    uatomic_read(&crdp->qlen));
-		cds_list_del(&crdp->list);
-		free(crdp);
-	}
-}
-
-/*
- * Clean up all the per-CPU call_rcu threads.
- */
-void free_all_cpu_call_rcu_data(void)
-{
-	int cpu;
-	struct call_rcu_data *crdp;
-
-	if (maxcpus <= 0)
-		return;
-	for (cpu = 0; cpu < maxcpus; cpu++) {
-		crdp = get_cpu_call_rcu_data(cpu);
-		if (crdp == NULL)
-			continue;
-		set_cpu_call_rcu_data(cpu, NULL);
-		call_rcu_data_free(crdp);
-	}
-}
-
-/*
- * Acquire the call_rcu_mutex in order to ensure that the child sees
- * all of the call_rcu() data structures in a consistent state.
- * Suitable for pthread_atfork() and friends.
- */
-void call_rcu_before_fork(void)
-{
-	call_rcu_lock(&call_rcu_mutex);
-}
-
-/*
- * Clean up call_rcu data structures in the parent of a successful fork()
- * that is not followed by exec() in the child.  Suitable for
- * pthread_atfork() and friends.
- */
-void call_rcu_after_fork_parent(void)
-{
-	call_rcu_unlock(&call_rcu_mutex);
-}
-
-/*
- * Clean up call_rcu data structures in the child of a successful fork()
- * that is not followed by exec().  Suitable for pthread_atfork() and
- * friends.
- */
-void call_rcu_after_fork_child(void)
-{
-	struct call_rcu_data *crdp;
-
-	/* Release the mutex. */
-	call_rcu_unlock(&call_rcu_mutex);
-
-	/*
-	 * Allocate a new default call_rcu_data structure in order
-	 * to get a working call_rcu thread to go with it.
-	 */
-	default_call_rcu_data = NULL;
-	(void)get_default_call_rcu_data();
-
-	/* Dispose of all of the rest of the call_rcu_data structures. */
-	while (call_rcu_data_list.next != call_rcu_data_list.prev) {
-		crdp = cds_list_entry(call_rcu_data_list.prev,
-				      struct call_rcu_data, list);
-		if (crdp == default_call_rcu_data)
-			crdp = cds_list_entry(crdp->list.prev,
-					      struct call_rcu_data, list);
-		crdp->flags = URCU_CALL_RCU_STOPPED;
-		call_rcu_data_free(crdp);
-	}
-}
diff --git a/urcu-map.h b/urcu-map.h
new file mode 100644
index 0000000..eccaac2
--- /dev/null
+++ b/urcu-map.h
@@ -0,0 +1,117 @@
+#ifndef _URCU_MAP_H
+#define _URCU_MAP_H
+
+/*
+ * urcu-map.h
+ *
+ * Userspace RCU header -- name mapping to allow multiple flavors to be
+ * used in the same executable.
+ *
+ * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
+ * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
+ *
+ * LGPL-compatible code should include this header with :
+ *
+ * #define _LGPL_SOURCE
+ * #include <urcu.h>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ *
+ * IBM's contributions to this file may be relicensed under LGPLv2 or later.
+ */
+
+/* Mapping macros to allow multiple flavors in a single binary. */
+
+#ifdef RCU_MEMBARRIER
+
+#define rcu_read_lock			rcu_read_lock_memb
+#define _rcu_read_lock			_rcu_read_lock_memb
+#define rcu_read_unlock			rcu_read_unlock_memb
+#define _rcu_read_unlock		_rcu_read_unlock_memb
+#define rcu_register_thread		rcu_register_thread_memb
+#define rcu_unregister_thread		rcu_unregister_thread_memb
+#define rcu_init			rcu_init_memb
+#define rcu_exit			rcu_exit_memb
+#define synchronize_rcu			synchronize_rcu_memb
+#define rcu_reader			rcu_reader_memb
+#define rcu_gp_ctr			rcu_gp_ctr_memb
+
+#define get_cpu_call_rcu_data		get_cpu_call_rcu_data_memb
+#define get_call_rcu_thread		get_call_rcu_thread_memb
+#define create_call_rcu_data		create_call_rcu_data_memb
+#define set_cpu_call_rcu_data		set_cpu_call_rcu_data_memb
+#define get_default_call_rcu_data	get_default_call_rcu_data_memb
+#define get_call_rcu_data		get_call_rcu_data_memb
+#define get_thread_call_rcu_data	get_thread_call_rcu_data_memb
+#define set_thread_call_rcu_data	set_thread_call_rcu_data_memb
+#define create_all_cpu_call_rcu_data	create_all_cpu_call_rcu_data_memb
+#define free_all_cpu_call_rcu_data	free_all_cpu_call_rcu_data_memb
+#define call_rcu			call_rcu_memb
+
+#elif defined(RCU_SIGNAL)
+
+#define rcu_read_lock			rcu_read_lock_sig
+#define _rcu_read_lock			_rcu_read_lock_sig
+#define rcu_read_unlock			rcu_read_unlock_sig
+#define _rcu_read_unlock		_rcu_read_unlock_sig
+#define rcu_register_thread		rcu_register_thread_sig
+#define rcu_unregister_thread		rcu_unregister_thread_sig
+#define rcu_init			rcu_init_sig
+#define rcu_exit			rcu_exit_sig
+#define synchronize_rcu			synchronize_rcu_sig
+#define rcu_reader			rcu_reader_sig
+#define rcu_gp_ctr			rcu_gp_ctr_sig
+
+#define get_cpu_call_rcu_data		get_cpu_call_rcu_data_sig
+#define get_call_rcu_thread		get_call_rcu_thread_sig
+#define create_call_rcu_data		create_call_rcu_data_sig
+#define set_cpu_call_rcu_data		set_cpu_call_rcu_data_sig
+#define get_default_call_rcu_data	get_default_call_rcu_data_sig
+#define get_call_rcu_data		get_call_rcu_data_sig
+#define get_thread_call_rcu_data	get_thread_call_rcu_data_sig
+#define set_thread_call_rcu_data	set_thread_call_rcu_data_sig
+#define create_all_cpu_call_rcu_data	create_all_cpu_call_rcu_data_sig
+#define free_all_cpu_call_rcu_data	free_all_cpu_call_rcu_data_sig
+#define call_rcu			call_rcu_sig
+
+#elif defined(RCU_MB)
+
+#define rcu_read_lock			rcu_read_lock_mb
+#define _rcu_read_lock			_rcu_read_lock_mb
+#define rcu_read_unlock			rcu_read_unlock_mb
+#define _rcu_read_unlock		_rcu_read_unlock_mb
+#define rcu_register_thread		rcu_register_thread_mb
+#define rcu_unregister_thread		rcu_unregister_thread_mb
+#define rcu_init			rcu_init_mb
+#define rcu_exit			rcu_exit_mb
+#define synchronize_rcu			synchronize_rcu_mb
+#define rcu_reader			rcu_reader_mb
+#define rcu_gp_ctr			rcu_gp_ctr_mb
+
+#define get_cpu_call_rcu_data		get_cpu_call_rcu_data_mb
+#define get_call_rcu_thread		get_call_rcu_thread_mb
+#define create_call_rcu_data		create_call_rcu_data_mb
+#define set_cpu_call_rcu_data		set_cpu_call_rcu_data_mb
+#define get_default_call_rcu_data	get_default_call_rcu_data_mb
+#define get_call_rcu_data		get_call_rcu_data_mb
+#define get_thread_call_rcu_data	get_thread_call_rcu_data_mb
+#define set_thread_call_rcu_data	set_thread_call_rcu_data_mb
+#define create_all_cpu_call_rcu_data	create_all_cpu_call_rcu_data_mb
+#define free_all_cpu_call_rcu_data	free_all_cpu_call_rcu_data_mb
+#define call_rcu			call_rcu_mb
+
+#endif
+
+#endif /* _URCU_MAP_H */
diff --git a/urcu-qsbr-map.h b/urcu-qsbr-map.h
new file mode 100644
index 0000000..2bce1b6
--- /dev/null
+++ b/urcu-qsbr-map.h
@@ -0,0 +1,63 @@
+#ifndef _URCU_QSBR_MAP_H
+#define _URCU_QSBR_MAP_H
+
+/*
+ * urcu-map.h
+ *
+ * Userspace RCU header -- name mapping to allow multiple flavors to be
+ * used in the same executable.
+ *
+ * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
+ * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
+ *
+ * LGPL-compatible code should include this header with :
+ *
+ * #define _LGPL_SOURCE
+ * #include <urcu.h>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ *
+ * IBM's contributions to this file may be relicensed under LGPLv2 or later.
+ */
+
+/* Mapping macros to allow multiple flavors in a single binary. */
+
+#define rcu_read_lock			rcu_read_lock_qsbr
+#define _rcu_read_lock			_rcu_read_lock_qsbr
+#define rcu_read_unlock			rcu_read_unlock_qsbr
+#define _rcu_read_unlock		_rcu_read_unlock_qsbr
+#define rcu_quiescent_state		rcu_quiescent_state_qsbr
+#define _rcu_quiescent_state		_rcu_quiescent_state_qsbr
+#define rcu_thread_offline		rcu_thread_offline_qsbr
+#define rcu_thread_online		rcu_thread_online_qsbr
+#define rcu_register_thread		rcu_register_thread_qsbr
+#define rcu_unregister_thread		rcu_unregister_thread_qsbr
+#define rcu_exit			rcu_exit_qsbr
+#define synchronize_rcu			synchronize_rcu_qsbr
+#define rcu_reader			rcu_reader_qsbr
+#define rcu_gp_ctr			rcu_gp_ctr_qsbr
+
+#define get_cpu_call_rcu_data		get_cpu_call_rcu_data_qsbr
+#define get_call_rcu_thread		get_call_rcu_thread_qsbr
+#define create_call_rcu_data		create_call_rcu_data_qsbr
+#define set_cpu_call_rcu_data		set_cpu_call_rcu_data_qsbr
+#define get_default_call_rcu_data	get_default_call_rcu_data_qsbr
+#define get_call_rcu_data		get_call_rcu_data_qsbr
+#define get_thread_call_rcu_data	get_thread_call_rcu_data_qsbr
+#define set_thread_call_rcu_data	set_thread_call_rcu_data_qsbr
+#define create_all_cpu_call_rcu_data	create_all_cpu_call_rcu_data_qsbr
+#define call_rcu			call_rcu_qsbr
+
+#endif /* _URCU_QSBR_MAP_H */
diff --git a/urcu-qsbr.c b/urcu-qsbr.c
index 69effd5..8dcad33 100644
--- a/urcu-qsbr.c
+++ b/urcu-qsbr.c
@@ -32,6 +32,8 @@
 #include <errno.h>
 #include <poll.h>
 
+#include "urcu-qsbr-map.h"
+
 #define BUILD_QSBR_LIB
 #include "urcu-qsbr-static.h"
 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
@@ -121,10 +123,11 @@ static void update_counter_and_wait(void)
 #endif	/* !(CAA_BITS_PER_LONG < 64) */
 
 	/*
-	 * Must commit rcu_gp_ctr update to memory before waiting for quiescent
-	 * state. Failure to do so could result in the writer waiting forever
-	 * while new readers are always accessing data (no progress). Enforce
-	 * compiler-order of store to rcu_gp_ctr before load rcu_reader ctr.
+	 * Must commit rcu_gp_ctr update to memory before waiting for
+	 * quiescent state. Failure to do so could result in the writer
+	 * waiting forever while new readers are always accessing data
+	 * (no progress). Enforce compiler-order of store to rcu_gp_ctr
+	 * before load rcu_reader ctr.
 	 */
 	cmm_barrier();
 
@@ -194,8 +197,8 @@ void synchronize_rcu(void)
 
 	/*
 	 * Mark the writer thread offline to make sure we don't wait for
-	 * our own quiescent state. This allows using synchronize_rcu() in
-	 * threads registered as readers.
+	 * our own quiescent state. This allows using synchronize_rcu()
+	 * in threads registered as readers.
 	 */
 	if (was_online)
 		CMM_STORE_SHARED(rcu_reader.ctr, 0);
@@ -212,10 +215,11 @@ void synchronize_rcu(void)
 
 	/*
 	 * Must finish waiting for quiescent state for parity 0 before
-	 * committing next rcu_gp_ctr update to memory. Failure to do so could
-	 * result in the writer waiting forever while new readers are always
-	 * accessing data (no progress).  Enforce compiler-order of load
-	 * rcu_reader ctr before store to rcu_gp_ctr.
+	 * committing next rcu_gp_ctr update to memory. Failure to
+	 * do so could result in the writer waiting forever while new
+	 * readers are always accessing data (no progress).  Enforce
+	 * compiler-order of load rcu_reader ctr before store to
+	 * rcu_gp_ctr.
 	 */
 	cmm_barrier();
 
@@ -238,7 +242,8 @@ out:
 	 * freed.
 	 */
 	if (was_online)
-		_CMM_STORE_SHARED(rcu_reader.ctr, CMM_LOAD_SHARED(rcu_gp_ctr));
+		_CMM_STORE_SHARED(rcu_reader.ctr,
+				  CMM_LOAD_SHARED(rcu_gp_ctr));
 	cmm_smp_mb();
 }
 #else /* !(CAA_BITS_PER_LONG < 64) */
@@ -250,8 +255,8 @@ void synchronize_rcu(void)
 
 	/*
 	 * Mark the writer thread offline to make sure we don't wait for
-	 * our own quiescent state. This allows using synchronize_rcu() in
-	 * threads registered as readers.
+	 * our own quiescent state. This allows using synchronize_rcu()
+	 * in threads registered as readers.
 	 */
 	cmm_smp_mb();
 	if (was_online)
@@ -265,7 +270,8 @@ out:
 	mutex_unlock(&rcu_gp_lock);
 
 	if (was_online)
-		_CMM_STORE_SHARED(rcu_reader.ctr, CMM_LOAD_SHARED(rcu_gp_ctr));
+		_CMM_STORE_SHARED(rcu_reader.ctr,
+				  CMM_LOAD_SHARED(rcu_gp_ctr));
 	cmm_smp_mb();
 }
 #endif  /* !(CAA_BITS_PER_LONG < 64) */
@@ -326,3 +332,5 @@ void rcu_exit(void)
 {
 	assert(cds_list_empty(&registry));
 }
+
+#include "urcu-call-rcu-impl.h"
diff --git a/urcu-qsbr.h b/urcu-qsbr.h
index 116fd77..984d70c 100644
--- a/urcu-qsbr.h
+++ b/urcu-qsbr.h
@@ -40,6 +40,8 @@
 extern "C" {
 #endif 
 
+#include "urcu-qsbr-map.h"
+
 /*
  * Important !
  *
@@ -62,15 +64,15 @@ extern "C" {
  * rcu_read_unlock()
  *
  * Mark the beginning and end of a read-side critical section.
- * DON'T FORGET TO USE rcu_register_thread/rcu_unregister_thread() FOR EACH
- * THREAD WITH READ-SIDE CRITICAL SECTION.
+ * DON'T FORGET TO USE rcu_register_thread/rcu_unregister_thread()
+ * FOR EACH THREAD WITH READ-SIDE CRITICAL SECTION.
  */
-#define rcu_read_lock()		_rcu_read_lock()
-#define rcu_read_unlock()	_rcu_read_unlock()
+#define rcu_read_lock_qsbr()		_rcu_read_lock()
+#define rcu_read_unlock_qsbr()		_rcu_read_unlock()
 
-#define rcu_quiescent_state()	_rcu_quiescent_state()
-#define rcu_thread_offline()	_rcu_thread_offline()
-#define rcu_thread_online()	_rcu_thread_online()
+#define rcu_quiescent_state_qsbr()	_rcu_quiescent_state()
+#define rcu_thread_offline_qsbr()	_rcu_thread_offline()
+#define rcu_thread_online_qsbr()	_rcu_thread_online()
 
 #else /* !_LGPL_SOURCE */
 
@@ -122,4 +124,6 @@ extern void rcu_unregister_thread(void);
 }
 #endif
 
+#include "urcu-call-rcu.h"
+
 #endif /* _URCU_QSBR_H */
diff --git a/urcu.c b/urcu.c
index e529ac0..4ee9e3b 100644
--- a/urcu.c
+++ b/urcu.c
@@ -33,6 +33,8 @@
 #include <errno.h>
 #include <poll.h>
 
+#include "urcu-map.h"
+
 #include "urcu-static.h"
 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
 #include "urcu.h"
@@ -428,4 +430,7 @@ void rcu_exit(void)
 	assert(act.sa_sigaction == sigrcu_handler);
 	assert(cds_list_empty(&registry));
 }
+
 #endif /* #ifdef RCU_SIGNAL */
+
+#include "urcu-call-rcu-impl.h"
diff --git a/urcu.h b/urcu.h
index c6c54e7..00d9b75 100644
--- a/urcu.h
+++ b/urcu.h
@@ -43,12 +43,14 @@
 extern "C" {
 #endif 
 
+#include "urcu-map.h"
+
 /*
  * Important !
  *
  * Each thread containing read-side critical sections must be registered
- * with rcu_register_thread() before calling rcu_read_lock().
- * rcu_unregister_thread() should be called before the thread exits.
+ * with rcu_register_thread_mb() before calling rcu_read_lock_mb().
+ * rcu_unregister_thread_mb() should be called before the thread exits.
  */
 
 #ifdef _LGPL_SOURCE
@@ -68,8 +70,16 @@ extern "C" {
  * DON'T FORGET TO USE RCU_REGISTER/UNREGISTER_THREAD() FOR EACH THREAD WITH
  * READ-SIDE CRITICAL SECTION.
  */
-#define rcu_read_lock()		_rcu_read_lock()
-#define rcu_read_unlock()	_rcu_read_unlock()
+#ifdef RCU_MEMBARRIER
+#define rcu_read_lock_memb()		_rcu_read_lock()
+#define rcu_read_unlock_memb()		_rcu_read_unlock()
+#elif defined(RCU_SIGNAL)
+#define rcu_read_lock_sig()		_rcu_read_lock()
+#define rcu_read_unlock_sig()		_rcu_read_unlock()
+#elif defined(RCU_MB)
+#define rcu_read_lock_mb()		_rcu_read_lock()
+#define rcu_read_unlock_mb()		_rcu_read_unlock()
+#endif
 
 #else /* !_LGPL_SOURCE */
 
@@ -100,4 +110,6 @@ extern void rcu_init(void);
 }
 #endif
 
+#include "urcu-call-rcu.h"
+
 #endif /* _URCU_H */
-- 
1.7.3.2

[ltt-dev] [PATCH urcu 3/5] Allow taking address of rcu_read_lock() and rcu_read_unlock()

[Paul E. McKenney]

Also incorporate README feedback from Mathieu and Josh.

Signed-off-by: Paul E. McKenney <paulmck at linux.vnet.ibm.com>
---
 README      |    9 ++++++---
 urcu-bp.h   |    4 ++--
 urcu-qsbr.h |   10 +++++-----
 urcu.h      |   12 ++++++------
 4 files changed, 19 insertions(+), 16 deletions(-)

diff --git a/README b/README
index 56e98d7..659511f 100644
--- a/README
+++ b/README
@@ -205,6 +205,9 @@ Interaction with fork()
 	rcu_bp_before_fork, rcu_bp_after_fork_parent and
 	rcu_bp_after_fork_child.
 
-	Applications that use call_rcu() are required to invoke
-	call_rcu_after_fork_child() from the child process after a
-	successful fork() system call that is not followed by exec().
+	Applications that use call_rcu() and that fork() without
+	doing an immediate exec() must take special action.  The parent
+	must invoke call_rcu_before_fork() before the fork() and
+	call_rcu_after_fork_parent() after the fork().  The child
+	process must invoke call_rcu_after_fork_child().
+	These three APIs are suitable for passing to pthread_atfork().
diff --git a/urcu-bp.h b/urcu-bp.h
index fdf885c..bc2dbc3 100644
--- a/urcu-bp.h
+++ b/urcu-bp.h
@@ -71,8 +71,8 @@ extern "C" {
  *
  * Mark the beginning and end of a read-side critical section.
  */
-#define rcu_read_lock_bp()		_rcu_read_lock()
-#define rcu_read_unlock_bp()		_rcu_read_unlock()
+#define rcu_read_lock_bp		_rcu_read_lock
+#define rcu_read_unlock_bp		_rcu_read_unlock
 
 #else /* !_LGPL_SOURCE */
 
diff --git a/urcu-qsbr.h b/urcu-qsbr.h
index 984d70c..7ef1bfe 100644
--- a/urcu-qsbr.h
+++ b/urcu-qsbr.h
@@ -67,12 +67,12 @@ extern "C" {
  * DON'T FORGET TO USE rcu_register_thread/rcu_unregister_thread()
  * FOR EACH THREAD WITH READ-SIDE CRITICAL SECTION.
  */
-#define rcu_read_lock_qsbr()		_rcu_read_lock()
-#define rcu_read_unlock_qsbr()		_rcu_read_unlock()
+#define rcu_read_lock_qsbr		_rcu_read_lock
+#define rcu_read_unlock_qsbr		_rcu_read_unlock
 
-#define rcu_quiescent_state_qsbr()	_rcu_quiescent_state()
-#define rcu_thread_offline_qsbr()	_rcu_thread_offline()
-#define rcu_thread_online_qsbr()	_rcu_thread_online()
+#define rcu_quiescent_state_qsbr	_rcu_quiescent_state
+#define rcu_thread_offline_qsbr		_rcu_thread_offline
+#define rcu_thread_online_qsbr		_rcu_thread_online
 
 #else /* !_LGPL_SOURCE */
 
diff --git a/urcu.h b/urcu.h
index 00d9b75..417e609 100644
--- a/urcu.h
+++ b/urcu.h
@@ -71,14 +71,14 @@ extern "C" {
  * READ-SIDE CRITICAL SECTION.
  */
 #ifdef RCU_MEMBARRIER
-#define rcu_read_lock_memb()		_rcu_read_lock()
-#define rcu_read_unlock_memb()		_rcu_read_unlock()
+#define rcu_read_lock_memb		_rcu_read_lock
+#define rcu_read_unlock_memb		_rcu_read_unlock
 #elif defined(RCU_SIGNAL)
-#define rcu_read_lock_sig()		_rcu_read_lock()
-#define rcu_read_unlock_sig()		_rcu_read_unlock()
+#define rcu_read_lock_sig		_rcu_read_lock
+#define rcu_read_unlock_sig		_rcu_read_unlock
 #elif defined(RCU_MB)
-#define rcu_read_lock_mb()		_rcu_read_lock()
-#define rcu_read_unlock_mb()		_rcu_read_unlock()
+#define rcu_read_lock_mb		_rcu_read_lock
+#define rcu_read_unlock_mb		_rcu_read_unlock
 #endif
 
 #else /* !_LGPL_SOURCE */
-- 
1.7.3.2

[ltt-dev] [PATCH urcu 4/5] Make defer_rcu() usable from library using multiple URCU implementations

[Paul E. McKenney]

Signed-off-by: Paul E. McKenney <paulmck at linux.vnet.ibm.com>
---
 Makefile.am         |    4 +-
 README              |   11 +-
 tests/Makefile.am   |    2 +-
 urcu-bp-map.h       |    6 +
 urcu-bp.c           |    1 +
 urcu-bp.h           |    1 +
 urcu-defer-impl.h   |  450 +++++++++++++++++++++++++++++++++++++++++++++++++++
 urcu-defer-static.h |  108 ------------
 urcu-defer.c        |  385 -------------------------------------------
 urcu-map.h          |   26 +++
 urcu-qsbr-map.h     |    6 +
 urcu-qsbr.c         |    1 +
 urcu-qsbr.h         |    1 +
 urcu.c              |    1 +
 urcu.h              |    1 +
 15 files changed, 501 insertions(+), 503 deletions(-)
 create mode 100644 urcu-defer-impl.h
 delete mode 100644 urcu-defer-static.h
 delete mode 100644 urcu-defer.c

diff --git a/Makefile.am b/Makefile.am
index 4b70644..1589884 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -30,7 +30,7 @@ COMPAT+=compat_futex.c
 endif
 
 lib_LTLIBRARIES = liburcu.la liburcu-qsbr.la liburcu-mb.la liburcu-signal.la \
-		  liburcu-bp.la liburcu-defer.la \
+		  liburcu-bp.la \
 		  libwfqueue.la libwfstack.la librculfqueue.la librculfstack.la
 
 liburcu_la_SOURCES = urcu.c urcu-pointer.c $(COMPAT)
@@ -45,8 +45,6 @@ liburcu_signal_la_CFLAGS = -DRCU_SIGNAL
 
 liburcu_bp_la_SOURCES = urcu-bp.c urcu-pointer.c $(COMPAT)
 
-liburcu_defer_la_SOURCES = urcu-defer.c $(COMPAT)
-
 libwfqueue_la_SOURCES = wfqueue.c $(COMPAT)
 libwfstack_la_SOURCES = wfstack.c $(COMPAT)
 librculfqueue_la_SOURCES = rculfqueue.c $(COMPAT)
diff --git a/README b/README
index 659511f..7d97f19 100644
--- a/README
+++ b/README
@@ -124,16 +124,15 @@ Writing
 
 Usage of liburcu-defer
 
-	* #include <urcu-defer.h>
-	* Link with "-lurcu-defer", and also with one of the urcu library
-	  (either urcu, urcu-bp, urcu-mb or urcu-qsbr).
+	* Follow instructions for either liburcu, liburcu-qsbr,
+	  liburcu-mb, liburcu-signal, or liburcu-bp above.
+	  The liburcu-defer functionality is pulled into each of
+	  those library modules.
 	* Provides defer_rcu() primitive to enqueue delayed callbacks. Queued
 	  callbacks are executed in batch periodically after a grace period.
 	  Do _not_ use defer_rcu() within a read-side critical section, because
 	  it may call synchronize_rcu() if the thread queue is full.
-	* Provides defer_rcu_ratelimit() primitive, which acts just like
-	  defer_rcu(), but takes an additional rate limiter callback forcing
-	  synchronized callback execution of the limiter returns non-zero.
+	  This can lead to deadlock or worse.
 	* Requires that rcu_defer_barrier() must be called in library destructor
 	  if a library queues callbacks and is expected to be unloaded with
 	  dlclose().
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 8dacb11..5598689 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -35,7 +35,7 @@ URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.
 # URCU_SIGNAL uses urcu.c but -DRCU_SIGNAL must be defined
 URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
 URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
-URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-defer.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
 
 URCU_LIB=$(top_builddir)/liburcu.la
 URCU_QSBR_LIB=$(top_builddir)/liburcu-qsbr.la
diff --git a/urcu-bp-map.h b/urcu-bp-map.h
index 6321802..1d77006 100644
--- a/urcu-bp-map.h
+++ b/urcu-bp-map.h
@@ -58,4 +58,10 @@
 #define free_all_cpu_call_rcu_data	free_all_cpu_call_rcu_data_bp
 #define call_rcu			call_rcu_bp
 
+#define defer_rcu			defer_rcu_bp
+#define rcu_defer_register_thread	rcu_defer_register_thread_bp
+#define rcu_defer_unregister_thread	rcu_defer_unregister_thread_bp
+#define	rcu_defer_barrier		rcu_defer_barrier_bp
+#define rcu_defer_barrier_thread	rcu_defer_barrier_thread_bp
+
 #endif /* _URCU_BP_MAP_H */
diff --git a/urcu-bp.c b/urcu-bp.c
index 5474f9f..14b2001 100644
--- a/urcu-bp.c
+++ b/urcu-bp.c
@@ -379,3 +379,4 @@ void rcu_bp_after_fork_child(void)
 }
 
 #include "urcu-call-rcu-impl.h"
+#include "urcu-defer-impl.h"
diff --git a/urcu-bp.h b/urcu-bp.h
index bc2dbc3..21fe66a 100644
--- a/urcu-bp.h
+++ b/urcu-bp.h
@@ -118,5 +118,6 @@ static inline void rcu_init(void)
 #endif
 
 #include "urcu-call-rcu.h"
+#include "urcu-defer.h"
 
 #endif /* _URCU_BP_H */
diff --git a/urcu-defer-impl.h b/urcu-defer-impl.h
new file mode 100644
index 0000000..0aedd53
--- /dev/null
+++ b/urcu-defer-impl.h
@@ -0,0 +1,450 @@
+#ifndef _URCU_DEFER_IMPL_H
+#define _URCU_DEFER_IMPL_H
+
+/*
+ * urcu-defer-impl.h
+ *
+ * Userspace RCU header - memory reclamation.
+ *
+ * TO BE INCLUDED ONLY FROM URCU LIBRARY CODE. See urcu-defer.h for linking
+ * dynamically with the userspace rcu reclamation library.
+ *
+ * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
+ * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ *
+ * IBM's contributions to this file may be relicensed under LGPLv2 or later.
+ */
+
+#include <stdlib.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <signal.h>
+#include <assert.h>
+#include <string.h>
+#include <errno.h>
+#include <poll.h>
+#include <sys/time.h>
+#include <syscall.h>
+#include <unistd.h>
+
+#include "urcu/urcu-futex.h"
+
+#include <urcu/compiler.h>
+#include <urcu/arch.h>
+#include <urcu/uatomic_arch.h>
+#include <urcu/list.h>
+#include <urcu/system.h>
+
+/*
+ * Number of entries in the per-thread defer queue. Must be power of 2.
+ */
+#define DEFER_QUEUE_SIZE	(1 << 12)
+#define DEFER_QUEUE_MASK	(DEFER_QUEUE_SIZE - 1)
+
+/*
+ * Typically, data is aligned at least on the architecture size.
+ * Use lowest bit to indicate that the current callback is changing.
+ * Assumes that (void *)-2L is not used often. Used to encode non-aligned
+ * functions and non-aligned data using extra space.
+ * We encode the (void *)-2L fct as: -2L, fct, data.
+ * We encode the (void *)-2L data as: -2L, fct, data.
+ * Here, DQ_FCT_MARK == ~DQ_FCT_BIT. Required for the test order.
+ */
+#define DQ_FCT_BIT		(1 << 0)
+#define DQ_IS_FCT_BIT(x)	((unsigned long)(x) & DQ_FCT_BIT)
+#define DQ_SET_FCT_BIT(x)	\
+	(x = (void *)((unsigned long)(x) | DQ_FCT_BIT))
+#define DQ_CLEAR_FCT_BIT(x)	\
+	(x = (void *)((unsigned long)(x) & ~DQ_FCT_BIT))
+#define DQ_FCT_MARK		((void *)(~DQ_FCT_BIT))
+
+/*
+ * This code section can only be included in LGPL 2.1 compatible source code.
+ * See below for the function call wrappers which can be used in code meant to
+ * be only linked with the Userspace RCU library. This comes with a small
+ * performance degradation on the read-side due to the added function calls.
+ * This is required to permit relinking with newer versions of the library.
+ */
+
+#ifdef DEBUG_RCU
+#define rcu_assert(args...)	assert(args)
+#else
+#define rcu_assert(args...)
+#endif
+
+/*
+ * defer queue.
+ * Contains pointers. Encoded to save space when same callback is often used.
+ * When looking up the next item:
+ * - if DQ_FCT_BIT is set, set the current callback to DQ_CLEAR_FCT_BIT(ptr)
+ *   - next element contains pointer to data.
+ * - else if item == DQ_FCT_MARK
+ *   - set the current callback to next element ptr
+ *   - following next element contains pointer to data.
+ * - else current element contains data
+ */
+struct defer_queue {
+	unsigned long head;	/* add element at head */
+	void *last_fct_in;	/* last fct pointer encoded */
+	unsigned long tail;	/* next element to remove at tail */
+	void *last_fct_out;	/* last fct pointer encoded */
+	void **q;
+	/* registry information */
+	unsigned long last_head;
+	struct cds_list_head list;	/* list of thread queues */
+};
+
+/* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
+#include "urcu-defer.h"
+
+void __attribute__((destructor)) rcu_defer_exit(void);
+
+extern void synchronize_rcu(void);
+
+/*
+ * rcu_defer_mutex nests inside defer_thread_mutex.
+ */
+static pthread_mutex_t rcu_defer_mutex = PTHREAD_MUTEX_INITIALIZER;
+static pthread_mutex_t defer_thread_mutex = PTHREAD_MUTEX_INITIALIZER;
+
+static int defer_thread_futex;
+
+/*
+ * Written to only by each individual deferer. Read by both the deferer and
+ * the reclamation tread.
+ */
+static struct defer_queue __thread defer_queue;
+static CDS_LIST_HEAD(registry_defer);
+static pthread_t tid_defer;
+
+static void mutex_lock_defer(pthread_mutex_t *mutex)
+{
+	int ret;
+
+#ifndef DISTRUST_SIGNALS_EXTREME
+	ret = pthread_mutex_lock(mutex);
+	if (ret) {
+		perror("Error in pthread mutex lock");
+		exit(-1);
+	}
+#else /* #ifndef DISTRUST_SIGNALS_EXTREME */
+	while ((ret = pthread_mutex_trylock(mutex)) != 0) {
+		if (ret != EBUSY && ret != EINTR) {
+			printf("ret = %d, errno = %d\n", ret, errno);
+			perror("Error in pthread mutex lock");
+			exit(-1);
+		}
+		pthread_testcancel();
+		poll(NULL,0,10);
+	}
+#endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
+}
+
+/*
+ * Wake-up any waiting defer thread. Called from many concurrent threads.
+ */
+static void wake_up_defer(void)
+{
+	if (unlikely(uatomic_read(&defer_thread_futex) == -1)) {
+		uatomic_set(&defer_thread_futex, 0);
+		futex_noasync(&defer_thread_futex, FUTEX_WAKE, 1,
+		      NULL, NULL, 0);
+	}
+}
+
+static unsigned long rcu_defer_num_callbacks(void)
+{
+	unsigned long num_items = 0, head;
+	struct defer_queue *index;
+
+	mutex_lock_defer(&rcu_defer_mutex);
+	cds_list_for_each_entry(index, &registry_defer, list) {
+		head = CMM_LOAD_SHARED(index->head);
+		num_items += head - index->tail;
+	}
+	mutex_unlock(&rcu_defer_mutex);
+	return num_items;
+}
+
+/*
+ * Defer thread waiting. Single thread.
+ */
+static void wait_defer(void)
+{
+	uatomic_dec(&defer_thread_futex);
+	cmm_smp_mb();	/* Write futex before read queue */
+	if (rcu_defer_num_callbacks()) {
+		cmm_smp_mb();	/* Read queue before write futex */
+		/* Callbacks are queued, don't wait. */
+		uatomic_set(&defer_thread_futex, 0);
+	} else {
+		cmm_smp_rmb();	/* Read queue before read futex */
+		if (uatomic_read(&defer_thread_futex) == -1)
+			futex_noasync(&defer_thread_futex, FUTEX_WAIT, -1,
+			      NULL, NULL, 0);
+	}
+}
+
+/*
+ * Must be called after Q.S. is reached.
+ */
+static void rcu_defer_barrier_queue(struct defer_queue *queue,
+				    unsigned long head)
+{
+	unsigned long i;
+	void (*fct)(void *p);
+	void *p;
+
+	/*
+	 * Tail is only modified when lock is held.
+	 * Head is only modified by owner thread.
+	 */
+
+	for (i = queue->tail; i != head;) {
+		cmm_smp_rmb();       /* read head before q[]. */
+		p = CMM_LOAD_SHARED(queue->q[i++ & DEFER_QUEUE_MASK]);
+		if (unlikely(DQ_IS_FCT_BIT(p))) {
+			DQ_CLEAR_FCT_BIT(p);
+			queue->last_fct_out = p;
+			p = CMM_LOAD_SHARED(queue->q[i++ & DEFER_QUEUE_MASK]);
+		} else if (unlikely(p == DQ_FCT_MARK)) {
+			p = CMM_LOAD_SHARED(queue->q[i++ & DEFER_QUEUE_MASK]);
+			queue->last_fct_out = p;
+			p = CMM_LOAD_SHARED(queue->q[i++ & DEFER_QUEUE_MASK]);
+		}
+		fct = queue->last_fct_out;
+		fct(p);
+	}
+	cmm_smp_mb();	/* push tail after having used q[] */
+	CMM_STORE_SHARED(queue->tail, i);
+}
+
+static void _rcu_defer_barrier_thread(void)
+{
+	unsigned long head, num_items;
+
+	head = defer_queue.head;
+	num_items = head - defer_queue.tail;
+	if (unlikely(!num_items))
+		return;
+	synchronize_rcu();
+	rcu_defer_barrier_queue(&defer_queue, head);
+}
+
+void rcu_defer_barrier_thread(void)
+{
+	mutex_lock_defer(&rcu_defer_mutex);
+	_rcu_defer_barrier_thread();
+	mutex_unlock(&rcu_defer_mutex);
+}
+
+/*
+ * rcu_defer_barrier - Execute all queued rcu callbacks.
+ *
+ * Execute all RCU callbacks queued before rcu_defer_barrier() execution.
+ * All callbacks queued on the local thread prior to a rcu_defer_barrier() call
+ * are guaranteed to be executed.
+ * Callbacks queued by other threads concurrently with rcu_defer_barrier()
+ * execution are not guaranteed to be executed in the current batch (could
+ * be left for the next batch). These callbacks queued by other threads are only
+ * guaranteed to be executed if there is explicit synchronization between
+ * the thread adding to the queue and the thread issuing the defer_barrier call.
+ */
+
+void rcu_defer_barrier(void)
+{
+	struct defer_queue *index;
+	unsigned long num_items = 0;
+
+	if (cds_list_empty(&registry_defer))
+		return;
+
+	mutex_lock_defer(&rcu_defer_mutex);
+	cds_list_for_each_entry(index, &registry_defer, list) {
+		index->last_head = CMM_LOAD_SHARED(index->head);
+		num_items += index->last_head - index->tail;
+	}
+	if (likely(!num_items)) {
+		/*
+		 * We skip the grace period because there are no queued
+		 * callbacks to execute.
+		 */
+		goto end;
+	}
+	synchronize_rcu();
+	cds_list_for_each_entry(index, &registry_defer, list)
+		rcu_defer_barrier_queue(index, index->last_head);
+end:
+	mutex_unlock(&rcu_defer_mutex);
+}
+
+/*
+ * _defer_rcu - Queue a RCU callback.
+ */
+void _defer_rcu(void (*fct)(void *p), void *p)
+{
+	unsigned long head, tail;
+
+	/*
+	 * Head is only modified by ourself. Tail can be modified by reclamation
+	 * thread.
+	 */
+	head = defer_queue.head;
+	tail = CMM_LOAD_SHARED(defer_queue.tail);
+
+	/*
+	 * If queue is full, or reached threshold. Empty queue ourself.
+	 * Worse-case: must allow 2 supplementary entries for fct pointer.
+	 */
+	if (unlikely(head - tail >= DEFER_QUEUE_SIZE - 2)) {
+		assert(head - tail <= DEFER_QUEUE_SIZE);
+		rcu_defer_barrier_thread();
+		assert(head - CMM_LOAD_SHARED(defer_queue.tail) == 0);
+	}
+
+	if (unlikely(defer_queue.last_fct_in != fct)) {
+		defer_queue.last_fct_in = fct;
+		if (unlikely(DQ_IS_FCT_BIT(fct) || fct == DQ_FCT_MARK)) {
+			/*
+			 * If the function to encode is not aligned or the
+			 * marker, write DQ_FCT_MARK followed by the function
+			 * pointer.
+			 */
+			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
+				      DQ_FCT_MARK);
+			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
+				      fct);
+		} else {
+			DQ_SET_FCT_BIT(fct);
+			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
+				      fct);
+		}
+	} else {
+		if (unlikely(DQ_IS_FCT_BIT(p) || p == DQ_FCT_MARK)) {
+			/*
+			 * If the data to encode is not aligned or the marker,
+			 * write DQ_FCT_MARK followed by the function pointer.
+			 */
+			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
+				      DQ_FCT_MARK);
+			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
+				      fct);
+		}
+	}
+	_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK], p);
+	cmm_smp_wmb();	/* Publish new pointer before head */
+			/* Write q[] before head. */
+	CMM_STORE_SHARED(defer_queue.head, head);
+	cmm_smp_mb();	/* Write queue head before read futex */
+	/*
+	 * Wake-up any waiting defer thread.
+	 */
+	wake_up_defer();
+}
+
+void *thr_defer(void *args)
+{
+	for (;;) {
+		pthread_testcancel();
+		/*
+		 * "Be green". Don't wake up the CPU if there is no RCU work
+		 * to perform whatsoever. Aims at saving laptop battery life by
+		 * leaving the processor in sleep state when idle.
+		 */
+		wait_defer();
+		/* Sleeping after wait_defer to let many callbacks enqueue */
+		poll(NULL,0,100);	/* wait for 100ms */
+		rcu_defer_barrier();
+	}
+
+	return NULL;
+}
+
+/*
+ * library wrappers to be used by non-LGPL compatible source code.
+ */
+
+void defer_rcu(void (*fct)(void *p), void *p)
+{
+	_defer_rcu(fct, p);
+}
+
+static void start_defer_thread(void)
+{
+	int ret;
+
+	ret = pthread_create(&tid_defer, NULL, thr_defer, NULL);
+	assert(!ret);
+}
+
+static void stop_defer_thread(void)
+{
+	int ret;
+	void *tret;
+
+	pthread_cancel(tid_defer);
+	wake_up_defer();
+	ret = pthread_join(tid_defer, &tret);
+	assert(!ret);
+}
+
+int rcu_defer_register_thread(void)
+{
+	int was_empty;
+
+	assert(defer_queue.last_head == 0);
+	assert(defer_queue.q == NULL);
+	defer_queue.q = malloc(sizeof(void *) * DEFER_QUEUE_SIZE);
+	if (!defer_queue.q)
+		return -ENOMEM;
+
+	mutex_lock_defer(&defer_thread_mutex);
+	mutex_lock_defer(&rcu_defer_mutex);
+	was_empty = cds_list_empty(&registry_defer);
+	cds_list_add(&defer_queue.list, &registry_defer);
+	mutex_unlock(&rcu_defer_mutex);
+
+	if (was_empty)
+		start_defer_thread();
+	mutex_unlock(&defer_thread_mutex);
+	return 0;
+}
+
+void rcu_defer_unregister_thread(void)
+{
+	int is_empty;
+
+	mutex_lock_defer(&defer_thread_mutex);
+	mutex_lock_defer(&rcu_defer_mutex);
+	cds_list_del(&defer_queue.list);
+	_rcu_defer_barrier_thread();
+	free(defer_queue.q);
+	defer_queue.q = NULL;
+	is_empty = cds_list_empty(&registry_defer);
+	mutex_unlock(&rcu_defer_mutex);
+
+	if (is_empty)
+		stop_defer_thread();
+	mutex_unlock(&defer_thread_mutex);
+}
+
+void rcu_defer_exit(void)
+{
+	assert(cds_list_empty(&registry_defer));
+}
+
+#endif /* _URCU_DEFER_IMPL_H */
diff --git a/urcu-defer-static.h b/urcu-defer-static.h
deleted file mode 100644
index 3e5df3e..0000000
--- a/urcu-defer-static.h
+++ /dev/null
@@ -1,108 +0,0 @@
-#ifndef _URCU_DEFER_STATIC_H
-#define _URCU_DEFER_STATIC_H
-
-/*
- * urcu-defer-static.h
- *
- * Userspace RCU header - memory reclamation.
- *
- * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See urcu-defer.h for linking
- * dynamically with the userspace rcu reclamation library.
- *
- * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
- * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
- *
- * IBM's contributions to this file may be relicensed under LGPLv2 or later.
- */
-
-#include <stdlib.h>
-#include <pthread.h>
-
-#include <urcu/compiler.h>
-#include <urcu/arch.h>
-#include <urcu/uatomic_arch.h>
-#include <urcu/list.h>
-#include <urcu/system.h>
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-/*
- * Number of entries in the per-thread defer queue. Must be power of 2.
- */
-#define DEFER_QUEUE_SIZE	(1 << 12)
-#define DEFER_QUEUE_MASK	(DEFER_QUEUE_SIZE - 1)
-
-/*
- * Typically, data is aligned at least on the architecture size.
- * Use lowest bit to indicate that the current callback is changing.
- * Assumes that (void *)-2L is not used often. Used to encode non-aligned
- * functions and non-aligned data using extra space.
- * We encode the (void *)-2L fct as: -2L, fct, data.
- * We encode the (void *)-2L data as: -2L, fct, data.
- * Here, DQ_FCT_MARK == ~DQ_FCT_BIT. Required for the test order.
- */
-#define DQ_FCT_BIT		(1 << 0)
-#define DQ_IS_FCT_BIT(x)	((unsigned long)(x) & DQ_FCT_BIT)
-#define DQ_SET_FCT_BIT(x)	\
-	(x = (void *)((unsigned long)(x) | DQ_FCT_BIT))
-#define DQ_CLEAR_FCT_BIT(x)	\
-	(x = (void *)((unsigned long)(x) & ~DQ_FCT_BIT))
-#define DQ_FCT_MARK		((void *)(~DQ_FCT_BIT))
-
-/*
- * This code section can only be included in LGPL 2.1 compatible source code.
- * See below for the function call wrappers which can be used in code meant to
- * be only linked with the Userspace RCU library. This comes with a small
- * performance degradation on the read-side due to the added function calls.
- * This is required to permit relinking with newer versions of the library.
- */
-
-#ifdef DEBUG_RCU
-#define rcu_assert(args...)	assert(args)
-#else
-#define rcu_assert(args...)
-#endif
-
-/*
- * defer queue.
- * Contains pointers. Encoded to save space when same callback is often used.
- * When looking up the next item:
- * - if DQ_FCT_BIT is set, set the current callback to DQ_CLEAR_FCT_BIT(ptr)
- *   - next element contains pointer to data.
- * - else if item == DQ_FCT_MARK
- *   - set the current callback to next element ptr
- *   - following next element contains pointer to data.
- * - else current element contains data
- */
-struct defer_queue {
-	unsigned long head;	/* add element at head */
-	void *last_fct_in;	/* last fct pointer encoded */
-	unsigned long tail;	/* next element to remove at tail */
-	void *last_fct_out;	/* last fct pointer encoded */
-	void **q;
-	/* registry information */
-	unsigned long last_head;
-	struct cds_list_head list;	/* list of thread queues */
-};
-
-#ifdef __cplusplus 
-}
-#endif
-
-#endif /* _URCU_DEFER_STATIC_H */
diff --git a/urcu-defer.c b/urcu-defer.c
deleted file mode 100644
index 3f596ae..0000000
--- a/urcu-defer.c
+++ /dev/null
@@ -1,385 +0,0 @@
-/*
- * urcu-defer.c
- *
- * Userspace RCU library - batch memory reclamation
- *
- * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
- */
-
-#include <stdio.h>
-#include <pthread.h>
-#include <signal.h>
-#include <assert.h>
-#include <stdlib.h>
-#include <string.h>
-#include <errno.h>
-#include <poll.h>
-#include <sys/time.h>
-#include <syscall.h>
-#include <unistd.h>
-
-#include "urcu/urcu-futex.h"
-#include "urcu-defer-static.h"
-/* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
-#include "urcu-defer.h"
-
-void __attribute__((destructor)) rcu_defer_exit(void);
-
-extern void synchronize_rcu(void);
-
-/*
- * rcu_defer_mutex nests inside defer_thread_mutex.
- */
-static pthread_mutex_t rcu_defer_mutex = PTHREAD_MUTEX_INITIALIZER;
-static pthread_mutex_t defer_thread_mutex = PTHREAD_MUTEX_INITIALIZER;
-
-static int defer_thread_futex;
-
-/*
- * Written to only by each individual deferer. Read by both the deferer and
- * the reclamation tread.
- */
-static struct defer_queue __thread defer_queue;
-static CDS_LIST_HEAD(registry);
-static pthread_t tid_defer;
-
-static void mutex_lock(pthread_mutex_t *mutex)
-{
-	int ret;
-
-#ifndef DISTRUST_SIGNALS_EXTREME
-	ret = pthread_mutex_lock(mutex);
-	if (ret) {
-		perror("Error in pthread mutex lock");
-		exit(-1);
-	}
-#else /* #ifndef DISTRUST_SIGNALS_EXTREME */
-	while ((ret = pthread_mutex_trylock(mutex)) != 0) {
-		if (ret != EBUSY && ret != EINTR) {
-			printf("ret = %d, errno = %d\n", ret, errno);
-			perror("Error in pthread mutex lock");
-			exit(-1);
-		}
-		pthread_testcancel();
-		poll(NULL,0,10);
-	}
-#endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
-}
-
-static void mutex_unlock(pthread_mutex_t *mutex)
-{
-	int ret;
-
-	ret = pthread_mutex_unlock(mutex);
-	if (ret) {
-		perror("Error in pthread mutex unlock");
-		exit(-1);
-	}
-}
-
-/*
- * Wake-up any waiting defer thread. Called from many concurrent threads.
- */
-static void wake_up_defer(void)
-{
-	if (unlikely(uatomic_read(&defer_thread_futex) == -1)) {
-		uatomic_set(&defer_thread_futex, 0);
-		futex_noasync(&defer_thread_futex, FUTEX_WAKE, 1,
-		      NULL, NULL, 0);
-	}
-}
-
-static unsigned long rcu_defer_num_callbacks(void)
-{
-	unsigned long num_items = 0, head;
-	struct defer_queue *index;
-
-	mutex_lock(&rcu_defer_mutex);
-	cds_list_for_each_entry(index, &registry, list) {
-		head = CMM_LOAD_SHARED(index->head);
-		num_items += head - index->tail;
-	}
-	mutex_unlock(&rcu_defer_mutex);
-	return num_items;
-}
-
-/*
- * Defer thread waiting. Single thread.
- */
-static void wait_defer(void)
-{
-	uatomic_dec(&defer_thread_futex);
-	cmm_smp_mb();	/* Write futex before read queue */
-	if (rcu_defer_num_callbacks()) {
-		cmm_smp_mb();	/* Read queue before write futex */
-		/* Callbacks are queued, don't wait. */
-		uatomic_set(&defer_thread_futex, 0);
-	} else {
-		cmm_smp_rmb();	/* Read queue before read futex */
-		if (uatomic_read(&defer_thread_futex) == -1)
-			futex_noasync(&defer_thread_futex, FUTEX_WAIT, -1,
-			      NULL, NULL, 0);
-	}
-}
-
-/*
- * Must be called after Q.S. is reached.
- */
-static void rcu_defer_barrier_queue(struct defer_queue *queue,
-				    unsigned long head)
-{
-	unsigned long i;
-	void (*fct)(void *p);
-	void *p;
-
-	/*
-	 * Tail is only modified when lock is held.
-	 * Head is only modified by owner thread.
-	 */
-
-	for (i = queue->tail; i != head;) {
-		cmm_smp_rmb();       /* read head before q[]. */
-		p = CMM_LOAD_SHARED(queue->q[i++ & DEFER_QUEUE_MASK]);
-		if (unlikely(DQ_IS_FCT_BIT(p))) {
-			DQ_CLEAR_FCT_BIT(p);
-			queue->last_fct_out = p;
-			p = CMM_LOAD_SHARED(queue->q[i++ & DEFER_QUEUE_MASK]);
-		} else if (unlikely(p == DQ_FCT_MARK)) {
-			p = CMM_LOAD_SHARED(queue->q[i++ & DEFER_QUEUE_MASK]);
-			queue->last_fct_out = p;
-			p = CMM_LOAD_SHARED(queue->q[i++ & DEFER_QUEUE_MASK]);
-		}
-		fct = queue->last_fct_out;
-		fct(p);
-	}
-	cmm_smp_mb();	/* push tail after having used q[] */
-	CMM_STORE_SHARED(queue->tail, i);
-}
-
-static void _rcu_defer_barrier_thread(void)
-{
-	unsigned long head, num_items;
-
-	head = defer_queue.head;
-	num_items = head - defer_queue.tail;
-	if (unlikely(!num_items))
-		return;
-	synchronize_rcu();
-	rcu_defer_barrier_queue(&defer_queue, head);
-}
-
-void rcu_defer_barrier_thread(void)
-{
-	mutex_lock(&rcu_defer_mutex);
-	_rcu_defer_barrier_thread();
-	mutex_unlock(&rcu_defer_mutex);
-}
-
-/*
- * rcu_defer_barrier - Execute all queued rcu callbacks.
- *
- * Execute all RCU callbacks queued before rcu_defer_barrier() execution.
- * All callbacks queued on the local thread prior to a rcu_defer_barrier() call
- * are guaranteed to be executed.
- * Callbacks queued by other threads concurrently with rcu_defer_barrier()
- * execution are not guaranteed to be executed in the current batch (could
- * be left for the next batch). These callbacks queued by other threads are only
- * guaranteed to be executed if there is explicit synchronization between
- * the thread adding to the queue and the thread issuing the defer_barrier call.
- */
-
-void rcu_defer_barrier(void)
-{
-	struct defer_queue *index;
-	unsigned long num_items = 0;
-
-	if (cds_list_empty(&registry))
-		return;
-
-	mutex_lock(&rcu_defer_mutex);
-	cds_list_for_each_entry(index, &registry, list) {
-		index->last_head = CMM_LOAD_SHARED(index->head);
-		num_items += index->last_head - index->tail;
-	}
-	if (likely(!num_items)) {
-		/*
-		 * We skip the grace period because there are no queued
-		 * callbacks to execute.
-		 */
-		goto end;
-	}
-	synchronize_rcu();
-	cds_list_for_each_entry(index, &registry, list)
-		rcu_defer_barrier_queue(index, index->last_head);
-end:
-	mutex_unlock(&rcu_defer_mutex);
-}
-
-/*
- * _defer_rcu - Queue a RCU callback.
- */
-void _defer_rcu(void (*fct)(void *p), void *p)
-{
-	unsigned long head, tail;
-
-	/*
-	 * Head is only modified by ourself. Tail can be modified by reclamation
-	 * thread.
-	 */
-	head = defer_queue.head;
-	tail = CMM_LOAD_SHARED(defer_queue.tail);
-
-	/*
-	 * If queue is full, or reached threshold. Empty queue ourself.
-	 * Worse-case: must allow 2 supplementary entries for fct pointer.
-	 */
-	if (unlikely(head - tail >= DEFER_QUEUE_SIZE - 2)) {
-		assert(head - tail <= DEFER_QUEUE_SIZE);
-		rcu_defer_barrier_thread();
-		assert(head - CMM_LOAD_SHARED(defer_queue.tail) == 0);
-	}
-
-	if (unlikely(defer_queue.last_fct_in != fct)) {
-		defer_queue.last_fct_in = fct;
-		if (unlikely(DQ_IS_FCT_BIT(fct) || fct == DQ_FCT_MARK)) {
-			/*
-			 * If the function to encode is not aligned or the
-			 * marker, write DQ_FCT_MARK followed by the function
-			 * pointer.
-			 */
-			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
-				      DQ_FCT_MARK);
-			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
-				      fct);
-		} else {
-			DQ_SET_FCT_BIT(fct);
-			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
-				      fct);
-		}
-	} else {
-		if (unlikely(DQ_IS_FCT_BIT(p) || p == DQ_FCT_MARK)) {
-			/*
-			 * If the data to encode is not aligned or the marker,
-			 * write DQ_FCT_MARK followed by the function pointer.
-			 */
-			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
-				      DQ_FCT_MARK);
-			_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK],
-				      fct);
-		}
-	}
-	_CMM_STORE_SHARED(defer_queue.q[head++ & DEFER_QUEUE_MASK], p);
-	cmm_smp_wmb();	/* Publish new pointer before head */
-			/* Write q[] before head. */
-	CMM_STORE_SHARED(defer_queue.head, head);
-	cmm_smp_mb();	/* Write queue head before read futex */
-	/*
-	 * Wake-up any waiting defer thread.
-	 */
-	wake_up_defer();
-}
-
-void *thr_defer(void *args)
-{
-	for (;;) {
-		pthread_testcancel();
-		/*
-		 * "Be green". Don't wake up the CPU if there is no RCU work
-		 * to perform whatsoever. Aims at saving laptop battery life by
-		 * leaving the processor in sleep state when idle.
-		 */
-		wait_defer();
-		/* Sleeping after wait_defer to let many callbacks enqueue */
-		poll(NULL,0,100);	/* wait for 100ms */
-		rcu_defer_barrier();
-	}
-
-	return NULL;
-}
-
-/*
- * library wrappers to be used by non-LGPL compatible source code.
- */
-
-void defer_rcu(void (*fct)(void *p), void *p)
-{
-	_defer_rcu(fct, p);
-}
-
-static void start_defer_thread(void)
-{
-	int ret;
-
-	ret = pthread_create(&tid_defer, NULL, thr_defer, NULL);
-	assert(!ret);
-}
-
-static void stop_defer_thread(void)
-{
-	int ret;
-	void *tret;
-
-	pthread_cancel(tid_defer);
-	wake_up_defer();
-	ret = pthread_join(tid_defer, &tret);
-	assert(!ret);
-}
-
-int rcu_defer_register_thread(void)
-{
-	int was_empty;
-
-	assert(defer_queue.last_head == 0);
-	assert(defer_queue.q == NULL);
-	defer_queue.q = malloc(sizeof(void *) * DEFER_QUEUE_SIZE);
-	if (!defer_queue.q)
-		return -ENOMEM;
-
-	mutex_lock(&defer_thread_mutex);
-	mutex_lock(&rcu_defer_mutex);
-	was_empty = cds_list_empty(&registry);
-	cds_list_add(&defer_queue.list, &registry);
-	mutex_unlock(&rcu_defer_mutex);
-
-	if (was_empty)
-		start_defer_thread();
-	mutex_unlock(&defer_thread_mutex);
-	return 0;
-}
-
-void rcu_defer_unregister_thread(void)
-{
-	int is_empty;
-
-	mutex_lock(&defer_thread_mutex);
-	mutex_lock(&rcu_defer_mutex);
-	cds_list_del(&defer_queue.list);
-	_rcu_defer_barrier_thread();
-	free(defer_queue.q);
-	defer_queue.q = NULL;
-	is_empty = cds_list_empty(&registry);
-	mutex_unlock(&rcu_defer_mutex);
-
-	if (is_empty)
-		stop_defer_thread();
-	mutex_unlock(&defer_thread_mutex);
-}
-
-void rcu_defer_exit(void)
-{
-	assert(cds_list_empty(&registry));
-}
diff --git a/urcu-map.h b/urcu-map.h
index eccaac2..93ac65c 100644
--- a/urcu-map.h
+++ b/urcu-map.h
@@ -34,6 +34,10 @@
 
 /* Mapping macros to allow multiple flavors in a single binary. */
 
+#if !defined(RCU_MEMBARRIER) && !defined(RCU_SIGNAL) && !defined(RCU_MB)
+#define RCU_MB
+#endif
+
 #ifdef RCU_MEMBARRIER
 
 #define rcu_read_lock			rcu_read_lock_memb
@@ -60,6 +64,12 @@
 #define free_all_cpu_call_rcu_data	free_all_cpu_call_rcu_data_memb
 #define call_rcu			call_rcu_memb
 
+#define defer_rcu			defer_rcu_memb
+#define rcu_defer_register_thread	rcu_defer_register_thread_memb
+#define rcu_defer_unregister_thread	rcu_defer_unregister_thread_memb
+#define	rcu_defer_barrier		rcu_defer_barrier_memb
+#define rcu_defer_barrier_thread	rcu_defer_barrier_thread_memb
+
 #elif defined(RCU_SIGNAL)
 
 #define rcu_read_lock			rcu_read_lock_sig
@@ -86,6 +96,12 @@
 #define free_all_cpu_call_rcu_data	free_all_cpu_call_rcu_data_sig
 #define call_rcu			call_rcu_sig
 
+#define defer_rcu			defer_rcu_sig
+#define rcu_defer_register_thread	rcu_defer_register_thread_sig
+#define rcu_defer_unregister_thread	rcu_defer_unregister_thread_sig
+#define	rcu_defer_barrier		rcu_defer_barrier_sig
+#define rcu_defer_barrier_thread	rcu_defer_barrier_thread_sig
+
 #elif defined(RCU_MB)
 
 #define rcu_read_lock			rcu_read_lock_mb
@@ -112,6 +128,16 @@
 #define free_all_cpu_call_rcu_data	free_all_cpu_call_rcu_data_mb
 #define call_rcu			call_rcu_mb
 
+#define defer_rcu			defer_rcu_mb
+#define rcu_defer_register_thread	rcu_defer_register_thread_mb
+#define rcu_defer_unregister_thread	rcu_defer_unregister_thread_mb
+#define	rcu_defer_barrier		rcu_defer_barrier_mb
+#define rcu_defer_barrier_thread	rcu_defer_barrier_thread_mb
+
+#else
+
+#error "Undefined selection"
+
 #endif
 
 #endif /* _URCU_MAP_H */
diff --git a/urcu-qsbr-map.h b/urcu-qsbr-map.h
index 2bce1b6..0d88d83 100644
--- a/urcu-qsbr-map.h
+++ b/urcu-qsbr-map.h
@@ -60,4 +60,10 @@
 #define create_all_cpu_call_rcu_data	create_all_cpu_call_rcu_data_qsbr
 #define call_rcu			call_rcu_qsbr
 
+#define defer_rcu			defer_rcu_qsbr
+#define rcu_defer_register_thread	rcu_defer_register_thread_qsbr
+#define rcu_defer_unregister_thread	rcu_defer_unregister_thread_qsbr
+#define	rcu_defer_barrier		rcu_defer_barrier_qsbr
+#define rcu_defer_barrier_thread	rcu_defer_barrier_thread_qsbr
+
 #endif /* _URCU_QSBR_MAP_H */
diff --git a/urcu-qsbr.c b/urcu-qsbr.c
index 8dcad33..cf8b5ce 100644
--- a/urcu-qsbr.c
+++ b/urcu-qsbr.c
@@ -334,3 +334,4 @@ void rcu_exit(void)
 }
 
 #include "urcu-call-rcu-impl.h"
+#include "urcu-defer-impl.h"
diff --git a/urcu-qsbr.h b/urcu-qsbr.h
index 7ef1bfe..a691c52 100644
--- a/urcu-qsbr.h
+++ b/urcu-qsbr.h
@@ -125,5 +125,6 @@ extern void rcu_unregister_thread(void);
 #endif
 
 #include "urcu-call-rcu.h"
+#include "urcu-defer.h"
 
 #endif /* _URCU_QSBR_H */
diff --git a/urcu.c b/urcu.c
index 4ee9e3b..d356f54 100644
--- a/urcu.c
+++ b/urcu.c
@@ -434,3 +434,4 @@ void rcu_exit(void)
 #endif /* #ifdef RCU_SIGNAL */
 
 #include "urcu-call-rcu-impl.h"
+#include "urcu-defer-impl.h"
diff --git a/urcu.h b/urcu.h
index 417e609..15c8c38 100644
--- a/urcu.h
+++ b/urcu.h
@@ -111,5 +111,6 @@ extern void rcu_init(void);
 #endif
 
 #include "urcu-call-rcu.h"
+#include "urcu-defer.h"
 
 #endif /* _URCU_H */
-- 
1.7.3.2

[ltt-dev] [PATCH urcu 5/5] Document the new call_rcu() primitives.

[Paul E. McKenney]

Signed-off-by: Paul E. McKenney <paulmck at linux.vnet.ibm.com>
---
 API.txt |  120 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 README  |   16 ++++++++
 2 files changed, 136 insertions(+), 0 deletions(-)
 create mode 100644 API.txt

diff --git a/API.txt b/API.txt
new file mode 100644
index 0000000..162ca93
--- /dev/null
+++ b/API.txt
@@ -0,0 +1,120 @@
+Userspace RCU API
+by Mathieu Desnoyers and Paul E. McKenney
+
+
+void rcu_init(void);
+
+	This must be called before any of the following functions
+	are invoked.
+
+void rcu_read_lock(void);
+
+	Begin an RCU read-side critical section.  These critical
+	sections may be nested.
+
+void rcu_read_unlock(void);
+
+	End an RCU read-side critical section.
+
+void rcu_register_thread(void)
+
+	Each thread must invoke this function before its first call to
+	rcu_read_lock().  Threads that never call rcu_read_lock() need
+	not invoke this function.  In addition, rcu-bp ("bullet proof"
+	RCU) does not require any thread to invoke rcu_register_thread().
+
+void rcu_unregister_thread(void)
+
+	Each thread that invokes rcu_register_thread() must invoke
+	rcu_unregister_thread() before invoking pthread_exit()
+	or before returning from its top-level function.
+
+void synchronize_rcu(void);
+
+	Wait until every pre-existing RCU read-side critical section
+	has completed.  Note that this primitive will not necessarily
+	wait for RCU read-side critical sections that have not yet
+	started: this is not a reader-writer lock.  The duration
+	actually waited is called an RCU grace period.
+
+void call_rcu(struct rcu_head *head,
+	      void (*func)(struct rcu_head *head));
+
+	Registers the callback indicated by "head".  This means
+	that "func" will be invoked after the end of a future
+	RCU grace period.  The rcu_head structure referenced
+	by "head" will normally be a field in a larger RCU-protected
+	structure.  A typical implementation of "func" is as
+	follows:
+
+		void func(struct rcu_head *head)
+		{
+			struct foo *p = container_of(head, struct foo, rcu);
+
+			free(p);
+		}
+
+	This RCU callback function can be registered as follows
+	given a pointer "p" to the enclosing structure:
+
+		call_rcu(&p->rcu, func);
+
+struct call_rcu_data *create_call_rcu_data(unsigned long flags);
+
+	Returns a handle that can be passed to the following
+	primitives.  The "flags" argument can be zero, or can be
+	URCU_CALL_RCU_RT if the worker threads associated with the
+	new helper thread are to get real-time response.
+
+struct call_rcu_data *get_default_call_rcu_data(void);
+
+	Returns the handle of the default call_rcu() helper thread.
+
+struct call_rcu_data *get_call_rcu_data(void);
+
+	Returns the handle of the current thread's call_rcu() helper
+	thread, which might well be the default helper thread.
+
+struct call_rcu_data *get_thread_call_rcu_data(void);
+
+	Returns the handle for the current thread's hard-assigned
+	call_rcu() helper thread, or NULL if the current thread is
+	instead using a per-CPU or the default helper thread.
+
+void set_thread_call_rcu_data(struct call_rcu_data *crdp);
+
+	Sets the current thread's hard-assigned call_rcu() helper to the
+	handle specified by "crdp".  Note that "crdp" can be NULL to
+	disassociate this thread from its helper.  Once a thread is
+	disassociated from its helper, further call_rcu() invocations
+	use the current CPU's helper if there is one and the default
+	helper otherwise.
+
+int set_cpu_call_rcu_data(int cpu, struct call_rcu_data *crdp);
+
+	Sets the specified CPU's call_rcu() helper to the handle
+	specified by "crdp".  Again, "crdp" can be NULL to disassociate
+	this CPU from its helper thread.  Once a CPU has been
+	disassociated from its helper, further call_rcu() invocations
+	that would otherwise have used this CPU's helper will instead
+	use the default helper.
+
+int create_all_cpu_call_rcu_data(unsigned long flags)
+
+	Creates a separate call_rcu() helper thread for each CPU.
+	After this primitive is invoked, the global default call_rcu()
+	helper thread will not be called.
+
+	The set_thread_call_rcu_data(), set_cpu_call_rcu_data(), and
+	create_all_cpu_call_rcu_data() functions may be combined to set up
+	pretty much any desired association between worker and call_rcu()
+	helper threads.  If a given executable calls only call_rcu(),
+	then that executable will have only the single global default
+	call_rcu() helper thread.  This will suffice in most cases.
+
+void call_rcu_data_free(struct call_rcu_data *crdp)
+
+	Terminates a call_rcu() helper thread and frees its associated
+	data.  The caller must have ensured that this thread is no longer
+	in use, for example, by passing NULL to set_thread_call_rcu_data()
+	and set_cpu_call_rcu_data() as required.
diff --git a/README b/README
index 7d97f19..a6e29c8 100644
--- a/README
+++ b/README
@@ -139,6 +139,22 @@ Usage of liburcu-defer
 	* Its API is currently experimental. It may change in future library
 	  releases.
 
+Usage of urcu-call-rcu
+
+	* Follow instructions for either liburcu, liburcu-qsbr,
+	  liburcu-mb, liburcu-signal, or liburcu-bp above.
+	  The urcu-call-rcu functionality is provided for each of
+	  these library modules.
+	* Provides the call_rcu() primitive to enqueue delayed callbacks
+	  in a manner similar to defer_rcu(), but without ever delaying
+	  for a grace period.  On the other hand, call_rcu()'s best-case
+	  overhead is not quite as good as that of defer_rcu().
+	* Provides call_rcu() to allow asynchronous handling of RCU
+	  grace periods.  A number of additional functions are provided
+	  to manage the helper threads used by call_rcu(), but reasonable
+	  defaults are used if these additional functions are not invoked.
+	  See API.txt for more details.
+
 Being careful with signals
 
 	The liburcu library uses signals internally. The signal handler is
-- 
1.7.3.2

[ltt-dev] [PATCH userspace-rcu] Allow multiple flavors of RCU in same executable

[Mathieu Desnoyers]

* Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> Hello!
> 
> This series of patches permits a single executable to use multiple
> flavors of userspace RCU.  This is important, as it allows userspace
> RCU to be used freely by multiple libraries without requiring that
> the multiple authors of this library code to come to agreement about
> which flavor of userspace RCU they will all use.
> 
> This is updated from the version posted on May 31st to add documentation
> and to make the child atfork interface use unlocking rather than
> re-initialization.

All merged and pushed, thanks Paul !

Mathieu

> 
> 							Thanx, Paul
> 
> ------------------------------------------------------------------------
> 
>  Makefile.am            |    4 
>  README                 |   36 ++
>  a/urcu-call-rcu.c      |  617 -------------------------------------------------
>  b/API.txt              |  120 +++++++++
>  b/Makefile.am          |    3 
>  b/README               |    4 
>  b/tests/Makefile.am    |   33 +-
>  b/tests/rcutorture.h   |    1 
>  b/tests/urcutorture.c  |   13 -
>  b/urcu-bp-map.h        |   61 ++++
>  b/urcu-bp.c            |    4 
>  b/urcu-bp.h            |    8 
>  b/urcu-call-rcu-impl.h |  615 ++++++++++++++++++++++++++++++++++++++++++++++++
>  b/urcu-call-rcu.c      |   26 +-
>  b/urcu-defer-impl.h    |  450 +++++++++++++++++++++++++++++++++++
>  b/urcu-map.h           |  117 +++++++++
>  b/urcu-qsbr-map.h      |   63 +++++
>  b/urcu-qsbr.c          |   36 +-
>  b/urcu-qsbr.h          |   18 -
>  b/urcu.c               |    5 
>  b/urcu.h               |   20 +
>  tests/Makefile.am      |    2 
>  urcu-bp-map.h          |    6 
>  urcu-bp.c              |    1 
>  urcu-bp.h              |    5 
>  urcu-defer-static.h    |  108 --------
>  urcu-defer.c           |  385 ------------------------------
>  urcu-map.h             |   26 ++
>  urcu-qsbr-map.h        |    6 
>  urcu-qsbr.c            |    1 
>  urcu-qsbr.h            |   11 
>  urcu.c                 |    1 
>  urcu.h                 |   13 -
>  33 files changed, 1627 insertions(+), 1192 deletions(-)
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com