[userspace-rcu.git] / urcu / waitqueue-lifo.h

#ifndef _URCU_WAITQUEUE_LIFO_H
#define _URCU_WAITQUEUE_LIFO_H

/*
 * urcu/waitqueue-lifo.h
 *
 * Userspace RCU library - wait queue scheme with LIFO semantic
 *
 * Copyright (c) 2012-2014 Mathieu Desnoyers <mathieu.desnoyers@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 <urcu/uatomic.h>
#include <urcu/lfstack.h>
#include <urcu/futex.h>

/*
 * Number of busy-loop attempts before waiting on futex for grace period
 * batching.
 */
#define URCU_WAIT_ATTEMPTS 1000

enum urcu_wait_state {
	/* URCU_WAIT_WAITING is compared directly (futex compares it). */
	URCU_WAIT_WAITING =	0,
	/* non-zero are used as masks. */
	URCU_WAIT_WAKEUP =	(1 << 0),
	URCU_WAIT_RUNNING =	(1 << 1),
	URCU_WAIT_TEARDOWN =	(1 << 2),
};

struct urcu_wait_node {
	struct cds_lfs_node node;
	int32_t state;	/* enum urcu_wait_state */
	int in_waitqueue;
};

#define URCU_WAIT_NODE_INIT(name, _state)		\
	{ .state = _state }

#define DEFINE_URCU_WAIT_NODE(name, state)		\
	struct urcu_wait_node name = URCU_WAIT_NODE_INIT(name, state)

#define DECLARE_URCU_WAIT_NODE(name)			\
	struct urcu_wait_node name

struct urcu_wait_queue {
	struct __cds_lfs_stack stack;
};

#define URCU_WAIT_QUEUE_HEAD_INIT(name)			\
	{ .stack.head = CDS_LFS_END, }

#define DECLARE_URCU_WAIT_QUEUE(name)			\
	struct urcu_wait_queue name

#define DEFINE_URCU_WAIT_QUEUE(name)			\
	struct urcu_wait_queue name = URCU_WAIT_QUEUE_HEAD_INIT(name)

static inline
void urcu_wait_queue_init(struct urcu_wait_queue *queue)
{
	__cds_lfs_init(&queue->stack);
}

struct urcu_waiters {
	struct cds_lfs_head *head;
};

/*
 * Add ourself atomically to a wait queue. Return 0 if queue was
 * previously empty, else return 1.
 * A full memory barrier is issued before being added to the wait queue.
 */
static inline
bool urcu_wait_add(struct urcu_wait_queue *queue,
		struct urcu_wait_node *node)
{
	cds_lfs_node_init(&node->node);
	CMM_STORE_SHARED(node->in_waitqueue, true);
	return cds_lfs_push(&queue->stack, &node->node);
}

/*
 * Atomically move all waiters from wait queue into our local struct
 * urcu_waiters.
 */
static inline
void urcu_move_waiters(struct urcu_waiters *waiters,
		struct urcu_wait_queue *queue)
{
	waiters->head = __cds_lfs_pop_all(&queue->stack);
}

static inline
void urcu_wait_set_state(struct urcu_wait_node *node,
		enum urcu_wait_state state)
{
	node->state = state;
}

static inline
void urcu_wait_or_state(struct urcu_wait_node *node,
		enum urcu_wait_state state)
{
	uatomic_or(&node->state, state);
}

static inline
void urcu_wait_node_init(struct urcu_wait_node *node,
		enum urcu_wait_state state)
{
	urcu_wait_set_state(node, state);
	cds_lfs_node_init(&node->node);
	node->in_waitqueue = false;
}

static inline
bool urcu_in_waitqueue(struct urcu_wait_node *node)
{
	return CMM_LOAD_SHARED(node->in_waitqueue);
}

/*
 * Note: urcu_adaptative_wake_up needs "value" to stay allocated
 * throughout its execution. In this scheme, the waiter owns the node
 * memory, and we only allow it to free this memory when it receives the
 * URCU_WAIT_TEARDOWN flag.
 * Return true if wakeup is performed, false if thread was already
 * running.
 */
