lib/librte_ring/rte_ring_c11_mem.h

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