drivers/event/dlb2/dlb2.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2016-2020 Intel Corporation
   3  */
   4
   5 #include <assert.h>
   6 #include <errno.h>
   7 #include <nmmintrin.h>
   8 #include <pthread.h>
   9 #include <stdint.h>
  10 #include <stdbool.h>
  11 #include <stdio.h>
  12 #include <string.h>
  13 #include <sys/mman.h>
  14 #include <fcntl.h>
  15
  16 #include <rte_common.h>
  17 #include <rte_config.h>
  18 #include <rte_cycles.h>
  19 #include <rte_debug.h>
  20 #include <rte_dev.h>
  21 #include <rte_errno.h>
  22 #include <rte_eventdev.h>
  23 #include <eventdev_pmd.h>
  24 #include <rte_io.h>
  25 #include <rte_kvargs.h>
  26 #include <rte_log.h>
  27 #include <rte_malloc.h>
  28 #include <rte_mbuf.h>
  29 #include <rte_power_intrinsics.h>
  30 #include <rte_prefetch.h>
  31 #include <rte_ring.h>
  32 #include <rte_string_fns.h>
  33
  34 #include "dlb2_priv.h"
  35 #include "dlb2_iface.h"
  36 #include "dlb2_inline_fns.h"
  37
  38 /*
  39  * Resources exposed to eventdev. Some values overridden at runtime using
  40  * values returned by the DLB kernel driver.
  41  */
  42 #if (RTE_EVENT_MAX_QUEUES_PER_DEV > UINT8_MAX)
  43 #error "RTE_EVENT_MAX_QUEUES_PER_DEV cannot fit in member max_event_queues"
  44 #endif
  45 static struct rte_event_dev_info evdev_dlb2_default_info = {
  46         .driver_name = "", /* probe will set */
  47         .min_dequeue_timeout_ns = DLB2_MIN_DEQUEUE_TIMEOUT_NS,
  48         .max_dequeue_timeout_ns = DLB2_MAX_DEQUEUE_TIMEOUT_NS,
  49 #if (RTE_EVENT_MAX_QUEUES_PER_DEV < DLB2_MAX_NUM_LDB_QUEUES)
  50         .max_event_queues = RTE_EVENT_MAX_QUEUES_PER_DEV,
  51 #else
  52         .max_event_queues = DLB2_MAX_NUM_LDB_QUEUES,
  53 #endif
  54         .max_event_queue_flows = DLB2_MAX_NUM_FLOWS,
  55         .max_event_queue_priority_levels = DLB2_QID_PRIORITIES,
  56         .max_event_priority_levels = DLB2_QID_PRIORITIES,
  57         .max_event_ports = DLB2_MAX_NUM_LDB_PORTS,
  58         .max_event_port_dequeue_depth = DLB2_MAX_CQ_DEPTH,
  59         .max_event_port_enqueue_depth = DLB2_MAX_ENQUEUE_DEPTH,
  60         .max_event_port_links = DLB2_MAX_NUM_QIDS_PER_LDB_CQ,
  61         .max_num_events = DLB2_MAX_NUM_LDB_CREDITS,
  62         .max_single_link_event_port_queue_pairs =
  63                 DLB2_MAX_NUM_DIR_PORTS(DLB2_HW_V2),
  64         .event_dev_cap = (RTE_EVENT_DEV_CAP_QUEUE_QOS |
  65                           RTE_EVENT_DEV_CAP_EVENT_QOS |
  66                           RTE_EVENT_DEV_CAP_BURST_MODE |
  67                           RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED |
  68                           RTE_EVENT_DEV_CAP_IMPLICIT_RELEASE_DISABLE |
  69                           RTE_EVENT_DEV_CAP_QUEUE_ALL_TYPES),
  70 };
  71
  72 struct process_local_port_data
  73 dlb2_port[DLB2_MAX_NUM_PORTS_ALL][DLB2_NUM_PORT_TYPES];
  74
  75 static void
  76 dlb2_free_qe_mem(struct dlb2_port *qm_port)
  77 {
  78         if (qm_port == NULL)
  79                 return;
  80
  81         rte_free(qm_port->qe4);
  82         qm_port->qe4 = NULL;
  83
  84         rte_free(qm_port->int_arm_qe);
  85         qm_port->int_arm_qe = NULL;
  86
  87         rte_free(qm_port->consume_qe);
  88         qm_port->consume_qe = NULL;
  89
  90         rte_memzone_free(dlb2_port[qm_port->id][PORT_TYPE(qm_port)].mz);
  91         dlb2_port[qm_port->id][PORT_TYPE(qm_port)].mz = NULL;
  92 }
  93
  94 /* override defaults with value(s) provided on command line */
  95 static void
  96 dlb2_init_queue_depth_thresholds(struct dlb2_eventdev *dlb2,
  97                                  int *qid_depth_thresholds)
  98 {
  99         int q;
 100
 101         for (q = 0; q < DLB2_MAX_NUM_QUEUES(dlb2->version); q++) {
 102                 if (qid_depth_thresholds[q] != 0)
 103                         dlb2->ev_queues[q].depth_threshold =
 104                                 qid_depth_thresholds[q];
 105         }
 106 }
 107
 108 static int
 109 dlb2_hw_query_resources(struct dlb2_eventdev *dlb2)
 110 {
 111         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
 112         struct dlb2_hw_resource_info *dlb2_info = &handle->info;
 113         int ret;
 114
 115         /* Query driver resources provisioned for this device */
 116
 117         ret = dlb2_iface_get_num_resources(handle,
 118                                            &dlb2->hw_rsrc_query_results);
 119         if (ret) {
 120                 DLB2_LOG_ERR("ioctl get dlb2 num resources, err=%d\n", ret);
 121                 return ret;
 122         }
 123
 124         /* Complete filling in device resource info returned to evdev app,
 125          * overriding any default values.
 126          * The capabilities (CAPs) were set at compile time.
 127          */
 128
 129         evdev_dlb2_default_info.max_event_queues =
 130                 dlb2->hw_rsrc_query_results.num_ldb_queues;
 131
 132         evdev_dlb2_default_info.max_event_ports =
 133                 dlb2->hw_rsrc_query_results.num_ldb_ports;
 134
 135         if (dlb2->version == DLB2_HW_V2_5) {
 136                 evdev_dlb2_default_info.max_num_events =
 137                         dlb2->hw_rsrc_query_results.num_credits;
 138         } else {
 139                 evdev_dlb2_default_info.max_num_events =
 140                         dlb2->hw_rsrc_query_results.num_ldb_credits;
 141         }
 142         /* Save off values used when creating the scheduling domain. */
 143
 144         handle->info.num_sched_domains =
 145                 dlb2->hw_rsrc_query_results.num_sched_domains;
 146
 147         if (dlb2->version == DLB2_HW_V2_5) {
 148                 handle->info.hw_rsrc_max.nb_events_limit =
 149                         dlb2->hw_rsrc_query_results.num_credits;
 150         } else {
 151                 handle->info.hw_rsrc_max.nb_events_limit =
 152                         dlb2->hw_rsrc_query_results.num_ldb_credits;
 153         }
 154         handle->info.hw_rsrc_max.num_queues =
 155                 dlb2->hw_rsrc_query_results.num_ldb_queues +
 156                 dlb2->hw_rsrc_query_results.num_dir_ports;
 157
 158         handle->info.hw_rsrc_max.num_ldb_queues =
 159                 dlb2->hw_rsrc_query_results.num_ldb_queues;
 160
 161         handle->info.hw_rsrc_max.num_ldb_ports =
 162                 dlb2->hw_rsrc_query_results.num_ldb_ports;
 163
 164         handle->info.hw_rsrc_max.num_dir_ports =
 165                 dlb2->hw_rsrc_query_results.num_dir_ports;
 166
 167         handle->info.hw_rsrc_max.reorder_window_size =
 168                 dlb2->hw_rsrc_query_results.num_hist_list_entries;
 169
 170         rte_memcpy(dlb2_info, &handle->info.hw_rsrc_max, sizeof(*dlb2_info));
 171
 172         return 0;
 173 }
 174
 175 #define DLB2_BASE_10 10
 176
 177 static int
 178 dlb2_string_to_int(int *result, const char *str)
 179 {
 180         long ret;
 181         char *endptr;
 182
 183         if (str == NULL || result == NULL)
 184                 return -EINVAL;
 185
 186         errno = 0;
 187         ret = strtol(str, &endptr, DLB2_BASE_10);
 188         if (errno)
 189                 return -errno;
 190
 191         /* long int and int may be different width for some architectures */
 192         if (ret < INT_MIN || ret > INT_MAX || endptr == str)
 193                 return -EINVAL;
 194
 195         *result = ret;
 196         return 0;
 197 }
 198
 199 static int
 200 set_numa_node(const char *key __rte_unused, const char *value, void *opaque)
 201 {
 202         int *socket_id = opaque;
 203         int ret;
 204
 205         ret = dlb2_string_to_int(socket_id, value);
 206         if (ret < 0)
 207                 return ret;
 208
 209         if (*socket_id > RTE_MAX_NUMA_NODES)
 210                 return -EINVAL;
 211         return 0;
 212 }
 213
 214 static int
 215 set_max_num_events(const char *key __rte_unused,
 216                    const char *value,
 217                    void *opaque)
 218 {
 219         int *max_num_events = opaque;
 220         int ret;
 221
 222         if (value == NULL || opaque == NULL) {
 223                 DLB2_LOG_ERR("NULL pointer\n");
 224                 return -EINVAL;
 225         }
 226
 227         ret = dlb2_string_to_int(max_num_events, value);
 228         if (ret < 0)
 229                 return ret;
 230
 231         if (*max_num_events < 0 || *max_num_events >
 232                         DLB2_MAX_NUM_LDB_CREDITS) {
 233                 DLB2_LOG_ERR("dlb2: max_num_events must be between 0 and %d\n",
 234                              DLB2_MAX_NUM_LDB_CREDITS);
 235                 return -EINVAL;
 236         }
 237
 238         return 0;
 239 }
 240
 241 static int
 242 set_num_dir_credits(const char *key __rte_unused,
 243                     const char *value,
 244                     void *opaque)
 245 {
 246         int *num_dir_credits = opaque;
 247         int ret;
 248
 249         if (value == NULL || opaque == NULL) {
 250                 DLB2_LOG_ERR("NULL pointer\n");
 251                 return -EINVAL;
 252         }
 253
 254         ret = dlb2_string_to_int(num_dir_credits, value);
 255         if (ret < 0)
 256                 return ret;
 257
 258         if (*num_dir_credits < 0 ||
 259             *num_dir_credits > DLB2_MAX_NUM_DIR_CREDITS(DLB2_HW_V2)) {
 260                 DLB2_LOG_ERR("dlb2: num_dir_credits must be between 0 and %d\n",
 261                              DLB2_MAX_NUM_DIR_CREDITS(DLB2_HW_V2));
 262                 return -EINVAL;
 263         }
 264
 265         return 0;
 266 }
 267
 268 static int
 269 set_dev_id(const char *key __rte_unused,
 270            const char *value,
 271            void *opaque)
 272 {
 273         int *dev_id = opaque;
 274         int ret;
 275
 276         if (value == NULL || opaque == NULL) {
 277                 DLB2_LOG_ERR("NULL pointer\n");
 278                 return -EINVAL;
 279         }
 280
 281         ret = dlb2_string_to_int(dev_id, value);
 282         if (ret < 0)
 283                 return ret;
 284
 285         return 0;
 286 }
 287
 288 static int
 289 set_cos(const char *key __rte_unused,
 290         const char *value,
 291         void *opaque)
 292 {
 293         enum dlb2_cos *cos_id = opaque;
 294         int x = 0;
 295         int ret;
 296
 297         if (value == NULL || opaque == NULL) {
 298                 DLB2_LOG_ERR("NULL pointer\n");
 299                 return -EINVAL;
 300         }
 301
 302         ret = dlb2_string_to_int(&x, value);
 303         if (ret < 0)
 304                 return ret;
 305
 306         if (x != DLB2_COS_DEFAULT && (x < DLB2_COS_0 || x > DLB2_COS_3)) {
 307                 DLB2_LOG_ERR(
 308                         "COS %d out of range, must be DLB2_COS_DEFAULT or 0-3\n",
 309                         x);
 310                 return -EINVAL;
 311         }
 312
 313         *cos_id = x;
 314
 315         return 0;
 316 }
 317
 318 static int
 319 set_poll_interval(const char *key __rte_unused,
 320         const char *value,
 321         void *opaque)
 322 {
 323         int *poll_interval = opaque;
 324         int ret;
 325
 326         if (value == NULL || opaque == NULL) {
 327                 DLB2_LOG_ERR("NULL pointer\n");
 328                 return -EINVAL;
 329         }
 330
 331         ret = dlb2_string_to_int(poll_interval, value);
 332         if (ret < 0)
 333                 return ret;
 334
 335         return 0;
 336 }
 337
 338 static int
 339 set_sw_credit_quanta(const char *key __rte_unused,
 340         const char *value,
 341         void *opaque)
 342 {
 343         int *sw_credit_quanta = opaque;
 344         int ret;
 345
 346         if (value == NULL || opaque == NULL) {
 347                 DLB2_LOG_ERR("NULL pointer\n");
 348                 return -EINVAL;
 349         }
 350
 351         ret = dlb2_string_to_int(sw_credit_quanta, value);
 352         if (ret < 0)
 353                 return ret;
 354
 355         return 0;
 356 }
 357
 358 static int
 359 set_default_depth_thresh(const char *key __rte_unused,
 360         const char *value,
 361         void *opaque)
 362 {
 363         int *default_depth_thresh = opaque;
 364         int ret;
 365
 366         if (value == NULL || opaque == NULL) {
 367                 DLB2_LOG_ERR("NULL pointer\n");
 368                 return -EINVAL;
 369         }
 370
 371         ret = dlb2_string_to_int(default_depth_thresh, value);
 372         if (ret < 0)
 373                 return ret;
 374
 375         return 0;
 376 }
 377
 378 static int
 379 set_vector_opts_disab(const char *key __rte_unused,
 380         const char *value,
 381         void *opaque)
 382 {
 383         bool *dlb2_vector_opts_disabled = opaque;
 384
 385         if (value == NULL || opaque == NULL) {
 386                 DLB2_LOG_ERR("NULL pointer\n");
 387                 return -EINVAL;
 388         }
 389
 390         if ((*value == 'y') || (*value == 'Y'))
 391                 *dlb2_vector_opts_disabled = true;
 392         else
 393                 *dlb2_vector_opts_disabled = false;
 394
 395         return 0;
 396 }
 397
 398 static int
 399 set_qid_depth_thresh(const char *key __rte_unused,
 400                      const char *value,
 401                      void *opaque)
 402 {
 403         struct dlb2_qid_depth_thresholds *qid_thresh = opaque;
 404         int first, last, thresh, i;
 405
 406         if (value == NULL || opaque == NULL) {
 407                 DLB2_LOG_ERR("NULL pointer\n");
 408                 return -EINVAL;
 409         }
 410
 411         /* command line override may take one of the following 3 forms:
 412          * qid_depth_thresh=all:<threshold_value> ... all queues
 413          * qid_depth_thresh=qidA-qidB:<threshold_value> ... a range of queues
 414          * qid_depth_thresh=qid:<threshold_value> ... just one queue
 415          */
 416         if (sscanf(value, "all:%d", &thresh) == 1) {
 417                 first = 0;
 418                 last = DLB2_MAX_NUM_QUEUES(DLB2_HW_V2) - 1;
 419         } else if (sscanf(value, "%d-%d:%d", &first, &last, &thresh) == 3) {
 420                 /* we have everything we need */
 421         } else if (sscanf(value, "%d:%d", &first, &thresh) == 2) {
 422                 last = first;
 423         } else {
 424                 DLB2_LOG_ERR("Error parsing qid depth devarg. Should be all:val, qid-qid:val, or qid:val\n");
 425                 return -EINVAL;
 426         }
 427
 428         if (first > last || first < 0 ||
 429                 last >= DLB2_MAX_NUM_QUEUES(DLB2_HW_V2)) {
 430                 DLB2_LOG_ERR("Error parsing qid depth devarg, invalid qid value\n");
 431                 return -EINVAL;
 432         }
 433
 434         if (thresh < 0 || thresh > DLB2_MAX_QUEUE_DEPTH_THRESHOLD) {
 435                 DLB2_LOG_ERR("Error parsing qid depth devarg, threshold > %d\n",
 436                              DLB2_MAX_QUEUE_DEPTH_THRESHOLD);
 437                 return -EINVAL;
 438         }
 439
 440         for (i = first; i <= last; i++)
 441                 qid_thresh->val[i] = thresh; /* indexed by qid */
 442
 443         return 0;
 444 }
 445
 446 static int
 447 set_qid_depth_thresh_v2_5(const char *key __rte_unused,
 448                           const char *value,
 449                           void *opaque)
 450 {
 451         struct dlb2_qid_depth_thresholds *qid_thresh = opaque;
 452         int first, last, thresh, i;
 453
 454         if (value == NULL || opaque == NULL) {
 455                 DLB2_LOG_ERR("NULL pointer\n");
 456                 return -EINVAL;
 457         }
 458
 459         /* command line override may take one of the following 3 forms:
 460          * qid_depth_thresh=all:<threshold_value> ... all queues
 461          * qid_depth_thresh=qidA-qidB:<threshold_value> ... a range of queues
 462          * qid_depth_thresh=qid:<threshold_value> ... just one queue
 463          */
 464         if (sscanf(value, "all:%d", &thresh) == 1) {
 465                 first = 0;
 466                 last = DLB2_MAX_NUM_QUEUES(DLB2_HW_V2_5) - 1;
 467         } else if (sscanf(value, "%d-%d:%d", &first, &last, &thresh) == 3) {
 468                 /* we have everything we need */
 469         } else if (sscanf(value, "%d:%d", &first, &thresh) == 2) {
 470                 last = first;
 471         } else {
 472                 DLB2_LOG_ERR("Error parsing qid depth devarg. Should be all:val, qid-qid:val, or qid:val\n");
 473                 return -EINVAL;
 474         }
 475
 476         if (first > last || first < 0 ||
 477                 last >= DLB2_MAX_NUM_QUEUES(DLB2_HW_V2_5)) {
 478                 DLB2_LOG_ERR("Error parsing qid depth devarg, invalid qid value\n");
 479                 return -EINVAL;
 480         }
 481
 482         if (thresh < 0 || thresh > DLB2_MAX_QUEUE_DEPTH_THRESHOLD) {
 483                 DLB2_LOG_ERR("Error parsing qid depth devarg, threshold > %d\n",
 484                              DLB2_MAX_QUEUE_DEPTH_THRESHOLD);
 485                 return -EINVAL;
 486         }
 487
 488         for (i = first; i <= last; i++)
 489                 qid_thresh->val[i] = thresh; /* indexed by qid */
 490
 491         return 0;
 492 }
 493
 494 static void
 495 dlb2_eventdev_info_get(struct rte_eventdev *dev,
 496                        struct rte_event_dev_info *dev_info)
 497 {
 498         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
 499         int ret;
 500
 501         ret = dlb2_hw_query_resources(dlb2);
 502         if (ret) {
 503                 const struct rte_eventdev_data *data = dev->data;
 504
 505                 DLB2_LOG_ERR("get resources err=%d, devid=%d\n",
 506                              ret, data->dev_id);
 507                 /* fn is void, so fall through and return values set up in
 508                  * probe
 509                  */
 510         }
 511
 512         /* Add num resources currently owned by this domain.
 513          * These would become available if the scheduling domain were reset due
 514          * to the application recalling eventdev_configure to *reconfigure* the
 515          * domain.
 516          */
 517         evdev_dlb2_default_info.max_event_ports += dlb2->num_ldb_ports;
 518         evdev_dlb2_default_info.max_event_queues += dlb2->num_ldb_queues;
 519         if (dlb2->version == DLB2_HW_V2_5) {
 520                 evdev_dlb2_default_info.max_num_events +=
 521                         dlb2->max_credits;
 522         } else {
 523                 evdev_dlb2_default_info.max_num_events +=
 524                         dlb2->max_ldb_credits;
 525         }
 526         evdev_dlb2_default_info.max_event_queues =
 527                 RTE_MIN(evdev_dlb2_default_info.max_event_queues,
 528                         RTE_EVENT_MAX_QUEUES_PER_DEV);
 529
 530         evdev_dlb2_default_info.max_num_events =
 531                 RTE_MIN(evdev_dlb2_default_info.max_num_events,
 532                         dlb2->max_num_events_override);
 533
 534         *dev_info = evdev_dlb2_default_info;
 535 }
 536
 537 static int
 538 dlb2_hw_create_sched_domain(struct dlb2_hw_dev *handle,
 539                             const struct dlb2_hw_rsrcs *resources_asked,
 540                             uint8_t device_version)
 541 {
 542         int ret = 0;
 543         struct dlb2_create_sched_domain_args *cfg;
 544
 545         if (resources_asked == NULL) {
 546                 DLB2_LOG_ERR("dlb2: dlb2_create NULL parameter\n");
 547                 ret = EINVAL;
 548                 goto error_exit;
 549         }
 550
 551         /* Map generic qm resources to dlb2 resources */
 552         cfg = &handle->cfg.resources;
 553
 554         /* DIR ports and queues */
 555
 556         cfg->num_dir_ports = resources_asked->num_dir_ports;
 557         if (device_version == DLB2_HW_V2_5)
 558                 cfg->num_credits = resources_asked->num_credits;
 559         else
 560                 cfg->num_dir_credits = resources_asked->num_dir_credits;
 561
 562         /* LDB queues */
 563
 564         cfg->num_ldb_queues = resources_asked->num_ldb_queues;
 565
 566         /* LDB ports */
 567
 568         cfg->cos_strict = 0; /* Best effort */
 569         cfg->num_cos_ldb_ports[0] = 0;
 570         cfg->num_cos_ldb_ports[1] = 0;
 571         cfg->num_cos_ldb_ports[2] = 0;
 572         cfg->num_cos_ldb_ports[3] = 0;
 573
 574         switch (handle->cos_id) {
 575         case DLB2_COS_0:
 576                 cfg->num_ldb_ports = 0; /* no don't care ports */
 577                 cfg->num_cos_ldb_ports[0] =
 578                         resources_asked->num_ldb_ports;
 579                 break;
 580         case DLB2_COS_1:
 581                 cfg->num_ldb_ports = 0; /* no don't care ports */
 582                 cfg->num_cos_ldb_ports[1] = resources_asked->num_ldb_ports;
 583                 break;
 584         case DLB2_COS_2:
 585                 cfg->num_ldb_ports = 0; /* no don't care ports */
 586                 cfg->num_cos_ldb_ports[2] = resources_asked->num_ldb_ports;
 587                 break;
 588         case DLB2_COS_3:
 589                 cfg->num_ldb_ports = 0; /* no don't care ports */
 590                 cfg->num_cos_ldb_ports[3] =
 591                         resources_asked->num_ldb_ports;
 592                 break;
 593         case DLB2_COS_DEFAULT:
 594                 /* all ldb ports are don't care ports from a cos perspective */
 595                 cfg->num_ldb_ports =
 596                         resources_asked->num_ldb_ports;
 597                 break;
 598         }
 599
 600         if (device_version == DLB2_HW_V2)
 601                 cfg->num_ldb_credits = resources_asked->num_ldb_credits;
 602
 603         cfg->num_atomic_inflights =
 604                 DLB2_NUM_ATOMIC_INFLIGHTS_PER_QUEUE *
 605                 cfg->num_ldb_queues;
 606
 607         cfg->num_hist_list_entries = resources_asked->num_ldb_ports *
 608                 DLB2_NUM_HIST_LIST_ENTRIES_PER_LDB_PORT;
 609
 610         if (device_version == DLB2_HW_V2_5) {
 611                 DLB2_LOG_DBG("sched domain create - ldb_qs=%d, ldb_ports=%d, dir_ports=%d, atomic_inflights=%d, hist_list_entries=%d, credits=%d\n",
 612                              cfg->num_ldb_queues,
 613                              resources_asked->num_ldb_ports,
 614                              cfg->num_dir_ports,
 615                              cfg->num_atomic_inflights,
 616                              cfg->num_hist_list_entries,
 617                              cfg->num_credits);
 618         } else {
 619                 DLB2_LOG_DBG("sched domain create - ldb_qs=%d, ldb_ports=%d, dir_ports=%d, atomic_inflights=%d, hist_list_entries=%d, ldb_credits=%d, dir_credits=%d\n",
 620                              cfg->num_ldb_queues,
 621                              resources_asked->num_ldb_ports,
 622                              cfg->num_dir_ports,
 623                              cfg->num_atomic_inflights,
 624                              cfg->num_hist_list_entries,
 625                              cfg->num_ldb_credits,
 626                              cfg->num_dir_credits);
 627         }
 628
 629         /* Configure the QM */
 630
 631         ret = dlb2_iface_sched_domain_create(handle, cfg);
 632         if (ret < 0) {
 633                 DLB2_LOG_ERR("dlb2: domain create failed, ret = %d, extra status: %s\n",
 634                              ret,
 635                              dlb2_error_strings[cfg->response.status]);
 636
 637                 goto error_exit;
 638         }
 639
 640         handle->domain_id = cfg->response.id;
 641         handle->cfg.configured = true;
 642
 643 error_exit:
 644
 645         return ret;
 646 }
 647
 648 static void
 649 dlb2_hw_reset_sched_domain(const struct rte_eventdev *dev, bool reconfig)
 650 {
 651         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
 652         enum dlb2_configuration_state config_state;
 653         int i, j;
 654
 655         dlb2_iface_domain_reset(dlb2);
 656
 657         /* Free all dynamically allocated port memory */
 658         for (i = 0; i < dlb2->num_ports; i++)
 659                 dlb2_free_qe_mem(&dlb2->ev_ports[i].qm_port);
 660
 661         /* If reconfiguring, mark the device's queues and ports as "previously
 662          * configured." If the user doesn't reconfigure them, the PMD will
 663          * reapply their previous configuration when the device is started.
 664          */
 665         config_state = (reconfig) ? DLB2_PREV_CONFIGURED :
 666                 DLB2_NOT_CONFIGURED;
 667
 668         for (i = 0; i < dlb2->num_ports; i++) {
 669                 dlb2->ev_ports[i].qm_port.config_state = config_state;
 670                 /* Reset setup_done so ports can be reconfigured */
 671                 dlb2->ev_ports[i].setup_done = false;
 672                 for (j = 0; j < DLB2_MAX_NUM_QIDS_PER_LDB_CQ; j++)
 673                         dlb2->ev_ports[i].link[j].mapped = false;
 674         }
 675
 676         for (i = 0; i < dlb2->num_queues; i++)
 677                 dlb2->ev_queues[i].qm_queue.config_state = config_state;
 678
 679         for (i = 0; i < DLB2_MAX_NUM_QUEUES(DLB2_HW_V2_5); i++)
 680                 dlb2->ev_queues[i].setup_done = false;
 681
 682         dlb2->num_ports = 0;
 683         dlb2->num_ldb_ports = 0;
 684         dlb2->num_dir_ports = 0;
 685         dlb2->num_queues = 0;
 686         dlb2->num_ldb_queues = 0;
 687         dlb2->num_dir_queues = 0;
 688         dlb2->configured = false;
 689 }
 690
 691 /* Note: 1 QM instance per QM device, QM instance/device == event device */
 692 static int
 693 dlb2_eventdev_configure(const struct rte_eventdev *dev)
 694 {
 695         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
 696         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
 697         struct dlb2_hw_rsrcs *rsrcs = &handle->info.hw_rsrc_max;
 698         const struct rte_eventdev_data *data = dev->data;
 699         const struct rte_event_dev_config *config = &data->dev_conf;
 700         int ret;
 701
 702         /* If this eventdev is already configured, we must release the current
 703          * scheduling domain before attempting to configure a new one.
 704          */
 705         if (dlb2->configured) {
 706                 dlb2_hw_reset_sched_domain(dev, true);
 707                 ret = dlb2_hw_query_resources(dlb2);
 708                 if (ret) {
 709                         DLB2_LOG_ERR("get resources err=%d, devid=%d\n",
 710                                      ret, data->dev_id);
 711                         return ret;
 712                 }
 713         }
 714
 715         if (config->nb_event_queues > rsrcs->num_queues) {
 716                 DLB2_LOG_ERR("nb_event_queues parameter (%d) exceeds the QM device's capabilities (%d).\n",
 717                              config->nb_event_queues,
 718                              rsrcs->num_queues);
 719                 return -EINVAL;
 720         }
 721         if (config->nb_event_ports > (rsrcs->num_ldb_ports
 722                         + rsrcs->num_dir_ports)) {
 723                 DLB2_LOG_ERR("nb_event_ports parameter (%d) exceeds the QM device's capabilities (%d).\n",
 724                              config->nb_event_ports,
 725                              (rsrcs->num_ldb_ports + rsrcs->num_dir_ports));
 726                 return -EINVAL;
 727         }
 728         if (config->nb_events_limit > rsrcs->nb_events_limit) {
 729                 DLB2_LOG_ERR("nb_events_limit parameter (%d) exceeds the QM device's capabilities (%d).\n",
 730                              config->nb_events_limit,
 731                              rsrcs->nb_events_limit);
 732                 return -EINVAL;
 733         }
 734
 735         if (config->event_dev_cfg & RTE_EVENT_DEV_CFG_PER_DEQUEUE_TIMEOUT)
 736                 dlb2->global_dequeue_wait = false;
 737         else {
 738                 uint32_t timeout32;
 739
 740                 dlb2->global_dequeue_wait = true;
 741
 742                 /* note size mismatch of timeout vals in eventdev lib. */
 743                 timeout32 = config->dequeue_timeout_ns;
 744
 745                 dlb2->global_dequeue_wait_ticks =
 746                         timeout32 * (rte_get_timer_hz() / 1E9);
 747         }
 748
 749         /* Does this platform support umonitor/umwait? */
 750         if (rte_cpu_get_flag_enabled(RTE_CPUFLAG_WAITPKG))
 751                 dlb2->umwait_allowed = true;
 752
 753         rsrcs->num_dir_ports = config->nb_single_link_event_port_queues;
 754         rsrcs->num_ldb_ports  = config->nb_event_ports - rsrcs->num_dir_ports;
 755         /* 1 dir queue per dir port */
 756         rsrcs->num_ldb_queues = config->nb_event_queues - rsrcs->num_dir_ports;
 757
 758         if (dlb2->version == DLB2_HW_V2_5) {
 759                 rsrcs->num_credits = 0;
 760                 if (rsrcs->num_ldb_queues || rsrcs->num_dir_ports)
 761                         rsrcs->num_credits = config->nb_events_limit;
 762         } else {
 763                 /* Scale down nb_events_limit by 4 for directed credits,
 764                  * since there are 4x as many load-balanced credits.
 765                  */
 766                 rsrcs->num_ldb_credits = 0;
 767                 rsrcs->num_dir_credits = 0;
 768
 769                 if (rsrcs->num_ldb_queues)
 770                         rsrcs->num_ldb_credits = config->nb_events_limit;
 771                 if (rsrcs->num_dir_ports)
 772                         rsrcs->num_dir_credits = config->nb_events_limit / 4;
 773                 if (dlb2->num_dir_credits_override != -1)
 774                         rsrcs->num_dir_credits = dlb2->num_dir_credits_override;
 775         }
 776
 777         if (dlb2_hw_create_sched_domain(handle, rsrcs, dlb2->version) < 0) {
 778                 DLB2_LOG_ERR("dlb2_hw_create_sched_domain failed\n");
 779                 return -ENODEV;
 780         }
 781
 782         dlb2->new_event_limit = config->nb_events_limit;
 783         __atomic_store_n(&dlb2->inflights, 0, __ATOMIC_SEQ_CST);
 784
 785         /* Save number of ports/queues for this event dev */
 786         dlb2->num_ports = config->nb_event_ports;
 787         dlb2->num_queues = config->nb_event_queues;
 788         dlb2->num_dir_ports = rsrcs->num_dir_ports;
 789         dlb2->num_ldb_ports = dlb2->num_ports - dlb2->num_dir_ports;
 790         dlb2->num_ldb_queues = dlb2->num_queues - dlb2->num_dir_ports;
 791         dlb2->num_dir_queues = dlb2->num_dir_ports;
 792         if (dlb2->version == DLB2_HW_V2_5) {
 793                 dlb2->credit_pool = rsrcs->num_credits;
 794                 dlb2->max_credits = rsrcs->num_credits;
 795         } else {
 796                 dlb2->ldb_credit_pool = rsrcs->num_ldb_credits;
 797                 dlb2->max_ldb_credits = rsrcs->num_ldb_credits;
 798                 dlb2->dir_credit_pool = rsrcs->num_dir_credits;
 799                 dlb2->max_dir_credits = rsrcs->num_dir_credits;
 800         }
 801
 802         dlb2->configured = true;
 803
 804         return 0;
 805 }
 806
 807 static void
 808 dlb2_eventdev_port_default_conf_get(struct rte_eventdev *dev,
 809                                     uint8_t port_id,
 810                                     struct rte_event_port_conf *port_conf)
 811 {
 812         RTE_SET_USED(port_id);
 813         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
 814
 815         port_conf->new_event_threshold = dlb2->new_event_limit;
 816         port_conf->dequeue_depth = 32;
 817         port_conf->enqueue_depth = DLB2_MAX_ENQUEUE_DEPTH;
 818         port_conf->event_port_cfg = 0;
 819 }
 820
 821 static void
 822 dlb2_eventdev_queue_default_conf_get(struct rte_eventdev *dev,
 823                                      uint8_t queue_id,
 824                                      struct rte_event_queue_conf *queue_conf)
 825 {
 826         RTE_SET_USED(dev);
 827         RTE_SET_USED(queue_id);
 828
 829         queue_conf->nb_atomic_flows = 1024;
 830         queue_conf->nb_atomic_order_sequences = 64;
 831         queue_conf->event_queue_cfg = 0;
 832         queue_conf->priority = 0;
 833 }
 834
 835 static int32_t
 836 dlb2_get_sn_allocation(struct dlb2_eventdev *dlb2, int group)
 837 {
 838         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
 839         struct dlb2_get_sn_allocation_args cfg;
 840         int ret;
 841
 842         cfg.group = group;
 843
 844         ret = dlb2_iface_get_sn_allocation(handle, &cfg);
 845         if (ret < 0) {
 846                 DLB2_LOG_ERR("dlb2: get_sn_allocation ret=%d (driver status: %s)\n",
 847                              ret, dlb2_error_strings[cfg.response.status]);
 848                 return ret;
 849         }
 850
 851         return cfg.response.id;
 852 }
 853
 854 static int
 855 dlb2_set_sn_allocation(struct dlb2_eventdev *dlb2, int group, int num)
 856 {
 857         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
 858         struct dlb2_set_sn_allocation_args cfg;
 859         int ret;
 860
 861         cfg.num = num;
 862         cfg.group = group;
 863
 864         ret = dlb2_iface_set_sn_allocation(handle, &cfg);
 865         if (ret < 0) {
 866                 DLB2_LOG_ERR("dlb2: set_sn_allocation ret=%d (driver status: %s)\n",
 867                              ret, dlb2_error_strings[cfg.response.status]);
 868                 return ret;
 869         }
 870
 871         return ret;
 872 }
 873
 874 static int32_t
 875 dlb2_get_sn_occupancy(struct dlb2_eventdev *dlb2, int group)
 876 {
 877         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
 878         struct dlb2_get_sn_occupancy_args cfg;
 879         int ret;
 880
 881         cfg.group = group;
 882
 883         ret = dlb2_iface_get_sn_occupancy(handle, &cfg);
 884         if (ret < 0) {
 885                 DLB2_LOG_ERR("dlb2: get_sn_occupancy ret=%d (driver status: %s)\n",
 886                              ret, dlb2_error_strings[cfg.response.status]);
 887                 return ret;
 888         }
 889
 890         return cfg.response.id;
 891 }
 892
 893 /* Query the current sequence number allocations and, if they conflict with the
 894  * requested LDB queue configuration, attempt to re-allocate sequence numbers.
 895  * This is best-effort; if it fails, the PMD will attempt to configure the
 896  * load-balanced queue and return an error.
 897  */
 898 static void
 899 dlb2_program_sn_allocation(struct dlb2_eventdev *dlb2,
 900                            const struct rte_event_queue_conf *queue_conf)
 901 {
 902         int grp_occupancy[DLB2_NUM_SN_GROUPS];
 903         int grp_alloc[DLB2_NUM_SN_GROUPS];
 904         int i, sequence_numbers;
 905
 906         sequence_numbers = (int)queue_conf->nb_atomic_order_sequences;
 907
 908         for (i = 0; i < DLB2_NUM_SN_GROUPS; i++) {
 909                 int total_slots;
 910
 911                 grp_alloc[i] = dlb2_get_sn_allocation(dlb2, i);
 912                 if (grp_alloc[i] < 0)
 913                         return;
 914
 915                 total_slots = DLB2_MAX_LDB_SN_ALLOC / grp_alloc[i];
 916
 917                 grp_occupancy[i] = dlb2_get_sn_occupancy(dlb2, i);
 918                 if (grp_occupancy[i] < 0)
 919                         return;
 920
 921                 /* DLB has at least one available slot for the requested
 922                  * sequence numbers, so no further configuration required.
 923                  */
 924                 if (grp_alloc[i] == sequence_numbers &&
 925                     grp_occupancy[i] < total_slots)
 926                         return;
 927         }
 928
 929         /* None of the sequence number groups are configured for the requested
 930          * sequence numbers, so we have to reconfigure one of them. This is
 931          * only possible if a group is not in use.
 932          */
 933         for (i = 0; i < DLB2_NUM_SN_GROUPS; i++) {
 934                 if (grp_occupancy[i] == 0)
 935                         break;
 936         }
 937
 938         if (i == DLB2_NUM_SN_GROUPS) {
 939                 DLB2_LOG_ERR("[%s()] No groups with %d sequence_numbers are available or have free slots\n",
 940                        __func__, sequence_numbers);
 941                 return;
 942         }
 943
 944         /* Attempt to configure slot i with the requested number of sequence
 945          * numbers. Ignore the return value -- if this fails, the error will be
 946          * caught during subsequent queue configuration.
 947          */
 948         dlb2_set_sn_allocation(dlb2, i, sequence_numbers);
 949 }
 950
 951 static int32_t
 952 dlb2_hw_create_ldb_queue(struct dlb2_eventdev *dlb2,
 953                          struct dlb2_eventdev_queue *ev_queue,
 954                          const struct rte_event_queue_conf *evq_conf)
 955 {
 956         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
 957         struct dlb2_queue *queue = &ev_queue->qm_queue;
 958         struct dlb2_create_ldb_queue_args cfg;
 959         int32_t ret;
 960         uint32_t qm_qid;
 961         int sched_type = -1;
 962
 963         if (evq_conf == NULL)
 964                 return -EINVAL;
 965
 966         if (evq_conf->event_queue_cfg & RTE_EVENT_QUEUE_CFG_ALL_TYPES) {
 967                 if (evq_conf->nb_atomic_order_sequences != 0)
 968                         sched_type = RTE_SCHED_TYPE_ORDERED;
 969                 else
 970                         sched_type = RTE_SCHED_TYPE_PARALLEL;
 971         } else
 972                 sched_type = evq_conf->schedule_type;
 973
 974         cfg.num_atomic_inflights = DLB2_NUM_ATOMIC_INFLIGHTS_PER_QUEUE;
 975         cfg.num_sequence_numbers = evq_conf->nb_atomic_order_sequences;
 976         cfg.num_qid_inflights = evq_conf->nb_atomic_order_sequences;
 977
 978         if (sched_type != RTE_SCHED_TYPE_ORDERED) {
 979                 cfg.num_sequence_numbers = 0;
 980                 cfg.num_qid_inflights = 2048;
 981         }
 982
 983         /* App should set this to the number of hardware flows they want, not
 984          * the overall number of flows they're going to use. E.g. if app is
 985          * using 64 flows and sets compression to 64, best-case they'll get
 986          * 64 unique hashed flows in hardware.
 987          */
 988         switch (evq_conf->nb_atomic_flows) {
 989         /* Valid DLB2 compression levels */
 990         case 64:
 991         case 128:
 992         case 256:
 993         case 512:
 994         case (1 * 1024): /* 1K */
 995         case (2 * 1024): /* 2K */
 996         case (4 * 1024): /* 4K */
 997         case (64 * 1024): /* 64K */
 998                 cfg.lock_id_comp_level = evq_conf->nb_atomic_flows;
 999                 break;
1000         default:
1001                 /* Invalid compression level */
1002                 cfg.lock_id_comp_level = 0; /* no compression */
1003         }
1004
1005         if (ev_queue->depth_threshold == 0) {
1006                 cfg.depth_threshold = dlb2->default_depth_thresh;
1007                 ev_queue->depth_threshold =
1008                         dlb2->default_depth_thresh;
1009         } else
1010                 cfg.depth_threshold = ev_queue->depth_threshold;
1011
1012         ret = dlb2_iface_ldb_queue_create(handle, &cfg);
1013         if (ret < 0) {
1014                 DLB2_LOG_ERR("dlb2: create LB event queue error, ret=%d (driver status: %s)\n",
1015                              ret, dlb2_error_strings[cfg.response.status]);
1016                 return -EINVAL;
1017         }
1018
1019         qm_qid = cfg.response.id;
1020
1021         /* Save off queue config for debug, resource lookups, and reconfig */
1022         queue->num_qid_inflights = cfg.num_qid_inflights;
1023         queue->num_atm_inflights = cfg.num_atomic_inflights;
1024
1025         queue->sched_type = sched_type;
1026         queue->config_state = DLB2_CONFIGURED;
1027
1028         DLB2_LOG_DBG("Created LB event queue %d, nb_inflights=%d, nb_seq=%d, qid inflights=%d\n",
1029                      qm_qid,
1030                      cfg.num_atomic_inflights,
1031                      cfg.num_sequence_numbers,
1032                      cfg.num_qid_inflights);
1033
1034         return qm_qid;
1035 }
1036
1037 static int
1038 dlb2_eventdev_ldb_queue_setup(struct rte_eventdev *dev,
1039                               struct dlb2_eventdev_queue *ev_queue,
1040                               const struct rte_event_queue_conf *queue_conf)
1041 {
1042         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
1043         int32_t qm_qid;
1044
1045         if (queue_conf->nb_atomic_order_sequences)
1046                 dlb2_program_sn_allocation(dlb2, queue_conf);
1047
1048         qm_qid = dlb2_hw_create_ldb_queue(dlb2, ev_queue, queue_conf);
1049         if (qm_qid < 0) {
1050                 DLB2_LOG_ERR("Failed to create the load-balanced queue\n");
1051
1052                 return qm_qid;
1053         }
1054
1055         dlb2->qm_ldb_to_ev_queue_id[qm_qid] = ev_queue->id;
1056
1057         ev_queue->qm_queue.id = qm_qid;
1058
1059         return 0;
1060 }
1061
1062 static int dlb2_num_dir_queues_setup(struct dlb2_eventdev *dlb2)
1063 {
1064         int i, num = 0;
1065
1066         for (i = 0; i < dlb2->num_queues; i++) {
1067                 if (dlb2->ev_queues[i].setup_done &&
1068                     dlb2->ev_queues[i].qm_queue.is_directed)
1069                         num++;
1070         }
1071
1072         return num;
1073 }
1074
1075 static void
1076 dlb2_queue_link_teardown(struct dlb2_eventdev *dlb2,
1077                          struct dlb2_eventdev_queue *ev_queue)
1078 {
1079         struct dlb2_eventdev_port *ev_port;
1080         int i, j;
1081
1082         for (i = 0; i < dlb2->num_ports; i++) {
1083                 ev_port = &dlb2->ev_ports[i];
1084
1085                 for (j = 0; j < DLB2_MAX_NUM_QIDS_PER_LDB_CQ; j++) {
1086                         if (!ev_port->link[j].valid ||
1087                             ev_port->link[j].queue_id != ev_queue->id)
1088                                 continue;
1089
1090                         ev_port->link[j].valid = false;
1091                         ev_port->num_links--;
1092                 }
1093         }
1094
1095         ev_queue->num_links = 0;
1096 }
1097
1098 static int
1099 dlb2_eventdev_queue_setup(struct rte_eventdev *dev,
1100                           uint8_t ev_qid,
1101                           const struct rte_event_queue_conf *queue_conf)
1102 {
1103         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
1104         struct dlb2_eventdev_queue *ev_queue;
1105         int ret;
1106
1107         if (queue_conf == NULL)
1108                 return -EINVAL;
1109
1110         if (ev_qid >= dlb2->num_queues)
1111                 return -EINVAL;
1112
1113         ev_queue = &dlb2->ev_queues[ev_qid];
1114
1115         ev_queue->qm_queue.is_directed = queue_conf->event_queue_cfg &
1116                 RTE_EVENT_QUEUE_CFG_SINGLE_LINK;
1117         ev_queue->id = ev_qid;
1118         ev_queue->conf = *queue_conf;
1119
1120         if (!ev_queue->qm_queue.is_directed) {
1121                 ret = dlb2_eventdev_ldb_queue_setup(dev, ev_queue, queue_conf);
1122         } else {
1123                 /* The directed queue isn't setup until link time, at which
1124                  * point we know its directed port ID. Directed queue setup
1125                  * will only fail if this queue is already setup or there are
1126                  * no directed queues left to configure.
1127                  */
1128                 ret = 0;
1129
1130                 ev_queue->qm_queue.config_state = DLB2_NOT_CONFIGURED;
1131
1132                 if (ev_queue->setup_done ||
1133                     dlb2_num_dir_queues_setup(dlb2) == dlb2->num_dir_queues)
1134                         ret = -EINVAL;
1135         }
1136
1137         /* Tear down pre-existing port->queue links */
1138         if (!ret && dlb2->run_state == DLB2_RUN_STATE_STOPPED)
1139                 dlb2_queue_link_teardown(dlb2, ev_queue);
1140
1141         if (!ret)
1142                 ev_queue->setup_done = true;
1143
1144         return ret;
1145 }
1146
1147 static int
1148 dlb2_init_consume_qe(struct dlb2_port *qm_port, char *mz_name)
1149 {
1150         struct dlb2_cq_pop_qe *qe;
1151
1152         qe = rte_zmalloc(mz_name,
1153                         DLB2_NUM_QES_PER_CACHE_LINE *
1154                                 sizeof(struct dlb2_cq_pop_qe),
1155                         RTE_CACHE_LINE_SIZE);
1156
1157         if (qe == NULL) {
1158                 DLB2_LOG_ERR("dlb2: no memory for consume_qe\n");
1159                 return -ENOMEM;
1160         }
1161         qm_port->consume_qe = qe;
1162
1163         qe->qe_valid = 0;
1164         qe->qe_frag = 0;
1165         qe->qe_comp = 0;
1166         qe->cq_token = 1;
1167         /* Tokens value is 0-based; i.e. '0' returns 1 token, '1' returns 2,
1168          * and so on.
1169          */
1170         qe->tokens = 0; /* set at run time */
1171         qe->meas_lat = 0;
1172         qe->no_dec = 0;
1173         /* Completion IDs are disabled */
1174         qe->cmp_id = 0;
1175
1176         return 0;
1177 }
1178
1179 static int
1180 dlb2_init_int_arm_qe(struct dlb2_port *qm_port, char *mz_name)
1181 {
1182         struct dlb2_enqueue_qe *qe;
1183
1184         qe = rte_zmalloc(mz_name,
1185                         DLB2_NUM_QES_PER_CACHE_LINE *
1186                                 sizeof(struct dlb2_enqueue_qe),
1187                         RTE_CACHE_LINE_SIZE);
1188
1189         if (qe == NULL) {
1190                 DLB2_LOG_ERR("dlb2: no memory for complete_qe\n");
1191                 return -ENOMEM;
1192         }
1193         qm_port->int_arm_qe = qe;
1194
1195         /* V2 - INT ARM is CQ_TOKEN + FRAG */
1196         qe->qe_valid = 0;
1197         qe->qe_frag = 1;
1198         qe->qe_comp = 0;
1199         qe->cq_token = 1;
1200         qe->meas_lat = 0;
1201         qe->no_dec = 0;
1202         /* Completion IDs are disabled */
1203         qe->cmp_id = 0;
1204
1205         return 0;
1206 }
1207
1208 static int
1209 dlb2_init_qe_mem(struct dlb2_port *qm_port, char *mz_name)
1210 {
1211         int ret, sz;
1212
1213         sz = DLB2_NUM_QES_PER_CACHE_LINE * sizeof(struct dlb2_enqueue_qe);
1214
1215         qm_port->qe4 = rte_zmalloc(mz_name, sz, RTE_CACHE_LINE_SIZE);
1216
1217         if (qm_port->qe4 == NULL) {
1218                 DLB2_LOG_ERR("dlb2: no qe4 memory\n");
1219                 ret = -ENOMEM;
1220                 goto error_exit;
1221         }
1222
1223         ret = dlb2_init_int_arm_qe(qm_port, mz_name);
1224         if (ret < 0) {
1225                 DLB2_LOG_ERR("dlb2: dlb2_init_int_arm_qe ret=%d\n", ret);
1226                 goto error_exit;
1227         }
1228
1229         ret = dlb2_init_consume_qe(qm_port, mz_name);
1230         if (ret < 0) {
1231                 DLB2_LOG_ERR("dlb2: dlb2_init_consume_qe ret=%d\n", ret);
1232                 goto error_exit;
1233         }
1234
1235         return 0;
1236
1237 error_exit:
1238
1239         dlb2_free_qe_mem(qm_port);
1240
1241         return ret;
1242 }
1243
1244 static inline uint16_t
1245 dlb2_event_enqueue_delayed(void *event_port,
1246                            const struct rte_event events[]);
1247
1248 static inline uint16_t
1249 dlb2_event_enqueue_burst_delayed(void *event_port,
1250                                  const struct rte_event events[],
1251                                  uint16_t num);
1252
1253 static inline uint16_t
1254 dlb2_event_enqueue_new_burst_delayed(void *event_port,
1255                                      const struct rte_event events[],
1256                                      uint16_t num);
1257
1258 static inline uint16_t
1259 dlb2_event_enqueue_forward_burst_delayed(void *event_port,
1260                                          const struct rte_event events[],
1261                                          uint16_t num);
1262
1263 /* Generate the required bitmask for rotate-style expected QE gen bits.
1264  * This requires a pattern of 1's and zeros, starting with expected as
1265  * 1 bits, so when hardware writes 0's they're "new". This requires the
1266  * ring size to be powers of 2 to wrap correctly.
1267  */
1268 static void
1269 dlb2_hw_cq_bitmask_init(struct dlb2_port *qm_port, uint32_t cq_depth)
1270 {
1271         uint64_t cq_build_mask = 0;
1272         uint32_t i;
1273
1274         if (cq_depth > 64)
1275                 return; /* need to fall back to scalar code */
1276
1277         /*
1278          * all 1's in first u64, all zeros in second is correct bit pattern to
1279          * start. Special casing == 64 easier than adapting complex loop logic.
1280          */
1281         if (cq_depth == 64) {
1282                 qm_port->cq_rolling_mask = 0;
1283                 qm_port->cq_rolling_mask_2 = -1;
1284                 return;
1285         }
1286
1287         for (i = 0; i < 64; i += (cq_depth * 2))
1288                 cq_build_mask |= ((1ULL << cq_depth) - 1) << (i + cq_depth);
1289
1290         qm_port->cq_rolling_mask = cq_build_mask;
1291         qm_port->cq_rolling_mask_2 = cq_build_mask;
1292 }
1293
1294 static int
1295 dlb2_hw_create_ldb_port(struct dlb2_eventdev *dlb2,
1296                         struct dlb2_eventdev_port *ev_port,
1297                         uint32_t dequeue_depth,
1298                         uint32_t enqueue_depth)
1299 {
1300         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
1301         struct dlb2_create_ldb_port_args cfg = { {0} };
1302         int ret;
1303         struct dlb2_port *qm_port = NULL;
1304         char mz_name[RTE_MEMZONE_NAMESIZE];
1305         uint32_t qm_port_id;
1306         uint16_t ldb_credit_high_watermark = 0;
1307         uint16_t dir_credit_high_watermark = 0;
1308         uint16_t credit_high_watermark = 0;
1309
1310         if (handle == NULL)
1311                 return -EINVAL;
1312
1313         if (dequeue_depth < DLB2_MIN_CQ_DEPTH) {
1314                 DLB2_LOG_ERR("dlb2: invalid enqueue_depth, must be at least %d\n",
1315                              DLB2_MIN_CQ_DEPTH);
1316                 return -EINVAL;
1317         }
1318
1319         if (enqueue_depth < DLB2_MIN_ENQUEUE_DEPTH) {
1320                 DLB2_LOG_ERR("dlb2: invalid enqueue_depth, must be at least %d\n",
1321                              DLB2_MIN_ENQUEUE_DEPTH);
1322                 return -EINVAL;
1323         }
1324
1325         rte_spinlock_lock(&handle->resource_lock);
1326
1327         /* We round up to the next power of 2 if necessary */
1328         cfg.cq_depth = rte_align32pow2(dequeue_depth);
1329         cfg.cq_depth_threshold = 1;
1330
1331         cfg.cq_history_list_size = DLB2_NUM_HIST_LIST_ENTRIES_PER_LDB_PORT;
1332
1333         if (handle->cos_id == DLB2_COS_DEFAULT)
1334                 cfg.cos_id = 0;
1335         else
1336                 cfg.cos_id = handle->cos_id;
1337
1338         cfg.cos_strict = 0;
1339
1340         /* User controls the LDB high watermark via enqueue depth. The DIR high
1341          * watermark is equal, unless the directed credit pool is too small.
1342          */
1343         if (dlb2->version == DLB2_HW_V2) {
1344                 ldb_credit_high_watermark = enqueue_depth;
1345                 /* If there are no directed ports, the kernel driver will
1346                  * ignore this port's directed credit settings. Don't use
1347                  * enqueue_depth if it would require more directed credits
1348                  * than are available.
1349                  */
1350                 dir_credit_high_watermark =
1351                         RTE_MIN(enqueue_depth,
1352                                 handle->cfg.num_dir_credits / dlb2->num_ports);
1353         } else
1354                 credit_high_watermark = enqueue_depth;
1355
1356         /* Per QM values */
1357
1358         ret = dlb2_iface_ldb_port_create(handle, &cfg,  dlb2->poll_mode);
1359         if (ret < 0) {
1360                 DLB2_LOG_ERR("dlb2: dlb2_ldb_port_create error, ret=%d (driver status: %s)\n",
1361                              ret, dlb2_error_strings[cfg.response.status]);
1362                 goto error_exit;
1363         }
1364
1365         qm_port_id = cfg.response.id;
1366
1367         DLB2_LOG_DBG("dlb2: ev_port %d uses qm LB port %d <<<<<\n",
1368                      ev_port->id, qm_port_id);
1369
1370         qm_port = &ev_port->qm_port;
1371         qm_port->ev_port = ev_port; /* back ptr */
1372         qm_port->dlb2 = dlb2; /* back ptr */
1373         /*
1374          * Allocate and init local qe struct(s).
1375          * Note: MOVDIR64 requires the enqueue QE (qe4) to be aligned.
1376          */
1377
1378         snprintf(mz_name, sizeof(mz_name), "dlb2_ldb_port%d",
1379                  ev_port->id);
1380
1381         ret = dlb2_init_qe_mem(qm_port, mz_name);
1382         if (ret < 0) {
1383                 DLB2_LOG_ERR("dlb2: init_qe_mem failed, ret=%d\n", ret);
1384                 goto error_exit;
1385         }
1386
1387         qm_port->id = qm_port_id;
1388
1389         if (dlb2->version == DLB2_HW_V2) {
1390                 qm_port->cached_ldb_credits = 0;
1391                 qm_port->cached_dir_credits = 0;
1392         } else
1393                 qm_port->cached_credits = 0;
1394
1395         /* CQs with depth < 8 use an 8-entry queue, but withhold credits so
1396          * the effective depth is smaller.
1397          */
1398         qm_port->cq_depth = cfg.cq_depth <= 8 ? 8 : cfg.cq_depth;
1399         qm_port->cq_idx = 0;
1400         qm_port->cq_idx_unmasked = 0;
1401
1402         if (dlb2->poll_mode == DLB2_CQ_POLL_MODE_SPARSE)
1403                 qm_port->cq_depth_mask = (qm_port->cq_depth * 4) - 1;
1404         else
1405                 qm_port->cq_depth_mask = qm_port->cq_depth - 1;
1406
1407         qm_port->gen_bit_shift = __builtin_popcount(qm_port->cq_depth_mask);
1408         /* starting value of gen bit - it toggles at wrap time */
1409         qm_port->gen_bit = 1;
1410
1411         dlb2_hw_cq_bitmask_init(qm_port, qm_port->cq_depth);
1412
1413         qm_port->int_armed = false;
1414
1415         /* Save off for later use in info and lookup APIs. */
1416         qm_port->qid_mappings = &dlb2->qm_ldb_to_ev_queue_id[0];
1417
1418         qm_port->dequeue_depth = dequeue_depth;
1419         qm_port->token_pop_thresh = dequeue_depth;
1420
1421         /* The default enqueue functions do not include delayed-pop support for
1422          * performance reasons.
1423          */
1424         if (qm_port->token_pop_mode == DELAYED_POP) {
1425                 dlb2->event_dev->enqueue = dlb2_event_enqueue_delayed;
1426                 dlb2->event_dev->enqueue_burst =
1427                         dlb2_event_enqueue_burst_delayed;
1428                 dlb2->event_dev->enqueue_new_burst =
1429                         dlb2_event_enqueue_new_burst_delayed;
1430                 dlb2->event_dev->enqueue_forward_burst =
1431                         dlb2_event_enqueue_forward_burst_delayed;
1432         }
1433
1434         qm_port->owed_tokens = 0;
1435         qm_port->issued_releases = 0;
1436
1437         /* Save config message too. */
1438         rte_memcpy(&qm_port->cfg.ldb, &cfg, sizeof(qm_port->cfg.ldb));
1439
1440         /* update state */
1441         qm_port->state = PORT_STARTED; /* enabled at create time */
1442         qm_port->config_state = DLB2_CONFIGURED;
1443
1444         if (dlb2->version == DLB2_HW_V2) {
1445                 qm_port->dir_credits = dir_credit_high_watermark;
1446                 qm_port->ldb_credits = ldb_credit_high_watermark;
1447                 qm_port->credit_pool[DLB2_DIR_QUEUE] = &dlb2->dir_credit_pool;
1448                 qm_port->credit_pool[DLB2_LDB_QUEUE] = &dlb2->ldb_credit_pool;
1449
1450                 DLB2_LOG_DBG("dlb2: created ldb port %d, depth = %d, ldb credits=%d, dir credits=%d\n",
1451                              qm_port_id,
1452                              dequeue_depth,
1453                              qm_port->ldb_credits,
1454                              qm_port->dir_credits);
1455         } else {
1456                 qm_port->credits = credit_high_watermark;
1457                 qm_port->credit_pool[DLB2_COMBINED_POOL] = &dlb2->credit_pool;
1458
1459                 DLB2_LOG_DBG("dlb2: created ldb port %d, depth = %d, credits=%d\n",
1460                              qm_port_id,
1461                              dequeue_depth,
1462                              qm_port->credits);
1463         }
1464
1465         qm_port->use_scalar = false;
1466
1467 #if (!defined RTE_ARCH_X86_64)
1468         qm_port->use_scalar = true;
1469 #else
1470         if ((qm_port->cq_depth > 64) ||
1471             (!rte_is_power_of_2(qm_port->cq_depth)) ||
1472             (dlb2->vector_opts_disabled == true))
1473                 qm_port->use_scalar = true;
1474 #endif
1475
1476         rte_spinlock_unlock(&handle->resource_lock);
1477
1478         return 0;
1479
1480 error_exit:
1481
1482         if (qm_port)
1483                 dlb2_free_qe_mem(qm_port);
1484
1485         rte_spinlock_unlock(&handle->resource_lock);
1486
1487         DLB2_LOG_ERR("dlb2: create ldb port failed!\n");
1488
1489         return ret;
1490 }
1491
1492 static void
1493 dlb2_port_link_teardown(struct dlb2_eventdev *dlb2,
1494                         struct dlb2_eventdev_port *ev_port)
1495 {
1496         struct dlb2_eventdev_queue *ev_queue;
1497         int i;
1498
1499         for (i = 0; i < DLB2_MAX_NUM_QIDS_PER_LDB_CQ; i++) {
1500                 if (!ev_port->link[i].valid)
1501                         continue;
1502
1503                 ev_queue = &dlb2->ev_queues[ev_port->link[i].queue_id];
1504
1505                 ev_port->link[i].valid = false;
1506                 ev_port->num_links--;
1507                 ev_queue->num_links--;
1508         }
1509 }
1510
1511 static int
1512 dlb2_hw_create_dir_port(struct dlb2_eventdev *dlb2,
1513                         struct dlb2_eventdev_port *ev_port,
1514                         uint32_t dequeue_depth,
1515                         uint32_t enqueue_depth)
1516 {
1517         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
1518         struct dlb2_create_dir_port_args cfg = { {0} };
1519         int ret;
1520         struct dlb2_port *qm_port = NULL;
1521         char mz_name[RTE_MEMZONE_NAMESIZE];
1522         uint32_t qm_port_id;
1523         uint16_t ldb_credit_high_watermark = 0;
1524         uint16_t dir_credit_high_watermark = 0;
1525         uint16_t credit_high_watermark = 0;
1526
1527         if (dlb2 == NULL || handle == NULL)
1528                 return -EINVAL;
1529
1530         if (dequeue_depth < DLB2_MIN_CQ_DEPTH) {
1531                 DLB2_LOG_ERR("dlb2: invalid dequeue_depth, must be %d-%d\n",
1532                              DLB2_MIN_CQ_DEPTH, DLB2_MAX_INPUT_QUEUE_DEPTH);
1533                 return -EINVAL;
1534         }
1535
1536         if (enqueue_depth < DLB2_MIN_ENQUEUE_DEPTH) {
1537                 DLB2_LOG_ERR("dlb2: invalid enqueue_depth, must be at least %d\n",
1538                              DLB2_MIN_ENQUEUE_DEPTH);
1539                 return -EINVAL;
1540         }
1541
1542         rte_spinlock_lock(&handle->resource_lock);
1543
1544         /* Directed queues are configured at link time. */
1545         cfg.queue_id = -1;
1546
1547         /* We round up to the next power of 2 if necessary */
1548         cfg.cq_depth = rte_align32pow2(dequeue_depth);
1549         cfg.cq_depth_threshold = 1;
1550
1551         /* User controls the LDB high watermark via enqueue depth. The DIR high
1552          * watermark is equal, unless the directed credit pool is too small.
1553          */
1554         if (dlb2->version == DLB2_HW_V2) {
1555                 ldb_credit_high_watermark = enqueue_depth;
1556                 /* Don't use enqueue_depth if it would require more directed
1557                  * credits than are available.
1558                  */
1559                 dir_credit_high_watermark =
1560                         RTE_MIN(enqueue_depth,
1561                                 handle->cfg.num_dir_credits / dlb2->num_ports);
1562         } else
1563                 credit_high_watermark = enqueue_depth;
1564
1565         /* Per QM values */
1566
1567         ret = dlb2_iface_dir_port_create(handle, &cfg,  dlb2->poll_mode);
1568         if (ret < 0) {
1569                 DLB2_LOG_ERR("dlb2: dlb2_dir_port_create error, ret=%d (driver status: %s)\n",
1570                              ret, dlb2_error_strings[cfg.response.status]);
1571                 goto error_exit;
1572         }
1573
1574         qm_port_id = cfg.response.id;
1575
1576         DLB2_LOG_DBG("dlb2: ev_port %d uses qm DIR port %d <<<<<\n",
1577                      ev_port->id, qm_port_id);
1578
1579         qm_port = &ev_port->qm_port;
1580         qm_port->ev_port = ev_port; /* back ptr */
1581         qm_port->dlb2 = dlb2;  /* back ptr */
1582
1583         /*
1584          * Init local qe struct(s).
1585          * Note: MOVDIR64 requires the enqueue QE to be aligned
1586          */
1587
1588         snprintf(mz_name, sizeof(mz_name), "dlb2_dir_port%d",
1589                  ev_port->id);
1590
1591         ret = dlb2_init_qe_mem(qm_port, mz_name);
1592
1593         if (ret < 0) {
1594                 DLB2_LOG_ERR("dlb2: init_qe_mem failed, ret=%d\n", ret);
1595                 goto error_exit;
1596         }
1597
1598         qm_port->id = qm_port_id;
1599
1600         if (dlb2->version == DLB2_HW_V2) {
1601                 qm_port->cached_ldb_credits = 0;
1602                 qm_port->cached_dir_credits = 0;
1603         } else
1604                 qm_port->cached_credits = 0;
1605
1606         /* CQs with depth < 8 use an 8-entry queue, but withhold credits so
1607          * the effective depth is smaller.
1608          */
1609         qm_port->cq_depth = cfg.cq_depth <= 8 ? 8 : cfg.cq_depth;
1610         qm_port->cq_idx = 0;
1611         qm_port->cq_idx_unmasked = 0;
1612
1613         if (dlb2->poll_mode == DLB2_CQ_POLL_MODE_SPARSE)
1614                 qm_port->cq_depth_mask = (cfg.cq_depth * 4) - 1;
1615         else
1616                 qm_port->cq_depth_mask = cfg.cq_depth - 1;
1617
1618         qm_port->gen_bit_shift = __builtin_popcount(qm_port->cq_depth_mask);
1619         /* starting value of gen bit - it toggles at wrap time */
1620         qm_port->gen_bit = 1;
1621         dlb2_hw_cq_bitmask_init(qm_port, qm_port->cq_depth);
1622
1623         qm_port->int_armed = false;
1624
1625         /* Save off for later use in info and lookup APIs. */
1626         qm_port->qid_mappings = &dlb2->qm_dir_to_ev_queue_id[0];
1627
1628         qm_port->dequeue_depth = dequeue_depth;
1629
1630         /* Directed ports are auto-pop, by default. */
1631         qm_port->token_pop_mode = AUTO_POP;
1632         qm_port->owed_tokens = 0;
1633         qm_port->issued_releases = 0;
1634
1635         /* Save config message too. */
1636         rte_memcpy(&qm_port->cfg.dir, &cfg, sizeof(qm_port->cfg.dir));
1637
1638         /* update state */
1639         qm_port->state = PORT_STARTED; /* enabled at create time */
1640         qm_port->config_state = DLB2_CONFIGURED;
1641
1642         if (dlb2->version == DLB2_HW_V2) {
1643                 qm_port->dir_credits = dir_credit_high_watermark;
1644                 qm_port->ldb_credits = ldb_credit_high_watermark;
1645                 qm_port->credit_pool[DLB2_DIR_QUEUE] = &dlb2->dir_credit_pool;
1646                 qm_port->credit_pool[DLB2_LDB_QUEUE] = &dlb2->ldb_credit_pool;
1647
1648                 DLB2_LOG_DBG("dlb2: created dir port %d, depth = %d cr=%d,%d\n",
1649                              qm_port_id,
1650                              dequeue_depth,
1651                              dir_credit_high_watermark,
1652                              ldb_credit_high_watermark);
1653         } else {
1654                 qm_port->credits = credit_high_watermark;
1655                 qm_port->credit_pool[DLB2_COMBINED_POOL] = &dlb2->credit_pool;
1656
1657                 DLB2_LOG_DBG("dlb2: created dir port %d, depth = %d cr=%d\n",
1658                              qm_port_id,
1659                              dequeue_depth,
1660                              credit_high_watermark);
1661         }
1662
1663 #if (!defined RTE_ARCH_X86_64)
1664         qm_port->use_scalar = true;
1665 #else
1666         if ((qm_port->cq_depth > 64) ||
1667             (!rte_is_power_of_2(qm_port->cq_depth)) ||
1668             (dlb2->vector_opts_disabled == true))
1669                 qm_port->use_scalar = true;
1670 #endif
1671
1672         rte_spinlock_unlock(&handle->resource_lock);
1673
1674         return 0;
1675
1676 error_exit:
1677
1678         if (qm_port)
1679                 dlb2_free_qe_mem(qm_port);
1680
1681         rte_spinlock_unlock(&handle->resource_lock);
1682
1683         DLB2_LOG_ERR("dlb2: create dir port failed!\n");
1684
1685         return ret;
1686 }
1687
1688 static int
1689 dlb2_eventdev_port_setup(struct rte_eventdev *dev,
1690                          uint8_t ev_port_id,
1691                          const struct rte_event_port_conf *port_conf)
1692 {
1693         struct dlb2_eventdev *dlb2;
1694         struct dlb2_eventdev_port *ev_port;
1695         int ret;
1696
1697         if (dev == NULL || port_conf == NULL) {
1698                 DLB2_LOG_ERR("Null parameter\n");
1699                 return -EINVAL;
1700         }
1701
1702         dlb2 = dlb2_pmd_priv(dev);
1703
1704         if (ev_port_id >= DLB2_MAX_NUM_PORTS(dlb2->version))
1705                 return -EINVAL;
1706
1707         if (port_conf->dequeue_depth >
1708                 evdev_dlb2_default_info.max_event_port_dequeue_depth ||
1709             port_conf->enqueue_depth >
1710                 evdev_dlb2_default_info.max_event_port_enqueue_depth)
1711                 return -EINVAL;
1712
1713         ev_port = &dlb2->ev_ports[ev_port_id];
1714         /* configured? */
1715         if (ev_port->setup_done) {
1716                 DLB2_LOG_ERR("evport %d is already configured\n", ev_port_id);
1717                 return -EINVAL;
1718         }
1719
1720         ev_port->qm_port.is_directed = port_conf->event_port_cfg &
1721                 RTE_EVENT_PORT_CFG_SINGLE_LINK;
1722
1723         if (!ev_port->qm_port.is_directed) {
1724                 ret = dlb2_hw_create_ldb_port(dlb2,
1725                                               ev_port,
1726                                               port_conf->dequeue_depth,
1727                                               port_conf->enqueue_depth);
1728                 if (ret < 0) {
1729                         DLB2_LOG_ERR("Failed to create the lB port ve portId=%d\n",
1730                                      ev_port_id);
1731
1732                         return ret;
1733                 }
1734         } else {
1735                 ret = dlb2_hw_create_dir_port(dlb2,
1736                                               ev_port,
1737                                               port_conf->dequeue_depth,
1738                                               port_conf->enqueue_depth);
1739                 if (ret < 0) {
1740                         DLB2_LOG_ERR("Failed to create the DIR port\n");
1741                         return ret;
1742                 }
1743         }
1744
1745         /* Save off port config for reconfig */
1746         ev_port->conf = *port_conf;
1747
1748         ev_port->id = ev_port_id;
1749         ev_port->enq_configured = true;
1750         ev_port->setup_done = true;
1751         ev_port->inflight_max = port_conf->new_event_threshold;
1752         ev_port->implicit_release = !(port_conf->event_port_cfg &
1753                   RTE_EVENT_PORT_CFG_DISABLE_IMPL_REL);
1754         ev_port->outstanding_releases = 0;
1755         ev_port->inflight_credits = 0;
1756         ev_port->credit_update_quanta = dlb2->sw_credit_quanta;
1757         ev_port->dlb2 = dlb2; /* reverse link */
1758
1759         /* Tear down pre-existing port->queue links */
1760         if (dlb2->run_state == DLB2_RUN_STATE_STOPPED)
1761                 dlb2_port_link_teardown(dlb2, &dlb2->ev_ports[ev_port_id]);
1762
1763         dev->data->ports[ev_port_id] = &dlb2->ev_ports[ev_port_id];
1764
1765         return 0;
1766 }
1767
1768 static int16_t
1769 dlb2_hw_map_ldb_qid_to_port(struct dlb2_hw_dev *handle,
1770                             uint32_t qm_port_id,
1771                             uint16_t qm_qid,
1772                             uint8_t priority)
1773 {
1774         struct dlb2_map_qid_args cfg;
1775         int32_t ret;
1776
1777         if (handle == NULL)
1778                 return -EINVAL;
1779
1780         /* Build message */
1781         cfg.port_id = qm_port_id;
1782         cfg.qid = qm_qid;
1783         cfg.priority = EV_TO_DLB2_PRIO(priority);
1784
1785         ret = dlb2_iface_map_qid(handle, &cfg);
1786         if (ret < 0) {
1787                 DLB2_LOG_ERR("dlb2: map qid error, ret=%d (driver status: %s)\n",
1788                              ret, dlb2_error_strings[cfg.response.status]);
1789                 DLB2_LOG_ERR("dlb2: grp=%d, qm_port=%d, qm_qid=%d prio=%d\n",
1790                              handle->domain_id, cfg.port_id,
1791                              cfg.qid,
1792                              cfg.priority);
1793         } else {
1794                 DLB2_LOG_DBG("dlb2: mapped queue %d to qm_port %d\n",
1795                              qm_qid, qm_port_id);
1796         }
1797
1798         return ret;
1799 }
1800
1801 static int
1802 dlb2_event_queue_join_ldb(struct dlb2_eventdev *dlb2,
1803                           struct dlb2_eventdev_port *ev_port,
1804                           struct dlb2_eventdev_queue *ev_queue,
1805                           uint8_t priority)
1806 {
1807         int first_avail = -1;
1808         int ret, i;
1809
1810         for (i = 0; i < DLB2_MAX_NUM_QIDS_PER_LDB_CQ; i++) {
1811                 if (ev_port->link[i].valid) {
1812                         if (ev_port->link[i].queue_id == ev_queue->id &&
1813                             ev_port->link[i].priority == priority) {
1814                                 if (ev_port->link[i].mapped)
1815                                         return 0; /* already mapped */
1816                                 first_avail = i;
1817                         }
1818                 } else if (first_avail == -1)
1819                         first_avail = i;
1820         }
1821         if (first_avail == -1) {
1822                 DLB2_LOG_ERR("dlb2: qm_port %d has no available QID slots.\n",
1823                              ev_port->qm_port.id);
1824                 return -EINVAL;
1825         }
1826
1827         ret = dlb2_hw_map_ldb_qid_to_port(&dlb2->qm_instance,
1828                                           ev_port->qm_port.id,
1829                                           ev_queue->qm_queue.id,
1830                                           priority);
1831
1832         if (!ret)
1833                 ev_port->link[first_avail].mapped = true;
1834
1835         return ret;
1836 }
1837
1838 static int32_t
1839 dlb2_hw_create_dir_queue(struct dlb2_eventdev *dlb2,
1840                          struct dlb2_eventdev_queue *ev_queue,
1841                          int32_t qm_port_id)
1842 {
1843         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
1844         struct dlb2_create_dir_queue_args cfg;
1845         int32_t ret;
1846
1847         /* The directed port is always configured before its queue */
1848         cfg.port_id = qm_port_id;
1849
1850         if (ev_queue->depth_threshold == 0) {
1851                 cfg.depth_threshold = dlb2->default_depth_thresh;
1852                 ev_queue->depth_threshold =
1853                         dlb2->default_depth_thresh;
1854         } else
1855                 cfg.depth_threshold = ev_queue->depth_threshold;
1856
1857         ret = dlb2_iface_dir_queue_create(handle, &cfg);
1858         if (ret < 0) {
1859                 DLB2_LOG_ERR("dlb2: create DIR event queue error, ret=%d (driver status: %s)\n",
1860                              ret, dlb2_error_strings[cfg.response.status]);
1861                 return -EINVAL;
1862         }
1863
1864         return cfg.response.id;
1865 }
1866
1867 static int
1868 dlb2_eventdev_dir_queue_setup(struct dlb2_eventdev *dlb2,
1869                               struct dlb2_eventdev_queue *ev_queue,
1870                               struct dlb2_eventdev_port *ev_port)
1871 {
1872         int32_t qm_qid;
1873
1874         qm_qid = dlb2_hw_create_dir_queue(dlb2, ev_queue, ev_port->qm_port.id);
1875
1876         if (qm_qid < 0) {
1877                 DLB2_LOG_ERR("Failed to create the DIR queue\n");
1878                 return qm_qid;
1879         }
1880
1881         dlb2->qm_dir_to_ev_queue_id[qm_qid] = ev_queue->id;
1882
1883         ev_queue->qm_queue.id = qm_qid;
1884
1885         return 0;
1886 }
1887
1888 static int
1889 dlb2_do_port_link(struct rte_eventdev *dev,
1890                   struct dlb2_eventdev_queue *ev_queue,
1891                   struct dlb2_eventdev_port *ev_port,
1892                   uint8_t prio)
1893 {
1894         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
1895         int err;
1896
1897         /* Don't link until start time. */
1898         if (dlb2->run_state == DLB2_RUN_STATE_STOPPED)
1899                 return 0;
1900
1901         if (ev_queue->qm_queue.is_directed)
1902                 err = dlb2_eventdev_dir_queue_setup(dlb2, ev_queue, ev_port);
1903         else
1904                 err = dlb2_event_queue_join_ldb(dlb2, ev_port, ev_queue, prio);
1905
1906         if (err) {
1907                 DLB2_LOG_ERR("port link failure for %s ev_q %d, ev_port %d\n",
1908                              ev_queue->qm_queue.is_directed ? "DIR" : "LDB",
1909                              ev_queue->id, ev_port->id);
1910
1911                 rte_errno = err;
1912                 return -1;
1913         }
1914
1915         return 0;
1916 }
1917
1918 static int
1919 dlb2_validate_port_link(struct dlb2_eventdev_port *ev_port,
1920                         uint8_t queue_id,
1921                         bool link_exists,
1922                         int index)
1923 {
1924         struct dlb2_eventdev *dlb2 = ev_port->dlb2;
1925         struct dlb2_eventdev_queue *ev_queue;
1926         bool port_is_dir, queue_is_dir;
1927
1928         if (queue_id > dlb2->num_queues) {
1929                 rte_errno = -EINVAL;
1930                 return -1;
1931         }
1932
1933         ev_queue = &dlb2->ev_queues[queue_id];
1934
1935         if (!ev_queue->setup_done &&
1936             ev_queue->qm_queue.config_state != DLB2_PREV_CONFIGURED) {
1937                 rte_errno = -EINVAL;
1938                 return -1;
1939         }
1940
1941         port_is_dir = ev_port->qm_port.is_directed;
1942         queue_is_dir = ev_queue->qm_queue.is_directed;
1943
1944         if (port_is_dir != queue_is_dir) {
1945                 DLB2_LOG_ERR("%s queue %u can't link to %s port %u\n",
1946                              queue_is_dir ? "DIR" : "LDB", ev_queue->id,
1947                              port_is_dir ? "DIR" : "LDB", ev_port->id);
1948
1949                 rte_errno = -EINVAL;
1950                 return -1;
1951         }
1952
1953         /* Check if there is space for the requested link */
1954         if (!link_exists && index == -1) {
1955                 DLB2_LOG_ERR("no space for new link\n");
1956                 rte_errno = -ENOSPC;
1957                 return -1;
1958         }
1959
1960         /* Check if the directed port is already linked */
1961         if (ev_port->qm_port.is_directed && ev_port->num_links > 0 &&
1962             !link_exists) {
1963                 DLB2_LOG_ERR("Can't link DIR port %d to >1 queues\n",
1964                              ev_port->id);
1965                 rte_errno = -EINVAL;
1966                 return -1;
1967         }
1968
1969         /* Check if the directed queue is already linked */
1970         if (ev_queue->qm_queue.is_directed && ev_queue->num_links > 0 &&
1971             !link_exists) {
1972                 DLB2_LOG_ERR("Can't link DIR queue %d to >1 ports\n",
1973                              ev_queue->id);
1974                 rte_errno = -EINVAL;
1975                 return -1;
1976         }
1977
1978         return 0;
1979 }
1980
1981 static int
1982 dlb2_eventdev_port_link(struct rte_eventdev *dev, void *event_port,
1983                         const uint8_t queues[], const uint8_t priorities[],
1984                         uint16_t nb_links)
1985
1986 {
1987         struct dlb2_eventdev_port *ev_port = event_port;
1988         struct dlb2_eventdev *dlb2;
1989         int i, j;
1990
1991         RTE_SET_USED(dev);
1992
1993         if (ev_port == NULL) {
1994                 DLB2_LOG_ERR("dlb2: evport not setup\n");
1995                 rte_errno = -EINVAL;
1996                 return 0;
1997         }
1998
1999         if (!ev_port->setup_done &&
2000             ev_port->qm_port.config_state != DLB2_PREV_CONFIGURED) {
2001                 DLB2_LOG_ERR("dlb2: evport not setup\n");
2002                 rte_errno = -EINVAL;
2003                 return 0;
2004         }
2005
2006         /* Note: rte_event_port_link() ensures the PMD won't receive a NULL
2007          * queues pointer.
2008          */
2009         if (nb_links == 0) {
2010                 DLB2_LOG_DBG("dlb2: nb_links is 0\n");
2011                 return 0; /* Ignore and return success */
2012         }
2013
2014         dlb2 = ev_port->dlb2;
2015
2016         DLB2_LOG_DBG("Linking %u queues to %s port %d\n",
2017                      nb_links,
2018                      ev_port->qm_port.is_directed ? "DIR" : "LDB",
2019                      ev_port->id);
2020
2021         for (i = 0; i < nb_links; i++) {
2022                 struct dlb2_eventdev_queue *ev_queue;
2023                 uint8_t queue_id, prio;
2024                 bool found = false;
2025                 int index = -1;
2026
2027                 queue_id = queues[i];
2028                 prio = priorities[i];
2029
2030                 /* Check if the link already exists. */
2031                 for (j = 0; j < DLB2_MAX_NUM_QIDS_PER_LDB_CQ; j++)
2032                         if (ev_port->link[j].valid) {
2033                                 if (ev_port->link[j].queue_id == queue_id) {
2034                                         found = true;
2035                                         index = j;
2036                                         break;
2037                                 }
2038                         } else if (index == -1) {
2039                                 index = j;
2040                         }
2041
2042                 /* could not link */
2043                 if (index == -1)
2044                         break;
2045
2046                 /* Check if already linked at the requested priority */
2047                 if (found && ev_port->link[j].priority == prio)
2048                         continue;
2049
2050                 if (dlb2_validate_port_link(ev_port, queue_id, found, index))
2051                         break; /* return index of offending queue */
2052
2053                 ev_queue = &dlb2->ev_queues[queue_id];
2054
2055                 if (dlb2_do_port_link(dev, ev_queue, ev_port, prio))
2056                         break; /* return index of offending queue */
2057
2058                 ev_queue->num_links++;
2059
2060                 ev_port->link[index].queue_id = queue_id;
2061                 ev_port->link[index].priority = prio;
2062                 ev_port->link[index].valid = true;
2063                 /* Entry already exists?  If so, then must be prio change */
2064                 if (!found)
2065                         ev_port->num_links++;
2066         }
2067         return i;
2068 }
2069
2070 static int16_t
2071 dlb2_hw_unmap_ldb_qid_from_port(struct dlb2_hw_dev *handle,
2072                                 uint32_t qm_port_id,
2073                                 uint16_t qm_qid)
2074 {
2075         struct dlb2_unmap_qid_args cfg;
2076         int32_t ret;
2077
2078         if (handle == NULL)
2079                 return -EINVAL;
2080
2081         cfg.port_id = qm_port_id;
2082         cfg.qid = qm_qid;
2083
2084         ret = dlb2_iface_unmap_qid(handle, &cfg);
2085         if (ret < 0)
2086                 DLB2_LOG_ERR("dlb2: unmap qid error, ret=%d (driver status: %s)\n",
2087                              ret, dlb2_error_strings[cfg.response.status]);
2088
2089         return ret;
2090 }
2091
2092 static int
2093 dlb2_event_queue_detach_ldb(struct dlb2_eventdev *dlb2,
2094                             struct dlb2_eventdev_port *ev_port,
2095                             struct dlb2_eventdev_queue *ev_queue)
2096 {
2097         int ret, i;
2098
2099         /* Don't unlink until start time. */
2100         if (dlb2->run_state == DLB2_RUN_STATE_STOPPED)
2101                 return 0;
2102
2103         for (i = 0; i < DLB2_MAX_NUM_QIDS_PER_LDB_CQ; i++) {
2104                 if (ev_port->link[i].valid &&
2105                     ev_port->link[i].queue_id == ev_queue->id)
2106                         break; /* found */
2107         }
2108
2109         /* This is expected with eventdev API!
2110          * It blindly attemmpts to unmap all queues.
2111          */
2112         if (i == DLB2_MAX_NUM_QIDS_PER_LDB_CQ) {
2113                 DLB2_LOG_DBG("dlb2: ignoring LB QID %d not mapped for qm_port %d.\n",
2114                              ev_queue->qm_queue.id,
2115                              ev_port->qm_port.id);
2116                 return 0;
2117         }
2118
2119         ret = dlb2_hw_unmap_ldb_qid_from_port(&dlb2->qm_instance,
2120                                               ev_port->qm_port.id,
2121                                               ev_queue->qm_queue.id);
2122         if (!ret)
2123                 ev_port->link[i].mapped = false;
2124
2125         return ret;
2126 }
2127
2128 static int
2129 dlb2_eventdev_port_unlink(struct rte_eventdev *dev, void *event_port,
2130                           uint8_t queues[], uint16_t nb_unlinks)
2131 {
2132         struct dlb2_eventdev_port *ev_port = event_port;
2133         struct dlb2_eventdev *dlb2;
2134         int i;
2135
2136         RTE_SET_USED(dev);
2137
2138         if (!ev_port->setup_done) {
2139                 DLB2_LOG_ERR("dlb2: evport %d is not configured\n",
2140                              ev_port->id);
2141                 rte_errno = -EINVAL;
2142                 return 0;
2143         }
2144
2145         if (queues == NULL || nb_unlinks == 0) {
2146                 DLB2_LOG_DBG("dlb2: queues is NULL or nb_unlinks is 0\n");
2147                 return 0; /* Ignore and return success */
2148         }
2149
2150         if (ev_port->qm_port.is_directed) {
2151                 DLB2_LOG_DBG("dlb2: ignore unlink from dir port %d\n",
2152                              ev_port->id);
2153                 rte_errno = 0;
2154                 return nb_unlinks; /* as if success */
2155         }
2156
2157         dlb2 = ev_port->dlb2;
2158
2159         for (i = 0; i < nb_unlinks; i++) {
2160                 struct dlb2_eventdev_queue *ev_queue;
2161                 int ret, j;
2162
2163                 if (queues[i] >= dlb2->num_queues) {
2164                         DLB2_LOG_ERR("dlb2: invalid queue id %d\n", queues[i]);
2165                         rte_errno = -EINVAL;
2166                         return i; /* return index of offending queue */
2167                 }
2168
2169                 ev_queue = &dlb2->ev_queues[queues[i]];
2170
2171                 /* Does a link exist? */
2172                 for (j = 0; j < DLB2_MAX_NUM_QIDS_PER_LDB_CQ; j++)
2173                         if (ev_port->link[j].queue_id == queues[i] &&
2174                             ev_port->link[j].valid)
2175                                 break;
2176
2177                 if (j == DLB2_MAX_NUM_QIDS_PER_LDB_CQ)
2178                         continue;
2179
2180                 ret = dlb2_event_queue_detach_ldb(dlb2, ev_port, ev_queue);
2181                 if (ret) {
2182                         DLB2_LOG_ERR("unlink err=%d for port %d queue %d\n",
2183                                      ret, ev_port->id, queues[i]);
2184                         rte_errno = -ENOENT;
2185                         return i; /* return index of offending queue */
2186                 }
2187
2188                 ev_port->link[j].valid = false;
2189                 ev_port->num_links--;
2190                 ev_queue->num_links--;
2191         }
2192
2193         return nb_unlinks;
2194 }
2195
2196 static int
2197 dlb2_eventdev_port_unlinks_in_progress(struct rte_eventdev *dev,
2198                                        void *event_port)
2199 {
2200         struct dlb2_eventdev_port *ev_port = event_port;
2201         struct dlb2_eventdev *dlb2;
2202         struct dlb2_hw_dev *handle;
2203         struct dlb2_pending_port_unmaps_args cfg;
2204         int ret;
2205
2206         RTE_SET_USED(dev);
2207
2208         if (!ev_port->setup_done) {
2209                 DLB2_LOG_ERR("dlb2: evport %d is not configured\n",
2210                              ev_port->id);
2211                 rte_errno = -EINVAL;
2212                 return 0;
2213         }
2214
2215         cfg.port_id = ev_port->qm_port.id;
2216         dlb2 = ev_port->dlb2;
2217         handle = &dlb2->qm_instance;
2218         ret = dlb2_iface_pending_port_unmaps(handle, &cfg);
2219
2220         if (ret < 0) {
2221                 DLB2_LOG_ERR("dlb2: num_unlinks_in_progress ret=%d (driver status: %s)\n",
2222                              ret, dlb2_error_strings[cfg.response.status]);
2223                 return ret;
2224         }
2225
2226         return cfg.response.id;
2227 }
2228
2229 static int
2230 dlb2_eventdev_reapply_configuration(struct rte_eventdev *dev)
2231 {
2232         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
2233         int ret, i;
2234
2235         /* If an event queue or port was previously configured, but hasn't been
2236          * reconfigured, reapply its original configuration.
2237          */
2238         for (i = 0; i < dlb2->num_queues; i++) {
2239                 struct dlb2_eventdev_queue *ev_queue;
2240
2241                 ev_queue = &dlb2->ev_queues[i];
2242
2243                 if (ev_queue->qm_queue.config_state != DLB2_PREV_CONFIGURED)
2244                         continue;
2245
2246                 ret = dlb2_eventdev_queue_setup(dev, i, &ev_queue->conf);
2247                 if (ret < 0) {
2248                         DLB2_LOG_ERR("dlb2: failed to reconfigure queue %d", i);
2249                         return ret;
2250                 }
2251         }
2252
2253         for (i = 0; i < dlb2->num_ports; i++) {
2254                 struct dlb2_eventdev_port *ev_port = &dlb2->ev_ports[i];
2255
2256                 if (ev_port->qm_port.config_state != DLB2_PREV_CONFIGURED)
2257                         continue;
2258
2259                 ret = dlb2_eventdev_port_setup(dev, i, &ev_port->conf);
2260                 if (ret < 0) {
2261                         DLB2_LOG_ERR("dlb2: failed to reconfigure ev_port %d",
2262                                      i);
2263                         return ret;
2264                 }
2265         }
2266
2267         return 0;
2268 }
2269
2270 static int
2271 dlb2_eventdev_apply_port_links(struct rte_eventdev *dev)
2272 {
2273         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
2274         int i;
2275
2276         /* Perform requested port->queue links */
2277         for (i = 0; i < dlb2->num_ports; i++) {
2278                 struct dlb2_eventdev_port *ev_port = &dlb2->ev_ports[i];
2279                 int j;
2280
2281                 for (j = 0; j < DLB2_MAX_NUM_QIDS_PER_LDB_CQ; j++) {
2282                         struct dlb2_eventdev_queue *ev_queue;
2283                         uint8_t prio, queue_id;
2284
2285                         if (!ev_port->link[j].valid)
2286                                 continue;
2287
2288                         prio = ev_port->link[j].priority;
2289                         queue_id = ev_port->link[j].queue_id;
2290
2291                         if (dlb2_validate_port_link(ev_port, queue_id, true, j))
2292                                 return -EINVAL;
2293
2294                         ev_queue = &dlb2->ev_queues[queue_id];
2295
2296                         if (dlb2_do_port_link(dev, ev_queue, ev_port, prio))
2297                                 return -EINVAL;
2298                 }
2299         }
2300
2301         return 0;
2302 }
2303
2304 static int
2305 dlb2_eventdev_start(struct rte_eventdev *dev)
2306 {
2307         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
2308         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
2309         struct dlb2_start_domain_args cfg;
2310         int ret, i;
2311
2312         rte_spinlock_lock(&dlb2->qm_instance.resource_lock);
2313         if (dlb2->run_state != DLB2_RUN_STATE_STOPPED) {
2314                 DLB2_LOG_ERR("bad state %d for dev_start\n",
2315                              (int)dlb2->run_state);
2316                 rte_spinlock_unlock(&dlb2->qm_instance.resource_lock);
2317                 return -EINVAL;
2318         }
2319         dlb2->run_state = DLB2_RUN_STATE_STARTING;
2320         rte_spinlock_unlock(&dlb2->qm_instance.resource_lock);
2321
2322         /* If the device was configured more than once, some event ports and/or
2323          * queues may need to be reconfigured.
2324          */
2325         ret = dlb2_eventdev_reapply_configuration(dev);
2326         if (ret)
2327                 return ret;
2328
2329         /* The DLB PMD delays port links until the device is started. */
2330         ret = dlb2_eventdev_apply_port_links(dev);
2331         if (ret)
2332                 return ret;
2333
2334         for (i = 0; i < dlb2->num_ports; i++) {
2335                 if (!dlb2->ev_ports[i].setup_done) {
2336                         DLB2_LOG_ERR("dlb2: port %d not setup", i);
2337                         return -ESTALE;
2338                 }
2339         }
2340
2341         for (i = 0; i < dlb2->num_queues; i++) {
2342                 if (dlb2->ev_queues[i].num_links == 0) {
2343                         DLB2_LOG_ERR("dlb2: queue %d is not linked", i);
2344                         return -ENOLINK;
2345                 }
2346         }
2347
2348         ret = dlb2_iface_sched_domain_start(handle, &cfg);
2349         if (ret < 0) {
2350                 DLB2_LOG_ERR("dlb2: sched_domain_start ret=%d (driver status: %s)\n",
2351                              ret, dlb2_error_strings[cfg.response.status]);
2352                 return ret;
2353         }
2354
2355         dlb2->run_state = DLB2_RUN_STATE_STARTED;
2356         DLB2_LOG_DBG("dlb2: sched_domain_start completed OK\n");
2357
2358         return 0;
2359 }
2360
2361 static uint8_t cmd_byte_map[DLB2_NUM_PORT_TYPES][DLB2_NUM_HW_SCHED_TYPES] = {
2362         {
2363                 /* Load-balanced cmd bytes */
2364                 [RTE_EVENT_OP_NEW] = DLB2_NEW_CMD_BYTE,
2365                 [RTE_EVENT_OP_FORWARD] = DLB2_FWD_CMD_BYTE,
2366                 [RTE_EVENT_OP_RELEASE] = DLB2_COMP_CMD_BYTE,
2367         },
2368         {
2369                 /* Directed cmd bytes */
2370                 [RTE_EVENT_OP_NEW] = DLB2_NEW_CMD_BYTE,
2371                 [RTE_EVENT_OP_FORWARD] = DLB2_NEW_CMD_BYTE,
2372                 [RTE_EVENT_OP_RELEASE] = DLB2_NOOP_CMD_BYTE,
2373         },
2374 };
2375
2376 static inline uint32_t
2377 dlb2_port_credits_get(struct dlb2_port *qm_port,
2378                       enum dlb2_hw_queue_types type)
2379 {
2380         uint32_t credits = *qm_port->credit_pool[type];
2381         uint32_t batch_size = DLB2_SW_CREDIT_BATCH_SZ;
2382
2383         if (unlikely(credits < batch_size))
2384                 batch_size = credits;
2385
2386         if (likely(credits &&
2387                    __atomic_compare_exchange_n(
2388                         qm_port->credit_pool[type],
2389                         &credits, credits - batch_size, false,
2390                         __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)))
2391                 return batch_size;
2392         else
2393                 return 0;
2394 }
2395
2396 static inline void
2397 dlb2_replenish_sw_credits(struct dlb2_eventdev *dlb2,
2398                           struct dlb2_eventdev_port *ev_port)
2399 {
2400         uint16_t quanta = ev_port->credit_update_quanta;
2401
2402         if (ev_port->inflight_credits >= quanta * 2) {
2403                 /* Replenish credits, saving one quanta for enqueues */
2404                 uint16_t val = ev_port->inflight_credits - quanta;
2405
2406                 __atomic_fetch_sub(&dlb2->inflights, val, __ATOMIC_SEQ_CST);
2407                 ev_port->inflight_credits -= val;
2408         }
2409 }
2410
2411 static inline int
2412 dlb2_check_enqueue_sw_credits(struct dlb2_eventdev *dlb2,
2413                               struct dlb2_eventdev_port *ev_port)
2414 {
2415         uint32_t sw_inflights = __atomic_load_n(&dlb2->inflights,
2416                                                 __ATOMIC_SEQ_CST);
2417         const int num = 1;
2418
2419         if (unlikely(ev_port->inflight_max < sw_inflights)) {
2420                 DLB2_INC_STAT(ev_port->stats.traffic.tx_nospc_inflight_max, 1);
2421                 rte_errno = -ENOSPC;
2422                 return 1;
2423         }
2424
2425         if (ev_port->inflight_credits < num) {
2426                 /* check if event enqueue brings ev_port over max threshold */
2427                 uint32_t credit_update_quanta = ev_port->credit_update_quanta;
2428
2429                 if (sw_inflights + credit_update_quanta >
2430                                 dlb2->new_event_limit) {
2431                         DLB2_INC_STAT(
2432                         ev_port->stats.traffic.tx_nospc_new_event_limit,
2433                         1);
2434                         rte_errno = -ENOSPC;
2435                         return 1;
2436                 }
2437
2438                 __atomic_fetch_add(&dlb2->inflights, credit_update_quanta,
2439                                    __ATOMIC_SEQ_CST);
2440                 ev_port->inflight_credits += (credit_update_quanta);
2441
2442                 if (ev_port->inflight_credits < num) {
2443                         DLB2_INC_STAT(
2444                         ev_port->stats.traffic.tx_nospc_inflight_credits,
2445                         1);
2446                         rte_errno = -ENOSPC;
2447                         return 1;
2448                 }
2449         }
2450
2451         return 0;
2452 }
2453
2454 static inline int
2455 dlb2_check_enqueue_hw_ldb_credits(struct dlb2_port *qm_port)
2456 {
2457         if (unlikely(qm_port->cached_ldb_credits == 0)) {
2458                 qm_port->cached_ldb_credits =
2459                         dlb2_port_credits_get(qm_port,
2460                                               DLB2_LDB_QUEUE);
2461                 if (unlikely(qm_port->cached_ldb_credits == 0)) {
2462                         DLB2_INC_STAT(
2463                         qm_port->ev_port->stats.traffic.tx_nospc_ldb_hw_credits,
2464                         1);
2465                         DLB2_LOG_DBG("ldb credits exhausted\n");
2466                         return 1; /* credits exhausted */
2467                 }
2468         }
2469
2470         return 0;
2471 }
2472
2473 static inline int
2474 dlb2_check_enqueue_hw_dir_credits(struct dlb2_port *qm_port)
2475 {
2476         if (unlikely(qm_port->cached_dir_credits == 0)) {
2477                 qm_port->cached_dir_credits =
2478                         dlb2_port_credits_get(qm_port,
2479                                               DLB2_DIR_QUEUE);
2480                 if (unlikely(qm_port->cached_dir_credits == 0)) {
2481                         DLB2_INC_STAT(
2482                         qm_port->ev_port->stats.traffic.tx_nospc_dir_hw_credits,
2483                         1);
2484                         DLB2_LOG_DBG("dir credits exhausted\n");
2485                         return 1; /* credits exhausted */
2486                 }
2487         }
2488
2489         return 0;
2490 }
2491
2492 static inline int
2493 dlb2_check_enqueue_hw_credits(struct dlb2_port *qm_port)
2494 {
2495         if (unlikely(qm_port->cached_credits == 0)) {
2496                 qm_port->cached_credits =
2497                         dlb2_port_credits_get(qm_port,
2498                                               DLB2_COMBINED_POOL);
2499                 if (unlikely(qm_port->cached_credits == 0)) {
2500                         DLB2_INC_STAT(
2501                         qm_port->ev_port->stats.traffic.tx_nospc_hw_credits, 1);
2502                         DLB2_LOG_DBG("credits exhausted\n");
2503                         return 1; /* credits exhausted */
2504                 }
2505         }
2506
2507         return 0;
2508 }
2509
2510 static __rte_always_inline void
2511 dlb2_pp_write(struct dlb2_enqueue_qe *qe4,
2512               struct process_local_port_data *port_data)
2513 {
2514         dlb2_movdir64b(port_data->pp_addr, qe4);
2515 }
2516
2517 static inline int
2518 dlb2_consume_qe_immediate(struct dlb2_port *qm_port, int num)
2519 {
2520         struct process_local_port_data *port_data;
2521         struct dlb2_cq_pop_qe *qe;
2522
2523         RTE_ASSERT(qm_port->config_state == DLB2_CONFIGURED);
2524
2525         qe = qm_port->consume_qe;
2526
2527         qe->tokens = num - 1;
2528
2529         /* No store fence needed since no pointer is being sent, and CQ token
2530          * pops can be safely reordered with other HCWs.
2531          */
2532         port_data = &dlb2_port[qm_port->id][PORT_TYPE(qm_port)];
2533
2534         dlb2_movntdq_single(port_data->pp_addr, qe);
2535
2536         DLB2_LOG_DBG("dlb2: consume immediate - %d QEs\n", num);
2537
2538         qm_port->owed_tokens = 0;
2539
2540         return 0;
2541 }
2542
2543 static inline void
2544 dlb2_hw_do_enqueue(struct dlb2_port *qm_port,
2545                    bool do_sfence,
2546                    struct process_local_port_data *port_data)
2547 {
2548         /* Since MOVDIR64B is weakly-ordered, use an SFENCE to ensure that
2549          * application writes complete before enqueueing the QE.
2550          */
2551         if (do_sfence)
2552                 rte_wmb();
2553
2554         dlb2_pp_write(qm_port->qe4, port_data);
2555 }
2556
2557 static inline void
2558 dlb2_construct_token_pop_qe(struct dlb2_port *qm_port, int idx)
2559 {
2560         struct dlb2_cq_pop_qe *qe = (void *)qm_port->qe4;
2561         int num = qm_port->owed_tokens;
2562
2563         qe[idx].cmd_byte = DLB2_POP_CMD_BYTE;
2564         qe[idx].tokens = num - 1;
2565
2566         qm_port->owed_tokens = 0;
2567 }
2568
2569 static inline void
2570 dlb2_event_build_hcws(struct dlb2_port *qm_port,
2571                       const struct rte_event ev[],
2572                       int num,
2573                       uint8_t *sched_type,
2574                       uint8_t *queue_id)
2575 {
2576         struct dlb2_enqueue_qe *qe;
2577         uint16_t sched_word[4];
2578         __m128i sse_qe[2];
2579         int i;
2580
2581         qe = qm_port->qe4;
2582
2583         sse_qe[0] = _mm_setzero_si128();
2584         sse_qe[1] = _mm_setzero_si128();
2585
2586         switch (num) {
2587         case 4:
2588                 /* Construct the metadata portion of two HCWs in one 128b SSE
2589                  * register. HCW metadata is constructed in the SSE registers
2590                  * like so:
2591                  * sse_qe[0][63:0]:   qe[0]'s metadata
2592                  * sse_qe[0][127:64]: qe[1]'s metadata
2593                  * sse_qe[1][63:0]:   qe[2]'s metadata
2594                  * sse_qe[1][127:64]: qe[3]'s metadata
2595                  */
2596
2597                 /* Convert the event operation into a command byte and store it
2598                  * in the metadata:
2599                  * sse_qe[0][63:56]   = cmd_byte_map[is_directed][ev[0].op]
2600                  * sse_qe[0][127:120] = cmd_byte_map[is_directed][ev[1].op]
2601                  * sse_qe[1][63:56]   = cmd_byte_map[is_directed][ev[2].op]
2602                  * sse_qe[1][127:120] = cmd_byte_map[is_directed][ev[3].op]
2603                  */
2604 #define DLB2_QE_CMD_BYTE 7
2605                 sse_qe[0] = _mm_insert_epi8(sse_qe[0],
2606                                 cmd_byte_map[qm_port->is_directed][ev[0].op],
2607                                 DLB2_QE_CMD_BYTE);
2608                 sse_qe[0] = _mm_insert_epi8(sse_qe[0],
2609                                 cmd_byte_map[qm_port->is_directed][ev[1].op],
2610                                 DLB2_QE_CMD_BYTE + 8);
2611                 sse_qe[1] = _mm_insert_epi8(sse_qe[1],
2612                                 cmd_byte_map[qm_port->is_directed][ev[2].op],
2613                                 DLB2_QE_CMD_BYTE);
2614                 sse_qe[1] = _mm_insert_epi8(sse_qe[1],
2615                                 cmd_byte_map[qm_port->is_directed][ev[3].op],
2616                                 DLB2_QE_CMD_BYTE + 8);
2617
2618                 /* Store priority, scheduling type, and queue ID in the sched
2619                  * word array because these values are re-used when the
2620                  * destination is a directed queue.
2621                  */
2622                 sched_word[0] = EV_TO_DLB2_PRIO(ev[0].priority) << 10 |
2623                                 sched_type[0] << 8 |
2624                                 queue_id[0];
2625                 sched_word[1] = EV_TO_DLB2_PRIO(ev[1].priority) << 10 |
2626                                 sched_type[1] << 8 |
2627                                 queue_id[1];
2628                 sched_word[2] = EV_TO_DLB2_PRIO(ev[2].priority) << 10 |
2629                                 sched_type[2] << 8 |
2630                                 queue_id[2];
2631                 sched_word[3] = EV_TO_DLB2_PRIO(ev[3].priority) << 10 |
2632                                 sched_type[3] << 8 |
2633                                 queue_id[3];
2634
2635                 /* Store the event priority, scheduling type, and queue ID in
2636                  * the metadata:
2637                  * sse_qe[0][31:16] = sched_word[0]
2638                  * sse_qe[0][95:80] = sched_word[1]
2639                  * sse_qe[1][31:16] = sched_word[2]
2640                  * sse_qe[1][95:80] = sched_word[3]
2641                  */
2642 #define DLB2_QE_QID_SCHED_WORD 1
2643                 sse_qe[0] = _mm_insert_epi16(sse_qe[0],
2644                                              sched_word[0],
2645                                              DLB2_QE_QID_SCHED_WORD);
2646                 sse_qe[0] = _mm_insert_epi16(sse_qe[0],
2647                                              sched_word[1],
2648                                              DLB2_QE_QID_SCHED_WORD + 4);
2649                 sse_qe[1] = _mm_insert_epi16(sse_qe[1],
2650                                              sched_word[2],
2651                                              DLB2_QE_QID_SCHED_WORD);
2652                 sse_qe[1] = _mm_insert_epi16(sse_qe[1],
2653                                              sched_word[3],
2654                                              DLB2_QE_QID_SCHED_WORD + 4);
2655
2656                 /* If the destination is a load-balanced queue, store the lock
2657                  * ID. If it is a directed queue, DLB places this field in
2658                  * bytes 10-11 of the received QE, so we format it accordingly:
2659                  * sse_qe[0][47:32]  = dir queue ? sched_word[0] : flow_id[0]
2660                  * sse_qe[0][111:96] = dir queue ? sched_word[1] : flow_id[1]
2661                  * sse_qe[1][47:32]  = dir queue ? sched_word[2] : flow_id[2]
2662                  * sse_qe[1][111:96] = dir queue ? sched_word[3] : flow_id[3]
2663                  */
2664 #define DLB2_QE_LOCK_ID_WORD 2
2665                 sse_qe[0] = _mm_insert_epi16(sse_qe[0],
2666                                 (sched_type[0] == DLB2_SCHED_DIRECTED) ?
2667                                         sched_word[0] : ev[0].flow_id,
2668                                 DLB2_QE_LOCK_ID_WORD);
2669                 sse_qe[0] = _mm_insert_epi16(sse_qe[0],
2670                                 (sched_type[1] == DLB2_SCHED_DIRECTED) ?
2671                                         sched_word[1] : ev[1].flow_id,
2672                                 DLB2_QE_LOCK_ID_WORD + 4);
2673                 sse_qe[1] = _mm_insert_epi16(sse_qe[1],
2674                                 (sched_type[2] == DLB2_SCHED_DIRECTED) ?
2675                                         sched_word[2] : ev[2].flow_id,
2676                                 DLB2_QE_LOCK_ID_WORD);
2677                 sse_qe[1] = _mm_insert_epi16(sse_qe[1],
2678                                 (sched_type[3] == DLB2_SCHED_DIRECTED) ?
2679                                         sched_word[3] : ev[3].flow_id,
2680                                 DLB2_QE_LOCK_ID_WORD + 4);
2681
2682                 /* Store the event type and sub event type in the metadata:
2683                  * sse_qe[0][15:0]  = flow_id[0]
2684                  * sse_qe[0][79:64] = flow_id[1]
2685                  * sse_qe[1][15:0]  = flow_id[2]
2686                  * sse_qe[1][79:64] = flow_id[3]
2687                  */
2688 #define DLB2_QE_EV_TYPE_WORD 0
2689                 sse_qe[0] = _mm_insert_epi16(sse_qe[0],
2690                                              ev[0].sub_event_type << 8 |
2691                                                 ev[0].event_type,
2692                                              DLB2_QE_EV_TYPE_WORD);
2693                 sse_qe[0] = _mm_insert_epi16(sse_qe[0],
2694                                              ev[1].sub_event_type << 8 |
2695                                                 ev[1].event_type,
2696                                              DLB2_QE_EV_TYPE_WORD + 4);
2697                 sse_qe[1] = _mm_insert_epi16(sse_qe[1],
2698                                              ev[2].sub_event_type << 8 |
2699                                                 ev[2].event_type,
2700                                              DLB2_QE_EV_TYPE_WORD);
2701                 sse_qe[1] = _mm_insert_epi16(sse_qe[1],
2702                                              ev[3].sub_event_type << 8 |
2703                                                 ev[3].event_type,
2704                                              DLB2_QE_EV_TYPE_WORD + 4);
2705
2706                 /* Store the metadata to memory (use the double-precision
2707                  * _mm_storeh_pd because there is no integer function for
2708                  * storing the upper 64b):
2709                  * qe[0] metadata = sse_qe[0][63:0]
2710                  * qe[1] metadata = sse_qe[0][127:64]
2711                  * qe[2] metadata = sse_qe[1][63:0]
2712                  * qe[3] metadata = sse_qe[1][127:64]
2713                  */
2714                 _mm_storel_epi64((__m128i *)&qe[0].u.opaque_data, sse_qe[0]);
2715                 _mm_storeh_pd((double *)&qe[1].u.opaque_data,
2716                               (__m128d)sse_qe[0]);
2717                 _mm_storel_epi64((__m128i *)&qe[2].u.opaque_data, sse_qe[1]);
2718                 _mm_storeh_pd((double *)&qe[3].u.opaque_data,
2719                               (__m128d)sse_qe[1]);
2720
2721                 qe[0].data = ev[0].u64;
2722                 qe[1].data = ev[1].u64;
2723                 qe[2].data = ev[2].u64;
2724                 qe[3].data = ev[3].u64;
2725
2726                 break;
2727         case 3:
2728         case 2:
2729         case 1:
2730                 for (i = 0; i < num; i++) {
2731                         qe[i].cmd_byte =
2732                                 cmd_byte_map[qm_port->is_directed][ev[i].op];
2733                         qe[i].sched_type = sched_type[i];
2734                         qe[i].data = ev[i].u64;
2735                         qe[i].qid = queue_id[i];
2736                         qe[i].priority = EV_TO_DLB2_PRIO(ev[i].priority);
2737                         qe[i].lock_id = ev[i].flow_id;
2738                         if (sched_type[i] == DLB2_SCHED_DIRECTED) {
2739                                 struct dlb2_msg_info *info =
2740                                         (struct dlb2_msg_info *)&qe[i].lock_id;
2741
2742                                 info->qid = queue_id[i];
2743                                 info->sched_type = DLB2_SCHED_DIRECTED;
2744                                 info->priority = qe[i].priority;
2745                         }
2746                         qe[i].u.event_type.major = ev[i].event_type;
2747                         qe[i].u.event_type.sub = ev[i].sub_event_type;
2748                 }
2749                 break;
2750         case 0:
2751                 break;
2752         }
2753 }
2754
2755 static inline int
2756 dlb2_event_enqueue_prep(struct dlb2_eventdev_port *ev_port,
2757                         struct dlb2_port *qm_port,
2758                         const struct rte_event ev[],
2759                         uint8_t *sched_type,
2760                         uint8_t *queue_id)
2761 {
2762         struct dlb2_eventdev *dlb2 = ev_port->dlb2;
2763         struct dlb2_eventdev_queue *ev_queue;
2764         uint16_t *cached_credits = NULL;
2765         struct dlb2_queue *qm_queue;
2766
2767         ev_queue = &dlb2->ev_queues[ev->queue_id];
2768         qm_queue = &ev_queue->qm_queue;
2769         *queue_id = qm_queue->id;
2770
2771         /* Ignore sched_type and hardware credits on release events */
2772         if (ev->op == RTE_EVENT_OP_RELEASE)
2773                 goto op_check;
2774
2775         if (!qm_queue->is_directed) {
2776                 /* Load balanced destination queue */
2777
2778                 if (dlb2->version == DLB2_HW_V2) {
2779                         if (dlb2_check_enqueue_hw_ldb_credits(qm_port)) {
2780                                 rte_errno = -ENOSPC;
2781                                 return 1;
2782                         }
2783                         cached_credits = &qm_port->cached_ldb_credits;
2784                 } else {
2785                         if (dlb2_check_enqueue_hw_credits(qm_port)) {
2786                                 rte_errno = -ENOSPC;
2787                                 return 1;
2788                         }
2789                         cached_credits = &qm_port->cached_credits;
2790                 }
2791                 switch (ev->sched_type) {
2792                 case RTE_SCHED_TYPE_ORDERED:
2793                         DLB2_LOG_DBG("dlb2: put_qe: RTE_SCHED_TYPE_ORDERED\n");
2794                         if (qm_queue->sched_type != RTE_SCHED_TYPE_ORDERED) {
2795                                 DLB2_LOG_ERR("dlb2: tried to send ordered event to unordered queue %d\n",
2796                                              *queue_id);
2797                                 rte_errno = -EINVAL;
2798                                 return 1;
2799                         }
2800                         *sched_type = DLB2_SCHED_ORDERED;
2801                         break;
2802                 case RTE_SCHED_TYPE_ATOMIC:
2803                         DLB2_LOG_DBG("dlb2: put_qe: RTE_SCHED_TYPE_ATOMIC\n");
2804                         *sched_type = DLB2_SCHED_ATOMIC;
2805                         break;
2806                 case RTE_SCHED_TYPE_PARALLEL:
2807                         DLB2_LOG_DBG("dlb2: put_qe: RTE_SCHED_TYPE_PARALLEL\n");
2808                         if (qm_queue->sched_type == RTE_SCHED_TYPE_ORDERED)
2809                                 *sched_type = DLB2_SCHED_ORDERED;
2810                         else
2811                                 *sched_type = DLB2_SCHED_UNORDERED;
2812                         break;
2813                 default:
2814                         DLB2_LOG_ERR("Unsupported LDB sched type in put_qe\n");
2815                         DLB2_INC_STAT(ev_port->stats.tx_invalid, 1);
2816                         rte_errno = -EINVAL;
2817                         return 1;
2818                 }
2819         } else {
2820                 /* Directed destination queue */
2821
2822                 if (dlb2->version == DLB2_HW_V2) {
2823                         if (dlb2_check_enqueue_hw_dir_credits(qm_port)) {
2824                                 rte_errno = -ENOSPC;
2825                                 return 1;
2826                         }
2827                         cached_credits = &qm_port->cached_dir_credits;
2828                 } else {
2829                         if (dlb2_check_enqueue_hw_credits(qm_port)) {
2830                                 rte_errno = -ENOSPC;
2831                                 return 1;
2832                         }
2833                         cached_credits = &qm_port->cached_credits;
2834                 }
2835                 DLB2_LOG_DBG("dlb2: put_qe: RTE_SCHED_TYPE_DIRECTED\n");
2836
2837                 *sched_type = DLB2_SCHED_DIRECTED;
2838         }
2839
2840 op_check:
2841         switch (ev->op) {
2842         case RTE_EVENT_OP_NEW:
2843                 /* Check that a sw credit is available */
2844                 if (dlb2_check_enqueue_sw_credits(dlb2, ev_port)) {
2845                         rte_errno = -ENOSPC;
2846                         return 1;
2847                 }
2848                 ev_port->inflight_credits--;
2849                 (*cached_credits)--;
2850                 break;
2851         case RTE_EVENT_OP_FORWARD:
2852                 /* Check for outstanding_releases underflow. If this occurs,
2853                  * the application is not using the EVENT_OPs correctly; for
2854                  * example, forwarding or releasing events that were not
2855                  * dequeued.
2856                  */
2857                 RTE_ASSERT(ev_port->outstanding_releases > 0);
2858                 ev_port->outstanding_releases--;
2859                 qm_port->issued_releases++;
2860                 (*cached_credits)--;
2861                 break;
2862         case RTE_EVENT_OP_RELEASE:
2863                 ev_port->inflight_credits++;
2864                 /* Check for outstanding_releases underflow. If this occurs,
2865                  * the application is not using the EVENT_OPs correctly; for
2866                  * example, forwarding or releasing events that were not
2867                  * dequeued.
2868                  */
2869                 RTE_ASSERT(ev_port->outstanding_releases > 0);
2870                 ev_port->outstanding_releases--;
2871                 qm_port->issued_releases++;
2872
2873                 /* Replenish s/w credits if enough are cached */
2874                 dlb2_replenish_sw_credits(dlb2, ev_port);
2875                 break;
2876         }
2877
2878         DLB2_INC_STAT(ev_port->stats.tx_op_cnt[ev->op], 1);
2879         DLB2_INC_STAT(ev_port->stats.traffic.tx_ok, 1);
2880
2881 #ifndef RTE_LIBRTE_PMD_DLB_QUELL_STATS
2882         if (ev->op != RTE_EVENT_OP_RELEASE) {
2883                 DLB2_INC_STAT(ev_port->stats.queue[ev->queue_id].enq_ok, 1);
2884                 DLB2_INC_STAT(ev_port->stats.tx_sched_cnt[*sched_type], 1);
2885         }
2886 #endif
2887
2888         return 0;
2889 }
2890
2891 static inline uint16_t
2892 __dlb2_event_enqueue_burst(void *event_port,
2893                            const struct rte_event events[],
2894                            uint16_t num,
2895                            bool use_delayed)
2896 {
2897         struct dlb2_eventdev_port *ev_port = event_port;
2898         struct dlb2_port *qm_port = &ev_port->qm_port;
2899         struct process_local_port_data *port_data;
2900         int i;
2901
2902         RTE_ASSERT(ev_port->enq_configured);
2903         RTE_ASSERT(events != NULL);
2904
2905         i = 0;
2906
2907         port_data = &dlb2_port[qm_port->id][PORT_TYPE(qm_port)];
2908
2909         while (i < num) {
2910                 uint8_t sched_types[DLB2_NUM_QES_PER_CACHE_LINE];
2911                 uint8_t queue_ids[DLB2_NUM_QES_PER_CACHE_LINE];
2912                 int pop_offs = 0;
2913                 int j = 0;
2914
2915                 memset(qm_port->qe4,
2916                        0,
2917                        DLB2_NUM_QES_PER_CACHE_LINE *
2918                        sizeof(struct dlb2_enqueue_qe));
2919
2920                 for (; j < DLB2_NUM_QES_PER_CACHE_LINE && (i + j) < num; j++) {
2921                         const struct rte_event *ev = &events[i + j];
2922                         int16_t thresh = qm_port->token_pop_thresh;
2923
2924                         if (use_delayed &&
2925                             qm_port->token_pop_mode == DELAYED_POP &&
2926                             (ev->op == RTE_EVENT_OP_FORWARD ||
2927                              ev->op == RTE_EVENT_OP_RELEASE) &&
2928                             qm_port->issued_releases >= thresh - 1) {
2929                                 /* Insert the token pop QE and break out. This
2930                                  * may result in a partial HCW, but that is
2931                                  * simpler than supporting arbitrary QE
2932                                  * insertion.
2933                                  */
2934                                 dlb2_construct_token_pop_qe(qm_port, j);
2935
2936                                 /* Reset the releases for the next QE batch */
2937                                 qm_port->issued_releases -= thresh;
2938
2939                                 pop_offs = 1;
2940                                 j++;
2941                                 break;
2942                         }
2943
2944                         if (dlb2_event_enqueue_prep(ev_port, qm_port, ev,
2945                                                     &sched_types[j],
2946                                                     &queue_ids[j]))
2947                                 break;
2948                 }
2949
2950                 if (j == 0)
2951                         break;
2952
2953                 dlb2_event_build_hcws(qm_port, &events[i], j - pop_offs,
2954                                       sched_types, queue_ids);
2955
2956                 dlb2_hw_do_enqueue(qm_port, i == 0, port_data);
2957
2958                 /* Don't include the token pop QE in the enqueue count */
2959                 i += j - pop_offs;
2960
2961                 /* Don't interpret j < DLB2_NUM_... as out-of-credits if
2962                  * pop_offs != 0
2963                  */
2964                 if (j < DLB2_NUM_QES_PER_CACHE_LINE && pop_offs == 0)
2965                         break;
2966         }
2967
2968         return i;
2969 }
2970
2971 static uint16_t
2972 dlb2_event_enqueue_burst(void *event_port,
2973                              const struct rte_event events[],
2974                              uint16_t num)
2975 {
2976         return __dlb2_event_enqueue_burst(event_port, events, num, false);
2977 }
2978
2979 static uint16_t
2980 dlb2_event_enqueue_burst_delayed(void *event_port,
2981                                      const struct rte_event events[],
2982                                      uint16_t num)
2983 {
2984         return __dlb2_event_enqueue_burst(event_port, events, num, true);
2985 }
2986
2987 static inline uint16_t
2988 dlb2_event_enqueue(void *event_port,
2989                    const struct rte_event events[])
2990 {
2991         return __dlb2_event_enqueue_burst(event_port, events, 1, false);
2992 }
2993
2994 static inline uint16_t
2995 dlb2_event_enqueue_delayed(void *event_port,
2996                            const struct rte_event events[])
2997 {
2998         return __dlb2_event_enqueue_burst(event_port, events, 1, true);
2999 }
3000
3001 static uint16_t
3002 dlb2_event_enqueue_new_burst(void *event_port,
3003                              const struct rte_event events[],
3004                              uint16_t num)
3005 {
3006         return __dlb2_event_enqueue_burst(event_port, events, num, false);
3007 }
3008
3009 static uint16_t
3010 dlb2_event_enqueue_new_burst_delayed(void *event_port,
3011                                      const struct rte_event events[],
3012                                      uint16_t num)
3013 {
3014         return __dlb2_event_enqueue_burst(event_port, events, num, true);
3015 }
3016
3017 static uint16_t
3018 dlb2_event_enqueue_forward_burst(void *event_port,
3019                                  const struct rte_event events[],
3020                                  uint16_t num)
3021 {
3022         return __dlb2_event_enqueue_burst(event_port, events, num, false);
3023 }
3024
3025 static uint16_t
3026 dlb2_event_enqueue_forward_burst_delayed(void *event_port,
3027                                          const struct rte_event events[],
3028                                          uint16_t num)
3029 {
3030         return __dlb2_event_enqueue_burst(event_port, events, num, true);
3031 }
3032
3033 static void
3034 dlb2_event_release(struct dlb2_eventdev *dlb2,
3035                    uint8_t port_id,
3036                    int n)
3037 {
3038         struct process_local_port_data *port_data;
3039         struct dlb2_eventdev_port *ev_port;
3040         struct dlb2_port *qm_port;
3041         int i;
3042
3043         if (port_id > dlb2->num_ports) {
3044                 DLB2_LOG_ERR("Invalid port id %d in dlb2-event_release\n",
3045                              port_id);
3046                 rte_errno = -EINVAL;
3047                 return;
3048         }
3049
3050         ev_port = &dlb2->ev_ports[port_id];
3051         qm_port = &ev_port->qm_port;
3052         port_data = &dlb2_port[qm_port->id][PORT_TYPE(qm_port)];
3053
3054         i = 0;
3055
3056         if (qm_port->is_directed) {
3057                 i = n;
3058                 goto sw_credit_update;
3059         }
3060
3061         while (i < n) {
3062                 int pop_offs = 0;
3063                 int j = 0;
3064
3065                 /* Zero-out QEs */
3066                 _mm_storeu_si128((void *)&qm_port->qe4[0], _mm_setzero_si128());
3067                 _mm_storeu_si128((void *)&qm_port->qe4[1], _mm_setzero_si128());
3068                 _mm_storeu_si128((void *)&qm_port->qe4[2], _mm_setzero_si128());
3069                 _mm_storeu_si128((void *)&qm_port->qe4[3], _mm_setzero_si128());
3070
3071
3072                 for (; j < DLB2_NUM_QES_PER_CACHE_LINE && (i + j) < n; j++) {
3073                         int16_t thresh = qm_port->token_pop_thresh;
3074
3075                         if (qm_port->token_pop_mode == DELAYED_POP &&
3076                             qm_port->issued_releases >= thresh - 1) {
3077                                 /* Insert the token pop QE */
3078                                 dlb2_construct_token_pop_qe(qm_port, j);
3079
3080                                 /* Reset the releases for the next QE batch */
3081                                 qm_port->issued_releases -= thresh;
3082
3083                                 pop_offs = 1;
3084                                 j++;
3085                                 break;
3086                         }
3087
3088                         qm_port->qe4[j].cmd_byte = DLB2_COMP_CMD_BYTE;
3089                         qm_port->issued_releases++;
3090                 }
3091
3092                 dlb2_hw_do_enqueue(qm_port, i == 0, port_data);
3093
3094                 /* Don't include the token pop QE in the release count */
3095                 i += j - pop_offs;
3096         }
3097
3098 sw_credit_update:
3099         /* each release returns one credit */
3100         if (unlikely(!ev_port->outstanding_releases)) {
3101                 DLB2_LOG_ERR("%s: Outstanding releases underflowed.\n",
3102                              __func__);
3103                 return;
3104         }
3105         ev_port->outstanding_releases -= i;
3106         ev_port->inflight_credits += i;
3107
3108         /* Replenish s/w credits if enough releases are performed */
3109         dlb2_replenish_sw_credits(dlb2, ev_port);
3110 }
3111
3112 static inline void
3113 dlb2_port_credits_inc(struct dlb2_port *qm_port, int num)
3114 {
3115         uint32_t batch_size = DLB2_SW_CREDIT_BATCH_SZ;
3116
3117         /* increment port credits, and return to pool if exceeds threshold */
3118         if (!qm_port->is_directed) {
3119                 if (qm_port->dlb2->version == DLB2_HW_V2) {
3120                         qm_port->cached_ldb_credits += num;
3121                         if (qm_port->cached_ldb_credits >= 2 * batch_size) {
3122                                 __atomic_fetch_add(
3123                                         qm_port->credit_pool[DLB2_LDB_QUEUE],
3124                                         batch_size, __ATOMIC_SEQ_CST);
3125                                 qm_port->cached_ldb_credits -= batch_size;
3126                         }
3127                 } else {
3128                         qm_port->cached_credits += num;
3129                         if (qm_port->cached_credits >= 2 * batch_size) {
3130                                 __atomic_fetch_add(
3131                                       qm_port->credit_pool[DLB2_COMBINED_POOL],
3132                                       batch_size, __ATOMIC_SEQ_CST);
3133                                 qm_port->cached_credits -= batch_size;
3134                         }
3135                 }
3136         } else {
3137                 if (qm_port->dlb2->version == DLB2_HW_V2) {
3138                         qm_port->cached_dir_credits += num;
3139                         if (qm_port->cached_dir_credits >= 2 * batch_size) {
3140                                 __atomic_fetch_add(
3141                                         qm_port->credit_pool[DLB2_DIR_QUEUE],
3142                                         batch_size, __ATOMIC_SEQ_CST);
3143                                 qm_port->cached_dir_credits -= batch_size;
3144                         }
3145                 } else {
3146                         qm_port->cached_credits += num;
3147                         if (qm_port->cached_credits >= 2 * batch_size) {
3148                                 __atomic_fetch_add(
3149                                       qm_port->credit_pool[DLB2_COMBINED_POOL],
3150                                       batch_size, __ATOMIC_SEQ_CST);
3151                                 qm_port->cached_credits -= batch_size;
3152                         }
3153                 }
3154         }
3155 }
3156
3157 static inline int
3158 dlb2_dequeue_wait(struct dlb2_eventdev *dlb2,
3159                   struct dlb2_eventdev_port *ev_port,
3160                   struct dlb2_port *qm_port,
3161                   uint64_t timeout,
3162                   uint64_t start_ticks)
3163 {
3164         struct process_local_port_data *port_data;
3165         uint64_t elapsed_ticks;
3166
3167         port_data = &dlb2_port[qm_port->id][PORT_TYPE(qm_port)];
3168
3169         elapsed_ticks = rte_get_timer_cycles() - start_ticks;
3170
3171         /* Wait/poll time expired */
3172         if (elapsed_ticks >= timeout) {
3173                 return 1;
3174         } else if (dlb2->umwait_allowed) {
3175                 struct rte_power_monitor_cond pmc;
3176                 volatile struct dlb2_dequeue_qe *cq_base;
3177                 union {
3178                         uint64_t raw_qe[2];
3179                         struct dlb2_dequeue_qe qe;
3180                 } qe_mask;
3181                 uint64_t expected_value;
3182                 volatile uint64_t *monitor_addr;
3183
3184                 qe_mask.qe.cq_gen = 1; /* set mask */
3185
3186                 cq_base = port_data->cq_base;
3187                 monitor_addr = (volatile uint64_t *)(volatile void *)
3188                         &cq_base[qm_port->cq_idx];
3189                 monitor_addr++; /* cq_gen bit is in second 64bit location */
3190
3191                 if (qm_port->gen_bit)
3192                         expected_value = qe_mask.raw_qe[1];
3193                 else
3194                         expected_value = 0;
3195
3196                 pmc.addr = monitor_addr;
3197                 pmc.val = expected_value;
3198                 pmc.mask = qe_mask.raw_qe[1];
3199                 pmc.size = sizeof(uint64_t);
3200
3201                 rte_power_monitor(&pmc, timeout + start_ticks);
3202
3203                 DLB2_INC_STAT(ev_port->stats.traffic.rx_umonitor_umwait, 1);
3204         } else {
3205                 uint64_t poll_interval = dlb2->poll_interval;
3206                 uint64_t curr_ticks = rte_get_timer_cycles();
3207                 uint64_t init_ticks = curr_ticks;
3208
3209                 while ((curr_ticks - start_ticks < timeout) &&
3210                        (curr_ticks - init_ticks < poll_interval))
3211                         curr_ticks = rte_get_timer_cycles();
3212         }
3213
3214         return 0;
3215 }
3216
3217 static __rte_noinline int
3218 dlb2_process_dequeue_qes(struct dlb2_eventdev_port *ev_port,
3219                          struct dlb2_port *qm_port,
3220                          struct rte_event *events,
3221                          struct dlb2_dequeue_qe *qes,
3222                          int cnt)
3223 {
3224         uint8_t *qid_mappings = qm_port->qid_mappings;
3225         int i, num, evq_id;
3226
3227         for (i = 0, num = 0; i < cnt; i++) {
3228                 struct dlb2_dequeue_qe *qe = &qes[i];
3229                 int sched_type_map[DLB2_NUM_HW_SCHED_TYPES] = {
3230                         [DLB2_SCHED_ATOMIC] = RTE_SCHED_TYPE_ATOMIC,
3231                         [DLB2_SCHED_UNORDERED] = RTE_SCHED_TYPE_PARALLEL,
3232                         [DLB2_SCHED_ORDERED] = RTE_SCHED_TYPE_ORDERED,
3233                         [DLB2_SCHED_DIRECTED] = RTE_SCHED_TYPE_ATOMIC,
3234                 };
3235
3236                 /* Fill in event information.
3237                  * Note that flow_id must be embedded in the data by
3238                  * the app, such as the mbuf RSS hash field if the data
3239                  * buffer is a mbuf.
3240                  */
3241                 if (unlikely(qe->error)) {
3242                         DLB2_LOG_ERR("QE error bit ON\n");
3243                         DLB2_INC_STAT(ev_port->stats.traffic.rx_drop, 1);
3244                         dlb2_consume_qe_immediate(qm_port, 1);
3245                         continue; /* Ignore */
3246                 }
3247
3248                 events[num].u64 = qe->data;
3249                 events[num].flow_id = qe->flow_id;
3250                 events[num].priority = DLB2_TO_EV_PRIO((uint8_t)qe->priority);
3251                 events[num].event_type = qe->u.event_type.major;
3252                 events[num].sub_event_type = qe->u.event_type.sub;
3253                 events[num].sched_type = sched_type_map[qe->sched_type];
3254                 events[num].impl_opaque = qe->qid_depth;
3255
3256                 /* qid not preserved for directed queues */
3257                 if (qm_port->is_directed)
3258                         evq_id = ev_port->link[0].queue_id;
3259                 else
3260                         evq_id = qid_mappings[qe->qid];
3261
3262                 events[num].queue_id = evq_id;
3263                 DLB2_INC_STAT(
3264                         ev_port->stats.queue[evq_id].qid_depth[qe->qid_depth],
3265                         1);
3266                 DLB2_INC_STAT(ev_port->stats.rx_sched_cnt[qe->sched_type], 1);
3267                 num++;
3268         }
3269
3270         DLB2_INC_STAT(ev_port->stats.traffic.rx_ok, num);
3271
3272         return num;
3273 }
3274
3275 static inline int
3276 dlb2_process_dequeue_four_qes(struct dlb2_eventdev_port *ev_port,
3277                               struct dlb2_port *qm_port,
3278                               struct rte_event *events,
3279                               struct dlb2_dequeue_qe *qes)
3280 {
3281         int sched_type_map[] = {
3282                 [DLB2_SCHED_ATOMIC] = RTE_SCHED_TYPE_ATOMIC,
3283                 [DLB2_SCHED_UNORDERED] = RTE_SCHED_TYPE_PARALLEL,
3284                 [DLB2_SCHED_ORDERED] = RTE_SCHED_TYPE_ORDERED,
3285                 [DLB2_SCHED_DIRECTED] = RTE_SCHED_TYPE_ATOMIC,
3286         };
3287         const int num_events = DLB2_NUM_QES_PER_CACHE_LINE;
3288         uint8_t *qid_mappings = qm_port->qid_mappings;
3289         __m128i sse_evt[2];
3290
3291         /* In the unlikely case that any of the QE error bits are set, process
3292          * them one at a time.
3293          */
3294         if (unlikely(qes[0].error || qes[1].error ||
3295                      qes[2].error || qes[3].error))
3296                 return dlb2_process_dequeue_qes(ev_port, qm_port, events,
3297                                                  qes, num_events);
3298
3299         events[0].u64 = qes[0].data;
3300         events[1].u64 = qes[1].data;
3301         events[2].u64 = qes[2].data;
3302         events[3].u64 = qes[3].data;
3303
3304         /* Construct the metadata portion of two struct rte_events
3305          * in one 128b SSE register. Event metadata is constructed in the SSE
3306          * registers like so:
3307          * sse_evt[0][63:0]:   event[0]'s metadata
3308          * sse_evt[0][127:64]: event[1]'s metadata
3309          * sse_evt[1][63:0]:   event[2]'s metadata
3310          * sse_evt[1][127:64]: event[3]'s metadata
3311          */
3312         sse_evt[0] = _mm_setzero_si128();
3313         sse_evt[1] = _mm_setzero_si128();
3314
3315         /* Convert the hardware queue ID to an event queue ID and store it in
3316          * the metadata:
3317          * sse_evt[0][47:40]   = qid_mappings[qes[0].qid]
3318          * sse_evt[0][111:104] = qid_mappings[qes[1].qid]
3319          * sse_evt[1][47:40]   = qid_mappings[qes[2].qid]
3320          * sse_evt[1][111:104] = qid_mappings[qes[3].qid]
3321          */
3322 #define DLB_EVENT_QUEUE_ID_BYTE 5
3323         sse_evt[0] = _mm_insert_epi8(sse_evt[0],
3324                                      qid_mappings[qes[0].qid],
3325                                      DLB_EVENT_QUEUE_ID_BYTE);
3326         sse_evt[0] = _mm_insert_epi8(sse_evt[0],
3327                                      qid_mappings[qes[1].qid],
3328                                      DLB_EVENT_QUEUE_ID_BYTE + 8);
3329         sse_evt[1] = _mm_insert_epi8(sse_evt[1],
3330                                      qid_mappings[qes[2].qid],
3331                                      DLB_EVENT_QUEUE_ID_BYTE);
3332         sse_evt[1] = _mm_insert_epi8(sse_evt[1],
3333                                      qid_mappings[qes[3].qid],
3334                                      DLB_EVENT_QUEUE_ID_BYTE + 8);
3335
3336         /* Convert the hardware priority to an event priority and store it in
3337          * the metadata, while also returning the queue depth status
3338          * value captured by the hardware, storing it in impl_opaque, which can
3339          * be read by the application but not modified
3340          * sse_evt[0][55:48]   = DLB2_TO_EV_PRIO(qes[0].priority)
3341          * sse_evt[0][63:56]   = qes[0].qid_depth
3342          * sse_evt[0][119:112] = DLB2_TO_EV_PRIO(qes[1].priority)
3343          * sse_evt[0][127:120] = qes[1].qid_depth
3344          * sse_evt[1][55:48]   = DLB2_TO_EV_PRIO(qes[2].priority)
3345          * sse_evt[1][63:56]   = qes[2].qid_depth
3346          * sse_evt[1][119:112] = DLB2_TO_EV_PRIO(qes[3].priority)
3347          * sse_evt[1][127:120] = qes[3].qid_depth
3348          */
3349 #define DLB_EVENT_PRIO_IMPL_OPAQUE_WORD 3
3350 #define DLB_BYTE_SHIFT 8
3351         sse_evt[0] =
3352                 _mm_insert_epi16(sse_evt[0],
3353                         DLB2_TO_EV_PRIO((uint8_t)qes[0].priority) |
3354                         (qes[0].qid_depth << DLB_BYTE_SHIFT),
3355                         DLB_EVENT_PRIO_IMPL_OPAQUE_WORD);
3356         sse_evt[0] =
3357                 _mm_insert_epi16(sse_evt[0],
3358                         DLB2_TO_EV_PRIO((uint8_t)qes[1].priority) |
3359                         (qes[1].qid_depth << DLB_BYTE_SHIFT),
3360                         DLB_EVENT_PRIO_IMPL_OPAQUE_WORD + 4);
3361         sse_evt[1] =
3362                 _mm_insert_epi16(sse_evt[1],
3363                         DLB2_TO_EV_PRIO((uint8_t)qes[2].priority) |
3364                         (qes[2].qid_depth << DLB_BYTE_SHIFT),
3365                         DLB_EVENT_PRIO_IMPL_OPAQUE_WORD);
3366         sse_evt[1] =
3367                 _mm_insert_epi16(sse_evt[1],
3368                         DLB2_TO_EV_PRIO((uint8_t)qes[3].priority) |
3369                         (qes[3].qid_depth << DLB_BYTE_SHIFT),
3370                         DLB_EVENT_PRIO_IMPL_OPAQUE_WORD + 4);
3371
3372         /* Write the event type, sub event type, and flow_id to the event
3373          * metadata.
3374          * sse_evt[0][31:0]   = qes[0].flow_id |
3375          *                      qes[0].u.event_type.major << 28 |
3376          *                      qes[0].u.event_type.sub << 20;
3377          * sse_evt[0][95:64]  = qes[1].flow_id |
3378          *                      qes[1].u.event_type.major << 28 |
3379          *                      qes[1].u.event_type.sub << 20;
3380          * sse_evt[1][31:0]   = qes[2].flow_id |
3381          *                      qes[2].u.event_type.major << 28 |
3382          *                      qes[2].u.event_type.sub << 20;
3383          * sse_evt[1][95:64]  = qes[3].flow_id |
3384          *                      qes[3].u.event_type.major << 28 |
3385          *                      qes[3].u.event_type.sub << 20;
3386          */
3387 #define DLB_EVENT_EV_TYPE_DW 0
3388 #define DLB_EVENT_EV_TYPE_SHIFT 28
3389 #define DLB_EVENT_SUB_EV_TYPE_SHIFT 20
3390         sse_evt[0] = _mm_insert_epi32(sse_evt[0],
3391                         qes[0].flow_id |
3392                         qes[0].u.event_type.major << DLB_EVENT_EV_TYPE_SHIFT |
3393                         qes[0].u.event_type.sub <<  DLB_EVENT_SUB_EV_TYPE_SHIFT,
3394                         DLB_EVENT_EV_TYPE_DW);
3395         sse_evt[0] = _mm_insert_epi32(sse_evt[0],
3396                         qes[1].flow_id |
3397                         qes[1].u.event_type.major << DLB_EVENT_EV_TYPE_SHIFT |
3398                         qes[1].u.event_type.sub <<  DLB_EVENT_SUB_EV_TYPE_SHIFT,
3399                         DLB_EVENT_EV_TYPE_DW + 2);
3400         sse_evt[1] = _mm_insert_epi32(sse_evt[1],
3401                         qes[2].flow_id |
3402                         qes[2].u.event_type.major << DLB_EVENT_EV_TYPE_SHIFT |
3403                         qes[2].u.event_type.sub <<  DLB_EVENT_SUB_EV_TYPE_SHIFT,
3404                         DLB_EVENT_EV_TYPE_DW);
3405         sse_evt[1] = _mm_insert_epi32(sse_evt[1],
3406                         qes[3].flow_id |
3407                         qes[3].u.event_type.major << DLB_EVENT_EV_TYPE_SHIFT  |
3408                         qes[3].u.event_type.sub << DLB_EVENT_SUB_EV_TYPE_SHIFT,
3409                         DLB_EVENT_EV_TYPE_DW + 2);
3410
3411         /* Write the sched type to the event metadata. 'op' and 'rsvd' are not
3412          * set:
3413          * sse_evt[0][39:32]  = sched_type_map[qes[0].sched_type] << 6
3414          * sse_evt[0][103:96] = sched_type_map[qes[1].sched_type] << 6
3415          * sse_evt[1][39:32]  = sched_type_map[qes[2].sched_type] << 6
3416          * sse_evt[1][103:96] = sched_type_map[qes[3].sched_type] << 6
3417          */
3418 #define DLB_EVENT_SCHED_TYPE_BYTE 4
3419 #define DLB_EVENT_SCHED_TYPE_SHIFT 6
3420         sse_evt[0] = _mm_insert_epi8(sse_evt[0],
3421                 sched_type_map[qes[0].sched_type] << DLB_EVENT_SCHED_TYPE_SHIFT,
3422                 DLB_EVENT_SCHED_TYPE_BYTE);
3423         sse_evt[0] = _mm_insert_epi8(sse_evt[0],
3424                 sched_type_map[qes[1].sched_type] << DLB_EVENT_SCHED_TYPE_SHIFT,
3425                 DLB_EVENT_SCHED_TYPE_BYTE + 8);
3426         sse_evt[1] = _mm_insert_epi8(sse_evt[1],
3427                 sched_type_map[qes[2].sched_type] << DLB_EVENT_SCHED_TYPE_SHIFT,
3428                 DLB_EVENT_SCHED_TYPE_BYTE);
3429         sse_evt[1] = _mm_insert_epi8(sse_evt[1],
3430                 sched_type_map[qes[3].sched_type] << DLB_EVENT_SCHED_TYPE_SHIFT,
3431                 DLB_EVENT_SCHED_TYPE_BYTE + 8);
3432
3433         /* Store the metadata to the event (use the double-precision
3434          * _mm_storeh_pd because there is no integer function for storing the
3435          * upper 64b):
3436          * events[0].event = sse_evt[0][63:0]
3437          * events[1].event = sse_evt[0][127:64]
3438          * events[2].event = sse_evt[1][63:0]
3439          * events[3].event = sse_evt[1][127:64]
3440          */
3441         _mm_storel_epi64((__m128i *)&events[0].event, sse_evt[0]);
3442         _mm_storeh_pd((double *)&events[1].event, (__m128d) sse_evt[0]);
3443         _mm_storel_epi64((__m128i *)&events[2].event, sse_evt[1]);
3444         _mm_storeh_pd((double *)&events[3].event, (__m128d) sse_evt[1]);
3445
3446         DLB2_INC_STAT(ev_port->stats.rx_sched_cnt[qes[0].sched_type], 1);
3447         DLB2_INC_STAT(ev_port->stats.rx_sched_cnt[qes[1].sched_type], 1);
3448         DLB2_INC_STAT(ev_port->stats.rx_sched_cnt[qes[2].sched_type], 1);
3449         DLB2_INC_STAT(ev_port->stats.rx_sched_cnt[qes[3].sched_type], 1);
3450
3451         DLB2_INC_STAT(
3452                 ev_port->stats.queue[events[0].queue_id].
3453                         qid_depth[qes[0].qid_depth],
3454                 1);
3455         DLB2_INC_STAT(
3456                 ev_port->stats.queue[events[1].queue_id].
3457                         qid_depth[qes[1].qid_depth],
3458                 1);
3459         DLB2_INC_STAT(
3460                 ev_port->stats.queue[events[2].queue_id].
3461                         qid_depth[qes[2].qid_depth],
3462                 1);
3463         DLB2_INC_STAT(
3464                 ev_port->stats.queue[events[3].queue_id].
3465                         qid_depth[qes[3].qid_depth],
3466                 1);
3467
3468         DLB2_INC_STAT(ev_port->stats.traffic.rx_ok, num_events);
3469
3470         return num_events;
3471 }
3472
3473 static __rte_always_inline int
3474 dlb2_recv_qe_sparse(struct dlb2_port *qm_port, struct dlb2_dequeue_qe *qe)
3475 {
3476         volatile struct dlb2_dequeue_qe *cq_addr;
3477         uint8_t xor_mask[2] = {0x0F, 0x00};
3478         const uint8_t and_mask = 0x0F;
3479         __m128i *qes = (__m128i *)qe;
3480         uint8_t gen_bits, gen_bit;
3481         uintptr_t addr[4];
3482         uint16_t idx;
3483
3484         cq_addr = dlb2_port[qm_port->id][PORT_TYPE(qm_port)].cq_base;
3485
3486         idx = qm_port->cq_idx_unmasked & qm_port->cq_depth_mask;
3487         /* Load the next 4 QEs */
3488         addr[0] = (uintptr_t)&cq_addr[idx];
3489         addr[1] = (uintptr_t)&cq_addr[(idx +  4) & qm_port->cq_depth_mask];
3490         addr[2] = (uintptr_t)&cq_addr[(idx +  8) & qm_port->cq_depth_mask];
3491         addr[3] = (uintptr_t)&cq_addr[(idx + 12) & qm_port->cq_depth_mask];
3492
3493         /* Prefetch next batch of QEs (all CQs occupy minimum 8 cache lines) */
3494         rte_prefetch0(&cq_addr[(idx + 16) & qm_port->cq_depth_mask]);
3495         rte_prefetch0(&cq_addr[(idx + 20) & qm_port->cq_depth_mask]);
3496         rte_prefetch0(&cq_addr[(idx + 24) & qm_port->cq_depth_mask]);
3497         rte_prefetch0(&cq_addr[(idx + 28) & qm_port->cq_depth_mask]);
3498
3499         /* Correct the xor_mask for wrap-around QEs */
3500         gen_bit = qm_port->gen_bit;
3501         xor_mask[gen_bit] ^= !!((idx +  4) > qm_port->cq_depth_mask) << 1;
3502         xor_mask[gen_bit] ^= !!((idx +  8) > qm_port->cq_depth_mask) << 2;
3503         xor_mask[gen_bit] ^= !!((idx + 12) > qm_port->cq_depth_mask) << 3;
3504
3505         /* Read the cache lines backwards to ensure that if QE[N] (N > 0) is
3506          * valid, then QEs[0:N-1] are too.
3507          */
3508         qes[3] = _mm_load_si128((__m128i *)(void *)addr[3]);
3509         rte_compiler_barrier();
3510         qes[2] = _mm_load_si128((__m128i *)(void *)addr[2]);
3511         rte_compiler_barrier();
3512         qes[1] = _mm_load_si128((__m128i *)(void *)addr[1]);
3513         rte_compiler_barrier();
3514         qes[0] = _mm_load_si128((__m128i *)(void *)addr[0]);
3515
3516         /* Extract and combine the gen bits */
3517         gen_bits = ((_mm_extract_epi8(qes[0], 15) & 0x1) << 0) |
3518                    ((_mm_extract_epi8(qes[1], 15) & 0x1) << 1) |
3519                    ((_mm_extract_epi8(qes[2], 15) & 0x1) << 2) |
3520                    ((_mm_extract_epi8(qes[3], 15) & 0x1) << 3);
3521
3522         /* XOR the combined bits such that a 1 represents a valid QE */
3523         gen_bits ^= xor_mask[gen_bit];
3524
3525         /* Mask off gen bits we don't care about */
3526         gen_bits &= and_mask;
3527
3528         return __builtin_popcount(gen_bits);
3529 }
3530
3531 static inline void
3532 _process_deq_qes_vec_impl(struct dlb2_port *qm_port,
3533                           struct rte_event *events,
3534                           __m128i v_qe_3,
3535                           __m128i v_qe_2,
3536                           __m128i v_qe_1,
3537                           __m128i v_qe_0,
3538                           __m128i v_qe_meta,
3539                           __m128i v_qe_status,
3540                           uint32_t valid_events)
3541 {
3542         /* Look up the event QIDs, using the hardware QIDs to index the
3543          * port's QID mapping.
3544          *
3545          * Each v_qe_[0-4] is just a 16-byte load of the whole QE. It is
3546          * passed along in registers as the QE data is required later.
3547          *
3548          * v_qe_meta is an u32 unpack of all 4x QEs. A.k.a, it contains one
3549          * 32-bit slice of each QE, so makes up a full SSE register. This
3550          * allows parallel processing of 4x QEs in a single register.
3551          */
3552
3553         __m128i v_qid_done = {0};
3554         int hw_qid0 = _mm_extract_epi8(v_qe_meta, 2);
3555         int hw_qid1 = _mm_extract_epi8(v_qe_meta, 6);
3556         int hw_qid2 = _mm_extract_epi8(v_qe_meta, 10);
3557         int hw_qid3 = _mm_extract_epi8(v_qe_meta, 14);
3558
3559         int ev_qid0 = qm_port->qid_mappings[hw_qid0];
3560         int ev_qid1 = qm_port->qid_mappings[hw_qid1];
3561         int ev_qid2 = qm_port->qid_mappings[hw_qid2];
3562         int ev_qid3 = qm_port->qid_mappings[hw_qid3];
3563
3564         v_qid_done = _mm_insert_epi8(v_qid_done, ev_qid0, 2);
3565         v_qid_done = _mm_insert_epi8(v_qid_done, ev_qid1, 6);
3566         v_qid_done = _mm_insert_epi8(v_qid_done, ev_qid2, 10);
3567         v_qid_done = _mm_insert_epi8(v_qid_done, ev_qid3, 14);
3568
3569         /* Schedule field remapping using byte shuffle
3570          * - Full byte containing sched field handled here (op, rsvd are zero)
3571          * - Note sanitizing the register requires two masking ANDs:
3572          *   1) to strip prio/msg_type from byte for correct shuffle lookup
3573          *   2) to strip any non-sched-field lanes from any results to OR later
3574          * - Final byte result is >> 10 to another byte-lane inside the u32.
3575          *   This makes the final combination OR easier to make the rte_event.
3576          */
3577         __m128i v_sched_done;
3578         __m128i v_sched_bits;
3579         {
3580                 static const uint8_t sched_type_map[16] = {
3581                         [DLB2_SCHED_ATOMIC] = RTE_SCHED_TYPE_ATOMIC,
3582                         [DLB2_SCHED_UNORDERED] = RTE_SCHED_TYPE_PARALLEL,
3583                         [DLB2_SCHED_ORDERED] = RTE_SCHED_TYPE_ORDERED,
3584                         [DLB2_SCHED_DIRECTED] = RTE_SCHED_TYPE_ATOMIC,
3585                 };
3586                 static const uint8_t sched_and_mask[16] = {
3587                         0x00, 0x00, 0x00, 0x03,
3588                         0x00, 0x00, 0x00, 0x03,
3589                         0x00, 0x00, 0x00, 0x03,
3590                         0x00, 0x00, 0x00, 0x03,
3591                 };
3592                 const __m128i v_sched_map = _mm_loadu_si128(
3593                                              (const __m128i *)sched_type_map);
3594                 __m128i v_sched_mask = _mm_loadu_si128(
3595                                              (const __m128i *)&sched_and_mask);
3596                 v_sched_bits = _mm_and_si128(v_qe_meta, v_sched_mask);
3597                 __m128i v_sched_remapped = _mm_shuffle_epi8(v_sched_map,
3598                                                             v_sched_bits);
3599                 __m128i v_preshift = _mm_and_si128(v_sched_remapped,
3600                                                    v_sched_mask);
3601                 v_sched_done = _mm_srli_epi32(v_preshift, 10);
3602         }
3603
3604         /* Priority handling
3605          * - QE provides 3 bits of priority
3606          * - Shift << 3 to move to MSBs for byte-prio in rte_event
3607          * - Mask bits to avoid pollution, leaving only 3 prio MSBs in reg
3608          */
3609         __m128i v_prio_done;
3610         {
3611                 static const uint8_t prio_mask[16] = {
3612                         0x00, 0x00, 0x00, 0x07 << 5,
3613                         0x00, 0x00, 0x00, 0x07 << 5,
3614                         0x00, 0x00, 0x00, 0x07 << 5,
3615                         0x00, 0x00, 0x00, 0x07 << 5,
3616                 };
3617                 __m128i v_prio_mask  = _mm_loadu_si128(
3618                                                 (const __m128i *)prio_mask);
3619                 __m128i v_prio_shifted = _mm_slli_epi32(v_qe_meta, 3);
3620                 v_prio_done = _mm_and_si128(v_prio_shifted, v_prio_mask);
3621         }
3622
3623         /* Event Sub/Type handling:
3624          * we want to keep the lower 12 bits of each QE. Shift up by 20 bits
3625          * to get the sub/ev type data into rte_event location, clearing the
3626          * lower 20 bits in the process.
3627          */
3628         __m128i v_types_done;
3629         {
3630                 static const uint8_t event_mask[16] = {
3631                         0x0f, 0x00, 0x00, 0x00,
3632                         0x0f, 0x00, 0x00, 0x00,
3633                         0x0f, 0x00, 0x00, 0x00,
3634                         0x0f, 0x00, 0x00, 0x00,
3635                 };
3636                 static const uint8_t sub_event_mask[16] = {
3637                         0xff, 0x00, 0x00, 0x00,
3638                         0xff, 0x00, 0x00, 0x00,
3639                         0xff, 0x00, 0x00, 0x00,
3640                         0xff, 0x00, 0x00, 0x00,
3641                 };
3642                 static const uint8_t flow_mask[16] = {
3643                         0xff, 0xff, 0x00, 0x00,
3644                         0xff, 0xff, 0x00, 0x00,
3645                         0xff, 0xff, 0x00, 0x00,
3646                         0xff, 0xff, 0x00, 0x00,
3647                 };
3648                 __m128i v_event_mask  = _mm_loadu_si128(
3649                                         (const __m128i *)event_mask);
3650                 __m128i v_sub_event_mask  = _mm_loadu_si128(
3651                                         (const __m128i *)sub_event_mask);
3652                 __m128i v_flow_mask  = _mm_loadu_si128(
3653                                        (const __m128i *)flow_mask);
3654                 __m128i v_sub = _mm_srli_epi32(v_qe_meta, 8);
3655                 v_sub = _mm_and_si128(v_sub, v_sub_event_mask);
3656                 __m128i v_type = _mm_and_si128(v_qe_meta, v_event_mask);
3657                 v_type = _mm_slli_epi32(v_type, 8);
3658                 v_types_done = _mm_or_si128(v_type, v_sub);
3659                 v_types_done = _mm_slli_epi32(v_types_done, 20);
3660                 __m128i v_flow = _mm_and_si128(v_qe_status, v_flow_mask);
3661                 v_types_done = _mm_or_si128(v_types_done, v_flow);
3662         }
3663
3664         /* Combine QID, Sched and Prio fields, then Shift >> 8 bits to align
3665          * with the rte_event, allowing unpacks to move/blend with payload.
3666          */
3667         __m128i v_q_s_p_done;
3668         {
3669                 __m128i v_qid_sched = _mm_or_si128(v_qid_done, v_sched_done);
3670                 __m128i v_q_s_prio = _mm_or_si128(v_qid_sched, v_prio_done);
3671                 v_q_s_p_done = _mm_srli_epi32(v_q_s_prio, 8);
3672         }
3673
3674         __m128i v_unpk_ev_23, v_unpk_ev_01, v_ev_2, v_ev_3, v_ev_0, v_ev_1;
3675
3676         /* Unpack evs into u64 metadata, then indiv events */
3677         v_unpk_ev_23 = _mm_unpackhi_epi32(v_types_done, v_q_s_p_done);
3678         v_unpk_ev_01 = _mm_unpacklo_epi32(v_types_done, v_q_s_p_done);
3679
3680         switch (valid_events) {
3681         case 4:
3682                 v_ev_3 = _mm_blend_epi16(v_unpk_ev_23, v_qe_3, 0x0F);
3683                 v_ev_3 = _mm_alignr_epi8(v_ev_3, v_ev_3, 8);
3684                 _mm_storeu_si128((__m128i *)&events[3], v_ev_3);
3685                 /* fallthrough */
3686         case 3:
3687                 v_ev_2 = _mm_unpacklo_epi64(v_unpk_ev_23, v_qe_2);
3688                 _mm_storeu_si128((__m128i *)&events[2], v_ev_2);
3689                 /* fallthrough */
3690         case 2:
3691                 v_ev_1 = _mm_blend_epi16(v_unpk_ev_01, v_qe_1, 0x0F);
3692                 v_ev_1 = _mm_alignr_epi8(v_ev_1, v_ev_1, 8);
3693                 _mm_storeu_si128((__m128i *)&events[1], v_ev_1);
3694                 /* fallthrough */
3695         case 1:
3696                 v_ev_0 = _mm_unpacklo_epi64(v_unpk_ev_01, v_qe_0);
3697                 _mm_storeu_si128((__m128i *)&events[0], v_ev_0);
3698         }
3699 }
3700
3701 static __rte_always_inline int
3702 dlb2_recv_qe_sparse_vec(struct dlb2_port *qm_port, void *events,
3703                         uint32_t max_events)
3704 {
3705         /* Using unmasked idx for perf, and masking manually */
3706         uint16_t idx = qm_port->cq_idx_unmasked;
3707         volatile struct dlb2_dequeue_qe *cq_addr;
3708
3709         cq_addr = dlb2_port[qm_port->id][PORT_TYPE(qm_port)].cq_base;
3710
3711         uintptr_t qe_ptr_3 = (uintptr_t)&cq_addr[(idx + 12) &
3712                                                  qm_port->cq_depth_mask];
3713         uintptr_t qe_ptr_2 = (uintptr_t)&cq_addr[(idx +  8) &
3714                                                  qm_port->cq_depth_mask];
3715         uintptr_t qe_ptr_1 = (uintptr_t)&cq_addr[(idx +  4) &
3716                                                  qm_port->cq_depth_mask];
3717         uintptr_t qe_ptr_0 = (uintptr_t)&cq_addr[(idx +  0) &
3718                                                  qm_port->cq_depth_mask];
3719
3720         /* Load QEs from CQ: use compiler barriers to avoid load reordering */
3721         __m128i v_qe_3 = _mm_loadu_si128((const __m128i *)qe_ptr_3);
3722         rte_compiler_barrier();
3723         __m128i v_qe_2 = _mm_loadu_si128((const __m128i *)qe_ptr_2);
3724         rte_compiler_barrier();
3725         __m128i v_qe_1 = _mm_loadu_si128((const __m128i *)qe_ptr_1);
3726         rte_compiler_barrier();
3727         __m128i v_qe_0 = _mm_loadu_si128((const __m128i *)qe_ptr_0);
3728
3729         /* Generate the pkt_shuffle mask;
3730          * - Avoids load in otherwise load-heavy section of code
3731          * - Moves bytes 3,7,11,15 (gen bit bytes) to LSB bytes in XMM
3732          */
3733         const uint32_t stat_shuf_bytes = (15 << 24) | (11 << 16) | (7 << 8) | 3;
3734         __m128i v_zeros = _mm_setzero_si128();
3735         __m128i v_ffff = _mm_cmpeq_epi8(v_zeros, v_zeros);
3736         __m128i v_stat_shuf_mask = _mm_insert_epi32(v_ffff, stat_shuf_bytes, 0);
3737
3738         /* Extract u32 components required from the QE
3739          * - QE[64 to 95 ] for metadata (qid, sched, prio, event type, ...)
3740          * - QE[96 to 127] for status (cq gen bit, error)
3741          *
3742          * Note that stage 1 of the unpacking is re-used for both u32 extracts
3743          */
3744         __m128i v_qe_02 = _mm_unpackhi_epi32(v_qe_0, v_qe_2);
3745         __m128i v_qe_13 = _mm_unpackhi_epi32(v_qe_1, v_qe_3);
3746         __m128i v_qe_status = _mm_unpackhi_epi32(v_qe_02, v_qe_13);
3747         __m128i v_qe_meta   = _mm_unpacklo_epi32(v_qe_02, v_qe_13);
3748
3749         /* Status byte (gen_bit, error) handling:
3750          * - Shuffle to lanes 0,1,2,3, clear all others
3751          * - Shift right by 7 for gen bit to MSB, movemask to scalar
3752          * - Shift right by 2 for error bit to MSB, movemask to scalar
3753          */
3754         __m128i v_qe_shuffled = _mm_shuffle_epi8(v_qe_status, v_stat_shuf_mask);
3755         __m128i v_qes_shift_gen_bit = _mm_slli_epi32(v_qe_shuffled, 7);
3756         int32_t qe_gen_bits = _mm_movemask_epi8(v_qes_shift_gen_bit) & 0xf;
3757
3758         /* Expected vs Reality of QE Gen bits
3759          * - cq_rolling_mask provides expected bits
3760          * - QE loads, unpacks/shuffle and movemask provides reality
3761          * - XOR of the two gives bitmask of new packets
3762          * - POPCNT to get the number of new events
3763          */
3764         uint64_t rolling = qm_port->cq_rolling_mask & 0xF;
3765         uint64_t qe_xor_bits = (qe_gen_bits ^ rolling);
3766         uint32_t count_new = __builtin_popcount(qe_xor_bits);
3767         count_new = RTE_MIN(count_new, max_events);
3768         if (!count_new)
3769                 return 0;
3770
3771         /* emulate a 128 bit rotate using 2x 64-bit numbers and bit-shifts */
3772
3773         uint64_t m_rshift = qm_port->cq_rolling_mask >> count_new;
3774         uint64_t m_lshift = qm_port->cq_rolling_mask << (64 - count_new);
3775         uint64_t m2_rshift = qm_port->cq_rolling_mask_2 >> count_new;
3776         uint64_t m2_lshift = qm_port->cq_rolling_mask_2 << (64 - count_new);
3777
3778         /* shifted out of m2 into MSB of m */
3779         qm_port->cq_rolling_mask = (m_rshift | m2_lshift);
3780
3781         /* shifted out of m "looped back" into MSB of m2 */
3782         qm_port->cq_rolling_mask_2 = (m2_rshift | m_lshift);
3783
3784         /* Prefetch the next QEs - should run as IPC instead of cycles */
3785         rte_prefetch0(&cq_addr[(idx + 16) & qm_port->cq_depth_mask]);
3786         rte_prefetch0(&cq_addr[(idx + 20) & qm_port->cq_depth_mask]);
3787         rte_prefetch0(&cq_addr[(idx + 24) & qm_port->cq_depth_mask]);
3788         rte_prefetch0(&cq_addr[(idx + 28) & qm_port->cq_depth_mask]);
3789
3790         /* Convert QEs from XMM regs to events and store events directly */
3791         _process_deq_qes_vec_impl(qm_port, events, v_qe_3, v_qe_2, v_qe_1,
3792                                   v_qe_0, v_qe_meta, v_qe_status, count_new);
3793
3794         return count_new;
3795 }
3796
3797 static inline void
3798 dlb2_inc_cq_idx(struct dlb2_port *qm_port, int cnt)
3799 {
3800         uint16_t idx = qm_port->cq_idx_unmasked + cnt;
3801
3802         qm_port->cq_idx_unmasked = idx;
3803         qm_port->cq_idx = idx & qm_port->cq_depth_mask;
3804         qm_port->gen_bit = (~(idx >> qm_port->gen_bit_shift)) & 0x1;
3805 }
3806
3807 static inline int16_t
3808 dlb2_hw_dequeue_sparse(struct dlb2_eventdev *dlb2,
3809                        struct dlb2_eventdev_port *ev_port,
3810                        struct rte_event *events,
3811                        uint16_t max_num,
3812                        uint64_t dequeue_timeout_ticks)
3813 {
3814         uint64_t start_ticks = 0ULL;
3815         struct dlb2_port *qm_port;
3816         int num = 0;
3817         bool use_scalar;
3818         uint64_t timeout;
3819
3820         qm_port = &ev_port->qm_port;
3821         use_scalar = qm_port->use_scalar;
3822
3823         if (!dlb2->global_dequeue_wait)
3824                 timeout = dequeue_timeout_ticks;
3825         else
3826                 timeout = dlb2->global_dequeue_wait_ticks;
3827
3828         start_ticks = rte_get_timer_cycles();
3829
3830         use_scalar = use_scalar || (max_num & 0x3);
3831
3832         while (num < max_num) {
3833                 struct dlb2_dequeue_qe qes[DLB2_NUM_QES_PER_CACHE_LINE];
3834                 int num_avail;
3835                 if (use_scalar) {
3836                         num_avail = dlb2_recv_qe_sparse(qm_port, qes);
3837                         num_avail = RTE_MIN(num_avail, max_num - num);
3838                         dlb2_inc_cq_idx(qm_port, num_avail << 2);
3839                         if (num_avail == DLB2_NUM_QES_PER_CACHE_LINE)
3840                                 num += dlb2_process_dequeue_four_qes(ev_port,
3841                                                                   qm_port,
3842                                                                   &events[num],
3843                                                                   &qes[0]);
3844                         else if (num_avail)
3845                                 num += dlb2_process_dequeue_qes(ev_port,
3846                                                                 qm_port,
3847                                                                 &events[num],
3848                                                                 &qes[0],
3849                                                                 num_avail);
3850                 } else { /* !use_scalar */
3851                         num_avail = dlb2_recv_qe_sparse_vec(qm_port,
3852                                                             &events[num],
3853                                                             max_num - num);
3854                         num += num_avail;
3855                         dlb2_inc_cq_idx(qm_port, num_avail << 2);
3856                         DLB2_INC_STAT(ev_port->stats.traffic.rx_ok, num_avail);
3857                 }
3858                 if (!num_avail) {
3859                         if (num > 0)
3860                                 break;
3861                         else if (dlb2_dequeue_wait(dlb2, ev_port, qm_port,
3862                                                    timeout, start_ticks))
3863                                 break;
3864                 }
3865         }
3866
3867         qm_port->owed_tokens += num;
3868
3869         if (num) {
3870                 if (qm_port->token_pop_mode == AUTO_POP)
3871                         dlb2_consume_qe_immediate(qm_port, num);
3872
3873                 ev_port->outstanding_releases += num;
3874
3875                 dlb2_port_credits_inc(qm_port, num);
3876         }
3877
3878         return num;
3879 }
3880
3881 static __rte_always_inline int
3882 dlb2_recv_qe(struct dlb2_port *qm_port, struct dlb2_dequeue_qe *qe,
3883              uint8_t *offset)
3884 {
3885         uint8_t xor_mask[2][4] = { {0x0F, 0x0E, 0x0C, 0x08},
3886                                    {0x00, 0x01, 0x03, 0x07} };
3887         uint8_t and_mask[4] = {0x0F, 0x0E, 0x0C, 0x08};
3888         volatile struct dlb2_dequeue_qe *cq_addr;
3889         __m128i *qes = (__m128i *)qe;
3890         uint64_t *cache_line_base;
3891         uint8_t gen_bits;
3892
3893         cq_addr = dlb2_port[qm_port->id][PORT_TYPE(qm_port)].cq_base;
3894         cq_addr = &cq_addr[qm_port->cq_idx];
3895
3896         cache_line_base = (void *)(((uintptr_t)cq_addr) & ~0x3F);
3897         *offset = ((uintptr_t)cq_addr & 0x30) >> 4;
3898
3899         /* Load the next CQ cache line from memory. Pack these reads as tight
3900          * as possible to reduce the chance that DLB invalidates the line while
3901          * the CPU is reading it. Read the cache line backwards to ensure that
3902          * if QE[N] (N > 0) is valid, then QEs[0:N-1] are too.
3903          *
3904          * (Valid QEs start at &qe[offset])
3905          */
3906         qes[3] = _mm_load_si128((__m128i *)&cache_line_base[6]);
3907         qes[2] = _mm_load_si128((__m128i *)&cache_line_base[4]);
3908         qes[1] = _mm_load_si128((__m128i *)&cache_line_base[2]);
3909         qes[0] = _mm_load_si128((__m128i *)&cache_line_base[0]);
3910
3911         /* Evict the cache line ASAP */
3912         rte_cldemote(cache_line_base);
3913
3914         /* Extract and combine the gen bits */
3915         gen_bits = ((_mm_extract_epi8(qes[0], 15) & 0x1) << 0) |
3916                    ((_mm_extract_epi8(qes[1], 15) & 0x1) << 1) |
3917                    ((_mm_extract_epi8(qes[2], 15) & 0x1) << 2) |
3918                    ((_mm_extract_epi8(qes[3], 15) & 0x1) << 3);
3919
3920         /* XOR the combined bits such that a 1 represents a valid QE */
3921         gen_bits ^= xor_mask[qm_port->gen_bit][*offset];
3922
3923         /* Mask off gen bits we don't care about */
3924         gen_bits &= and_mask[*offset];
3925
3926         return __builtin_popcount(gen_bits);
3927 }
3928
3929 static inline int16_t
3930 dlb2_hw_dequeue(struct dlb2_eventdev *dlb2,
3931                 struct dlb2_eventdev_port *ev_port,
3932                 struct rte_event *events,
3933                 uint16_t max_num,
3934                 uint64_t dequeue_timeout_ticks)
3935 {
3936         uint64_t timeout;
3937         uint64_t start_ticks = 0ULL;
3938         struct dlb2_port *qm_port;
3939         int num = 0;
3940
3941         qm_port = &ev_port->qm_port;
3942
3943         /* We have a special implementation for waiting. Wait can be:
3944          * 1) no waiting at all
3945          * 2) busy poll only
3946          * 3) wait for interrupt. If wakeup and poll time
3947          * has expired, then return to caller
3948          * 4) umonitor/umwait repeatedly up to poll time
3949          */
3950
3951         /* If configured for per dequeue wait, then use wait value provided
3952          * to this API. Otherwise we must use the global
3953          * value from eventdev config time.
3954          */
3955         if (!dlb2->global_dequeue_wait)
3956                 timeout = dequeue_timeout_ticks;
3957         else
3958                 timeout = dlb2->global_dequeue_wait_ticks;
3959
3960         start_ticks = rte_get_timer_cycles();
3961
3962         while (num < max_num) {
3963                 struct dlb2_dequeue_qe qes[DLB2_NUM_QES_PER_CACHE_LINE];
3964                 uint8_t offset;
3965                 int num_avail;
3966
3967                 /* Copy up to 4 QEs from the current cache line into qes */
3968                 num_avail = dlb2_recv_qe(qm_port, qes, &offset);
3969
3970                 /* But don't process more than the user requested */
3971                 num_avail = RTE_MIN(num_avail, max_num - num);
3972
3973                 dlb2_inc_cq_idx(qm_port, num_avail);
3974
3975                 if (num_avail == DLB2_NUM_QES_PER_CACHE_LINE)
3976                         num += dlb2_process_dequeue_four_qes(ev_port,
3977                                                              qm_port,
3978                                                              &events[num],
3979                                                              &qes[offset]);
3980                 else if (num_avail)
3981                         num += dlb2_process_dequeue_qes(ev_port,
3982                                                         qm_port,
3983                                                         &events[num],
3984                                                         &qes[offset],
3985                                                         num_avail);
3986                 else if ((timeout == 0) || (num > 0))
3987                         /* Not waiting in any form, or 1+ events received? */
3988                         break;
3989                 else if (dlb2_dequeue_wait(dlb2, ev_port, qm_port,
3990                                            timeout, start_ticks))
3991                         break;
3992         }
3993
3994         qm_port->owed_tokens += num;
3995
3996         if (num) {
3997                 if (qm_port->token_pop_mode == AUTO_POP)
3998                         dlb2_consume_qe_immediate(qm_port, num);
3999
4000                 ev_port->outstanding_releases += num;
4001
4002                 dlb2_port_credits_inc(qm_port, num);
4003         }
4004
4005         return num;
4006 }
4007
4008 static uint16_t
4009 dlb2_event_dequeue_burst(void *event_port, struct rte_event *ev, uint16_t num,
4010                          uint64_t wait)
4011 {
4012         struct dlb2_eventdev_port *ev_port = event_port;
4013         struct dlb2_port *qm_port = &ev_port->qm_port;
4014         struct dlb2_eventdev *dlb2 = ev_port->dlb2;
4015         uint16_t cnt;
4016
4017         RTE_ASSERT(ev_port->setup_done);
4018         RTE_ASSERT(ev != NULL);
4019
4020         if (ev_port->implicit_release && ev_port->outstanding_releases > 0) {
4021                 uint16_t out_rels = ev_port->outstanding_releases;
4022
4023                 dlb2_event_release(dlb2, ev_port->id, out_rels);
4024
4025                 DLB2_INC_STAT(ev_port->stats.tx_implicit_rel, out_rels);
4026         }
4027
4028         if (qm_port->token_pop_mode == DEFERRED_POP && qm_port->owed_tokens)
4029                 dlb2_consume_qe_immediate(qm_port, qm_port->owed_tokens);
4030
4031         cnt = dlb2_hw_dequeue(dlb2, ev_port, ev, num, wait);
4032
4033         DLB2_INC_STAT(ev_port->stats.traffic.total_polls, 1);
4034         DLB2_INC_STAT(ev_port->stats.traffic.zero_polls, ((cnt == 0) ? 1 : 0));
4035
4036         return cnt;
4037 }
4038
4039 static uint16_t
4040 dlb2_event_dequeue(void *event_port, struct rte_event *ev, uint64_t wait)
4041 {
4042         return dlb2_event_dequeue_burst(event_port, ev, 1, wait);
4043 }
4044
4045 static uint16_t
4046 dlb2_event_dequeue_burst_sparse(void *event_port, struct rte_event *ev,
4047                                 uint16_t num, uint64_t wait)
4048 {
4049         struct dlb2_eventdev_port *ev_port = event_port;
4050         struct dlb2_port *qm_port = &ev_port->qm_port;
4051         struct dlb2_eventdev *dlb2 = ev_port->dlb2;
4052         uint16_t cnt;
4053
4054         RTE_ASSERT(ev_port->setup_done);
4055         RTE_ASSERT(ev != NULL);
4056
4057         if (ev_port->implicit_release && ev_port->outstanding_releases > 0) {
4058                 uint16_t out_rels = ev_port->outstanding_releases;
4059
4060                 dlb2_event_release(dlb2, ev_port->id, out_rels);
4061
4062                 DLB2_INC_STAT(ev_port->stats.tx_implicit_rel, out_rels);
4063         }
4064
4065         if (qm_port->token_pop_mode == DEFERRED_POP && qm_port->owed_tokens)
4066                 dlb2_consume_qe_immediate(qm_port, qm_port->owed_tokens);
4067
4068         cnt = dlb2_hw_dequeue_sparse(dlb2, ev_port, ev, num, wait);
4069
4070         DLB2_INC_STAT(ev_port->stats.traffic.total_polls, 1);
4071         DLB2_INC_STAT(ev_port->stats.traffic.zero_polls, ((cnt == 0) ? 1 : 0));
4072         return cnt;
4073 }
4074
4075 static uint16_t
4076 dlb2_event_dequeue_sparse(void *event_port, struct rte_event *ev,
4077                           uint64_t wait)
4078 {
4079         return dlb2_event_dequeue_burst_sparse(event_port, ev, 1, wait);
4080 }
4081
4082 static void
4083 dlb2_flush_port(struct rte_eventdev *dev, int port_id)
4084 {
4085         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
4086         eventdev_stop_flush_t flush;
4087         struct rte_event ev;
4088         uint8_t dev_id;
4089         void *arg;
4090         int i;
4091
4092         flush = dev->dev_ops->dev_stop_flush;
4093         dev_id = dev->data->dev_id;
4094         arg = dev->data->dev_stop_flush_arg;
4095
4096         while (rte_event_dequeue_burst(dev_id, port_id, &ev, 1, 0)) {
4097                 if (flush)
4098                         flush(dev_id, ev, arg);
4099
4100                 if (dlb2->ev_ports[port_id].qm_port.is_directed)
4101                         continue;
4102
4103                 ev.op = RTE_EVENT_OP_RELEASE;
4104
4105                 rte_event_enqueue_burst(dev_id, port_id, &ev, 1);
4106         }
4107
4108         /* Enqueue any additional outstanding releases */
4109         ev.op = RTE_EVENT_OP_RELEASE;
4110
4111         for (i = dlb2->ev_ports[port_id].outstanding_releases; i > 0; i--)
4112                 rte_event_enqueue_burst(dev_id, port_id, &ev, 1);
4113 }
4114
4115 static uint32_t
4116 dlb2_get_ldb_queue_depth(struct dlb2_eventdev *dlb2,
4117                          struct dlb2_eventdev_queue *queue)
4118 {
4119         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
4120         struct dlb2_get_ldb_queue_depth_args cfg;
4121         int ret;
4122
4123         cfg.queue_id = queue->qm_queue.id;
4124
4125         ret = dlb2_iface_get_ldb_queue_depth(handle, &cfg);
4126         if (ret < 0) {
4127                 DLB2_LOG_ERR("dlb2: get_ldb_queue_depth ret=%d (driver status: %s)\n",
4128                              ret, dlb2_error_strings[cfg.response.status]);
4129                 return ret;
4130         }
4131
4132         return cfg.response.id;
4133 }
4134
4135 static uint32_t
4136 dlb2_get_dir_queue_depth(struct dlb2_eventdev *dlb2,
4137                          struct dlb2_eventdev_queue *queue)
4138 {
4139         struct dlb2_hw_dev *handle = &dlb2->qm_instance;
4140         struct dlb2_get_dir_queue_depth_args cfg;
4141         int ret;
4142
4143         cfg.queue_id = queue->qm_queue.id;
4144
4145         ret = dlb2_iface_get_dir_queue_depth(handle, &cfg);
4146         if (ret < 0) {
4147                 DLB2_LOG_ERR("dlb2: get_dir_queue_depth ret=%d (driver status: %s)\n",
4148                              ret, dlb2_error_strings[cfg.response.status]);
4149                 return ret;
4150         }
4151
4152         return cfg.response.id;
4153 }
4154
4155 uint32_t
4156 dlb2_get_queue_depth(struct dlb2_eventdev *dlb2,
4157                      struct dlb2_eventdev_queue *queue)
4158 {
4159         if (queue->qm_queue.is_directed)
4160                 return dlb2_get_dir_queue_depth(dlb2, queue);
4161         else
4162                 return dlb2_get_ldb_queue_depth(dlb2, queue);
4163 }
4164
4165 static bool
4166 dlb2_queue_is_empty(struct dlb2_eventdev *dlb2,
4167                     struct dlb2_eventdev_queue *queue)
4168 {
4169         return dlb2_get_queue_depth(dlb2, queue) == 0;
4170 }
4171
4172 static bool
4173 dlb2_linked_queues_empty(struct dlb2_eventdev *dlb2)
4174 {
4175         int i;
4176
4177         for (i = 0; i < dlb2->num_queues; i++) {
4178                 if (dlb2->ev_queues[i].num_links == 0)
4179                         continue;
4180                 if (!dlb2_queue_is_empty(dlb2, &dlb2->ev_queues[i]))
4181                         return false;
4182         }
4183
4184         return true;
4185 }
4186
4187 static bool
4188 dlb2_queues_empty(struct dlb2_eventdev *dlb2)
4189 {
4190         int i;
4191
4192         for (i = 0; i < dlb2->num_queues; i++) {
4193                 if (!dlb2_queue_is_empty(dlb2, &dlb2->ev_queues[i]))
4194                         return false;
4195         }
4196
4197         return true;
4198 }
4199
4200 static void
4201 dlb2_drain(struct rte_eventdev *dev)
4202 {
4203         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
4204         struct dlb2_eventdev_port *ev_port = NULL;
4205         uint8_t dev_id;
4206         int i;
4207
4208         dev_id = dev->data->dev_id;
4209
4210         while (!dlb2_linked_queues_empty(dlb2)) {
4211                 /* Flush all the ev_ports, which will drain all their connected
4212                  * queues.
4213                  */
4214                 for (i = 0; i < dlb2->num_ports; i++)
4215                         dlb2_flush_port(dev, i);
4216         }
4217
4218         /* The queues are empty, but there may be events left in the ports. */
4219         for (i = 0; i < dlb2->num_ports; i++)
4220                 dlb2_flush_port(dev, i);
4221
4222         /* If the domain's queues are empty, we're done. */
4223         if (dlb2_queues_empty(dlb2))
4224                 return;
4225
4226         /* Else, there must be at least one unlinked load-balanced queue.
4227          * Select a load-balanced port with which to drain the unlinked
4228          * queue(s).
4229          */
4230         for (i = 0; i < dlb2->num_ports; i++) {
4231                 ev_port = &dlb2->ev_ports[i];
4232
4233                 if (!ev_port->qm_port.is_directed)
4234                         break;
4235         }
4236
4237         if (i == dlb2->num_ports) {
4238                 DLB2_LOG_ERR("internal error: no LDB ev_ports\n");
4239                 return;
4240         }
4241
4242         rte_errno = 0;
4243         rte_event_port_unlink(dev_id, ev_port->id, NULL, 0);
4244
4245         if (rte_errno) {
4246                 DLB2_LOG_ERR("internal error: failed to unlink ev_port %d\n",
4247                              ev_port->id);
4248                 return;
4249         }
4250
4251         for (i = 0; i < dlb2->num_queues; i++) {
4252                 uint8_t qid, prio;
4253                 int ret;
4254
4255                 if (dlb2_queue_is_empty(dlb2, &dlb2->ev_queues[i]))
4256                         continue;
4257
4258                 qid = i;
4259                 prio = 0;
4260
4261                 /* Link the ev_port to the queue */
4262                 ret = rte_event_port_link(dev_id, ev_port->id, &qid, &prio, 1);
4263                 if (ret != 1) {
4264                         DLB2_LOG_ERR("internal error: failed to link ev_port %d to queue %d\n",
4265                                      ev_port->id, qid);
4266                         return;
4267                 }
4268
4269                 /* Flush the queue */
4270                 while (!dlb2_queue_is_empty(dlb2, &dlb2->ev_queues[i]))
4271                         dlb2_flush_port(dev, ev_port->id);
4272
4273                 /* Drain any extant events in the ev_port. */
4274                 dlb2_flush_port(dev, ev_port->id);
4275
4276                 /* Unlink the ev_port from the queue */
4277                 ret = rte_event_port_unlink(dev_id, ev_port->id, &qid, 1);
4278                 if (ret != 1) {
4279                         DLB2_LOG_ERR("internal error: failed to unlink ev_port %d to queue %d\n",
4280                                      ev_port->id, qid);
4281                         return;
4282                 }
4283         }
4284 }
4285
4286 static void
4287 dlb2_eventdev_stop(struct rte_eventdev *dev)
4288 {
4289         struct dlb2_eventdev *dlb2 = dlb2_pmd_priv(dev);
4290
4291         rte_spinlock_lock(&dlb2->qm_instance.resource_lock);
4292
4293         if (dlb2->run_state == DLB2_RUN_STATE_STOPPED) {
4294                 DLB2_LOG_DBG("Internal error: already stopped\n");
4295                 rte_spinlock_unlock(&dlb2->qm_instance.resource_lock);
4296                 return;
4297         } else if (dlb2->run_state != DLB2_RUN_STATE_STARTED) {
4298                 DLB2_LOG_ERR("Internal error: bad state %d for dev_stop\n",
4299                              (int)dlb2->run_state);
4300                 rte_spinlock_unlock(&dlb2->qm_instance.resource_lock);
4301                 return;
4302         }
4303
4304         dlb2->run_state = DLB2_RUN_STATE_STOPPING;
4305
4306         rte_spinlock_unlock(&dlb2->qm_instance.resource_lock);
4307
4308         dlb2_drain(dev);
4309
4310         dlb2->run_state = DLB2_RUN_STATE_STOPPED;
4311 }
4312
4313 static int
4314 dlb2_eventdev_close(struct rte_eventdev *dev)
4315 {
4316         dlb2_hw_reset_sched_domain(dev, false);
4317
4318         return 0;
4319 }
4320
4321 static void
4322 dlb2_eventdev_queue_release(struct rte_eventdev *dev, uint8_t id)
4323 {
4324         RTE_SET_USED(dev);
4325         RTE_SET_USED(id);
4326
4327         /* This function intentionally left blank. */
4328 }
4329
4330 static void
4331 dlb2_eventdev_port_release(void *port)
4332 {
4333         struct dlb2_eventdev_port *ev_port = port;
4334         struct dlb2_port *qm_port;
4335
4336         if (ev_port) {
4337                 qm_port = &ev_port->qm_port;
4338                 if (qm_port->config_state == DLB2_CONFIGURED)
4339                         dlb2_free_qe_mem(qm_port);
4340         }
4341 }
4342
4343 static int
4344 dlb2_eventdev_timeout_ticks(struct rte_eventdev *dev, uint64_t ns,
4345                             uint64_t *timeout_ticks)
4346 {
4347         RTE_SET_USED(dev);
4348         uint64_t cycles_per_ns = rte_get_timer_hz() / 1E9;
4349
4350         *timeout_ticks = ns * cycles_per_ns;
4351
4352         return 0;
4353 }
4354
4355 static void
4356 dlb2_entry_points_init(struct rte_eventdev *dev)
4357 {
4358         struct dlb2_eventdev *dlb2;
4359
4360         /* Expose PMD's eventdev interface */
4361         static struct rte_eventdev_ops dlb2_eventdev_entry_ops = {
4362                 .dev_infos_get    = dlb2_eventdev_info_get,
4363                 .dev_configure    = dlb2_eventdev_configure,
4364                 .dev_start        = dlb2_eventdev_start,
4365                 .dev_stop         = dlb2_eventdev_stop,
4366                 .dev_close        = dlb2_eventdev_close,
4367                 .queue_def_conf   = dlb2_eventdev_queue_default_conf_get,
4368                 .queue_setup      = dlb2_eventdev_queue_setup,
4369                 .queue_release    = dlb2_eventdev_queue_release,
4370                 .port_def_conf    = dlb2_eventdev_port_default_conf_get,
4371                 .port_setup       = dlb2_eventdev_port_setup,
4372                 .port_release     = dlb2_eventdev_port_release,
4373                 .port_link        = dlb2_eventdev_port_link,
4374                 .port_unlink      = dlb2_eventdev_port_unlink,
4375                 .port_unlinks_in_progress =
4376                                     dlb2_eventdev_port_unlinks_in_progress,
4377                 .timeout_ticks    = dlb2_eventdev_timeout_ticks,
4378                 .dump             = dlb2_eventdev_dump,
4379                 .xstats_get       = dlb2_eventdev_xstats_get,
4380                 .xstats_get_names = dlb2_eventdev_xstats_get_names,
4381                 .xstats_get_by_name = dlb2_eventdev_xstats_get_by_name,
4382                 .xstats_reset       = dlb2_eventdev_xstats_reset,
4383                 .dev_selftest     = test_dlb2_eventdev,
4384         };
4385
4386         /* Expose PMD's eventdev interface */
4387
4388         dev->dev_ops = &dlb2_eventdev_entry_ops;
4389         dev->enqueue = dlb2_event_enqueue;
4390         dev->enqueue_burst = dlb2_event_enqueue_burst;
4391         dev->enqueue_new_burst = dlb2_event_enqueue_new_burst;
4392         dev->enqueue_forward_burst = dlb2_event_enqueue_forward_burst;
4393
4394         dlb2 = dev->data->dev_private;
4395         if (dlb2->poll_mode == DLB2_CQ_POLL_MODE_SPARSE) {
4396                 dev->dequeue = dlb2_event_dequeue_sparse;
4397                 dev->dequeue_burst = dlb2_event_dequeue_burst_sparse;
4398         } else {
4399                 dev->dequeue = dlb2_event_dequeue;
4400                 dev->dequeue_burst = dlb2_event_dequeue_burst;
4401         }
4402 }
4403
4404 int
4405 dlb2_primary_eventdev_probe(struct rte_eventdev *dev,
4406                             const char *name,
4407                             struct dlb2_devargs *dlb2_args)
4408 {
4409         struct dlb2_eventdev *dlb2;
4410         int err, i;
4411
4412         dlb2 = dev->data->dev_private;
4413
4414         dlb2->event_dev = dev; /* backlink */
4415
4416         evdev_dlb2_default_info.driver_name = name;
4417
4418         dlb2->max_num_events_override = dlb2_args->max_num_events;
4419         dlb2->num_dir_credits_override = dlb2_args->num_dir_credits_override;
4420         dlb2->qm_instance.cos_id = dlb2_args->cos_id;
4421         dlb2->poll_interval = dlb2_args->poll_interval;
4422         dlb2->sw_credit_quanta = dlb2_args->sw_credit_quanta;
4423         dlb2->default_depth_thresh = dlb2_args->default_depth_thresh;
4424         dlb2->vector_opts_disabled = dlb2_args->vector_opts_disabled;
4425
4426         err = dlb2_iface_open(&dlb2->qm_instance, name);
4427         if (err < 0) {
4428                 DLB2_LOG_ERR("could not open event hardware device, err=%d\n",
4429                              err);
4430                 return err;
4431         }
4432
4433         err = dlb2_iface_get_device_version(&dlb2->qm_instance,
4434                                             &dlb2->revision);
4435         if (err < 0) {
4436                 DLB2_LOG_ERR("dlb2: failed to get the device version, err=%d\n",
4437                              err);
4438                 return err;
4439         }
4440
4441         err = dlb2_hw_query_resources(dlb2);
4442         if (err) {
4443                 DLB2_LOG_ERR("get resources err=%d for %s\n",
4444                              err, name);
4445                 return err;
4446         }
4447
4448         dlb2_iface_hardware_init(&dlb2->qm_instance);
4449
4450         err = dlb2_iface_get_cq_poll_mode(&dlb2->qm_instance, &dlb2->poll_mode);
4451         if (err < 0) {
4452                 DLB2_LOG_ERR("dlb2: failed to get the poll mode, err=%d\n",
4453                              err);
4454                 return err;
4455         }
4456
4457         /* Complete xtstats runtime initialization */
4458         err = dlb2_xstats_init(dlb2);
4459         if (err) {
4460                 DLB2_LOG_ERR("dlb2: failed to init xstats, err=%d\n", err);
4461                 return err;
4462         }
4463
4464         /* Initialize each port's token pop mode */
4465         for (i = 0; i < DLB2_MAX_NUM_PORTS(dlb2->version); i++)
4466                 dlb2->ev_ports[i].qm_port.token_pop_mode = AUTO_POP;
4467
4468         rte_spinlock_init(&dlb2->qm_instance.resource_lock);
4469
4470         dlb2_iface_low_level_io_init();
4471
4472         dlb2_entry_points_init(dev);
4473
4474         dlb2_init_queue_depth_thresholds(dlb2,
4475                                          dlb2_args->qid_depth_thresholds.val);
4476
4477         return 0;
4478 }
4479
4480 int
4481 dlb2_secondary_eventdev_probe(struct rte_eventdev *dev,
4482                               const char *name)
4483 {
4484         struct dlb2_eventdev *dlb2;
4485         int err;
4486
4487         dlb2 = dev->data->dev_private;
4488
4489         evdev_dlb2_default_info.driver_name = name;
4490
4491         err = dlb2_iface_open(&dlb2->qm_instance, name);
4492         if (err < 0) {
4493                 DLB2_LOG_ERR("could not open event hardware device, err=%d\n",
4494                              err);
4495                 return err;
4496         }
4497
4498         err = dlb2_hw_query_resources(dlb2);
4499         if (err) {
4500                 DLB2_LOG_ERR("get resources err=%d for %s\n",
4501                              err, name);
4502                 return err;
4503         }
4504
4505         dlb2_iface_low_level_io_init();
4506
4507         dlb2_entry_points_init(dev);
4508
4509         return 0;
4510 }
4511
4512 int
4513 dlb2_parse_params(const char *params,
4514                   const char *name,
4515                   struct dlb2_devargs *dlb2_args,
4516                   uint8_t version)
4517 {
4518         int ret = 0;
4519         static const char * const args[] = { NUMA_NODE_ARG,
4520                                              DLB2_MAX_NUM_EVENTS,
4521                                              DLB2_NUM_DIR_CREDITS,
4522                                              DEV_ID_ARG,
4523                                              DLB2_QID_DEPTH_THRESH_ARG,
4524                                              DLB2_COS_ARG,
4525                                              DLB2_POLL_INTERVAL_ARG,
4526                                              DLB2_SW_CREDIT_QUANTA_ARG,
4527                                              DLB2_DEPTH_THRESH_ARG,
4528                                              DLB2_VECTOR_OPTS_DISAB_ARG,
4529                                              NULL };
4530
4531         if (params != NULL && params[0] != '\0') {
4532                 struct rte_kvargs *kvlist = rte_kvargs_parse(params, args);
4533
4534                 if (kvlist == NULL) {
4535                         RTE_LOG(INFO, PMD,
4536                                 "Ignoring unsupported parameters when creating device '%s'\n",
4537                                 name);
4538                 } else {
4539                         int ret = rte_kvargs_process(kvlist, NUMA_NODE_ARG,
4540                                                      set_numa_node,
4541                                                      &dlb2_args->socket_id);
4542                         if (ret != 0) {
4543                                 DLB2_LOG_ERR("%s: Error parsing numa node parameter",
4544                                              name);
4545                                 rte_kvargs_free(kvlist);
4546                                 return ret;
4547                         }
4548
4549                         ret = rte_kvargs_process(kvlist, DLB2_MAX_NUM_EVENTS,
4550                                                  set_max_num_events,
4551                                                  &dlb2_args->max_num_events);
4552                         if (ret != 0) {
4553                                 DLB2_LOG_ERR("%s: Error parsing max_num_events parameter",
4554                                              name);
4555                                 rte_kvargs_free(kvlist);
4556                                 return ret;
4557                         }
4558
4559                         if (version == DLB2_HW_V2) {
4560                                 ret = rte_kvargs_process(kvlist,
4561                                         DLB2_NUM_DIR_CREDITS,
4562                                         set_num_dir_credits,
4563                                         &dlb2_args->num_dir_credits_override);
4564                                 if (ret != 0) {
4565                                         DLB2_LOG_ERR("%s: Error parsing num_dir_credits parameter",
4566                                                      name);
4567                                         rte_kvargs_free(kvlist);
4568                                         return ret;
4569                                 }
4570                         }
4571                         ret = rte_kvargs_process(kvlist, DEV_ID_ARG,
4572                                                  set_dev_id,
4573                                                  &dlb2_args->dev_id);
4574                         if (ret != 0) {
4575                                 DLB2_LOG_ERR("%s: Error parsing dev_id parameter",
4576                                              name);
4577                                 rte_kvargs_free(kvlist);
4578                                 return ret;
4579                         }
4580
4581                         if (version == DLB2_HW_V2) {
4582                                 ret = rte_kvargs_process(
4583                                         kvlist,
4584                                         DLB2_QID_DEPTH_THRESH_ARG,
4585                                         set_qid_depth_thresh,
4586                                         &dlb2_args->qid_depth_thresholds);
4587                         } else {
4588                                 ret = rte_kvargs_process(
4589                                         kvlist,
4590                                         DLB2_QID_DEPTH_THRESH_ARG,
4591                                         set_qid_depth_thresh_v2_5,
4592                                         &dlb2_args->qid_depth_thresholds);
4593                         }
4594                         if (ret != 0) {
4595                                 DLB2_LOG_ERR("%s: Error parsing qid_depth_thresh parameter",
4596                                              name);
4597                                 rte_kvargs_free(kvlist);
4598                                 return ret;
4599                         }
4600
4601                         ret = rte_kvargs_process(kvlist, DLB2_COS_ARG,
4602                                                  set_cos,
4603                                                  &dlb2_args->cos_id);
4604                         if (ret != 0) {
4605                                 DLB2_LOG_ERR("%s: Error parsing cos parameter",
4606                                              name);
4607                                 rte_kvargs_free(kvlist);
4608                                 return ret;
4609                         }
4610
4611                         ret = rte_kvargs_process(kvlist, DLB2_POLL_INTERVAL_ARG,
4612                                                  set_poll_interval,
4613                                                  &dlb2_args->poll_interval);
4614                         if (ret != 0) {
4615                                 DLB2_LOG_ERR("%s: Error parsing poll interval parameter",
4616                                              name);
4617                                 rte_kvargs_free(kvlist);
4618                                 return ret;
4619                         }
4620
4621                         ret = rte_kvargs_process(kvlist,
4622                                                  DLB2_SW_CREDIT_QUANTA_ARG,
4623                                                  set_sw_credit_quanta,
4624                                                  &dlb2_args->sw_credit_quanta);
4625                         if (ret != 0) {
4626                                 DLB2_LOG_ERR("%s: Error parsing sw xredit quanta parameter",
4627                                              name);
4628                                 rte_kvargs_free(kvlist);
4629                                 return ret;
4630                         }
4631
4632                         ret = rte_kvargs_process(kvlist, DLB2_DEPTH_THRESH_ARG,
4633                                         set_default_depth_thresh,
4634                                         &dlb2_args->default_depth_thresh);
4635                         if (ret != 0) {
4636                                 DLB2_LOG_ERR("%s: Error parsing set depth thresh parameter",
4637                                              name);
4638                                 rte_kvargs_free(kvlist);
4639                                 return ret;
4640                         }
4641
4642                         ret = rte_kvargs_process(kvlist,
4643                                         DLB2_VECTOR_OPTS_DISAB_ARG,
4644                                         set_vector_opts_disab,
4645                                         &dlb2_args->vector_opts_disabled);
4646                         if (ret != 0) {
4647                                 DLB2_LOG_ERR("%s: Error parsing vector opts disabled",
4648                                              name);
4649                                 rte_kvargs_free(kvlist);
4650                                 return ret;
4651                         }
4652
4653                         rte_kvargs_free(kvlist);
4654                 }
4655         }
4656         return ret;
4657 }
4658 RTE_LOG_REGISTER(eventdev_dlb2_log_level, pmd.event.dlb2, NOTICE);