Branch data Line data Source code
1 : : /* SPDX-License-Identifier: BSD-3-Clause
2 : : * Copyright(c) 2024 Huawei Technologies Co., Ltd
3 : : */
4 : :
5 : : /**
6 : : * @file
7 : : * This file contains implementation of SORING 'datapath' functions.
8 : : *
9 : : * Brief description:
10 : : * ==================
11 : : * enqueue/dequeue works the same as for conventional rte_ring:
12 : : * any rte_ring sync types can be used, etc.
13 : : * Plus there could be multiple 'stages'.
14 : : * For each stage there is an acquire (start) and release (finish) operation.
15 : : * After some elems are 'acquired' - user can safely assume that he has
16 : : * exclusive possession of these elems till 'release' for them is done.
17 : : * Note that right now user has to release exactly the same number of elems
18 : : * he acquired before.
19 : : * After 'release', elems can be 'acquired' by next stage and/or dequeued
20 : : * (in case of last stage).
21 : : *
22 : : * Internal structure:
23 : : * ===================
24 : : * In addition to 'normal' ring of elems, it also has a ring of states of the
25 : : * same size. Each state[] corresponds to exactly one elem[].
26 : : * state[] will be used by acquire/release/dequeue functions to store internal
27 : : * information and should not be accessed by the user directly.
28 : : *
29 : : * How it works:
30 : : * =============
31 : : * 'acquire()' just moves stage's head (same as rte_ring move_head does),
32 : : * plus it saves in state[stage.cur_head] information about how many elems
33 : : * were acquired, current head position and special flag value to indicate
34 : : * that elems are acquired (SORING_ST_START).
35 : : * Note that 'acquire()' returns to the user a special 'ftoken' that user has
36 : : * to provide for 'release()' (in fact it is just a position for current head
37 : : * plus current stage index).
38 : : * 'release()' extracts old head value from provided ftoken and checks that
39 : : * corresponding 'state[]' contains expected values(mostly for sanity
40 : : * purposes). Then it marks this state[] with 'SORING_ST_FINISH' flag to
41 : : * indicate that given subset of objects was released.
42 : : * After that, it calls 'finalize()'.
43 : : * Main purpose of 'finalize()' operation is to walk through 'state[]'
44 : : * from current stage tail up to its head, check state[] and move stage tail
45 : : * through elements that already are in SORING_ST_FINISH state.
46 : : * Along with that, corresponding state[] values are reset to zero.
47 : : * Note that updated state[] is shared by all threads, so
48 : : * 'finalize()' for given stage can be done from multiple places:
49 : : * 'release()' for that stage or from 'acquire()' for next stage
50 : : * even from consumer's 'dequeue()' - in case given stage is the last one.
51 : : * So 'finalize()' has to be MT-safe and inside it we have to
52 : : * guarantee that only one thread at a time will update state[] and
53 : : * stage's tail values (sort of critical-section).
54 : : * When multiple threads trying to do finalize() for the same stage,
55 : : * simultaneously one thread will win the race and do all the pending
56 : : * updates, while others will simply return (kind of try-lock scenario).
57 : : * That allows 'release()' to avoid excessive waits on the tail value.
58 : : */
59 : :
60 : : #include "soring.h"
61 : :
62 : : #include <eal_export.h>
63 : :
64 : : /*
65 : : * Inline functions (fastpath) start here.
66 : : */
67 : : static __rte_always_inline uint32_t
68 : : __rte_soring_stage_finalize(struct soring_stage_headtail *sht, uint32_t stage,
69 : : union soring_state *rstate, uint32_t rmask, uint32_t maxn)
70 : : {
71 : : int32_t rc;
72 : : uint32_t ftkn, head, i, idx, k, n, tail;
73 : : union soring_stage_tail nt, ot;
74 : : union soring_state st;
75 : :
76 : : /* try to grab exclusive right to update tail value */
77 : 20012 : ot.raw = rte_atomic_load_explicit(&sht->tail.raw,
78 : : rte_memory_order_acquire);
79 : :
80 : : /* other thread already finalizing it for us */
81 : 20012 : if (ot.sync != 0)
82 : : return 0;
83 : :
84 : 20012 : nt.pos = ot.pos;
85 : 20012 : nt.sync = 1;
86 : 20012 : rc = rte_atomic_compare_exchange_strong_explicit(&sht->tail.raw,
87 : : (uint64_t *)(uintptr_t)&ot.raw, nt.raw,
88 : : rte_memory_order_release, rte_memory_order_relaxed);
89 : :
90 : : /* other thread won the race */
91 [ + - + - : 20012 : if (rc == 0)
+ - - - -
- - - -
- ]
92 : : return 0;
93 : :
94 : : /* Ensure the head is read before rstate[] */
95 : 20012 : head = rte_atomic_load_explicit(&sht->head, rte_memory_order_relaxed);
96 : : rte_atomic_thread_fence(rte_memory_order_acquire);
97 : :
98 : : /*
99 : : * start with current tail and walk through states that are
100 : : * already finished.
101 : : */
102 : :
103 : 20012 : n = RTE_MIN(head - ot.pos, maxn);
104 [ - + + + : 30019 : for (i = 0, tail = ot.pos; i < n; i += k, tail += k) {
+ + - - -
- - - -
- ]
105 : :
106 : 10007 : idx = tail & rmask;
107 : 10007 : ftkn = SORING_FTKN_MAKE(tail, stage);
108 : :
109 : 10007 : st.raw = rte_atomic_load_explicit(&rstate[idx].raw,
110 : : rte_memory_order_relaxed);
111 [ - - - - : 10007 : if ((st.stnum & SORING_ST_MASK) != SORING_ST_FINISH ||
+ - + - +
- + - - -
- - - - -
- - - - -
- - - - ]
112 : : st.ftoken != ftkn)
113 : : break;
114 : :
115 : 10007 : k = st.stnum & ~SORING_ST_MASK;
116 : 10007 : rte_atomic_store_explicit(&rstate[idx].raw, 0,
117 : : rte_memory_order_relaxed);
118 : : }
119 : :
120 : :
121 : : /* release exclusive right to update along with new tail value */
122 : 20012 : ot.pos = tail;
123 : 20012 : rte_atomic_store_explicit(&sht->tail.raw, ot.raw,
124 : : rte_memory_order_release);
125 : :
126 : 20012 : return i;
127 : : }
128 : :
129 : : static __rte_always_inline uint32_t
130 : : __rte_soring_move_prod_head(struct rte_soring *r, uint32_t num,
131 : : enum rte_ring_queue_behavior behavior, enum rte_ring_sync_type st,
132 : : uint32_t *head, uint32_t *next, uint32_t *free)
133 : : {
134 : : uint32_t n;
135 : :
136 : 6 : switch (st) {
137 : 0 : case RTE_RING_SYNC_ST:
138 : 0 : n = __rte_ring_headtail_move_head_st(&r->prod.ht, &r->cons.ht,
139 : : r->capacity, num, behavior, head, next, free);
140 : : break;
141 : 6 : case RTE_RING_SYNC_MT:
142 : 6 : n = __rte_ring_headtail_move_head_mt(&r->prod.ht, &r->cons.ht,
143 : : r->capacity, num, behavior, head, next, free);
144 : : break;
145 : 0 : case RTE_RING_SYNC_MT_RTS:
146 : 0 : n = __rte_ring_rts_move_head(&r->prod.rts, &r->cons.ht,
147 : : r->capacity, num, behavior, head, free);
148 : 0 : *next = *head + n;
149 : 0 : break;
150 : 0 : case RTE_RING_SYNC_MT_HTS:
151 : 0 : n = __rte_ring_hts_move_head(&r->prod.hts, &r->cons.ht,
152 : : r->capacity, num, behavior, head, free);
153 : 0 : *next = *head + n;
154 : 0 : break;
155 : : default:
156 : : /* unsupported mode, shouldn't be here */
157 : : RTE_ASSERT(0);
158 : : *free = 0;
159 : : n = 0;
160 : : }
161 : :
162 : : return n;
163 : : }
164 : :
165 : : static __rte_always_inline uint32_t
166 : : __rte_soring_move_cons_head(struct rte_soring *r, uint32_t stage, uint32_t num,
167 : : enum rte_ring_queue_behavior behavior, enum rte_ring_sync_type st,
168 : : uint32_t *head, uint32_t *next, uint32_t *avail)
169 : : {
170 : : uint32_t n;
171 : :
172 [ - + - - : 3 : switch (st) {
- ]
173 : 0 : case RTE_RING_SYNC_ST:
174 : : n = __rte_ring_headtail_move_head_st(&r->cons.ht,
175 : 0 : &r->stage[stage].ht, 0, num, behavior,
176 : : head, next, avail);
177 : : break;
178 : 10011 : case RTE_RING_SYNC_MT:
179 : : n = __rte_ring_headtail_move_head_mt(&r->cons.ht,
180 : 10011 : &r->stage[stage].ht, 0, num, behavior,
181 : : head, next, avail);
182 : : break;
183 : 0 : case RTE_RING_SYNC_MT_RTS:
184 : 0 : n = __rte_ring_rts_move_head(&r->cons.rts, &r->stage[stage].ht,
185 : : 0, num, behavior, head, avail);
186 : 0 : *next = *head + n;
187 : 0 : break;
188 : 0 : case RTE_RING_SYNC_MT_HTS:
189 : 0 : n = __rte_ring_hts_move_head(&r->cons.hts, &r->stage[stage].ht,
190 : : 0, num, behavior, head, avail);
191 : 0 : *next = *head + n;
192 : 0 : break;
193 : : default:
194 : : /* unsupported mode, shouldn't be here */
195 : : RTE_ASSERT(0);
196 : : *avail = 0;
197 : : n = 0;
198 : : }
199 : :
200 : : return n;
201 : : }
202 : :
203 : : static __rte_always_inline void
204 : : __rte_soring_update_tail(struct __rte_ring_headtail *rht,
205 : : enum rte_ring_sync_type st, uint32_t head, uint32_t next)
206 : : {
207 : : uint32_t n;
208 : :
209 [ + - - - : 9 : switch (st) {
+ - - - ]
210 [ + - + - ]: 9 : case RTE_RING_SYNC_ST:
211 : : case RTE_RING_SYNC_MT:
212 : : __rte_ring_update_tail(&rht->ht, head, next, st);
213 : : break;
214 : 0 : case RTE_RING_SYNC_MT_RTS:
215 : : __rte_ring_rts_update_tail(&rht->rts);
216 : : break;
217 : 0 : case RTE_RING_SYNC_MT_HTS:
218 : : n = next - head;
219 : : __rte_ring_hts_update_tail(&rht->hts, head, n);
220 : : break;
221 : : default:
222 : : /* unsupported mode, shouldn't be here */
223 : : RTE_ASSERT(0);
224 : : }
225 : : }
226 : :
227 : : static __rte_always_inline uint32_t
228 : : __rte_soring_stage_move_head(struct soring_stage_headtail *d,
229 : : const struct rte_ring_headtail *s, uint32_t capacity, uint32_t num,
230 : : enum rte_ring_queue_behavior behavior,
231 : : uint32_t *old_head, uint32_t *new_head, uint32_t *avail)
232 : : {
233 : : uint32_t n, tail;
234 : :
235 : 7 : *old_head = rte_atomic_load_explicit(&d->head,
236 : : rte_memory_order_relaxed);
237 : :
238 : : do {
239 : : n = num;
240 : :
241 : : /* Ensure the head is read before tail */
242 : : rte_atomic_thread_fence(rte_memory_order_acquire);
243 : :
244 : 10007 : tail = rte_atomic_load_explicit(&s->tail,
245 : : rte_memory_order_acquire);
246 : 10007 : *avail = capacity + tail - *old_head;
247 [ + - - - : 10005 : if (n > *avail)
+ - + - -
- + - - -
- - ]
248 : : n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *avail;
249 [ + - + - : 10007 : if (n == 0)
- - - - -
- - - + -
+ - - - +
- - - -
- ]
250 : : return 0;
251 : 10007 : *new_head = *old_head + n;
252 : 10007 : } while (rte_atomic_compare_exchange_strong_explicit(&d->head,
253 : : old_head, *new_head, rte_memory_order_acq_rel,
254 [ - + - + : 10007 : rte_memory_order_relaxed) == 0);
- - - - -
- - - - +
- + - - -
+ - - -
- ]
255 : :
256 : : return n;
257 : : }
258 : :
259 : : static inline void
260 : 9 : __enqueue_elems(struct rte_soring *r, const void *objs, const void *meta,
261 : : uint32_t head, uint32_t n)
262 : : {
263 [ + + ]: 9 : __rte_ring_do_enqueue_elems(&r[1], objs, r->size, head & r->mask,
264 : : r->esize, n);
265 [ + + ]: 9 : if (meta != NULL)
266 : 2 : __rte_ring_do_enqueue_elems(r->meta, meta, r->size,
267 [ - + ]: 2 : head & r->mask, r->msize, n);
268 : 9 : }
269 : :
270 : : static inline void
271 : 8 : __dequeue_elems(const struct rte_soring *r, void *objs, void *meta,
272 : : uint32_t head, uint32_t n)
273 : : {
274 : 8 : __rte_ring_do_dequeue_elems(objs, &r[1], r->size, head & r->mask,
275 [ + + ]: 8 : r->esize, n);
276 [ + + ]: 8 : if (meta != NULL)
277 : 2 : __rte_ring_do_dequeue_elems(meta, r->meta, r->size,
278 [ - + ]: 2 : head & r->mask, r->msize, n);
279 : 8 : }
280 : :
281 : : static inline uint32_t
282 : 6 : soring_enqueue(struct rte_soring *r, const void *objs,
283 : : const void *meta, uint32_t n, enum rte_ring_queue_behavior behavior,
284 : : uint32_t *free_space)
285 : : {
286 : : enum rte_ring_sync_type st;
287 : : uint32_t nb_free, prod_head, prod_next;
288 : :
289 : : RTE_ASSERT(r != NULL && r->nb_stage > 0);
290 : : RTE_ASSERT(meta == NULL || r->meta != NULL);
291 : :
292 [ - + - - : 6 : st = r->prod.ht.sync_type;
- ]
293 : :
294 : : n = __rte_soring_move_prod_head(r, n, behavior, st,
295 : : &prod_head, &prod_next, &nb_free);
296 [ + + ]: 6 : if (n != 0) {
297 : 5 : __enqueue_elems(r, objs, meta, prod_head, n);
298 : : __rte_soring_update_tail(&r->prod, st, prod_head, prod_next);
299 : : }
300 : :
301 [ + + ]: 6 : if (free_space != NULL)
302 : 5 : *free_space = nb_free - n;
303 : 6 : return n;
304 : : }
305 : :
306 : : static inline uint32_t
307 : 4 : soring_enqueue_start(struct rte_soring *r, uint32_t num,
308 : : enum rte_ring_queue_behavior behavior, uint32_t *free_space)
309 : : {
310 : : enum rte_ring_sync_type st;
311 : : uint32_t free, head, n, next;
312 : :
313 : : RTE_ASSERT(r != NULL && r->nb_stage > 0);
314 : :
315 : 4 : st = r->prod.ht.sync_type;
316 : :
317 [ + + - ]: 4 : switch (st) {
318 : 2 : case RTE_RING_SYNC_ST:
319 : 2 : n = __rte_ring_headtail_move_head_st(&r->prod.ht, &r->cons.ht,
320 : : r->capacity, num, behavior, &head, &next, &free);
321 : : break;
322 : 2 : case RTE_RING_SYNC_MT_HTS:
323 : 2 : n = __rte_ring_hts_move_head(&r->prod.hts, &r->cons.ht,
324 : : r->capacity, num, behavior, &head, &free);
325 : 2 : break;
326 : : default:
327 : : /* unsupported mode, shouldn't be here */
328 : : RTE_ASSERT(0);
329 : : free = 0;
330 : : n = 0;
331 : : }
332 : :
333 [ + - ]: 4 : if (free_space != NULL)
334 : 4 : *free_space = free - n;
335 : 4 : return n;
336 : : }
337 : :
338 : : static inline void
339 : 4 : soring_enqueue_finish(struct rte_soring *r, const void *objs, const void *meta,
340 : : uint32_t num)
341 : : {
342 : : enum rte_ring_sync_type st;
343 : : uint32_t n, tail;
344 : :
345 : : RTE_ASSERT(r != NULL && r->nb_stage > 0);
346 : : RTE_ASSERT(meta == NULL || r->meta != NULL);
347 : :
348 : 4 : st = r->prod.ht.sync_type;
349 : :
350 [ + + - ]: 4 : switch (st) {
351 [ + - ]: 2 : case RTE_RING_SYNC_ST:
352 : : n = __rte_ring_st_get_tail(&r->prod.ht, &tail, num);
353 [ + - ]: 2 : if (n != 0)
354 : 2 : __enqueue_elems(r, objs, meta, tail, n);
355 : : __rte_ring_st_set_head_tail(&r->prod.ht, tail, n, 1);
356 : : break;
357 [ + - ]: 2 : case RTE_RING_SYNC_MT_HTS:
358 : : n = __rte_ring_hts_get_tail(&r->prod.hts, &tail, num);
359 [ + - ]: 2 : if (n != 0)
360 : 2 : __enqueue_elems(r, objs, meta, tail, n);
361 : : __rte_ring_hts_set_head_tail(&r->prod.hts, tail, n, 1);
362 : : break;
363 : 4 : default:
364 : : /* unsupported mode, shouldn't be here */
365 : : RTE_ASSERT(0);
366 : : }
367 : 4 : }
368 : :
369 : : static inline uint32_t
370 : 10008 : soring_dequeue(struct rte_soring *r, void *objs, void *meta,
371 : : uint32_t num, enum rte_ring_queue_behavior behavior,
372 : : uint32_t *available)
373 : : {
374 : : enum rte_ring_sync_type st;
375 : : uint32_t entries, cons_head, cons_next, n, ns, reqn;
376 : :
377 : : RTE_ASSERT(r != NULL && r->nb_stage > 0);
378 : : RTE_ASSERT(meta == NULL || r->meta != NULL);
379 : :
380 : 10008 : ns = r->nb_stage - 1;
381 [ - + - - : 10008 : st = r->cons.ht.sync_type;
- ]
382 : :
383 : : /* try to grab exactly @num elems first */
384 : : n = __rte_soring_move_cons_head(r, ns, num, RTE_RING_QUEUE_FIXED, st,
385 : : &cons_head, &cons_next, &entries);
386 [ + + ]: 10008 : if (n == 0) {
387 : : /* try to finalize some elems from previous stage */
388 [ + - ]: 10005 : n = __rte_soring_stage_finalize(&r->stage[ns].sht, ns,
389 : : r->state, r->mask, 2 * num);
390 : 10005 : entries += n;
391 : :
392 : : /* repeat attempt to grab elems */
393 [ + + ]: 10005 : reqn = (behavior == RTE_RING_QUEUE_FIXED) ? num : 0;
394 [ + + ]: 10005 : if (entries >= reqn)
395 : : n = __rte_soring_move_cons_head(r, ns, num, behavior,
396 : : st, &cons_head, &cons_next, &entries);
397 : : else
398 : : n = 0;
399 : : }
400 : :
401 : : /* we have some elems to consume */
402 [ + + ]: 10008 : if (n != 0) {
403 : 4 : __dequeue_elems(r, objs, meta, cons_head, n);
404 : : __rte_soring_update_tail(&r->cons, st, cons_head, cons_next);
405 : : }
406 : :
407 [ - + ]: 10008 : if (available != NULL)
408 : 0 : *available = entries - n;
409 : 10008 : return n;
410 : : }
411 : :
412 : : static inline uint32_t
413 : 8 : soring_dequeue_start(struct rte_soring *r, void *objs, void *meta,
414 : : uint32_t num, enum rte_ring_queue_behavior behavior,
415 : : uint32_t *available)
416 : : {
417 : : enum rte_ring_sync_type st;
418 : : uint32_t avail, head, next, n, ns;
419 : :
420 : : RTE_ASSERT(r != NULL && r->nb_stage > 0);
421 : : RTE_ASSERT(meta == NULL || r->meta != NULL);
422 : :
423 : 8 : ns = r->nb_stage - 1;
424 : 8 : st = r->cons.ht.sync_type;
425 : :
426 [ + + - ]: 8 : switch (st) {
427 : 4 : case RTE_RING_SYNC_ST:
428 : 4 : n = __rte_ring_headtail_move_head_st(&r->cons.ht, &r->stage[ns].ht,
429 : : 0, num, behavior, &head, &next,
430 : : &avail);
431 : : break;
432 : 4 : case RTE_RING_SYNC_MT_HTS:
433 : 4 : n = __rte_ring_hts_move_head(&r->cons.hts, &r->stage[ns].ht,
434 : : 0, num, behavior, &head, &avail);
435 : 4 : break;
436 : : default:
437 : : /* unsupported mode, shouldn't be here */
438 : : RTE_ASSERT(0);
439 : : avail = 0;
440 : : n = 0;
441 : : }
442 : :
443 : : /* we have some elems to consume */
444 [ + + ]: 8 : if (n != 0)
445 : 4 : __dequeue_elems(r, objs, meta, head, n);
446 : :
447 [ + + ]: 8 : if (available != NULL)
448 : 4 : *available = avail - n;
449 : 8 : return n;
450 : : }
451 : :
452 : :
453 : : static inline void
454 : 4 : soring_dequeue_finish(struct rte_soring *r, uint32_t num)
455 : : {
456 : : uint32_t n, tail;
457 : :
458 : : RTE_ASSERT(r != NULL && r->nb_stage > 0);
459 : :
460 [ + + - ]: 4 : switch (r->cons.ht.sync_type) {
461 [ - + ]: 2 : case RTE_RING_SYNC_ST:
462 : : n = __rte_ring_st_get_tail(&r->cons.ht, &tail, num);
463 : : __rte_ring_st_set_head_tail(&r->cons.ht, tail, n, 0);
464 : : break;
465 [ - + ]: 2 : case RTE_RING_SYNC_MT_HTS:
466 : : n = __rte_ring_hts_get_tail(&r->cons.hts, &tail, num);
467 : : __rte_ring_hts_set_head_tail(&r->cons.hts, tail, n, 0);
468 : : break;
469 : 4 : default:
470 : : /* unsupported mode, shouldn't be here */
471 : : RTE_ASSERT(0);
472 : : }
473 : 4 : }
474 : :
475 : : /*
476 : : * Verify internal SORING state.
477 : : * WARNING: if expected value is not equal to actual one, it means that for
478 : : * whatever reason SORING data constancy is broken. That is a very serious
479 : : * problem that most likely will cause race-conditions, memory corruption,
480 : : * program crash.
481 : : * To ease debugging it user might rebuild ring library with
482 : : * RTE_SORING_DEBUG enabled.
483 : : */
484 : : static __rte_always_inline void
485 : : soring_verify_state(const struct rte_soring *r, uint32_t stage, uint32_t idx,
486 : : const char *msg, union soring_state val, union soring_state exp)
487 : : {
488 : 20014 : if (val.raw != exp.raw) {
489 : : #ifdef RTE_SORING_DEBUG
490 : : rte_soring_dump(stderr, r);
491 : : rte_panic("line:%d from:%s: soring=%p, stage=%#x, idx=%#x, "
492 : : "expected={.stnum=%#x, .ftoken=%#x}, "
493 : : "actual={.stnum=%#x, .ftoken=%#x};\n",
494 : : __LINE__, msg, r, stage, idx,
495 : : exp.stnum, exp.ftoken,
496 : : val.stnum, val.ftoken);
497 : : #else
498 : 0 : SORING_LOG(EMERG, "from:%s: soring=%p, stage=%#x, idx=%#x, "
499 : : "expected={.stnum=%#x, .ftoken=%#x}, "
500 : : "actual={.stnum=%#x, .ftoken=%#x};",
501 : : msg, r, stage, idx,
502 : : exp.stnum, exp.ftoken,
503 : : val.stnum, val.ftoken);
504 : : #endif
505 : : }
506 : : }
507 : :
508 : : /* check and update state ring at acquire op*/
509 : : static __rte_always_inline void
510 : : acquire_state_update(const struct rte_soring *r, uint32_t stage, uint32_t idx,
511 : : uint32_t ftoken, uint32_t num)
512 : : {
513 : : union soring_state st;
514 : : const union soring_state est = {.raw = 0};
515 : :
516 : 10007 : st.raw = rte_atomic_load_explicit(&r->state[idx].raw,
517 : : rte_memory_order_relaxed);
518 : : soring_verify_state(r, stage, idx, __func__, st, est);
519 : :
520 : 10007 : st.ftoken = ftoken;
521 : 10007 : st.stnum = (SORING_ST_START | num);
522 : :
523 : 10007 : rte_atomic_store_explicit(&r->state[idx].raw, st.raw,
524 : : rte_memory_order_relaxed);
525 : : }
526 : :
527 : : static __rte_always_inline uint32_t
528 : : soring_acquire(struct rte_soring *r, void *objs, void *meta,
529 : : uint32_t stage, uint32_t num, enum rte_ring_queue_behavior behavior,
530 : : uint32_t *ftoken, uint32_t *available)
531 : : {
532 : : uint32_t avail, head, idx, n, next, reqn;
533 : : struct soring_stage *pstg;
534 : : struct soring_stage_headtail *cons;
535 : :
536 : : RTE_ASSERT(r != NULL && stage < r->nb_stage);
537 : : RTE_ASSERT(meta == NULL || r->meta != NULL);
538 : :
539 : 10007 : cons = &r->stage[stage].sht;
540 : :
541 : 10007 : if (stage == 0)
542 : : n = __rte_soring_stage_move_head(cons, &r->prod.ht, 0, num,
543 : : behavior, &head, &next, &avail);
544 : : else {
545 : 10000 : pstg = r->stage + stage - 1;
546 : :
547 : : /* try to grab exactly @num elems */
548 : : n = __rte_soring_stage_move_head(cons, &pstg->ht, 0, num,
549 : : RTE_RING_QUEUE_FIXED, &head, &next, &avail);
550 [ - + - - : 10000 : if (n == 0) {
- + - - ]
551 : : /* try to finalize some elems from previous stage */
552 [ # # # # : 0 : n = __rte_soring_stage_finalize(&pstg->sht, stage - 1,
# # # # ]
553 : : r->state, r->mask, 2 * num);
554 : 0 : avail += n;
555 : :
556 : : /* repeat attempt to grab elems */
557 : : reqn = (behavior == RTE_RING_QUEUE_FIXED) ? num : 0;
558 [ # # # # ]: 0 : if (avail >= reqn)
559 : : n = __rte_soring_stage_move_head(cons,
560 : : &pstg->ht, 0, num, behavior, &head,
561 : : &next, &avail);
562 : : else
563 : : n = 0;
564 : : }
565 : : }
566 : :
567 [ + - - - : 10007 : if (n != 0) {
+ - + - ]
568 : :
569 : 10007 : idx = head & r->mask;
570 [ - + - - : 10007 : *ftoken = SORING_FTKN_MAKE(head, stage);
- + - + ]
571 : :
572 : : /* check and update state value */
573 : : acquire_state_update(r, stage, idx, *ftoken, n);
574 : :
575 : : /* copy elems that are ready for given stage */
576 [ - + - - : 10007 : __rte_ring_do_dequeue_elems(objs, &r[1], r->size, idx,
- + + - ]
577 : : r->esize, n);
578 [ + - + - ]: 10003 : if (meta != NULL)
579 [ - + - + ]: 10003 : __rte_ring_do_dequeue_elems(meta, r->meta,
580 : : r->size, idx, r->msize, n);
581 : : }
582 : :
583 [ - + - - : 10007 : if (available != NULL)
- + - + ]
584 : 0 : *available = avail - n;
585 : : return n;
586 : : }
587 : :
588 : : static __rte_always_inline void
589 : : soring_release(struct rte_soring *r, const void *objs,
590 : : const void *meta, uint32_t stage, uint32_t n, uint32_t ftoken)
591 : : {
592 : : uint32_t idx, pos;
593 : : struct soring_stage *stg;
594 : : union soring_state st;
595 : :
596 : : const union soring_state est = {
597 : 10007 : .stnum = (SORING_ST_START | n),
598 : : .ftoken = ftoken,
599 : : };
600 : :
601 : : RTE_ASSERT(r != NULL && stage < r->nb_stage);
602 : : RTE_ASSERT(meta == NULL || r->meta != NULL);
603 : :
604 : 10007 : stg = r->stage + stage;
605 : :
606 : 10007 : pos = SORING_FTKN_POS(ftoken, stage);
607 : 10007 : idx = pos & r->mask;
608 : 10007 : st.raw = rte_atomic_load_explicit(&r->state[idx].raw,
609 : : rte_memory_order_relaxed);
610 : :
611 : : /* check state ring contents */
612 : 10007 : soring_verify_state(r, stage, idx, __func__, st, est);
613 : :
614 : : /* update contents of the ring, if necessary */
615 [ + + - + ]: 10007 : if (objs != NULL)
616 [ - + - - ]: 10000 : __rte_ring_do_enqueue_elems(&r[1], objs, r->size, idx,
617 : : r->esize, n);
618 [ + - ]: 10001 : if (meta != NULL)
619 [ - + ]: 10001 : __rte_ring_do_enqueue_elems(r->meta, meta, r->size, idx,
620 : : r->msize, n);
621 : :
622 : : /* set state to FINISH, make sure it is not reordered */
623 : : rte_atomic_thread_fence(rte_memory_order_release);
624 : :
625 : 10007 : st.stnum = SORING_ST_FINISH | n;
626 : 10007 : rte_atomic_store_explicit(&r->state[idx].raw, st.raw,
627 : : rte_memory_order_relaxed);
628 : :
629 : : /* now, try to do finalize() */
630 [ + - + - ]: 10007 : __rte_soring_stage_finalize(&stg->sht, stage, r->state, r->mask,
631 : : r->capacity);
632 : : }
633 : :
634 : : /*
635 : : * Public functions (data-path) start here.
636 : : */
637 : :
638 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_release, 25.03)
639 : : void
640 [ - + ]: 6 : rte_soring_release(struct rte_soring *r, const void *objs,
641 : : uint32_t stage, uint32_t n, uint32_t ftoken)
642 : : {
643 : : soring_release(r, objs, NULL, stage, n, ftoken);
644 : 6 : }
645 : :
646 : :
647 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_releasx, 25.03)
648 : : void
649 [ - + ]: 10001 : rte_soring_releasx(struct rte_soring *r, const void *objs,
650 : : const void *meta, uint32_t stage, uint32_t n, uint32_t ftoken)
651 : : {
652 : : soring_release(r, objs, meta, stage, n, ftoken);
653 : 10001 : }
654 : :
655 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_enqueue_bulk, 25.03)
656 : : uint32_t
657 : 1 : rte_soring_enqueue_bulk(struct rte_soring *r, const void *objs, uint32_t n,
658 : : uint32_t *free_space)
659 : : {
660 : 1 : return soring_enqueue(r, objs, NULL, n, RTE_RING_QUEUE_FIXED,
661 : : free_space);
662 : : }
663 : :
664 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_enqueux_bulk, 25.03)
665 : : uint32_t
666 : 0 : rte_soring_enqueux_bulk(struct rte_soring *r, const void *objs,
667 : : const void *meta, uint32_t n, uint32_t *free_space)
668 : : {
669 : 0 : return soring_enqueue(r, objs, meta, n, RTE_RING_QUEUE_FIXED,
670 : : free_space);
671 : : }
672 : :
673 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_enqueue_burst, 25.03)
674 : : uint32_t
675 : 3 : rte_soring_enqueue_burst(struct rte_soring *r, const void *objs, uint32_t n,
676 : : uint32_t *free_space)
677 : : {
678 : 3 : return soring_enqueue(r, objs, NULL, n, RTE_RING_QUEUE_VARIABLE,
679 : : free_space);
680 : : }
681 : :
682 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_enqueux_burst, 25.03)
683 : : uint32_t
684 : 2 : rte_soring_enqueux_burst(struct rte_soring *r, const void *objs,
685 : : const void *meta, uint32_t n, uint32_t *free_space)
686 : : {
687 : 2 : return soring_enqueue(r, objs, meta, n, RTE_RING_QUEUE_VARIABLE,
688 : : free_space);
689 : : }
690 : :
691 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeue_bulk, 25.03)
692 : : uint32_t
693 : 10002 : rte_soring_dequeue_bulk(struct rte_soring *r, void *objs, uint32_t num,
694 : : uint32_t *available)
695 : : {
696 : 10002 : return soring_dequeue(r, objs, NULL, num, RTE_RING_QUEUE_FIXED,
697 : : available);
698 : : }
699 : :
700 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeux_bulk, 25.03)
701 : : uint32_t
702 : 1 : rte_soring_dequeux_bulk(struct rte_soring *r, void *objs, void *meta,
703 : : uint32_t num, uint32_t *available)
704 : : {
705 : 1 : return soring_dequeue(r, objs, meta, num, RTE_RING_QUEUE_FIXED,
706 : : available);
707 : : }
708 : :
709 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeue_burst, 25.03)
710 : : uint32_t
711 : 4 : rte_soring_dequeue_burst(struct rte_soring *r, void *objs, uint32_t num,
712 : : uint32_t *available)
713 : : {
714 : 4 : return soring_dequeue(r, objs, NULL, num, RTE_RING_QUEUE_VARIABLE,
715 : : available);
716 : : }
717 : :
718 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeux_burst, 25.03)
719 : : uint32_t
720 : 1 : rte_soring_dequeux_burst(struct rte_soring *r, void *objs, void *meta,
721 : : uint32_t num, uint32_t *available)
722 : : {
723 : 1 : return soring_dequeue(r, objs, meta, num, RTE_RING_QUEUE_VARIABLE,
724 : : available);
725 : : }
726 : :
727 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_acquire_bulk, 25.03)
728 : : uint32_t
729 [ + - ]: 4 : rte_soring_acquire_bulk(struct rte_soring *r, void *objs,
730 : : uint32_t stage, uint32_t num, uint32_t *ftoken, uint32_t *available)
731 : : {
732 : 4 : return soring_acquire(r, objs, NULL, stage, num,
733 : : RTE_RING_QUEUE_FIXED, ftoken, available);
734 : : }
735 : :
736 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_acquirx_bulk, 25.03)
737 : : uint32_t
738 [ + + ]: 10000 : rte_soring_acquirx_bulk(struct rte_soring *r, void *objs, void *meta,
739 : : uint32_t stage, uint32_t num, uint32_t *ftoken, uint32_t *available)
740 : : {
741 : 10000 : return soring_acquire(r, objs, meta, stage, num,
742 : : RTE_RING_QUEUE_FIXED, ftoken, available);
743 : : }
744 : :
745 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_acquire_burst, 25.03)
746 : : uint32_t
747 [ # # ]: 0 : rte_soring_acquire_burst(struct rte_soring *r, void *objs,
748 : : uint32_t stage, uint32_t num, uint32_t *ftoken, uint32_t *available)
749 : : {
750 : 0 : return soring_acquire(r, objs, NULL, stage, num,
751 : : RTE_RING_QUEUE_VARIABLE, ftoken, available);
752 : : }
753 : :
754 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_acquirx_burst, 25.03)
755 : : uint32_t
756 [ + + ]: 3 : rte_soring_acquirx_burst(struct rte_soring *r, void *objs, void *meta,
757 : : uint32_t stage, uint32_t num, uint32_t *ftoken, uint32_t *available)
758 : : {
759 : 3 : return soring_acquire(r, objs, meta, stage, num,
760 : : RTE_RING_QUEUE_VARIABLE, ftoken, available);
761 : : }
762 : :
763 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_count, 25.03)
764 : : unsigned int
765 : 8 : rte_soring_count(const struct rte_soring *r)
766 : : {
767 : 8 : uint32_t prod_tail = r->prod.ht.tail;
768 : 8 : uint32_t cons_tail = r->cons.ht.tail;
769 : 8 : uint32_t count = (prod_tail - cons_tail) & r->mask;
770 : 8 : return (count > r->capacity) ? r->capacity : count;
771 : : }
772 : :
773 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_free_count, 25.03)
774 : : unsigned int
775 : 4 : rte_soring_free_count(const struct rte_soring *r)
776 : : {
777 : 4 : return r->capacity - rte_soring_count(r);
778 : : }
779 : :
780 : : /*
781 : : * SORING public peek API
782 : : */
783 : :
784 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_enqueue_bulk_start, 26.07)
785 : : uint32_t
786 : 0 : rte_soring_enqueue_bulk_start(struct rte_soring *r, uint32_t n,
787 : : uint32_t *free_space)
788 : : {
789 : 0 : return soring_enqueue_start(r, n, RTE_RING_QUEUE_FIXED, free_space);
790 : : }
791 : :
792 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_enqueue_burst_start, 26.07)
793 : : uint32_t
794 : 4 : rte_soring_enqueue_burst_start(struct rte_soring *r, uint32_t n,
795 : : uint32_t *free_space)
796 : : {
797 : 4 : return soring_enqueue_start(r, n, RTE_RING_QUEUE_VARIABLE, free_space);
798 : : }
799 : :
800 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_enqueue_finish, 26.07)
801 : : void
802 : 4 : rte_soring_enqueue_finish(struct rte_soring *r, const void *objs, uint32_t n)
803 : : {
804 : 4 : soring_enqueue_finish(r, objs, NULL, n);
805 : 4 : }
806 : :
807 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_enqueux_finish, 26.07)
808 : : void
809 : 0 : rte_soring_enqueux_finish(struct rte_soring *r, const void *objs,
810 : : const void *meta, uint32_t n)
811 : : {
812 : 0 : soring_enqueue_finish(r, objs, meta, n);
813 : 0 : }
814 : :
815 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeue_bulk_start, 26.07)
816 : : uint32_t
817 : 4 : rte_soring_dequeue_bulk_start(struct rte_soring *r, void *objs, uint32_t num,
818 : : uint32_t *available)
819 : : {
820 : 4 : return soring_dequeue_start(r, objs, NULL, num, RTE_RING_QUEUE_FIXED,
821 : : available);
822 : : }
823 : :
824 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeux_bulk_start, 26.07)
825 : : uint32_t
826 : 0 : rte_soring_dequeux_bulk_start(struct rte_soring *r, void *objs, void *meta,
827 : : uint32_t num, uint32_t *available)
828 : : {
829 : 0 : return soring_dequeue_start(r, objs, meta, num, RTE_RING_QUEUE_FIXED,
830 : : available);
831 : : }
832 : :
833 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeue_burst_start, 26.07)
834 : : uint32_t
835 : 4 : rte_soring_dequeue_burst_start(struct rte_soring *r, void *objs, uint32_t num,
836 : : uint32_t *available)
837 : : {
838 : 4 : return soring_dequeue_start(r, objs, NULL, num, RTE_RING_QUEUE_VARIABLE,
839 : : available);
840 : : }
841 : :
842 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeux_burst_start, 26.07)
843 : : uint32_t
844 : 0 : rte_soring_dequeux_burst_start(struct rte_soring *r, void *objs, void *meta,
845 : : uint32_t num, uint32_t *available)
846 : : {
847 : 0 : return soring_dequeue_start(r, objs, meta, num, RTE_RING_QUEUE_VARIABLE,
848 : : available);
849 : : }
850 : :
851 : : RTE_EXPORT_EXPERIMENTAL_SYMBOL(rte_soring_dequeue_finish, 26.07)
852 : : void
853 : 4 : rte_soring_dequeue_finish(struct rte_soring *r, uint32_t n)
854 : : {
855 : 4 : soring_dequeue_finish(r, n);
856 : 4 : }
|
