DPDK  25.03.0
rte_ring_rts_elem_pvt.h
Go to the documentation of this file.
1 /* SPDX-License-Identifier: BSD-3-Clause
2  *
3  * Copyright (c) 2010-2020 Intel Corporation
4  * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
5  * All rights reserved.
6  * Derived from FreeBSD's bufring.h
7  * Used as BSD-3 Licensed with permission from Kip Macy.
8  */
9 
10 #ifndef _RTE_RING_RTS_ELEM_PVT_H_
11 #define _RTE_RING_RTS_ELEM_PVT_H_
12 
24 static __rte_always_inline void
25 __rte_ring_rts_update_tail(struct rte_ring_rts_headtail *ht)
26 {
27  union __rte_ring_rts_poscnt h, ot, nt;
28 
29  /*
30  * If there are other enqueues/dequeues in progress that
31  * might preceded us, then don't update tail with new value.
32  */
33 
34  ot.raw = rte_atomic_load_explicit(&ht->tail.raw, rte_memory_order_acquire);
35 
36  do {
37  /* on 32-bit systems we have to do atomic read here */
38  h.raw = rte_atomic_load_explicit(&ht->head.raw, rte_memory_order_relaxed);
39 
40  nt.raw = ot.raw;
41  if (++nt.val.cnt == h.val.cnt)
42  nt.val.pos = h.val.pos;
43 
44  } while (rte_atomic_compare_exchange_strong_explicit(&ht->tail.raw,
45  (uint64_t *)(uintptr_t)&ot.raw, nt.raw,
46  rte_memory_order_release, rte_memory_order_acquire) == 0);
47 }
48 
53 static __rte_always_inline void
54 __rte_ring_rts_head_wait(const struct rte_ring_rts_headtail *ht,
55  union __rte_ring_rts_poscnt *h)
56 {
57  uint32_t max;
58 
59  max = ht->htd_max;
60 
61  while (h->val.pos - ht->tail.val.pos > max) {
62  rte_pause();
63  h->raw = rte_atomic_load_explicit(&ht->head.raw, rte_memory_order_acquire);
64  }
65 }
66 
91 static __rte_always_inline uint32_t
92 __rte_ring_rts_move_head(struct rte_ring_rts_headtail *d,
93  const struct rte_ring_headtail *s, uint32_t capacity, uint32_t num,
94  enum rte_ring_queue_behavior behavior, uint32_t *old_head,
95  uint32_t *entries)
96 {
97  uint32_t n;
98  union __rte_ring_rts_poscnt nh, oh;
99 
100  oh.raw = rte_atomic_load_explicit(&d->head.raw,
101  rte_memory_order_acquire);
102 
103  do {
104  /* Reset n to the initial burst count */
105  n = num;
106 
107  /*
108  * wait for prod head/tail distance,
109  * make sure that we read prod head *before*
110  * reading cons tail.
111  */
112  __rte_ring_rts_head_wait(d, &oh);
113 
114  /*
115  * The subtraction is done between two unsigned 32bits value
116  * (the result is always modulo 32 bits even if we have
117  * *old_head > cons_tail). So 'entries' is always between 0
118  * and capacity (which is < size).
119  */
120  *entries = capacity + s->tail - oh.val.pos;
121 
122  /* check that we have enough room in ring */
123  if (unlikely(n > *entries))
124  n = (behavior == RTE_RING_QUEUE_FIXED) ?
125  0 : *entries;
126 
127  if (n == 0)
128  break;
129 
130  nh.val.pos = oh.val.pos + n;
131  nh.val.cnt = oh.val.cnt + 1;
132 
133  /*
134  * this CAS(ACQUIRE, ACQUIRE) serves as a hoist barrier to prevent:
135  * - OOO reads of cons tail value
136  * - OOO copy of elems to the ring
137  */
138  } while (rte_atomic_compare_exchange_strong_explicit(&d->head.raw,
139  (uint64_t *)(uintptr_t)&oh.raw, nh.raw,
140  rte_memory_order_acquire,
141  rte_memory_order_acquire) == 0);
142 
143  *old_head = oh.val.pos;
144  return n;
145 }
146 
150 static __rte_always_inline uint32_t
151 __rte_ring_rts_move_prod_head(struct rte_ring *r, uint32_t num,
152  enum rte_ring_queue_behavior behavior, uint32_t *old_head,
153  uint32_t *free_entries)
154 {
155  return __rte_ring_rts_move_head(&r->rts_prod, &r->cons,
156  r->capacity, num, behavior, old_head, free_entries);
157 }
158 
162 static __rte_always_inline unsigned int
163 __rte_ring_rts_move_cons_head(struct rte_ring *r, uint32_t num,
164  enum rte_ring_queue_behavior behavior, uint32_t *old_head,
165  uint32_t *entries)
166 {
167  return __rte_ring_rts_move_head(&r->rts_cons, &r->prod,
168  0, num, behavior, old_head, entries);
169 }
170 
193 static __rte_always_inline unsigned int
194 __rte_ring_do_rts_enqueue_elem(struct rte_ring *r, const void *obj_table,
195  uint32_t esize, uint32_t n, enum rte_ring_queue_behavior behavior,
196  uint32_t *free_space)
197 {
198  uint32_t free, head;
199 
200  n = __rte_ring_rts_move_prod_head(r, n, behavior, &head, &free);
201 
202  if (n != 0) {
203  __rte_ring_enqueue_elems(r, head, obj_table, esize, n);
204  __rte_ring_rts_update_tail(&r->rts_prod);
205  }
206 
207  if (free_space != NULL)
208  *free_space = free - n;
209  return n;
210 }
211 
234 static __rte_always_inline unsigned int
235 __rte_ring_do_rts_dequeue_elem(struct rte_ring *r, void *obj_table,
236  uint32_t esize, uint32_t n, enum rte_ring_queue_behavior behavior,
237  uint32_t *available)
238 {
239  uint32_t entries, head;
240 
241  n = __rte_ring_rts_move_cons_head(r, n, behavior, &head, &entries);
242 
243  if (n != 0) {
244  __rte_ring_dequeue_elems(r, head, obj_table, esize, n);
245  __rte_ring_rts_update_tail(&r->rts_cons);
246  }
247 
248  if (available != NULL)
249  *available = entries - n;
250  return n;
251 }
252 
253 #endif /* _RTE_RING_RTS_ELEM_PVT_H_ */
#define __rte_always_inline
Definition: rte_common.h:456
rte_ring_queue_behavior
Definition: rte_ring_core.h:40
#define unlikely(x)
static void rte_pause(void)
uint32_t capacity