[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 */
};

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

/*
 * 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->node.next, NULL);
	/* 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->node.next, NULL);
		/* 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 */
cba82d7b MD	48	};
cba82d7b MD	49
bf6822a6 MD	50	#define URCU_WAIT_NODE_INIT(name, _state) \
	51	{ .state = _state }
	52
	53	#define DEFINE_URCU_WAIT_NODE(name, state) \
	54	struct urcu_wait_node name = URCU_WAIT_NODE_INIT(name, state)
	55
	56	#define DECLARE_URCU_WAIT_NODE(name) \
	57	struct urcu_wait_node name
	58
	59	struct urcu_wait_queue {
7a618cf7	60	struct __cds_lfs_stack stack;
bf6822a6 MD	61	};
	62
	63	#define URCU_WAIT_QUEUE_HEAD_INIT(name) \
7a618cf7	64	{ .stack.head = CDS_LFS_END, }
bf6822a6 MD	65
	66	#define DECLARE_URCU_WAIT_QUEUE(name) \
	67	struct urcu_wait_queue name
	68
	69	#define DEFINE_URCU_WAIT_QUEUE(name) \
	70	struct urcu_wait_queue name = URCU_WAIT_QUEUE_HEAD_INIT(name)
	71
13652c4b MD	72	static inline
	73	void urcu_wait_queue_init(struct urcu_wait_queue *queue)
	74	{
7a618cf7	75	__cds_lfs_init(&queue->stack);
13652c4b MD	76	}
13652c4b MD	77
bf6822a6	78	struct urcu_waiters {
7a618cf7	79	struct cds_lfs_head *head;
bf6822a6 MD	80	};
	81
	82	/*
	83	* Add ourself atomically to a wait queue. Return 0 if queue was
	84	* previously empty, else return 1.
	85	* A full memory barrier is issued before being added to the wait queue.
	86	*/
	87	static inline
	88	bool urcu_wait_add(struct urcu_wait_queue *queue,
	89	struct urcu_wait_node *node)
	90	{
7a618cf7	91	return cds_lfs_push(&queue->stack, &node->node);
bf6822a6 MD	92	}
	93
	94	/*
	95	* Atomically move all waiters from wait queue into our local struct
	96	* urcu_waiters.
	97	*/
	98	static inline
	99	void urcu_move_waiters(struct urcu_waiters *waiters,
	100	struct urcu_wait_queue *queue)
	101	{
7a618cf7	102	waiters->head = __cds_lfs_pop_all(&queue->stack);
bf6822a6 MD	103	}
	104
	105	static inline
	106	void urcu_wait_set_state(struct urcu_wait_node *node,
	107	enum urcu_wait_state state)
	108	{
	109	node->state = state;
	110	}
	111
13652c4b MD	112	static inline
	113	void urcu_wait_or_state(struct urcu_wait_node *node,
	114	enum urcu_wait_state state)
	115	{
	116	uatomic_or(&node->state, state);
	117	}
	118
cba82d7b	119	static inline
bf6822a6 MD	120	void urcu_wait_node_init(struct urcu_wait_node *node,
bf6822a6 MD	121	enum urcu_wait_state state)
cba82d7b	122	{
bf6822a6	123	urcu_wait_set_state(node, state);
7a618cf7	124	cds_lfs_node_init(&node->node);
cba82d7b MD	125	}
	126
	127	/*
	128	* Note: urcu_adaptative_wake_up needs "value" to stay allocated
bf6822a6	129	* throughout its execution. In this scheme, the waiter owns the node
cba82d7b MD	130	* memory, and we only allow it to free this memory when it receives the
cba82d7b MD	131	* URCU_WAIT_TEARDOWN flag.
5d30bf32 MD	132	* Return true if wakeup is performed, false if thread was already
5d30bf32 MD	133	* running.
cba82d7b MD	134	*/
cba82d7b MD	135	static inline
5d30bf32	136	bool urcu_adaptative_wake_up(struct urcu_wait_node *wait)
cba82d7b	137	{
5d30bf32 MD	138	bool wakeup_performed = false;
5d30bf32 MD	139
cba82d7b	140	cmm_smp_mb();
13652c4b MD	141	/*
	142	* "or" of WAKEUP flag rather than "set" is useful for multiple
	143	* concurrent wakeup sources. Note that "WAIT_TEARDOWN" becomes
	144	* useless when we use multiple wakeup sources: lifetime of the
	145	* "value" should then be handled by the caller.
	146	*/
	147	uatomic_or(&wait->state, URCU_WAIT_WAKEUP);
5d30bf32	148	if (!(uatomic_read(&wait->state) & URCU_WAIT_RUNNING)) {
bf6822a6	149	futex_noasync(&wait->state, FUTEX_WAKE, 1, NULL, NULL, 0);
5d30bf32 MD	150	wakeup_performed = true;
5d30bf32 MD	151	}
cba82d7b	152	/* Allow teardown of struct urcu_wait memory. */
bf6822a6	153	uatomic_or(&wait->state, URCU_WAIT_TEARDOWN);
5d30bf32	154	return wakeup_performed;
cba82d7b MD	155	}
	156
	157	/*
	158	* Caller must initialize "value" to URCU_WAIT_WAITING before passing its
	159	* memory to waker thread.
	160	*/
