[ceph.git] / ceph / src / dpdk / examples / performance-thread / common / lthread_queue.h

/*
 *-
 *   BSD LICENSE
 *
 *   Copyright(c) 2015 Intel Corporation. All rights reserved.
 *   All rights reserved.
 *
 *   Redistribution and use in source and binary forms, with or without
 *   modification, are permitted provided that the following conditions
 *   are met:
 *
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in
 *       the documentation and/or other materials provided with the
 *       distribution.
 *     * Neither the name of Intel Corporation nor the names of its
 *       contributors may be used to endorse or promote products derived
 *       from this software without specific prior written permission.
 *
 *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

/*
 * Some portions of this software is derived from the producer
 * consumer queues described by Dmitry Vyukov and published  here
 * http://www.1024cores.net
 *
 * Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer in the documentation
 * and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
 * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL DMITRY VYUKOV OR CONTRIBUTORS
 * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
 * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
 * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
 * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
 * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * The views and conclusions contained in the software and documentation are
 * those of the authors and should not be interpreted as representing official
 * policies, either expressed or implied, of Dmitry Vyukov.
 */

#ifndef LTHREAD_QUEUE_H_
#define LTHREAD_QUEUE_H_

#include <string.h>

#include <rte_prefetch.h>
#include <rte_per_lcore.h>

#include "lthread_int.h"
#include "lthread.h"
#include "lthread_diag.h"
#include "lthread_pool.h"

struct lthread_queue;

/*
 * This file implements an unbounded FIFO queue based on a lock free
 * linked list.
 *
 * The queue is non-intrusive in that it uses intermediate nodes, and does
 * not require these nodes to be inserted into the object being placed
 * in the queue.
 *
 * This is slightly more efficient than the very similar queue in lthread_pool
 * in that it does not have to swing a stub node as the queue becomes empty.
 *
 * The queue access functions allocate and free intermediate node
 * transparently from/to a per scheduler pool ( see lthread_pool.h ).
 *
 * The queue provides both MPSC and SPSC insert methods
 */

/*
 * define a queue of lthread nodes
 */
struct lthread_queue {
	struct qnode *head;
	struct qnode *tail __rte_cache_aligned;
	struct lthread_queue *p;
	char name[LT_MAX_NAME_SIZE];

	DIAG_COUNT_DEFINE(rd);
	DIAG_COUNT_DEFINE(wr);
	DIAG_COUNT_DEFINE(size);

} __rte_cache_aligned;


static inline struct lthread_queue *
_lthread_queue_create(const char *name)
{
	struct qnode *stub;
	struct lthread_queue *new_queue;

	new_queue = rte_malloc_socket(NULL, sizeof(struct lthread_queue),
					RTE_CACHE_LINE_SIZE,
					rte_socket_id());
	if (new_queue == NULL)
		return NULL;

	/* allocated stub node */
	stub = _qnode_alloc();
	RTE_ASSERT(stub);

	if (name != NULL)
		strncpy(new_queue->name, name, sizeof(new_queue->name));
	new_queue->name[sizeof(new_queue->name)-1] = 0;

	/* initialize queue as empty */
	stub->next = NULL;
	new_queue->head = stub;
	new_queue->tail = stub;

	DIAG_COUNT_INIT(new_queue, rd);
	DIAG_COUNT_INIT(new_queue, wr);
	DIAG_COUNT_INIT(new_queue, size);

	return new_queue;
}

/**
 * Return true if the queue is empty
 */
static inline int __attribute__ ((always_inline))
_lthread_queue_empty(struct lthread_queue *q)
{
	return q->tail == q->head;
}


/**
 * Destroy a queue
 * fail if queue is not empty
 */
static inline int _lthread_queue_destroy(struct lthread_queue *q)
{
	if (q == NULL)
		return -1;

	if (!_lthread_queue_empty(q))
		return -1;

	_qnode_free(q->head);
	rte_free(q);
	return 0;
}

RTE_DECLARE_PER_LCORE(struct lthread_sched *, this_sched);

