examples/distributor/main.c

   1 /*-
   2  *   BSD LICENSE
   3  *
   4  *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
   5  *   All rights reserved.
   6  *
   7  *   Redistribution and use in source and binary forms, with or without
   8  *   modification, are permitted provided that the following conditions
   9  *   are met:
  10  *
  11  *     * Redistributions of source code must retain the above copyright
  12  *       notice, this list of conditions and the following disclaimer.
  13  *     * Redistributions in binary form must reproduce the above copyright
  14  *       notice, this list of conditions and the following disclaimer in
  15  *       the documentation and/or other materials provided with the
  16  *       distribution.
  17  *     * Neither the name of Intel Corporation nor the names of its
  18  *       contributors may be used to endorse or promote products derived
  19  *       from this software without specific prior written permission.
  20  *
  21  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  24  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  25  *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  26  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  27  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  28  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  29  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  30  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  31  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  32  */
  33
  34 #include <stdint.h>
  35 #include <inttypes.h>
  36 #include <unistd.h>
  37 #include <signal.h>
  38 #include <getopt.h>
  39
  40 #include <rte_eal.h>
  41 #include <rte_ethdev.h>
  42 #include <rte_cycles.h>
  43 #include <rte_malloc.h>
  44 #include <rte_debug.h>
  45 #include <rte_prefetch.h>
  46 #include <rte_distributor.h>
  47
  48 #define RX_RING_SIZE 256
  49 #define TX_RING_SIZE 512
  50 #define NUM_MBUFS ((64*1024)-1)
  51 #define MBUF_CACHE_SIZE 250
  52 #define BURST_SIZE 32
  53 #define RTE_RING_SZ 1024
  54
  55 #define RTE_LOGTYPE_DISTRAPP RTE_LOGTYPE_USER1
  56
  57 /* mask of enabled ports */
  58 static uint32_t enabled_port_mask;
  59 volatile uint8_t quit_signal;
  60 volatile uint8_t quit_signal_rx;
  61
  62 static volatile struct app_stats {
  63         struct {
  64                 uint64_t rx_pkts;
  65                 uint64_t returned_pkts;
  66                 uint64_t enqueued_pkts;
  67         } rx __rte_cache_aligned;
  68
  69         struct {
  70                 uint64_t dequeue_pkts;
  71                 uint64_t tx_pkts;
  72         } tx __rte_cache_aligned;
  73 } app_stats;
  74
  75 static const struct rte_eth_conf port_conf_default = {
  76         .rxmode = {
  77                 .mq_mode = ETH_MQ_RX_RSS,
  78                 .max_rx_pkt_len = ETHER_MAX_LEN,
  79         },
  80         .txmode = {
  81                 .mq_mode = ETH_MQ_TX_NONE,
  82         },
  83         .rx_adv_conf = {
  84                 .rss_conf = {
  85                         .rss_hf = ETH_RSS_IP | ETH_RSS_UDP |
  86                                 ETH_RSS_TCP | ETH_RSS_SCTP,
  87                 }
  88         },
  89 };
  90
  91 struct output_buffer {
  92         unsigned count;
  93         struct rte_mbuf *mbufs[BURST_SIZE];
  94 };
  95
  96 /*
  97  * Initialises a given port using global settings and with the rx buffers
  98  * coming from the mbuf_pool passed as parameter
  99  */
 100 static inline int
 101 port_init(uint8_t port, struct rte_mempool *mbuf_pool)
 102 {
 103         struct rte_eth_conf port_conf = port_conf_default;
 104         const uint16_t rxRings = 1, txRings = rte_lcore_count() - 1;
 105         int retval;
 106         uint16_t q;
 107
 108         if (port >= rte_eth_dev_count())
 109                 return -1;
 110
 111         retval = rte_eth_dev_configure(port, rxRings, txRings, &port_conf);
 112         if (retval != 0)
 113                 return retval;
 114
 115         for (q = 0; q < rxRings; q++) {
 116                 retval = rte_eth_rx_queue_setup(port, q, RX_RING_SIZE,
 117                                                 rte_eth_dev_socket_id(port),
 118                                                 NULL, mbuf_pool);
 119                 if (retval < 0)
 120                         return retval;
 121         }
 122
 123         for (q = 0; q < txRings; q++) {
 124                 retval = rte_eth_tx_queue_setup(port, q, TX_RING_SIZE,
 125                                                 rte_eth_dev_socket_id(port),
 126                                                 NULL);
 127                 if (retval < 0)
 128                         return retval;
 129         }
 130
 131         retval = rte_eth_dev_start(port);
 132         if (retval < 0)
 133                 return retval;
 134
 135         struct rte_eth_link link;
 136         rte_eth_link_get_nowait(port, &link);
 137         if (!link.link_status) {
 138                 sleep(1);
 139                 rte_eth_link_get_nowait(port, &link);
 140         }
 141
 142         if (!link.link_status) {
 143                 printf("Link down on port %"PRIu8"\n", port);
 144                 return 0;
 145         }
 146
 147         struct ether_addr addr;
 148         rte_eth_macaddr_get(port, &addr);
 149         printf("Port %u MAC: %02"PRIx8" %02"PRIx8" %02"PRIx8
 150                         " %02"PRIx8" %02"PRIx8" %02"PRIx8"\n",
 151                         (unsigned)port,
 152                         addr.addr_bytes[0], addr.addr_bytes[1],
 153                         addr.addr_bytes[2], addr.addr_bytes[3],
 154                         addr.addr_bytes[4], addr.addr_bytes[5]);
 155
 156         rte_eth_promiscuous_enable(port);
 157
 158         return 0;
 159 }
 160
 161 struct lcore_params {
 162         unsigned worker_id;
 163         struct rte_distributor *d;
 164         struct rte_ring *r;
 165         struct rte_mempool *mem_pool;
 166 };
 167
 168 static int
 169 quit_workers(struct rte_distributor *d, struct rte_mempool *p)
 170 {
 171         const unsigned num_workers = rte_lcore_count() - 2;
 172         unsigned i;
 173         struct rte_mbuf *bufs[num_workers];
 174
 175         if (rte_mempool_get_bulk(p, (void *)bufs, num_workers) != 0) {
 176                 printf("line %d: Error getting mbufs from pool\n", __LINE__);
 177                 return -1;
 178         }
 179
 180         for (i = 0; i < num_workers; i++)
 181                 bufs[i]->hash.rss = i << 1;
 182
 183         rte_distributor_process(d, bufs, num_workers);
 184         rte_mempool_put_bulk(p, (void *)bufs, num_workers);
 185
 186         return 0;
 187 }
 188
 189 static int
 190 lcore_rx(struct lcore_params *p)
 191 {
 192         struct rte_distributor *d = p->d;
 193         struct rte_mempool *mem_pool = p->mem_pool;
 194         struct rte_ring *r = p->r;
 195         const uint8_t nb_ports = rte_eth_dev_count();
 196         const int socket_id = rte_socket_id();
 197         uint8_t port;
 198
 199         for (port = 0; port < nb_ports; port++) {
 200                 /* skip ports that are not enabled */
 201                 if ((enabled_port_mask & (1 << port)) == 0)
 202                         continue;
 203
 204                 if (rte_eth_dev_socket_id(port) > 0 &&
 205                                 rte_eth_dev_socket_id(port) != socket_id)
 206                         printf("WARNING, port %u is on remote NUMA node to "
 207                                         "RX thread.\n\tPerformance will not "
 208                                         "be optimal.\n", port);
 209         }
 210
 211         printf("\nCore %u doing packet RX.\n", rte_lcore_id());
 212         port = 0;
 213         while (!quit_signal_rx) {
 214
 215                 /* skip ports that are not enabled */
 216                 if ((enabled_port_mask & (1 << port)) == 0) {
 217                         if (++port == nb_ports)
 218                                 port = 0;
 219                         continue;
 220                 }
 221                 struct rte_mbuf *bufs[BURST_SIZE*2];
 222                 const uint16_t nb_rx = rte_eth_rx_burst(port, 0, bufs,
 223                                 BURST_SIZE);
 224                 if (unlikely(nb_rx == 0)) {
 225                         if (++port == nb_ports)
 226                                 port = 0;
 227                         continue;
 228                 }
 229                 app_stats.rx.rx_pkts += nb_rx;
 230
 231                 rte_distributor_process(d, bufs, nb_rx);
 232                 const uint16_t nb_ret = rte_distributor_returned_pkts(d,
 233                                 bufs, BURST_SIZE*2);
 234                 app_stats.rx.returned_pkts += nb_ret;
 235                 if (unlikely(nb_ret == 0)) {
 236                         if (++port == nb_ports)
 237                                 port = 0;
 238                         continue;
 239                 }
 240
 241                 uint16_t sent = rte_ring_enqueue_burst(r, (void *)bufs, nb_ret);
 242                 app_stats.rx.enqueued_pkts += sent;
 243                 if (unlikely(sent < nb_ret)) {
 244                         RTE_LOG_DP(DEBUG, DISTRAPP,
 245                                 "%s:Packet loss due to full ring\n", __func__);
 246                         while (sent < nb_ret)
 247                                 rte_pktmbuf_free(bufs[sent++]);
 248                 }
 249                 if (++port == nb_ports)
 250                         port = 0;
 251         }
 252         rte_distributor_process(d, NULL, 0);
 253         /* flush distributor to bring to known state */
 254         rte_distributor_flush(d);
 255         /* set worker & tx threads quit flag */
 256         quit_signal = 1;
 257         /*
 258          * worker threads may hang in get packet as
 259          * distributor process is not running, just make sure workers
 260          * get packets till quit_signal is actually been
 261          * received and they gracefully shutdown
 262          */
 263         if (quit_workers(d, mem_pool) != 0)
 264                 return -1;
 265         /* rx thread should quit at last */
 266         return 0;
 267 }
 268
 269 static inline void
 270 flush_one_port(struct output_buffer *outbuf, uint8_t outp)
 271 {
 272         unsigned nb_tx = rte_eth_tx_burst(outp, 0, outbuf->mbufs,
 273                         outbuf->count);
 274         app_stats.tx.tx_pkts += nb_tx;
 275
 276         if (unlikely(nb_tx < outbuf->count)) {
 277                 RTE_LOG_DP(DEBUG, DISTRAPP,
 278                         "%s:Packet loss with tx_burst\n", __func__);
 279                 do {
 280                         rte_pktmbuf_free(outbuf->mbufs[nb_tx]);
 281                 } while (++nb_tx < outbuf->count);
 282         }
 283         outbuf->count = 0;
 284 }
 285
 286 static inline void
 287 flush_all_ports(struct output_buffer *tx_buffers, uint8_t nb_ports)
 288 {
 289         uint8_t outp;
 290         for (outp = 0; outp < nb_ports; outp++) {
 291                 /* skip ports that are not enabled */
 292                 if ((enabled_port_mask & (1 << outp)) == 0)
 293                         continue;
 294
 295                 if (tx_buffers[outp].count == 0)
 296                         continue;
 297
 298                 flush_one_port(&tx_buffers[outp], outp);
 299         }
 300 }
 301
 302 static int
 303 lcore_tx(struct rte_ring *in_r)
 304 {
 305         static struct output_buffer tx_buffers[RTE_MAX_ETHPORTS];
 306         const uint8_t nb_ports = rte_eth_dev_count();
 307         const int socket_id = rte_socket_id();
 308         uint8_t port;
 309
 310         for (port = 0; port < nb_ports; port++) {
 311                 /* skip ports that are not enabled */
 312                 if ((enabled_port_mask & (1 << port)) == 0)
 313                         continue;
 314
 315                 if (rte_eth_dev_socket_id(port) > 0 &&
 316                                 rte_eth_dev_socket_id(port) != socket_id)
 317                         printf("WARNING, port %u is on remote NUMA node to "
 318                                         "TX thread.\n\tPerformance will not "
 319                                         "be optimal.\n", port);
 320         }
 321
 322         printf("\nCore %u doing packet TX.\n", rte_lcore_id());
 323         while (!quit_signal) {
 324
 325                 for (port = 0; port < nb_ports; port++) {
 326                         /* skip ports that are not enabled */
 327                         if ((enabled_port_mask & (1 << port)) == 0)
 328                                 continue;
 329
 330                         struct rte_mbuf *bufs[BURST_SIZE];
 331                         const uint16_t nb_rx = rte_ring_dequeue_burst(in_r,
 332                                         (void *)bufs, BURST_SIZE);
 333                         app_stats.tx.dequeue_pkts += nb_rx;
 334
 335                         /* if we get no traffic, flush anything we have */
 336                         if (unlikely(nb_rx == 0)) {
 337                                 flush_all_ports(tx_buffers, nb_ports);
 338                                 continue;
 339                         }
 340
 341                         /* for traffic we receive, queue it up for transmit */
 342                         uint16_t i;
 343                         rte_prefetch_non_temporal((void *)bufs[0]);
 344                         rte_prefetch_non_temporal((void *)bufs[1]);
 345                         rte_prefetch_non_temporal((void *)bufs[2]);
 346                         for (i = 0; i < nb_rx; i++) {
 347                                 struct output_buffer *outbuf;
 348                                 uint8_t outp;
 349                                 rte_prefetch_non_temporal((void *)bufs[i + 3]);
 350                                 /*
 351                                  * workers should update in_port to hold the
 352                                  * output port value
 353                                  */
 354                                 outp = bufs[i]->port;
 355                                 /* skip ports that are not enabled */
 356                                 if ((enabled_port_mask & (1 << outp)) == 0)
 357                                         continue;
 358
 359                                 outbuf = &tx_buffers[outp];
 360                                 outbuf->mbufs[outbuf->count++] = bufs[i];
 361                                 if (outbuf->count == BURST_SIZE)
 362                                         flush_one_port(outbuf, outp);
 363                         }
 364                 }
 365         }
 366         return 0;
 367 }
 368
 369 static void
 370 int_handler(int sig_num)
 371 {
 372         printf("Exiting on signal %d\n", sig_num);
 373         /* set quit flag for rx thread to exit */
 374         quit_signal_rx = 1;
 375 }
 376
 377 static void
 378 print_stats(void)
 379 {
 380         struct rte_eth_stats eth_stats;
 381         unsigned i;
 382
 383         printf("\nRX thread stats:\n");
 384         printf(" - Received:    %"PRIu64"\n", app_stats.rx.rx_pkts);
 385         printf(" - Processed:   %"PRIu64"\n", app_stats.rx.returned_pkts);
 386         printf(" - Enqueued:    %"PRIu64"\n", app_stats.rx.enqueued_pkts);
 387
 388         printf("\nTX thread stats:\n");
 389         printf(" - Dequeued:    %"PRIu64"\n", app_stats.tx.dequeue_pkts);
 390         printf(" - Transmitted: %"PRIu64"\n", app_stats.tx.tx_pkts);
 391
 392         for (i = 0; i < rte_eth_dev_count(); i++) {
 393                 rte_eth_stats_get(i, &eth_stats);
 394                 printf("\nPort %u stats:\n", i);
 395                 printf(" - Pkts in:   %"PRIu64"\n", eth_stats.ipackets);
 396                 printf(" - Pkts out:  %"PRIu64"\n", eth_stats.opackets);
 397                 printf(" - In Errs:   %"PRIu64"\n", eth_stats.ierrors);
 398                 printf(" - Out Errs:  %"PRIu64"\n", eth_stats.oerrors);
 399                 printf(" - Mbuf Errs: %"PRIu64"\n", eth_stats.rx_nombuf);
 400         }
 401 }
 402
 403 static int
 404 lcore_worker(struct lcore_params *p)
 405 {
 406         struct rte_distributor *d = p->d;
 407         const unsigned id = p->worker_id;
 408         unsigned int num = 0;
 409         unsigned int i;
 410
 411         /*
 412          * for single port, xor_val will be zero so we won't modify the output
 413          * port, otherwise we send traffic from 0 to 1, 2 to 3, and vice versa
 414          */
 415         const unsigned xor_val = (rte_eth_dev_count() > 1);
 416         struct rte_mbuf *buf[8] __rte_cache_aligned;
 417
 418         for (i = 0; i < 8; i++)
 419                 buf[i] = NULL;
 420
 421         printf("\nCore %u acting as worker core.\n", rte_lcore_id());
 422         while (!quit_signal) {
 423                 num = rte_distributor_get_pkt(d, id, buf, buf, num);
 424                 /* Do a little bit of work for each packet */
 425                 for (i = 0; i < num; i++) {
 426                         uint64_t t = rte_rdtsc()+100;
 427
 428                         while (rte_rdtsc() < t)
 429                                 rte_pause();
 430                         buf[i]->port ^= xor_val;
 431                 }
 432         }
 433         return 0;
 434 }
 435
 436 /* display usage */
 437 static void
 438 print_usage(const char *prgname)
 439 {
 440         printf("%s [EAL options] -- -p PORTMASK\n"
 441                         "  -p PORTMASK: hexadecimal bitmask of ports to configure\n",
 442                         prgname);
 443 }
 444
 445 static int
 446 parse_portmask(const char *portmask)
 447 {
 448         char *end = NULL;
 449         unsigned long pm;
 450
 451         /* parse hexadecimal string */
 452         pm = strtoul(portmask, &end, 16);
 453         if ((portmask[0] == '\0') || (end == NULL) || (*end != '\0'))
 454                 return -1;
 455
 456         if (pm == 0)
 457                 return -1;
 458
 459         return pm;
 460 }
 461
 462 /* Parse the argument given in the command line of the application */
 463 static int
 464 parse_args(int argc, char **argv)
 465 {
 466         int opt;
 467         char **argvopt;
 468         int option_index;
 469         char *prgname = argv[0];
 470         static struct option lgopts[] = {
 471                 {NULL, 0, 0, 0}
 472         };
 473
 474         argvopt = argv;
 475
 476         while ((opt = getopt_long(argc, argvopt, "p:",
 477                         lgopts, &option_index)) != EOF) {
 478
 479                 switch (opt) {
 480                 /* portmask */
 481                 case 'p':
 482                         enabled_port_mask = parse_portmask(optarg);
 483                         if (enabled_port_mask == 0) {
 484                                 printf("invalid portmask\n");
 485                                 print_usage(prgname);
 486                                 return -1;
 487                         }
 488                         break;
 489
 490                 default:
 491                         print_usage(prgname);
 492                         return -1;
 493                 }
 494         }
 495
 496         if (optind <= 1) {
 497                 print_usage(prgname);
 498                 return -1;
 499         }
 500
 501         argv[optind-1] = prgname;
 502
 503         optind = 1; /* reset getopt lib */
 504         return 0;
 505 }
 506
 507 /* Main function, does initialization and calls the per-lcore functions */
 508 int
 509 main(int argc, char *argv[])
 510 {
 511         struct rte_mempool *mbuf_pool;
 512         struct rte_distributor *d;
 513         struct rte_ring *output_ring;
 514         unsigned lcore_id, worker_id = 0;
 515         unsigned nb_ports;
 516         uint8_t portid;
 517         uint8_t nb_ports_available;
 518
 519         /* catch ctrl-c so we can print on exit */
 520         signal(SIGINT, int_handler);
 521
 522         /* init EAL */
 523         int ret = rte_eal_init(argc, argv);
 524         if (ret < 0)
 525                 rte_exit(EXIT_FAILURE, "Error with EAL initialization\n");
 526         argc -= ret;
 527         argv += ret;
 528
 529         /* parse application arguments (after the EAL ones) */
 530         ret = parse_args(argc, argv);
 531         if (ret < 0)
 532                 rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 533
 534         if (rte_lcore_count() < 3)
 535                 rte_exit(EXIT_FAILURE, "Error, This application needs at "
 536                                 "least 3 logical cores to run:\n"
 537                                 "1 lcore for packet RX and distribution\n"
 538                                 "1 lcore for packet TX\n"
 539                                 "and at least 1 lcore for worker threads\n");
 540
 541         nb_ports = rte_eth_dev_count();
 542         if (nb_ports == 0)
 543                 rte_exit(EXIT_FAILURE, "Error: no ethernet ports detected\n");
 544         if (nb_ports != 1 && (nb_ports & 1))
 545                 rte_exit(EXIT_FAILURE, "Error: number of ports must be even, except "
 546                                 "when using a single port\n");
 547
 548         mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL",
 549                 NUM_MBUFS * nb_ports, MBUF_CACHE_SIZE, 0,
 550                 RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id());
 551         if (mbuf_pool == NULL)
 552                 rte_exit(EXIT_FAILURE, "Cannot create mbuf pool\n");
 553         nb_ports_available = nb_ports;
 554
 555         /* initialize all ports */
 556         for (portid = 0; portid < nb_ports; portid++) {
 557                 /* skip ports that are not enabled */
 558                 if ((enabled_port_mask & (1 << portid)) == 0) {
 559                         printf("\nSkipping disabled port %d\n", portid);
 560                         nb_ports_available--;
 561                         continue;
 562                 }
 563                 /* init port */
 564                 printf("Initializing port %u... done\n", (unsigned) portid);
 565
 566                 if (port_init(portid, mbuf_pool) != 0)
 567                         rte_exit(EXIT_FAILURE, "Cannot initialize port %"PRIu8"\n",
 568                                         portid);
 569         }
 570
 571         if (!nb_ports_available) {
 572                 rte_exit(EXIT_FAILURE,
 573                                 "All available ports are disabled. Please set portmask.\n");
 574         }
 575
 576         d = rte_distributor_create("PKT_DIST", rte_socket_id(),
 577                         rte_lcore_count() - 2,
 578                         RTE_DIST_ALG_BURST);
 579         if (d == NULL)
 580                 rte_exit(EXIT_FAILURE, "Cannot create distributor\n");
 581
 582         /*
 583          * scheduler ring is read only by the transmitter core, but written to
 584          * by multiple threads
 585          */
 586         output_ring = rte_ring_create("Output_ring", RTE_RING_SZ,
 587                         rte_socket_id(), RING_F_SC_DEQ);
 588         if (output_ring == NULL)
 589                 rte_exit(EXIT_FAILURE, "Cannot create output ring\n");
 590
 591         RTE_LCORE_FOREACH_SLAVE(lcore_id) {
 592                 if (worker_id == rte_lcore_count() - 2)
 593                         rte_eal_remote_launch((lcore_function_t *)lcore_tx,
 594                                         output_ring, lcore_id);
 595                 else {
 596                         struct lcore_params *p =
 597                                         rte_malloc(NULL, sizeof(*p), 0);
 598                         if (!p)
 599                                 rte_panic("malloc failure\n");
 600                         *p = (struct lcore_params){worker_id, d, output_ring, mbuf_pool};
 601
 602                         rte_eal_remote_launch((lcore_function_t *)lcore_worker,
 603                                         p, lcore_id);
 604                 }
 605                 worker_id++;
 606         }
 607         /* call lcore_main on master core only */
 608         struct lcore_params p = { 0, d, output_ring, mbuf_pool};
 609
 610         if (lcore_rx(&p) != 0)
 611                 return -1;
 612
 613         RTE_LCORE_FOREACH_SLAVE(lcore_id) {
 614                 if (rte_eal_wait_lcore(lcore_id) < 0)
 615                         return -1;
 616         }
 617
 618         print_stats();
 619         return 0;
 620 }