bus/pci: reference driver structure before mapping
[dpdk.git] / drivers / event / opdl / opdl_ring.c
1 /* SPDX-License-Identifier: BSD-3-Clause
2  * Copyright(c) 2017 Intel Corporation
3  */
4
5 #include <stdbool.h>
6 #include <stddef.h>
7 #include <stdint.h>
8 #include <stdio.h>
9
10 #include <rte_branch_prediction.h>
11 #include <rte_debug.h>
12 #include <rte_lcore.h>
13 #include <rte_log.h>
14 #include <rte_malloc.h>
15 #include <rte_memcpy.h>
16 #include <rte_memory.h>
17 #include <rte_memzone.h>
18 #include <rte_eal_memconfig.h>
19
20 #include "opdl_ring.h"
21 #include "opdl_log.h"
22
23 #define LIB_NAME "opdl_ring"
24
25 #define OPDL_NAME_SIZE 64
26
27
28 #define OPDL_EVENT_MASK  (0x00000000000FFFFFULL)
29 #define OPDL_FLOWID_MASK (0xFFFFF)
30 #define OPDL_OPA_MASK    (0xFF)
31 #define OPDL_OPA_OFFSET  (0x38)
32
33 int opdl_logtype_driver;
34
35 /* Types of dependency between stages */
36 enum dep_type {
37         DEP_NONE = 0,  /* no dependency */
38         DEP_DIRECT,  /* stage has direct dependency */
39         DEP_INDIRECT,  /* in-direct dependency through other stage(s) */
40         DEP_SELF,  /* stage dependency on itself, used to detect loops */
41 };
42
43 /* Shared section of stage state.
44  * Care is needed when accessing and the layout is important, especially to
45  * limit the adjacent cache-line HW prefetcher from impacting performance.
46  */
47 struct shared_state {
48         /* Last known minimum sequence number of dependencies, used for multi
49          * thread operation
50          */
51         uint32_t available_seq;
52         char _pad1[RTE_CACHE_LINE_SIZE * 3];
53         uint32_t head;  /* Head sequence number (for multi thread operation) */
54         char _pad2[RTE_CACHE_LINE_SIZE * 3];
55         struct opdl_stage *stage;  /* back pointer */
56         uint32_t tail;  /* Tail sequence number */
57         char _pad3[RTE_CACHE_LINE_SIZE * 2];
58 } __rte_cache_aligned;
59
60 /* A structure to keep track of "unfinished" claims. This is only used for
61  * stages that are threadsafe. Each lcore accesses its own instance of this
62  * structure to record the entries it has claimed. This allows one lcore to make
63  * multiple claims without being blocked by another. When disclaiming it moves
64  * forward the shared tail when the shared tail matches the tail value recorded
65  * here.
66  */
67 struct claim_manager {
68         uint32_t num_to_disclaim;
69         uint32_t num_claimed;
70         uint32_t mgr_head;
71         uint32_t mgr_tail;
72         struct {
73                 uint32_t head;
74                 uint32_t tail;
75         } claims[OPDL_DISCLAIMS_PER_LCORE];
76 } __rte_cache_aligned;
77
78 /* Context for each stage of opdl_ring.
79  * Calculations on sequence numbers need to be done with other uint32_t values
80  * so that results are modulus 2^32, and not undefined.
81  */
82 struct opdl_stage {
83         struct opdl_ring *t;  /* back pointer, set at init */
84         uint32_t num_slots;  /* Number of slots for entries, set at init */
85         uint32_t index;  /* ID for this stage, set at init */
86         bool threadsafe;  /* Set to 1 if this stage supports threadsafe use */
87         /* Last known min seq number of dependencies for used for single thread
88          * operation
89          */
90         uint32_t available_seq;
91         uint32_t head;  /* Current head for single-thread operation */
92         uint32_t nb_instance;  /* Number of instances */
93         uint32_t instance_id;  /* ID of this stage instance */
94         uint16_t num_claimed;  /* Number of slots claimed */
95         uint16_t num_event;             /* Number of events */
96         uint32_t seq;                   /* sequence number  */
97         uint32_t num_deps;  /* Number of direct dependencies */
98         /* Keep track of all dependencies, used during init only */
99         enum dep_type *dep_tracking;
100         /* Direct dependencies of this stage */
101         struct shared_state **deps;
102         /* Other stages read this! */
103         struct shared_state shared __rte_cache_aligned;
104         /* For managing disclaims in multi-threaded processing stages */
105         struct claim_manager pending_disclaims[RTE_MAX_LCORE]
106                                                __rte_cache_aligned;
107         uint32_t shadow_head;  /* Shadow head for single-thread operation */
108         uint32_t queue_id;     /* ID of Queue which is assigned to this stage */
109         uint32_t pos;           /* Atomic scan position */
110 } __rte_cache_aligned;
111
112 /* Context for opdl_ring */
113 struct opdl_ring {
114         char name[OPDL_NAME_SIZE];  /* OPDL queue instance name */
115         int socket;  /* NUMA socket that memory is allocated on */
116         uint32_t num_slots;  /* Number of slots for entries */
117         uint32_t mask;  /* Mask for sequence numbers (num_slots - 1) */
118         uint32_t slot_size;  /* Size of each slot in bytes */
119         uint32_t num_stages;  /* Number of stages that have been added */
120         uint32_t max_num_stages;  /* Max number of stages */
121         /* Stages indexed by ID */
122         struct opdl_stage *stages;
123         /* Memory for storing slot data */
124         uint8_t slots[0] __rte_cache_aligned;
125 };
126
127
128 /* Return input stage of a opdl_ring */
129 static __rte_always_inline struct opdl_stage *
130 input_stage(const struct opdl_ring *t)
131 {
132         return &t->stages[0];
133 }
134
135 /* Check if a stage is the input stage */
136 static __rte_always_inline bool
137 is_input_stage(const struct opdl_stage *s)
138 {
139         return s->index == 0;
140 }
141
142 /* Get slot pointer from sequence number */
143 static __rte_always_inline void *
144 get_slot(const struct opdl_ring *t, uint32_t n)
145 {
146         return (void *)(uintptr_t)&t->slots[(n & t->mask) * t->slot_size];
147 }
148
149 /* Find how many entries are available for processing */
150 static __rte_always_inline uint32_t
151 available(const struct opdl_stage *s)
152 {
153         if (s->threadsafe == true) {
154                 uint32_t n = __atomic_load_n(&s->shared.available_seq,
155                                 __ATOMIC_ACQUIRE) -
156                                 __atomic_load_n(&s->shared.head,
157                                 __ATOMIC_ACQUIRE);
158
159                 /* Return 0 if available_seq needs to be updated */
160                 return (n <= s->num_slots) ? n : 0;
161         }
162
163         /* Single threaded */
164         return s->available_seq - s->head;
165 }
166
167 /* Read sequence number of dependencies and find minimum */
168 static __rte_always_inline void
169 update_available_seq(struct opdl_stage *s)
170 {
171         uint32_t i;
172         uint32_t this_tail = s->shared.tail;
173         uint32_t min_seq = __atomic_load_n(&s->deps[0]->tail, __ATOMIC_ACQUIRE);
174         /* Input stage sequence numbers are greater than the sequence numbers of
175          * its dependencies so an offset of t->num_slots is needed when
176          * calculating available slots and also the condition which is used to
177          * determine the dependencies minimum sequence number must be reverted.
178          */
179         uint32_t wrap;
180
181         if (is_input_stage(s)) {
182                 wrap = s->num_slots;
183                 for (i = 1; i < s->num_deps; i++) {
184                         uint32_t seq = __atomic_load_n(&s->deps[i]->tail,
185                                         __ATOMIC_ACQUIRE);
186                         if ((this_tail - seq) > (this_tail - min_seq))
187                                 min_seq = seq;
188                 }
189         } else {
190                 wrap = 0;
191                 for (i = 1; i < s->num_deps; i++) {
192                         uint32_t seq = __atomic_load_n(&s->deps[i]->tail,
193                                         __ATOMIC_ACQUIRE);
194                         if ((seq - this_tail) < (min_seq - this_tail))
195                                 min_seq = seq;
196                 }
197         }
198
199         if (s->threadsafe == false)
200                 s->available_seq = min_seq + wrap;
201         else
202                 __atomic_store_n(&s->shared.available_seq, min_seq + wrap,
203                                 __ATOMIC_RELEASE);
204 }
205
206 /* Wait until the number of available slots reaches number requested */
207 static __rte_always_inline void
208 wait_for_available(struct opdl_stage *s, uint32_t n)
209 {
210         while (available(s) < n) {
211                 rte_pause();
212                 update_available_seq(s);
213         }
214 }
215
216 /* Return number of slots to process based on number requested and mode */
217 static __rte_always_inline uint32_t
218 num_to_process(struct opdl_stage *s, uint32_t n, bool block)
219 {
220         /* Don't read tail sequences of dependencies if not needed */
221         if (available(s) >= n)
222                 return n;
223
224         update_available_seq(s);
225
226         if (block == false) {
227                 uint32_t avail = available(s);
228
229                 if (avail == 0) {
230                         rte_pause();
231                         return 0;
232                 }
233                 return (avail <= n) ? avail : n;
234         }
235
236         if (unlikely(n > s->num_slots)) {
237                 PMD_DRV_LOG(ERR, "%u entries is more than max (%u)",
238                                 n, s->num_slots);
239                 return 0;  /* Avoid infinite loop */
240         }
241         /* blocking */
242         wait_for_available(s, n);
243         return n;
244 }
245
246 /* Copy entries in to slots with wrap-around */
247 static __rte_always_inline void
248 copy_entries_in(struct opdl_ring *t, uint32_t start, const void *entries,
249                 uint32_t num_entries)
250 {
251         uint32_t slot_size = t->slot_size;
252         uint32_t slot_index = start & t->mask;
253
254         if (slot_index + num_entries <= t->num_slots) {
255                 rte_memcpy(get_slot(t, start), entries,
256                                 num_entries * slot_size);
257         } else {
258                 uint32_t split = t->num_slots - slot_index;
259
260                 rte_memcpy(get_slot(t, start), entries, split * slot_size);
261                 rte_memcpy(get_slot(t, 0),
262                                 RTE_PTR_ADD(entries, split * slot_size),
263                                 (num_entries - split) * slot_size);
264         }
265 }
266
267 /* Copy entries out from slots with wrap-around */
268 static __rte_always_inline void
269 copy_entries_out(struct opdl_ring *t, uint32_t start, void *entries,
270                 uint32_t num_entries)
271 {
272         uint32_t slot_size = t->slot_size;
273         uint32_t slot_index = start & t->mask;
274
275         if (slot_index + num_entries <= t->num_slots) {
276                 rte_memcpy(entries, get_slot(t, start),
277                                 num_entries * slot_size);
278         } else {
279                 uint32_t split = t->num_slots - slot_index;
280
281                 rte_memcpy(entries, get_slot(t, start), split * slot_size);
282                 rte_memcpy(RTE_PTR_ADD(entries, split * slot_size),
283                                 get_slot(t, 0),
284                                 (num_entries - split) * slot_size);
285         }
286 }
287
288 /* Input function optimised for single thread */
289 static __rte_always_inline uint32_t
290 opdl_ring_input_singlethread(struct opdl_ring *t, const void *entries,
291                 uint32_t num_entries, bool block)
292 {
293         struct opdl_stage *s = input_stage(t);
294         uint32_t head = s->head;
295
296         num_entries = num_to_process(s, num_entries, block);
297         if (num_entries == 0)
298                 return 0;
299
300         copy_entries_in(t, head, entries, num_entries);
301
302         s->head += num_entries;
303         __atomic_store_n(&s->shared.tail, s->head, __ATOMIC_RELEASE);
304
305         return num_entries;
306 }
307
308 /* Convert head and tail of claim_manager into valid index */
309 static __rte_always_inline uint32_t
310 claim_mgr_index(uint32_t n)
311 {
312         return n & (OPDL_DISCLAIMS_PER_LCORE - 1);
313 }
314
315 /* Check if there are available slots in claim_manager */
316 static __rte_always_inline bool
317 claim_mgr_available(struct claim_manager *mgr)
318 {
319         return (mgr->mgr_head < (mgr->mgr_tail + OPDL_DISCLAIMS_PER_LCORE)) ?
320                         true : false;
321 }
322
323 /* Record a new claim. Only use after first checking an entry is available */
324 static __rte_always_inline void
325 claim_mgr_add(struct claim_manager *mgr, uint32_t tail, uint32_t head)
326 {
327         if ((mgr->mgr_head != mgr->mgr_tail) &&
328                         (mgr->claims[claim_mgr_index(mgr->mgr_head - 1)].head ==
329                         tail)) {
330                 /* Combine with previous claim */
331                 mgr->claims[claim_mgr_index(mgr->mgr_head - 1)].head = head;
332         } else {
333                 mgr->claims[claim_mgr_index(mgr->mgr_head)].head = head;
334                 mgr->claims[claim_mgr_index(mgr->mgr_head)].tail = tail;
335                 mgr->mgr_head++;
336         }
337
338         mgr->num_claimed += (head - tail);
339 }
340
341 /* Read the oldest recorded claim */
342 static __rte_always_inline bool
343 claim_mgr_read(struct claim_manager *mgr, uint32_t *tail, uint32_t *head)
344 {
345         if (mgr->mgr_head == mgr->mgr_tail)
346                 return false;
347
348         *head = mgr->claims[claim_mgr_index(mgr->mgr_tail)].head;
349         *tail = mgr->claims[claim_mgr_index(mgr->mgr_tail)].tail;
350         return true;
351 }
352
353 /* Remove the oldest recorded claim. Only use after first reading the entry */
354 static __rte_always_inline void
355 claim_mgr_remove(struct claim_manager *mgr)
356 {
357         mgr->num_claimed -= (mgr->claims[claim_mgr_index(mgr->mgr_tail)].head -
358                         mgr->claims[claim_mgr_index(mgr->mgr_tail)].tail);
359         mgr->mgr_tail++;
360 }
361
362 /* Update tail in the oldest claim. Only use after first reading the entry */
363 static __rte_always_inline void
364 claim_mgr_move_tail(struct claim_manager *mgr, uint32_t num_entries)
365 {
366         mgr->num_claimed -= num_entries;
367         mgr->claims[claim_mgr_index(mgr->mgr_tail)].tail += num_entries;
368 }
369
370 static __rte_always_inline void
371 opdl_stage_disclaim_multithread_n(struct opdl_stage *s,
372                 uint32_t num_entries, bool block)
373 {
374         struct claim_manager *disclaims = &s->pending_disclaims[rte_lcore_id()];
375         uint32_t head;
376         uint32_t tail;
377
378         while (num_entries) {
379                 bool ret = claim_mgr_read(disclaims, &tail, &head);
380
381                 if (ret == false)
382                         break;  /* nothing is claimed */
383                 /* There should be no race condition here. If shared.tail
384                  * matches, no other core can update it until this one does.
385                  */
386                 if (__atomic_load_n(&s->shared.tail, __ATOMIC_ACQUIRE) ==
387                                 tail) {
388                         if (num_entries >= (head - tail)) {
389                                 claim_mgr_remove(disclaims);
390                                 __atomic_store_n(&s->shared.tail, head,
391                                                 __ATOMIC_RELEASE);
392                                 num_entries -= (head - tail);
393                         } else {
394                                 claim_mgr_move_tail(disclaims, num_entries);
395                                 __atomic_store_n(&s->shared.tail,
396                                                 num_entries + tail,
397                                                 __ATOMIC_RELEASE);
398                                 num_entries = 0;
399                         }
400                 } else if (block == false)
401                         break;  /* blocked by other thread */
402                 /* Keep going until num_entries are disclaimed. */
403                 rte_pause();
404         }
405
406         disclaims->num_to_disclaim = num_entries;
407 }
408
409 /* Move head atomically, returning number of entries available to process and
410  * the original value of head. For non-input stages, the claim is recorded
411  * so that the tail can be updated later by opdl_stage_disclaim().
412  */
413 static __rte_always_inline void
414 move_head_atomically(struct opdl_stage *s, uint32_t *num_entries,
415                 uint32_t *old_head, bool block, bool claim_func)
416 {
417         uint32_t orig_num_entries = *num_entries;
418         uint32_t ret;
419         struct claim_manager *disclaims = &s->pending_disclaims[rte_lcore_id()];
420
421         /* Attempt to disclaim any outstanding claims */
422         opdl_stage_disclaim_multithread_n(s, disclaims->num_to_disclaim,
423                         false);
424
425         *old_head = __atomic_load_n(&s->shared.head, __ATOMIC_ACQUIRE);
426         while (true) {
427                 bool success;
428                 /* If called by opdl_ring_input(), claim does not need to be
429                  * recorded, as there will be no disclaim.
430                  */
431                 if (claim_func) {
432                         /* Check that the claim can be recorded */
433                         ret = claim_mgr_available(disclaims);
434                         if (ret == false) {
435                                 /* exit out if claim can't be recorded */
436                                 *num_entries = 0;
437                                 return;
438                         }
439                 }
440
441                 *num_entries = num_to_process(s, orig_num_entries, block);
442                 if (*num_entries == 0)
443                         return;
444
445                 success = __atomic_compare_exchange_n(&s->shared.head, old_head,
446                                 *old_head + *num_entries,
447                                 true,  /* may fail spuriously */
448                                 __ATOMIC_RELEASE,  /* memory order on success */
449                                 __ATOMIC_ACQUIRE);  /* memory order on fail */
450                 if (likely(success))
451                         break;
452                 rte_pause();
453         }
454
455         if (claim_func)
456                 /* Store the claim record */
457                 claim_mgr_add(disclaims, *old_head, *old_head + *num_entries);
458 }
459
460 /* Input function that supports multiple threads */
461 static __rte_always_inline uint32_t
462 opdl_ring_input_multithread(struct opdl_ring *t, const void *entries,
463                 uint32_t num_entries, bool block)
464 {
465         struct opdl_stage *s = input_stage(t);
466         uint32_t old_head;
467
468         move_head_atomically(s, &num_entries, &old_head, block, false);
469         if (num_entries == 0)
470                 return 0;
471
472         copy_entries_in(t, old_head, entries, num_entries);
473
474         /* If another thread started inputting before this one, but hasn't
475          * finished, we need to wait for it to complete to update the tail.
476          */
477         while (unlikely(__atomic_load_n(&s->shared.tail, __ATOMIC_ACQUIRE) !=
478                         old_head))
479                 rte_pause();
480
481         __atomic_store_n(&s->shared.tail, old_head + num_entries,
482                         __ATOMIC_RELEASE);
483
484         return num_entries;
485 }
486
487 static __rte_always_inline uint32_t
488 opdl_first_entry_id(uint32_t start_seq, uint8_t nb_p_lcores,
489                 uint8_t this_lcore)
490 {
491         return ((nb_p_lcores <= 1) ? 0 :
492                         (nb_p_lcores - (start_seq % nb_p_lcores) + this_lcore) %
493                         nb_p_lcores);
494 }
495
496 /* Claim slots to process, optimised for single-thread operation */
497 static __rte_always_inline uint32_t
498 opdl_stage_claim_singlethread(struct opdl_stage *s, void *entries,
499                 uint32_t num_entries, uint32_t *seq, bool block, bool atomic)
500 {
501         uint32_t i = 0, j = 0,  offset;
502         uint32_t opa_id   = 0;
503         uint32_t flow_id  = 0;
504         uint64_t event    = 0;
505         void *get_slots;
506         struct rte_event *ev;
507         RTE_SET_USED(seq);
508         struct opdl_ring *t = s->t;
509         uint8_t *entries_offset = (uint8_t *)entries;
510
511         if (!atomic) {
512
513                 offset = opdl_first_entry_id(s->seq, s->nb_instance,
514                                 s->instance_id);
515
516                 num_entries = s->nb_instance * num_entries;
517
518                 num_entries = num_to_process(s, num_entries, block);
519
520                 for (; offset < num_entries; offset += s->nb_instance) {
521                         get_slots = get_slot(t, s->head + offset);
522                         memcpy(entries_offset, get_slots, t->slot_size);
523                         entries_offset += t->slot_size;
524                         i++;
525                 }
526         } else {
527                 num_entries = num_to_process(s, num_entries, block);
528
529                 for (j = 0; j < num_entries; j++) {
530                         ev = (struct rte_event *)get_slot(t, s->head+j);
531
532                         event  = __atomic_load_n(&(ev->event),
533                                         __ATOMIC_ACQUIRE);
534
535                         opa_id = OPDL_OPA_MASK & (event >> OPDL_OPA_OFFSET);
536                         flow_id  = OPDL_FLOWID_MASK & event;
537
538                         if (opa_id >= s->queue_id)
539                                 continue;
540
541                         if ((flow_id % s->nb_instance) == s->instance_id) {
542                                 memcpy(entries_offset, ev, t->slot_size);
543                                 entries_offset += t->slot_size;
544                                 i++;
545                         }
546                 }
547         }
548         s->shadow_head = s->head;
549         s->head += num_entries;
550         s->num_claimed = num_entries;
551         s->num_event = i;
552         s->pos = 0;
553
554         /* automatically disclaim entries if number of rte_events is zero */
555         if (unlikely(i == 0))
556                 opdl_stage_disclaim(s, 0, false);
557
558         return i;
559 }
560
561 /* Thread-safe version of function to claim slots for processing */
562 static __rte_always_inline uint32_t
563 opdl_stage_claim_multithread(struct opdl_stage *s, void *entries,
564                 uint32_t num_entries, uint32_t *seq, bool block)
565 {
566         uint32_t old_head;
567         struct opdl_ring *t = s->t;
568         uint32_t i = 0, offset;
569         uint8_t *entries_offset = (uint8_t *)entries;
570
571         if (seq == NULL) {
572                 PMD_DRV_LOG(ERR, "Invalid seq PTR");
573                 return 0;
574         }
575         offset = opdl_first_entry_id(*seq, s->nb_instance, s->instance_id);
576         num_entries = offset + (s->nb_instance * num_entries);
577
578         move_head_atomically(s, &num_entries, &old_head, block, true);
579
580         for (; offset < num_entries; offset += s->nb_instance) {
581                 memcpy(entries_offset, get_slot(t, s->head + offset),
582                         t->slot_size);
583                 entries_offset += t->slot_size;
584                 i++;
585         }
586
587         *seq = old_head;
588
589         return i;
590 }
591
592 /* Claim and copy slot pointers, optimised for single-thread operation */
593 static __rte_always_inline uint32_t
594 opdl_stage_claim_copy_singlethread(struct opdl_stage *s, void *entries,
595                 uint32_t num_entries, uint32_t *seq, bool block)
596 {
597         num_entries = num_to_process(s, num_entries, block);
598         if (num_entries == 0)
599                 return 0;
600         copy_entries_out(s->t, s->head, entries, num_entries);
601         if (seq != NULL)
602                 *seq = s->head;
603         s->head += num_entries;
604         return num_entries;
605 }
606
607 /* Thread-safe version of function to claim and copy pointers to slots */
608 static __rte_always_inline uint32_t
609 opdl_stage_claim_copy_multithread(struct opdl_stage *s, void *entries,
610                 uint32_t num_entries, uint32_t *seq, bool block)
611 {
612         uint32_t old_head;
613
614         move_head_atomically(s, &num_entries, &old_head, block, true);
615         if (num_entries == 0)
616                 return 0;
617         copy_entries_out(s->t, old_head, entries, num_entries);
618         if (seq != NULL)
619                 *seq = old_head;
620         return num_entries;
621 }
622
623 static __rte_always_inline void
624 opdl_stage_disclaim_singlethread_n(struct opdl_stage *s,
625                 uint32_t num_entries)
626 {
627         uint32_t old_tail = s->shared.tail;
628
629         if (unlikely(num_entries > (s->head - old_tail))) {
630                 PMD_DRV_LOG(WARNING, "Attempt to disclaim (%u) more than claimed (%u)",
631                                 num_entries, s->head - old_tail);
632                 num_entries = s->head - old_tail;
633         }
634         __atomic_store_n(&s->shared.tail, num_entries + old_tail,
635                         __ATOMIC_RELEASE);
636 }
637
638 uint32_t
639 opdl_ring_input(struct opdl_ring *t, const void *entries, uint32_t num_entries,
640                 bool block)
641 {
642         if (input_stage(t)->threadsafe == false)
643                 return opdl_ring_input_singlethread(t, entries, num_entries,
644                                 block);
645         else
646                 return opdl_ring_input_multithread(t, entries, num_entries,
647                                 block);
648 }
649
650 uint32_t
651 opdl_ring_copy_from_burst(struct opdl_ring *t, struct opdl_stage *s,
652                 const void *entries, uint32_t num_entries, bool block)
653 {
654         uint32_t head = s->head;
655
656         num_entries = num_to_process(s, num_entries, block);
657
658         if (num_entries == 0)
659                 return 0;
660
661         copy_entries_in(t, head, entries, num_entries);
662
663         s->head += num_entries;
664         __atomic_store_n(&s->shared.tail, s->head, __ATOMIC_RELEASE);
665
666         return num_entries;
667
668 }
669
670 uint32_t
671 opdl_ring_copy_to_burst(struct opdl_ring *t, struct opdl_stage *s,
672                 void *entries, uint32_t num_entries, bool block)
673 {
674         uint32_t head = s->head;
675
676         num_entries = num_to_process(s, num_entries, block);
677         if (num_entries == 0)
678                 return 0;
679
680         copy_entries_out(t, head, entries, num_entries);
681
682         s->head += num_entries;
683         __atomic_store_n(&s->shared.tail, s->head, __ATOMIC_RELEASE);
684
685         return num_entries;
686 }
687
688 uint32_t
689 opdl_stage_find_num_available(struct opdl_stage *s, uint32_t num_entries)
690 {
691         /* return (num_to_process(s, num_entries, false)); */
692
693         if (available(s) >= num_entries)
694                 return num_entries;
695
696         update_available_seq(s);
697
698         uint32_t avail = available(s);
699
700         if (avail == 0) {
701                 rte_pause();
702                 return 0;
703         }
704         return (avail <= num_entries) ? avail : num_entries;
705 }
706
707 uint32_t
708 opdl_stage_claim(struct opdl_stage *s, void *entries,
709                 uint32_t num_entries, uint32_t *seq, bool block, bool atomic)
710 {
711         if (s->threadsafe == false)
712                 return opdl_stage_claim_singlethread(s, entries, num_entries,
713                                 seq, block, atomic);
714         else
715                 return opdl_stage_claim_multithread(s, entries, num_entries,
716                                 seq, block);
717 }
718
719 uint32_t
720 opdl_stage_claim_copy(struct opdl_stage *s, void *entries,
721                 uint32_t num_entries, uint32_t *seq, bool block)
722 {
723         if (s->threadsafe == false)
724                 return opdl_stage_claim_copy_singlethread(s, entries,
725                                 num_entries, seq, block);
726         else
727                 return opdl_stage_claim_copy_multithread(s, entries,
728                                 num_entries, seq, block);
729 }
730
731 void
732 opdl_stage_disclaim_n(struct opdl_stage *s, uint32_t num_entries,
733                 bool block)
734 {
735
736         if (s->threadsafe == false) {
737                 opdl_stage_disclaim_singlethread_n(s, s->num_claimed);
738         } else {
739                 struct claim_manager *disclaims =
740                         &s->pending_disclaims[rte_lcore_id()];
741
742                 if (unlikely(num_entries > s->num_slots)) {
743                         PMD_DRV_LOG(WARNING, "Attempt to disclaim (%u) more than claimed (%u)",
744                                         num_entries, disclaims->num_claimed);
745                         num_entries = disclaims->num_claimed;
746                 }
747
748                 num_entries = RTE_MIN(num_entries + disclaims->num_to_disclaim,
749                                 disclaims->num_claimed);
750                 opdl_stage_disclaim_multithread_n(s, num_entries, block);
751         }
752 }
753
754 int
755 opdl_stage_disclaim(struct opdl_stage *s, uint32_t num_entries, bool block)
756 {
757         if (num_entries != s->num_event) {
758                 rte_errno = -EINVAL;
759                 return 0;
760         }
761         if (s->threadsafe == false) {
762                 __atomic_store_n(&s->shared.tail, s->head, __ATOMIC_RELEASE);
763                 s->seq += s->num_claimed;
764                 s->shadow_head = s->head;
765                 s->num_claimed = 0;
766         } else {
767                 struct claim_manager *disclaims =
768                                 &s->pending_disclaims[rte_lcore_id()];
769                 opdl_stage_disclaim_multithread_n(s, disclaims->num_claimed,
770                                 block);
771         }
772         return num_entries;
773 }
774
775 uint32_t
776 opdl_ring_available(struct opdl_ring *t)
777 {
778         return opdl_stage_available(&t->stages[0]);
779 }
780
781 uint32_t
782 opdl_stage_available(struct opdl_stage *s)
783 {
784         update_available_seq(s);
785         return available(s);
786 }
787
788 void
789 opdl_ring_flush(struct opdl_ring *t)
790 {
791         struct opdl_stage *s = input_stage(t);
792
793         wait_for_available(s, s->num_slots);
794 }
795
796 /******************** Non performance sensitive functions ********************/
797
798 /* Initial setup of a new stage's context */
799 static int
800 init_stage(struct opdl_ring *t, struct opdl_stage *s, bool threadsafe,
801                 bool is_input)
802 {
803         uint32_t available = (is_input) ? t->num_slots : 0;
804
805         s->t = t;
806         s->num_slots = t->num_slots;
807         s->index = t->num_stages;
808         s->threadsafe = threadsafe;
809         s->shared.stage = s;
810
811         /* Alloc memory for deps */
812         s->dep_tracking = rte_zmalloc_socket(LIB_NAME,
813                         t->max_num_stages * sizeof(enum dep_type),
814                         0, t->socket);
815         if (s->dep_tracking == NULL)
816                 return -ENOMEM;
817
818         s->deps = rte_zmalloc_socket(LIB_NAME,
819                         t->max_num_stages * sizeof(struct shared_state *),
820                         0, t->socket);
821         if (s->deps == NULL) {
822                 rte_free(s->dep_tracking);
823                 return -ENOMEM;
824         }
825
826         s->dep_tracking[s->index] = DEP_SELF;
827
828         if (threadsafe == true)
829                 s->shared.available_seq = available;
830         else
831                 s->available_seq = available;
832
833         return 0;
834 }
835
836 /* Add direct or indirect dependencies between stages */
837 static int
838 add_dep(struct opdl_stage *dependent, const struct opdl_stage *dependency,
839                 enum dep_type type)
840 {
841         struct opdl_ring *t = dependent->t;
842         uint32_t i;
843
844         /* Add new direct dependency */
845         if ((type == DEP_DIRECT) &&
846                         (dependent->dep_tracking[dependency->index] ==
847                                         DEP_NONE)) {
848                 PMD_DRV_LOG(DEBUG, "%s:%u direct dependency on %u",
849                                 t->name, dependent->index, dependency->index);
850                 dependent->dep_tracking[dependency->index] = DEP_DIRECT;
851         }
852
853         /* Add new indirect dependency or change direct to indirect */
854         if ((type == DEP_INDIRECT) &&
855                         ((dependent->dep_tracking[dependency->index] ==
856                         DEP_NONE) ||
857                         (dependent->dep_tracking[dependency->index] ==
858                         DEP_DIRECT))) {
859                 PMD_DRV_LOG(DEBUG, "%s:%u indirect dependency on %u",
860                                 t->name, dependent->index, dependency->index);
861                 dependent->dep_tracking[dependency->index] = DEP_INDIRECT;
862         }
863
864         /* Shouldn't happen... */
865         if ((dependent->dep_tracking[dependency->index] == DEP_SELF) &&
866                         (dependent != input_stage(t))) {
867                 PMD_DRV_LOG(ERR, "Loop in dependency graph %s:%u",
868                                 t->name, dependent->index);
869                 return -EINVAL;
870         }
871
872         /* Keep going to dependencies of the dependency, until input stage */
873         if (dependency != input_stage(t))
874                 for (i = 0; i < dependency->num_deps; i++) {
875                         int ret = add_dep(dependent, dependency->deps[i]->stage,
876                                         DEP_INDIRECT);
877
878                         if (ret < 0)
879                                 return ret;
880                 }
881
882         /* Make list of sequence numbers for direct dependencies only */
883         if (type == DEP_DIRECT)
884                 for (i = 0, dependent->num_deps = 0; i < t->num_stages; i++)
885                         if (dependent->dep_tracking[i] == DEP_DIRECT) {
886                                 if ((i == 0) && (dependent->num_deps > 1))
887                                         rte_panic("%s:%u depends on > input",
888                                                         t->name,
889                                                         dependent->index);
890                                 dependent->deps[dependent->num_deps++] =
891                                                 &t->stages[i].shared;
892                         }
893
894         return 0;
895 }
896
897 struct opdl_ring *
898 opdl_ring_create(const char *name, uint32_t num_slots, uint32_t slot_size,
899                 uint32_t max_num_stages, int socket)
900 {
901         struct opdl_ring *t;
902         char mz_name[RTE_MEMZONE_NAMESIZE];
903         int mz_flags = 0;
904         struct opdl_stage *st = NULL;
905         const struct rte_memzone *mz = NULL;
906         size_t alloc_size = RTE_CACHE_LINE_ROUNDUP(sizeof(*t) +
907                         (num_slots * slot_size));
908
909         /* Compile time checking */
910         RTE_BUILD_BUG_ON((sizeof(struct shared_state) & RTE_CACHE_LINE_MASK) !=
911                         0);
912         RTE_BUILD_BUG_ON((offsetof(struct opdl_stage, shared) &
913                         RTE_CACHE_LINE_MASK) != 0);
914         RTE_BUILD_BUG_ON((offsetof(struct opdl_ring, slots) &
915                         RTE_CACHE_LINE_MASK) != 0);
916         RTE_BUILD_BUG_ON(!rte_is_power_of_2(OPDL_DISCLAIMS_PER_LCORE));
917
918         /* Parameter checking */
919         if (name == NULL) {
920                 PMD_DRV_LOG(ERR, "name param is NULL");
921                 return NULL;
922         }
923         if (!rte_is_power_of_2(num_slots)) {
924                 PMD_DRV_LOG(ERR, "num_slots (%u) for %s is not power of 2",
925                                 num_slots, name);
926                 return NULL;
927         }
928
929         /* Alloc memory for stages */
930         st = rte_zmalloc_socket(LIB_NAME,
931                 max_num_stages * sizeof(struct opdl_stage),
932                 RTE_CACHE_LINE_SIZE, socket);
933         if (st == NULL)
934                 goto exit_fail;
935
936         snprintf(mz_name, sizeof(mz_name), "%s%s", LIB_NAME, name);
937
938         /* Alloc memory for memzone */
939         mz = rte_memzone_reserve(mz_name, alloc_size, socket, mz_flags);
940         if (mz == NULL)
941                 goto exit_fail;
942
943         t = mz->addr;
944
945         /* Initialise opdl_ring queue */
946         memset(t, 0, sizeof(*t));
947         snprintf(t->name, sizeof(t->name), "%s", name);
948         t->socket = socket;
949         t->num_slots = num_slots;
950         t->mask = num_slots - 1;
951         t->slot_size = slot_size;
952         t->max_num_stages = max_num_stages;
953         t->stages = st;
954
955         PMD_DRV_LOG(DEBUG, "Created %s at %p (num_slots=%u,socket=%i,slot_size=%u)",
956                         t->name, t, num_slots, socket, slot_size);
957
958         return t;
959
960 exit_fail:
961         PMD_DRV_LOG(ERR, "Cannot reserve memory");
962         rte_free(st);
963         rte_memzone_free(mz);
964
965         return NULL;
966 }
967
968 void *
969 opdl_ring_get_slot(const struct opdl_ring *t, uint32_t index)
970 {
971         return get_slot(t, index);
972 }
973
974 bool
975 opdl_ring_cas_slot(struct opdl_stage *s, const struct rte_event *ev,
976                 uint32_t index, bool atomic)
977 {
978         uint32_t i = 0, offset;
979         struct opdl_ring *t = s->t;
980         struct rte_event *ev_orig = NULL;
981         bool ev_updated = false;
982         uint64_t ev_temp    = 0;
983         uint64_t ev_update  = 0;
984
985         uint32_t opa_id   = 0;
986         uint32_t flow_id  = 0;
987         uint64_t event    = 0;
988
989         if (index > s->num_event) {
990                 PMD_DRV_LOG(ERR, "index is overflow");
991                 return ev_updated;
992         }
993
994         ev_temp = ev->event & OPDL_EVENT_MASK;
995
996         if (!atomic) {
997                 offset = opdl_first_entry_id(s->seq, s->nb_instance,
998                                 s->instance_id);
999                 offset += index*s->nb_instance;
1000                 ev_orig = get_slot(t, s->shadow_head+offset);
1001                 if ((ev_orig->event&OPDL_EVENT_MASK) != ev_temp) {
1002                         ev_orig->event = ev->event;
1003                         ev_updated = true;
1004                 }
1005                 if (ev_orig->u64 != ev->u64) {
1006                         ev_orig->u64 = ev->u64;
1007                         ev_updated = true;
1008                 }
1009
1010         } else {
1011                 for (i = s->pos; i < s->num_claimed; i++) {
1012                         ev_orig = (struct rte_event *)
1013                                 get_slot(t, s->shadow_head+i);
1014
1015                         event  = __atomic_load_n(&(ev_orig->event),
1016                                         __ATOMIC_ACQUIRE);
1017
1018                         opa_id = OPDL_OPA_MASK & (event >> OPDL_OPA_OFFSET);
1019                         flow_id  = OPDL_FLOWID_MASK & event;
1020
1021                         if (opa_id >= s->queue_id)
1022                                 continue;
1023
1024                         if ((flow_id % s->nb_instance) == s->instance_id) {
1025                                 ev_update = s->queue_id;
1026                                 ev_update = (ev_update << OPDL_OPA_OFFSET)
1027                                         | ev->event;
1028
1029                                 s->pos = i + 1;
1030
1031                                 if ((event & OPDL_EVENT_MASK) !=
1032                                                 ev_temp) {
1033                                         __atomic_store_n(&(ev_orig->event),
1034                                                         ev_update,
1035                                                         __ATOMIC_RELEASE);
1036                                         ev_updated = true;
1037                                 }
1038                                 if (ev_orig->u64 != ev->u64) {
1039                                         ev_orig->u64 = ev->u64;
1040                                         ev_updated = true;
1041                                 }
1042
1043                                 break;
1044                         }
1045                 }
1046
1047         }
1048
1049         return ev_updated;
1050 }
1051
1052 int
1053 opdl_ring_get_socket(const struct opdl_ring *t)
1054 {
1055         return t->socket;
1056 }
1057
1058 uint32_t
1059 opdl_ring_get_num_slots(const struct opdl_ring *t)
1060 {
1061         return t->num_slots;
1062 }
1063
1064 const char *
1065 opdl_ring_get_name(const struct opdl_ring *t)
1066 {
1067         return t->name;
1068 }
1069
1070 /* Check dependency list is valid for a given opdl_ring */
1071 static int
1072 check_deps(struct opdl_ring *t, struct opdl_stage *deps[],
1073                 uint32_t num_deps)
1074 {
1075         unsigned int i;
1076
1077         for (i = 0; i < num_deps; ++i) {
1078                 if (!deps[i]) {
1079                         PMD_DRV_LOG(ERR, "deps[%u] is NULL", i);
1080                         return -EINVAL;
1081                 }
1082                 if (t != deps[i]->t) {
1083                         PMD_DRV_LOG(ERR, "deps[%u] is in opdl_ring %s, not %s",
1084                                         i, deps[i]->t->name, t->name);
1085                         return -EINVAL;
1086                 }
1087         }
1088
1089         return 0;
1090 }
1091
1092 struct opdl_stage *
1093 opdl_stage_add(struct opdl_ring *t, bool threadsafe, bool is_input)
1094 {
1095         struct opdl_stage *s;
1096
1097         /* Parameter checking */
1098         if (!t) {
1099                 PMD_DRV_LOG(ERR, "opdl_ring is NULL");
1100                 return NULL;
1101         }
1102         if (t->num_stages == t->max_num_stages) {
1103                 PMD_DRV_LOG(ERR, "%s has max number of stages (%u)",
1104                                 t->name, t->max_num_stages);
1105                 return NULL;
1106         }
1107
1108         s = &t->stages[t->num_stages];
1109
1110         if (((uintptr_t)&s->shared & RTE_CACHE_LINE_MASK) != 0)
1111                 PMD_DRV_LOG(WARNING, "Tail seq num (%p) of %s stage not cache aligned",
1112                                 &s->shared, t->name);
1113
1114         if (init_stage(t, s, threadsafe, is_input) < 0) {
1115                 PMD_DRV_LOG(ERR, "Cannot reserve memory");
1116                 return NULL;
1117         }
1118         t->num_stages++;
1119
1120         return s;
1121 }
1122
1123 uint32_t
1124 opdl_stage_deps_add(struct opdl_ring *t, struct opdl_stage *s,
1125                 uint32_t nb_instance, uint32_t instance_id,
1126                 struct opdl_stage *deps[],
1127                 uint32_t num_deps)
1128 {
1129         uint32_t i;
1130         int ret = 0;
1131
1132         if ((num_deps > 0) && (!deps)) {
1133                 PMD_DRV_LOG(ERR, "%s stage has NULL dependencies", t->name);
1134                 return -1;
1135         }
1136         ret = check_deps(t, deps, num_deps);
1137         if (ret < 0)
1138                 return ret;
1139
1140         for (i = 0; i < num_deps; i++) {
1141                 ret = add_dep(s, deps[i], DEP_DIRECT);
1142                 if (ret < 0)
1143                         return ret;
1144         }
1145
1146         s->nb_instance = nb_instance;
1147         s->instance_id = instance_id;
1148
1149         return ret;
1150 }
1151
1152 struct opdl_stage *
1153 opdl_ring_get_input_stage(const struct opdl_ring *t)
1154 {
1155         return input_stage(t);
1156 }
1157
1158 int
1159 opdl_stage_set_deps(struct opdl_stage *s, struct opdl_stage *deps[],
1160                 uint32_t num_deps)
1161 {
1162         unsigned int i;
1163         int ret;
1164
1165         if ((num_deps == 0) || (!deps)) {
1166                 PMD_DRV_LOG(ERR, "cannot set NULL dependencies");
1167                 return -EINVAL;
1168         }
1169
1170         ret = check_deps(s->t, deps, num_deps);
1171         if (ret < 0)
1172                 return ret;
1173
1174         /* Update deps */
1175         for (i = 0; i < num_deps; i++)
1176                 s->deps[i] = &deps[i]->shared;
1177         s->num_deps = num_deps;
1178
1179         return 0;
1180 }
1181
1182 struct opdl_ring *
1183 opdl_stage_get_opdl_ring(const struct opdl_stage *s)
1184 {
1185         return s->t;
1186 }
1187
1188 void
1189 opdl_stage_set_queue_id(struct opdl_stage *s,
1190                 uint32_t queue_id)
1191 {
1192         s->queue_id = queue_id;
1193 }
1194
1195 void
1196 opdl_ring_dump(const struct opdl_ring *t, FILE *f)
1197 {
1198         uint32_t i;
1199
1200         if (t == NULL) {
1201                 fprintf(f, "NULL OPDL!\n");
1202                 return;
1203         }
1204         fprintf(f, "OPDL \"%s\": num_slots=%u; mask=%#x; slot_size=%u; num_stages=%u; socket=%i\n",
1205                         t->name, t->num_slots, t->mask, t->slot_size,
1206                         t->num_stages, t->socket);
1207         for (i = 0; i < t->num_stages; i++) {
1208                 uint32_t j;
1209                 const struct opdl_stage *s = &t->stages[i];
1210
1211                 fprintf(f, "  %s[%u]: threadsafe=%s; head=%u; available_seq=%u; tail=%u; deps=%u",
1212                                 t->name, i, (s->threadsafe) ? "true" : "false",
1213                                 (s->threadsafe) ? s->shared.head : s->head,
1214                                 (s->threadsafe) ? s->shared.available_seq :
1215                                 s->available_seq,
1216                                 s->shared.tail, (s->num_deps > 0) ?
1217                                 s->deps[0]->stage->index : 0);
1218                 for (j = 1; j < s->num_deps; j++)
1219                         fprintf(f, ",%u", s->deps[j]->stage->index);
1220                 fprintf(f, "\n");
1221         }
1222         fflush(f);
1223 }
1224
1225 void
1226 opdl_ring_free(struct opdl_ring *t)
1227 {
1228         uint32_t i;
1229         const struct rte_memzone *mz;
1230         char mz_name[RTE_MEMZONE_NAMESIZE];
1231
1232         if (t == NULL) {
1233                 PMD_DRV_LOG(DEBUG, "Freeing NULL OPDL Ring!");
1234                 return;
1235         }
1236
1237         PMD_DRV_LOG(DEBUG, "Freeing %s opdl_ring at %p", t->name, t);
1238
1239         for (i = 0; i < t->num_stages; ++i) {
1240                 rte_free(t->stages[i].deps);
1241                 rte_free(t->stages[i].dep_tracking);
1242         }
1243
1244         rte_free(t->stages);
1245
1246         snprintf(mz_name, sizeof(mz_name), "%s%s", LIB_NAME, t->name);
1247         mz = rte_memzone_lookup(mz_name);
1248         if (rte_memzone_free(mz) != 0)
1249                 PMD_DRV_LOG(ERR, "Cannot free memzone for %s", t->name);
1250 }
1251
1252 /* search a opdl_ring from its name */
1253 struct opdl_ring *
1254 opdl_ring_lookup(const char *name)
1255 {
1256         const struct rte_memzone *mz;
1257         char mz_name[RTE_MEMZONE_NAMESIZE];
1258
1259         snprintf(mz_name, sizeof(mz_name), "%s%s", LIB_NAME, name);
1260
1261         mz = rte_memzone_lookup(mz_name);
1262         if (mz == NULL)
1263                 return NULL;
1264
1265         return mz->addr;
1266 }
1267
1268 void
1269 opdl_ring_set_stage_threadsafe(struct opdl_stage *s, bool threadsafe)
1270 {
1271         s->threadsafe = threadsafe;
1272 }