X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=examples%2Fdistributor%2Fmain.c;h=567c5e98919d0a998882ff7875aa334ecdd1d7e8;hb=36f9eba4710b521b95d72f0e65e85f38c1cff9b5;hp=75c001d7ed5ba369efbfcea33e0f8d52ffb3afb7;hpb=9872251dfe4e7bd5bbb95ef12d5781c23a75854b;p=dpdk.git diff --git a/examples/distributor/main.c b/examples/distributor/main.c index 75c001d7ed..567c5e9891 100644 --- a/examples/distributor/main.c +++ b/examples/distributor/main.c @@ -1,33 +1,5 @@ -/*- - * BSD LICENSE - * - * 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 - * are met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above copyright - * notice, this list of conditions and the following disclaimer in - * the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Intel Corporation nor the names of its - * contributors may be used to endorse or promote products derived - * from this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +/* SPDX-License-Identifier: BSD-3-Clause + * Copyright(c) 2010-2017 Intel Corporation */ #include @@ -43,13 +15,17 @@ #include #include #include +#include +#include -#define RX_RING_SIZE 256 -#define TX_RING_SIZE 512 +#define RX_RING_SIZE 1024 +#define TX_RING_SIZE 1024 #define NUM_MBUFS ((64*1024)-1) -#define MBUF_CACHE_SIZE 250 -#define BURST_SIZE 32 -#define RTE_RING_SZ 1024 +#define MBUF_CACHE_SIZE 128 +#define BURST_SIZE 64 +#define SCHED_RX_RING_SZ 8192 +#define SCHED_TX_RING_SZ 65536 +#define BURST_SIZE_TX 32 #define RTE_LOGTYPE_DISTRAPP RTE_LOGTYPE_USER1 @@ -62,6 +38,7 @@ volatile uint8_t quit_signal; volatile uint8_t quit_signal_rx; volatile uint8_t quit_signal_dist; volatile uint8_t quit_signal_work; +unsigned int power_lib_initialised; static volatile struct app_stats { struct { @@ -104,7 +81,7 @@ struct app_stats prev_app_stats; static const struct rte_eth_conf port_conf_default = { .rxmode = { .mq_mode = ETH_MQ_RX_RSS, - .max_rx_pkt_len = ETHER_MAX_LEN, + .max_rx_pkt_len = RTE_ETHER_MAX_LEN, }, .txmode = { .mq_mode = ETH_MQ_TX_NONE, @@ -129,32 +106,64 @@ static void print_stats(void); * coming from the mbuf_pool passed as parameter */ static inline int -port_init(uint8_t port, struct rte_mempool *mbuf_pool) +port_init(uint16_t port, struct rte_mempool *mbuf_pool) { struct rte_eth_conf port_conf = port_conf_default; const uint16_t rxRings = 1, txRings = rte_lcore_count() - 1; int retval; uint16_t q; + uint16_t nb_rxd = RX_RING_SIZE; + uint16_t nb_txd = TX_RING_SIZE; + struct rte_eth_dev_info dev_info; + struct rte_eth_txconf txconf; - if (port >= rte_eth_dev_count()) + if (!rte_eth_dev_is_valid_port(port)) return -1; + retval = rte_eth_dev_info_get(port, &dev_info); + if (retval != 0) { + printf("Error during getting device (port %u) info: %s\n", + port, strerror(-retval)); + return retval; + } + + if (dev_info.tx_offload_capa & DEV_TX_OFFLOAD_MBUF_FAST_FREE) + port_conf.txmode.offloads |= + DEV_TX_OFFLOAD_MBUF_FAST_FREE; + + port_conf.rx_adv_conf.rss_conf.rss_hf &= + dev_info.flow_type_rss_offloads; + if (port_conf.rx_adv_conf.rss_conf.rss_hf != + port_conf_default.rx_adv_conf.rss_conf.rss_hf) { + printf("Port %u modified RSS hash function based on hardware support," + "requested:%#"PRIx64" configured:%#"PRIx64"\n", + port, + port_conf_default.rx_adv_conf.rss_conf.rss_hf, + port_conf.rx_adv_conf.rss_conf.rss_hf); + } + retval = rte_eth_dev_configure(port, rxRings, txRings, &port_conf); if (retval != 0) return retval; + retval = rte_eth_dev_adjust_nb_rx_tx_desc(port, &nb_rxd, &nb_txd); + if (retval != 0) + return retval; + for (q = 0; q < rxRings; q++) { - retval = rte_eth_rx_queue_setup(port, q, RX_RING_SIZE, + retval = rte_eth_rx_queue_setup(port, q, nb_rxd, rte_eth_dev_socket_id(port), NULL, mbuf_pool); if (retval < 0) return retval; } + txconf = dev_info.default_txconf; + txconf.offloads = port_conf.txmode.offloads; for (q = 0; q < txRings; q++) { - retval = rte_eth_tx_queue_setup(port, q, TX_RING_SIZE, + retval = rte_eth_tx_queue_setup(port, q, nb_txd, rte_eth_dev_socket_id(port), - NULL); + &txconf); if (retval < 0) return retval; } @@ -164,28 +173,42 @@ port_init(uint8_t port, struct rte_mempool *mbuf_pool) return retval; struct rte_eth_link link; - rte_eth_link_get_nowait(port, &link); - while (!link.link_status) { - printf("Waiting for Link up on port %"PRIu8"\n", port); + do { + retval = rte_eth_link_get_nowait(port, &link); + if (retval < 0) { + printf("Failed link get (port %u): %s\n", + port, rte_strerror(-retval)); + return retval; + } else if (link.link_status) + break; + + printf("Waiting for Link up on port %"PRIu16"\n", port); sleep(1); - rte_eth_link_get_nowait(port, &link); - } + } while (!link.link_status); if (!link.link_status) { - printf("Link down on port %"PRIu8"\n", port); + printf("Link down on port %"PRIu16"\n", port); return 0; } - struct ether_addr addr; - rte_eth_macaddr_get(port, &addr); + struct rte_ether_addr addr; + retval = rte_eth_macaddr_get(port, &addr); + if (retval < 0) { + printf("Failed to get MAC address (port %u): %s\n", + port, rte_strerror(-retval)); + return retval; + } + printf("Port %u MAC: %02"PRIx8" %02"PRIx8" %02"PRIx8 " %02"PRIx8" %02"PRIx8" %02"PRIx8"\n", - (unsigned)port, + port, addr.addr_bytes[0], addr.addr_bytes[1], addr.addr_bytes[2], addr.addr_bytes[3], addr.addr_bytes[4], addr.addr_bytes[5]); - rte_eth_promiscuous_enable(port); + retval = rte_eth_promiscuous_enable(port); + if (retval != 0) + return retval; return 0; } @@ -193,42 +216,20 @@ port_init(uint8_t port, struct rte_mempool *mbuf_pool) struct lcore_params { unsigned worker_id; struct rte_distributor *d; - struct rte_ring *r; + struct rte_ring *rx_dist_ring; + struct rte_ring *dist_tx_ring; struct rte_mempool *mem_pool; }; -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]; - - 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 lcore_rx(struct lcore_params *p) { - struct rte_distributor *d = p->d; - struct rte_mempool *mem_pool = p->mem_pool; - struct rte_ring *r = p->r; - const uint8_t nb_ports = rte_eth_dev_count(); + const uint16_t nb_ports = rte_eth_dev_count_avail(); const int socket_id = rte_socket_id(); - uint8_t port; + uint16_t port; + struct rte_mbuf *bufs[BURST_SIZE*2]; - for (port = 0; port < nb_ports; port++) { + RTE_ETH_FOREACH_DEV(port) { /* skip ports that are not enabled */ if ((enabled_port_mask & (1 << port)) == 0) continue; @@ -250,7 +251,6 @@ lcore_rx(struct lcore_params *p) port = 0; continue; } - 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)) { @@ -260,9 +260,15 @@ lcore_rx(struct lcore_params *p) } 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); +/* + * You can run the distributor on the rx core with this code. Returned + * packets are then send straight to the tx core. + */ +#if 0 + rte_distributor_process(d, bufs, nb_rx); + const uint16_t nb_ret = rte_distributor_returned_pktsd, + bufs, BURST_SIZE*2); + app_stats.rx.returned_pkts += nb_ret; if (unlikely(nb_ret == 0)) { if (++port == nb_ports) @@ -270,9 +276,25 @@ lcore_rx(struct lcore_params *p) continue; } - uint16_t sent = rte_ring_enqueue_burst(r, (void *)bufs, nb_ret); + struct rte_ring *tx_ring = p->dist_tx_ring; + uint16_t sent = rte_ring_enqueue_burst(tx_ring, + (void *)bufs, nb_ret, NULL); +#else + uint16_t nb_ret = nb_rx; + /* + * Swap the following two lines if you want the rx traffic + * to go directly to tx, no distribution. + */ + struct rte_ring *out_ring = p->rx_dist_ring; + /* struct rte_ring *out_ring = p->dist_tx_ring; */ + + uint16_t sent = rte_ring_enqueue_burst(out_ring, + (void *)bufs, nb_ret, NULL); +#endif + app_stats.rx.enqueued_pkts += sent; if (unlikely(sent < nb_ret)) { + app_stats.rx.enqdrop_pkts += nb_ret - sent; RTE_LOG_DP(DEBUG, DISTRAPP, "%s:Packet loss due to full ring\n", __func__); while (sent < nb_ret) @@ -281,33 +303,23 @@ lcore_rx(struct lcore_params *p) if (++port == nb_ports) port = 0; } - rte_distributor_process(d, NULL, 0); - /* flush distributor to bring to known state */ - rte_distributor_flush(d); + if (power_lib_initialised) + rte_power_exit(rte_lcore_id()); /* set worker & tx threads quit flag */ + printf("\nCore %u exiting rx task.\n", rte_lcore_id()); quit_signal = 1; - /* - * worker threads may hang in get packet as - * distributor process is not running, just make sure workers - * get packets till quit_signal is actually been - * received and they gracefully shutdown - */ - if (quit_workers(d, mem_pool) != 0) - return -1; - /* rx thread should quit at last */ return 0; } static inline void flush_one_port(struct output_buffer *outbuf, uint8_t outp) { - unsigned nb_tx = rte_eth_tx_burst(outp, 0, outbuf->mbufs, - outbuf->count); - app_stats.tx.tx_pkts += nb_tx; + unsigned int nb_tx = rte_eth_tx_burst(outp, 0, + outbuf->mbufs, outbuf->count); + app_stats.tx.tx_pkts += outbuf->count; if (unlikely(nb_tx < outbuf->count)) { - RTE_LOG_DP(DEBUG, DISTRAPP, - "%s:Packet loss with tx_burst\n", __func__); + app_stats.tx.enqdrop_pkts += outbuf->count - nb_tx; do { rte_pktmbuf_free(outbuf->mbufs[nb_tx]); } while (++nb_tx < outbuf->count); @@ -316,10 +328,11 @@ flush_one_port(struct output_buffer *outbuf, uint8_t outp) } static inline void -flush_all_ports(struct output_buffer *tx_buffers, uint8_t nb_ports) +flush_all_ports(struct output_buffer *tx_buffers) { - uint8_t outp; - for (outp = 0; outp < nb_ports; outp++) { + uint16_t outp; + + RTE_ETH_FOREACH_DEV(outp) { /* skip ports that are not enabled */ if ((enabled_port_mask & (1 << outp)) == 0) continue; @@ -331,15 +344,67 @@ flush_all_ports(struct output_buffer *tx_buffers, uint8_t nb_ports) } } + + +static int +lcore_distributor(struct lcore_params *p) +{ + struct rte_ring *in_r = p->rx_dist_ring; + struct rte_ring *out_r = p->dist_tx_ring; + struct rte_mbuf *bufs[BURST_SIZE * 4]; + struct rte_distributor *d = p->d; + + printf("\nCore %u acting as distributor core.\n", rte_lcore_id()); + while (!quit_signal_dist) { + const uint16_t nb_rx = rte_ring_dequeue_burst(in_r, + (void *)bufs, BURST_SIZE*1, NULL); + if (nb_rx) { + app_stats.dist.in_pkts += nb_rx; + + /* Distribute the packets */ + rte_distributor_process(d, bufs, nb_rx); + /* Handle Returns */ + const uint16_t nb_ret = + rte_distributor_returned_pkts(d, + bufs, BURST_SIZE*2); + + if (unlikely(nb_ret == 0)) + continue; + app_stats.dist.ret_pkts += nb_ret; + + uint16_t sent = rte_ring_enqueue_burst(out_r, + (void *)bufs, nb_ret, NULL); + app_stats.dist.sent_pkts += sent; + if (unlikely(sent < nb_ret)) { + app_stats.dist.enqdrop_pkts += nb_ret - sent; + RTE_LOG(DEBUG, DISTRAPP, + "%s:Packet loss due to full out ring\n", + __func__); + while (sent < nb_ret) + rte_pktmbuf_free(bufs[sent++]); + } + } + } + printf("\nCore %u exiting distributor task.\n", rte_lcore_id()); + quit_signal_work = 1; + if (power_lib_initialised) + rte_power_exit(rte_lcore_id()); + rte_distributor_flush(d); + /* Unblock any returns so workers can exit */ + rte_distributor_clear_returns(d); + quit_signal_rx = 1; + return 0; +} + + static int lcore_tx(struct rte_ring *in_r) { static struct output_buffer tx_buffers[RTE_MAX_ETHPORTS]; - const uint8_t nb_ports = rte_eth_dev_count(); const int socket_id = rte_socket_id(); - uint8_t port; + uint16_t port; - for (port = 0; port < nb_ports; port++) { + RTE_ETH_FOREACH_DEV(port) { /* skip ports that are not enabled */ if ((enabled_port_mask & (1 << port)) == 0) continue; @@ -354,19 +419,19 @@ lcore_tx(struct rte_ring *in_r) printf("\nCore %u doing packet TX.\n", rte_lcore_id()); while (!quit_signal) { - for (port = 0; port < nb_ports; port++) { + RTE_ETH_FOREACH_DEV(port) { /* skip ports that are not enabled */ if ((enabled_port_mask & (1 << port)) == 0) continue; - struct rte_mbuf *bufs[BURST_SIZE]; + struct rte_mbuf *bufs[BURST_SIZE_TX]; const uint16_t nb_rx = rte_ring_dequeue_burst(in_r, - (void *)bufs, BURST_SIZE); + (void *)bufs, BURST_SIZE_TX, NULL); app_stats.tx.dequeue_pkts += nb_rx; /* if we get no traffic, flush anything we have */ if (unlikely(nb_rx == 0)) { - flush_all_ports(tx_buffers, nb_ports); + flush_all_ports(tx_buffers); continue; } @@ -390,11 +455,14 @@ lcore_tx(struct rte_ring *in_r) outbuf = &tx_buffers[outp]; outbuf->mbufs[outbuf->count++] = bufs[i]; - if (outbuf->count == BURST_SIZE) + if (outbuf->count == BURST_SIZE_TX) flush_one_port(outbuf, outp); } } } + if (power_lib_initialised) + rte_power_exit(rte_lcore_id()); + printf("\nCore %u exiting tx task.\n", rte_lcore_id()); return 0; } @@ -403,7 +471,7 @@ int_handler(int sig_num) { printf("Exiting on signal %d\n", sig_num); /* set quit flag for rx thread to exit */ - quit_signal_rx = 1; + quit_signal_dist = 1; } static void @@ -413,14 +481,14 @@ print_stats(void) unsigned int i, j; const unsigned int num_workers = rte_lcore_count() - 4; - for (i = 0; i < rte_eth_dev_count(); i++) { + RTE_ETH_FOREACH_DEV(i) { rte_eth_stats_get(i, ð_stats); 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++) { + RTE_ETH_FOREACH_DEV(i) { printf("Port %u Pktsin : %5.2f\n", i, (app_stats.port_rx_pkts[i] - prev_app_stats.port_rx_pkts[i])/1000000.0); @@ -459,7 +527,7 @@ print_stats(void) 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++) { + RTE_ETH_FOREACH_DEV(i) { printf("Port %u Pktsout: %5.2f\n", i, (app_stats.port_tx_pkts[i] - prev_app_stats.port_tx_pkts[i])/1000000.0); @@ -510,14 +578,16 @@ lcore_worker(struct lcore_params *p) * 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); + const unsigned xor_val = (rte_eth_dev_count_avail() > 1); struct rte_mbuf *buf[8] __rte_cache_aligned; for (i = 0; i < 8; i++) buf[i] = NULL; + app_stats.worker_pkts[p->worker_id] = 1; + printf("\nCore %u acting as worker core.\n", rte_lcore_id()); - while (!quit_signal) { + while (!quit_signal_work) { 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++) { @@ -527,10 +597,38 @@ lcore_worker(struct lcore_params *p) rte_pause(); buf[i]->port ^= xor_val; } + + app_stats.worker_pkts[p->worker_id] += num; + if (num > 0) + app_stats.worker_bursts[p->worker_id][num-1]++; } + if (power_lib_initialised) + rte_power_exit(rte_lcore_id()); + rte_free(p); return 0; } +static int +init_power_library(void) +{ + int ret = 0, lcore_id; + RTE_LCORE_FOREACH_SLAVE(lcore_id) { + /* init power management library */ + ret = rte_power_init(lcore_id); + if (ret) { + RTE_LOG(ERR, POWER, + "Library initialization failed on core %u\n", + lcore_id); + /* + * Return on first failure, we'll fall back + * to non-power operation + */ + return ret; + } + } + return ret; +} + /* display usage */ static void print_usage(const char *prgname) @@ -608,11 +706,14 @@ main(int argc, char *argv[]) { struct rte_mempool *mbuf_pool; struct rte_distributor *d; - struct rte_ring *output_ring; - unsigned lcore_id, worker_id = 0; + struct rte_ring *dist_tx_ring; + struct rte_ring *rx_dist_ring; + struct rte_power_core_capabilities lcore_cap; + unsigned int lcore_id, worker_id = 0; + int distr_core_id = -1, rx_core_id = -1, tx_core_id = -1; unsigned nb_ports; - uint8_t portid; - uint8_t nb_ports_available; + uint16_t portid; + uint16_t nb_ports_available; uint64_t t, freq; /* catch ctrl-c so we can print on exit */ @@ -630,14 +731,19 @@ main(int argc, char *argv[]) if (ret < 0) rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n"); - if (rte_lcore_count() < 3) + if (rte_lcore_count() < 5) rte_exit(EXIT_FAILURE, "Error, This application needs at " - "least 3 logical cores to run:\n" - "1 lcore for packet RX and distribution\n" + "least 5 logical cores to run:\n" + "1 lcore for stats (can be core 0)\n" + "1 lcore for packet RX\n" + "1 lcore for distribution\n" "1 lcore for packet TX\n" "and at least 1 lcore for worker threads\n"); - nb_ports = rte_eth_dev_count(); + if (init_power_library() == 0) + power_lib_initialised = 1; + + nb_ports = rte_eth_dev_count_avail(); if (nb_ports == 0) rte_exit(EXIT_FAILURE, "Error: no ethernet ports detected\n"); if (nb_ports != 1 && (nb_ports & 1)) @@ -652,7 +758,7 @@ main(int argc, char *argv[]) nb_ports_available = nb_ports; /* initialize all ports */ - for (portid = 0; portid < nb_ports; portid++) { + RTE_ETH_FOREACH_DEV(portid) { /* skip ports that are not enabled */ if ((enabled_port_mask & (1 << portid)) == 0) { printf("\nSkipping disabled port %d\n", portid); @@ -660,10 +766,10 @@ main(int argc, char *argv[]) continue; } /* init port */ - printf("Initializing port %u... done\n", (unsigned) portid); + printf("Initializing port %u... done\n", portid); if (port_init(portid, mbuf_pool) != 0) - rte_exit(EXIT_FAILURE, "Cannot initialize port %"PRIu8"\n", + rte_exit(EXIT_FAILURE, "Cannot initialize port %u\n", portid); } @@ -673,41 +779,141 @@ main(int argc, char *argv[]) } d = rte_distributor_create("PKT_DIST", rte_socket_id(), - rte_lcore_count() - 2, + rte_lcore_count() - 4, RTE_DIST_ALG_BURST); if (d == NULL) rte_exit(EXIT_FAILURE, "Cannot create distributor\n"); /* - * scheduler ring is read only by the transmitter core, but written to - * by multiple threads + * scheduler ring is read by the transmitter core, and written to + * by scheduler core */ - output_ring = rte_ring_create("Output_ring", RTE_RING_SZ, - rte_socket_id(), RING_F_SC_DEQ); - if (output_ring == NULL) + dist_tx_ring = rte_ring_create("Output_ring", SCHED_TX_RING_SZ, + rte_socket_id(), RING_F_SC_DEQ | RING_F_SP_ENQ); + if (dist_tx_ring == NULL) + rte_exit(EXIT_FAILURE, "Cannot create output ring\n"); + + rx_dist_ring = rte_ring_create("Input_ring", SCHED_RX_RING_SZ, + rte_socket_id(), RING_F_SC_DEQ | RING_F_SP_ENQ); + if (rx_dist_ring == NULL) rte_exit(EXIT_FAILURE, "Cannot create output ring\n"); + if (power_lib_initialised) { + /* + * Here we'll pre-assign lcore ids to the rx, tx and + * distributor workloads if there's higher frequency + * on those cores e.g. if Turbo Boost is enabled. + * It's also worth mentioning that it will assign cores in a + * specific order, so that if there's less than three + * available, the higher frequency cores will go to the + * distributor first, then rx, then tx. + */ + RTE_LCORE_FOREACH_SLAVE(lcore_id) { + + rte_power_get_capabilities(lcore_id, &lcore_cap); + + if (lcore_cap.priority != 1) + continue; + + if (distr_core_id < 0) { + distr_core_id = lcore_id; + printf("Distributor on priority core %d\n", + lcore_id); + continue; + } + if (rx_core_id < 0) { + rx_core_id = lcore_id; + printf("Rx on priority core %d\n", + lcore_id); + continue; + } + if (tx_core_id < 0) { + tx_core_id = lcore_id; + printf("Tx on priority core %d\n", + lcore_id); + continue; + } + } + } + + /* + * If there's any of the key workloads left without an lcore_id + * after the high performing core assignment above, pre-assign + * them here. + */ RTE_LCORE_FOREACH_SLAVE(lcore_id) { - if (worker_id == rte_lcore_count() - 2) - rte_eal_remote_launch((lcore_function_t *)lcore_tx, - output_ring, lcore_id); - else { - struct lcore_params *p = - rte_malloc(NULL, sizeof(*p), 0); - if (!p) - rte_panic("malloc failure\n"); - *p = (struct lcore_params){worker_id, d, output_ring, mbuf_pool}; - - rte_eal_remote_launch((lcore_function_t *)lcore_worker, - p, lcore_id); + if (lcore_id == (unsigned int)distr_core_id || + lcore_id == (unsigned int)rx_core_id || + lcore_id == (unsigned int)tx_core_id) + continue; + if (distr_core_id < 0) { + distr_core_id = lcore_id; + printf("Distributor on core %d\n", lcore_id); + continue; + } + if (rx_core_id < 0) { + rx_core_id = lcore_id; + printf("Rx on core %d\n", lcore_id); + continue; + } + if (tx_core_id < 0) { + tx_core_id = lcore_id; + printf("Tx on core %d\n", lcore_id); + continue; } - worker_id++; } - /* call lcore_main on master core only */ - struct lcore_params p = { 0, d, output_ring, mbuf_pool}; - if (lcore_rx(&p) != 0) - return -1; + printf(" tx id %d, dist id %d, rx id %d\n", + tx_core_id, + distr_core_id, + rx_core_id); + + /* + * Kick off all the worker threads first, avoiding the pre-assigned + * lcore_ids for tx, rx and distributor workloads. + */ + RTE_LCORE_FOREACH_SLAVE(lcore_id) { + if (lcore_id == (unsigned int)distr_core_id || + lcore_id == (unsigned int)rx_core_id || + lcore_id == (unsigned int)tx_core_id) + continue; + printf("Starting thread %d as worker, lcore_id %d\n", + worker_id, lcore_id); + struct lcore_params *p = + rte_malloc(NULL, sizeof(*p), 0); + if (!p) + rte_panic("malloc failure\n"); + *p = (struct lcore_params){worker_id++, d, rx_dist_ring, + dist_tx_ring, mbuf_pool}; + + rte_eal_remote_launch((lcore_function_t *)lcore_worker, + p, lcore_id); + } + + /* Start tx core */ + rte_eal_remote_launch((lcore_function_t *)lcore_tx, + dist_tx_ring, tx_core_id); + + /* Start distributor core */ + struct lcore_params *pd = + rte_malloc(NULL, sizeof(*pd), 0); + if (!pd) + rte_panic("malloc failure\n"); + *pd = (struct lcore_params){worker_id++, d, + rx_dist_ring, dist_tx_ring, mbuf_pool}; + rte_eal_remote_launch( + (lcore_function_t *)lcore_distributor, + pd, distr_core_id); + + /* Start rx core */ + struct lcore_params *pr = + rte_malloc(NULL, sizeof(*pr), 0); + if (!pr) + rte_panic("malloc failure\n"); + *pr = (struct lcore_params){worker_id++, d, rx_dist_ring, + dist_tx_ring, mbuf_pool}; + rte_eal_remote_launch((lcore_function_t *)lcore_rx, + pr, rx_core_id); freq = rte_get_timer_hz(); t = rte_rdtsc() + freq; @@ -725,5 +931,9 @@ main(int argc, char *argv[]) } print_stats(); + + rte_free(pd); + rte_free(pr); + return 0; }