1 /* SPDX-License-Identifier: BSD-3-Clause
3 * Copyright (c) 2010-2020 Intel Corporation
4 * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
6 * Derived from FreeBSD's bufring.h
7 * Used as BSD-3 Licensed with permission from Kip Macy.
10 #ifndef _RTE_RING_RTS_H_
11 #define _RTE_RING_RTS_H_
14 * @file rte_ring_rts.h
15 * It is not recommended to include this file directly.
16 * Please include <rte_ring.h> instead.
18 * Contains functions for Relaxed Tail Sync (RTS) ring mode.
19 * The main idea remains the same as for our original MP/MC synchronization
21 * The main difference is that tail value is increased not
22 * by every thread that finished enqueue/dequeue,
23 * but only by the current last one doing enqueue/dequeue.
24 * That allows threads to skip spinning on tail value,
25 * leaving actual tail value change to last thread at a given instance.
26 * RTS requires 2 64-bit CAS for each enqueue(/dequeue) operation:
27 * one for head update, second for tail update.
28 * As a gain it allows thread to avoid spinning/waiting on tail value.
29 * In comparison original MP/MC algorithm requires one 32-bit CAS
30 * for head update and waiting/spinning on tail value.
33 * - introduce update counter (cnt) for both head and tail.
34 * - increment head.cnt for each head.value update
35 * - write head.value and head.cnt atomically (64-bit CAS)
36 * - move tail.value ahead only when tail.cnt + 1 == head.cnt
37 * (indicating that this is the last thread updating the tail)
38 * - increment tail.cnt when each enqueue/dequeue op finishes
39 * (no matter if tail.value going to change or not)
40 * - write tail.value and tail.cnt atomically (64-bit CAS)
42 * To avoid producer/consumer starvation:
43 * - limit max allowed distance between head and tail value (HTD_MAX).
44 * I.E. thread is allowed to proceed with changing head.value,
45 * only when: head.value - tail.value <= HTD_MAX
46 * HTD_MAX is an optional parameter.
47 * With HTD_MAX == 0 we'll have fully serialized ring -
48 * i.e. only one thread at a time will be able to enqueue/dequeue
50 * With HTD_MAX >= ring.capacity - no limitation.
51 * By default HTD_MAX == ring.capacity / 8.
58 #include <rte_ring_rts_elem_pvt.h>
61 * Enqueue several objects on the RTS ring (multi-producers safe).
64 * A pointer to the ring structure.
66 * A pointer to a table of objects.
68 * The size of ring element, in bytes. It must be a multiple of 4.
69 * This must be the same value used while creating the ring. Otherwise
70 * the results are undefined.
72 * The number of objects to add in the ring from the obj_table.
74 * if non-NULL, returns the amount of space in the ring after the
75 * enqueue operation has finished.
77 * The number of objects enqueued, either 0 or n
79 static __rte_always_inline unsigned int
80 rte_ring_mp_rts_enqueue_bulk_elem(struct rte_ring *r, const void *obj_table,
81 unsigned int esize, unsigned int n, unsigned int *free_space)
83 return __rte_ring_do_rts_enqueue_elem(r, obj_table, esize, n,
84 RTE_RING_QUEUE_FIXED, free_space);
88 * Dequeue several objects from an RTS ring (multi-consumers safe).
91 * A pointer to the ring structure.
93 * A pointer to a table of objects that will be filled.
95 * The size of ring element, in bytes. It must be a multiple of 4.
96 * This must be the same value used while creating the ring. Otherwise
97 * the results are undefined.
99 * The number of objects to dequeue from the ring to the obj_table.
101 * If non-NULL, returns the number of remaining ring entries after the
102 * dequeue has finished.
104 * The number of objects dequeued, either 0 or n
106 static __rte_always_inline unsigned int
107 rte_ring_mc_rts_dequeue_bulk_elem(struct rte_ring *r, void *obj_table,
108 unsigned int esize, unsigned int n, unsigned int *available)
110 return __rte_ring_do_rts_dequeue_elem(r, obj_table, esize, n,
111 RTE_RING_QUEUE_FIXED, available);
115 * Enqueue several objects on the RTS ring (multi-producers safe).
118 * A pointer to the ring structure.
120 * A pointer to a table of objects.
122 * The size of ring element, in bytes. It must be a multiple of 4.
123 * This must be the same value used while creating the ring. Otherwise
124 * the results are undefined.
126 * The number of objects to add in the ring from the obj_table.
128 * if non-NULL, returns the amount of space in the ring after the
129 * enqueue operation has finished.
131 * - n: Actual number of objects enqueued.
133 static __rte_always_inline unsigned int
134 rte_ring_mp_rts_enqueue_burst_elem(struct rte_ring *r, const void *obj_table,
135 unsigned int esize, unsigned int n, unsigned int *free_space)
137 return __rte_ring_do_rts_enqueue_elem(r, obj_table, esize, n,
138 RTE_RING_QUEUE_VARIABLE, free_space);
142 * Dequeue several objects from an RTS ring (multi-consumers safe).
143 * When the requested objects are more than the available objects,
144 * only dequeue the actual number of objects.
147 * A pointer to the ring structure.
149 * A pointer to a table of objects that will be filled.
151 * The size of ring element, in bytes. It must be a multiple of 4.
152 * This must be the same value used while creating the ring. Otherwise
153 * the results are undefined.
155 * The number of objects to dequeue from the ring to the obj_table.
157 * If non-NULL, returns the number of remaining ring entries after the
158 * dequeue has finished.
160 * - n: Actual number of objects dequeued, 0 if ring is empty
162 static __rte_always_inline unsigned int
163 rte_ring_mc_rts_dequeue_burst_elem(struct rte_ring *r, void *obj_table,
164 unsigned int esize, unsigned int n, unsigned int *available)
166 return __rte_ring_do_rts_dequeue_elem(r, obj_table, esize, n,
167 RTE_RING_QUEUE_VARIABLE, available);
171 * Enqueue several objects on the RTS ring (multi-producers safe).
174 * A pointer to the ring structure.
176 * A pointer to a table of void * pointers (objects).
178 * The number of objects to add in the ring from the obj_table.
180 * if non-NULL, returns the amount of space in the ring after the
181 * enqueue operation has finished.
183 * The number of objects enqueued, either 0 or n
185 static __rte_always_inline unsigned int
186 rte_ring_mp_rts_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
187 unsigned int n, unsigned int *free_space)
189 return rte_ring_mp_rts_enqueue_bulk_elem(r, obj_table,
190 sizeof(uintptr_t), n, free_space);
194 * Dequeue several objects from an RTS ring (multi-consumers safe).
197 * A pointer to the ring structure.
199 * A pointer to a table of void * pointers (objects) that will be filled.
201 * The number of objects to dequeue from the ring to the obj_table.
203 * If non-NULL, returns the number of remaining ring entries after the
204 * dequeue has finished.
206 * The number of objects dequeued, either 0 or n
208 static __rte_always_inline unsigned int
209 rte_ring_mc_rts_dequeue_bulk(struct rte_ring *r, void **obj_table,
210 unsigned int n, unsigned int *available)
212 return rte_ring_mc_rts_dequeue_bulk_elem(r, obj_table,
213 sizeof(uintptr_t), n, available);
217 * Enqueue several objects on the RTS ring (multi-producers safe).
220 * A pointer to the ring structure.
222 * A pointer to a table of void * pointers (objects).
224 * The number of objects to add in the ring from the obj_table.
226 * if non-NULL, returns the amount of space in the ring after the
227 * enqueue operation has finished.
229 * - n: Actual number of objects enqueued.
231 static __rte_always_inline unsigned int
232 rte_ring_mp_rts_enqueue_burst(struct rte_ring *r, void * const *obj_table,
233 unsigned int n, unsigned int *free_space)
235 return rte_ring_mp_rts_enqueue_burst_elem(r, obj_table,
236 sizeof(uintptr_t), n, free_space);
240 * Dequeue several objects from an RTS ring (multi-consumers safe).
241 * When the requested objects are more than the available objects,
242 * only dequeue the actual number of objects.
245 * A pointer to the ring structure.
247 * A pointer to a table of void * pointers (objects) that will be filled.
249 * The number of objects to dequeue from the ring to the obj_table.
251 * If non-NULL, returns the number of remaining ring entries after the
252 * dequeue has finished.
254 * - n: Actual number of objects dequeued, 0 if ring is empty
256 static __rte_always_inline unsigned int
257 rte_ring_mc_rts_dequeue_burst(struct rte_ring *r, void **obj_table,
258 unsigned int n, unsigned int *available)
260 return rte_ring_mc_rts_dequeue_burst_elem(r, obj_table,
261 sizeof(uintptr_t), n, available);
265 * Return producer max Head-Tail-Distance (HTD).
268 * A pointer to the ring structure.
270 * Producer HTD value, if producer is set in appropriate sync mode,
271 * or UINT32_MAX otherwise.
273 static inline uint32_t
274 rte_ring_get_prod_htd_max(const struct rte_ring *r)
276 if (r->prod.sync_type == RTE_RING_SYNC_MT_RTS)
277 return r->rts_prod.htd_max;
282 * Set producer max Head-Tail-Distance (HTD).
283 * Note that producer has to use appropriate sync mode (RTS).
286 * A pointer to the ring structure.
288 * new HTD value to setup.
290 * Zero on success, or negative error code otherwise.
293 rte_ring_set_prod_htd_max(struct rte_ring *r, uint32_t v)
295 if (r->prod.sync_type != RTE_RING_SYNC_MT_RTS)
298 r->rts_prod.htd_max = v;
303 * Return consumer max Head-Tail-Distance (HTD).
306 * A pointer to the ring structure.
308 * Consumer HTD value, if consumer is set in appropriate sync mode,
309 * or UINT32_MAX otherwise.
311 static inline uint32_t
312 rte_ring_get_cons_htd_max(const struct rte_ring *r)
314 if (r->cons.sync_type == RTE_RING_SYNC_MT_RTS)
315 return r->rts_cons.htd_max;
320 * Set consumer max Head-Tail-Distance (HTD).
321 * Note that consumer has to use appropriate sync mode (RTS).
324 * A pointer to the ring structure.
326 * new HTD value to setup.
328 * Zero on success, or negative error code otherwise.
331 rte_ring_set_cons_htd_max(struct rte_ring *r, uint32_t v)
333 if (r->cons.sync_type != RTE_RING_SYNC_MT_RTS)
336 r->rts_cons.htd_max = v;
344 #endif /* _RTE_RING_RTS_H_ */