examples/eventdev_pipeline_sw_pmd/pipeline_worker_tx.c

   1 /*
   2  * SPDX-License-Identifier: BSD-3-Clause
   3  * Copyright(c) 2010-2014 Intel Corporation
   4  * Copyright 2017 Cavium, Inc.
   5  */
   6
   7 #include "pipeline_common.h"
   8
   9 static __rte_always_inline void
  10 worker_fwd_event(struct rte_event *ev, uint8_t sched)
  11 {
  12         ev->event_type = RTE_EVENT_TYPE_CPU;
  13         ev->op = RTE_EVENT_OP_FORWARD;
  14         ev->sched_type = sched;
  15 }
  16
  17 static __rte_always_inline void
  18 worker_event_enqueue(const uint8_t dev, const uint8_t port,
  19                 struct rte_event *ev)
  20 {
  21         while (rte_event_enqueue_burst(dev, port, ev, 1) != 1)
  22                 rte_pause();
  23 }
  24
  25 static __rte_always_inline void
  26 worker_event_enqueue_burst(const uint8_t dev, const uint8_t port,
  27                 struct rte_event *ev, const uint16_t nb_rx)
  28 {
  29         uint16_t enq;
  30
  31         enq = rte_event_enqueue_burst(dev, port, ev, nb_rx);
  32         while (enq < nb_rx) {
  33                 enq += rte_event_enqueue_burst(dev, port,
  34                                                 ev + enq, nb_rx - enq);
  35         }
  36 }
  37
  38 static __rte_always_inline void
  39 worker_tx_pkt(struct rte_mbuf *mbuf)
  40 {
  41         exchange_mac(mbuf);
  42         while (rte_eth_tx_burst(mbuf->port, 0, &mbuf, 1) != 1)
  43                 rte_pause();
  44 }
  45
  46 /* Multi stage Pipeline Workers */
  47
  48 static int
  49 worker_do_tx(void *arg)
  50 {
  51         struct rte_event ev;
  52
  53         struct worker_data *data = (struct worker_data *)arg;
  54         const uint8_t dev = data->dev_id;
  55         const uint8_t port = data->port_id;
  56         const uint8_t lst_qid = cdata.num_stages - 1;
  57         size_t fwd = 0, received = 0, tx = 0;
  58
  59
  60         while (!fdata->done) {
  61
  62                 if (!rte_event_dequeue_burst(dev, port, &ev, 1, 0)) {
  63                         rte_pause();
  64                         continue;
  65                 }
  66
  67                 received++;
  68                 const uint8_t cq_id = ev.queue_id % cdata.num_stages;
  69
  70                 if (cq_id >= lst_qid) {
  71                         if (ev.sched_type == RTE_SCHED_TYPE_ATOMIC) {
  72                                 worker_tx_pkt(ev.mbuf);
  73                                 tx++;
  74                                 continue;
  75                         }
  76
  77                         worker_fwd_event(&ev, RTE_SCHED_TYPE_ATOMIC);
  78                         ev.queue_id = (cq_id == lst_qid) ?
  79                                 cdata.next_qid[ev.queue_id] : ev.queue_id;
  80                 } else {
  81                         ev.queue_id = cdata.next_qid[ev.queue_id];
  82                         worker_fwd_event(&ev, cdata.queue_type);
  83                 }
  84                 work();
  85
  86                 worker_event_enqueue(dev, port, &ev);
  87                 fwd++;
  88         }
  89
  90         if (!cdata.quiet)
  91                 printf("  worker %u thread done. RX=%zu FWD=%zu TX=%zu\n",
  92                                 rte_lcore_id(), received, fwd, tx);
  93
  94         return 0;
  95 }
  96
  97 static int
  98 worker_do_tx_atq(void *arg)
  99 {
 100         struct rte_event ev;
 101
 102         struct worker_data *data = (struct worker_data *)arg;
 103         const uint8_t dev = data->dev_id;
 104         const uint8_t port = data->port_id;
 105         const uint8_t lst_qid = cdata.num_stages - 1;
 106         size_t fwd = 0, received = 0, tx = 0;
 107
 108         while (!fdata->done) {
 109
 110                 if (!rte_event_dequeue_burst(dev, port, &ev, 1, 0)) {
 111                         rte_pause();
 112                         continue;
 113                 }
 114
 115                 received++;
 116                 const uint8_t cq_id = ev.sub_event_type % cdata.num_stages;
 117
 118                 if (cq_id == lst_qid) {
 119                         if (ev.sched_type == RTE_SCHED_TYPE_ATOMIC) {
 120                                 worker_tx_pkt(ev.mbuf);
 121                                 tx++;
 122                                 continue;
 123                         }
 124
 125                         worker_fwd_event(&ev, RTE_SCHED_TYPE_ATOMIC);
 126                 } else {
 127                         ev.sub_event_type++;
 128                         worker_fwd_event(&ev, cdata.queue_type);
 129                 }
 130                 work();
 131
 132                 worker_event_enqueue(dev, port, &ev);
 133                 fwd++;
 134         }
 135
 136         if (!cdata.quiet)
 137                 printf("  worker %u thread done. RX=%zu FWD=%zu TX=%zu\n",
 138                                 rte_lcore_id(), received, fwd, tx);
 139
 140         return 0;
 141 }
 142
 143 static int
 144 worker_do_tx_burst(void *arg)
 145 {
 146         struct rte_event ev[BATCH_SIZE];
 147
 148         struct worker_data *data = (struct worker_data *)arg;
 149         uint8_t dev = data->dev_id;
 150         uint8_t port = data->port_id;
 151         uint8_t lst_qid = cdata.num_stages - 1;
 152         size_t fwd = 0, received = 0, tx = 0;
 153
 154         while (!fdata->done) {
 155                 uint16_t i;
 156                 const uint16_t nb_rx = rte_event_dequeue_burst(dev, port,
 157                                 ev, BATCH_SIZE, 0);
 158
 159                 if (nb_rx == 0) {
 160                         rte_pause();
 161                         continue;
 162                 }
 163                 received += nb_rx;
 164
 165                 for (i = 0; i < nb_rx; i++) {
 166                         const uint8_t cq_id = ev[i].queue_id % cdata.num_stages;
 167
 168                         if (cq_id >= lst_qid) {
 169                                 if (ev[i].sched_type == RTE_SCHED_TYPE_ATOMIC) {
 170                                         worker_tx_pkt(ev[i].mbuf);
 171                                         tx++;
 172                                         ev[i].op = RTE_EVENT_OP_RELEASE;
 173                                         continue;
 174                                 }
 175                                 ev[i].queue_id = (cq_id == lst_qid) ?
 176                                         cdata.next_qid[ev[i].queue_id] :
 177                                         ev[i].queue_id;
 178
 179                                 worker_fwd_event(&ev[i], RTE_SCHED_TYPE_ATOMIC);
 180                         } else {
 181                                 ev[i].queue_id = cdata.next_qid[ev[i].queue_id];
 182                                 worker_fwd_event(&ev[i], cdata.queue_type);
 183                         }
 184                         work();
 185                 }
 186                 worker_event_enqueue_burst(dev, port, ev, nb_rx);
 187
 188                 fwd += nb_rx;
 189         }
 190
 191         if (!cdata.quiet)
 192                 printf("  worker %u thread done. RX=%zu FWD=%zu TX=%zu\n",
 193                                 rte_lcore_id(), received, fwd, tx);
 194
 195         return 0;
 196 }
 197
 198 static int
 199 worker_do_tx_burst_atq(void *arg)
 200 {
 201         struct rte_event ev[BATCH_SIZE];
 202
 203         struct worker_data *data = (struct worker_data *)arg;
 204         uint8_t dev = data->dev_id;
 205         uint8_t port = data->port_id;
 206         uint8_t lst_qid = cdata.num_stages - 1;
 207         size_t fwd = 0, received = 0, tx = 0;
 208
 209         while (!fdata->done) {
 210                 uint16_t i;
 211
 212                 const uint16_t nb_rx = rte_event_dequeue_burst(dev, port,
 213                                 ev, BATCH_SIZE, 0);
 214
 215                 if (nb_rx == 0) {
 216                         rte_pause();
 217                         continue;
 218                 }
 219                 received += nb_rx;
 220
 221                 for (i = 0; i < nb_rx; i++) {
 222                         const uint8_t cq_id = ev[i].sub_event_type %
 223                                 cdata.num_stages;
 224
 225                         if (cq_id == lst_qid) {
 226                                 if (ev[i].sched_type == RTE_SCHED_TYPE_ATOMIC) {
 227                                         worker_tx_pkt(ev[i].mbuf);
 228                                         tx++;
 229                                         ev[i].op = RTE_EVENT_OP_RELEASE;
 230                                         continue;
 231                                 }
 232
 233                                 worker_fwd_event(&ev[i], RTE_SCHED_TYPE_ATOMIC);
 234                         } else {
 235                                 ev[i].sub_event_type++;
 236                                 worker_fwd_event(&ev[i], cdata.queue_type);
 237                         }
 238                         work();
 239                 }
 240
 241                 worker_event_enqueue_burst(dev, port, ev, nb_rx);
 242                 fwd += nb_rx;
 243         }
 244
 245         if (!cdata.quiet)
 246                 printf("  worker %u thread done. RX=%zu FWD=%zu TX=%zu\n",
 247                                 rte_lcore_id(), received, fwd, tx);
 248
 249         return 0;
 250 }
 251
 252 static int
 253 setup_eventdev_worker_tx(struct cons_data *cons_data,
 254                 struct worker_data *worker_data)
 255 {
 256         RTE_SET_USED(cons_data);
 257         uint8_t i;
 258         const uint8_t atq = cdata.all_type_queues ? 1 : 0;
 259         const uint8_t dev_id = 0;
 260         const uint8_t nb_ports = cdata.num_workers;
 261         uint8_t nb_slots = 0;
 262         uint8_t nb_queues = rte_eth_dev_count();
 263
 264         /*
 265          * In case where all type queues are not enabled, use queues equal to
 266          * number of stages * eth_dev_count and one extra queue per pipeline
 267          * for Tx.
 268          */
 269         if (!atq) {
 270                 nb_queues *= cdata.num_stages;
 271                 nb_queues += rte_eth_dev_count();
 272         }
 273
 274         struct rte_event_dev_config config = {
 275                         .nb_event_queues = nb_queues,
 276                         .nb_event_ports = nb_ports,
 277                         .nb_events_limit  = 4096,
 278                         .nb_event_queue_flows = 1024,
 279                         .nb_event_port_dequeue_depth = 128,
 280                         .nb_event_port_enqueue_depth = 128,
 281         };
 282         struct rte_event_port_conf wkr_p_conf = {
 283                         .dequeue_depth = cdata.worker_cq_depth,
 284                         .enqueue_depth = 64,
 285                         .new_event_threshold = 4096,
 286         };
 287         struct rte_event_queue_conf wkr_q_conf = {
 288                         .schedule_type = cdata.queue_type,
 289                         .priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
 290                         .nb_atomic_flows = 1024,
 291                         .nb_atomic_order_sequences = 1024,
 292         };
 293
 294         int ret, ndev = rte_event_dev_count();
 295
 296         if (ndev < 1) {
 297                 printf("%d: No Eventdev Devices Found\n", __LINE__);
 298                 return -1;
 299         }
 300
 301
 302         struct rte_event_dev_info dev_info;
 303         ret = rte_event_dev_info_get(dev_id, &dev_info);
 304         printf("\tEventdev %d: %s\n", dev_id, dev_info.driver_name);
 305
 306         if (dev_info.max_event_port_dequeue_depth <
 307                         config.nb_event_port_dequeue_depth)
 308                 config.nb_event_port_dequeue_depth =
 309                                 dev_info.max_event_port_dequeue_depth;
 310         if (dev_info.max_event_port_enqueue_depth <
 311                         config.nb_event_port_enqueue_depth)
 312                 config.nb_event_port_enqueue_depth =
 313                                 dev_info.max_event_port_enqueue_depth;
 314
 315         ret = rte_event_dev_configure(dev_id, &config);
 316         if (ret < 0) {
 317                 printf("%d: Error configuring device\n", __LINE__);
 318                 return -1;
 319         }
 320
 321         printf("  Stages:\n");
 322         for (i = 0; i < nb_queues; i++) {
 323
 324                 if (atq) {
 325
 326                         nb_slots = cdata.num_stages;
 327                         wkr_q_conf.event_queue_cfg =
 328                                 RTE_EVENT_QUEUE_CFG_ALL_TYPES;
 329                 } else {
 330                         uint8_t slot;
 331
 332                         nb_slots = cdata.num_stages + 1;
 333                         slot = i % nb_slots;
 334                         wkr_q_conf.schedule_type = slot == cdata.num_stages ?
 335                                 RTE_SCHED_TYPE_ATOMIC : cdata.queue_type;
 336                 }
 337
 338                 if (rte_event_queue_setup(dev_id, i, &wkr_q_conf) < 0) {
 339                         printf("%d: error creating qid %d\n", __LINE__, i);
 340                         return -1;
 341                 }
 342                 cdata.qid[i] = i;
 343                 cdata.next_qid[i] = i+1;
 344                 if (cdata.enable_queue_priorities) {
 345                         const uint32_t prio_delta =
 346                                 (RTE_EVENT_DEV_PRIORITY_LOWEST) /
 347                                 nb_slots;
 348
 349                         /* higher priority for queues closer to tx */
 350                         wkr_q_conf.priority =
 351                                 RTE_EVENT_DEV_PRIORITY_LOWEST - prio_delta *
 352                                 (i % nb_slots);
 353                 }
 354
 355                 const char *type_str = "Atomic";
 356                 switch (wkr_q_conf.schedule_type) {
 357                 case RTE_SCHED_TYPE_ORDERED:
 358                         type_str = "Ordered";
 359                         break;
 360                 case RTE_SCHED_TYPE_PARALLEL:
 361                         type_str = "Parallel";
 362                         break;
 363                 }
 364                 printf("\tStage %d, Type %s\tPriority = %d\n", i, type_str,
 365                                 wkr_q_conf.priority);
 366         }
 367
 368         printf("\n");
 369         if (wkr_p_conf.dequeue_depth > config.nb_event_port_dequeue_depth)
 370                 wkr_p_conf.dequeue_depth = config.nb_event_port_dequeue_depth;
 371         if (wkr_p_conf.enqueue_depth > config.nb_event_port_enqueue_depth)
 372                 wkr_p_conf.enqueue_depth = config.nb_event_port_enqueue_depth;
 373
 374         /* set up one port per worker, linking to all stage queues */
 375         for (i = 0; i < cdata.num_workers; i++) {
 376                 struct worker_data *w = &worker_data[i];
 377                 w->dev_id = dev_id;
 378                 if (rte_event_port_setup(dev_id, i, &wkr_p_conf) < 0) {
 379                         printf("Error setting up port %d\n", i);
 380                         return -1;
 381                 }
 382
 383                 if (rte_event_port_link(dev_id, i, NULL, NULL, 0)
 384                                 != nb_queues) {
 385                         printf("%d: error creating link for port %d\n",
 386                                         __LINE__, i);
 387                         return -1;
 388                 }
 389                 w->port_id = i;
 390         }
 391         /*
 392          * Reduce the load on ingress event queue by splitting the traffic
 393          * across multiple event queues.
 394          * for example, nb_stages =  2 and nb_ethdev = 2 then
 395          *
 396          *      nb_queues = (2 * 2) + 2 = 6 (non atq)
 397          *      rx_stride = 3
 398          *
 399          * So, traffic is split across queue 0 and queue 3 since queue id for
 400          * rx adapter is chosen <ethport_id> * <rx_stride> i.e in the above
 401          * case eth port 0, 1 will inject packets into event queue 0, 3
 402          * respectively.
 403          *
 404          * This forms two set of queue pipelines 0->1->2->tx and 3->4->5->tx.
 405          */
 406         cdata.rx_stride = atq ? 1 : nb_slots;
 407         ret = rte_event_dev_service_id_get(dev_id,
 408                                 &fdata->evdev_service_id);
 409         if (ret != -ESRCH && ret != 0) {
 410                 printf("Error getting the service ID\n");
 411                 return -1;
 412         }
 413         rte_service_runstate_set(fdata->evdev_service_id, 1);
 414         rte_service_set_runstate_mapped_check(fdata->evdev_service_id, 0);
 415         if (rte_event_dev_start(dev_id) < 0) {
 416                 printf("Error starting eventdev\n");
 417                 return -1;
 418         }
 419
 420         return dev_id;
 421 }
 422
 423
 424 struct rx_adptr_services {
 425         uint16_t nb_rx_adptrs;
 426         uint32_t *rx_adpt_arr;
 427 };
 428
 429 static int32_t
 430 service_rx_adapter(void *arg)
 431 {
 432         int i;
 433         struct rx_adptr_services *adptr_services = arg;
 434
 435         for (i = 0; i < adptr_services->nb_rx_adptrs; i++)
 436                 rte_service_run_iter_on_app_lcore(
 437                                 adptr_services->rx_adpt_arr[i], 1);
 438         return 0;
 439 }
 440
 441 static void
 442 init_rx_adapter(uint16_t nb_ports)
 443 {
 444         int i;
 445         int ret;
 446         uint8_t evdev_id = 0;
 447         struct rx_adptr_services *adptr_services = NULL;
 448         struct rte_event_dev_info dev_info;
 449
 450         ret = rte_event_dev_info_get(evdev_id, &dev_info);
 451         adptr_services = rte_zmalloc(NULL, sizeof(struct rx_adptr_services), 0);
 452
 453         struct rte_event_port_conf rx_p_conf = {
 454                 .dequeue_depth = 8,
 455                 .enqueue_depth = 8,
 456                 .new_event_threshold = 1200,
 457         };
 458
 459         if (rx_p_conf.dequeue_depth > dev_info.max_event_port_dequeue_depth)
 460                 rx_p_conf.dequeue_depth = dev_info.max_event_port_dequeue_depth;
 461         if (rx_p_conf.enqueue_depth > dev_info.max_event_port_enqueue_depth)
 462                 rx_p_conf.enqueue_depth = dev_info.max_event_port_enqueue_depth;
 463
 464
 465         struct rte_event_eth_rx_adapter_queue_conf queue_conf = {
 466                 .ev.sched_type = cdata.queue_type,
 467         };
 468
 469         for (i = 0; i < nb_ports; i++) {
 470                 uint32_t cap;
 471                 uint32_t service_id;
 472
 473                 ret = rte_event_eth_rx_adapter_create(i, evdev_id, &rx_p_conf);
 474                 if (ret)
 475                         rte_exit(EXIT_FAILURE,
 476                                         "failed to create rx adapter[%d]",
 477                                         cdata.rx_adapter_id);
 478
 479                 ret = rte_event_eth_rx_adapter_caps_get(evdev_id, i, &cap);
 480                 if (ret)
 481                         rte_exit(EXIT_FAILURE,
 482                                         "failed to get event rx adapter "
 483                                         "capabilities");
 484
 485                 queue_conf.ev.queue_id = cdata.rx_stride ?
 486                         (i * cdata.rx_stride)
 487                         : (uint8_t)cdata.qid[0];
 488
 489                 ret = rte_event_eth_rx_adapter_queue_add(i, i, -1, &queue_conf);
 490                 if (ret)
 491                         rte_exit(EXIT_FAILURE,
 492                                         "Failed to add queues to Rx adapter");
 493
 494
 495                 /* Producer needs to be scheduled. */
 496                 if (!(cap & RTE_EVENT_ETH_RX_ADAPTER_CAP_INTERNAL_PORT)) {
 497                         ret = rte_event_eth_rx_adapter_service_id_get(i,
 498                                         &service_id);
 499                         if (ret != -ESRCH && ret != 0) {
 500                                 rte_exit(EXIT_FAILURE,
 501                                 "Error getting the service ID for rx adptr\n");
 502                         }
 503
 504                         rte_service_runstate_set(service_id, 1);
 505                         rte_service_set_runstate_mapped_check(service_id, 0);
 506
 507                         adptr_services->nb_rx_adptrs++;
 508                         adptr_services->rx_adpt_arr = rte_realloc(
 509                                         adptr_services->rx_adpt_arr,
 510                                         adptr_services->nb_rx_adptrs *
 511                                         sizeof(uint32_t), 0);
 512                         adptr_services->rx_adpt_arr[
 513                                 adptr_services->nb_rx_adptrs - 1] =
 514                                 service_id;
 515                 }
 516
 517                 ret = rte_event_eth_rx_adapter_start(i);
 518                 if (ret)
 519                         rte_exit(EXIT_FAILURE, "Rx adapter[%d] start failed",
 520                                         cdata.rx_adapter_id);
 521         }
 522
 523         if (adptr_services->nb_rx_adptrs) {
 524                 struct rte_service_spec service;
 525
 526                 memset(&service, 0, sizeof(struct rte_service_spec));
 527                 snprintf(service.name, sizeof(service.name), "rx_service");
 528                 service.callback = service_rx_adapter;
 529                 service.callback_userdata = (void *)adptr_services;
 530
 531                 int32_t ret = rte_service_component_register(&service,
 532                                 &fdata->rxadptr_service_id);
 533                 if (ret)
 534                         rte_exit(EXIT_FAILURE,
 535                                 "Rx adapter[%d] service register failed",
 536                                 cdata.rx_adapter_id);
 537
 538                 rte_service_runstate_set(fdata->rxadptr_service_id, 1);
 539                 rte_service_component_runstate_set(fdata->rxadptr_service_id,
 540                                 1);
 541                 rte_service_set_runstate_mapped_check(fdata->rxadptr_service_id,
 542                                 0);
 543         } else {
 544                 memset(fdata->rx_core, 0, sizeof(unsigned int) * MAX_NUM_CORE);
 545                 rte_free(adptr_services);
 546         }
 547
 548         if (!adptr_services->nb_rx_adptrs && fdata->cap.consumer == NULL &&
 549                         (dev_info.event_dev_cap &
 550                          RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED))
 551                 fdata->cap.scheduler = NULL;
 552
 553         if (dev_info.event_dev_cap & RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED)
 554                 memset(fdata->sched_core, 0,
 555                                 sizeof(unsigned int) * MAX_NUM_CORE);
 556 }
 557
 558 static void
 559 worker_tx_opt_check(void)
 560 {
 561         int i;
 562         int ret;
 563         uint32_t cap = 0;
 564         uint8_t rx_needed = 0;
 565         struct rte_event_dev_info eventdev_info;
 566
 567         memset(&eventdev_info, 0, sizeof(struct rte_event_dev_info));
 568         rte_event_dev_info_get(0, &eventdev_info);
 569
 570         if (cdata.all_type_queues && !(eventdev_info.event_dev_cap &
 571                                 RTE_EVENT_DEV_CAP_QUEUE_ALL_TYPES))
 572                 rte_exit(EXIT_FAILURE,
 573                                 "Event dev doesn't support all type queues\n");
 574
 575         for (i = 0; i < rte_eth_dev_count(); i++) {
 576                 ret = rte_event_eth_rx_adapter_caps_get(0, i, &cap);
 577                 if (ret)
 578                         rte_exit(EXIT_FAILURE,
 579                                         "failed to get event rx adapter "
 580                                         "capabilities");
 581                 rx_needed |=
 582                         !(cap & RTE_EVENT_ETH_RX_ADAPTER_CAP_INTERNAL_PORT);
 583         }
 584
 585         if (cdata.worker_lcore_mask == 0 ||
 586                         (rx_needed && cdata.rx_lcore_mask == 0) ||
 587                         (cdata.sched_lcore_mask == 0 &&
 588                          !(eventdev_info.event_dev_cap &
 589                                  RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED))) {
 590                 printf("Core part of pipeline was not assigned any cores. "
 591                         "This will stall the pipeline, please check core masks "
 592                         "(use -h for details on setting core masks):\n"
 593                         "\trx: %"PRIu64"\n\ttx: %"PRIu64"\n\tsched: %"PRIu64
 594                         "\n\tworkers: %"PRIu64"\n",
 595                         cdata.rx_lcore_mask, cdata.tx_lcore_mask,
 596                         cdata.sched_lcore_mask,
 597                         cdata.worker_lcore_mask);
 598                 rte_exit(-1, "Fix core masks\n");
 599         }
 600 }
 601
 602 static worker_loop
 603 get_worker_loop_burst(uint8_t atq)
 604 {
 605         if (atq)
 606                 return worker_do_tx_burst_atq;
 607
 608         return worker_do_tx_burst;
 609 }
 610
 611 static worker_loop
 612 get_worker_loop_non_burst(uint8_t atq)
 613 {
 614         if (atq)
 615                 return worker_do_tx_atq;
 616
 617         return worker_do_tx;
 618 }
 619
 620 void
 621 set_worker_tx_setup_data(struct setup_data *caps, bool burst)
 622 {
 623         uint8_t atq = cdata.all_type_queues ? 1 : 0;
 624
 625         if (burst)
 626                 caps->worker = get_worker_loop_burst(atq);
 627         else
 628                 caps->worker = get_worker_loop_non_burst(atq);
 629
 630         memset(fdata->tx_core, 0, sizeof(unsigned int) * MAX_NUM_CORE);
 631
 632         caps->check_opt = worker_tx_opt_check;
 633         caps->consumer = NULL;
 634         caps->scheduler = schedule_devices;
 635         caps->evdev_setup = setup_eventdev_worker_tx;
 636         caps->adptr_setup = init_rx_adapter;
 637 }