1 /* SPDX-License-Identifier: BSD-3-Clause 2 * 3 * Copyright (c) 2010-2017 Intel Corporation 4 * Copyright (c) 2007-2009 Kip Macy [email protected] 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_GENERIC_H_ 11 #define _RTE_RING_GENERIC_H_ 12 13 static __rte_always_inline void 14 update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val, 15 uint32_t single, uint32_t enqueue) 16 { 17 if (enqueue) 18 rte_smp_wmb(); 19 else 20 rte_smp_rmb(); 21 /* 22 * If there are other enqueues/dequeues in progress that preceded us, 23 * we need to wait for them to complete 24 */ 25 if (!single) 26 while (unlikely(ht->tail != old_val)) 27 rte_pause(); 28 29 ht->tail = new_val; 30 } 31 32 /** 33 * @internal This function updates the producer head for enqueue 34 * 35 * @param r 36 * A pointer to the ring structure 37 * @param is_sp 38 * Indicates whether multi-producer path is needed or not 39 * @param n 40 * The number of elements we will want to enqueue, i.e. how far should the 41 * head be moved 42 * @param behavior 43 * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring 44 * RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring 45 * @param old_head 46 * Returns head value as it was before the move, i.e. where enqueue starts 47 * @param new_head 48 * Returns the current/new head value i.e. where enqueue finishes 49 * @param free_entries 50 * Returns the amount of free space in the ring BEFORE head was moved 51 * @return 52 * Actual number of objects enqueued. 53 * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only. 54 */ 55 static __rte_always_inline unsigned int 56 __rte_ring_move_prod_head(struct rte_ring *r, unsigned int is_sp, 57 unsigned int n, enum rte_ring_queue_behavior behavior, 58 uint32_t *old_head, uint32_t *new_head, 59 uint32_t *free_entries) 60 { 61 const uint32_t capacity = r->capacity; 62 unsigned int max = n; 63 int success; 64 65 do { 66 /* Reset n to the initial burst count */ 67 n = max; 68 69 *old_head = r->prod.head; 70 71 /* add rmb barrier to avoid load/load reorder in weak 72 * memory model. It is noop on x86 73 */ 74 rte_smp_rmb(); 75 76 /* 77 * The subtraction is done between two unsigned 32bits value 78 * (the result is always modulo 32 bits even if we have 79 * *old_head > cons_tail). So 'free_entries' is always between 0 80 * and capacity (which is < size). 81 */ 82 *free_entries = (capacity + r->cons.tail - *old_head); 83 84 /* check that we have enough room in ring */ 85 if (unlikely(n > *free_entries)) 86 n = (behavior == RTE_RING_QUEUE_FIXED) ? 87 0 : *free_entries; 88 89 if (n == 0) 90 return 0; 91 92 *new_head = *old_head + n; 93 if (is_sp) 94 r->prod.head = *new_head, success = 1; 95 else 96 success = rte_atomic32_cmpset(&r->prod.head, 97 *old_head, *new_head); 98 } while (unlikely(success == 0)); 99 return n; 100 } 101 102 /** 103 * @internal This function updates the consumer head for dequeue 104 * 105 * @param r 106 * A pointer to the ring structure 107 * @param is_sc 108 * Indicates whether multi-consumer path is needed or not 109 * @param n 110 * The number of elements we will want to enqueue, i.e. how far should the 111 * head be moved 112 * @param behavior 113 * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring 114 * RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring 115 * @param old_head 116 * Returns head value as it was before the move, i.e. where dequeue starts 117 * @param new_head 118 * Returns the current/new head value i.e. where dequeue finishes 119 * @param entries 120 * Returns the number of entries in the ring BEFORE head was moved 121 * @return 122 * - Actual number of objects dequeued. 123 * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only. 124 */ 125 static __rte_always_inline unsigned int 126 __rte_ring_move_cons_head(struct rte_ring *r, unsigned int is_sc, 127 unsigned int n, enum rte_ring_queue_behavior behavior, 128 uint32_t *old_head, uint32_t *new_head, 129 uint32_t *entries) 130 { 131 unsigned int max = n; 132 int success; 133 134 /* move cons.head atomically */ 135 do { 136 /* Restore n as it may change every loop */ 137 n = max; 138 139 *old_head = r->cons.head; 140 141 /* add rmb barrier to avoid load/load reorder in weak 142 * memory model. It is noop on x86 143 */ 144 rte_smp_rmb(); 145 146 /* The subtraction is done between two unsigned 32bits value 147 * (the result is always modulo 32 bits even if we have 148 * cons_head > prod_tail). So 'entries' is always between 0 149 * and size(ring)-1. 150 */ 151 *entries = (r->prod.tail - *old_head); 152 153 /* Set the actual entries for dequeue */ 154 if (n > *entries) 155 n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries; 156 157 if (unlikely(n == 0)) 158 return 0; 159 160 *new_head = *old_head + n; 161 if (is_sc) { 162 r->cons.head = *new_head; 163 rte_smp_rmb(); 164 success = 1; 165 } else { 166 success = rte_atomic32_cmpset(&r->cons.head, *old_head, 167 *new_head); 168 } 169 } while (unlikely(success == 0)); 170 return n; 171 } 172 173 #endif /* _RTE_RING_GENERIC_H_ */ 174