Branch data Line data Source code
1 : : /* SPDX-License-Identifier: BSD-3-Clause
2 : : *
3 : : * Copyright (c) 2018-2020 Arm Limited
4 : : */
5 : :
6 : : #include <stdio.h>
7 : : #include <string.h>
8 : : #include <stdint.h>
9 : : #include <inttypes.h>
10 : : #include <errno.h>
11 : :
12 : : #include <rte_common.h>
13 : : #include <rte_log.h>
14 : : #include <rte_memory.h>
15 : : #include <rte_malloc.h>
16 : : #include <rte_errno.h>
17 : : #include <rte_ring_elem.h>
18 : :
19 : : #include "rte_rcu_qsbr.h"
20 : : #include "rcu_qsbr_pvt.h"
21 : :
22 : : #define RCU_LOG(level, fmt, args...) \
23 : : RTE_LOG_LINE(level, RCU, "%s(): " fmt, __func__, ## args)
24 : :
25 : : /* Get the memory size of QSBR variable */
26 : : size_t
27 : 46 : rte_rcu_qsbr_get_memsize(uint32_t max_threads)
28 : : {
29 : : size_t sz;
30 : :
31 [ + + ]: 46 : if (max_threads == 0) {
32 : 1 : RCU_LOG(ERR, "Invalid max_threads %u", max_threads);
33 : 1 : rte_errno = EINVAL;
34 : :
35 : 1 : return 1;
36 : : }
37 : :
38 : : sz = sizeof(struct rte_rcu_qsbr);
39 : :
40 : : /* Add the size of quiescent state counter array */
41 : 45 : sz += sizeof(struct rte_rcu_qsbr_cnt) * max_threads;
42 : :
43 : : /* Add the size of the registered thread ID bitmap array */
44 : 45 : sz += __RTE_QSBR_THRID_ARRAY_SIZE(max_threads);
45 : :
46 : 45 : return sz;
47 : : }
48 : :
49 : : /* Initialize a quiescent state variable */
50 : : int
51 : 33 : rte_rcu_qsbr_init(struct rte_rcu_qsbr *v, uint32_t max_threads)
52 : : {
53 : : size_t sz;
54 : :
55 [ + + ]: 33 : if (v == NULL) {
56 : 1 : RCU_LOG(ERR, "Invalid input parameter");
57 : 1 : rte_errno = EINVAL;
58 : :
59 : 1 : return 1;
60 : : }
61 : :
62 : 32 : sz = rte_rcu_qsbr_get_memsize(max_threads);
63 [ + - ]: 32 : if (sz == 1)
64 : : return 1;
65 : :
66 : : /* Set all the threads to offline */
67 : : memset(v, 0, sz);
68 : 32 : v->max_threads = max_threads;
69 : 32 : v->num_elems = RTE_ALIGN_MUL_CEIL(max_threads,
70 : : __RTE_QSBR_THRID_ARRAY_ELM_SIZE) /
71 : : __RTE_QSBR_THRID_ARRAY_ELM_SIZE;
72 : 32 : v->token = __RTE_QSBR_CNT_INIT;
73 : 32 : v->acked_token = __RTE_QSBR_CNT_INIT - 1;
74 : :
75 : 32 : return 0;
76 : : }
77 : :
78 : : /* Register a reader thread to report its quiescent state
79 : : * on a QS variable.
80 : : */
81 : : int
82 : 541 : rte_rcu_qsbr_thread_register(struct rte_rcu_qsbr *v, unsigned int thread_id)
83 : : {
84 : : unsigned int i, id, success;
85 : : uint64_t old_bmap, new_bmap;
86 : :
87 [ + + + + ]: 541 : if (v == NULL || thread_id >= v->max_threads) {
88 : 3 : RCU_LOG(ERR, "Invalid input parameter");
89 : 3 : rte_errno = EINVAL;
90 : :
91 : 3 : return 1;
92 : : }
93 : :
94 : : __RTE_RCU_IS_LOCK_CNT_ZERO(v, thread_id, ERR, "Lock counter %u",
95 : : v->qsbr_cnt[thread_id].lock_cnt);
96 : :
97 : 538 : id = thread_id & __RTE_QSBR_THRID_MASK;
98 : 538 : i = thread_id >> __RTE_QSBR_THRID_INDEX_SHIFT;
99 : :
100 : : /* Make sure that the counter for registered threads does not
101 : : * go out of sync. Hence, additional checks are required.
102 : : */
103 : : /* Check if the thread is already registered */
104 : 538 : old_bmap = rte_atomic_load_explicit(__RTE_QSBR_THRID_ARRAY_ELM(v, i),
105 : : rte_memory_order_relaxed);
106 [ + + ]: 538 : if (old_bmap & 1UL << id)
107 : : return 0;
108 : :
109 : : do {
110 : 537 : new_bmap = old_bmap | (1UL << id);
111 : 537 : success = rte_atomic_compare_exchange_strong_explicit(
112 : : __RTE_QSBR_THRID_ARRAY_ELM(v, i),
113 : : &old_bmap, new_bmap,
114 : : rte_memory_order_release, rte_memory_order_relaxed);
115 : :
116 [ + - ]: 537 : if (success)
117 : 537 : rte_atomic_fetch_add_explicit(&v->num_threads,
118 : : 1, rte_memory_order_relaxed);
119 [ # # ]: 0 : else if (old_bmap & (1UL << id))
120 : : /* Someone else registered this thread.
121 : : * Counter should not be incremented.
122 : : */
123 : : return 0;
124 [ - + ]: 537 : } while (success == 0);
125 : :
126 : : return 0;
127 : : }
128 : :
129 : : /* Remove a reader thread, from the list of threads reporting their
130 : : * quiescent state on a QS variable.
131 : : */
132 : : int
133 : 275 : rte_rcu_qsbr_thread_unregister(struct rte_rcu_qsbr *v, unsigned int thread_id)
134 : : {
135 : : unsigned int i, id, success;
136 : : uint64_t old_bmap, new_bmap;
137 : :
138 [ + + + + ]: 275 : if (v == NULL || thread_id >= v->max_threads) {
139 : 3 : RCU_LOG(ERR, "Invalid input parameter");
140 : 3 : rte_errno = EINVAL;
141 : :
142 : 3 : return 1;
143 : : }
144 : :
145 : : __RTE_RCU_IS_LOCK_CNT_ZERO(v, thread_id, ERR, "Lock counter %u",
146 : : v->qsbr_cnt[thread_id].lock_cnt);
147 : :
148 : 272 : id = thread_id & __RTE_QSBR_THRID_MASK;
149 : 272 : i = thread_id >> __RTE_QSBR_THRID_INDEX_SHIFT;
150 : :
151 : : /* Make sure that the counter for registered threads does not
152 : : * go out of sync. Hence, additional checks are required.
153 : : */
154 : : /* Check if the thread is already unregistered */
155 : 272 : old_bmap = rte_atomic_load_explicit(__RTE_QSBR_THRID_ARRAY_ELM(v, i),
156 : : rte_memory_order_relaxed);
157 [ + + ]: 272 : if (!(old_bmap & (1UL << id)))
158 : : return 0;
159 : :
160 : : do {
161 : 269 : new_bmap = old_bmap & ~(1UL << id);
162 : : /* Make sure any loads of the shared data structure are
163 : : * completed before removal of the thread from the list of
164 : : * reporting threads.
165 : : */
166 : 269 : success = rte_atomic_compare_exchange_strong_explicit(
167 : : __RTE_QSBR_THRID_ARRAY_ELM(v, i),
168 : : &old_bmap, new_bmap,
169 : : rte_memory_order_release, rte_memory_order_relaxed);
170 : :
171 [ + - ]: 269 : if (success)
172 : 269 : rte_atomic_fetch_sub_explicit(&v->num_threads,
173 : : 1, rte_memory_order_relaxed);
174 [ # # ]: 0 : else if (!(old_bmap & (1UL << id)))
175 : : /* Someone else unregistered this thread.
176 : : * Counter should not be incremented.
177 : : */
178 : : return 0;
179 [ - + ]: 269 : } while (success == 0);
180 : :
181 : : return 0;
182 : : }
183 : :
184 : : /* Wait till the reader threads have entered quiescent state. */
185 : : void
186 [ + + ]: 1546 : rte_rcu_qsbr_synchronize(struct rte_rcu_qsbr *v, unsigned int thread_id)
187 : : {
188 : : uint64_t t;
189 : :
190 : : RTE_ASSERT(v != NULL);
191 : :
192 : : t = rte_rcu_qsbr_start(v);
193 : :
194 : : /* If the current thread has readside critical section,
195 : : * update its quiescent state status.
196 : : */
197 [ + + ]: 1546 : if (thread_id != RTE_QSBR_THRID_INVALID)
198 : : rte_rcu_qsbr_quiescent(v, thread_id);
199 : :
200 : : /* Wait for other readers to enter quiescent state */
201 : : rte_rcu_qsbr_check(v, t, true);
202 : 1546 : }
203 : :
204 : : /* Dump the details of a single quiescent state variable to a file. */
205 : : int
206 : 6 : rte_rcu_qsbr_dump(FILE *f, struct rte_rcu_qsbr *v)
207 : : {
208 : : uint64_t bmap;
209 : : uint32_t i, t, id;
210 : :
211 [ + + ]: 6 : if (v == NULL || f == NULL) {
212 : 3 : RCU_LOG(ERR, "Invalid input parameter");
213 : 3 : rte_errno = EINVAL;
214 : :
215 : 3 : return 1;
216 : : }
217 : :
218 : : fprintf(f, "\nQuiescent State Variable @%p\n", v);
219 : :
220 : 3 : fprintf(f, " QS variable memory size = %zu\n",
221 : : rte_rcu_qsbr_get_memsize(v->max_threads));
222 : 3 : fprintf(f, " Given # max threads = %u\n", v->max_threads);
223 : 3 : fprintf(f, " Current # threads = %u\n", v->num_threads);
224 : :
225 : : fprintf(f, " Registered thread IDs = ");
226 [ + + ]: 9 : for (i = 0; i < v->num_elems; i++) {
227 : 6 : bmap = rte_atomic_load_explicit(__RTE_QSBR_THRID_ARRAY_ELM(v, i),
228 : : rte_memory_order_acquire);
229 : 6 : id = i << __RTE_QSBR_THRID_INDEX_SHIFT;
230 [ + + ]: 7 : while (bmap) {
231 : : t = rte_ctz64(bmap);
232 : 1 : fprintf(f, "%u ", id + t);
233 : :
234 : 1 : bmap &= ~(1UL << t);
235 : : }
236 : : }
237 : :
238 : : fprintf(f, "\n");
239 : :
240 : 3 : fprintf(f, " Token = %" PRIu64 "\n",
241 : 3 : rte_atomic_load_explicit(&v->token, rte_memory_order_acquire));
242 : :
243 : 3 : fprintf(f, " Least Acknowledged Token = %" PRIu64 "\n",
244 : 3 : rte_atomic_load_explicit(&v->acked_token, rte_memory_order_acquire));
245 : :
246 : : fprintf(f, "Quiescent State Counts for readers:\n");
247 [ + + ]: 9 : for (i = 0; i < v->num_elems; i++) {
248 : 6 : bmap = rte_atomic_load_explicit(__RTE_QSBR_THRID_ARRAY_ELM(v, i),
249 : : rte_memory_order_acquire);
250 : 6 : id = i << __RTE_QSBR_THRID_INDEX_SHIFT;
251 [ + + ]: 7 : while (bmap) {
252 : : t = rte_ctz64(bmap);
253 : 1 : fprintf(f, "thread ID = %u, count = %" PRIu64 ", lock count = %u\n",
254 : : id + t,
255 : 1 : rte_atomic_load_explicit(
256 : : &v->qsbr_cnt[id + t].cnt,
257 : : rte_memory_order_relaxed),
258 : 1 : rte_atomic_load_explicit(
259 : : &v->qsbr_cnt[id + t].lock_cnt,
260 : : rte_memory_order_relaxed));
261 : 1 : bmap &= ~(1UL << t);
262 : : }
263 : : }
264 : :
265 : : return 0;
266 : : }
267 : :
268 : : /* Create a queue used to store the data structure elements that can
269 : : * be freed later. This queue is referred to as 'defer queue'.
270 : : */
271 : : struct rte_rcu_qsbr_dq *
272 : 22 : rte_rcu_qsbr_dq_create(const struct rte_rcu_qsbr_dq_parameters *params)
273 : : {
274 : : struct rte_rcu_qsbr_dq *dq;
275 : : uint32_t qs_fifo_size;
276 : : unsigned int flags;
277 : :
278 [ + + + + ]: 22 : if (params == NULL || params->free_fn == NULL ||
279 [ + + + - ]: 19 : params->v == NULL || params->name == NULL ||
280 [ + + + + ]: 18 : params->size == 0 || params->esize == 0 ||
281 [ + + ]: 16 : (params->esize % 4 != 0)) {
282 : 9 : RCU_LOG(ERR, "Invalid input parameter");
283 : 9 : rte_errno = EINVAL;
284 : :
285 : 9 : return NULL;
286 : : }
287 : : /* If auto reclamation is configured, reclaim limit
288 : : * should be a valid value.
289 : : */
290 [ + - ]: 13 : if ((params->trigger_reclaim_limit <= params->size) &&
291 [ - + ]: 13 : (params->max_reclaim_size == 0)) {
292 : 0 : RCU_LOG(ERR,
293 : : "Invalid input parameter, size = %u, trigger_reclaim_limit = %u, "
294 : : "max_reclaim_size = %u",
295 : : params->size, params->trigger_reclaim_limit,
296 : : params->max_reclaim_size);
297 : 0 : rte_errno = EINVAL;
298 : :
299 : 0 : return NULL;
300 : : }
301 : :
302 : 13 : dq = rte_zmalloc(NULL, sizeof(struct rte_rcu_qsbr_dq),
303 : : RTE_CACHE_LINE_SIZE);
304 [ - + ]: 13 : if (dq == NULL) {
305 : 0 : rte_errno = ENOMEM;
306 : :
307 : 0 : return NULL;
308 : : }
309 : :
310 : : /* Decide the flags for the ring.
311 : : * If MT safety is requested, use RTS for ring enqueue as most
312 : : * use cases involve dq-enqueue happening on the control plane.
313 : : * Ring dequeue is always HTS due to the possibility of revert.
314 : : */
315 : : flags = RING_F_MP_RTS_ENQ;
316 [ + + ]: 13 : if (params->flags & RTE_RCU_QSBR_DQ_MT_UNSAFE)
317 : : flags = RING_F_SP_ENQ;
318 : 13 : flags |= RING_F_MC_HTS_DEQ;
319 : : /* round up qs_fifo_size to next power of two that is not less than
320 : : * max_size.
321 : : */
322 : 13 : qs_fifo_size = rte_align32pow2(params->size + 1);
323 : : /* Add token size to ring element size */
324 : 26 : dq->r = rte_ring_create_elem(params->name,
325 : 13 : __RTE_QSBR_TOKEN_SIZE + params->esize,
326 : : qs_fifo_size, SOCKET_ID_ANY, flags);
327 [ - + ]: 13 : if (dq->r == NULL) {
328 : 0 : RCU_LOG(ERR, "defer queue create failed");
329 : 0 : rte_free(dq);
330 : 0 : return NULL;
331 : : }
332 : :
333 : 13 : dq->v = params->v;
334 : 13 : dq->size = params->size;
335 : 13 : dq->esize = __RTE_QSBR_TOKEN_SIZE + params->esize;
336 : 13 : dq->trigger_reclaim_limit = params->trigger_reclaim_limit;
337 : 13 : dq->max_reclaim_size = params->max_reclaim_size;
338 : 13 : dq->free_fn = params->free_fn;
339 : 13 : dq->p = params->p;
340 : :
341 : 13 : return dq;
342 : : }
343 : :
344 : : /* Enqueue one resource to the defer queue to free after the grace
345 : : * period is over.
346 : : */
347 : 1098 : int rte_rcu_qsbr_dq_enqueue(struct rte_rcu_qsbr_dq *dq, void *e)
348 : 1098 : {
349 : : __rte_rcu_qsbr_dq_elem_t *dq_elem;
350 : : uint32_t cur_size;
351 : :
352 [ + + ]: 1098 : if (dq == NULL || e == NULL) {
353 : 3 : RCU_LOG(ERR, "Invalid input parameter");
354 : 3 : rte_errno = EINVAL;
355 : :
356 : 3 : return 1;
357 : : }
358 : :
359 : 1095 : char data[dq->esize];
360 : : dq_elem = (__rte_rcu_qsbr_dq_elem_t *)data;
361 : : /* Start the grace period */
362 [ + + ]: 1095 : dq_elem->token = rte_rcu_qsbr_start(dq->v);
363 : :
364 : : /* Reclaim resources if the queue size has hit the reclaim
365 : : * limit. This helps the queue from growing too large and
366 : : * allows time for reader threads to report their quiescent state.
367 : : */
368 [ + + ]: 1095 : cur_size = rte_ring_count(dq->r);
369 [ + + ]: 1095 : if (cur_size > dq->trigger_reclaim_limit) {
370 : 1033 : RCU_LOG(INFO, "Triggering reclamation");
371 : 1033 : rte_rcu_qsbr_dq_reclaim(dq, dq->max_reclaim_size,
372 : : NULL, NULL, NULL);
373 : : }
374 : :
375 : : /* Enqueue the token and resource. Generating the token and
376 : : * enqueuing (token + resource) on the queue is not an
377 : : * atomic operation. When the defer queue is shared by multiple
378 : : * writers, this might result in tokens enqueued out of order
379 : : * on the queue. So, some tokens might wait longer than they
380 : : * are required to be reclaimed.
381 : : */
382 [ - + + - : 1095 : memcpy(dq_elem->elem, e, dq->esize - __RTE_QSBR_TOKEN_SIZE);
- ]
383 : : /* Check the status as enqueue might fail since the other threads
384 : : * might have used up the freed space.
385 : : * Enqueue uses the configured flags when the DQ was created.
386 : : */
387 [ - + + - : 1095 : if (rte_ring_enqueue_elem(dq->r, data, dq->esize) != 0) {
- ]
388 : 8 : RCU_LOG(ERR, "Enqueue failed");
389 : : /* Note that the token generated above is not used.
390 : : * Other than wasting tokens, it should not cause any
391 : : * other issues.
392 : : */
393 : 8 : RCU_LOG(INFO, "Skipped enqueuing token = %" PRIu64, dq_elem->token);
394 : :
395 : 8 : rte_errno = ENOSPC;
396 : 8 : return 1;
397 : : }
398 : :
399 : 1087 : RCU_LOG(INFO, "Enqueued token = %" PRIu64, dq_elem->token);
400 : :
401 : 1087 : return 0;
402 : : }
403 : :
404 : : /* Reclaim resources from the defer queue. */
405 : : int
406 : 1058 : rte_rcu_qsbr_dq_reclaim(struct rte_rcu_qsbr_dq *dq, unsigned int n,
407 : : unsigned int *freed, unsigned int *pending,
408 : : unsigned int *available)
409 : 1058 : {
410 : : uint32_t cnt;
411 : : __rte_rcu_qsbr_dq_elem_t *dq_elem;
412 : :
413 [ + + ]: 1058 : if (dq == NULL || n == 0) {
414 : 2 : RCU_LOG(ERR, "Invalid input parameter");
415 : 2 : rte_errno = EINVAL;
416 : :
417 : 2 : return 1;
418 : : }
419 : :
420 : : cnt = 0;
421 : :
422 : 1056 : char data[dq->esize];
423 : : /* Check reader threads quiescent state and reclaim resources */
424 [ + + + + ]: 4216 : while (cnt < n &&
425 [ - + - ]: 2073 : rte_ring_dequeue_bulk_elem_start(dq->r, &data,
426 : : dq->esize, 1, available) != 0) {
427 : : dq_elem = (__rte_rcu_qsbr_dq_elem_t *)data;
428 : :
429 : : /* Reclaim the resource */
430 [ + + ]: 2058 : if (rte_rcu_qsbr_check(dq->v, dq_elem->token, false) != 1) {
431 [ - + - ]: 971 : rte_ring_dequeue_elem_finish(dq->r, 0);
432 : : break;
433 : : }
434 [ - + - ]: 1087 : rte_ring_dequeue_elem_finish(dq->r, 1);
435 : :
436 : 1087 : RCU_LOG(INFO, "Reclaimed token = %" PRIu64, dq_elem->token);
437 : :
438 : 1087 : dq->free_fn(dq->p, dq_elem->elem, 1);
439 : :
440 : 1087 : cnt++;
441 : : }
442 : :
443 : 1056 : RCU_LOG(INFO, "Reclaimed %u resources", cnt);
444 : :
445 [ - + ]: 1056 : if (freed != NULL)
446 : 0 : *freed = cnt;
447 [ + + ]: 1056 : if (pending != NULL)
448 : 17 : *pending = rte_ring_count(dq->r);
449 : :
450 : : return 0;
451 : : }
452 : :
453 : : /* Delete a defer queue. */
454 : : int
455 : 18 : rte_rcu_qsbr_dq_delete(struct rte_rcu_qsbr_dq *dq)
456 : : {
457 : : unsigned int pending;
458 : :
459 [ + + ]: 18 : if (dq == NULL) {
460 : 1 : RCU_LOG(DEBUG, "Invalid input parameter");
461 : :
462 : 1 : return 0;
463 : : }
464 : :
465 : : /* Reclaim all the resources */
466 : 17 : rte_rcu_qsbr_dq_reclaim(dq, ~0, NULL, &pending, NULL);
467 [ + + ]: 17 : if (pending != 0) {
468 : 4 : rte_errno = EAGAIN;
469 : :
470 : 4 : return 1;
471 : : }
472 : :
473 : 13 : rte_ring_free(dq->r);
474 : 13 : rte_free(dq);
475 : :
476 : 13 : return 0;
477 : : }
478 : :
479 [ - + ]: 235 : RTE_LOG_REGISTER_DEFAULT(rte_rcu_log_type, ERR);
|