/*
 * Insert a node into a queue
 * this implementation is multi producer safe
 */
static inline struct qnode *__attribute__ ((always_inline))
_lthread_queue_insert_mp(struct lthread_queue
							  *q, void *data)
{
	struct qnode *prev;
	struct qnode *n = _qnode_alloc();

	if (n == NULL)
		return NULL;

	/* set object in node */
	n->data = data;
	n->next = NULL;

	/* this is an MPSC method, perform a locked update */
	prev = n;
	prev =
	    (struct qnode *)__sync_lock_test_and_set((uint64_t *) &(q)->head,
					       (uint64_t) prev);
	/* there is a window of inconsistency until prev next is set,
	 * which is why remove must retry
	 */
	prev->next = n;

	DIAG_COUNT_INC(q, wr);
	DIAG_COUNT_INC(q, size);

	return n;
}

/*
 * Insert an node into a queue in single producer mode
 * this implementation is NOT mult producer safe
 */
static inline struct qnode *__attribute__ ((always_inline))
_lthread_queue_insert_sp(struct lthread_queue
							  *q, void *data)
{
	/* allocate a queue node */
	struct qnode *prev;
	struct qnode *n = _qnode_alloc();

	if (n == NULL)
		return NULL;

	/* set data in node */
	n->data = data;
	n->next = NULL;

	/* this is an SPSC method, no need for locked exchange operation */
	prev = q->head;
	prev->next = q->head = n;

	DIAG_COUNT_INC(q, wr);
	DIAG_COUNT_INC(q, size);

	return n;
}

/*
 * Remove a node from a queue
 */
static inline void *__attribute__ ((always_inline))
_lthread_queue_poll(struct lthread_queue *q)
{
	void *data = NULL;
	struct qnode *tail = q->tail;
	struct qnode *next = (struct qnode *)tail->next;
	/*
	 * There is a small window of inconsistency between producer and
	 * consumer whereby the queue may appear empty if consumer and
	 * producer access it at the same time.
	 * The consumer must handle this by retrying
	 */

	if (likely(next != NULL)) {
		q->tail = next;
		tail->data = next->data;
		data = tail->data;

		/* free the node */
		_qnode_free(tail);

		DIAG_COUNT_INC(q, rd);
		DIAG_COUNT_DEC(q, size);
		return data;
	}
	return NULL;
}

/*
 * Remove a node from a queue
 */
static inline void *__attribute__ ((always_inline))
_lthread_queue_remove(struct lthread_queue *q)
{
	void *data = NULL;

	/*
	 * There is a small window of inconsistency between producer and
	 * consumer whereby the queue may appear empty if consumer and
	 * producer access it at the same time. We handle this by retrying
	 */
	do {
		data = _lthread_queue_poll(q);

		if (likely(data != NULL)) {

			DIAG_COUNT_INC(q, rd);
			DIAG_COUNT_DEC(q, size);
			return data;
		}
		rte_compiler_barrier();
	} while (unlikely(!_lthread_queue_empty(q)));
	return NULL;
}


