examples/distributor/main.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2010-2017 Intel Corporation
   3  */
   4
   5 #include <stdint.h>
   6 #include <inttypes.h>
   7 #include <unistd.h>
   8 #include <signal.h>
   9 #include <getopt.h>
  10
  11 #include <rte_eal.h>
  12 #include <rte_ethdev.h>
  13 #include <rte_cycles.h>
  14 #include <rte_malloc.h>
  15 #include <rte_debug.h>
  16 #include <rte_prefetch.h>
  17 #include <rte_distributor.h>
  18 #include <rte_pause.h>
  19 #include <rte_power.h>
  20
  21 #define RX_RING_SIZE 1024
  22 #define TX_RING_SIZE 1024
  23 #define NUM_MBUFS ((64*1024)-1)
  24 #define MBUF_CACHE_SIZE 128
  25 #define BURST_SIZE 64
  26 #define SCHED_RX_RING_SZ 8192
  27 #define SCHED_TX_RING_SZ 65536
  28 #define BURST_SIZE_TX 32
  29
  30 #define RTE_LOGTYPE_DISTRAPP RTE_LOGTYPE_USER1
  31
  32 #define ANSI_COLOR_RED     "\x1b[31m"
  33 #define ANSI_COLOR_RESET   "\x1b[0m"
  34
  35 /* mask of enabled ports */
  36 static uint32_t enabled_port_mask;
  37 volatile uint8_t quit_signal;
  38 volatile uint8_t quit_signal_rx;
  39 volatile uint8_t quit_signal_dist;
  40 volatile uint8_t quit_signal_work;
  41 unsigned int power_lib_initialised;
  42
  43 static volatile struct app_stats {
  44         struct {
  45                 uint64_t rx_pkts;
  46                 uint64_t returned_pkts;
  47                 uint64_t enqueued_pkts;
  48                 uint64_t enqdrop_pkts;
  49         } rx __rte_cache_aligned;
  50         int pad1 __rte_cache_aligned;
  51
  52         struct {
  53                 uint64_t in_pkts;
  54                 uint64_t ret_pkts;
  55                 uint64_t sent_pkts;
  56                 uint64_t enqdrop_pkts;
  57         } dist __rte_cache_aligned;
  58         int pad2 __rte_cache_aligned;
  59
  60         struct {
  61                 uint64_t dequeue_pkts;
  62                 uint64_t tx_pkts;
  63                 uint64_t enqdrop_pkts;
  64         } tx __rte_cache_aligned;
  65         int pad3 __rte_cache_aligned;
  66
  67         uint64_t worker_pkts[64] __rte_cache_aligned;
  68
  69         int pad4 __rte_cache_aligned;
  70
  71         uint64_t worker_bursts[64][8] __rte_cache_aligned;
  72
  73         int pad5 __rte_cache_aligned;
  74
  75         uint64_t port_rx_pkts[64] __rte_cache_aligned;
  76         uint64_t port_tx_pkts[64] __rte_cache_aligned;
  77 } app_stats;
  78
  79 struct app_stats prev_app_stats;
  80
  81 static const struct rte_eth_conf port_conf_default = {
  82         .rxmode = {
  83                 .mq_mode = ETH_MQ_RX_RSS,
  84                 .max_rx_pkt_len = RTE_ETHER_MAX_LEN,
  85         },
  86         .txmode = {
  87                 .mq_mode = ETH_MQ_TX_NONE,
  88         },
  89         .rx_adv_conf = {
  90                 .rss_conf = {
  91                         .rss_hf = ETH_RSS_IP | ETH_RSS_UDP |
  92                                 ETH_RSS_TCP | ETH_RSS_SCTP,
  93                 }
  94         },
  95 };
  96
  97 struct output_buffer {
  98         unsigned count;
  99         struct rte_mbuf *mbufs[BURST_SIZE];
 100 };
 101
 102 static void print_stats(void);
 103
 104 /*
 105  * Initialises a given port using global settings and with the rx buffers
 106  * coming from the mbuf_pool passed as parameter
 107  */
 108 static inline int
 109 port_init(uint16_t port, struct rte_mempool *mbuf_pool)
 110 {
 111         struct rte_eth_conf port_conf = port_conf_default;
 112         const uint16_t rxRings = 1, txRings = rte_lcore_count() - 1;
 113         int retval;
 114         uint16_t q;
 115         uint16_t nb_rxd = RX_RING_SIZE;
 116         uint16_t nb_txd = TX_RING_SIZE;
 117         struct rte_eth_dev_info dev_info;
 118         struct rte_eth_txconf txconf;
 119
 120         if (!rte_eth_dev_is_valid_port(port))
 121                 return -1;
 122
 123         retval = rte_eth_dev_info_get(port, &dev_info);
 124         if (retval != 0) {
 125                 printf("Error during getting device (port %u) info: %s\n",
 126                                 port, strerror(-retval));
 127                 return retval;
 128         }
 129
 130         if (dev_info.tx_offload_capa & DEV_TX_OFFLOAD_MBUF_FAST_FREE)
 131                 port_conf.txmode.offloads |=
 132                         DEV_TX_OFFLOAD_MBUF_FAST_FREE;
 133
 134         port_conf.rx_adv_conf.rss_conf.rss_hf &=
 135                 dev_info.flow_type_rss_offloads;
 136         if (port_conf.rx_adv_conf.rss_conf.rss_hf !=
 137                         port_conf_default.rx_adv_conf.rss_conf.rss_hf) {
 138                 printf("Port %u modified RSS hash function based on hardware support,"
 139                         "requested:%#"PRIx64" configured:%#"PRIx64"\n",
 140                         port,
 141                         port_conf_default.rx_adv_conf.rss_conf.rss_hf,
 142                         port_conf.rx_adv_conf.rss_conf.rss_hf);
 143         }
 144
 145         retval = rte_eth_dev_configure(port, rxRings, txRings, &port_conf);
 146         if (retval != 0)
 147                 return retval;
 148
 149         retval = rte_eth_dev_adjust_nb_rx_tx_desc(port, &nb_rxd, &nb_txd);
 150         if (retval != 0)
 151                 return retval;
 152
 153         for (q = 0; q < rxRings; q++) {
 154                 retval = rte_eth_rx_queue_setup(port, q, nb_rxd,
 155                                                 rte_eth_dev_socket_id(port),
 156                                                 NULL, mbuf_pool);
 157                 if (retval < 0)
 158                         return retval;
 159         }
 160
 161         txconf = dev_info.default_txconf;
 162         txconf.offloads = port_conf.txmode.offloads;
 163         for (q = 0; q < txRings; q++) {
 164                 retval = rte_eth_tx_queue_setup(port, q, nb_txd,
 165                                                 rte_eth_dev_socket_id(port),
 166                                                 &txconf);
 167                 if (retval < 0)
 168                         return retval;
 169         }
 170
 171         retval = rte_eth_dev_start(port);
 172         if (retval < 0)
 173                 return retval;
 174
 175         struct rte_eth_link link;
 176         do {
 177                 retval = rte_eth_link_get_nowait(port, &link);
 178                 if (retval < 0) {
 179                         printf("Failed link get (port %u): %s\n",
 180                                 port, rte_strerror(-retval));
 181                         return retval;
 182                 } else if (link.link_status)
 183                         break;
 184
 185                 printf("Waiting for Link up on port %"PRIu16"\n", port);
 186                 sleep(1);
 187         } while (!link.link_status);
 188
 189         if (!link.link_status) {
 190                 printf("Link down on port %"PRIu16"\n", port);
 191                 return 0;
 192         }
 193
 194         struct rte_ether_addr addr;
 195         rte_eth_macaddr_get(port, &addr);
 196         printf("Port %u MAC: %02"PRIx8" %02"PRIx8" %02"PRIx8
 197                         " %02"PRIx8" %02"PRIx8" %02"PRIx8"\n",
 198                         port,
 199                         addr.addr_bytes[0], addr.addr_bytes[1],
 200                         addr.addr_bytes[2], addr.addr_bytes[3],
 201                         addr.addr_bytes[4], addr.addr_bytes[5]);
 202
 203         retval = rte_eth_promiscuous_enable(port);
 204         if (retval != 0)
 205                 return retval;
 206
 207         return 0;
 208 }
 209
 210 struct lcore_params {
 211         unsigned worker_id;
 212         struct rte_distributor *d;
 213         struct rte_ring *rx_dist_ring;
 214         struct rte_ring *dist_tx_ring;
 215         struct rte_mempool *mem_pool;
 216 };
 217
 218 static int
 219 lcore_rx(struct lcore_params *p)
 220 {
 221         const uint16_t nb_ports = rte_eth_dev_count_avail();
 222         const int socket_id = rte_socket_id();
 223         uint16_t port;
 224         struct rte_mbuf *bufs[BURST_SIZE*2];
 225
 226         RTE_ETH_FOREACH_DEV(port) {
 227                 /* skip ports that are not enabled */
 228                 if ((enabled_port_mask & (1 << port)) == 0)
 229                         continue;
 230
 231                 if (rte_eth_dev_socket_id(port) > 0 &&
 232                                 rte_eth_dev_socket_id(port) != socket_id)
 233                         printf("WARNING, port %u is on remote NUMA node to "
 234                                         "RX thread.\n\tPerformance will not "
 235                                         "be optimal.\n", port);
 236         }
 237
 238         printf("\nCore %u doing packet RX.\n", rte_lcore_id());
 239         port = 0;
 240         while (!quit_signal_rx) {
 241
 242                 /* skip ports that are not enabled */
 243                 if ((enabled_port_mask & (1 << port)) == 0) {
 244                         if (++port == nb_ports)
 245                                 port = 0;
 246                         continue;
 247                 }
 248                 const uint16_t nb_rx = rte_eth_rx_burst(port, 0, bufs,
 249                                 BURST_SIZE);
 250                 if (unlikely(nb_rx == 0)) {
 251                         if (++port == nb_ports)
 252                                 port = 0;
 253                         continue;
 254                 }
 255                 app_stats.rx.rx_pkts += nb_rx;
 256
 257 /*
 258  * You can run the distributor on the rx core with this code. Returned
 259  * packets are then send straight to the tx core.
 260  */
 261 #if 0
 262         rte_distributor_process(d, bufs, nb_rx);
 263         const uint16_t nb_ret = rte_distributor_returned_pktsd,
 264                         bufs, BURST_SIZE*2);
 265
 266                 app_stats.rx.returned_pkts += nb_ret;
 267                 if (unlikely(nb_ret == 0)) {
 268                         if (++port == nb_ports)
 269                                 port = 0;
 270                         continue;
 271                 }
 272
 273                 struct rte_ring *tx_ring = p->dist_tx_ring;
 274                 uint16_t sent = rte_ring_enqueue_burst(tx_ring,
 275                                 (void *)bufs, nb_ret, NULL);
 276 #else
 277                 uint16_t nb_ret = nb_rx;
 278                 /*
 279                  * Swap the following two lines if you want the rx traffic
 280                  * to go directly to tx, no distribution.
 281                  */
 282                 struct rte_ring *out_ring = p->rx_dist_ring;
 283                 /* struct rte_ring *out_ring = p->dist_tx_ring; */
 284
 285                 uint16_t sent = rte_ring_enqueue_burst(out_ring,
 286                                 (void *)bufs, nb_ret, NULL);
 287 #endif
 288
 289                 app_stats.rx.enqueued_pkts += sent;
 290                 if (unlikely(sent < nb_ret)) {
 291                         app_stats.rx.enqdrop_pkts +=  nb_ret - sent;
 292                         RTE_LOG_DP(DEBUG, DISTRAPP,
 293                                 "%s:Packet loss due to full ring\n", __func__);
 294                         while (sent < nb_ret)
 295                                 rte_pktmbuf_free(bufs[sent++]);
 296                 }
 297                 if (++port == nb_ports)
 298                         port = 0;
 299         }
 300         if (power_lib_initialised)
 301                 rte_power_exit(rte_lcore_id());
 302         /* set worker & tx threads quit flag */
 303         printf("\nCore %u exiting rx task.\n", rte_lcore_id());
 304         quit_signal = 1;
 305         return 0;
 306 }
 307
 308 static inline void
 309 flush_one_port(struct output_buffer *outbuf, uint8_t outp)
 310 {
 311         unsigned int nb_tx = rte_eth_tx_burst(outp, 0,
 312                         outbuf->mbufs, outbuf->count);
 313         app_stats.tx.tx_pkts += outbuf->count;
 314
 315         if (unlikely(nb_tx < outbuf->count)) {
 316                 app_stats.tx.enqdrop_pkts +=  outbuf->count - nb_tx;
 317                 do {
 318                         rte_pktmbuf_free(outbuf->mbufs[nb_tx]);
 319                 } while (++nb_tx < outbuf->count);
 320         }
 321         outbuf->count = 0;
 322 }
 323
 324 static inline void
 325 flush_all_ports(struct output_buffer *tx_buffers)
 326 {
 327         uint16_t outp;
 328
 329         RTE_ETH_FOREACH_DEV(outp) {
 330                 /* skip ports that are not enabled */
 331                 if ((enabled_port_mask & (1 << outp)) == 0)
 332                         continue;
 333
 334                 if (tx_buffers[outp].count == 0)
 335                         continue;
 336
 337                 flush_one_port(&tx_buffers[outp], outp);
 338         }
 339 }
 340
 341
 342
 343 static int
 344 lcore_distributor(struct lcore_params *p)
 345 {
 346         struct rte_ring *in_r = p->rx_dist_ring;
 347         struct rte_ring *out_r = p->dist_tx_ring;
 348         struct rte_mbuf *bufs[BURST_SIZE * 4];
 349         struct rte_distributor *d = p->d;
 350
 351         printf("\nCore %u acting as distributor core.\n", rte_lcore_id());
 352         while (!quit_signal_dist) {
 353                 const uint16_t nb_rx = rte_ring_dequeue_burst(in_r,
 354                                 (void *)bufs, BURST_SIZE*1, NULL);
 355                 if (nb_rx) {
 356                         app_stats.dist.in_pkts += nb_rx;
 357
 358                         /* Distribute the packets */
 359                         rte_distributor_process(d, bufs, nb_rx);
 360                         /* Handle Returns */
 361                         const uint16_t nb_ret =
 362                                 rte_distributor_returned_pkts(d,
 363                                         bufs, BURST_SIZE*2);
 364
 365                         if (unlikely(nb_ret == 0))
 366                                 continue;
 367                         app_stats.dist.ret_pkts += nb_ret;
 368
 369                         uint16_t sent = rte_ring_enqueue_burst(out_r,
 370                                         (void *)bufs, nb_ret, NULL);
 371                         app_stats.dist.sent_pkts += sent;
 372                         if (unlikely(sent < nb_ret)) {
 373                                 app_stats.dist.enqdrop_pkts += nb_ret - sent;
 374                                 RTE_LOG(DEBUG, DISTRAPP,
 375                                         "%s:Packet loss due to full out ring\n",
 376                                         __func__);
 377                                 while (sent < nb_ret)
 378                                         rte_pktmbuf_free(bufs[sent++]);
 379                         }
 380                 }
 381         }
 382         printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
 383         quit_signal_work = 1;
 384         if (power_lib_initialised)
 385                 rte_power_exit(rte_lcore_id());
 386         rte_distributor_flush(d);
 387         /* Unblock any returns so workers can exit */
 388         rte_distributor_clear_returns(d);
 389         quit_signal_rx = 1;
 390         return 0;
 391 }
 392
 393
 394 static int
 395 lcore_tx(struct rte_ring *in_r)
 396 {
 397         static struct output_buffer tx_buffers[RTE_MAX_ETHPORTS];
 398         const int socket_id = rte_socket_id();
 399         uint16_t port;
 400
 401         RTE_ETH_FOREACH_DEV(port) {
 402                 /* skip ports that are not enabled */
 403                 if ((enabled_port_mask & (1 << port)) == 0)
 404                         continue;
 405
 406                 if (rte_eth_dev_socket_id(port) > 0 &&
 407                                 rte_eth_dev_socket_id(port) != socket_id)
 408                         printf("WARNING, port %u is on remote NUMA node to "
 409                                         "TX thread.\n\tPerformance will not "
 410                                         "be optimal.\n", port);
 411         }
 412
 413         printf("\nCore %u doing packet TX.\n", rte_lcore_id());
 414         while (!quit_signal) {
 415
 416                 RTE_ETH_FOREACH_DEV(port) {
 417                         /* skip ports that are not enabled */
 418                         if ((enabled_port_mask & (1 << port)) == 0)
 419                                 continue;
 420
 421                         struct rte_mbuf *bufs[BURST_SIZE_TX];
 422                         const uint16_t nb_rx = rte_ring_dequeue_burst(in_r,
 423                                         (void *)bufs, BURST_SIZE_TX, NULL);
 424                         app_stats.tx.dequeue_pkts += nb_rx;
 425
 426                         /* if we get no traffic, flush anything we have */
 427                         if (unlikely(nb_rx == 0)) {
 428                                 flush_all_ports(tx_buffers);
 429                                 continue;
 430                         }
 431
 432                         /* for traffic we receive, queue it up for transmit */
 433                         uint16_t i;
 434                         rte_prefetch_non_temporal((void *)bufs[0]);
 435                         rte_prefetch_non_temporal((void *)bufs[1]);
 436                         rte_prefetch_non_temporal((void *)bufs[2]);
 437                         for (i = 0; i < nb_rx; i++) {
 438                                 struct output_buffer *outbuf;
 439                                 uint8_t outp;
 440                                 rte_prefetch_non_temporal((void *)bufs[i + 3]);
 441                                 /*
 442                                  * workers should update in_port to hold the
 443                                  * output port value
 444                                  */
 445                                 outp = bufs[i]->port;
 446                                 /* skip ports that are not enabled */
 447                                 if ((enabled_port_mask & (1 << outp)) == 0)
 448                                         continue;
 449
 450                                 outbuf = &tx_buffers[outp];
 451                                 outbuf->mbufs[outbuf->count++] = bufs[i];
 452                                 if (outbuf->count == BURST_SIZE_TX)
 453                                         flush_one_port(outbuf, outp);
 454                         }
 455                 }
 456         }
 457         if (power_lib_initialised)
 458                 rte_power_exit(rte_lcore_id());
 459         printf("\nCore %u exiting tx task.\n", rte_lcore_id());
 460         return 0;
 461 }
 462
 463 static void
 464 int_handler(int sig_num)
 465 {
 466         printf("Exiting on signal %d\n", sig_num);
 467         /* set quit flag for rx thread to exit */
 468         quit_signal_dist = 1;
 469 }
 470
 471 static void
 472 print_stats(void)
 473 {
 474         struct rte_eth_stats eth_stats;
 475         unsigned int i, j;
 476         const unsigned int num_workers = rte_lcore_count() - 4;
 477
 478         RTE_ETH_FOREACH_DEV(i) {
 479                 rte_eth_stats_get(i, &eth_stats);
 480                 app_stats.port_rx_pkts[i] = eth_stats.ipackets;
 481                 app_stats.port_tx_pkts[i] = eth_stats.opackets;
 482         }
 483
 484         printf("\n\nRX Thread:\n");
 485         RTE_ETH_FOREACH_DEV(i) {
 486                 printf("Port %u Pktsin : %5.2f\n", i,
 487                                 (app_stats.port_rx_pkts[i] -
 488                                 prev_app_stats.port_rx_pkts[i])/1000000.0);
 489                 prev_app_stats.port_rx_pkts[i] = app_stats.port_rx_pkts[i];
 490         }
 491         printf(" - Received:    %5.2f\n",
 492                         (app_stats.rx.rx_pkts -
 493                         prev_app_stats.rx.rx_pkts)/1000000.0);
 494         printf(" - Returned:    %5.2f\n",
 495                         (app_stats.rx.returned_pkts -
 496                         prev_app_stats.rx.returned_pkts)/1000000.0);
 497         printf(" - Enqueued:    %5.2f\n",
 498                         (app_stats.rx.enqueued_pkts -
 499                         prev_app_stats.rx.enqueued_pkts)/1000000.0);
 500         printf(" - Dropped:     %s%5.2f%s\n", ANSI_COLOR_RED,
 501                         (app_stats.rx.enqdrop_pkts -
 502                         prev_app_stats.rx.enqdrop_pkts)/1000000.0,
 503                         ANSI_COLOR_RESET);
 504
 505         printf("Distributor thread:\n");
 506         printf(" - In:          %5.2f\n",
 507                         (app_stats.dist.in_pkts -
 508                         prev_app_stats.dist.in_pkts)/1000000.0);
 509         printf(" - Returned:    %5.2f\n",
 510                         (app_stats.dist.ret_pkts -
 511                         prev_app_stats.dist.ret_pkts)/1000000.0);
 512         printf(" - Sent:        %5.2f\n",
 513                         (app_stats.dist.sent_pkts -
 514                         prev_app_stats.dist.sent_pkts)/1000000.0);
 515         printf(" - Dropped      %s%5.2f%s\n", ANSI_COLOR_RED,
 516                         (app_stats.dist.enqdrop_pkts -
 517                         prev_app_stats.dist.enqdrop_pkts)/1000000.0,
 518                         ANSI_COLOR_RESET);
 519
 520         printf("TX thread:\n");
 521         printf(" - Dequeued:    %5.2f\n",
 522                         (app_stats.tx.dequeue_pkts -
 523                         prev_app_stats.tx.dequeue_pkts)/1000000.0);
 524         RTE_ETH_FOREACH_DEV(i) {
 525                 printf("Port %u Pktsout: %5.2f\n",
 526                                 i, (app_stats.port_tx_pkts[i] -
 527                                 prev_app_stats.port_tx_pkts[i])/1000000.0);
 528                 prev_app_stats.port_tx_pkts[i] = app_stats.port_tx_pkts[i];
 529         }
 530         printf(" - Transmitted: %5.2f\n",
 531                         (app_stats.tx.tx_pkts -
 532                         prev_app_stats.tx.tx_pkts)/1000000.0);
 533         printf(" - Dropped:     %s%5.2f%s\n", ANSI_COLOR_RED,
 534                         (app_stats.tx.enqdrop_pkts -
 535                         prev_app_stats.tx.enqdrop_pkts)/1000000.0,
 536                         ANSI_COLOR_RESET);
 537
 538         prev_app_stats.rx.rx_pkts = app_stats.rx.rx_pkts;
 539         prev_app_stats.rx.returned_pkts = app_stats.rx.returned_pkts;
 540         prev_app_stats.rx.enqueued_pkts = app_stats.rx.enqueued_pkts;
 541         prev_app_stats.rx.enqdrop_pkts = app_stats.rx.enqdrop_pkts;
 542         prev_app_stats.dist.in_pkts = app_stats.dist.in_pkts;
 543         prev_app_stats.dist.ret_pkts = app_stats.dist.ret_pkts;
 544         prev_app_stats.dist.sent_pkts = app_stats.dist.sent_pkts;
 545         prev_app_stats.dist.enqdrop_pkts = app_stats.dist.enqdrop_pkts;
 546         prev_app_stats.tx.dequeue_pkts = app_stats.tx.dequeue_pkts;
 547         prev_app_stats.tx.tx_pkts = app_stats.tx.tx_pkts;
 548         prev_app_stats.tx.enqdrop_pkts = app_stats.tx.enqdrop_pkts;
 549
 550         for (i = 0; i < num_workers; i++) {
 551                 printf("Worker %02u Pkts: %5.2f. Bursts(1-8): ", i,
 552                                 (app_stats.worker_pkts[i] -
 553                                 prev_app_stats.worker_pkts[i])/1000000.0);
 554                 for (j = 0; j < 8; j++) {
 555                         printf("%"PRIu64" ", app_stats.worker_bursts[i][j]);
 556                         app_stats.worker_bursts[i][j] = 0;
 557                 }
 558                 printf("\n");
 559                 prev_app_stats.worker_pkts[i] = app_stats.worker_pkts[i];
 560         }
 561 }
 562
 563 static int
 564 lcore_worker(struct lcore_params *p)
 565 {
 566         struct rte_distributor *d = p->d;
 567         const unsigned id = p->worker_id;
 568         unsigned int num = 0;
 569         unsigned int i;
 570
 571         /*
 572          * for single port, xor_val will be zero so we won't modify the output
 573          * port, otherwise we send traffic from 0 to 1, 2 to 3, and vice versa
 574          */
 575         const unsigned xor_val = (rte_eth_dev_count_avail() > 1);
 576         struct rte_mbuf *buf[8] __rte_cache_aligned;
 577
 578         for (i = 0; i < 8; i++)
 579                 buf[i] = NULL;
 580
 581         app_stats.worker_pkts[p->worker_id] = 1;
 582
 583         printf("\nCore %u acting as worker core.\n", rte_lcore_id());
 584         while (!quit_signal_work) {
 585                 num = rte_distributor_get_pkt(d, id, buf, buf, num);
 586                 /* Do a little bit of work for each packet */
 587                 for (i = 0; i < num; i++) {
 588                         uint64_t t = rte_rdtsc()+100;
 589
 590                         while (rte_rdtsc() < t)
 591                                 rte_pause();
 592                         buf[i]->port ^= xor_val;
 593                 }
 594
 595                 app_stats.worker_pkts[p->worker_id] += num;
 596                 if (num > 0)
 597                         app_stats.worker_bursts[p->worker_id][num-1]++;
 598         }
 599         if (power_lib_initialised)
 600                 rte_power_exit(rte_lcore_id());
 601         rte_free(p);
 602         return 0;
 603 }
 604
 605 static int
 606 init_power_library(void)
 607 {
 608         int ret = 0, lcore_id;
 609         RTE_LCORE_FOREACH_SLAVE(lcore_id) {
 610                 /* init power management library */
 611                 ret = rte_power_init(lcore_id);
 612                 if (ret) {
 613                         RTE_LOG(ERR, POWER,
 614                                 "Library initialization failed on core %u\n",
 615                                 lcore_id);
 616                         /*
 617                          * Return on first failure, we'll fall back
 618                          * to non-power operation
 619                          */
 620                         return ret;
 621                 }
 622         }
 623         return ret;
 624 }
 625
 626 /* display usage */
 627 static void
 628 print_usage(const char *prgname)
 629 {
 630         printf("%s [EAL options] -- -p PORTMASK\n"
 631                         "  -p PORTMASK: hexadecimal bitmask of ports to configure\n",
 632                         prgname);
 633 }
 634
 635 static int
 636 parse_portmask(const char *portmask)
 637 {
 638         char *end = NULL;
 639         unsigned long pm;
 640
 641         /* parse hexadecimal string */
 642         pm = strtoul(portmask, &end, 16);
 643         if ((portmask[0] == '\0') || (end == NULL) || (*end != '\0'))
 644                 return -1;
 645
 646         if (pm == 0)
 647                 return -1;
 648
 649         return pm;
 650 }
 651
 652 /* Parse the argument given in the command line of the application */
 653 static int
 654 parse_args(int argc, char **argv)
 655 {
 656         int opt;
 657         char **argvopt;
 658         int option_index;
 659         char *prgname = argv[0];
 660         static struct option lgopts[] = {
 661                 {NULL, 0, 0, 0}
 662         };
 663
 664         argvopt = argv;
 665
 666         while ((opt = getopt_long(argc, argvopt, "p:",
 667                         lgopts, &option_index)) != EOF) {
 668
 669                 switch (opt) {
 670                 /* portmask */
 671                 case 'p':
 672                         enabled_port_mask = parse_portmask(optarg);
 673                         if (enabled_port_mask == 0) {
 674                                 printf("invalid portmask\n");
 675                                 print_usage(prgname);
 676                                 return -1;
 677                         }
 678                         break;
 679
 680                 default:
 681                         print_usage(prgname);
 682                         return -1;
 683                 }
 684         }
 685
 686         if (optind <= 1) {
 687                 print_usage(prgname);
 688                 return -1;
 689         }
 690
 691         argv[optind-1] = prgname;
 692
 693         optind = 1; /* reset getopt lib */
 694         return 0;
 695 }
 696
 697 /* Main function, does initialization and calls the per-lcore functions */
 698 int
 699 main(int argc, char *argv[])
 700 {
 701         struct rte_mempool *mbuf_pool;
 702         struct rte_distributor *d;
 703         struct rte_ring *dist_tx_ring;
 704         struct rte_ring *rx_dist_ring;
 705         struct rte_power_core_capabilities lcore_cap;
 706         unsigned int lcore_id, worker_id = 0;
 707         int distr_core_id = -1, rx_core_id = -1, tx_core_id = -1;
 708         unsigned nb_ports;
 709         uint16_t portid;
 710         uint16_t nb_ports_available;
 711         uint64_t t, freq;
 712
 713         /* catch ctrl-c so we can print on exit */
 714         signal(SIGINT, int_handler);
 715
 716         /* init EAL */
 717         int ret = rte_eal_init(argc, argv);
 718         if (ret < 0)
 719                 rte_exit(EXIT_FAILURE, "Error with EAL initialization\n");
 720         argc -= ret;
 721         argv += ret;
 722
 723         /* parse application arguments (after the EAL ones) */
 724         ret = parse_args(argc, argv);
 725         if (ret < 0)
 726                 rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 727
 728         if (rte_lcore_count() < 5)
 729                 rte_exit(EXIT_FAILURE, "Error, This application needs at "
 730                                 "least 5 logical cores to run:\n"
 731                                 "1 lcore for stats (can be core 0)\n"
 732                                 "1 lcore for packet RX\n"
 733                                 "1 lcore for distribution\n"
 734                                 "1 lcore for packet TX\n"
 735                                 "and at least 1 lcore for worker threads\n");
 736
 737         if (init_power_library() == 0)
 738                 power_lib_initialised = 1;
 739
 740         nb_ports = rte_eth_dev_count_avail();
 741         if (nb_ports == 0)
 742                 rte_exit(EXIT_FAILURE, "Error: no ethernet ports detected\n");
 743         if (nb_ports != 1 && (nb_ports & 1))
 744                 rte_exit(EXIT_FAILURE, "Error: number of ports must be even, except "
 745                                 "when using a single port\n");
 746
 747         mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL",
 748                 NUM_MBUFS * nb_ports, MBUF_CACHE_SIZE, 0,
 749                 RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id());
 750         if (mbuf_pool == NULL)
 751                 rte_exit(EXIT_FAILURE, "Cannot create mbuf pool\n");
 752         nb_ports_available = nb_ports;
 753
 754         /* initialize all ports */
 755         RTE_ETH_FOREACH_DEV(portid) {
 756                 /* skip ports that are not enabled */
 757                 if ((enabled_port_mask & (1 << portid)) == 0) {
 758                         printf("\nSkipping disabled port %d\n", portid);
 759                         nb_ports_available--;
 760                         continue;
 761                 }
 762                 /* init port */
 763                 printf("Initializing port %u... done\n", portid);
 764
 765                 if (port_init(portid, mbuf_pool) != 0)
 766                         rte_exit(EXIT_FAILURE, "Cannot initialize port %u\n",
 767                                         portid);
 768         }
 769
 770         if (!nb_ports_available) {
 771                 rte_exit(EXIT_FAILURE,
 772                                 "All available ports are disabled. Please set portmask.\n");
 773         }
 774
 775         d = rte_distributor_create("PKT_DIST", rte_socket_id(),
 776                         rte_lcore_count() - 4,
 777                         RTE_DIST_ALG_BURST);
 778         if (d == NULL)
 779                 rte_exit(EXIT_FAILURE, "Cannot create distributor\n");
 780
 781         /*
 782          * scheduler ring is read by the transmitter core, and written to
 783          * by scheduler core
 784          */
 785         dist_tx_ring = rte_ring_create("Output_ring", SCHED_TX_RING_SZ,
 786                         rte_socket_id(), RING_F_SC_DEQ | RING_F_SP_ENQ);
 787         if (dist_tx_ring == NULL)
 788                 rte_exit(EXIT_FAILURE, "Cannot create output ring\n");
 789
 790         rx_dist_ring = rte_ring_create("Input_ring", SCHED_RX_RING_SZ,
 791                         rte_socket_id(), RING_F_SC_DEQ | RING_F_SP_ENQ);
 792         if (rx_dist_ring == NULL)
 793                 rte_exit(EXIT_FAILURE, "Cannot create output ring\n");
 794
 795         if (power_lib_initialised) {
 796                 /*
 797                  * Here we'll pre-assign lcore ids to the rx, tx and
 798                  * distributor workloads if there's higher frequency
 799                  * on those cores e.g. if Turbo Boost is enabled.
 800                  * It's also worth mentioning that it will assign cores in a
 801                  * specific order, so that if there's less than three
 802                  * available, the higher frequency cores will go to the
 803                  * distributor first, then rx, then tx.
 804                  */
 805                 RTE_LCORE_FOREACH_SLAVE(lcore_id) {
 806
 807                         rte_power_get_capabilities(lcore_id, &lcore_cap);
 808
 809                         if (lcore_cap.priority != 1)
 810                                 continue;
 811
 812                         if (distr_core_id < 0) {
 813                                 distr_core_id = lcore_id;
 814                                 printf("Distributor on priority core %d\n",
 815                                         lcore_id);
 816                                 continue;
 817                         }
 818                         if (rx_core_id < 0) {
 819                                 rx_core_id = lcore_id;
 820                                 printf("Rx on priority core %d\n",
 821                                         lcore_id);
 822                                 continue;
 823                         }
 824                         if (tx_core_id < 0) {
 825                                 tx_core_id = lcore_id;
 826                                 printf("Tx on priority core %d\n",
 827                                         lcore_id);
 828                                 continue;
 829                         }
 830                 }
 831         }
 832
 833         /*
 834          * If there's any of the key workloads left without an lcore_id
 835          * after the high performing core assignment above, pre-assign
 836          * them here.
 837          */
 838         RTE_LCORE_FOREACH_SLAVE(lcore_id) {
 839                 if (lcore_id == (unsigned int)distr_core_id ||
 840                                 lcore_id == (unsigned int)rx_core_id ||
 841                                 lcore_id == (unsigned int)tx_core_id)
 842                         continue;
 843                 if (distr_core_id < 0) {
 844                         distr_core_id = lcore_id;
 845                         printf("Distributor on core %d\n", lcore_id);
 846                         continue;
 847                 }
 848                 if (rx_core_id < 0) {
 849                         rx_core_id = lcore_id;
 850                         printf("Rx on core %d\n", lcore_id);
 851                         continue;
 852                 }
 853                 if (tx_core_id < 0) {
 854                         tx_core_id = lcore_id;
 855                         printf("Tx on core %d\n", lcore_id);
 856                         continue;
 857                 }
 858         }
 859
 860         printf(" tx id %d, dist id %d, rx id %d\n",
 861                         tx_core_id,
 862                         distr_core_id,
 863                         rx_core_id);
 864
 865         /*
 866          * Kick off all the worker threads first, avoiding the pre-assigned
 867          * lcore_ids for tx, rx and distributor workloads.
 868          */
 869         RTE_LCORE_FOREACH_SLAVE(lcore_id) {
 870                 if (lcore_id == (unsigned int)distr_core_id ||
 871                                 lcore_id == (unsigned int)rx_core_id ||
 872                                 lcore_id == (unsigned int)tx_core_id)
 873                         continue;
 874                 printf("Starting thread %d as worker, lcore_id %d\n",
 875                                 worker_id, lcore_id);
 876                 struct lcore_params *p =
 877                         rte_malloc(NULL, sizeof(*p), 0);
 878                 if (!p)
 879                         rte_panic("malloc failure\n");
 880                 *p = (struct lcore_params){worker_id++, d, rx_dist_ring,
 881                         dist_tx_ring, mbuf_pool};
 882
 883                 rte_eal_remote_launch((lcore_function_t *)lcore_worker,
 884                                 p, lcore_id);
 885         }
 886
 887         /* Start tx core */
 888         rte_eal_remote_launch((lcore_function_t *)lcore_tx,
 889                         dist_tx_ring, tx_core_id);
 890
 891         /* Start distributor core */
 892         struct lcore_params *pd =
 893                 rte_malloc(NULL, sizeof(*pd), 0);
 894         if (!pd)
 895                 rte_panic("malloc failure\n");
 896         *pd = (struct lcore_params){worker_id++, d,
 897                 rx_dist_ring, dist_tx_ring, mbuf_pool};
 898         rte_eal_remote_launch(
 899                         (lcore_function_t *)lcore_distributor,
 900                         pd, distr_core_id);
 901
 902         /* Start rx core */
 903         struct lcore_params *pr =
 904                 rte_malloc(NULL, sizeof(*pr), 0);
 905         if (!pr)
 906                 rte_panic("malloc failure\n");
 907         *pr = (struct lcore_params){worker_id++, d, rx_dist_ring,
 908                 dist_tx_ring, mbuf_pool};
 909         rte_eal_remote_launch((lcore_function_t *)lcore_rx,
 910                         pr, rx_core_id);
 911
 912         freq = rte_get_timer_hz();
 913         t = rte_rdtsc() + freq;
 914         while (!quit_signal_dist) {
 915                 if (t < rte_rdtsc()) {
 916                         print_stats();
 917                         t = rte_rdtsc() + freq;
 918                 }
 919                 usleep(1000);
 920         }
 921
 922         RTE_LCORE_FOREACH_SLAVE(lcore_id) {
 923                 if (rte_eal_wait_lcore(lcore_id) < 0)
 924                         return -1;
 925         }
 926
 927         print_stats();
 928
 929         rte_free(pd);
 930         rte_free(pr);
 931
 932         return 0;
 933 }