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