[urcu.git] / urcu-qsbr.c

/*
 * urcu-qsbr.c
 *
 * Userspace RCU QSBR library
 *
 * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers@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.
 */

#define _GNU_SOURCE
#include <stdio.h>
#include <pthread.h>
#include <signal.h>
#include <assert.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <errno.h>
#include <poll.h>

#include "urcu/map/urcu-qsbr.h"

#define BUILD_QSBR_LIB
#include "urcu/static/urcu-qsbr.h"
/* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
#include "urcu-qsbr.h"

void __attribute__((destructor)) rcu_exit(void);

static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;

int32_t gp_futex;

/*
 * Global grace period counter.
 */
unsigned long rcu_gp_ctr = RCU_GP_ONLINE;

/*
 * Written to only by each individual reader. Read by both the reader and the
 * writers.
 */
struct rcu_reader __thread rcu_reader;

#ifdef DEBUG_YIELD
unsigned int yield_active;
unsigned int __thread rand_yield;
#endif

static CDS_LIST_HEAD(registry);

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);
		}
		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);
	}
}

/*
 * synchronize_rcu() waiting. Single thread.
 */
static void wait_gp(void)
{
	/* Read reader_gp before read futex */
	cmm_smp_rmb();
	if (uatomic_read(&gp_futex) == -1)
		futex_noasync(&gp_futex, FUTEX_WAIT, -1,
		      NULL, NULL, 0);
}

static void update_counter_and_wait(void)
{
	CDS_LIST_HEAD(qsreaders);
	int wait_loops = 0;
	struct rcu_reader *index, *tmp;

#if (CAA_BITS_PER_LONG < 64)
	/* Switch parity: 0 -> 1, 1 -> 0 */
	CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR);
#else	/* !(CAA_BITS_PER_LONG < 64) */
	/* Increment current G.P. */
	CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr + RCU_GP_CTR);
#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.
	 */
	cmm_barrier();

	/*
	 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
	 * model easier to understand. It does not have a big performance impact
	 * anyway, given this is the write-side.
	 */
	cmm_smp_mb();

	/*
	 * Wait for each thread rcu_reader_qs_gp count to become 0.
	 */
	for (;;) {
		wait_loops++;
		if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
			uatomic_dec(&gp_futex);
			/* Write futex before read reader_gp */
			cmm_smp_mb();
		}

		cds_list_for_each_entry_safe(index, tmp, &registry, node) {
			if (!rcu_gp_ongoing(&index->ctr))
				cds_list_move(&index->node, &qsreaders);
		}

		if (cds_list_empty(&registry)) {
			if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
				/* Read reader_gp before write futex */
				cmm_smp_mb();
				uatomic_set(&gp_futex, 0);
			}
			break;
		} else {
			if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
				wait_gp();
			} else {
#ifndef HAS_INCOHERENT_CACHES
				caa_cpu_relax();
#else /* #ifndef HAS_INCOHERENT_CACHES */
				cmm_smp_mb();
#endif /* #else #ifndef HAS_INCOHERENT_CACHES */
			}
		}
	}
	/* put back the reader list in the registry */
	cds_list_splice(&qsreaders, &registry);
}

/*
 * Using a two-subphases algorithm for architectures with smaller than 64-bit
 * long-size to ensure we do not encounter an overflow bug.
 */

#if (CAA_BITS_PER_LONG < 64)
void synchronize_rcu(void)
{
	unsigned long was_online;

	was_online = rcu_reader.ctr;

	/* All threads should read qparity before accessing data structure
	 * where new ptr points to.
	 */
	/* Write new ptr before changing the qparity */
	cmm_smp_mb();

	/*
	 * 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.
	 */
	if (was_online)
		CMM_STORE_SHARED(rcu_reader.ctr, 0);

	mutex_lock(&rcu_gp_lock);

	if (cds_list_empty(&registry))
		goto out;

	/*
	 * Wait for previous parity to be empty of readers.
	 */
	update_counter_and_wait();	/* 0 -> 1, wait readers in parity 0 */

	/*
	 * 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.
	 */
	cmm_barrier();

	/*
	 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
	 * model easier to understand. It does not have a big performance impact
	 * anyway, given this is the write-side.
	 */
	cmm_smp_mb();

	/*
	 * Wait for previous parity to be empty of readers.
	 */
	update_counter_and_wait();	/* 1 -> 0, wait readers in parity 1 */
out:
	mutex_unlock(&rcu_gp_lock);

	/*
	 * Finish waiting for reader threads before letting the old ptr being
	 * freed.
	 */
	if (was_online)
		_CMM_STORE_SHARED(rcu_reader.ctr,
				  CMM_LOAD_SHARED(rcu_gp_ctr));
	cmm_smp_mb();
}
#else /* !(CAA_BITS_PER_LONG < 64) */
void synchronize_rcu(void)
{
	unsigned long was_online;

	was_online = rcu_reader.ctr;

	/*
	 * 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.
	 */
	cmm_smp_mb();
	if (was_online)
		CMM_STORE_SHARED(rcu_reader.ctr, 0);

	mutex_lock(&rcu_gp_lock);
	if (cds_list_empty(&registry))
		goto out;
	update_counter_and_wait();
out:
	mutex_unlock(&rcu_gp_lock);

	if (was_online)
		_CMM_STORE_SHARED(rcu_reader.ctr,
				  CMM_LOAD_SHARED(rcu_gp_ctr));
	cmm_smp_mb();
}
#endif  /* !(CAA_BITS_PER_LONG < 64) */

