[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
13652c4b MD	1	#ifndef _URCU_WAITQUEUE_LIFO_H
13652c4b MD	2	#define _URCU_WAITQUEUE_LIFO_H
cba82d7b MD	3
cba82d7b MD	4	/*
13652c4b	5	* urcu/waitqueue-lifo.h
cba82d7b	6	*
13652c4b	7	* Userspace RCU library - wait queue scheme with LIFO semantic
cba82d7b	8	*
13652c4b	9	* Copyright (c) 2012-2014 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
cba82d7b MD	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>
7a618cf7	27	#include <urcu/lfstack.h>
13652c4b	28	#include <urcu/futex.h>
cba82d7b MD	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),
bf6822a6	41	URCU_WAIT_RUNNING = (1 << 1),
cba82d7b MD	42	URCU_WAIT_TEARDOWN = (1 << 2),
	43	};
	44
bf6822a6	45	struct urcu_wait_node {
7a618cf7	46	struct cds_lfs_node node;
bf6822a6	47	int32_t state; /* enum urcu_wait_state */
aa46e09f	48	int in_waitqueue;
cba82d7b MD	49	};
cba82d7b MD	50
bf6822a6 MD	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 {
7a618cf7	61	struct __cds_lfs_stack stack;
bf6822a6 MD	62	};
	63
	64	#define URCU_WAIT_QUEUE_HEAD_INIT(name) \
7a618cf7	65	{ .stack.head = CDS_LFS_END, }
bf6822a6 MD	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
13652c4b MD	73	static inline
	74	void urcu_wait_queue_init(struct urcu_wait_queue *queue)
	75	{
7a618cf7	76	__cds_lfs_init(&queue->stack);
13652c4b MD	77	}
13652c4b MD	78
bf6822a6	79	struct urcu_waiters {
7a618cf7	80	struct cds_lfs_head *head;
bf6822a6 MD	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	{
aa46e09f MD	92	cds_lfs_node_init(&node->node);
aa46e09f MD	93	CMM_STORE_SHARED(node->in_waitqueue, true);
7a618cf7	94	return cds_lfs_push(&queue->stack, &node->node);
bf6822a6 MD	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	{
7a618cf7	105	waiters->head = __cds_lfs_pop_all(&queue->stack);
bf6822a6 MD	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
13652c4b MD	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
cba82d7b	122	static inline
bf6822a6 MD	123	void urcu_wait_node_init(struct urcu_wait_node *node,
bf6822a6 MD	124	enum urcu_wait_state state)
cba82d7b	125	{
bf6822a6	126	urcu_wait_set_state(node, state);
7a618cf7	127	cds_lfs_node_init(&node->node);
aa46e09f MD	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);
cba82d7b MD	135	}
	136
	137	/*
	138	* Note: urcu_adaptative_wake_up needs "value" to stay allocated
bf6822a6	139	* throughout its execution. In this scheme, the waiter owns the node
cba82d7b MD	140	* memory, and we only allow it to free this memory when it receives the
cba82d7b MD	141	* URCU_WAIT_TEARDOWN flag.
5d30bf32 MD	142	* Return true if wakeup is performed, false if thread was already
5d30bf32 MD	143	* running.
cba82d7b MD	144	*/
cba82d7b MD	145	static inline
5d30bf32	146	bool urcu_adaptative_wake_up(struct urcu_wait_node *wait)
cba82d7b	147	{
5d30bf32 MD	148	bool wakeup_performed = false;
5d30bf32 MD	149
cba82d7b	150	cmm_smp_mb();
13652c4b MD	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);
5d30bf32	158	if (!(uatomic_read(&wait->state) & URCU_WAIT_RUNNING)) {
bf6822a6	159	futex_noasync(&wait->state, FUTEX_WAKE, 1, NULL, NULL, 0);
5d30bf32 MD	160	wakeup_performed = true;
5d30bf32 MD	161	}
cba82d7b	162	/* Allow teardown of struct urcu_wait memory. */
bf6822a6	163	uatomic_or(&wait->state, URCU_WAIT_TEARDOWN);
5d30bf32	164	return wakeup_performed;
cba82d7b MD	165	}
	166
	167	/*
	168	* Caller must initialize "value" to URCU_WAIT_WAITING before passing its
	169	* memory to waker thread.
	170	*/
bf6822a6 MD	171	static inline
bf6822a6 MD	172	void urcu_adaptative_busy_wait(struct urcu_wait_node *wait)
cba82d7b MD	173	{
	174	unsigned int i;
	175
bf6822a6	176	/* Load and test condition before read state */
cba82d7b MD	177	cmm_smp_rmb();
cba82d7b MD	178	for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
bf6822a6	179	if (uatomic_read(&wait->state) != URCU_WAIT_WAITING)
cba82d7b MD	180	goto skip_futex_wait;
	181	caa_cpu_relax();
	182	}
bf6822a6	183	futex_noasync(&wait->state, FUTEX_WAIT,
cba82d7b MD	184	URCU_WAIT_WAITING, NULL, NULL, 0);
	185	skip_futex_wait:
	186
ffa11a18	187	/* Tell waker thread than we are running. */
bf6822a6	188	uatomic_or(&wait->state, URCU_WAIT_RUNNING);
cba82d7b MD	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++) {
bf6822a6	195	if (uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN)
cba82d7b MD	196	break;
	197	caa_cpu_relax();
	198	}
bf6822a6	199	while (!(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN))
cba82d7b	200	poll(NULL, 0, 10);
bf6822a6 MD	201	assert(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN);
	202	}
	203
13652c4b MD	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	{
7a618cf7	211	struct cds_lfs_node *node;
13652c4b	212	struct urcu_wait_node *wait_node;
5d30bf32	213	int ret = 0;
13652c4b	214
7a618cf7	215	node = __cds_lfs_pop(&queue->stack);
13652c4b MD	216	if (!node)
	217	return -ENOENT;
	218	wait_node = caa_container_of(node, struct urcu_wait_node, node);
aa46e09f	219	CMM_STORE_SHARED(wait_node->in_waitqueue, false);
13652c4b	220	/* Don't wake already running threads */
5d30bf32 MD	221	if (!(wait_node->state & URCU_WAIT_RUNNING))
	222	ret = urcu_adaptative_wake_up(wait_node);
	223	return ret;
13652c4b MD	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
bf6822a6	249	static inline
13652c4b	250	int urcu_wake_all_waiters(struct urcu_waiters *waiters)
bf6822a6	251	{
7a618cf7	252	struct cds_lfs_node iter, iter_n;
13652c4b	253	int nr_wakeup = 0;
bf6822a6 MD	254
bf6822a6 MD	255	/* Wake all waiters in our stack head */
7a618cf7	256	cds_lfs_for_each_safe(waiters->head, iter, iter_n) {
bf6822a6 MD	257	struct urcu_wait_node *wait_node =
	258	caa_container_of(iter, struct urcu_wait_node, node);
	259
aa46e09f	260	CMM_STORE_SHARED(wait_node->in_waitqueue, false);
bf6822a6 MD	261	/* Don't wake already running threads */
	262	if (wait_node->state & URCU_WAIT_RUNNING)
	263	continue;
5d30bf32 MD	264	if (urcu_adaptative_wake_up(wait_node))
5d30bf32 MD	265	nr_wakeup++;
bf6822a6	266	}
13652c4b	267	return nr_wakeup;
cba82d7b MD	268	}
cba82d7b MD	269
13652c4b	270	#endif /* _URCU_WAITQUEUE_LIFO_H */