static inline
bool urcu_adaptative_wake_up(struct urcu_wait_node *wait)
{
	bool wakeup_performed = false;

	cmm_smp_mb();
	/*
	 * "or" of WAKEUP flag rather than "set" is useful for multiple
	 * concurrent wakeup sources. Note that "WAIT_TEARDOWN" becomes
	 * useless when we use multiple wakeup sources: lifetime of the
	 * "value" should then be handled by the caller.
	 */
	uatomic_or(&wait->state, URCU_WAIT_WAKEUP);
	if (!(uatomic_read(&wait->state) & URCU_WAIT_RUNNING)) {
		futex_noasync(&wait->state, FUTEX_WAKE, 1, NULL, NULL, 0);
		wakeup_performed = true;
	}
	/* Allow teardown of struct urcu_wait memory. */
	uatomic_or(&wait->state, URCU_WAIT_TEARDOWN);
	return wakeup_performed;
}

/*
 * Caller must initialize "value" to URCU_WAIT_WAITING before passing its
 * memory to waker thread.
 */
static inline
void urcu_adaptative_busy_wait(struct urcu_wait_node *wait)
{
	unsigned int i;

	/* Load and test condition before read state */
	cmm_smp_rmb();
	for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
		if (uatomic_read(&wait->state) != URCU_WAIT_WAITING)
			goto skip_futex_wait;
		caa_cpu_relax();
	}
	futex_noasync(&wait->state, FUTEX_WAIT,
		URCU_WAIT_WAITING, NULL, NULL, 0);
skip_futex_wait:

	/* Tell waker thread than we are running. */
	uatomic_or(&wait->state, URCU_WAIT_RUNNING);

	/*
	 * Wait until waker thread lets us know it's ok to tear down
	 * memory allocated for struct urcu_wait.
	 */
	for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
		if (uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN)
			break;
		caa_cpu_relax();
	}
	while (!(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN))
		poll(NULL, 0, 10);
	assert(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN);
}

/*
 * Need mutual exclusion against other wakeup and move waiters
 * operations. It is provided by the caller.
 */
static inline
int urcu_dequeue_wake_single(struct urcu_wait_queue *queue)
{
	struct cds_lfs_node *node;
	struct urcu_wait_node *wait_node;
	int ret = 0;

	node = __cds_lfs_pop(&queue->stack);
	if (!node)
		return -ENOENT;
	wait_node = caa_container_of(node, struct urcu_wait_node, node);
	CMM_STORE_SHARED(wait_node->in_waitqueue, false);
	/* Don't wake already running threads */
	if (!(wait_node->state & URCU_WAIT_RUNNING))
		ret = urcu_adaptative_wake_up(wait_node);
	return ret;
}

/*
 * Need mutual exclusion against other wakeup and move waiters
 * operations. It is provided by the caller.
 */
static inline
int urcu_dequeue_wake_n(struct urcu_wait_queue *queue, int n)
{
	int nr_wakeup = 0;

	for (;;) {
		int ret;

		ret = urcu_dequeue_wake_single(queue);
		if (ret < 0)
			return nr_wakeup;
		else if (ret > 0)
			nr_wakeup++;
		else
			break;
	}
	return nr_wakeup;
}

static inline
int urcu_wake_all_waiters(struct urcu_waiters *waiters)
{
	struct cds_lfs_node *iter, *iter_n;
	int nr_wakeup = 0;

	/* Wake all waiters in our stack head */
	cds_lfs_for_each_safe(waiters->head, iter, iter_n) {
		struct urcu_wait_node *wait_node =
			caa_container_of(iter, struct urcu_wait_node, node);

		CMM_STORE_SHARED(wait_node->in_waitqueue, false);
		/* Don't wake already running threads */
		if (wait_node->state & URCU_WAIT_RUNNING)
			continue;
		if (urcu_adaptative_wake_up(wait_node))
			nr_wakeup++;
	}
	return nr_wakeup;
}