/*
 * library wrappers to be used by non-LGPL compatible source code.
 */

void rcu_read_lock(void)
{
	_rcu_read_lock();
}

void rcu_read_unlock(void)
{
	_rcu_read_unlock();
}

void rcu_quiescent_state(void)
{
	_rcu_quiescent_state();
}

void rcu_thread_offline(void)
{
	_rcu_thread_offline();
}

void rcu_thread_online(void)
{
	_rcu_thread_online();
}

void rcu_register_thread(void)
{
	rcu_reader.tid = pthread_self();
	assert(rcu_reader.ctr == 0);

	mutex_lock(&rcu_gp_lock);
	cds_list_add(&rcu_reader.node, &registry);
	mutex_unlock(&rcu_gp_lock);
	_rcu_thread_online();
}

void rcu_unregister_thread(void)
{
	/*
	 * We have to make the thread offline otherwise we end up dealocking
	 * with a waiting writer.
	 */
	_rcu_thread_offline();
	mutex_lock(&rcu_gp_lock);
	cds_list_del(&rcu_reader.node);
	mutex_unlock(&rcu_gp_lock);
}

void rcu_exit(void)
{
	assert(cds_list_empty(&registry));
}

#include "urcu-call-rcu-impl.h"
#include "urcu-defer-impl.h"
Commit	Line	Data
	1	/*
	2	* urcu-qsbr.c
	3	*
	4	* Userspace RCU QSBR library
	5	*
	6	* Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	7	* Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
	8	*
	9	* This library is free software; you can redistribute it and/or
	10	* modify it under the terms of the GNU Lesser General Public
	11	* License as published by the Free Software Foundation; either
	12	* version 2.1 of the License, or (at your option) any later version.
	13	*
	14	* This library is distributed in the hope that it will be useful,
	15	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	16	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	17	* Lesser General Public License for more details.
	18	*
	19	* You should have received a copy of the GNU Lesser General Public
	20	* License along with this library; if not, write to the Free Software
	21	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
	22	*
	23	* IBM's contributions to this file may be relicensed under LGPLv2 or later.
	24	*/
	25
	26	#define _GNU_SOURCE
	27	#include <stdio.h>
	28	#include <pthread.h>
	29	#include <signal.h>
	30	#include <assert.h>
	31	#include <stdlib.h>
	32	#include <stdint.h>
	33	#include <string.h>
	34	#include <errno.h>
	35	#include <poll.h>
	36
	37	#include "urcu/map/urcu-qsbr.h"
	38
	39	#define BUILD_QSBR_LIB
	40	#include "urcu/static/urcu-qsbr.h"
	41	/* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
	42	#include "urcu-qsbr.h"
	43
	44	void __attribute__((destructor)) rcu_exit(void);
	45
	46	static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
	47
	48	int32_t gp_futex;
	49
	50	/*
	51	* Global grace period counter.
	52	*/
	53	unsigned long rcu_gp_ctr = RCU_GP_ONLINE;
	54
	55	/*
	56	* Written to only by each individual reader. Read by both the reader and the
	57	* writers.
	58	*/
	59	struct rcu_reader __thread rcu_reader;
	60
	61	#ifdef DEBUG_YIELD
	62	unsigned int yield_active;
	63	unsigned int __thread rand_yield;
	64	#endif
	65
	66	static CDS_LIST_HEAD(registry);
	67
	68	static void mutex_lock(pthread_mutex_t *mutex)
	69	{
	70	int ret;
	71
	72	#ifndef DISTRUST_SIGNALS_EXTREME
	73	ret = pthread_mutex_lock(mutex);
	74	if (ret) {
	75	perror("Error in pthread mutex lock");
	76	exit(-1);
	77	}
	78	#else /* #ifndef DISTRUST_SIGNALS_EXTREME */
	79	while ((ret = pthread_mutex_trylock(mutex)) != 0) {
	80	if (ret != EBUSY && ret != EINTR) {
	81	printf("ret = %d, errno = %d\n", ret, errno);
	82	perror("Error in pthread mutex lock");
	83	exit(-1);
	84	}
	85	poll(NULL,0,10);
	86	}
	87	#endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
	88	}
	89
	90	static void mutex_unlock(pthread_mutex_t *mutex)
	91	{
	92	int ret;
	93
	94	ret = pthread_mutex_unlock(mutex);
	95	if (ret) {
	96	perror("Error in pthread mutex unlock");
	97	exit(-1);
	98	}
	99	}
	100
	101	/*
	102	* synchronize_rcu() waiting. Single thread.
	103	*/
	104	static void wait_gp(void)
	105	{
	106	/* Read reader_gp before read futex */
	107	cmm_smp_rmb();
	108	if (uatomic_read(&gp_futex) == -1)
	109	futex_noasync(&gp_futex, FUTEX_WAIT, -1,
	110	NULL, NULL, 0);
	111	}
	112
	113	static void update_counter_and_wait(void)
	114	{
	115	CDS_LIST_HEAD(qsreaders);
	116	int wait_loops = 0;
	117	struct rcu_reader index, tmp;
	118
	119	#if (CAA_BITS_PER_LONG < 64)
	120	/* Switch parity: 0 -> 1, 1 -> 0 */
	121	CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR);
	122	#else /* !(CAA_BITS_PER_LONG < 64) */
	123	/* Increment current G.P. */
	124	CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr + RCU_GP_CTR);
	125	#endif /* !(CAA_BITS_PER_LONG < 64) */
	126
	127	/*
	128	* Must commit rcu_gp_ctr update to memory before waiting for
	129	* quiescent state. Failure to do so could result in the writer
	130	* waiting forever while new readers are always accessing data
	131	* (no progress). Enforce compiler-order of store to rcu_gp_ctr
	132	* before load rcu_reader ctr.
	133	*/
	134	cmm_barrier();
	135
	136	/*
	137	* Adding a cmm_smp_mb() which is _not_ formally required, but makes the
	138	* model easier to understand. It does not have a big performance impact
	139	* anyway, given this is the write-side.
	140	*/
	141	cmm_smp_mb();
	142
	143	/*
	144	* Wait for each thread rcu_reader_qs_gp count to become 0.
	145	*/
	146	for (;;) {
	147	wait_loops++;
	148	if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
	149	uatomic_dec(&gp_futex);
	150	/* Write futex before read reader_gp */
	151	cmm_smp_mb();
	152	}
	153
	154	cds_list_for_each_entry_safe(index, tmp, &registry, node) {
	155	if (!rcu_gp_ongoing(&index->ctr))
	156	cds_list_move(&index->node, &qsreaders);
	157	}
	158
	159	if (cds_list_empty(&registry)) {
	160	if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
	161	/* Read reader_gp before write futex */
	162	cmm_smp_mb();
	163	uatomic_set(&gp_futex, 0);
	164	}
	165	break;
	166	} else {
	167	if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
	168	wait_gp();
	169	} else {
	170	#ifndef HAS_INCOHERENT_CACHES
	171	caa_cpu_relax();
	172	#else /* #ifndef HAS_INCOHERENT_CACHES */
	173	cmm_smp_mb();
	174	#endif /* #else #ifndef HAS_INCOHERENT_CACHES */
	175	}
	176	}
	177	}
	178	/* put back the reader list in the registry */
	179	cds_list_splice(&qsreaders, &registry);
	180	}
	181
	182	/*
	183	* Using a two-subphases algorithm for architectures with smaller than 64-bit
	184	* long-size to ensure we do not encounter an overflow bug.
	185	*/
	186
	187	#if (CAA_BITS_PER_LONG < 64)
	188	void synchronize_rcu(void)
	189	{
	190	unsigned long was_online;
	191
	192	was_online = rcu_reader.ctr;
	193
	194	/* All threads should read qparity before accessing data structure
	195	* where new ptr points to.
	196	*/
	197	/* Write new ptr before changing the qparity */
	198	cmm_smp_mb();
	199
	200	/*
	201	* Mark the writer thread offline to make sure we don't wait for
	202	* our own quiescent state. This allows using synchronize_rcu()
	203	* in threads registered as readers.
	204	*/
	205	if (was_online)
	206	CMM_STORE_SHARED(rcu_reader.ctr, 0);
	207
	208	mutex_lock(&rcu_gp_lock);
	209
	210	if (cds_list_empty(&registry))
	211	goto out;
	212
	213	/*
	214	* Wait for previous parity to be empty of readers.
	215	*/
	216	update_counter_and_wait(); /* 0 -> 1, wait readers in parity 0 */
	217
	218	/*
	219	* Must finish waiting for quiescent state for parity 0 before
	220	* committing next rcu_gp_ctr update to memory. Failure to
	221	* do so could result in the writer waiting forever while new
	222	* readers are always accessing data (no progress). Enforce
	223	* compiler-order of load rcu_reader ctr before store to
	224	* rcu_gp_ctr.
	225	*/
	226	cmm_barrier();
	227
	228	/*
	229	* Adding a cmm_smp_mb() which is _not_ formally required, but makes the
	230	* model easier to understand. It does not have a big performance impact
	231	* anyway, given this is the write-side.
	232	*/
	233	cmm_smp_mb();
	234
	235	/*
	236	* Wait for previous parity to be empty of readers.
	237	*/
	238	update_counter_and_wait(); /* 1 -> 0, wait readers in parity 1 */
	239	out:
	240	mutex_unlock(&rcu_gp_lock);
	241
	242	/*
	243	* Finish waiting for reader threads before letting the old ptr being
	244	* freed.
	245	*/
	246	if (was_online)
	247	_CMM_STORE_SHARED(rcu_reader.ctr,
	248	CMM_LOAD_SHARED(rcu_gp_ctr));
	249	cmm_smp_mb();
	250	}
	251	#else /* !(CAA_BITS_PER_LONG < 64) */
	252	void synchronize_rcu(void)
	253	{
	254	unsigned long was_online;
	255
	256	was_online = rcu_reader.ctr;
	257
	258	/*
	259	* Mark the writer thread offline to make sure we don't wait for
	260	* our own quiescent state. This allows using synchronize_rcu()
	261	* in threads registered as readers.
	262	*/
	263	cmm_smp_mb();
	264	if (was_online)
	265	CMM_STORE_SHARED(rcu_reader.ctr, 0);
	266
	267	mutex_lock(&rcu_gp_lock);
	268	if (cds_list_empty(&registry))
	269	goto out;
	270	update_counter_and_wait();
	271	out:
	272	mutex_unlock(&rcu_gp_lock);
	273
	274	if (was_online)
	275	_CMM_STORE_SHARED(rcu_reader.ctr,
	276	CMM_LOAD_SHARED(rcu_gp_ctr));
	277	cmm_smp_mb();
	278	}
	279	#endif /* !(CAA_BITS_PER_LONG < 64) */
	280
	281	/*
	282	* library wrappers to be used by non-LGPL compatible source code.
	283	*/
	284
	285	void rcu_read_lock(void)
	286	{
	287	_rcu_read_lock();
	288	}
	289
	290	void rcu_read_unlock(void)
	291	{
	292	_rcu_read_unlock();
	293	}
	294
	295	void rcu_quiescent_state(void)
	296	{
	297	_rcu_quiescent_state();
	298	}
	299
	300	void rcu_thread_offline(void)
	301	{
	302	_rcu_thread_offline();
	303	}
	304
	305	void rcu_thread_online(void)
	306	{
	307	_rcu_thread_online();
	308	}
	309
	310	void rcu_register_thread(void)
	311	{
	312	rcu_reader.tid = pthread_self();
	313	assert(rcu_reader.ctr == 0);
	314
	315	mutex_lock(&rcu_gp_lock);
	316	cds_list_add(&rcu_reader.node, &registry);
	317	mutex_unlock(&rcu_gp_lock);
	318	_rcu_thread_online();
	319	}
	320
	321	void rcu_unregister_thread(void)
	322	{
	323	/*
	324	* We have to make the thread offline otherwise we end up dealocking
	325	* with a waiting writer.
	326	*/
	327	_rcu_thread_offline();
	328	mutex_lock(&rcu_gp_lock);
	329	cds_list_del(&rcu_reader.node);
	330	mutex_unlock(&rcu_gp_lock);
	331	}
	332
	333	void rcu_exit(void)
	334	{
	335	assert(cds_list_empty(&registry));
	336	}
	337
	338	#include "urcu-call-rcu-impl.h"
	339	#include "urcu-defer-impl.h"