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