5 * Copyright(c) 2015 Intel Corporation. All rights reserved.
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions
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
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.
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.
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
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
45 * 1. Redistributions of source code must retain the above copyright notice,
46 * this list of conditions and the following disclaimer.
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.
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.
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.
69 #ifndef LTHREAD_POOL_H_
70 #define LTHREAD_POOL_H_
72 #include <rte_malloc.h>
73 #include <rte_per_lcore.h>
76 #include "lthread_int.h"
77 #include "lthread_diag.h"
80 * This file implements pool of queue nodes used by the queue implemented
83 * The pool is an intrusive lock free MPSC queue.
85 * The pool is created empty and populated lazily, i.e. on first attempt to
86 * allocate a the pool.
88 * Whenever the pool is empty more nodes are added to the pool
89 * The number of nodes preallocated in this way is a parameter of
90 * _qnode_pool_create. Freeing an object returns it to the pool.
92 * Each lthread scheduler maintains its own pool of nodes. L-threads must always
93 * allocate from this local pool ( because it is a single consumer queue ).
94 * L-threads can free nodes to any pool (because it is a multi producer queue)
95 * This enables threads that have affined to a different scheduler to free
103 * define intermediate node
108 struct qnode_pool *pool;
109 } __rte_cache_aligned;
117 struct qnode *fast_alloc;
118 struct qnode *tail __rte_cache_aligned;
120 char name[LT_MAX_NAME_SIZE];
122 DIAG_COUNT_DEFINE(rd);
123 DIAG_COUNT_DEFINE(wr);
124 DIAG_COUNT_DEFINE(available);
125 DIAG_COUNT_DEFINE(prealloc);
126 DIAG_COUNT_DEFINE(capacity);
127 } __rte_cache_aligned;
130 * Create a pool of qnodes
133 static inline struct qnode_pool *
134 _qnode_pool_create(const char *name, int prealloc_size) {
136 struct qnode_pool *p = rte_malloc_socket(NULL,
137 sizeof(struct qnode_pool),
143 p->stub = rte_malloc_socket(NULL,
144 sizeof(struct qnode),
151 strncpy(p->name, name, LT_MAX_NAME_SIZE);
152 p->name[sizeof(p->name)-1] = 0;
155 p->stub->next = NULL;
158 p->pre_alloc = prealloc_size;
160 DIAG_COUNT_INIT(p, rd);
161 DIAG_COUNT_INIT(p, wr);
162 DIAG_COUNT_INIT(p, available);
163 DIAG_COUNT_INIT(p, prealloc);
164 DIAG_COUNT_INIT(p, capacity);
171 * Insert a node into the pool
173 static inline void __attribute__ ((always_inline))
174 _qnode_pool_insert(struct qnode_pool *p, struct qnode *n)
177 struct qnode *prev = n;
178 /* We insert at the head */
179 prev = (struct qnode *) __sync_lock_test_and_set((uint64_t *)&p->head,
181 /* there is a window of inconsistency until prev next is set */
182 /* which is why remove must retry */
187 * Remove a node from the pool
189 * There is a race with _qnode_pool_insert() whereby the queue could appear
190 * empty during a concurrent insert, this is handled by retrying
192 * The queue uses a stub node, which must be swung as the queue becomes
193 * empty, this requires an insert of the stub, which means that removing the
194 * last item from the queue incurs the penalty of an atomic exchange. Since the
195 * pool is maintained with a bulk pre-allocation the cost of this is amortised.
197 static inline struct qnode *__attribute__ ((always_inline))
198 _pool_remove(struct qnode_pool *p)
201 struct qnode *tail = p->tail;
202 struct qnode *next = tail->next;
204 /* we remove from the tail */
205 if (tail == p->stub) {
208 /* advance the tail */
213 if (likely(next != NULL)) {
222 /* swing stub node */
223 _qnode_pool_insert(p, p->stub);
235 * This adds a retry to the _pool_remove function
238 static inline struct qnode *__attribute__ ((always_inline))
239 _qnode_pool_remove(struct qnode_pool *p)
245 if (likely(n != NULL))
248 rte_compiler_barrier();
249 } while ((p->head != p->tail) &&
250 (p->tail != p->stub));
255 * Allocate a node from the pool
256 * If the pool is empty add mode nodes
258 static inline struct qnode *__attribute__ ((always_inline))
261 struct qnode_pool *p = (THIS_SCHED)->qnode_pool;
262 int prealloc_size = p->pre_alloc;
266 if (likely(p->fast_alloc != NULL)) {
268 p->fast_alloc = NULL;
272 n = _qnode_pool_remove(p);
274 if (unlikely(n == NULL)) {
275 DIAG_COUNT_INC(p, prealloc);
276 for (i = 0; i < prealloc_size; i++) {
277 n = rte_malloc_socket(NULL,
278 sizeof(struct qnode),
284 DIAG_COUNT_INC(p, available);
285 DIAG_COUNT_INC(p, capacity);
288 _qnode_pool_insert(p, n);
290 n = _qnode_pool_remove(p);
293 DIAG_COUNT_INC(p, rd);
294 DIAG_COUNT_DEC(p, available);
301 * free a queue node to the per scheduler pool from which it came
303 static inline void __attribute__ ((always_inline))
304 _qnode_free(struct qnode *n)
306 struct qnode_pool *p = n->pool;
309 if (unlikely(p->fast_alloc != NULL) ||
310 unlikely(n->pool != (THIS_SCHED)->qnode_pool)) {
311 DIAG_COUNT_INC(p, wr);
312 DIAG_COUNT_INC(p, available);
313 _qnode_pool_insert(p, n);
320 * Destroy an qnode pool
321 * queue must be empty when this is called
324 _qnode_pool_destroy(struct qnode_pool *p)
332 #endif /* LTHREAD_POOL_H_ */