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