#endif				/* LTHREAD_QUEUE_H_ */
Commit	Line	Data
7c673cae FG	1	/*
	2	*-
	3	* BSD LICENSE
	4	*
	5	* Copyright(c) 2015 Intel Corporation. All rights reserved.
	6	* All rights reserved.
	7	*
	8	* Redistribution and use in source and binary forms, with or without
	9	* modification, are permitted provided that the following conditions
	10	* are met:
	11	*
	12	* * Redistributions of source code must retain the above copyright
	13	* notice, this list of conditions and the following disclaimer.
	14	* * Redistributions in binary form must reproduce the above copyright
	15	* notice, this list of conditions and the following disclaimer in
	16	* the documentation and/or other materials provided with the
	17	* distribution.
	18	* * Neither the name of Intel Corporation nor the names of its
	19	* contributors may be used to endorse or promote products derived
	20	* from this software without specific prior written permission.
	21	*
	22	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
	23	* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
	24	* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
	25	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
	26	* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
	27	* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
	28	* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
	29	* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
	30	* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
	31	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
	32	* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
	33	*/
	34
	35	/*
	36	* Some portions of this software is derived from the producer
	37	* consumer queues described by Dmitry Vyukov and published here
	38	* http://www.1024cores.net
	39	*
	40	* Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
	41	* Redistribution and use in source and binary forms, with or without
	42	* modification, are permitted provided that the following conditions
	43	* are met:
	44	*
	45	* 1. Redistributions of source code must retain the above copyright notice,
	46	* this list of conditions and the following disclaimer.
	47	*
	48	* 2. Redistributions in binary form must reproduce the above copyright notice,
	49	* this list of conditions and the following disclaimer in the documentation
	50	* and/or other materials provided with the distribution.
	51	*
	52	* THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS"
	53	* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
	54	* THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
	55	* PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL DMITRY VYUKOV OR CONTRIBUTORS
	56	* BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
	57	* OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
	58	* OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
	59	* OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
	60	* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
	61	* OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
	62	* EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
	63	*
	64	* The views and conclusions contained in the software and documentation are
65	* those of the authors and should not be interpreted as representing official
66	* policies, either expressed or implied, of Dmitry Vyukov.
67	*/
68
69	#ifndef LTHREAD_QUEUE_H_
70	#define LTHREAD_QUEUE_H_
71
72	#include <string.h>
73
74	#include <rte_prefetch.h>
75	#include <rte_per_lcore.h>
76
77	#include "lthread_int.h"
78	#include "lthread.h"
79	#include "lthread_diag.h"
80	#include "lthread_pool.h"
81
82	struct lthread_queue;
83
84	/*
85	* This file implements an unbounded FIFO queue based on a lock free
86	* linked list.
87	*
88	* The queue is non-intrusive in that it uses intermediate nodes, and does
89	* not require these nodes to be inserted into the object being placed
90	* in the queue.
91	*
92	* This is slightly more efficient than the very similar queue in lthread_pool
93	* in that it does not have to swing a stub node as the queue becomes empty.
94	*
95	* The queue access functions allocate and free intermediate node
96	* transparently from/to a per scheduler pool ( see lthread_pool.h ).
97	*
98	* The queue provides both MPSC and SPSC insert methods
99	*/
100
101	/*
102	* define a queue of lthread nodes
103	*/
104	struct lthread_queue {
105	struct qnode *head;
106	struct qnode *tail __rte_cache_aligned;
107	struct lthread_queue *p;
108	char name[LT_MAX_NAME_SIZE];
109
110	DIAG_COUNT_DEFINE(rd);
111	DIAG_COUNT_DEFINE(wr);
112	DIAG_COUNT_DEFINE(size);
113
114	} __rte_cache_aligned;
115
116
117
118	static inline struct lthread_queue *
119	_lthread_queue_create(const char *name)
120	{
121	struct qnode *stub;
122	struct lthread_queue *new_queue;
123
124	new_queue = rte_malloc_socket(NULL, sizeof(struct lthread_queue),
125	RTE_CACHE_LINE_SIZE,
126	rte_socket_id());
127	if (new_queue == NULL)
128	return NULL;
129
130	/* allocated stub node */
131	stub = _qnode_alloc();
132	RTE_ASSERT(stub);
133
134	if (name != NULL)
135	strncpy(new_queue->name, name, sizeof(new_queue->name));
136	new_queue->name[sizeof(new_queue->name)-1] = 0;
137
138	/* initialize queue as empty */
139	stub->next = NULL;
140	new_queue->head = stub;
141	new_queue->tail = stub;
142
143	DIAG_COUNT_INIT(new_queue, rd);
144	DIAG_COUNT_INIT(new_queue, wr);
145	DIAG_COUNT_INIT(new_queue, size);
146
147	return new_queue;
148	}
149
150	/**
151	* Return true if the queue is empty
152	*/
153	static inline int __attribute__ ((always_inline))
154	_lthread_queue_empty(struct lthread_queue *q)
155	{
156	return q->tail == q->head;
157	}
158
159
160
161	/**
162	* Destroy a queue
163	* fail if queue is not empty
164	*/
165	static inline int _lthread_queue_destroy(struct lthread_queue *q)
166	{
167	if (q == NULL)
168	return -1;
169
170	if (!_lthread_queue_empty(q))
171	return -1;
172
173	_qnode_free(q->head);
174	rte_free(q);
175	return 0;
176	}
177
178	RTE_DECLARE_PER_LCORE(struct lthread_sched *, this_sched);
179
180	/*
181	* Insert a node into a queue
182	* this implementation is multi producer safe
183	*/
184	static inline struct qnode *__attribute__ ((always_inline))
185	_lthread_queue_insert_mp(struct lthread_queue
186	q, void data)
187	{
188	struct qnode *prev;
189	struct qnode *n = _qnode_alloc();
190
191	if (n == NULL)
192	return NULL;
193
194	/* set object in node */
195	n->data = data;
196	n->next = NULL;
197
198	/* this is an MPSC method, perform a locked update */
199	prev = n;
200	prev =
201	(struct qnode )__sync_lock_test_and_set((uint64_t ) &(q)->head,
202	(uint64_t) prev);
203	/* there is a window of inconsistency until prev next is set,
204	* which is why remove must retry
205	*/
206	prev->next = n;
207
208	DIAG_COUNT_INC(q, wr);
209	DIAG_COUNT_INC(q, size);
210
211	return n;
212	}
213
214	/*
215	* Insert an node into a queue in single producer mode
216	* this implementation is NOT mult producer safe
217	*/
218	static inline struct qnode *__attribute__ ((always_inline))
219	_lthread_queue_insert_sp(struct lthread_queue
220	q, void data)
221	{
222	/* allocate a queue node */
223	struct qnode *prev;
224	struct qnode *n = _qnode_alloc();
225
226	if (n == NULL)
227	return NULL;
228
229	/* set data in node */
230	n->data = data;
231	n->next = NULL;
232
233	/* this is an SPSC method, no need for locked exchange operation */
234	prev = q->head;
235	prev->next = q->head = n;
236
237	DIAG_COUNT_INC(q, wr);
238	DIAG_COUNT_INC(q, size);
239
240	return n;
241	}
242
243	/*
244	* Remove a node from a queue
245	*/
246	static inline void *__attribute__ ((always_inline))
247	_lthread_queue_poll(struct lthread_queue *q)
248	{
249	void *data = NULL;
250	struct qnode *tail = q->tail;
251	struct qnode next = (struct qnode )tail->next;
252	/*
253	* There is a small window of inconsistency between producer and
254	* consumer whereby the queue may appear empty if consumer and
255	* producer access it at the same time.
256	* The consumer must handle this by retrying
257	*/
258
259	if (likely(next != NULL)) {
260	q->tail = next;
261	tail->data = next->data;
262	data = tail->data;
263
264	/* free the node */
265	_qnode_free(tail);
266
267	DIAG_COUNT_INC(q, rd);
268	DIAG_COUNT_DEC(q, size);
269	return data;
270	}
271	return NULL;
272	}
273
274	/*
275	* Remove a node from a queue
276	*/
277	static inline void *__attribute__ ((always_inline))
278	_lthread_queue_remove(struct lthread_queue *q)
279	{
280	void *data = NULL;
281
282	/*
283	* There is a small window of inconsistency between producer and
284	* consumer whereby the queue may appear empty if consumer and
285	* producer access it at the same time. We handle this by retrying
286	*/
287	do {
288	data = _lthread_queue_poll(q);
289
290	if (likely(data != NULL)) {
291
292	DIAG_COUNT_INC(q, rd);
293	DIAG_COUNT_DEC(q, size);
294	return data;
295	}
296	rte_compiler_barrier();
297	} while (unlikely(!_lthread_queue_empty(q)));
298	return NULL;
299	}
300
301
302	#endif /* LTHREAD_QUEUE_H_ */