examples: fix port mask parsing failure handling
[dpdk.git] / examples / distributor / main.c
index 96d6454..dca48c2 100644 (file)
@@ -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 <stdint.h>
 #include <rte_debug.h>
 #include <rte_prefetch.h>
 #include <rte_distributor.h>
+#include <rte_pause.h>
+#include <rte_power.h>
 
-#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 MBUF_CACHE_SIZE 128
+#define BURST_SIZE 64
 #define SCHED_RX_RING_SZ 8192
 #define SCHED_TX_RING_SZ 65536
-#define RTE_RING_SZ 1024
+#define BURST_SIZE_TX 32
 
 #define RTE_LOGTYPE_DISTRAPP RTE_LOGTYPE_USER1
 
@@ -64,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 {
@@ -106,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,
@@ -131,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;
        }
@@ -166,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;
 }
@@ -203,11 +224,12 @@ struct lcore_params {
 static int
 lcore_rx(struct lcore_params *p)
 {
-       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;
@@ -229,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)) {
@@ -257,7 +278,7 @@ lcore_rx(struct lcore_params *p)
 
                struct rte_ring *tx_ring = p->dist_tx_ring;
                uint16_t sent = rte_ring_enqueue_burst(tx_ring,
-                               (void *)bufs, nb_ret);
+                               (void *)bufs, nb_ret, NULL);
 #else
                uint16_t nb_ret = nb_rx;
                /*
@@ -268,11 +289,12 @@ lcore_rx(struct lcore_params *p)
                /* struct rte_ring *out_ring = p->dist_tx_ring; */
 
                uint16_t sent = rte_ring_enqueue_burst(out_ring,
-                               (void *)bufs, nb_ret);
+                               (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,6 +303,8 @@ lcore_rx(struct lcore_params *p)
                if (++port == nb_ports)
                        port = 0;
        }
+       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;
@@ -290,13 +314,12 @@ lcore_rx(struct lcore_params *p)
 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);
@@ -305,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;
@@ -333,7 +357,7 @@ lcore_distributor(struct lcore_params *p)
        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);
+                               (void *)bufs, BURST_SIZE*1, NULL);
                if (nb_rx) {
                        app_stats.dist.in_pkts += nb_rx;
 
@@ -349,7 +373,7 @@ lcore_distributor(struct lcore_params *p)
                        app_stats.dist.ret_pkts += nb_ret;
 
                        uint16_t sent = rte_ring_enqueue_burst(out_r,
-                                       (void *)bufs, nb_ret);
+                                       (void *)bufs, nb_ret, NULL);
                        app_stats.dist.sent_pkts += sent;
                        if (unlikely(sent < nb_ret)) {
                                app_stats.dist.enqdrop_pkts += nb_ret - sent;
@@ -363,7 +387,8 @@ lcore_distributor(struct lcore_params *p)
        }
        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);
@@ -376,11 +401,10 @@ 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;
@@ -395,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;
                        }
 
@@ -431,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;
 }
 
@@ -454,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, &eth_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);
@@ -500,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);
@@ -551,12 +578,14 @@ 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_work) {
                num = rte_distributor_get_pkt(d, id, buf, buf, num);
@@ -568,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)
@@ -590,10 +647,7 @@ parse_portmask(const char *portmask)
        /* parse hexadecimal string */
        pm = strtoul(portmask, &end, 16);
        if ((portmask[0] == '\0') || (end == NULL) || (*end != '\0'))
-               return -1;
-
-       if (pm == 0)
-               return -1;
+               return 0;
 
        return pm;
 }
@@ -651,10 +705,12 @@ main(int argc, char *argv[])
        struct rte_distributor *d;
        struct rte_ring *dist_tx_ring;
        struct rte_ring *rx_dist_ring;
-       unsigned lcore_id, worker_id = 0;
+       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 */
@@ -672,15 +728,19 @@ main(int argc, char *argv[])
        if (ret < 0)
                rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
-       if (rte_lcore_count() < 4)
+       if (rte_lcore_count() < 5)
                rte_exit(EXIT_FAILURE, "Error, This application needs at "
-                               "least 4 logical cores to run:\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))
@@ -695,7 +755,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);
@@ -703,10 +763,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);
        }
 
@@ -716,7 +776,7 @@ main(int argc, char *argv[])
        }
 
        d = rte_distributor_create("PKT_DIST", rte_socket_id(),
-                       rte_lcore_count() - 3,
+                       rte_lcore_count() - 4,
                        RTE_DIST_ALG_BURST);
        if (d == NULL)
                rte_exit(EXIT_FAILURE, "Cannot create distributor\n");
@@ -735,44 +795,122 @@ main(int argc, char *argv[])
        if (rx_dist_ring == NULL)
                rte_exit(EXIT_FAILURE, "Cannot create output ring\n");
 
-       RTE_LCORE_FOREACH_SLAVE(lcore_id) {
-               if (worker_id == rte_lcore_count() - 3) {
-                       printf("Starting distributor on lcore_id %d\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);
-                       /* distributor core */
-                       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_distributor,
-                               p, lcore_id);
-               } else if (worker_id == rte_lcore_count() - 4) {
-                       printf("Starting tx  on worker_id %d, lcore_id %d\n",
-                                       worker_id, lcore_id);
-                       /* tx core */
-                       rte_eal_remote_launch((lcore_function_t *)lcore_tx,
-                                       dist_tx_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, rx_dist_ring,
-                                       dist_tx_ring, mbuf_pool};
-
-                       rte_eal_remote_launch((lcore_function_t *)lcore_worker,
-                                       p, 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;
+                       }
                }
-               worker_id++;
        }
-       /* call lcore_main on master core only */
-       struct lcore_params p = { 0, d, rx_dist_ring, dist_tx_ring, mbuf_pool};
 
-       if (lcore_rx(&p) != 0)
-               return -1;
+       /*
+        * 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 (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;
+               }
+       }
+
+       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;
@@ -790,5 +928,9 @@ main(int argc, char *argv[])
        }
 
        print_stats();
+
+       rte_free(pd);
+       rte_free(pr);
+
        return 0;
 }