examples/eventdev_pipeline_sw_pmd/main.c

   1 /*-
   2  *   BSD LICENSE
   3  *
   4  *   Copyright(c) 2016-2017 Intel Corporation. All rights reserved.
   5  *
   6  *   Redistribution and use in source and binary forms, with or without
   7  *   modification, are permitted provided that the following conditions
   8  *   are met:
   9  *
  10  *     * Redistributions of source code must retain the above copyright
  11  *       notice, this list of conditions and the following disclaimer.
  12  *     * Redistributions in binary form must reproduce the above copyright
  13  *       notice, this list of conditions and the following disclaimer in
  14  *       the documentation and/or other materials provided with the
  15  *       distribution.
  16  *     * Neither the name of Intel Corporation nor the names of its
  17  *       contributors may be used to endorse or promote products derived
  18  *       from this software without specific prior written permission.
  19  *
  20  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  21  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  22  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  23  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  24  *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  25  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  26  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  27  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  28  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  29  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  30  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  31  */
  32
  33 #include <getopt.h>
  34 #include <stdint.h>
  35 #include <stdio.h>
  36 #include <signal.h>
  37 #include <sched.h>
  38 #include <stdbool.h>
  39
  40 #include <rte_eal.h>
  41 #include <rte_mempool.h>
  42 #include <rte_mbuf.h>
  43 #include <rte_launch.h>
  44 #include <rte_malloc.h>
  45 #include <rte_random.h>
  46 #include <rte_cycles.h>
  47 #include <rte_ethdev.h>
  48 #include <rte_eventdev.h>
  49 #include <rte_service.h>
  50
  51 #define MAX_NUM_STAGES 8
  52 #define BATCH_SIZE 16
  53 #define MAX_NUM_CORE 64
  54
  55 struct prod_data {
  56         uint8_t dev_id;
  57         uint8_t port_id;
  58         int32_t qid;
  59         unsigned int num_nic_ports;
  60 } __rte_cache_aligned;
  61
  62 struct cons_data {
  63         uint8_t dev_id;
  64         uint8_t port_id;
  65 } __rte_cache_aligned;
  66
  67 static struct prod_data prod_data;
  68 static struct cons_data cons_data;
  69
  70 struct worker_data {
  71         uint8_t dev_id;
  72         uint8_t port_id;
  73 } __rte_cache_aligned;
  74
  75 struct fastpath_data {
  76         volatile int done;
  77         uint32_t rx_lock;
  78         uint32_t tx_lock;
  79         uint32_t sched_lock;
  80         uint32_t evdev_service_id;
  81         bool rx_single;
  82         bool tx_single;
  83         bool sched_single;
  84         unsigned int rx_core[MAX_NUM_CORE];
  85         unsigned int tx_core[MAX_NUM_CORE];
  86         unsigned int sched_core[MAX_NUM_CORE];
  87         unsigned int worker_core[MAX_NUM_CORE];
  88         struct rte_eth_dev_tx_buffer *tx_buf[RTE_MAX_ETHPORTS];
  89 };
  90
  91 static struct fastpath_data *fdata;
  92
  93 struct config_data {
  94         unsigned int active_cores;
  95         unsigned int num_workers;
  96         int64_t num_packets;
  97         unsigned int num_fids;
  98         int queue_type;
  99         int worker_cycles;
 100         int enable_queue_priorities;
 101         int quiet;
 102         int dump_dev;
 103         int dump_dev_signal;
 104         unsigned int num_stages;
 105         unsigned int worker_cq_depth;
 106         int16_t next_qid[MAX_NUM_STAGES+2];
 107         int16_t qid[MAX_NUM_STAGES];
 108 };
 109
 110 static struct config_data cdata = {
 111         .num_packets = (1L << 25), /* do ~32M packets */
 112         .num_fids = 512,
 113         .queue_type = RTE_SCHED_TYPE_ATOMIC,
 114         .next_qid = {-1},
 115         .qid = {-1},
 116         .num_stages = 1,
 117         .worker_cq_depth = 16
 118 };
 119
 120 static bool
 121 core_in_use(unsigned int lcore_id) {
 122         return (fdata->rx_core[lcore_id] || fdata->sched_core[lcore_id] ||
 123                 fdata->tx_core[lcore_id] || fdata->worker_core[lcore_id]);
 124 }
 125
 126 static void
 127 eth_tx_buffer_retry(struct rte_mbuf **pkts, uint16_t unsent,
 128                         void *userdata)
 129 {
 130         int port_id = (uintptr_t) userdata;
 131         unsigned int _sent = 0;
 132
 133         do {
 134                 /* Note: hard-coded TX queue */
 135                 _sent += rte_eth_tx_burst(port_id, 0, &pkts[_sent],
 136                                           unsent - _sent);
 137         } while (_sent != unsent);
 138 }
 139
 140 static int
 141 consumer(void)
 142 {
 143         const uint64_t freq_khz = rte_get_timer_hz() / 1000;
 144         struct rte_event packets[BATCH_SIZE];
 145
 146         static uint64_t received;
 147         static uint64_t last_pkts;
 148         static uint64_t last_time;
 149         static uint64_t start_time;
 150         unsigned int i, j;
 151         uint8_t dev_id = cons_data.dev_id;
 152         uint8_t port_id = cons_data.port_id;
 153
 154         uint16_t n = rte_event_dequeue_burst(dev_id, port_id,
 155                         packets, RTE_DIM(packets), 0);
 156
 157         if (n == 0) {
 158                 for (j = 0; j < rte_eth_dev_count(); j++)
 159                         rte_eth_tx_buffer_flush(j, 0, fdata->tx_buf[j]);
 160                 return 0;
 161         }
 162         if (start_time == 0)
 163                 last_time = start_time = rte_get_timer_cycles();
 164
 165         received += n;
 166         for (i = 0; i < n; i++) {
 167                 uint8_t outport = packets[i].mbuf->port;
 168                 rte_eth_tx_buffer(outport, 0, fdata->tx_buf[outport],
 169                                 packets[i].mbuf);
 170         }
 171
 172         /* Print out mpps every 1<22 packets */
 173         if (!cdata.quiet && received >= last_pkts + (1<<22)) {
 174                 const uint64_t now = rte_get_timer_cycles();
 175                 const uint64_t total_ms = (now - start_time) / freq_khz;
 176                 const uint64_t delta_ms = (now - last_time) / freq_khz;
 177                 uint64_t delta_pkts = received - last_pkts;
 178
 179                 printf("# consumer RX=%"PRIu64", time %"PRIu64 "ms, "
 180                         "avg %.3f mpps [current %.3f mpps]\n",
 181                                 received,
 182                                 total_ms,
 183                                 received / (total_ms * 1000.0),
 184                                 delta_pkts / (delta_ms * 1000.0));
 185                 last_pkts = received;
 186                 last_time = now;
 187         }
 188
 189         cdata.num_packets -= n;
 190         if (cdata.num_packets <= 0)
 191                 fdata->done = 1;
 192
 193         return 0;
 194 }
 195
 196 static int
 197 producer(void)
 198 {
 199         static uint8_t eth_port;
 200         struct rte_mbuf *mbufs[BATCH_SIZE+2];
 201         struct rte_event ev[BATCH_SIZE+2];
 202         uint32_t i, num_ports = prod_data.num_nic_ports;
 203         int32_t qid = prod_data.qid;
 204         uint8_t dev_id = prod_data.dev_id;
 205         uint8_t port_id = prod_data.port_id;
 206         uint32_t prio_idx = 0;
 207
 208         const uint16_t nb_rx = rte_eth_rx_burst(eth_port, 0, mbufs, BATCH_SIZE);
 209         if (++eth_port == num_ports)
 210                 eth_port = 0;
 211         if (nb_rx == 0) {
 212                 rte_pause();
 213                 return 0;
 214         }
 215
 216         for (i = 0; i < nb_rx; i++) {
 217                 ev[i].flow_id = mbufs[i]->hash.rss;
 218                 ev[i].op = RTE_EVENT_OP_NEW;
 219                 ev[i].sched_type = cdata.queue_type;
 220                 ev[i].queue_id = qid;
 221                 ev[i].event_type = RTE_EVENT_TYPE_ETHDEV;
 222                 ev[i].sub_event_type = 0;
 223                 ev[i].priority = RTE_EVENT_DEV_PRIORITY_NORMAL;
 224                 ev[i].mbuf = mbufs[i];
 225                 RTE_SET_USED(prio_idx);
 226         }
 227
 228         const int nb_tx = rte_event_enqueue_burst(dev_id, port_id, ev, nb_rx);
 229         if (nb_tx != nb_rx) {
 230                 for (i = nb_tx; i < nb_rx; i++)
 231                         rte_pktmbuf_free(mbufs[i]);
 232         }
 233
 234         return 0;
 235 }
 236
 237 static inline void
 238 schedule_devices(unsigned int lcore_id)
 239 {
 240         if (fdata->rx_core[lcore_id] && (fdata->rx_single ||
 241             rte_atomic32_cmpset(&(fdata->rx_lock), 0, 1))) {
 242                 producer();
 243                 rte_atomic32_clear((rte_atomic32_t *)&(fdata->rx_lock));
 244         }
 245
 246         if (fdata->sched_core[lcore_id] && (fdata->sched_single ||
 247             rte_atomic32_cmpset(&(fdata->sched_lock), 0, 1))) {
 248                 rte_service_run_iter_on_app_lcore(fdata->evdev_service_id, 1);
 249                 if (cdata.dump_dev_signal) {
 250                         rte_event_dev_dump(0, stdout);
 251                         cdata.dump_dev_signal = 0;
 252                 }
 253                 rte_atomic32_clear((rte_atomic32_t *)&(fdata->sched_lock));
 254         }
 255
 256         if (fdata->tx_core[lcore_id] && (fdata->tx_single ||
 257             rte_atomic32_cmpset(&(fdata->tx_lock), 0, 1))) {
 258                 consumer();
 259                 rte_atomic32_clear((rte_atomic32_t *)&(fdata->tx_lock));
 260         }
 261 }
 262
 263 static inline void
 264 work(struct rte_mbuf *m)
 265 {
 266         struct ether_hdr *eth;
 267         struct ether_addr addr;
 268
 269         /* change mac addresses on packet (to use mbuf data) */
 270         /*
 271          * FIXME Swap mac address properly and also handle the
 272          * case for both odd and even number of stages that the
 273          * addresses end up the same at the end of the pipeline
 274          */
 275         eth = rte_pktmbuf_mtod(m, struct ether_hdr *);
 276         ether_addr_copy(&eth->d_addr, &addr);
 277         ether_addr_copy(&addr, &eth->d_addr);
 278
 279         /* do a number of cycles of work per packet */
 280         volatile uint64_t start_tsc = rte_rdtsc();
 281         while (rte_rdtsc() < start_tsc + cdata.worker_cycles)
 282                 rte_pause();
 283 }
 284
 285 static int
 286 worker(void *arg)
 287 {
 288         struct rte_event events[BATCH_SIZE];
 289
 290         struct worker_data *data = (struct worker_data *)arg;
 291         uint8_t dev_id = data->dev_id;
 292         uint8_t port_id = data->port_id;
 293         size_t sent = 0, received = 0;
 294         unsigned int lcore_id = rte_lcore_id();
 295
 296         while (!fdata->done) {
 297                 uint16_t i;
 298
 299                 schedule_devices(lcore_id);
 300
 301                 if (!fdata->worker_core[lcore_id]) {
 302                         rte_pause();
 303                         continue;
 304                 }
 305
 306                 const uint16_t nb_rx = rte_event_dequeue_burst(dev_id, port_id,
 307                                 events, RTE_DIM(events), 0);
 308
 309                 if (nb_rx == 0) {
 310                         rte_pause();
 311                         continue;
 312                 }
 313                 received += nb_rx;
 314
 315                 for (i = 0; i < nb_rx; i++) {
 316
 317                         /* The first worker stage does classification */
 318                         if (events[i].queue_id == cdata.qid[0])
 319                                 events[i].flow_id = events[i].mbuf->hash.rss
 320                                                         % cdata.num_fids;
 321
 322                         events[i].queue_id = cdata.next_qid[events[i].queue_id];
 323                         events[i].op = RTE_EVENT_OP_FORWARD;
 324                         events[i].sched_type = cdata.queue_type;
 325
 326                         work(events[i].mbuf);
 327                 }
 328                 uint16_t nb_tx = rte_event_enqueue_burst(dev_id, port_id,
 329                                 events, nb_rx);
 330                 while (nb_tx < nb_rx && !fdata->done)
 331                         nb_tx += rte_event_enqueue_burst(dev_id, port_id,
 332                                                         events + nb_tx,
 333                                                         nb_rx - nb_tx);
 334                 sent += nb_tx;
 335         }
 336
 337         if (!cdata.quiet)
 338                 printf("  worker %u thread done. RX=%zu TX=%zu\n",
 339                                 rte_lcore_id(), received, sent);
 340
 341         return 0;
 342 }
 343
 344 /*
 345  * Parse the coremask given as argument (hexadecimal string) and fill
 346  * the global configuration (core role and core count) with the parsed
 347  * value.
 348  */
 349 static int xdigit2val(unsigned char c)
 350 {
 351         int val;
 352
 353         if (isdigit(c))
 354                 val = c - '0';
 355         else if (isupper(c))
 356                 val = c - 'A' + 10;
 357         else
 358                 val = c - 'a' + 10;
 359         return val;
 360 }
 361
 362 static uint64_t
 363 parse_coremask(const char *coremask)
 364 {
 365         int i, j, idx = 0;
 366         unsigned int count = 0;
 367         char c;
 368         int val;
 369         uint64_t mask = 0;
 370         const int32_t BITS_HEX = 4;
 371
 372         if (coremask == NULL)
 373                 return -1;
 374         /* Remove all blank characters ahead and after .
 375          * Remove 0x/0X if exists.
 376          */
 377         while (isblank(*coremask))
 378                 coremask++;
 379         if (coremask[0] == '0' && ((coremask[1] == 'x')
 380                 || (coremask[1] == 'X')))
 381                 coremask += 2;
 382         i = strlen(coremask);
 383         while ((i > 0) && isblank(coremask[i - 1]))
 384                 i--;
 385         if (i == 0)
 386                 return -1;
 387
 388         for (i = i - 1; i >= 0 && idx < MAX_NUM_CORE; i--) {
 389                 c = coremask[i];
 390                 if (isxdigit(c) == 0) {
 391                         /* invalid characters */
 392                         return -1;
 393                 }
 394                 val = xdigit2val(c);
 395                 for (j = 0; j < BITS_HEX && idx < MAX_NUM_CORE; j++, idx++) {
 396                         if ((1 << j) & val) {
 397                                 mask |= (1UL << idx);
 398                                 count++;
 399                         }
 400                 }
 401         }
 402         for (; i >= 0; i--)
 403                 if (coremask[i] != '0')
 404                         return -1;
 405         if (count == 0)
 406                 return -1;
 407         return mask;
 408 }
 409
 410 static struct option long_options[] = {
 411         {"workers", required_argument, 0, 'w'},
 412         {"packets", required_argument, 0, 'n'},
 413         {"atomic-flows", required_argument, 0, 'f'},
 414         {"num_stages", required_argument, 0, 's'},
 415         {"rx-mask", required_argument, 0, 'r'},
 416         {"tx-mask", required_argument, 0, 't'},
 417         {"sched-mask", required_argument, 0, 'e'},
 418         {"cq-depth", required_argument, 0, 'c'},
 419         {"work-cycles", required_argument, 0, 'W'},
 420         {"queue-priority", no_argument, 0, 'P'},
 421         {"parallel", no_argument, 0, 'p'},
 422         {"ordered", no_argument, 0, 'o'},
 423         {"quiet", no_argument, 0, 'q'},
 424         {"dump", no_argument, 0, 'D'},
 425         {0, 0, 0, 0}
 426 };
 427
 428 static void
 429 usage(void)
 430 {
 431         const char *usage_str =
 432                 "  Usage: eventdev_demo [options]\n"
 433                 "  Options:\n"
 434                 "  -n, --packets=N              Send N packets (default ~32M), 0 implies no limit\n"
 435                 "  -f, --atomic-flows=N         Use N random flows from 1 to N (default 16)\n"
 436                 "  -s, --num_stages=N           Use N atomic stages (default 1)\n"
 437                 "  -r, --rx-mask=core mask      Run NIC rx on CPUs in core mask\n"
 438                 "  -w, --worker-mask=core mask  Run worker on CPUs in core mask\n"
 439                 "  -t, --tx-mask=core mask      Run NIC tx on CPUs in core mask\n"
 440                 "  -e  --sched-mask=core mask   Run scheduler on CPUs in core mask\n"
 441                 "  -c  --cq-depth=N             Worker CQ depth (default 16)\n"
 442                 "  -W  --work-cycles=N          Worker cycles (default 0)\n"
 443                 "  -P  --queue-priority         Enable scheduler queue prioritization\n"
 444                 "  -o, --ordered                Use ordered scheduling\n"
 445                 "  -p, --parallel               Use parallel scheduling\n"
 446                 "  -q, --quiet                  Minimize printed output\n"
 447                 "  -D, --dump                   Print detailed statistics before exit"
 448                 "\n";
 449         fprintf(stderr, "%s", usage_str);
 450         exit(1);
 451 }
 452
 453 static void
 454 parse_app_args(int argc, char **argv)
 455 {
 456         /* Parse cli options*/
 457         int option_index;
 458         int c;
 459         opterr = 0;
 460         uint64_t rx_lcore_mask = 0;
 461         uint64_t tx_lcore_mask = 0;
 462         uint64_t sched_lcore_mask = 0;
 463         uint64_t worker_lcore_mask = 0;
 464         int i;
 465
 466         for (;;) {
 467                 c = getopt_long(argc, argv, "r:t:e:c:w:n:f:s:poPqDW:",
 468                                 long_options, &option_index);
 469                 if (c == -1)
 470                         break;
 471
 472                 int popcnt = 0;
 473                 switch (c) {
 474                 case 'n':
 475                         cdata.num_packets = (int64_t)atol(optarg);
 476                         if (cdata.num_packets == 0)
 477                                 cdata.num_packets = INT64_MAX;
 478                         break;
 479                 case 'f':
 480                         cdata.num_fids = (unsigned int)atoi(optarg);
 481                         break;
 482                 case 's':
 483                         cdata.num_stages = (unsigned int)atoi(optarg);
 484                         break;
 485                 case 'c':
 486                         cdata.worker_cq_depth = (unsigned int)atoi(optarg);
 487                         break;
 488                 case 'W':
 489                         cdata.worker_cycles = (unsigned int)atoi(optarg);
 490                         break;
 491                 case 'P':
 492                         cdata.enable_queue_priorities = 1;
 493                         break;
 494                 case 'o':
 495                         cdata.queue_type = RTE_SCHED_TYPE_ORDERED;
 496                         break;
 497                 case 'p':
 498                         cdata.queue_type = RTE_SCHED_TYPE_PARALLEL;
 499                         break;
 500                 case 'q':
 501                         cdata.quiet = 1;
 502                         break;
 503                 case 'D':
 504                         cdata.dump_dev = 1;
 505                         break;
 506                 case 'w':
 507                         worker_lcore_mask = parse_coremask(optarg);
 508                         break;
 509                 case 'r':
 510                         rx_lcore_mask = parse_coremask(optarg);
 511                         popcnt = __builtin_popcountll(rx_lcore_mask);
 512                         fdata->rx_single = (popcnt == 1);
 513                         break;
 514                 case 't':
 515                         tx_lcore_mask = parse_coremask(optarg);
 516                         popcnt = __builtin_popcountll(tx_lcore_mask);
 517                         fdata->tx_single = (popcnt == 1);
 518                         break;
 519                 case 'e':
 520                         sched_lcore_mask = parse_coremask(optarg);
 521                         popcnt = __builtin_popcountll(sched_lcore_mask);
 522                         fdata->sched_single = (popcnt == 1);
 523                         break;
 524                 default:
 525                         usage();
 526                 }
 527         }
 528
 529         if (worker_lcore_mask == 0 || rx_lcore_mask == 0 ||
 530             sched_lcore_mask == 0 || tx_lcore_mask == 0) {
 531                 printf("Core part of pipeline was not assigned any cores. "
 532                         "This will stall the pipeline, please check core masks "
 533                         "(use -h for details on setting core masks):\n"
 534                         "\trx: %"PRIu64"\n\ttx: %"PRIu64"\n\tsched: %"PRIu64
 535                         "\n\tworkers: %"PRIu64"\n",
 536                         rx_lcore_mask, tx_lcore_mask, sched_lcore_mask,
 537                         worker_lcore_mask);
 538                 rte_exit(-1, "Fix core masks\n");
 539         }
 540         if (cdata.num_stages == 0 || cdata.num_stages > MAX_NUM_STAGES)
 541                 usage();
 542
 543         for (i = 0; i < MAX_NUM_CORE; i++) {
 544                 fdata->rx_core[i] = !!(rx_lcore_mask & (1UL << i));
 545                 fdata->tx_core[i] = !!(tx_lcore_mask & (1UL << i));
 546                 fdata->sched_core[i] = !!(sched_lcore_mask & (1UL << i));
 547                 fdata->worker_core[i] = !!(worker_lcore_mask & (1UL << i));
 548
 549                 if (fdata->worker_core[i])
 550                         cdata.num_workers++;
 551                 if (core_in_use(i))
 552                         cdata.active_cores++;
 553         }
 554 }
 555
 556 /*
 557  * Initializes a given port using global settings and with the RX buffers
 558  * coming from the mbuf_pool passed as a parameter.
 559  */
 560 static inline int
 561 port_init(uint8_t port, struct rte_mempool *mbuf_pool)
 562 {
 563         static const struct rte_eth_conf port_conf_default = {
 564                 .rxmode = {
 565                         .mq_mode = ETH_MQ_RX_RSS,
 566                         .max_rx_pkt_len = ETHER_MAX_LEN
 567                 },
 568                 .rx_adv_conf = {
 569                         .rss_conf = {
 570                                 .rss_hf = ETH_RSS_IP |
 571                                           ETH_RSS_TCP |
 572                                           ETH_RSS_UDP,
 573                         }
 574                 }
 575         };
 576         const uint16_t rx_rings = 1, tx_rings = 1;
 577         const uint16_t rx_ring_size = 512, tx_ring_size = 512;
 578         struct rte_eth_conf port_conf = port_conf_default;
 579         int retval;
 580         uint16_t q;
 581
 582         if (port >= rte_eth_dev_count())
 583                 return -1;
 584
 585         /* Configure the Ethernet device. */
 586         retval = rte_eth_dev_configure(port, rx_rings, tx_rings, &port_conf);
 587         if (retval != 0)
 588                 return retval;
 589
 590         /* Allocate and set up 1 RX queue per Ethernet port. */
 591         for (q = 0; q < rx_rings; q++) {
 592                 retval = rte_eth_rx_queue_setup(port, q, rx_ring_size,
 593                                 rte_eth_dev_socket_id(port), NULL, mbuf_pool);
 594                 if (retval < 0)
 595                         return retval;
 596         }
 597
 598         /* Allocate and set up 1 TX queue per Ethernet port. */
 599         for (q = 0; q < tx_rings; q++) {
 600                 retval = rte_eth_tx_queue_setup(port, q, tx_ring_size,
 601                                 rte_eth_dev_socket_id(port), NULL);
 602                 if (retval < 0)
 603                         return retval;
 604         }
 605
 606         /* Start the Ethernet port. */
 607         retval = rte_eth_dev_start(port);
 608         if (retval < 0)
 609                 return retval;
 610
 611         /* Display the port MAC address. */
 612         struct ether_addr addr;
 613         rte_eth_macaddr_get(port, &addr);
 614         printf("Port %u MAC: %02" PRIx8 " %02" PRIx8 " %02" PRIx8
 615                            " %02" PRIx8 " %02" PRIx8 " %02" PRIx8 "\n",
 616                         (unsigned int)port,
 617                         addr.addr_bytes[0], addr.addr_bytes[1],
 618                         addr.addr_bytes[2], addr.addr_bytes[3],
 619                         addr.addr_bytes[4], addr.addr_bytes[5]);
 620
 621         /* Enable RX in promiscuous mode for the Ethernet device. */
 622         rte_eth_promiscuous_enable(port);
 623
 624         return 0;
 625 }
 626
 627 static int
 628 init_ports(unsigned int num_ports)
 629 {
 630         uint8_t portid;
 631         unsigned int i;
 632
 633         struct rte_mempool *mp = rte_pktmbuf_pool_create("packet_pool",
 634                         /* mbufs */ 16384 * num_ports,
 635                         /* cache_size */ 512,
 636                         /* priv_size*/ 0,
 637                         /* data_room_size */ RTE_MBUF_DEFAULT_BUF_SIZE,
 638                         rte_socket_id());
 639
 640         for (portid = 0; portid < num_ports; portid++)
 641                 if (port_init(portid, mp) != 0)
 642                         rte_exit(EXIT_FAILURE, "Cannot init port %"PRIu8 "\n",
 643                                         portid);
 644
 645         for (i = 0; i < num_ports; i++) {
 646                 void *userdata = (void *)(uintptr_t) i;
 647                 fdata->tx_buf[i] =
 648                         rte_malloc(NULL, RTE_ETH_TX_BUFFER_SIZE(32), 0);
 649                 if (fdata->tx_buf[i] == NULL)
 650                         rte_panic("Out of memory\n");
 651                 rte_eth_tx_buffer_init(fdata->tx_buf[i], 32);
 652                 rte_eth_tx_buffer_set_err_callback(fdata->tx_buf[i],
 653                                                    eth_tx_buffer_retry,
 654                                                    userdata);
 655         }
 656
 657         return 0;
 658 }
 659
 660 struct port_link {
 661         uint8_t queue_id;
 662         uint8_t priority;
 663 };
 664
 665 static int
 666 setup_eventdev(struct prod_data *prod_data,
 667                 struct cons_data *cons_data,
 668                 struct worker_data *worker_data)
 669 {
 670         const uint8_t dev_id = 0;
 671         /* +1 stages is for a SINGLE_LINK TX stage */
 672         const uint8_t nb_queues = cdata.num_stages + 1;
 673         /* + 2 is one port for producer and one for consumer */
 674         const uint8_t nb_ports = cdata.num_workers + 2;
 675         struct rte_event_dev_config config = {
 676                         .nb_event_queues = nb_queues,
 677                         .nb_event_ports = nb_ports,
 678                         .nb_events_limit  = 4096,
 679                         .nb_event_queue_flows = 1024,
 680                         .nb_event_port_dequeue_depth = 128,
 681                         .nb_event_port_enqueue_depth = 128,
 682         };
 683         struct rte_event_port_conf wkr_p_conf = {
 684                         .dequeue_depth = cdata.worker_cq_depth,
 685                         .enqueue_depth = 64,
 686                         .new_event_threshold = 4096,
 687         };
 688         struct rte_event_queue_conf wkr_q_conf = {
 689                         .schedule_type = cdata.queue_type,
 690                         .priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
 691                         .nb_atomic_flows = 1024,
 692                         .nb_atomic_order_sequences = 1024,
 693         };
 694         struct rte_event_port_conf tx_p_conf = {
 695                         .dequeue_depth = 128,
 696                         .enqueue_depth = 128,
 697                         .new_event_threshold = 4096,
 698         };
 699         const struct rte_event_queue_conf tx_q_conf = {
 700                         .priority = RTE_EVENT_DEV_PRIORITY_HIGHEST,
 701                         .event_queue_cfg = RTE_EVENT_QUEUE_CFG_SINGLE_LINK,
 702         };
 703
 704         struct port_link worker_queues[MAX_NUM_STAGES];
 705         struct port_link tx_queue;
 706         unsigned int i;
 707
 708         int ret, ndev = rte_event_dev_count();
 709         if (ndev < 1) {
 710                 printf("%d: No Eventdev Devices Found\n", __LINE__);
 711                 return -1;
 712         }
 713
 714         struct rte_event_dev_info dev_info;
 715         ret = rte_event_dev_info_get(dev_id, &dev_info);
 716         printf("\tEventdev %d: %s\n", dev_id, dev_info.driver_name);
 717
 718         if (dev_info.max_event_port_dequeue_depth <
 719                         config.nb_event_port_dequeue_depth)
 720                 config.nb_event_port_dequeue_depth =
 721                                 dev_info.max_event_port_dequeue_depth;
 722         if (dev_info.max_event_port_enqueue_depth <
 723                         config.nb_event_port_enqueue_depth)
 724                 config.nb_event_port_enqueue_depth =
 725                                 dev_info.max_event_port_enqueue_depth;
 726
 727         ret = rte_event_dev_configure(dev_id, &config);
 728         if (ret < 0) {
 729                 printf("%d: Error configuring device\n", __LINE__);
 730                 return -1;
 731         }
 732
 733         /* Q creation - one load balanced per pipeline stage*/
 734         printf("  Stages:\n");
 735         for (i = 0; i < cdata.num_stages; i++) {
 736                 if (rte_event_queue_setup(dev_id, i, &wkr_q_conf) < 0) {
 737                         printf("%d: error creating qid %d\n", __LINE__, i);
 738                         return -1;
 739                 }
 740                 cdata.qid[i] = i;
 741                 cdata.next_qid[i] = i+1;
 742                 worker_queues[i].queue_id = i;
 743                 if (cdata.enable_queue_priorities) {
 744                         /* calculate priority stepping for each stage, leaving
 745                          * headroom of 1 for the SINGLE_LINK TX below
 746                          */
 747                         const uint32_t prio_delta =
 748                                 (RTE_EVENT_DEV_PRIORITY_LOWEST-1) /  nb_queues;
 749
 750                         /* higher priority for queues closer to tx */
 751                         wkr_q_conf.priority =
 752                                 RTE_EVENT_DEV_PRIORITY_LOWEST - prio_delta * i;
 753                 }
 754
 755                 const char *type_str = "Atomic";
 756                 switch (wkr_q_conf.schedule_type) {
 757                 case RTE_SCHED_TYPE_ORDERED:
 758                         type_str = "Ordered";
 759                         break;
 760                 case RTE_SCHED_TYPE_PARALLEL:
 761                         type_str = "Parallel";
 762                         break;
 763                 }
 764                 printf("\tStage %d, Type %s\tPriority = %d\n", i, type_str,
 765                                 wkr_q_conf.priority);
 766         }
 767         printf("\n");
 768
 769         /* final queue for sending to TX core */
 770         if (rte_event_queue_setup(dev_id, i, &tx_q_conf) < 0) {
 771                 printf("%d: error creating qid %d\n", __LINE__, i);
 772                 return -1;
 773         }
 774         tx_queue.queue_id = i;
 775         tx_queue.priority = RTE_EVENT_DEV_PRIORITY_HIGHEST;
 776
 777         if (wkr_p_conf.dequeue_depth > config.nb_event_port_dequeue_depth)
 778                 wkr_p_conf.dequeue_depth = config.nb_event_port_dequeue_depth;
 779         if (wkr_p_conf.enqueue_depth > config.nb_event_port_enqueue_depth)
 780                 wkr_p_conf.enqueue_depth = config.nb_event_port_enqueue_depth;
 781
 782         /* set up one port per worker, linking to all stage queues */
 783         for (i = 0; i < cdata.num_workers; i++) {
 784                 struct worker_data *w = &worker_data[i];
 785                 w->dev_id = dev_id;
 786                 if (rte_event_port_setup(dev_id, i, &wkr_p_conf) < 0) {
 787                         printf("Error setting up port %d\n", i);
 788                         return -1;
 789                 }
 790
 791                 uint32_t s;
 792                 for (s = 0; s < cdata.num_stages; s++) {
 793                         if (rte_event_port_link(dev_id, i,
 794                                                 &worker_queues[s].queue_id,
 795                                                 &worker_queues[s].priority,
 796                                                 1) != 1) {
 797                                 printf("%d: error creating link for port %d\n",
 798                                                 __LINE__, i);
 799                                 return -1;
 800                         }
 801                 }
 802                 w->port_id = i;
 803         }
 804
 805         if (tx_p_conf.dequeue_depth > config.nb_event_port_dequeue_depth)
 806                 tx_p_conf.dequeue_depth = config.nb_event_port_dequeue_depth;
 807         if (tx_p_conf.enqueue_depth > config.nb_event_port_enqueue_depth)
 808                 tx_p_conf.enqueue_depth = config.nb_event_port_enqueue_depth;
 809
 810         /* port for consumer, linked to TX queue */
 811         if (rte_event_port_setup(dev_id, i, &tx_p_conf) < 0) {
 812                 printf("Error setting up port %d\n", i);
 813                 return -1;
 814         }
 815         if (rte_event_port_link(dev_id, i, &tx_queue.queue_id,
 816                                 &tx_queue.priority, 1) != 1) {
 817                 printf("%d: error creating link for port %d\n",
 818                                 __LINE__, i);
 819                 return -1;
 820         }
 821         /* port for producer, no links */
 822         struct rte_event_port_conf rx_p_conf = {
 823                         .dequeue_depth = 8,
 824                         .enqueue_depth = 8,
 825                         .new_event_threshold = 1200,
 826         };
 827
 828         if (rx_p_conf.dequeue_depth > config.nb_event_port_dequeue_depth)
 829                 rx_p_conf.dequeue_depth = config.nb_event_port_dequeue_depth;
 830         if (rx_p_conf.enqueue_depth > config.nb_event_port_enqueue_depth)
 831                 rx_p_conf.enqueue_depth = config.nb_event_port_enqueue_depth;
 832
 833         if (rte_event_port_setup(dev_id, i + 1, &rx_p_conf) < 0) {
 834                 printf("Error setting up port %d\n", i);
 835                 return -1;
 836         }
 837
 838         *prod_data = (struct prod_data){.dev_id = dev_id,
 839                                         .port_id = i + 1,
 840                                         .qid = cdata.qid[0] };
 841         *cons_data = (struct cons_data){.dev_id = dev_id,
 842                                         .port_id = i };
 843
 844         ret = rte_event_dev_service_id_get(dev_id,
 845                                 &fdata->evdev_service_id);
 846         if (ret != -ESRCH && ret != 0) {
 847                 printf("Error getting the service ID for sw eventdev\n");
 848                 return -1;
 849         }
 850         rte_service_runstate_set(fdata->evdev_service_id, 1);
 851         rte_service_set_runstate_mapped_check(fdata->evdev_service_id, 0);
 852         if (rte_event_dev_start(dev_id) < 0) {
 853                 printf("Error starting eventdev\n");
 854                 return -1;
 855         }
 856
 857         return dev_id;
 858 }
 859
 860 static void
 861 signal_handler(int signum)
 862 {
 863         if (fdata->done)
 864                 rte_exit(1, "Exiting on signal %d\n", signum);
 865         if (signum == SIGINT || signum == SIGTERM) {
 866                 printf("\n\nSignal %d received, preparing to exit...\n",
 867                                 signum);
 868                 fdata->done = 1;
 869         }
 870         if (signum == SIGTSTP)
 871                 rte_event_dev_dump(0, stdout);
 872 }
 873
 874 static inline uint64_t
 875 port_stat(int dev_id, int32_t p)
 876 {
 877         char statname[64];
 878         snprintf(statname, sizeof(statname), "port_%u_rx", p);
 879         return rte_event_dev_xstats_by_name_get(dev_id, statname, NULL);
 880 }
 881
 882 int
 883 main(int argc, char **argv)
 884 {
 885         struct worker_data *worker_data;
 886         unsigned int num_ports;
 887         int lcore_id;
 888         int err;
 889
 890         signal(SIGINT, signal_handler);
 891         signal(SIGTERM, signal_handler);
 892         signal(SIGTSTP, signal_handler);
 893
 894         err = rte_eal_init(argc, argv);
 895         if (err < 0)
 896                 rte_panic("Invalid EAL arguments\n");
 897
 898         argc -= err;
 899         argv += err;
 900
 901         fdata = rte_malloc(NULL, sizeof(struct fastpath_data), 0);
 902         if (fdata == NULL)
 903                 rte_panic("Out of memory\n");
 904
 905         /* Parse cli options*/
 906         parse_app_args(argc, argv);
 907
 908         num_ports = rte_eth_dev_count();
 909         if (num_ports == 0)
 910                 rte_panic("No ethernet ports found\n");
 911
 912         const unsigned int cores_needed = cdata.active_cores;
 913
 914         if (!cdata.quiet) {
 915                 printf("  Config:\n");
 916                 printf("\tports: %u\n", num_ports);
 917                 printf("\tworkers: %u\n", cdata.num_workers);
 918                 printf("\tpackets: %"PRIi64"\n", cdata.num_packets);
 919                 printf("\tQueue-prio: %u\n", cdata.enable_queue_priorities);
 920                 if (cdata.queue_type == RTE_SCHED_TYPE_ORDERED)
 921                         printf("\tqid0 type: ordered\n");
 922                 if (cdata.queue_type == RTE_SCHED_TYPE_ATOMIC)
 923                         printf("\tqid0 type: atomic\n");
 924                 printf("\tCores available: %u\n", rte_lcore_count());
 925                 printf("\tCores used: %u\n", cores_needed);
 926         }
 927
 928         if (rte_lcore_count() < cores_needed)
 929                 rte_panic("Too few cores (%d < %d)\n", rte_lcore_count(),
 930                                 cores_needed);
 931
 932         const unsigned int ndevs = rte_event_dev_count();
 933         if (ndevs == 0)
 934                 rte_panic("No dev_id devs found. Pasl in a --vdev eventdev.\n");
 935         if (ndevs > 1)
 936                 fprintf(stderr, "Warning: More than one eventdev, using idx 0");
 937
 938         worker_data = rte_calloc(0, cdata.num_workers,
 939                         sizeof(worker_data[0]), 0);
 940         if (worker_data == NULL)
 941                 rte_panic("rte_calloc failed\n");
 942
 943         int dev_id = setup_eventdev(&prod_data, &cons_data, worker_data);
 944         if (dev_id < 0)
 945                 rte_exit(EXIT_FAILURE, "Error setting up eventdev\n");
 946
 947         prod_data.num_nic_ports = num_ports;
 948         init_ports(num_ports);
 949
 950         int worker_idx = 0;
 951         RTE_LCORE_FOREACH_SLAVE(lcore_id) {
 952                 if (lcore_id >= MAX_NUM_CORE)
 953                         break;
 954
 955                 if (!fdata->rx_core[lcore_id] &&
 956                         !fdata->worker_core[lcore_id] &&
 957                         !fdata->tx_core[lcore_id] &&
 958                         !fdata->sched_core[lcore_id])
 959                         continue;
 960
 961                 if (fdata->rx_core[lcore_id])
 962                         printf(
 963                                 "[%s()] lcore %d executing NIC Rx, and using eventdev port %u\n",
 964                                 __func__, lcore_id, prod_data.port_id);
 965
 966                 if (fdata->tx_core[lcore_id])
 967                         printf(
 968                                 "[%s()] lcore %d executing NIC Tx, and using eventdev port %u\n",
 969                                 __func__, lcore_id, cons_data.port_id);
 970
 971                 if (fdata->sched_core[lcore_id])
 972                         printf("[%s()] lcore %d executing scheduler\n",
 973                                         __func__, lcore_id);
 974
 975                 if (fdata->worker_core[lcore_id])
 976                         printf(
 977                                 "[%s()] lcore %d executing worker, using eventdev port %u\n",
 978                                 __func__, lcore_id,
 979                                 worker_data[worker_idx].port_id);
 980
 981                 err = rte_eal_remote_launch(worker, &worker_data[worker_idx],
 982                                             lcore_id);
 983                 if (err) {
 984                         rte_panic("Failed to launch worker on core %d\n",
 985                                         lcore_id);
 986                         continue;
 987                 }
 988                 if (fdata->worker_core[lcore_id])
 989                         worker_idx++;
 990         }
 991
 992         lcore_id = rte_lcore_id();
 993
 994         if (core_in_use(lcore_id))
 995                 worker(&worker_data[worker_idx++]);
 996
 997         rte_eal_mp_wait_lcore();
 998
 999         if (cdata.dump_dev)
1000                 rte_event_dev_dump(dev_id, stdout);
1001
1002         if (!cdata.quiet && (port_stat(dev_id, worker_data[0].port_id) !=
1003                         (uint64_t)-ENOTSUP)) {
1004                 printf("\nPort Workload distribution:\n");
1005                 uint32_t i;
1006                 uint64_t tot_pkts = 0;
1007                 uint64_t pkts_per_wkr[RTE_MAX_LCORE] = {0};
1008                 for (i = 0; i < cdata.num_workers; i++) {
1009                         pkts_per_wkr[i] =
1010                                 port_stat(dev_id, worker_data[i].port_id);
1011                         tot_pkts += pkts_per_wkr[i];
1012                 }
1013                 for (i = 0; i < cdata.num_workers; i++) {
1014                         float pc = pkts_per_wkr[i]  * 100 /
1015                                 ((float)tot_pkts);
1016                         printf("worker %i :\t%.1f %% (%"PRIu64" pkts)\n",
1017                                         i, pc, pkts_per_wkr[i]);
1018                 }
1019
1020         }
1021
1022         return 0;
1023 }