1 /* SPDX-License-Identifier: BSD-3-Clause
2 * Copyright(c) 2017 Intel Corporation
10 #include <rte_string_fns.h>
11 #include <rte_branch_prediction.h>
12 #include <rte_debug.h>
13 #include <rte_lcore.h>
15 #include <rte_malloc.h>
16 #include <rte_memcpy.h>
17 #include <rte_memory.h>
18 #include <rte_memzone.h>
19 #include <rte_atomic.h>
21 #include "opdl_ring.h"
24 #define LIB_NAME "opdl_ring"
26 #define OPDL_NAME_SIZE 64
29 #define OPDL_EVENT_MASK (0x00000000000FFFFFULL)
30 #define OPDL_FLOWID_MASK (0xFFFFF)
31 #define OPDL_OPA_MASK (0xFF)
32 #define OPDL_OPA_OFFSET (0x38)
34 /* Types of dependency between stages */
36 DEP_NONE = 0, /* no dependency */
37 DEP_DIRECT, /* stage has direct dependency */
38 DEP_INDIRECT, /* in-direct dependency through other stage(s) */
39 DEP_SELF, /* stage dependency on itself, used to detect loops */
42 /* Shared section of stage state.
43 * Care is needed when accessing and the layout is important, especially to
44 * limit the adjacent cache-line HW prefetcher from impacting performance.
47 /* Last known minimum sequence number of dependencies, used for multi
50 uint32_t available_seq;
51 char _pad1[RTE_CACHE_LINE_SIZE * 3];
52 uint32_t head; /* Head sequence number (for multi thread operation) */
53 char _pad2[RTE_CACHE_LINE_SIZE * 3];
54 struct opdl_stage *stage; /* back pointer */
55 uint32_t tail; /* Tail sequence number */
56 char _pad3[RTE_CACHE_LINE_SIZE * 2];
57 } __rte_cache_aligned;
59 /* A structure to keep track of "unfinished" claims. This is only used for
60 * stages that are threadsafe. Each lcore accesses its own instance of this
61 * structure to record the entries it has claimed. This allows one lcore to make
62 * multiple claims without being blocked by another. When disclaiming it moves
63 * forward the shared tail when the shared tail matches the tail value recorded
66 struct claim_manager {
67 uint32_t num_to_disclaim;
74 } claims[OPDL_DISCLAIMS_PER_LCORE];
75 } __rte_cache_aligned;
77 /* Context for each stage of opdl_ring.
78 * Calculations on sequence numbers need to be done with other uint32_t values
79 * so that results are modulus 2^32, and not undefined.
82 struct opdl_ring *t; /* back pointer, set at init */
83 uint32_t num_slots; /* Number of slots for entries, set at init */
84 uint32_t index; /* ID for this stage, set at init */
85 bool threadsafe; /* Set to 1 if this stage supports threadsafe use */
86 /* Last known min seq number of dependencies for used for single thread
89 uint32_t available_seq;
90 uint32_t head; /* Current head for single-thread operation */
91 uint32_t nb_instance; /* Number of instances */
92 uint32_t instance_id; /* ID of this stage instance */
93 uint16_t num_claimed; /* Number of slots claimed */
94 uint16_t num_event; /* Number of events */
95 uint32_t seq; /* sequence number */
96 uint32_t num_deps; /* Number of direct dependencies */
97 /* Keep track of all dependencies, used during init only */
98 enum dep_type *dep_tracking;
99 /* Direct dependencies of this stage */
100 struct shared_state **deps;
101 /* Other stages read this! */
102 struct shared_state shared __rte_cache_aligned;
103 /* For managing disclaims in multi-threaded processing stages */
104 struct claim_manager pending_disclaims[RTE_MAX_LCORE]
106 uint32_t shadow_head; /* Shadow head for single-thread operation */
107 uint32_t queue_id; /* ID of Queue which is assigned to this stage */
108 uint32_t pos; /* Atomic scan position */
109 } __rte_cache_aligned;
111 /* Context for opdl_ring */
113 char name[OPDL_NAME_SIZE]; /* OPDL queue instance name */
114 int socket; /* NUMA socket that memory is allocated on */
115 uint32_t num_slots; /* Number of slots for entries */
116 uint32_t mask; /* Mask for sequence numbers (num_slots - 1) */
117 uint32_t slot_size; /* Size of each slot in bytes */
118 uint32_t num_stages; /* Number of stages that have been added */
119 uint32_t max_num_stages; /* Max number of stages */
120 /* Stages indexed by ID */
121 struct opdl_stage *stages;
122 /* Memory for storing slot data */
123 uint8_t slots[0] __rte_cache_aligned;
127 /* Return input stage of a opdl_ring */
128 static __rte_always_inline struct opdl_stage *
129 input_stage(const struct opdl_ring *t)
131 return &t->stages[0];
134 /* Check if a stage is the input stage */
135 static __rte_always_inline bool
136 is_input_stage(const struct opdl_stage *s)
138 return s->index == 0;
141 /* Get slot pointer from sequence number */
142 static __rte_always_inline void *
143 get_slot(const struct opdl_ring *t, uint32_t n)
145 return (void *)(uintptr_t)&t->slots[(n & t->mask) * t->slot_size];
148 /* Find how many entries are available for processing */
149 static __rte_always_inline uint32_t
150 available(const struct opdl_stage *s)
152 if (s->threadsafe == true) {
153 uint32_t n = __atomic_load_n(&s->shared.available_seq,
155 __atomic_load_n(&s->shared.head,
158 /* Return 0 if available_seq needs to be updated */
159 return (n <= s->num_slots) ? n : 0;
162 /* Single threaded */
163 return s->available_seq - s->head;
166 /* Read sequence number of dependencies and find minimum */
167 static __rte_always_inline void
168 update_available_seq(struct opdl_stage *s)
171 uint32_t this_tail = s->shared.tail;
172 uint32_t min_seq = __atomic_load_n(&s->deps[0]->tail, __ATOMIC_ACQUIRE);
173 /* Input stage sequence numbers are greater than the sequence numbers of
174 * its dependencies so an offset of t->num_slots is needed when
175 * calculating available slots and also the condition which is used to
176 * determine the dependencies minimum sequence number must be reverted.
180 if (is_input_stage(s)) {
182 for (i = 1; i < s->num_deps; i++) {
183 uint32_t seq = __atomic_load_n(&s->deps[i]->tail,
185 if ((this_tail - seq) > (this_tail - min_seq))
190 for (i = 1; i < s->num_deps; i++) {
191 uint32_t seq = __atomic_load_n(&s->deps[i]->tail,
193 if ((seq - this_tail) < (min_seq - this_tail))
198 if (s->threadsafe == false)
199 s->available_seq = min_seq + wrap;
201 __atomic_store_n(&s->shared.available_seq, min_seq + wrap,
205 /* Wait until the number of available slots reaches number requested */
206 static __rte_always_inline void
207 wait_for_available(struct opdl_stage *s, uint32_t n)
209 while (available(s) < n) {
211 update_available_seq(s);
215 /* Return number of slots to process based on number requested and mode */
216 static __rte_always_inline uint32_t
217 num_to_process(struct opdl_stage *s, uint32_t n, bool block)
219 /* Don't read tail sequences of dependencies if not needed */
220 if (available(s) >= n)
223 update_available_seq(s);
225 if (block == false) {
226 uint32_t avail = available(s);
232 return (avail <= n) ? avail : n;
235 if (unlikely(n > s->num_slots)) {
236 PMD_DRV_LOG(ERR, "%u entries is more than max (%u)",
238 return 0; /* Avoid infinite loop */
241 wait_for_available(s, n);
245 /* Copy entries in to slots with wrap-around */
246 static __rte_always_inline void
247 copy_entries_in(struct opdl_ring *t, uint32_t start, const void *entries,
248 uint32_t num_entries)
250 uint32_t slot_size = t->slot_size;
251 uint32_t slot_index = start & t->mask;
253 if (slot_index + num_entries <= t->num_slots) {
254 rte_memcpy(get_slot(t, start), entries,
255 num_entries * slot_size);
257 uint32_t split = t->num_slots - slot_index;
259 rte_memcpy(get_slot(t, start), entries, split * slot_size);
260 rte_memcpy(get_slot(t, 0),
261 RTE_PTR_ADD(entries, split * slot_size),
262 (num_entries - split) * slot_size);
266 /* Copy entries out from slots with wrap-around */
267 static __rte_always_inline void
268 copy_entries_out(struct opdl_ring *t, uint32_t start, void *entries,
269 uint32_t num_entries)
271 uint32_t slot_size = t->slot_size;
272 uint32_t slot_index = start & t->mask;
274 if (slot_index + num_entries <= t->num_slots) {
275 rte_memcpy(entries, get_slot(t, start),
276 num_entries * slot_size);
278 uint32_t split = t->num_slots - slot_index;
280 rte_memcpy(entries, get_slot(t, start), split * slot_size);
281 rte_memcpy(RTE_PTR_ADD(entries, split * slot_size),
283 (num_entries - split) * slot_size);
287 /* Input function optimised for single thread */
288 static __rte_always_inline uint32_t
289 opdl_ring_input_singlethread(struct opdl_ring *t, const void *entries,
290 uint32_t num_entries, bool block)
292 struct opdl_stage *s = input_stage(t);
293 uint32_t head = s->head;
295 num_entries = num_to_process(s, num_entries, block);
296 if (num_entries == 0)
299 copy_entries_in(t, head, entries, num_entries);
301 s->head += num_entries;
302 __atomic_store_n(&s->shared.tail, s->head, __ATOMIC_RELEASE);
307 /* Convert head and tail of claim_manager into valid index */
308 static __rte_always_inline uint32_t
309 claim_mgr_index(uint32_t n)
311 return n & (OPDL_DISCLAIMS_PER_LCORE - 1);
314 /* Check if there are available slots in claim_manager */
315 static __rte_always_inline bool
316 claim_mgr_available(struct claim_manager *mgr)
318 return (mgr->mgr_head < (mgr->mgr_tail + OPDL_DISCLAIMS_PER_LCORE)) ?
322 /* Record a new claim. Only use after first checking an entry is available */
323 static __rte_always_inline void
324 claim_mgr_add(struct claim_manager *mgr, uint32_t tail, uint32_t head)
326 if ((mgr->mgr_head != mgr->mgr_tail) &&
327 (mgr->claims[claim_mgr_index(mgr->mgr_head - 1)].head ==
329 /* Combine with previous claim */
330 mgr->claims[claim_mgr_index(mgr->mgr_head - 1)].head = head;
332 mgr->claims[claim_mgr_index(mgr->mgr_head)].head = head;
333 mgr->claims[claim_mgr_index(mgr->mgr_head)].tail = tail;
337 mgr->num_claimed += (head - tail);
340 /* Read the oldest recorded claim */
341 static __rte_always_inline bool
342 claim_mgr_read(struct claim_manager *mgr, uint32_t *tail, uint32_t *head)
344 if (mgr->mgr_head == mgr->mgr_tail)
347 *head = mgr->claims[claim_mgr_index(mgr->mgr_tail)].head;
348 *tail = mgr->claims[claim_mgr_index(mgr->mgr_tail)].tail;
352 /* Remove the oldest recorded claim. Only use after first reading the entry */
353 static __rte_always_inline void
354 claim_mgr_remove(struct claim_manager *mgr)
356 mgr->num_claimed -= (mgr->claims[claim_mgr_index(mgr->mgr_tail)].head -
357 mgr->claims[claim_mgr_index(mgr->mgr_tail)].tail);
361 /* Update tail in the oldest claim. Only use after first reading the entry */
362 static __rte_always_inline void
363 claim_mgr_move_tail(struct claim_manager *mgr, uint32_t num_entries)
365 mgr->num_claimed -= num_entries;
366 mgr->claims[claim_mgr_index(mgr->mgr_tail)].tail += num_entries;
369 static __rte_always_inline void
370 opdl_stage_disclaim_multithread_n(struct opdl_stage *s,
371 uint32_t num_entries, bool block)
373 struct claim_manager *disclaims = &s->pending_disclaims[rte_lcore_id()];
377 while (num_entries) {
378 bool ret = claim_mgr_read(disclaims, &tail, &head);
381 break; /* nothing is claimed */
382 /* There should be no race condition here. If shared.tail
383 * matches, no other core can update it until this one does.
385 if (__atomic_load_n(&s->shared.tail, __ATOMIC_ACQUIRE) ==
387 if (num_entries >= (head - tail)) {
388 claim_mgr_remove(disclaims);
389 __atomic_store_n(&s->shared.tail, head,
391 num_entries -= (head - tail);
393 claim_mgr_move_tail(disclaims, num_entries);
394 __atomic_store_n(&s->shared.tail,
399 } else if (block == false)
400 break; /* blocked by other thread */
401 /* Keep going until num_entries are disclaimed. */
405 disclaims->num_to_disclaim = num_entries;
408 /* Move head atomically, returning number of entries available to process and
409 * the original value of head. For non-input stages, the claim is recorded
410 * so that the tail can be updated later by opdl_stage_disclaim().
412 static __rte_always_inline void
413 move_head_atomically(struct opdl_stage *s, uint32_t *num_entries,
414 uint32_t *old_head, bool block, bool claim_func)
416 uint32_t orig_num_entries = *num_entries;
418 struct claim_manager *disclaims = &s->pending_disclaims[rte_lcore_id()];
420 /* Attempt to disclaim any outstanding claims */
421 opdl_stage_disclaim_multithread_n(s, disclaims->num_to_disclaim,
424 *old_head = __atomic_load_n(&s->shared.head, __ATOMIC_ACQUIRE);
427 /* If called by opdl_ring_input(), claim does not need to be
428 * recorded, as there will be no disclaim.
431 /* Check that the claim can be recorded */
432 ret = claim_mgr_available(disclaims);
434 /* exit out if claim can't be recorded */
440 *num_entries = num_to_process(s, orig_num_entries, block);
441 if (*num_entries == 0)
444 success = __atomic_compare_exchange_n(&s->shared.head, old_head,
445 *old_head + *num_entries,
446 true, /* may fail spuriously */
447 __ATOMIC_RELEASE, /* memory order on success */
448 __ATOMIC_ACQUIRE); /* memory order on fail */
455 /* Store the claim record */
456 claim_mgr_add(disclaims, *old_head, *old_head + *num_entries);
459 /* Input function that supports multiple threads */
460 static __rte_always_inline uint32_t
461 opdl_ring_input_multithread(struct opdl_ring *t, const void *entries,
462 uint32_t num_entries, bool block)
464 struct opdl_stage *s = input_stage(t);
467 move_head_atomically(s, &num_entries, &old_head, block, false);
468 if (num_entries == 0)
471 copy_entries_in(t, old_head, entries, num_entries);
473 /* If another thread started inputting before this one, but hasn't
474 * finished, we need to wait for it to complete to update the tail.
476 rte_wait_until_equal_32(&s->shared.tail, old_head, __ATOMIC_ACQUIRE);
478 __atomic_store_n(&s->shared.tail, old_head + num_entries,
484 static __rte_always_inline uint32_t
485 opdl_first_entry_id(uint32_t start_seq, uint8_t nb_p_lcores,
488 return ((nb_p_lcores <= 1) ? 0 :
489 (nb_p_lcores - (start_seq % nb_p_lcores) + this_lcore) %
493 /* Claim slots to process, optimised for single-thread operation */
494 static __rte_always_inline uint32_t
495 opdl_stage_claim_singlethread(struct opdl_stage *s, void *entries,
496 uint32_t num_entries, uint32_t *seq, bool block, bool atomic)
498 uint32_t i = 0, j = 0, offset;
500 uint32_t flow_id = 0;
503 struct rte_event *ev;
505 struct opdl_ring *t = s->t;
506 uint8_t *entries_offset = (uint8_t *)entries;
510 offset = opdl_first_entry_id(s->seq, s->nb_instance,
513 num_entries = s->nb_instance * num_entries;
515 num_entries = num_to_process(s, num_entries, block);
517 for (; offset < num_entries; offset += s->nb_instance) {
518 get_slots = get_slot(t, s->head + offset);
519 memcpy(entries_offset, get_slots, t->slot_size);
520 entries_offset += t->slot_size;
524 num_entries = num_to_process(s, num_entries, block);
526 for (j = 0; j < num_entries; j++) {
527 ev = (struct rte_event *)get_slot(t, s->head+j);
529 event = __atomic_load_n(&(ev->event),
532 opa_id = OPDL_OPA_MASK & (event >> OPDL_OPA_OFFSET);
533 flow_id = OPDL_FLOWID_MASK & event;
535 if (opa_id >= s->queue_id)
538 if ((flow_id % s->nb_instance) == s->instance_id) {
539 memcpy(entries_offset, ev, t->slot_size);
540 entries_offset += t->slot_size;
545 s->shadow_head = s->head;
546 s->head += num_entries;
547 s->num_claimed = num_entries;
551 /* automatically disclaim entries if number of rte_events is zero */
552 if (unlikely(i == 0))
553 opdl_stage_disclaim(s, 0, false);
558 /* Thread-safe version of function to claim slots for processing */
559 static __rte_always_inline uint32_t
560 opdl_stage_claim_multithread(struct opdl_stage *s, void *entries,
561 uint32_t num_entries, uint32_t *seq, bool block)
564 struct opdl_ring *t = s->t;
565 uint32_t i = 0, offset;
566 uint8_t *entries_offset = (uint8_t *)entries;
569 PMD_DRV_LOG(ERR, "Invalid seq PTR");
572 offset = opdl_first_entry_id(*seq, s->nb_instance, s->instance_id);
573 num_entries = offset + (s->nb_instance * num_entries);
575 move_head_atomically(s, &num_entries, &old_head, block, true);
577 for (; offset < num_entries; offset += s->nb_instance) {
578 memcpy(entries_offset, get_slot(t, s->head + offset),
580 entries_offset += t->slot_size;
589 /* Claim and copy slot pointers, optimised for single-thread operation */
590 static __rte_always_inline uint32_t
591 opdl_stage_claim_copy_singlethread(struct opdl_stage *s, void *entries,
592 uint32_t num_entries, uint32_t *seq, bool block)
594 num_entries = num_to_process(s, num_entries, block);
595 if (num_entries == 0)
597 copy_entries_out(s->t, s->head, entries, num_entries);
600 s->head += num_entries;
604 /* Thread-safe version of function to claim and copy pointers to slots */
605 static __rte_always_inline uint32_t
606 opdl_stage_claim_copy_multithread(struct opdl_stage *s, void *entries,
607 uint32_t num_entries, uint32_t *seq, bool block)
611 move_head_atomically(s, &num_entries, &old_head, block, true);
612 if (num_entries == 0)
614 copy_entries_out(s->t, old_head, entries, num_entries);
620 static __rte_always_inline void
621 opdl_stage_disclaim_singlethread_n(struct opdl_stage *s,
622 uint32_t num_entries)
624 uint32_t old_tail = s->shared.tail;
626 if (unlikely(num_entries > (s->head - old_tail))) {
627 PMD_DRV_LOG(WARNING, "Attempt to disclaim (%u) more than claimed (%u)",
628 num_entries, s->head - old_tail);
629 num_entries = s->head - old_tail;
631 __atomic_store_n(&s->shared.tail, num_entries + old_tail,
636 opdl_ring_input(struct opdl_ring *t, const void *entries, uint32_t num_entries,
639 if (input_stage(t)->threadsafe == false)
640 return opdl_ring_input_singlethread(t, entries, num_entries,
643 return opdl_ring_input_multithread(t, entries, num_entries,
648 opdl_ring_copy_from_burst(struct opdl_ring *t, struct opdl_stage *s,
649 const void *entries, uint32_t num_entries, bool block)
651 uint32_t head = s->head;
653 num_entries = num_to_process(s, num_entries, block);
655 if (num_entries == 0)
658 copy_entries_in(t, head, entries, num_entries);
660 s->head += num_entries;
661 __atomic_store_n(&s->shared.tail, s->head, __ATOMIC_RELEASE);
668 opdl_ring_copy_to_burst(struct opdl_ring *t, struct opdl_stage *s,
669 void *entries, uint32_t num_entries, bool block)
671 uint32_t head = s->head;
673 num_entries = num_to_process(s, num_entries, block);
674 if (num_entries == 0)
677 copy_entries_out(t, head, entries, num_entries);
679 s->head += num_entries;
680 __atomic_store_n(&s->shared.tail, s->head, __ATOMIC_RELEASE);
686 opdl_stage_find_num_available(struct opdl_stage *s, uint32_t num_entries)
688 /* return (num_to_process(s, num_entries, false)); */
690 if (available(s) >= num_entries)
693 update_available_seq(s);
695 uint32_t avail = available(s);
701 return (avail <= num_entries) ? avail : num_entries;
705 opdl_stage_claim(struct opdl_stage *s, void *entries,
706 uint32_t num_entries, uint32_t *seq, bool block, bool atomic)
708 if (s->threadsafe == false)
709 return opdl_stage_claim_singlethread(s, entries, num_entries,
712 return opdl_stage_claim_multithread(s, entries, num_entries,
717 opdl_stage_claim_copy(struct opdl_stage *s, void *entries,
718 uint32_t num_entries, uint32_t *seq, bool block)
720 if (s->threadsafe == false)
721 return opdl_stage_claim_copy_singlethread(s, entries,
722 num_entries, seq, block);
724 return opdl_stage_claim_copy_multithread(s, entries,
725 num_entries, seq, block);
729 opdl_stage_disclaim_n(struct opdl_stage *s, uint32_t num_entries,
733 if (s->threadsafe == false) {
734 opdl_stage_disclaim_singlethread_n(s, s->num_claimed);
736 struct claim_manager *disclaims =
737 &s->pending_disclaims[rte_lcore_id()];
739 if (unlikely(num_entries > s->num_slots)) {
740 PMD_DRV_LOG(WARNING, "Attempt to disclaim (%u) more than claimed (%u)",
741 num_entries, disclaims->num_claimed);
742 num_entries = disclaims->num_claimed;
745 num_entries = RTE_MIN(num_entries + disclaims->num_to_disclaim,
746 disclaims->num_claimed);
747 opdl_stage_disclaim_multithread_n(s, num_entries, block);
752 opdl_stage_disclaim(struct opdl_stage *s, uint32_t num_entries, bool block)
754 if (num_entries != s->num_event) {
758 if (s->threadsafe == false) {
759 __atomic_store_n(&s->shared.tail, s->head, __ATOMIC_RELEASE);
760 s->seq += s->num_claimed;
761 s->shadow_head = s->head;
764 struct claim_manager *disclaims =
765 &s->pending_disclaims[rte_lcore_id()];
766 opdl_stage_disclaim_multithread_n(s, disclaims->num_claimed,
773 opdl_ring_available(struct opdl_ring *t)
775 return opdl_stage_available(&t->stages[0]);
779 opdl_stage_available(struct opdl_stage *s)
781 update_available_seq(s);
786 opdl_ring_flush(struct opdl_ring *t)
788 struct opdl_stage *s = input_stage(t);
790 wait_for_available(s, s->num_slots);
793 /******************** Non performance sensitive functions ********************/
795 /* Initial setup of a new stage's context */
797 init_stage(struct opdl_ring *t, struct opdl_stage *s, bool threadsafe,
800 uint32_t available = (is_input) ? t->num_slots : 0;
803 s->num_slots = t->num_slots;
804 s->index = t->num_stages;
805 s->threadsafe = threadsafe;
808 /* Alloc memory for deps */
809 s->dep_tracking = rte_zmalloc_socket(LIB_NAME,
810 t->max_num_stages * sizeof(enum dep_type),
812 if (s->dep_tracking == NULL)
815 s->deps = rte_zmalloc_socket(LIB_NAME,
816 t->max_num_stages * sizeof(struct shared_state *),
818 if (s->deps == NULL) {
819 rte_free(s->dep_tracking);
823 s->dep_tracking[s->index] = DEP_SELF;
825 if (threadsafe == true)
826 s->shared.available_seq = available;
828 s->available_seq = available;
833 /* Add direct or indirect dependencies between stages */
835 add_dep(struct opdl_stage *dependent, const struct opdl_stage *dependency,
838 struct opdl_ring *t = dependent->t;
841 /* Add new direct dependency */
842 if ((type == DEP_DIRECT) &&
843 (dependent->dep_tracking[dependency->index] ==
845 PMD_DRV_LOG(DEBUG, "%s:%u direct dependency on %u",
846 t->name, dependent->index, dependency->index);
847 dependent->dep_tracking[dependency->index] = DEP_DIRECT;
850 /* Add new indirect dependency or change direct to indirect */
851 if ((type == DEP_INDIRECT) &&
852 ((dependent->dep_tracking[dependency->index] ==
854 (dependent->dep_tracking[dependency->index] ==
856 PMD_DRV_LOG(DEBUG, "%s:%u indirect dependency on %u",
857 t->name, dependent->index, dependency->index);
858 dependent->dep_tracking[dependency->index] = DEP_INDIRECT;
861 /* Shouldn't happen... */
862 if ((dependent->dep_tracking[dependency->index] == DEP_SELF) &&
863 (dependent != input_stage(t))) {
864 PMD_DRV_LOG(ERR, "Loop in dependency graph %s:%u",
865 t->name, dependent->index);
869 /* Keep going to dependencies of the dependency, until input stage */
870 if (dependency != input_stage(t))
871 for (i = 0; i < dependency->num_deps; i++) {
872 int ret = add_dep(dependent, dependency->deps[i]->stage,
879 /* Make list of sequence numbers for direct dependencies only */
880 if (type == DEP_DIRECT)
881 for (i = 0, dependent->num_deps = 0; i < t->num_stages; i++)
882 if (dependent->dep_tracking[i] == DEP_DIRECT) {
883 if ((i == 0) && (dependent->num_deps > 1))
884 rte_panic("%s:%u depends on > input",
887 dependent->deps[dependent->num_deps++] =
888 &t->stages[i].shared;
895 opdl_ring_create(const char *name, uint32_t num_slots, uint32_t slot_size,
896 uint32_t max_num_stages, int socket)
899 char mz_name[RTE_MEMZONE_NAMESIZE];
901 struct opdl_stage *st = NULL;
902 const struct rte_memzone *mz = NULL;
903 size_t alloc_size = RTE_CACHE_LINE_ROUNDUP(sizeof(*t) +
904 (num_slots * slot_size));
906 /* Compile time checking */
907 RTE_BUILD_BUG_ON((sizeof(struct shared_state) & RTE_CACHE_LINE_MASK) !=
909 RTE_BUILD_BUG_ON((offsetof(struct opdl_stage, shared) &
910 RTE_CACHE_LINE_MASK) != 0);
911 RTE_BUILD_BUG_ON((offsetof(struct opdl_ring, slots) &
912 RTE_CACHE_LINE_MASK) != 0);
913 RTE_BUILD_BUG_ON(!rte_is_power_of_2(OPDL_DISCLAIMS_PER_LCORE));
915 /* Parameter checking */
917 PMD_DRV_LOG(ERR, "name param is NULL");
920 if (!rte_is_power_of_2(num_slots)) {
921 PMD_DRV_LOG(ERR, "num_slots (%u) for %s is not power of 2",
926 /* Alloc memory for stages */
927 st = rte_zmalloc_socket(LIB_NAME,
928 max_num_stages * sizeof(struct opdl_stage),
929 RTE_CACHE_LINE_SIZE, socket);
933 snprintf(mz_name, sizeof(mz_name), "%s%s", LIB_NAME, name);
935 /* Alloc memory for memzone */
936 mz = rte_memzone_reserve(mz_name, alloc_size, socket, mz_flags);
942 /* Initialise opdl_ring queue */
943 memset(t, 0, sizeof(*t));
944 strlcpy(t->name, name, sizeof(t->name));
946 t->num_slots = num_slots;
947 t->mask = num_slots - 1;
948 t->slot_size = slot_size;
949 t->max_num_stages = max_num_stages;
952 PMD_DRV_LOG(DEBUG, "Created %s at %p (num_slots=%u,socket=%i,slot_size=%u)",
953 t->name, t, num_slots, socket, slot_size);
958 PMD_DRV_LOG(ERR, "Cannot reserve memory");
960 rte_memzone_free(mz);
966 opdl_ring_get_slot(const struct opdl_ring *t, uint32_t index)
968 return get_slot(t, index);
972 opdl_ring_cas_slot(struct opdl_stage *s, const struct rte_event *ev,
973 uint32_t index, bool atomic)
975 uint32_t i = 0, offset;
976 struct opdl_ring *t = s->t;
977 struct rte_event *ev_orig = NULL;
978 bool ev_updated = false;
979 uint64_t ev_temp = 0;
980 uint64_t ev_update = 0;
983 uint32_t flow_id = 0;
986 if (index > s->num_event) {
987 PMD_DRV_LOG(ERR, "index is overflow");
991 ev_temp = ev->event & OPDL_EVENT_MASK;
994 offset = opdl_first_entry_id(s->seq, s->nb_instance,
996 offset += index*s->nb_instance;
997 ev_orig = get_slot(t, s->shadow_head+offset);
998 if ((ev_orig->event&OPDL_EVENT_MASK) != ev_temp) {
999 ev_orig->event = ev->event;
1002 if (ev_orig->u64 != ev->u64) {
1003 ev_orig->u64 = ev->u64;
1008 for (i = s->pos; i < s->num_claimed; i++) {
1009 ev_orig = (struct rte_event *)
1010 get_slot(t, s->shadow_head+i);
1012 event = __atomic_load_n(&(ev_orig->event),
1015 opa_id = OPDL_OPA_MASK & (event >> OPDL_OPA_OFFSET);
1016 flow_id = OPDL_FLOWID_MASK & event;
1018 if (opa_id >= s->queue_id)
1021 if ((flow_id % s->nb_instance) == s->instance_id) {
1022 ev_update = s->queue_id;
1023 ev_update = (ev_update << OPDL_OPA_OFFSET)
1028 if ((event & OPDL_EVENT_MASK) !=
1030 __atomic_store_n(&(ev_orig->event),
1035 if (ev_orig->u64 != ev->u64) {
1036 ev_orig->u64 = ev->u64;
1050 opdl_ring_get_socket(const struct opdl_ring *t)
1056 opdl_ring_get_num_slots(const struct opdl_ring *t)
1058 return t->num_slots;
1062 opdl_ring_get_name(const struct opdl_ring *t)
1067 /* Check dependency list is valid for a given opdl_ring */
1069 check_deps(struct opdl_ring *t, struct opdl_stage *deps[],
1074 for (i = 0; i < num_deps; ++i) {
1076 PMD_DRV_LOG(ERR, "deps[%u] is NULL", i);
1079 if (t != deps[i]->t) {
1080 PMD_DRV_LOG(ERR, "deps[%u] is in opdl_ring %s, not %s",
1081 i, deps[i]->t->name, t->name);
1090 opdl_stage_add(struct opdl_ring *t, bool threadsafe, bool is_input)
1092 struct opdl_stage *s;
1094 /* Parameter checking */
1096 PMD_DRV_LOG(ERR, "opdl_ring is NULL");
1099 if (t->num_stages == t->max_num_stages) {
1100 PMD_DRV_LOG(ERR, "%s has max number of stages (%u)",
1101 t->name, t->max_num_stages);
1105 s = &t->stages[t->num_stages];
1107 if (((uintptr_t)&s->shared & RTE_CACHE_LINE_MASK) != 0)
1108 PMD_DRV_LOG(WARNING, "Tail seq num (%p) of %s stage not cache aligned",
1109 &s->shared, t->name);
1111 if (init_stage(t, s, threadsafe, is_input) < 0) {
1112 PMD_DRV_LOG(ERR, "Cannot reserve memory");
1121 opdl_stage_deps_add(struct opdl_ring *t, struct opdl_stage *s,
1122 uint32_t nb_instance, uint32_t instance_id,
1123 struct opdl_stage *deps[],
1129 if ((num_deps > 0) && (!deps)) {
1130 PMD_DRV_LOG(ERR, "%s stage has NULL dependencies", t->name);
1133 ret = check_deps(t, deps, num_deps);
1137 for (i = 0; i < num_deps; i++) {
1138 ret = add_dep(s, deps[i], DEP_DIRECT);
1143 s->nb_instance = nb_instance;
1144 s->instance_id = instance_id;
1150 opdl_ring_get_input_stage(const struct opdl_ring *t)
1152 return input_stage(t);
1156 opdl_stage_set_deps(struct opdl_stage *s, struct opdl_stage *deps[],
1162 if ((num_deps == 0) || (!deps)) {
1163 PMD_DRV_LOG(ERR, "cannot set NULL dependencies");
1167 ret = check_deps(s->t, deps, num_deps);
1172 for (i = 0; i < num_deps; i++)
1173 s->deps[i] = &deps[i]->shared;
1174 s->num_deps = num_deps;
1180 opdl_stage_get_opdl_ring(const struct opdl_stage *s)
1186 opdl_stage_set_queue_id(struct opdl_stage *s,
1189 s->queue_id = queue_id;
1193 opdl_ring_dump(const struct opdl_ring *t, FILE *f)
1198 fprintf(f, "NULL OPDL!\n");
1201 fprintf(f, "OPDL \"%s\": num_slots=%u; mask=%#x; slot_size=%u; num_stages=%u; socket=%i\n",
1202 t->name, t->num_slots, t->mask, t->slot_size,
1203 t->num_stages, t->socket);
1204 for (i = 0; i < t->num_stages; i++) {
1206 const struct opdl_stage *s = &t->stages[i];
1208 fprintf(f, " %s[%u]: threadsafe=%s; head=%u; available_seq=%u; tail=%u; deps=%u",
1209 t->name, i, (s->threadsafe) ? "true" : "false",
1210 (s->threadsafe) ? s->shared.head : s->head,
1211 (s->threadsafe) ? s->shared.available_seq :
1213 s->shared.tail, (s->num_deps > 0) ?
1214 s->deps[0]->stage->index : 0);
1215 for (j = 1; j < s->num_deps; j++)
1216 fprintf(f, ",%u", s->deps[j]->stage->index);
1223 opdl_ring_free(struct opdl_ring *t)
1226 const struct rte_memzone *mz;
1227 char mz_name[RTE_MEMZONE_NAMESIZE];
1230 PMD_DRV_LOG(DEBUG, "Freeing NULL OPDL Ring!");
1234 PMD_DRV_LOG(DEBUG, "Freeing %s opdl_ring at %p", t->name, t);
1236 for (i = 0; i < t->num_stages; ++i) {
1237 rte_free(t->stages[i].deps);
1238 rte_free(t->stages[i].dep_tracking);
1241 rte_free(t->stages);
1243 snprintf(mz_name, sizeof(mz_name), "%s%s", LIB_NAME, t->name);
1244 mz = rte_memzone_lookup(mz_name);
1245 if (rte_memzone_free(mz) != 0)
1246 PMD_DRV_LOG(ERR, "Cannot free memzone for %s", t->name);
1249 /* search a opdl_ring from its name */
1251 opdl_ring_lookup(const char *name)
1253 const struct rte_memzone *mz;
1254 char mz_name[RTE_MEMZONE_NAMESIZE];
1256 snprintf(mz_name, sizeof(mz_name), "%s%s", LIB_NAME, name);
1258 mz = rte_memzone_lookup(mz_name);
1266 opdl_ring_set_stage_threadsafe(struct opdl_stage *s, bool threadsafe)
1268 s->threadsafe = threadsafe;