drivers/event/opdl/opdl_ring.c

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