X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=examples%2Fdistributor%2Fmain.c;h=75c001d7ed5ba369efbfcea33e0f8d52ffb3afb7;hb=9872251dfe4e7bd5bbb95ef12d5781c23a75854b;hp=78fe3e9042def08d1bc862fbdf835a4a27c850e4;hpb=ea0c20ea95fd5d71a10757e6598ac66233ea1495;p=dpdk.git diff --git a/examples/distributor/main.c b/examples/distributor/main.c index 78fe3e9042..75c001d7ed 100644 --- a/examples/distributor/main.c +++ b/examples/distributor/main.c @@ -1,8 +1,7 @@ /*- * BSD LICENSE * - * Copyright(c) 2010-2014 Intel Corporation. All rights reserved. - * All rights reserved. + * Copyright(c) 2010-2017 Intel Corporation. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions @@ -42,49 +41,66 @@ #include #include #include +#include #include #define RX_RING_SIZE 256 #define TX_RING_SIZE 512 #define NUM_MBUFS ((64*1024)-1) -#define MBUF_DATA_SIZE (2048 + RTE_PKTMBUF_HEADROOM) #define MBUF_CACHE_SIZE 250 #define BURST_SIZE 32 #define RTE_RING_SZ 1024 -/* uncommnet below line to enable debug logs */ -/* #define DEBUG */ - -#ifdef DEBUG -#define LOG_LEVEL RTE_LOG_DEBUG -#define LOG_DEBUG(log_type, fmt, args...) do { \ - RTE_LOG(DEBUG, log_type, fmt, ##args) \ -} while (0) -#else -#define LOG_LEVEL RTE_LOG_INFO -#define LOG_DEBUG(log_type, fmt, args...) do {} while (0) -#endif - #define RTE_LOGTYPE_DISTRAPP RTE_LOGTYPE_USER1 +#define ANSI_COLOR_RED "\x1b[31m" +#define ANSI_COLOR_RESET "\x1b[0m" + /* mask of enabled ports */ static uint32_t enabled_port_mask; volatile uint8_t quit_signal; volatile uint8_t quit_signal_rx; +volatile uint8_t quit_signal_dist; +volatile uint8_t quit_signal_work; static volatile struct app_stats { struct { uint64_t rx_pkts; uint64_t returned_pkts; uint64_t enqueued_pkts; + uint64_t enqdrop_pkts; } rx __rte_cache_aligned; + int pad1 __rte_cache_aligned; + + struct { + uint64_t in_pkts; + uint64_t ret_pkts; + uint64_t sent_pkts; + uint64_t enqdrop_pkts; + } dist __rte_cache_aligned; + int pad2 __rte_cache_aligned; struct { uint64_t dequeue_pkts; uint64_t tx_pkts; + uint64_t enqdrop_pkts; } tx __rte_cache_aligned; + int pad3 __rte_cache_aligned; + + uint64_t worker_pkts[64] __rte_cache_aligned; + + int pad4 __rte_cache_aligned; + + uint64_t worker_bursts[64][8] __rte_cache_aligned; + + int pad5 __rte_cache_aligned; + + uint64_t port_rx_pkts[64] __rte_cache_aligned; + uint64_t port_tx_pkts[64] __rte_cache_aligned; } app_stats; +struct app_stats prev_app_stats; + static const struct rte_eth_conf port_conf_default = { .rxmode = { .mq_mode = ETH_MQ_RX_RSS, @@ -106,6 +122,8 @@ struct output_buffer { struct rte_mbuf *mbufs[BURST_SIZE]; }; +static void print_stats(void); + /* * Initialises a given port using global settings and with the rx buffers * coming from the mbuf_pool passed as parameter @@ -147,7 +165,8 @@ port_init(uint8_t port, struct rte_mempool *mbuf_pool) struct rte_eth_link link; rte_eth_link_get_nowait(port, &link); - if (!link.link_status) { + while (!link.link_status) { + printf("Waiting for Link up on port %"PRIu8"\n", port); sleep(1); rte_eth_link_get_nowait(port, &link); } @@ -178,19 +197,25 @@ struct lcore_params { struct rte_mempool *mem_pool; }; -static void +static int quit_workers(struct rte_distributor *d, struct rte_mempool *p) { const unsigned num_workers = rte_lcore_count() - 2; unsigned i; struct rte_mbuf *bufs[num_workers]; - rte_mempool_get_bulk(p, (void *)bufs, num_workers); + + if (rte_mempool_get_bulk(p, (void *)bufs, num_workers) != 0) { + printf("line %d: Error getting mbufs from pool\n", __LINE__); + return -1; + } for (i = 0; i < num_workers; i++) bufs[i]->hash.rss = i << 1; rte_distributor_process(d, bufs, num_workers); rte_mempool_put_bulk(p, (void *)bufs, num_workers); + + return 0; } static int @@ -228,19 +253,28 @@ lcore_rx(struct lcore_params *p) struct rte_mbuf *bufs[BURST_SIZE*2]; const uint16_t nb_rx = rte_eth_rx_burst(port, 0, bufs, BURST_SIZE); + if (unlikely(nb_rx == 0)) { + if (++port == nb_ports) + port = 0; + continue; + } app_stats.rx.rx_pkts += nb_rx; rte_distributor_process(d, bufs, nb_rx); const uint16_t nb_ret = rte_distributor_returned_pkts(d, bufs, BURST_SIZE*2); app_stats.rx.returned_pkts += nb_ret; - if (unlikely(nb_ret == 0)) + if (unlikely(nb_ret == 0)) { + if (++port == nb_ports) + port = 0; continue; + } uint16_t sent = rte_ring_enqueue_burst(r, (void *)bufs, nb_ret); app_stats.rx.enqueued_pkts += sent; if (unlikely(sent < nb_ret)) { - LOG_DEBUG(DISTRAPP, "%s:Packet loss due to full ring\n", __func__); + RTE_LOG_DP(DEBUG, DISTRAPP, + "%s:Packet loss due to full ring\n", __func__); while (sent < nb_ret) rte_pktmbuf_free(bufs[sent++]); } @@ -258,7 +292,8 @@ lcore_rx(struct lcore_params *p) * get packets till quit_signal is actually been * received and they gracefully shutdown */ - quit_workers(d, mem_pool); + if (quit_workers(d, mem_pool) != 0) + return -1; /* rx thread should quit at last */ return 0; } @@ -271,7 +306,8 @@ flush_one_port(struct output_buffer *outbuf, uint8_t outp) app_stats.tx.tx_pkts += nb_tx; if (unlikely(nb_tx < outbuf->count)) { - LOG_DEBUG(DISTRAPP, "%s:Packet loss with tx_burst\n", __func__); + RTE_LOG_DP(DEBUG, DISTRAPP, + "%s:Packet loss with tx_burst\n", __func__); do { rte_pktmbuf_free(outbuf->mbufs[nb_tx]); } while (++nb_tx < outbuf->count); @@ -336,13 +372,13 @@ lcore_tx(struct rte_ring *in_r) /* for traffic we receive, queue it up for transmit */ uint16_t i; - _mm_prefetch(bufs[0], 0); - _mm_prefetch(bufs[1], 0); - _mm_prefetch(bufs[2], 0); + rte_prefetch_non_temporal((void *)bufs[0]); + rte_prefetch_non_temporal((void *)bufs[1]); + rte_prefetch_non_temporal((void *)bufs[2]); for (i = 0; i < nb_rx; i++) { struct output_buffer *outbuf; uint8_t outp; - _mm_prefetch(bufs[i + 3], 0); + rte_prefetch_non_temporal((void *)bufs[i + 3]); /* * workers should update in_port to hold the * output port value @@ -374,25 +410,91 @@ static void print_stats(void) { struct rte_eth_stats eth_stats; - unsigned i; - - printf("\nRX thread stats:\n"); - printf(" - Received: %"PRIu64"\n", app_stats.rx.rx_pkts); - printf(" - Processed: %"PRIu64"\n", app_stats.rx.returned_pkts); - printf(" - Enqueued: %"PRIu64"\n", app_stats.rx.enqueued_pkts); - - printf("\nTX thread stats:\n"); - printf(" - Dequeued: %"PRIu64"\n", app_stats.tx.dequeue_pkts); - printf(" - Transmitted: %"PRIu64"\n", app_stats.tx.tx_pkts); + unsigned int i, j; + const unsigned int num_workers = rte_lcore_count() - 4; for (i = 0; i < rte_eth_dev_count(); i++) { rte_eth_stats_get(i, ð_stats); - printf("\nPort %u stats:\n", i); - printf(" - Pkts in: %"PRIu64"\n", eth_stats.ipackets); - printf(" - Pkts out: %"PRIu64"\n", eth_stats.opackets); - printf(" - In Errs: %"PRIu64"\n", eth_stats.ierrors); - printf(" - Out Errs: %"PRIu64"\n", eth_stats.oerrors); - printf(" - Mbuf Errs: %"PRIu64"\n", eth_stats.rx_nombuf); + app_stats.port_rx_pkts[i] = eth_stats.ipackets; + app_stats.port_tx_pkts[i] = eth_stats.opackets; + } + + printf("\n\nRX Thread:\n"); + for (i = 0; i < rte_eth_dev_count(); i++) { + printf("Port %u Pktsin : %5.2f\n", i, + (app_stats.port_rx_pkts[i] - + prev_app_stats.port_rx_pkts[i])/1000000.0); + prev_app_stats.port_rx_pkts[i] = app_stats.port_rx_pkts[i]; + } + printf(" - Received: %5.2f\n", + (app_stats.rx.rx_pkts - + prev_app_stats.rx.rx_pkts)/1000000.0); + printf(" - Returned: %5.2f\n", + (app_stats.rx.returned_pkts - + prev_app_stats.rx.returned_pkts)/1000000.0); + printf(" - Enqueued: %5.2f\n", + (app_stats.rx.enqueued_pkts - + prev_app_stats.rx.enqueued_pkts)/1000000.0); + printf(" - Dropped: %s%5.2f%s\n", ANSI_COLOR_RED, + (app_stats.rx.enqdrop_pkts - + prev_app_stats.rx.enqdrop_pkts)/1000000.0, + ANSI_COLOR_RESET); + + printf("Distributor thread:\n"); + printf(" - In: %5.2f\n", + (app_stats.dist.in_pkts - + prev_app_stats.dist.in_pkts)/1000000.0); + printf(" - Returned: %5.2f\n", + (app_stats.dist.ret_pkts - + prev_app_stats.dist.ret_pkts)/1000000.0); + printf(" - Sent: %5.2f\n", + (app_stats.dist.sent_pkts - + prev_app_stats.dist.sent_pkts)/1000000.0); + printf(" - Dropped %s%5.2f%s\n", ANSI_COLOR_RED, + (app_stats.dist.enqdrop_pkts - + prev_app_stats.dist.enqdrop_pkts)/1000000.0, + ANSI_COLOR_RESET); + + printf("TX thread:\n"); + printf(" - Dequeued: %5.2f\n", + (app_stats.tx.dequeue_pkts - + prev_app_stats.tx.dequeue_pkts)/1000000.0); + for (i = 0; i < rte_eth_dev_count(); i++) { + printf("Port %u Pktsout: %5.2f\n", + i, (app_stats.port_tx_pkts[i] - + prev_app_stats.port_tx_pkts[i])/1000000.0); + prev_app_stats.port_tx_pkts[i] = app_stats.port_tx_pkts[i]; + } + printf(" - Transmitted: %5.2f\n", + (app_stats.tx.tx_pkts - + prev_app_stats.tx.tx_pkts)/1000000.0); + printf(" - Dropped: %s%5.2f%s\n", ANSI_COLOR_RED, + (app_stats.tx.enqdrop_pkts - + prev_app_stats.tx.enqdrop_pkts)/1000000.0, + ANSI_COLOR_RESET); + + prev_app_stats.rx.rx_pkts = app_stats.rx.rx_pkts; + prev_app_stats.rx.returned_pkts = app_stats.rx.returned_pkts; + prev_app_stats.rx.enqueued_pkts = app_stats.rx.enqueued_pkts; + prev_app_stats.rx.enqdrop_pkts = app_stats.rx.enqdrop_pkts; + prev_app_stats.dist.in_pkts = app_stats.dist.in_pkts; + prev_app_stats.dist.ret_pkts = app_stats.dist.ret_pkts; + prev_app_stats.dist.sent_pkts = app_stats.dist.sent_pkts; + prev_app_stats.dist.enqdrop_pkts = app_stats.dist.enqdrop_pkts; + prev_app_stats.tx.dequeue_pkts = app_stats.tx.dequeue_pkts; + prev_app_stats.tx.tx_pkts = app_stats.tx.tx_pkts; + prev_app_stats.tx.enqdrop_pkts = app_stats.tx.enqdrop_pkts; + + for (i = 0; i < num_workers; i++) { + printf("Worker %02u Pkts: %5.2f. Bursts(1-8): ", i, + (app_stats.worker_pkts[i] - + prev_app_stats.worker_pkts[i])/1000000.0); + for (j = 0; j < 8; j++) { + printf("%"PRIu64" ", app_stats.worker_bursts[i][j]); + app_stats.worker_bursts[i][j] = 0; + } + printf("\n"); + prev_app_stats.worker_pkts[i] = app_stats.worker_pkts[i]; } } @@ -401,17 +503,30 @@ lcore_worker(struct lcore_params *p) { struct rte_distributor *d = p->d; const unsigned id = p->worker_id; + unsigned int num = 0; + unsigned int i; + /* * for single port, xor_val will be zero so we won't modify the output * port, otherwise we send traffic from 0 to 1, 2 to 3, and vice versa */ const unsigned xor_val = (rte_eth_dev_count() > 1); - struct rte_mbuf *buf = NULL; + struct rte_mbuf *buf[8] __rte_cache_aligned; + + for (i = 0; i < 8; i++) + buf[i] = NULL; printf("\nCore %u acting as worker core.\n", rte_lcore_id()); while (!quit_signal) { - buf = rte_distributor_get_pkt(d, id, buf); - buf->port ^= xor_val; + num = rte_distributor_get_pkt(d, id, buf, buf, num); + /* Do a little bit of work for each packet */ + for (i = 0; i < num; i++) { + uint64_t t = rte_rdtsc()+100; + + while (rte_rdtsc() < t) + rte_pause(); + buf[i]->port ^= xor_val; + } } return 0; } @@ -483,7 +598,7 @@ parse_args(int argc, char **argv) argv[optind-1] = prgname; - optind = 0; /* reset getopt lib */ + optind = 1; /* reset getopt lib */ return 0; } @@ -498,6 +613,7 @@ main(int argc, char *argv[]) unsigned nb_ports; uint8_t portid; uint8_t nb_ports_available; + uint64_t t, freq; /* catch ctrl-c so we can print on exit */ signal(SIGINT, int_handler); @@ -529,8 +645,8 @@ main(int argc, char *argv[]) "when using a single port\n"); mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL", - NUM_MBUFS * nb_ports, MBUF_CACHE_SIZE, 0, MBUF_DATA_SIZE, - rte_socket_id()); + NUM_MBUFS * nb_ports, MBUF_CACHE_SIZE, 0, + RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id()); if (mbuf_pool == NULL) rte_exit(EXIT_FAILURE, "Cannot create mbuf pool\n"); nb_ports_available = nb_ports; @@ -557,7 +673,8 @@ main(int argc, char *argv[]) } d = rte_distributor_create("PKT_DIST", rte_socket_id(), - rte_lcore_count() - 2); + rte_lcore_count() - 2, + RTE_DIST_ALG_BURST); if (d == NULL) rte_exit(EXIT_FAILURE, "Cannot create distributor\n"); @@ -588,7 +705,19 @@ main(int argc, char *argv[]) } /* call lcore_main on master core only */ struct lcore_params p = { 0, d, output_ring, mbuf_pool}; - lcore_rx(&p); + + if (lcore_rx(&p) != 0) + return -1; + + freq = rte_get_timer_hz(); + t = rte_rdtsc() + freq; + while (!quit_signal_dist) { + if (t < rte_rdtsc()) { + print_stats(); + t = rte_rdtsc() + freq; + } + usleep(1000); + } RTE_LCORE_FOREACH_SLAVE(lcore_id) { if (rte_eal_wait_lcore(lcore_id) < 0)