bf6822a6 MD	161	static inline
bf6822a6 MD	162	void urcu_adaptative_busy_wait(struct urcu_wait_node *wait)
cba82d7b MD	163	{
	164	unsigned int i;
	165
bf6822a6	166	/* Load and test condition before read state */
cba82d7b MD	167	cmm_smp_rmb();
cba82d7b MD	168	for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
bf6822a6	169	if (uatomic_read(&wait->state) != URCU_WAIT_WAITING)
cba82d7b MD	170	goto skip_futex_wait;
	171	caa_cpu_relax();
	172	}
bf6822a6	173	futex_noasync(&wait->state, FUTEX_WAIT,
cba82d7b MD	174	URCU_WAIT_WAITING, NULL, NULL, 0);
	175	skip_futex_wait:
	176
ffa11a18	177	/* Tell waker thread than we are running. */
bf6822a6	178	uatomic_or(&wait->state, URCU_WAIT_RUNNING);
cba82d7b MD	179
	180	/*
	181	* Wait until waker thread lets us know it's ok to tear down
	182	* memory allocated for struct urcu_wait.
	183	*/
	184	for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
bf6822a6	185	if (uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN)
cba82d7b MD	186	break;
	187	caa_cpu_relax();
	188	}
bf6822a6	189	while (!(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN))
cba82d7b	190	poll(NULL, 0, 10);
bf6822a6 MD	191	assert(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN);
	192	}
	193
13652c4b MD	194	/*
	195	* Need mutual exclusion against other wakeup and move waiters
	196	* operations. It is provided by the caller.
	197	*/
	198	static inline
	199	int urcu_dequeue_wake_single(struct urcu_wait_queue *queue)
	200	{
7a618cf7	201	struct cds_lfs_node *node;
13652c4b	202	struct urcu_wait_node *wait_node;
5d30bf32	203	int ret = 0;
13652c4b	204
7a618cf7	205	node = __cds_lfs_pop(&queue->stack);
13652c4b MD	206	if (!node)
	207	return -ENOENT;
	208	wait_node = caa_container_of(node, struct urcu_wait_node, node);
	209	CMM_STORE_SHARED(wait_node->node.next, NULL);
	210	/* Don't wake already running threads */
5d30bf32 MD	211	if (!(wait_node->state & URCU_WAIT_RUNNING))
	212	ret = urcu_adaptative_wake_up(wait_node);
	213	return ret;
13652c4b MD	214	}
	215
	216	/*
	217	* Need mutual exclusion against other wakeup and move waiters
	218	* operations. It is provided by the caller.
	219	*/
	220	static inline
	221	int urcu_dequeue_wake_n(struct urcu_wait_queue *queue, int n)
	222	{
	223	int nr_wakeup = 0;
	224
	225	for (;;) {
	226	int ret;
	227
	228	ret = urcu_dequeue_wake_single(queue);
	229	if (ret < 0)
	230	return nr_wakeup;
	231	else if (ret > 0)
	232	nr_wakeup++;
	233	else
	234	break;
	235	}
	236	return nr_wakeup;
	237	}
	238
bf6822a6	239	static inline
13652c4b	240	int urcu_wake_all_waiters(struct urcu_waiters *waiters)
bf6822a6	241	{
7a618cf7	242	struct cds_lfs_node iter, iter_n;
13652c4b	243	int nr_wakeup = 0;
bf6822a6 MD	244
bf6822a6 MD	245	/* Wake all waiters in our stack head */
7a618cf7	246	cds_lfs_for_each_safe(waiters->head, iter, iter_n) {
bf6822a6 MD	247	struct urcu_wait_node *wait_node =
	248	caa_container_of(iter, struct urcu_wait_node, node);
	249
13652c4b	250	CMM_STORE_SHARED(wait_node->node.next, NULL);
bf6822a6 MD	251	/* Don't wake already running threads */
	252	if (wait_node->state & URCU_WAIT_RUNNING)
	253	continue;
5d30bf32 MD	254	if (urcu_adaptative_wake_up(wait_node))
5d30bf32 MD	255	nr_wakeup++;
bf6822a6	256	}
13652c4b	257	return nr_wakeup;
cba82d7b MD	258	}
cba82d7b MD	259
13652c4b	260	#endif /* _URCU_WAITQUEUE_LIFO_H */