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