#endif /* _URCU_WAITQUEUE_LIFO_H */
Commit	Line	Data
	1	#ifndef _URCU_WAITQUEUE_LIFO_H
	2	#define _URCU_WAITQUEUE_LIFO_H
	3
	4	/*
	5	* urcu/waitqueue-lifo.h
	6	*
	7	* Userspace RCU library - wait queue scheme with LIFO semantic
	8	*
	9	* Copyright (c) 2012-2014 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	10	*
	11	* This library is free software; you can redistribute it and/or
	12	* modify it under the terms of the GNU Lesser General Public
	13	* License as published by the Free Software Foundation; either
	14	* version 2.1 of the License, or (at your option) any later version.
	15	*
	16	* This library is distributed in the hope that it will be useful,
	17	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	18	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	19	* Lesser General Public License for more details.
	20	*
	21	* You should have received a copy of the GNU Lesser General Public
	22	* License along with this library; if not, write to the Free Software
	23	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
	24	*/
	25
	26	#include <urcu/uatomic.h>
	27	#include <urcu/lfstack.h>
	28	#include <urcu/futex.h>
	29
	30	/*
	31	* Number of busy-loop attempts before waiting on futex for grace period
	32	* batching.
	33	*/
	34	#define URCU_WAIT_ATTEMPTS 1000
	35
	36	enum urcu_wait_state {
	37	/* URCU_WAIT_WAITING is compared directly (futex compares it). */
	38	URCU_WAIT_WAITING = 0,
	39	/* non-zero are used as masks. */
	40	URCU_WAIT_WAKEUP = (1 << 0),
	41	URCU_WAIT_RUNNING = (1 << 1),
	42	URCU_WAIT_TEARDOWN = (1 << 2),
	43	};
	44
	45	struct urcu_wait_node {
	46	struct cds_lfs_node node;
	47	int32_t state; /* enum urcu_wait_state */
	48	int in_waitqueue;
	49	};
	50
	51	#define URCU_WAIT_NODE_INIT(name, _state) \
	52	{ .state = _state }
	53
	54	#define DEFINE_URCU_WAIT_NODE(name, state) \
	55	struct urcu_wait_node name = URCU_WAIT_NODE_INIT(name, state)
	56
	57	#define DECLARE_URCU_WAIT_NODE(name) \
	58	struct urcu_wait_node name
	59
	60	struct urcu_wait_queue {
	61	struct __cds_lfs_stack stack;
	62	};
	63
	64	#define URCU_WAIT_QUEUE_HEAD_INIT(name) \
	65	{ .stack.head = CDS_LFS_END, }
	66
	67	#define DECLARE_URCU_WAIT_QUEUE(name) \
	68	struct urcu_wait_queue name
	69
	70	#define DEFINE_URCU_WAIT_QUEUE(name) \
	71	struct urcu_wait_queue name = URCU_WAIT_QUEUE_HEAD_INIT(name)
	72
	73	static inline
	74	void urcu_wait_queue_init(struct urcu_wait_queue *queue)
	75	{
	76	__cds_lfs_init(&queue->stack);
	77	}
	78
	79	struct urcu_waiters {
	80	struct cds_lfs_head *head;
	81	};
	82
	83	/*
	84	* Add ourself atomically to a wait queue. Return 0 if queue was
	85	* previously empty, else return 1.
	86	* A full memory barrier is issued before being added to the wait queue.
	87	*/
	88	static inline
	89	bool urcu_wait_add(struct urcu_wait_queue *queue,
	90	struct urcu_wait_node *node)
	91	{
	92	cds_lfs_node_init(&node->node);
	93	CMM_STORE_SHARED(node->in_waitqueue, true);
	94	return cds_lfs_push(&queue->stack, &node->node);
	95	}
	96
	97	/*
	98	* Atomically move all waiters from wait queue into our local struct
	99	* urcu_waiters.
	100	*/
	101	static inline
	102	void urcu_move_waiters(struct urcu_waiters *waiters,
	103	struct urcu_wait_queue *queue)
	104	{
	105	waiters->head = __cds_lfs_pop_all(&queue->stack);
	106	}
	107
	108	static inline
	109	void urcu_wait_set_state(struct urcu_wait_node *node,
	110	enum urcu_wait_state state)
	111	{
	112	node->state = state;
	113	}
	114
	115	static inline
	116	void urcu_wait_or_state(struct urcu_wait_node *node,
	117	enum urcu_wait_state state)
	118	{
	119	uatomic_or(&node->state, state);
	120	}
	121
	122	static inline
	123	void urcu_wait_node_init(struct urcu_wait_node *node,
	124	enum urcu_wait_state state)
	125	{
	126	urcu_wait_set_state(node, state);
	127	cds_lfs_node_init(&node->node);
	128	node->in_waitqueue = false;
	129	}
	130
	131	static inline
	132	bool urcu_in_waitqueue(struct urcu_wait_node *node)
	133	{
	134	return CMM_LOAD_SHARED(node->in_waitqueue);
	135	}
	136
	137	/*
	138	* Note: urcu_adaptative_wake_up needs "value" to stay allocated
	139	* throughout its execution. In this scheme, the waiter owns the node
	140	* memory, and we only allow it to free this memory when it receives the
	141	* URCU_WAIT_TEARDOWN flag.
	142	* Return true if wakeup is performed, false if thread was already
	143	* running.
	144	*/
	145	static inline
	146	bool urcu_adaptative_wake_up(struct urcu_wait_node *wait)
	147	{
	148	bool wakeup_performed = false;
	149
	150	cmm_smp_mb();
	151	/*
	152	* "or" of WAKEUP flag rather than "set" is useful for multiple
	153	* concurrent wakeup sources. Note that "WAIT_TEARDOWN" becomes
	154	* useless when we use multiple wakeup sources: lifetime of the
	155	* "value" should then be handled by the caller.
	156	*/
	157	uatomic_or(&wait->state, URCU_WAIT_WAKEUP);
	158	if (!(uatomic_read(&wait->state) & URCU_WAIT_RUNNING)) {
	159	futex_noasync(&wait->state, FUTEX_WAKE, 1, NULL, NULL, 0);
	160	wakeup_performed = true;
	161	}
	162	/* Allow teardown of struct urcu_wait memory. */
	163	uatomic_or(&wait->state, URCU_WAIT_TEARDOWN);
	164	return wakeup_performed;
	165	}
	166
	167	/*
	168	* Caller must initialize "value" to URCU_WAIT_WAITING before passing its
	169	* memory to waker thread.
	170	*/
	171	static inline
	172	void urcu_adaptative_busy_wait(struct urcu_wait_node *wait)
	173	{
	174	unsigned int i;
	175
	176	/* Load and test condition before read state */
	177	cmm_smp_rmb();
	178	for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
	179	if (uatomic_read(&wait->state) != URCU_WAIT_WAITING)
	180	goto skip_futex_wait;
	181	caa_cpu_relax();
	182	}
	183	futex_noasync(&wait->state, FUTEX_WAIT,
	184	URCU_WAIT_WAITING, NULL, NULL, 0);
	185	skip_futex_wait:
	186
	187	/* Tell waker thread than we are running. */
	188	uatomic_or(&wait->state, URCU_WAIT_RUNNING);
	189
	190	/*
	191	* Wait until waker thread lets us know it's ok to tear down
	192	* memory allocated for struct urcu_wait.
	193	*/
	194	for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
	195	if (uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN)
	196	break;
	197	caa_cpu_relax();
	198	}
	199	while (!(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN))
	200	poll(NULL, 0, 10);
	201	assert(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN);
	202	}
	203
	204	/*
	205	* Need mutual exclusion against other wakeup and move waiters
	206	* operations. It is provided by the caller.
	207	*/
	208	static inline
	209	int urcu_dequeue_wake_single(struct urcu_wait_queue *queue)
	210	{
	211	struct cds_lfs_node *node;
	212	struct urcu_wait_node *wait_node;
	213	int ret = 0;
	214
	215	node = __cds_lfs_pop(&queue->stack);
	216	if (!node)
	217	return -ENOENT;
	218	wait_node = caa_container_of(node, struct urcu_wait_node, node);
	219	CMM_STORE_SHARED(wait_node->in_waitqueue, false);
	220	/* Don't wake already running threads */
	221	if (!(wait_node->state & URCU_WAIT_RUNNING))
	222	ret = urcu_adaptative_wake_up(wait_node);
	223	return ret;
	224	}
	225
	226	/*
	227	* Need mutual exclusion against other wakeup and move waiters
	228	* operations. It is provided by the caller.
	229	*/
	230	static inline
	231	int urcu_dequeue_wake_n(struct urcu_wait_queue *queue, int n)
	232	{
	233	int nr_wakeup = 0;
	234
	235	for (;;) {
	236	int ret;
	237
	238	ret = urcu_dequeue_wake_single(queue);
	239	if (ret < 0)
	240	return nr_wakeup;
	241	else if (ret > 0)
	242	nr_wakeup++;
	243	else
	244	break;
	245	}
	246	return nr_wakeup;
	247	}
	248
	249	static inline
	250	int urcu_wake_all_waiters(struct urcu_waiters *waiters)
	251	{
	252	struct cds_lfs_node iter, iter_n;
	253	int nr_wakeup = 0;
	254
	255	/* Wake all waiters in our stack head */
	256	cds_lfs_for_each_safe(waiters->head, iter, iter_n) {
	257	struct urcu_wait_node *wait_node =
	258	caa_container_of(iter, struct urcu_wait_node, node);
	259
	260	CMM_STORE_SHARED(wait_node->in_waitqueue, false);
	261	/* Don't wake already running threads */
	262	if (wait_node->state & URCU_WAIT_RUNNING)
	263	continue;
	264	if (urcu_adaptative_wake_up(wait_node))
	265	nr_wakeup++;
	266	}
	267	return nr_wakeup;
	268	}
	269
	270	#endif /* _URCU_WAITQUEUE_LIFO_H */