net/ice/base: remove RSS code as iavf host
[dpdk.git] / examples / performance-thread / common / lthread_queue.h
1 /*
2  * SPDX-License-Identifier: BSD-3-Clause
3  * Copyright 2015 Intel Corporation.
4  * Copyright 2010-2011 Dmitry Vyukov
5  */
6
7 #ifndef LTHREAD_QUEUE_H_
8 #define LTHREAD_QUEUE_H_
9
10 #ifdef __cplusplus
11 extern "C" {
12 #endif
13
14 #include <string.h>
15
16 #include <rte_prefetch.h>
17 #include <rte_per_lcore.h>
18
19 #include "lthread_int.h"
20 #include "lthread.h"
21 #include "lthread_diag.h"
22 #include "lthread_pool.h"
23
24 struct lthread_queue;
25
26 /*
27  * This file implements an unbounded FIFO queue based on a lock free
28  * linked list.
29  *
30  * The queue is non-intrusive in that it uses intermediate nodes, and does
31  * not require these nodes to be inserted into the object being placed
32  * in the queue.
33  *
34  * This is slightly more efficient than the very similar queue in lthread_pool
35  * in that it does not have to swing a stub node as the queue becomes empty.
36  *
37  * The queue access functions allocate and free intermediate node
38  * transparently from/to a per scheduler pool ( see lthread_pool.h ).
39  *
40  * The queue provides both MPSC and SPSC insert methods
41  */
42
43 /*
44  * define a queue of lthread nodes
45  */
46 struct lthread_queue {
47         struct qnode *head;
48         struct qnode *tail __rte_cache_aligned;
49         struct lthread_queue *p;
50         char name[LT_MAX_NAME_SIZE];
51
52         DIAG_COUNT_DEFINE(rd);
53         DIAG_COUNT_DEFINE(wr);
54         DIAG_COUNT_DEFINE(size);
55
56 } __rte_cache_aligned;
57
58
59
60 static inline struct lthread_queue *
61 _lthread_queue_create(const char *name)
62 {
63         struct qnode *stub;
64         struct lthread_queue *new_queue;
65
66         new_queue = rte_malloc_socket(NULL, sizeof(struct lthread_queue),
67                                         RTE_CACHE_LINE_SIZE,
68                                         rte_socket_id());
69         if (new_queue == NULL)
70                 return NULL;
71
72         /* allocated stub node */
73         stub = _qnode_alloc();
74         RTE_ASSERT(stub);
75
76         if (name != NULL)
77                 strncpy(new_queue->name, name, sizeof(new_queue->name));
78         new_queue->name[sizeof(new_queue->name)-1] = 0;
79
80         /* initialize queue as empty */
81         stub->next = NULL;
82         new_queue->head = stub;
83         new_queue->tail = stub;
84
85         DIAG_COUNT_INIT(new_queue, rd);
86         DIAG_COUNT_INIT(new_queue, wr);
87         DIAG_COUNT_INIT(new_queue, size);
88
89         return new_queue;
90 }
91
92 /**
93  * Return true if the queue is empty
94  */
95 static __rte_always_inline int
96 _lthread_queue_empty(struct lthread_queue *q)
97 {
98         return q->tail == q->head;
99 }
100
101
102
103 /**
104  * Destroy a queue
105  * fail if queue is not empty
106  */
107 static inline int _lthread_queue_destroy(struct lthread_queue *q)
108 {
109         if (q == NULL)
110                 return -1;
111
112         if (!_lthread_queue_empty(q))
113                 return -1;
114
115         _qnode_free(q->head);
116         rte_free(q);
117         return 0;
118 }
119
120 RTE_DECLARE_PER_LCORE(struct lthread_sched *, this_sched);
121
122 /*
123  * Insert a node into a queue
124  * this implementation is multi producer safe
125  */
126 static __rte_always_inline struct qnode *
127 _lthread_queue_insert_mp(struct lthread_queue
128                                                           *q, void *data)
129 {
130         struct qnode *prev;
131         struct qnode *n = _qnode_alloc();
132
133         if (n == NULL)
134                 return NULL;
135
136         /* set object in node */
137         n->data = data;
138         n->next = NULL;
139
140         /* this is an MPSC method, perform a locked update */
141         prev = n;
142         prev =
143             (struct qnode *)__sync_lock_test_and_set((uint64_t *) &(q)->head,
144                                                (uint64_t) prev);
145         /* there is a window of inconsistency until prev next is set,
146          * which is why remove must retry
147          */
148         prev->next = n;
149
150         DIAG_COUNT_INC(q, wr);
151         DIAG_COUNT_INC(q, size);
152
153         return n;
154 }
155
156 /*
157  * Insert an node into a queue in single producer mode
158  * this implementation is NOT mult producer safe
159  */
160 static __rte_always_inline struct qnode *
161 _lthread_queue_insert_sp(struct lthread_queue
162                                                           *q, void *data)
163 {
164         /* allocate a queue node */
165         struct qnode *prev;
166         struct qnode *n = _qnode_alloc();
167
168         if (n == NULL)
169                 return NULL;
170
171         /* set data in node */
172         n->data = data;
173         n->next = NULL;
174
175         /* this is an SPSC method, no need for locked exchange operation */
176         prev = q->head;
177         prev->next = q->head = n;
178
179         DIAG_COUNT_INC(q, wr);
180         DIAG_COUNT_INC(q, size);
181
182         return n;
183 }
184
185 /*
186  * Remove a node from a queue
187  */
188 static __rte_always_inline void *
189 _lthread_queue_poll(struct lthread_queue *q)
190 {
191         void *data = NULL;
192         struct qnode *tail = q->tail;
193         struct qnode *next = (struct qnode *)tail->next;
194         /*
195          * There is a small window of inconsistency between producer and
196          * consumer whereby the queue may appear empty if consumer and
197          * producer access it at the same time.
198          * The consumer must handle this by retrying
199          */
200
201         if (likely(next != NULL)) {
202                 q->tail = next;
203                 tail->data = next->data;
204                 data = tail->data;
205
206                 /* free the node */
207                 _qnode_free(tail);
208
209                 DIAG_COUNT_INC(q, rd);
210                 DIAG_COUNT_DEC(q, size);
211                 return data;
212         }
213         return NULL;
214 }
215
216 /*
217  * Remove a node from a queue
218  */
219 static __rte_always_inline void *
220 _lthread_queue_remove(struct lthread_queue *q)
221 {
222         void *data = NULL;
223
224         /*
225          * There is a small window of inconsistency between producer and
226          * consumer whereby the queue may appear empty if consumer and
227          * producer access it at the same time. We handle this by retrying
228          */
229         do {
230                 data = _lthread_queue_poll(q);
231
232                 if (likely(data != NULL)) {
233
234                         DIAG_COUNT_INC(q, rd);
235                         DIAG_COUNT_DEC(q, size);
236                         return data;
237                 }
238                 rte_compiler_barrier();
239         } while (unlikely(!_lthread_queue_empty(q)));
240         return NULL;
241 }
242
243 #ifdef __cplusplus
244 }
245 #endif
246
247 #endif                          /* LTHREAD_QUEUE_H_ */