DPDK  24.07.0
rte_ring_hts_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_HTS_ELEM_PVT_H_
11 #define _RTE_RING_HTS_ELEM_PVT_H_
12 
13 #include <rte_stdatomic.h>
14 
26 static __rte_always_inline void
27 __rte_ring_hts_update_tail(struct rte_ring_hts_headtail *ht, uint32_t old_tail,
28  uint32_t num, uint32_t enqueue)
29 {
30  uint32_t tail;
31 
32  RTE_SET_USED(enqueue);
33 
34  tail = old_tail + num;
35  rte_atomic_store_explicit(&ht->ht.pos.tail, tail, rte_memory_order_release);
36 }
37 
43 static __rte_always_inline void
44 __rte_ring_hts_head_wait(const struct rte_ring_hts_headtail *ht,
45  union __rte_ring_hts_pos *p)
46 {
47  while (p->pos.head != p->pos.tail) {
48  rte_pause();
49  p->raw = rte_atomic_load_explicit(&ht->ht.raw, rte_memory_order_acquire);
50  }
51 }
52 
56 static __rte_always_inline unsigned int
57 __rte_ring_hts_move_prod_head(struct rte_ring *r, unsigned int num,
58  enum rte_ring_queue_behavior behavior, uint32_t *old_head,
59  uint32_t *free_entries)
60 {
61  uint32_t n;
62  union __rte_ring_hts_pos np, op;
63 
64  const uint32_t capacity = r->capacity;
65 
66  op.raw = rte_atomic_load_explicit(&r->hts_prod.ht.raw, rte_memory_order_acquire);
67 
68  do {
69  /* Reset n to the initial burst count */
70  n = num;
71 
72  /*
73  * wait for tail to be equal to head,
74  * make sure that we read prod head/tail *before*
75  * reading cons tail.
76  */
77  __rte_ring_hts_head_wait(&r->hts_prod, &op);
78 
79  /*
80  * The subtraction is done between two unsigned 32bits value
81  * (the result is always modulo 32 bits even if we have
82  * *old_head > cons_tail). So 'free_entries' is always between 0
83  * and capacity (which is < size).
84  */
85  *free_entries = capacity + r->cons.tail - op.pos.head;
86 
87  /* check that we have enough room in ring */
88  if (unlikely(n > *free_entries))
89  n = (behavior == RTE_RING_QUEUE_FIXED) ?
90  0 : *free_entries;
91 
92  if (n == 0)
93  break;
94 
95  np.pos.tail = op.pos.tail;
96  np.pos.head = op.pos.head + n;
97 
98  /*
99  * this CAS(ACQUIRE, ACQUIRE) serves as a hoist barrier to prevent:
100  * - OOO reads of cons tail value
101  * - OOO copy of elems from the ring
102  */
103  } while (rte_atomic_compare_exchange_strong_explicit(&r->hts_prod.ht.raw,
104  (uint64_t *)(uintptr_t)&op.raw, np.raw,
105  rte_memory_order_acquire, rte_memory_order_acquire) == 0);
106 
107  *old_head = op.pos.head;
108  return n;
109 }
110 
114 static __rte_always_inline unsigned int
115 __rte_ring_hts_move_cons_head(struct rte_ring *r, unsigned int num,
116  enum rte_ring_queue_behavior behavior, uint32_t *old_head,
117  uint32_t *entries)
118 {
119  uint32_t n;
120  union __rte_ring_hts_pos np, op;
121 
122  op.raw = rte_atomic_load_explicit(&r->hts_cons.ht.raw, rte_memory_order_acquire);
123 
124  /* move cons.head atomically */
125  do {
126  /* Restore n as it may change every loop */
127  n = num;
128 
129  /*
130  * wait for tail to be equal to head,
131  * make sure that we read cons head/tail *before*
132  * reading prod tail.
133  */
134  __rte_ring_hts_head_wait(&r->hts_cons, &op);
135 
136  /* The subtraction is done between two unsigned 32bits value
137  * (the result is always modulo 32 bits even if we have
138  * cons_head > prod_tail). So 'entries' is always between 0
139  * and size(ring)-1.
140  */
141  *entries = r->prod.tail - op.pos.head;
142 
143  /* Set the actual entries for dequeue */
144  if (n > *entries)
145  n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;
146 
147  if (unlikely(n == 0))
148  break;
149 
150  np.pos.tail = op.pos.tail;
151  np.pos.head = op.pos.head + n;
152 
153  /*
154  * this CAS(ACQUIRE, ACQUIRE) serves as a hoist barrier to prevent:
155  * - OOO reads of prod tail value
156  * - OOO copy of elems from the ring
157  */
158  } while (rte_atomic_compare_exchange_strong_explicit(&r->hts_cons.ht.raw,
159  (uint64_t *)(uintptr_t)&op.raw, np.raw,
160  rte_memory_order_acquire, rte_memory_order_acquire) == 0);
161 
162  *old_head = op.pos.head;
163  return n;
164 }
165 
188 static __rte_always_inline unsigned int
189 __rte_ring_do_hts_enqueue_elem(struct rte_ring *r, const void *obj_table,
190  uint32_t esize, uint32_t n, enum rte_ring_queue_behavior behavior,
191  uint32_t *free_space)
192 {
193  uint32_t free, head;
194 
195  n = __rte_ring_hts_move_prod_head(r, n, behavior, &head, &free);
196 
197  if (n != 0) {
198  __rte_ring_enqueue_elems(r, head, obj_table, esize, n);
199  __rte_ring_hts_update_tail(&r->hts_prod, head, n, 1);
200  }
201 
202  if (free_space != NULL)
203  *free_space = free - n;
204  return n;
205 }
206 
229 static __rte_always_inline unsigned int
230 __rte_ring_do_hts_dequeue_elem(struct rte_ring *r, void *obj_table,
231  uint32_t esize, uint32_t n, enum rte_ring_queue_behavior behavior,
232  uint32_t *available)
233 {
234  uint32_t entries, head;
235 
236  n = __rte_ring_hts_move_cons_head(r, n, behavior, &head, &entries);
237 
238  if (n != 0) {
239  __rte_ring_dequeue_elems(r, head, obj_table, esize, n);
240  __rte_ring_hts_update_tail(&r->hts_cons, head, n, 0);
241  }
242 
243  if (available != NULL)
244  *available = entries - n;
245  return n;
246 }
247 
248 #endif /* _RTE_RING_HTS_ELEM_PVT_H_ */
#define __rte_always_inline
Definition: rte_common.h:370
rte_ring_queue_behavior
Definition: rte_ring_core.h:44
#define unlikely(x)
static void rte_pause(void)
uint32_t capacity
#define RTE_SET_USED(x)
Definition: rte_common.h:187