[PATCH] examples/distributor: update dynamic configuration

[PATCH] examples/distributor: update dynamic configuration

[Abdullah Ömer Yamaç]

In this patch,
	* It is possible to switch the running mode of the distributor
using the command line argument.
	* With "-c" parameter, you can run RX and Distributor
on the same core.
	* Without "-c" parameter, you can run RX and Distributor
on the different core.
	* Syntax error of the single RX and distributor core is fixed.
	* When "-c" parameter is active, the wasted distributor core is
also deactivated in the main function.

Fixes: 4a7f40c0ff9a ("examples/distributor: add dedicated core")
Cc: stable@dpdk.org

Signed-off-by: Abdullah Ömer Yamaç <omer.yamac@ceng.metu.edu.tr>

---
Cc: david.hunt@intel.com
---
 doc/guides/sample_app_ug/dist_app.rst |   3 +-
 examples/distributor/main.c           | 205 +++++++++++++++++++-------
 2 files changed, 152 insertions(+), 56 deletions(-)

diff --git a/doc/guides/sample_app_ug/dist_app.rst b/doc/guides/sample_app_ug/dist_app.rst
index 3bd03905c3..5c80561187 100644
--- a/doc/guides/sample_app_ug/dist_app.rst
+++ b/doc/guides/sample_app_ug/dist_app.rst
@@ -42,11 +42,12 @@ Running the Application
 
    ..  code-block:: console
 
-       ./<build-dir>/examples/dpdk-distributor [EAL options] -- -p PORTMASK
+       ./<build-dir>/examples/dpdk-distributor [EAL options] -- -p PORTMASK [-c]
 
    where,
 
    *   -p PORTMASK: Hexadecimal bitmask of ports to configure
+   *   -c: Combines the RX core with distribution core
 
 #. To run the application in linux environment with 10 lcores, 4 ports,
    issue the command:
diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 02bf91f555..40b30021ca 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -39,6 +39,7 @@ volatile uint8_t quit_signal_rx;
 volatile uint8_t quit_signal_dist;
 volatile uint8_t quit_signal_work;
 unsigned int power_lib_initialised;
+bool enable_lcore_rx_distributor;
 
 static volatile struct app_stats {
 	struct {
@@ -256,14 +257,82 @@ lcore_rx(struct lcore_params *p)
 		}
 		app_stats.rx.rx_pkts += nb_rx;
 
-/*
- * 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);
+		/*
+		 * 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_rx, NULL);
+
+		app_stats.rx.enqueued_pkts += sent;
+		if (unlikely(sent < nb_rx)) {
+			app_stats.rx.enqdrop_pkts +=  nb_rx - sent;
+			RTE_LOG_DP(DEBUG, DISTRAPP,
+				"%s:Packet loss due to full ring\n", __func__);
+			while (sent < nb_rx)
+				rte_pktmbuf_free(bufs[sent++]);
+		}
+		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;
+	return 0;
+}
+
+static int
+lcore_rx_and_distributor(struct lcore_params *p)
+{
+	struct rte_distributor *d = p->d;
+	const uint16_t nb_ports = rte_eth_dev_count_avail();
+	const int socket_id = rte_socket_id();
+	uint16_t port;
+	struct rte_mbuf *bufs[BURST_SIZE*2];
+
+	RTE_ETH_FOREACH_DEV(port) {
+		/* skip ports that are not enabled */
+		if ((enabled_port_mask & (1 << port)) == 0)
+			continue;
+
+		if (rte_eth_dev_socket_id(port) > 0 &&
+				rte_eth_dev_socket_id(port) != socket_id)
+			printf("WARNING, port %u is on remote NUMA node to "
+					"RX thread.\n\tPerformance will not "
+					"be optimal.\n", port);
+	}
+
+	printf("\nCore %u doing packet RX and Distributor.\n", rte_lcore_id());
+	port = 0;
+	while (!quit_signal_rx) {
+
+		/* skip ports that are not enabled */
+		if ((enabled_port_mask & (1 << port)) == 0) {
+			if (++port == nb_ports)
+				port = 0;
+			continue;
+		}
+		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;
+
+		/*
+		 * Run the distributor on the rx core. Returned
+		 * packets are then send straight to the tx core.
+		 */
+		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)) {
@@ -275,18 +344,6 @@ 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, 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)) {
@@ -475,7 +532,11 @@ print_stats(void)
 {
 	struct rte_eth_stats eth_stats;
 	unsigned int i, j;
-	const unsigned int num_workers = rte_lcore_count() - 4;
+	unsigned int num_workers;
+	if (enable_lcore_rx_distributor)
+		num_workers = rte_lcore_count() - 3;
+	else
+		num_workers = rte_lcore_count() - 4;
 
 	RTE_ETH_FOREACH_DEV(i) {
 		rte_eth_stats_get(i, &eth_stats);
@@ -504,20 +565,22 @@ print_stats(void)
 			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);
+	if (!enable_lcore_rx_distributor) {
+		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",
@@ -629,8 +692,9 @@ init_power_library(void)
 static void
 print_usage(const char *prgname)
 {
-	printf("%s [EAL options] -- -p PORTMASK\n"
-			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n",
+	printf("%s [EAL options] -- -p PORTMASK [-c]\n"
+			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n"
+			"  -c: Combines the RX core with the distribution core\n",
 			prgname);
 }
 
@@ -662,7 +726,7 @@ parse_args(int argc, char **argv)
 
 	argvopt = argv;
 
-	while ((opt = getopt_long(argc, argvopt, "p:",
+	while ((opt = getopt_long(argc, argvopt, "cp:",
 			lgopts, &option_index)) != EOF) {
 
 		switch (opt) {
@@ -676,6 +740,10 @@ parse_args(int argc, char **argv)
 			}
 			break;
 
+		case 'c':
+			enable_lcore_rx_distributor = true;
+			break;
+
 		default:
 			print_usage(prgname);
 			return -1;
@@ -705,9 +773,11 @@ main(int argc, char *argv[])
 	unsigned int lcore_id, worker_id = 0;
 	int distr_core_id = -1, rx_core_id = -1, tx_core_id = -1;
 	unsigned nb_ports;
+	unsigned int num_workers;
 	uint16_t portid;
 	uint16_t nb_ports_available;
 	uint64_t t, freq;
+	enable_lcore_rx_distributor = false;
 
 	/* catch ctrl-c so we can print on exit */
 	signal(SIGINT, int_handler);
@@ -724,7 +794,12 @@ main(int argc, char *argv[])
 	if (ret < 0)
 		rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
-	if (rte_lcore_count() < 5)
+	if (enable_lcore_rx_distributor)
+		num_workers = rte_lcore_count() - 3;
+	else
+		num_workers = rte_lcore_count() - 4;
+
+	if (rte_lcore_count() < 5 && !enable_lcore_rx_distributor)
 		rte_exit(EXIT_FAILURE, "Error, This application needs at "
 				"least 5 logical cores to run:\n"
 				"1 lcore for stats (can be core 0)\n"
@@ -733,6 +808,15 @@ main(int argc, char *argv[])
 				"1 lcore for packet TX\n"
 				"and at least 1 lcore for worker threads\n");
 
+	if (rte_lcore_count() < 4 && enable_lcore_rx_distributor)
+		rte_exit(EXIT_FAILURE, "Error, This application needs at "
+				"least 4 logical cores to run:\n"
+				"1 lcore for stats (can be core 0)\n"
+				"1 lcore for packet RX and distribution\n"
+				"1 lcore for packet TX\n"
+				"and at least 1 lcore for worker threads\n");
+
+
 	if (init_power_library() == 0)
 		power_lib_initialised = 1;
 
@@ -772,7 +856,7 @@ main(int argc, char *argv[])
 	}
 
 	d = rte_distributor_create("PKT_DIST", rte_socket_id(),
-			rte_lcore_count() - 4,
+			num_workers,
 			RTE_DIST_ALG_BURST);
 	if (d == NULL)
 		rte_exit(EXIT_FAILURE, "Cannot create distributor\n");
@@ -808,7 +892,7 @@ main(int argc, char *argv[])
 			if (lcore_cap.priority != 1)
 				continue;
 
-			if (distr_core_id < 0) {
+			if (distr_core_id < 0 && !enable_lcore_rx_distributor) {
 				distr_core_id = lcore_id;
 				printf("Distributor on priority core %d\n",
 					lcore_id);
@@ -839,7 +923,7 @@ main(int argc, char *argv[])
 				lcore_id == (unsigned int)rx_core_id ||
 				lcore_id == (unsigned int)tx_core_id)
 			continue;
-		if (distr_core_id < 0) {
+		if (distr_core_id < 0 && !enable_lcore_rx_distributor) {
 			distr_core_id = lcore_id;
 			printf("Distributor on core %d\n", lcore_id);
 			continue;
@@ -856,7 +940,12 @@ main(int argc, char *argv[])
 		}
 	}
 
-	printf(" tx id %d, dist id %d, rx id %d\n",
+	if (enable_lcore_rx_distributor)
+		printf(" tx id %d, rx id %d\n",
+			tx_core_id,
+			rx_core_id);
+	else
+		printf(" tx id %d, dist id %d, rx id %d\n",
 			tx_core_id,
 			distr_core_id,
 			rx_core_id);
@@ -888,15 +977,16 @@ main(int argc, char *argv[])
 			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);
+	struct lcore_params *pd = NULL;
+	if (!enable_lcore_rx_distributor) {
+		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 =
@@ -905,8 +995,12 @@ main(int argc, char *argv[])
 		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);
+	if (enable_lcore_rx_distributor)
+		rte_eal_remote_launch((lcore_function_t *)lcore_rx_and_distributor,
+				pr, rx_core_id);
+	else
+		rte_eal_remote_launch((lcore_function_t *)lcore_rx,
+				pr, rx_core_id);
 
 	freq = rte_get_timer_hz();
 	t = rte_rdtsc() + freq;
@@ -925,7 +1019,8 @@ main(int argc, char *argv[])
 
 	print_stats();
 
-	rte_free(pd);
+	if (pd)
+		rte_free(pd);
 	rte_free(pr);
 
 	/* clean up the EAL */
-- 
2.27.0

[PATCH v2] examples/distributor: update dynamic configuration

[Abdullah Ömer Yamaç]

In this patch,
    * It is possible to switch the running mode of the distributor
using the command line argument.
    * With "-c" parameter, you can run RX and Distributor
on the same core.
    * Without "-c" parameter, you can run RX and Distributor
on the different core.
    * Syntax error of the single RX and distributor core is fixed.
    * Consecutive termination of the lcores fixed.
The termination order was wrong, and you couldn't terminate the
application while traffic was capturing. The current order is
RX -> Distributor -> TX -> Workers
    * When "-c" parameter is active, the wasted distributor core is
also deactivated in the main function.

Fixes: 4a7f40c0ff9a ("examples/distributor: add dedicated core")
Cc: stable@dpdk.org

Signed-off-by: Abdullah Ömer Yamaç <omer.yamac@ceng.metu.edu.tr>

---
Cc: david.hunt@intel.com
---
 doc/guides/sample_app_ug/dist_app.rst |   3 +-
 examples/distributor/main.c           | 222 ++++++++++++++++++--------
 2 files changed, 159 insertions(+), 66 deletions(-)

diff --git a/doc/guides/sample_app_ug/dist_app.rst b/doc/guides/sample_app_ug/dist_app.rst
index 3bd03905c3..5c80561187 100644
--- a/doc/guides/sample_app_ug/dist_app.rst
+++ b/doc/guides/sample_app_ug/dist_app.rst
@@ -42,11 +42,12 @@ Running the Application
 
    ..  code-block:: console
 
-       ./<build-dir>/examples/dpdk-distributor [EAL options] -- -p PORTMASK
+       ./<build-dir>/examples/dpdk-distributor [EAL options] -- -p PORTMASK [-c]
 
    where,
 
    *   -p PORTMASK: Hexadecimal bitmask of ports to configure
+   *   -c: Combines the RX core with distribution core
 
 #. To run the application in linux environment with 10 lcores, 4 ports,
    issue the command:
diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 8995806b4e..632d5e5554 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -39,6 +39,8 @@ volatile uint8_t quit_signal_rx;
 volatile uint8_t quit_signal_dist;
 volatile uint8_t quit_signal_work;
 unsigned int power_lib_initialised;
+bool enable_lcore_rx_distributor;
+unsigned int num_workers;
 
 static volatile struct app_stats {
 	struct {
@@ -256,14 +258,82 @@ lcore_rx(struct lcore_params *p)
 		}
 		app_stats.rx.rx_pkts += nb_rx;
 
-/*
- * 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(p->d, bufs, nb_rx);
-		const uint16_t nb_ret = rte_distributor_returned_pkts(p->d,
-			bufs, BURST_SIZE*2);
+		/*
+		 * 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_rx, NULL);
+
+		app_stats.rx.enqueued_pkts += sent;
+		if (unlikely(sent < nb_rx)) {
+			app_stats.rx.enqdrop_pkts +=  nb_rx - sent;
+			RTE_LOG_DP(DEBUG, DISTRAPP,
+				"%s:Packet loss due to full ring\n", __func__);
+			while (sent < nb_rx)
+				rte_pktmbuf_free(bufs[sent++]);
+		}
+		if (++port == nb_ports)
+			port = 0;
+	}
+	if (power_lib_initialised)
+		rte_power_exit(rte_lcore_id());
+	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
+	/* set distributor threads quit flag */
+	quit_signal_dist = 1;
+	return 0;
+}
+
+static int
+lcore_rx_and_distributor(struct lcore_params *p)
+{
+	struct rte_distributor *d = p->d;
+	const uint16_t nb_ports = rte_eth_dev_count_avail();
+	const int socket_id = rte_socket_id();
+	uint16_t port;
+	struct rte_mbuf *bufs[BURST_SIZE*2];
+
+	RTE_ETH_FOREACH_DEV(port) {
+		/* skip ports that are not enabled */
+		if ((enabled_port_mask & (1 << port)) == 0)
+			continue;
+
+		if (rte_eth_dev_socket_id(port) > 0 &&
+				rte_eth_dev_socket_id(port) != socket_id)
+			printf("WARNING, port %u is on remote NUMA node to "
+					"RX thread.\n\tPerformance will not "
+					"be optimal.\n", port);
+	}
+
+	printf("\nCore %u doing packet RX and Distributor.\n", rte_lcore_id());
+	port = 0;
+	while (!quit_signal_rx) {
+
+		/* skip ports that are not enabled */
+		if ((enabled_port_mask & (1 << port)) == 0) {
+			if (++port == nb_ports)
+				port = 0;
+			continue;
+		}
+		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;
+
+		/*
+		 * Run the distributor on the rx core. Returned
+		 * packets are then send straight to the tx core.
+		 */
+		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)) {
@@ -275,18 +345,6 @@ 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, 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)) {
@@ -301,9 +359,14 @@ lcore_rx(struct lcore_params *p)
 	}
 	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());
+	/* set tx threads quit flag */
 	quit_signal = 1;
+	/* set worker threads quit flag */
+	quit_signal_work = 1;
+	rte_distributor_flush(d);
+	/* Unblock any returns so workers can exit */
+	rte_distributor_clear_returns(d);
 	return 0;
 }
 
@@ -381,14 +444,16 @@ 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());
+	printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
+	/* set tx threads quit flag */
+	quit_signal = 1;
+	/* set worker threads quit flag */
+	quit_signal_work = 1;
 	rte_distributor_flush(d);
 	/* Unblock any returns so workers can exit */
 	rte_distributor_clear_returns(d);
-	quit_signal_rx = 1;
 	return 0;
 }
 
@@ -467,7 +532,7 @@ int_handler(int sig_num)
 {
 	printf("Exiting on signal %d\n", sig_num);
 	/* set quit flag for rx thread to exit */
-	quit_signal_dist = 1;
+	quit_signal_rx = 1;
 }
 
 static void
@@ -475,7 +540,6 @@ print_stats(void)
 {
 	struct rte_eth_stats eth_stats;
 	unsigned int i, j;
-	const unsigned int num_workers = rte_lcore_count() - 4;
 
 	RTE_ETH_FOREACH_DEV(i) {
 		rte_eth_stats_get(i, &eth_stats);
@@ -504,20 +568,22 @@ print_stats(void)
 			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);
+	if (!enable_lcore_rx_distributor) {
+		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",
@@ -629,8 +695,9 @@ init_power_library(void)
 static void
 print_usage(const char *prgname)
 {
-	printf("%s [EAL options] -- -p PORTMASK\n"
-			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n",
+	printf("%s [EAL options] -- -p PORTMASK [-c]\n"
+			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n"
+			"  -c: Combines the RX core with the distribution core\n",
 			prgname);
 }
 
@@ -661,8 +728,8 @@ parse_args(int argc, char **argv)
 	};
 
 	argvopt = argv;
-
-	while ((opt = getopt_long(argc, argvopt, "p:",
+	enable_lcore_rx_distributor = false;
+	while ((opt = getopt_long(argc, argvopt, "cp:",
 			lgopts, &option_index)) != EOF) {
 
 		switch (opt) {
@@ -676,6 +743,10 @@ parse_args(int argc, char **argv)
 			}
 			break;
 
+		case 'c':
+			enable_lcore_rx_distributor = true;
+			break;
+
 		default:
 			print_usage(prgname);
 			return -1;
@@ -705,6 +776,7 @@ main(int argc, char *argv[])
 	unsigned int lcore_id, worker_id = 0;
 	int distr_core_id = -1, rx_core_id = -1, tx_core_id = -1;
 	unsigned nb_ports;
+	unsigned int min_cores;
 	uint16_t portid;
 	uint16_t nb_ports_available;
 	uint64_t t, freq;
@@ -724,12 +796,21 @@ main(int argc, char *argv[])
 	if (ret < 0)
 		rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
-	if (rte_lcore_count() < 5)
+	if (enable_lcore_rx_distributor) {
+	/* RX and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
+		min_cores = 4;
+		num_workers = rte_lcore_count() - 3;
+	} else {
+	/* separate RX and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
+		min_cores = 5;
+		num_workers = rte_lcore_count() - 4;
+	}
+
+	if (rte_lcore_count() < min_cores)
 		rte_exit(EXIT_FAILURE, "Error, This application needs at "
-				"least 5 logical cores to run:\n"
+				"least 4 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 or 2 lcore for packet RX and distribution\n"
 				"1 lcore for packet TX\n"
 				"and at least 1 lcore for worker threads\n");
 
@@ -772,7 +853,7 @@ main(int argc, char *argv[])
 	}
 
 	d = rte_distributor_create("PKT_DIST", rte_socket_id(),
-			rte_lcore_count() - 4,
+			num_workers,
 			RTE_DIST_ALG_BURST);
 	if (d == NULL)
 		rte_exit(EXIT_FAILURE, "Cannot create distributor\n");
@@ -808,7 +889,7 @@ main(int argc, char *argv[])
 			if (lcore_cap.priority != 1)
 				continue;
 
-			if (distr_core_id < 0) {
+			if (distr_core_id < 0 && !enable_lcore_rx_distributor) {
 				distr_core_id = lcore_id;
 				printf("Distributor on priority core %d\n",
 					lcore_id);
@@ -839,7 +920,7 @@ main(int argc, char *argv[])
 				lcore_id == (unsigned int)rx_core_id ||
 				lcore_id == (unsigned int)tx_core_id)
 			continue;
-		if (distr_core_id < 0) {
+		if (distr_core_id < 0 && !enable_lcore_rx_distributor) {
 			distr_core_id = lcore_id;
 			printf("Distributor on core %d\n", lcore_id);
 			continue;
@@ -856,7 +937,12 @@ main(int argc, char *argv[])
 		}
 	}
 
-	printf(" tx id %d, dist id %d, rx id %d\n",
+	if (enable_lcore_rx_distributor)
+		printf(" tx id %d, rx id %d\n",
+			tx_core_id,
+			rx_core_id);
+	else
+		printf(" tx id %d, dist id %d, rx id %d\n",
 			tx_core_id,
 			distr_core_id,
 			rx_core_id);
@@ -888,15 +974,16 @@ main(int argc, char *argv[])
 			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);
+	struct lcore_params *pd = NULL;
+	if (!enable_lcore_rx_distributor) {
+		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 =
@@ -905,12 +992,16 @@ main(int argc, char *argv[])
 		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);
+	if (enable_lcore_rx_distributor)
+		rte_eal_remote_launch((lcore_function_t *)lcore_rx_and_distributor,
+				pr, rx_core_id);
+	else
+		rte_eal_remote_launch((lcore_function_t *)lcore_rx,
+				pr, rx_core_id);
 
 	freq = rte_get_timer_hz();
 	t = rte_rdtsc() + freq;
-	while (!quit_signal_dist) {
+	while (!quit_signal) {
 		if (t < rte_rdtsc()) {
 			print_stats();
 			t = rte_rdtsc() + freq;
@@ -925,7 +1016,8 @@ main(int argc, char *argv[])
 
 	print_stats();
 
-	rte_free(pd);
+	if (pd)
+		rte_free(pd);
 	rte_free(pr);
 
 	/* clean up the EAL */
-- 
2.27.0

Re: [PATCH v2] examples/distributor: update dynamic configuration

[Hunt, David]

Hi Ömer,

On 28/06/2022 20:54, omer.yamac at ceng.metu.edu.tr (Abdullah Ömer 
Yamaç) wrote:
> In this patch,
>      * It is possible to switch the running mode of the distributor
> using the command line argument.
>      * With "-c" parameter, you can run RX and Distributor
> on the same core.
>      * Without "-c" parameter, you can run RX and Distributor
> on the different core.
>      * Syntax error of the single RX and distributor core is fixed.


I believe this particular fix is already merged and back-ported to 
stable. No need to include this line in the commit message.


>      * Consecutive termination of the lcores fixed.
> The termination order was wrong, and you couldn't terminate the
> application while traffic was capturing. The current order is
> RX -> Distributor -> TX -> Workers
>      * When "-c" parameter is active, the wasted distributor core is
> also deactivated in the main function.
>
> Fixes: 4a7f40c0ff9a ("examples/distributor: add dedicated core")
> Cc: stable at dpdk.org


This is a feature change, not a fix, so I don't believe you need the 
"Fixes" line or the "Cc: stable" line.


>
> Signed-off-by: Abdullah ?mer Yama? <omer.yamac at ceng.metu.edu.tr>


I've tested this with the "-c" option, works well. Traffic coming into 
the app is distributed among the core. With -c added to the command line 
parameters, I have an extra worker core, as expected. Looks good to me.

With the above suggested changes to the commit message:

Reviewed-by: David Hunt <david.hunt@intel.com>

Re: [PATCH v2] examples/distributor: update dynamic configuration

[Omer Yamac]

Hi David,

I applied the changes as new version (v3), Thank you

On 01.09.2022 13:58, Hunt, David wrote:
> Hi Ömer,
> 
> On 28/06/2022 20:54, omer.yamac at ceng.metu.edu.tr (Abdullah Ömer 
> Yamaç) wrote:
>> In this patch,
>>      * It is possible to switch the running mode of the distributor
>> using the command line argument.
>>      * With "-c" parameter, you can run RX and Distributor
>> on the same core.
>>      * Without "-c" parameter, you can run RX and Distributor
>> on the different core.
>>      * Syntax error of the single RX and distributor core is fixed.
> 
> 
> I believe this particular fix is already merged and back-ported to 
> stable. No need to include this line in the commit message.
> 
> 
>>      * Consecutive termination of the lcores fixed.
>> The termination order was wrong, and you couldn't terminate the
>> application while traffic was capturing. The current order is
>> RX -> Distributor -> TX -> Workers
>>      * When "-c" parameter is active, the wasted distributor core is
>> also deactivated in the main function.
>> 
>> Fixes: 4a7f40c0ff9a ("examples/distributor: add dedicated core")
>> Cc: stable at dpdk.org
> 
> 
> This is a feature change, not a fix, so I don't believe you need the 
> "Fixes" line or the "Cc: stable" line.
> 
> 
>> 
>> Signed-off-by: Abdullah ?mer Yama? <omer.yamac at ceng.metu.edu.tr>
> 
> 
> I've tested this with the "-c" option, works well. Traffic coming into 
> the app is distributed among the core. With -c added to the command 
> line parameters, I have an extra worker core, as expected. Looks good 
> to me.
> 
> With the above suggested changes to the commit message:
> 
> Reviewed-by: David Hunt <david.hunt@intel.com>

Re: [PATCH v2] examples/distributor: update dynamic configuration

[Hunt, David]

On 02/09/2022 06:20, Omer Yamac wrote:
> Hi David,
>
> I applied the changes as new version (v3), Thank you
>
--snip--


>> With the above suggested changes to the commit message:
>>
>> Reviewed-by: David Hunt <david.hunt@intel.com>


Hi Omer,

2 things:

Usually you submit subsequent versions of patches "in-reply-to" the 
previous versions to keep them together. Please see other examples of 
this on the mailing list.

It is normal to include any Reviewed-by or Acked-by tags in subsequent 
versions if the patch has not changed significantly.

Rgds,

Dave.

[PATCH v3] examples/distributor: update dynamic configuration

[Abdullah Ömer Yamaç]

In this patch,
    * It is possible to switch the running mode of the distributor
using the command line argument.
    * With "-c" parameter, you can run RX and Distributor
on the same core.
    * Without "-c" parameter, you can run RX and Distributor
on the different core.
    * Consecutive termination of the lcores fixed.
The termination order was wrong, and you couldn't terminate the
application while traffic was capturing. The current order is
RX -> Distributor -> TX -> Workers
    * When "-c" parameter is active, the wasted distributor core is
also deactivated in the main function.

Signed-off-by: Abdullah Ömer Yamaç <omer.yamac@ceng.metu.edu.tr>

---
Cc: david.hunt@intel.com
---
 doc/guides/sample_app_ug/dist_app.rst |   3 +-
 examples/distributor/main.c           | 222 ++++++++++++++++++--------
 2 files changed, 159 insertions(+), 66 deletions(-)

diff --git a/doc/guides/sample_app_ug/dist_app.rst b/doc/guides/sample_app_ug/dist_app.rst
index 3bd03905c3..5c80561187 100644
--- a/doc/guides/sample_app_ug/dist_app.rst
+++ b/doc/guides/sample_app_ug/dist_app.rst
@@ -42,11 +42,12 @@ Running the Application
 
    ..  code-block:: console
 
-       ./<build-dir>/examples/dpdk-distributor [EAL options] -- -p PORTMASK
+       ./<build-dir>/examples/dpdk-distributor [EAL options] -- -p PORTMASK [-c]
 
    where,
 
    *   -p PORTMASK: Hexadecimal bitmask of ports to configure
+   *   -c: Combines the RX core with distribution core
 
 #. To run the application in linux environment with 10 lcores, 4 ports,
    issue the command:
diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 8995806b4e..fc9a75f695 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -39,6 +39,8 @@ volatile uint8_t quit_signal_rx;
 volatile uint8_t quit_signal_dist;
 volatile uint8_t quit_signal_work;
 unsigned int power_lib_initialised;
+bool enable_lcore_rx_distributor;
+unsigned int num_workers;
 
 static volatile struct app_stats {
 	struct {
@@ -256,14 +258,82 @@ lcore_rx(struct lcore_params *p)
 		}
 		app_stats.rx.rx_pkts += nb_rx;
 
-/*
- * 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(p->d, bufs, nb_rx);
-		const uint16_t nb_ret = rte_distributor_returned_pkts(p->d,
-			bufs, BURST_SIZE*2);
+		/*
+		 * 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_rx, NULL);
+
+		app_stats.rx.enqueued_pkts += sent;
+		if (unlikely(sent < nb_rx)) {
+			app_stats.rx.enqdrop_pkts +=  nb_rx - sent;
+			RTE_LOG_DP(DEBUG, DISTRAPP,
+				"%s:Packet loss due to full ring\n", __func__);
+			while (sent < nb_rx)
+				rte_pktmbuf_free(bufs[sent++]);
+		}
+		if (++port == nb_ports)
+			port = 0;
+	}
+	if (power_lib_initialised)
+		rte_power_exit(rte_lcore_id());
+	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
+	/* set distributor threads quit flag */
+	quit_signal_dist = 1;
+	return 0;
+}
+
+static int
+lcore_rx_and_distributor(struct lcore_params *p)
+{
+	struct rte_distributor *d = p->d;
+	const uint16_t nb_ports = rte_eth_dev_count_avail();
+	const int socket_id = rte_socket_id();
+	uint16_t port;
+	struct rte_mbuf *bufs[BURST_SIZE*2];
+
+	RTE_ETH_FOREACH_DEV(port) {
+		/* skip ports that are not enabled */
+		if ((enabled_port_mask & (1 << port)) == 0)
+			continue;
+
+		if (rte_eth_dev_socket_id(port) > 0 &&
+				rte_eth_dev_socket_id(port) != socket_id)
+			printf("WARNING, port %u is on remote NUMA node to "
+					"RX thread.\n\tPerformance will not "
+					"be optimal.\n", port);
+	}
+
+	printf("\nCore %u doing packet RX and Distributor.\n", rte_lcore_id());
+	port = 0;
+	while (!quit_signal_rx) {
+
+		/* skip ports that are not enabled */
+		if ((enabled_port_mask & (1 << port)) == 0) {
+			if (++port == nb_ports)
+				port = 0;
+			continue;
+		}
+		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;
+
+		/*
+		 * Run the distributor on the rx core. Returned
+		 * packets are then send straight to the tx core.
+		 */
+		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)) {
@@ -275,18 +345,6 @@ 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, 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)) {
@@ -301,9 +359,14 @@ lcore_rx(struct lcore_params *p)
 	}
 	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());
+	/* set tx threads quit flag */
 	quit_signal = 1;
+	/* set worker threads quit flag */
+	quit_signal_work = 1;
+	rte_distributor_flush(d);
+	/* Unblock any returns so workers can exit */
+	rte_distributor_clear_returns(d);
 	return 0;
 }
 
@@ -381,14 +444,16 @@ 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());
+	printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
+	/* set tx threads quit flag */
+	quit_signal = 1;
+	/* set worker threads quit flag */
+	quit_signal_work = 1;
 	rte_distributor_flush(d);
 	/* Unblock any returns so workers can exit */
 	rte_distributor_clear_returns(d);
-	quit_signal_rx = 1;
 	return 0;
 }
 
@@ -467,7 +532,7 @@ int_handler(int sig_num)
 {
 	printf("Exiting on signal %d\n", sig_num);
 	/* set quit flag for rx thread to exit */
-	quit_signal_dist = 1;
+	quit_signal_rx = 1;
 }
 
 static void
@@ -475,7 +540,6 @@ print_stats(void)
 {
 	struct rte_eth_stats eth_stats;
 	unsigned int i, j;
-	const unsigned int num_workers = rte_lcore_count() - 4;
 
 	RTE_ETH_FOREACH_DEV(i) {
 		rte_eth_stats_get(i, &eth_stats);
@@ -504,20 +568,22 @@ print_stats(void)
 			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);
+	if (!enable_lcore_rx_distributor) {
+		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",
@@ -629,8 +695,9 @@ init_power_library(void)
 static void
 print_usage(const char *prgname)
 {
-	printf("%s [EAL options] -- -p PORTMASK\n"
-			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n",
+	printf("%s [EAL options] -- -p PORTMASK [-c]\n"
+			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n"
+			"  -c: Combines the RX core with the distribution core\n",
 			prgname);
 }
 
@@ -661,8 +728,8 @@ parse_args(int argc, char **argv)
 	};
 
 	argvopt = argv;
-
-	while ((opt = getopt_long(argc, argvopt, "p:",
+	enable_lcore_rx_distributor = false;
+	while ((opt = getopt_long(argc, argvopt, "cp:",
 			lgopts, &option_index)) != EOF) {
 
 		switch (opt) {
@@ -676,6 +743,10 @@ parse_args(int argc, char **argv)
 			}
 			break;
 
+		case 'c':
+			enable_lcore_rx_distributor = true;
+			break;
+
 		default:
 			print_usage(prgname);
 			return -1;
@@ -705,6 +776,7 @@ main(int argc, char *argv[])
 	unsigned int lcore_id, worker_id = 0;
 	int distr_core_id = -1, rx_core_id = -1, tx_core_id = -1;
 	unsigned nb_ports;
+	unsigned int min_cores;
 	uint16_t portid;
 	uint16_t nb_ports_available;
 	uint64_t t, freq;
@@ -724,12 +796,21 @@ main(int argc, char *argv[])
 	if (ret < 0)
 		rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
-	if (rte_lcore_count() < 5)
+	if (enable_lcore_rx_distributor) {
+	/* RX and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
+		min_cores = 4;
+		num_workers = rte_lcore_count() - 3;
+	} else {
+	/* separate RX and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
+		min_cores = 5;
+		num_workers = rte_lcore_count() - 4;
+	}
+
+	if (rte_lcore_count() < min_cores)
 		rte_exit(EXIT_FAILURE, "Error, This application needs at "
-				"least 5 logical cores to run:\n"
+				"least 4 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 or 2 lcore for packet RX and distribution\n"
 				"1 lcore for packet TX\n"
 				"and at least 1 lcore for worker threads\n");
 
@@ -772,7 +853,7 @@ main(int argc, char *argv[])
 	}
 
 	d = rte_distributor_create("PKT_DIST", rte_socket_id(),
-			rte_lcore_count() - 4,
+			num_workers,
 			RTE_DIST_ALG_BURST);
 	if (d == NULL)
 		rte_exit(EXIT_FAILURE, "Cannot create distributor\n");
@@ -808,7 +889,7 @@ main(int argc, char *argv[])
 			if (lcore_cap.priority != 1)
 				continue;
 
-			if (distr_core_id < 0) {
+			if (distr_core_id < 0 && !enable_lcore_rx_distributor) {
 				distr_core_id = lcore_id;
 				printf("Distributor on priority core %d\n",
 					lcore_id);
@@ -839,7 +920,7 @@ main(int argc, char *argv[])
 				lcore_id == (unsigned int)rx_core_id ||
 				lcore_id == (unsigned int)tx_core_id)
 			continue;
-		if (distr_core_id < 0) {
+		if (distr_core_id < 0 && !enable_lcore_rx_distributor) {
 			distr_core_id = lcore_id;
 			printf("Distributor on core %d\n", lcore_id);
 			continue;
@@ -856,7 +937,12 @@ main(int argc, char *argv[])
 		}
 	}
 
-	printf(" tx id %d, dist id %d, rx id %d\n",
+	if (enable_lcore_rx_distributor)
+		printf(" tx id %d, rx id %d\n",
+			tx_core_id,
+			rx_core_id);
+	else
+		printf(" tx id %d, dist id %d, rx id %d\n",
 			tx_core_id,
 			distr_core_id,
 			rx_core_id);
@@ -888,15 +974,16 @@ main(int argc, char *argv[])
 			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);
+	struct lcore_params *pd = NULL;
+	if (!enable_lcore_rx_distributor) {
+		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 =
@@ -905,12 +992,16 @@ main(int argc, char *argv[])
 		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);
+	if (enable_lcore_rx_distributor)
+		rte_eal_remote_launch((lcore_function_t *)lcore_rx_and_distributor,
+				pr, rx_core_id);
+	else
+		rte_eal_remote_launch((lcore_function_t *)lcore_rx,
+				pr, rx_core_id);
 
 	freq = rte_get_timer_hz();
 	t = rte_rdtsc() + freq;
-	while (!quit_signal_dist) {
+	while (!quit_signal) {
 		if (t < rte_rdtsc()) {
 			print_stats();
 			t = rte_rdtsc() + freq;
@@ -925,7 +1016,8 @@ main(int argc, char *argv[])
 
 	print_stats();
 
-	rte_free(pd);
+	if (pd)
+		rte_free(pd);
 	rte_free(pr);
 
 	/* clean up the EAL */
-- 
2.27.0

Re: [PATCH v3] examples/distributor: update dynamic configuration

[Hunt, David]

On 01/09/2022 15:09, omer.yamac at ceng.metu.edu.tr (Abdullah Ömer 
Yamaç) wrote:
> In this patch,
>      * It is possible to switch the running mode of the distributor
> using the command line argument.
>      * With "-c" parameter, you can run RX and Distributor
> on the same core.
>      * Without "-c" parameter, you can run RX and Distributor
> on the different core.
>      * Consecutive termination of the lcores fixed.
> The termination order was wrong, and you couldn't terminate the
> application while traffic was capturing. The current order is
> RX -> Distributor -> TX -> Workers
>      * When "-c" parameter is active, the wasted distributor core is
> also deactivated in the main function.
>
> Signed-off-by: Abdullah ?mer Yama? <omer.yamac at ceng.metu.edu.tr>
>
> ---
> Cc: david.hunt at intel.com
> ---

LGTM

Reviewed-by: David Hunt <david.hunt@intel.com>

Re: [PATCH v3] examples/distributor: update dynamic configuration

[Thomas Monjalon]

Hello,

Not a complete review, but few general comments to improve the patch below:

01/09/2022 16:09, Abdullah Ömer Yamaç:
> In this patch,
>     * It is possible to switch the running mode of the distributor
> using the command line argument.
>     * With "-c" parameter, you can run RX and Distributor
> on the same core.
>     * Without "-c" parameter, you can run RX and Distributor
> on the different core.
>     * Consecutive termination of the lcores fixed.
> The termination order was wrong, and you couldn't terminate the
> application while traffic was capturing. The current order is
> RX -> Distributor -> TX -> Workers
>     * When "-c" parameter is active, the wasted distributor core is
> also deactivated in the main function.

Please could you make clear what was the issue,
and what was changed in the commit message?

> -#if 0

It's good to remove such thing.
Dead code should not exist.

> +		/*
> +		 * Swap the following two lines if you want the rx traffic
> +		 * to go directly to tx, no distribution.
> +		 */

In DPDK, it is preferred to use uppercase Rx and Tx.

> +		struct rte_ring *out_ring = p->rx_dist_ring;
> +		/* struct rte_ring *out_ring = p->dist_tx_ring; */

This line is dead code, please remove.

> +		if (!pd)

It is preferred to not use boolean operator with pointer.
Explicit comparison is encouraged: pd == NULL

> -	rte_free(pd);
> +	if (pd)
> +		rte_free(pd);

This check is useless because redundant with rte_free behaviour.

[PATCH v4] examples/distributor: remove dead code and renaming Rx,Tx

[Abdullah Ömer Yamaç]

One line of commented code was dead code, this line and releated
comments are removed. Naming of rx,RX and tx,TX are replaced by Rx and
Tx.

Signed-off-by: Abdullah Ömer Yamaç <omer.yamac@ceng.metu.edu.tr>
---
CC: Thomas Monjalon <thomas@monjalon.net>
---
 examples/distributor/main.c | 51 +++++++++++++++++--------------------
 1 file changed, 23 insertions(+), 28 deletions(-)

diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 21304d6618..d54e241110 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -104,7 +104,7 @@ struct output_buffer {
 static void print_stats(void);
 
 /*
- * Initialises a given port using global settings and with the rx buffers
+ * Initialises a given port using global settings and with the Rx buffers
  * coming from the mbuf_pool passed as parameter
  */
 static inline int
@@ -259,12 +259,7 @@ lcore_rx(struct lcore_params *p)
 		}
 		app_stats.rx.rx_pkts += 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_rx, NULL);
@@ -282,7 +277,7 @@ lcore_rx(struct lcore_params *p)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
+	printf("\nCore %u exiting Rx task.\n", rte_lcore_id());
 	/* set distributor threads quit flag */
 	quit_signal_dist = 1;
 	return 0;
@@ -305,11 +300,11 @@ lcore_rx_and_distributor(struct lcore_params *p)
 		if (rte_eth_dev_socket_id(port) > 0 &&
 				rte_eth_dev_socket_id(port) != socket_id)
 			printf("WARNING, port %u is on remote NUMA node to "
-					"RX thread.\n\tPerformance will not "
+					"Rx thread.\n\tPerformance will not "
 					"be optimal.\n", port);
 	}
 
-	printf("\nCore %u doing packet RX and Distributor.\n", rte_lcore_id());
+	printf("\nCore %u doing packet Rx and Distributor.\n", rte_lcore_id());
 	port = 0;
 	while (!quit_signal_rx) {
 
@@ -329,8 +324,8 @@ lcore_rx_and_distributor(struct lcore_params *p)
 		app_stats.rx.rx_pkts += nb_rx;
 
 		/*
-		 * Run the distributor on the rx core. Returned
-		 * packets are then send straight to the tx core.
+		 * Run the distributor on the Rx core. Returned
+		 * packets are then send straight to the Tx core.
 		 */
 		rte_distributor_process(d, bufs, nb_rx);
 		const uint16_t nb_ret = rte_distributor_returned_pkts(d,
@@ -360,8 +355,8 @@ lcore_rx_and_distributor(struct lcore_params *p)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
-	/* set tx threads quit flag */
+	printf("\nCore %u exiting Rx task.\n", rte_lcore_id());
+	/* set Tx threads quit flag */
 	quit_signal = 1;
 	/* set worker threads quit flag */
 	quit_signal_work = 1;
@@ -448,7 +443,7 @@ lcore_distributor(struct lcore_params *p)
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
 	printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
-	/* set tx threads quit flag */
+	/* set Tx threads quit flag */
 	quit_signal = 1;
 	/* set worker threads quit flag */
 	quit_signal_work = 1;
@@ -524,7 +519,7 @@ lcore_tx(struct rte_ring *in_r)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting tx task.\n", rte_lcore_id());
+	printf("\nCore %u exiting Tx task.\n", rte_lcore_id());
 	return 0;
 }
 
@@ -532,7 +527,7 @@ static void
 int_handler(int sig_num)
 {
 	printf("Exiting on signal %d\n", sig_num);
-	/* set quit flag for rx thread to exit */
+	/* set quit flag for Rx thread to exit */
 	quit_signal_rx = 1;
 }
 
@@ -698,7 +693,7 @@ print_usage(const char *prgname)
 {
 	printf("%s [EAL options] -- -p PORTMASK [-c]\n"
 			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n"
-			"  -c: Combines the RX core with the distribution core\n",
+			"  -c: Combines the Rx core with the distribution core\n",
 			prgname);
 }
 
@@ -798,11 +793,11 @@ main(int argc, char *argv[])
 		rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
 	if (enable_lcore_rx_distributor) {
-	/* RX and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
+	/* Rx and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
 		min_cores = 4;
 		num_workers = rte_lcore_count() - 3;
 	} else {
-	/* separate RX and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
+	/* separate Rx and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
 		min_cores = 5;
 		num_workers = rte_lcore_count() - 4;
 	}
@@ -811,8 +806,8 @@ main(int argc, char *argv[])
 		rte_exit(EXIT_FAILURE, "Error, This application needs at "
 				"least 4 logical cores to run:\n"
 				"1 lcore for stats (can be core 0)\n"
-				"1 or 2 lcore for packet RX and distribution\n"
-				"1 lcore for packet TX\n"
+				"1 or 2 lcore for packet Rx and distribution\n"
+				"1 lcore for packet Tx\n"
 				"and at least 1 lcore for worker threads\n");
 
 	if (init_power_library() == 0)
@@ -875,7 +870,7 @@ main(int argc, char *argv[])
 
 	if (power_lib_initialised) {
 		/*
-		 * Here we'll pre-assign lcore ids to the rx, tx and
+		 * 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
@@ -939,18 +934,18 @@ main(int argc, char *argv[])
 	}
 
 	if (enable_lcore_rx_distributor)
-		printf(" tx id %d, rx id %d\n",
+		printf(" Tx id %d, Rx id %d\n",
 			tx_core_id,
 			rx_core_id);
 	else
-		printf(" tx id %d, dist id %d, rx id %d\n",
+		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.
+	 * lcore_ids for Tx, Rx and distributor workloads.
 	 */
 	RTE_LCORE_FOREACH_WORKER(lcore_id) {
 		if (lcore_id == (unsigned int)distr_core_id ||
@@ -970,7 +965,7 @@ main(int argc, char *argv[])
 				p, lcore_id);
 	}
 
-	/* Start tx core */
+	/* Start Tx core */
 	rte_eal_remote_launch((lcore_function_t *)lcore_tx,
 			dist_tx_ring, tx_core_id);
 
@@ -978,7 +973,7 @@ main(int argc, char *argv[])
 	struct lcore_params *pd = NULL;
 	if (!enable_lcore_rx_distributor) {
 		pd = rte_malloc(NULL, sizeof(*pd), 0);
-		if (!pd)
+		if (pd == NULL)
 			rte_panic("malloc failure\n");
 		*pd = (struct lcore_params){worker_id++, d,
 			rx_dist_ring, dist_tx_ring, mbuf_pool};
@@ -986,7 +981,7 @@ main(int argc, char *argv[])
 				pd, distr_core_id);
 	}
 
-	/* Start rx core */
+	/* Start Rx core */
 	struct lcore_params *pr =
 		rte_malloc(NULL, sizeof(*pr), 0);
 	if (!pr)
-- 
2.27.0

[PATCH v5] examples/distributor: remove dead code and renaming Rx,Tx

[Abdullah Ömer Yamaç]

One line of commented code was dead code, this line and related
comments are removed. Naming of rx,RX and tx,TX are replaced by Rx and
Tx.

Signed-off-by: Abdullah Ömer Yamaç <omer.yamac@ceng.metu.edu.tr>
---
CC: Thomas Monjalon <thomas@monjalon.net>
---
 examples/distributor/main.c | 51 +++++++++++++++++--------------------
 1 file changed, 23 insertions(+), 28 deletions(-)

diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 21304d6618..d54e241110 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -104,7 +104,7 @@ struct output_buffer {
 static void print_stats(void);
 
 /*
- * Initialises a given port using global settings and with the rx buffers
+ * Initialises a given port using global settings and with the Rx buffers
  * coming from the mbuf_pool passed as parameter
  */
 static inline int
@@ -259,12 +259,7 @@ lcore_rx(struct lcore_params *p)
 		}
 		app_stats.rx.rx_pkts += 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_rx, NULL);
@@ -282,7 +277,7 @@ lcore_rx(struct lcore_params *p)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
+	printf("\nCore %u exiting Rx task.\n", rte_lcore_id());
 	/* set distributor threads quit flag */
 	quit_signal_dist = 1;
 	return 0;
@@ -305,11 +300,11 @@ lcore_rx_and_distributor(struct lcore_params *p)
 		if (rte_eth_dev_socket_id(port) > 0 &&
 				rte_eth_dev_socket_id(port) != socket_id)
 			printf("WARNING, port %u is on remote NUMA node to "
-					"RX thread.\n\tPerformance will not "
+					"Rx thread.\n\tPerformance will not "
 					"be optimal.\n", port);
 	}
 
-	printf("\nCore %u doing packet RX and Distributor.\n", rte_lcore_id());
+	printf("\nCore %u doing packet Rx and Distributor.\n", rte_lcore_id());
 	port = 0;
 	while (!quit_signal_rx) {
 
@@ -329,8 +324,8 @@ lcore_rx_and_distributor(struct lcore_params *p)
 		app_stats.rx.rx_pkts += nb_rx;
 
 		/*
-		 * Run the distributor on the rx core. Returned
-		 * packets are then send straight to the tx core.
+		 * Run the distributor on the Rx core. Returned
+		 * packets are then send straight to the Tx core.
 		 */
 		rte_distributor_process(d, bufs, nb_rx);
 		const uint16_t nb_ret = rte_distributor_returned_pkts(d,
@@ -360,8 +355,8 @@ lcore_rx_and_distributor(struct lcore_params *p)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
-	/* set tx threads quit flag */
+	printf("\nCore %u exiting Rx task.\n", rte_lcore_id());
+	/* set Tx threads quit flag */
 	quit_signal = 1;
 	/* set worker threads quit flag */
 	quit_signal_work = 1;
@@ -448,7 +443,7 @@ lcore_distributor(struct lcore_params *p)
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
 	printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
-	/* set tx threads quit flag */
+	/* set Tx threads quit flag */
 	quit_signal = 1;
 	/* set worker threads quit flag */
 	quit_signal_work = 1;
@@ -524,7 +519,7 @@ lcore_tx(struct rte_ring *in_r)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting tx task.\n", rte_lcore_id());
+	printf("\nCore %u exiting Tx task.\n", rte_lcore_id());
 	return 0;
 }
 
@@ -532,7 +527,7 @@ static void
 int_handler(int sig_num)
 {
 	printf("Exiting on signal %d\n", sig_num);
-	/* set quit flag for rx thread to exit */
+	/* set quit flag for Rx thread to exit */
 	quit_signal_rx = 1;
 }
 
@@ -698,7 +693,7 @@ print_usage(const char *prgname)
 {
 	printf("%s [EAL options] -- -p PORTMASK [-c]\n"
 			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n"
-			"  -c: Combines the RX core with the distribution core\n",
+			"  -c: Combines the Rx core with the distribution core\n",
 			prgname);
 }
 
@@ -798,11 +793,11 @@ main(int argc, char *argv[])
 		rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
 	if (enable_lcore_rx_distributor) {
-	/* RX and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
+	/* Rx and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
 		min_cores = 4;
 		num_workers = rte_lcore_count() - 3;
 	} else {
-	/* separate RX and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
+	/* separate Rx and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
 		min_cores = 5;
 		num_workers = rte_lcore_count() - 4;
 	}
@@ -811,8 +806,8 @@ main(int argc, char *argv[])
 		rte_exit(EXIT_FAILURE, "Error, This application needs at "
 				"least 4 logical cores to run:\n"
 				"1 lcore for stats (can be core 0)\n"
-				"1 or 2 lcore for packet RX and distribution\n"
-				"1 lcore for packet TX\n"
+				"1 or 2 lcore for packet Rx and distribution\n"
+				"1 lcore for packet Tx\n"
 				"and at least 1 lcore for worker threads\n");
 
 	if (init_power_library() == 0)
@@ -875,7 +870,7 @@ main(int argc, char *argv[])
 
 	if (power_lib_initialised) {
 		/*
-		 * Here we'll pre-assign lcore ids to the rx, tx and
+		 * 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
@@ -939,18 +934,18 @@ main(int argc, char *argv[])
 	}
 
 	if (enable_lcore_rx_distributor)
-		printf(" tx id %d, rx id %d\n",
+		printf(" Tx id %d, Rx id %d\n",
 			tx_core_id,
 			rx_core_id);
 	else
-		printf(" tx id %d, dist id %d, rx id %d\n",
+		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.
+	 * lcore_ids for Tx, Rx and distributor workloads.
 	 */
 	RTE_LCORE_FOREACH_WORKER(lcore_id) {
 		if (lcore_id == (unsigned int)distr_core_id ||
@@ -970,7 +965,7 @@ main(int argc, char *argv[])
 				p, lcore_id);
 	}
 
-	/* Start tx core */
+	/* Start Tx core */
 	rte_eal_remote_launch((lcore_function_t *)lcore_tx,
 			dist_tx_ring, tx_core_id);
 
@@ -978,7 +973,7 @@ main(int argc, char *argv[])
 	struct lcore_params *pd = NULL;
 	if (!enable_lcore_rx_distributor) {
 		pd = rte_malloc(NULL, sizeof(*pd), 0);
-		if (!pd)
+		if (pd == NULL)
 			rte_panic("malloc failure\n");
 		*pd = (struct lcore_params){worker_id++, d,
 			rx_dist_ring, dist_tx_ring, mbuf_pool};
@@ -986,7 +981,7 @@ main(int argc, char *argv[])
 				pd, distr_core_id);
 	}
 
-	/* Start rx core */
+	/* Start Rx core */
 	struct lcore_params *pr =
 		rte_malloc(NULL, sizeof(*pr), 0);
 	if (!pr)
-- 
2.27.0

[PATCH v6] examples/distributor: remove dead code and renaming Rx,Tx

[Abdullah Ömer Yamaç]

One line of commented code was dead code, this line and related
comments are removed. Naming of rx,RX and tx,TX are replaced by Rx and
Tx.

Signed-off-by: Abdullah Ömer Yamaç <omer.yamac@ceng.metu.edu.tr>
---
CC: Thomas Monjalon <thomas@monjalon.net>
---
 examples/distributor/main.c | 53 +++++++++++++++++--------------------
 1 file changed, 24 insertions(+), 29 deletions(-)

diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 21304d6618..e12cd5760c 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -104,7 +104,7 @@ struct output_buffer {
 static void print_stats(void);
 
 /*
- * Initialises a given port using global settings and with the rx buffers
+ * Initialises a given port using global settings and with the Rx buffers
  * coming from the mbuf_pool passed as parameter
  */
 static inline int
@@ -259,12 +259,7 @@ lcore_rx(struct lcore_params *p)
 		}
 		app_stats.rx.rx_pkts += 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_rx, NULL);
@@ -282,7 +277,7 @@ lcore_rx(struct lcore_params *p)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
+	printf("\nCore %u exiting Rx task.\n", rte_lcore_id());
 	/* set distributor threads quit flag */
 	quit_signal_dist = 1;
 	return 0;
@@ -305,11 +300,11 @@ lcore_rx_and_distributor(struct lcore_params *p)
 		if (rte_eth_dev_socket_id(port) > 0 &&
 				rte_eth_dev_socket_id(port) != socket_id)
 			printf("WARNING, port %u is on remote NUMA node to "
-					"RX thread.\n\tPerformance will not "
+					"Rx thread.\n\tPerformance will not "
 					"be optimal.\n", port);
 	}
 
-	printf("\nCore %u doing packet RX and Distributor.\n", rte_lcore_id());
+	printf("\nCore %u doing packet Rx and Distributor.\n", rte_lcore_id());
 	port = 0;
 	while (!quit_signal_rx) {
 
@@ -329,8 +324,8 @@ lcore_rx_and_distributor(struct lcore_params *p)
 		app_stats.rx.rx_pkts += nb_rx;
 
 		/*
-		 * Run the distributor on the rx core. Returned
-		 * packets are then send straight to the tx core.
+		 * Run the distributor on the Rx core. Returned
+		 * packets are then send straight to the Tx core.
 		 */
 		rte_distributor_process(d, bufs, nb_rx);
 		const uint16_t nb_ret = rte_distributor_returned_pkts(d,
@@ -360,8 +355,8 @@ lcore_rx_and_distributor(struct lcore_params *p)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
-	/* set tx threads quit flag */
+	printf("\nCore %u exiting Rx task.\n", rte_lcore_id());
+	/* set Tx threads quit flag */
 	quit_signal = 1;
 	/* set worker threads quit flag */
 	quit_signal_work = 1;
@@ -448,7 +443,7 @@ lcore_distributor(struct lcore_params *p)
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
 	printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
-	/* set tx threads quit flag */
+	/* set Tx threads quit flag */
 	quit_signal = 1;
 	/* set worker threads quit flag */
 	quit_signal_work = 1;
@@ -524,7 +519,7 @@ lcore_tx(struct rte_ring *in_r)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting tx task.\n", rte_lcore_id());
+	printf("\nCore %u exiting Tx task.\n", rte_lcore_id());
 	return 0;
 }
 
@@ -532,7 +527,7 @@ static void
 int_handler(int sig_num)
 {
 	printf("Exiting on signal %d\n", sig_num);
-	/* set quit flag for rx thread to exit */
+	/* set quit flag for Rx thread to exit */
 	quit_signal_rx = 1;
 }
 
@@ -698,7 +693,7 @@ print_usage(const char *prgname)
 {
 	printf("%s [EAL options] -- -p PORTMASK [-c]\n"
 			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n"
-			"  -c: Combines the RX core with the distribution core\n",
+			"  -c: Combines the Rx core with the distribution core\n",
 			prgname);
 }
 
@@ -798,11 +793,11 @@ main(int argc, char *argv[])
 		rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
 	if (enable_lcore_rx_distributor) {
-	/* RX and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
+	/* Rx and distributor combined, 3 fixed function cores (stat, Tx, at least 1 worker) */
 		min_cores = 4;
 		num_workers = rte_lcore_count() - 3;
 	} else {
-	/* separate RX and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
+	/* separate Rx and distributor, 3 fixed function cores (stat, Tx, at least 1 worker) */
 		min_cores = 5;
 		num_workers = rte_lcore_count() - 4;
 	}
@@ -811,8 +806,8 @@ main(int argc, char *argv[])
 		rte_exit(EXIT_FAILURE, "Error, This application needs at "
 				"least 4 logical cores to run:\n"
 				"1 lcore for stats (can be core 0)\n"
-				"1 or 2 lcore for packet RX and distribution\n"
-				"1 lcore for packet TX\n"
+				"1 or 2 lcore for packet Rx and distribution\n"
+				"1 lcore for packet Tx\n"
 				"and at least 1 lcore for worker threads\n");
 
 	if (init_power_library() == 0)
@@ -875,13 +870,13 @@ main(int argc, char *argv[])
 
 	if (power_lib_initialised) {
 		/*
-		 * Here we'll pre-assign lcore ids to the rx, tx and
+		 * 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.
+		 * distributor first, then Rx, then tx.
 		 */
 		RTE_LCORE_FOREACH_WORKER(lcore_id) {
 
@@ -939,18 +934,18 @@ main(int argc, char *argv[])
 	}
 
 	if (enable_lcore_rx_distributor)
-		printf(" tx id %d, rx id %d\n",
+		printf(" Tx id %d, Rx id %d\n",
 			tx_core_id,
 			rx_core_id);
 	else
-		printf(" tx id %d, dist id %d, rx id %d\n",
+		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.
+	 * lcore_ids for Tx, Rx and distributor workloads.
 	 */
 	RTE_LCORE_FOREACH_WORKER(lcore_id) {
 		if (lcore_id == (unsigned int)distr_core_id ||
@@ -970,7 +965,7 @@ main(int argc, char *argv[])
 				p, lcore_id);
 	}
 
-	/* Start tx core */
+	/* Start Tx core */
 	rte_eal_remote_launch((lcore_function_t *)lcore_tx,
 			dist_tx_ring, tx_core_id);
 
@@ -978,7 +973,7 @@ main(int argc, char *argv[])
 	struct lcore_params *pd = NULL;
 	if (!enable_lcore_rx_distributor) {
 		pd = rte_malloc(NULL, sizeof(*pd), 0);
-		if (!pd)
+		if (pd == NULL)
 			rte_panic("malloc failure\n");
 		*pd = (struct lcore_params){worker_id++, d,
 			rx_dist_ring, dist_tx_ring, mbuf_pool};
@@ -986,7 +981,7 @@ main(int argc, char *argv[])
 				pd, distr_core_id);
 	}
 
-	/* Start rx core */
+	/* Start Rx core */
 	struct lcore_params *pr =
 		rte_malloc(NULL, sizeof(*pr), 0);
 	if (!pr)
-- 
2.27.0

[PATCH v7] examples/distributor: remove dead code and renaming Rx,Tx

[Abdullah Ömer Yamaç]

One line of commented code was dead code, this line and related
comments are removed. Naming of rx,RX and tx,TX are replaced by Rx and
Tx.

Signed-off-by: Abdullah Ömer Yamaç <omer.yamac@ceng.metu.edu.tr>
---
CC: Thomas Monjalon <thomas@monjalon.net>
---
 examples/distributor/main.c | 55 +++++++++++++++++--------------------
 1 file changed, 25 insertions(+), 30 deletions(-)

diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 21304d6618..9280a2694e 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -104,7 +104,7 @@ struct output_buffer {
 static void print_stats(void);
 
 /*
- * Initialises a given port using global settings and with the rx buffers
+ * Initialises a given port using global settings and with the Rx buffers
  * coming from the mbuf_pool passed as parameter
  */
 static inline int
@@ -259,12 +259,7 @@ lcore_rx(struct lcore_params *p)
 		}
 		app_stats.rx.rx_pkts += 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_rx, NULL);
@@ -282,7 +277,7 @@ lcore_rx(struct lcore_params *p)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
+	printf("\nCore %u exiting Rx task.\n", rte_lcore_id());
 	/* set distributor threads quit flag */
 	quit_signal_dist = 1;
 	return 0;
@@ -305,11 +300,11 @@ lcore_rx_and_distributor(struct lcore_params *p)
 		if (rte_eth_dev_socket_id(port) > 0 &&
 				rte_eth_dev_socket_id(port) != socket_id)
 			printf("WARNING, port %u is on remote NUMA node to "
-					"RX thread.\n\tPerformance will not "
+					"Rx thread.\n\tPerformance will not "
 					"be optimal.\n", port);
 	}
 
-	printf("\nCore %u doing packet RX and Distributor.\n", rte_lcore_id());
+	printf("\nCore %u doing packet Rx and Distributor.\n", rte_lcore_id());
 	port = 0;
 	while (!quit_signal_rx) {
 
@@ -329,8 +324,8 @@ lcore_rx_and_distributor(struct lcore_params *p)
 		app_stats.rx.rx_pkts += nb_rx;
 
 		/*
-		 * Run the distributor on the rx core. Returned
-		 * packets are then send straight to the tx core.
+		 * Run the distributor on the Rx core. Returned
+		 * packets are then send straight to the Tx core.
 		 */
 		rte_distributor_process(d, bufs, nb_rx);
 		const uint16_t nb_ret = rte_distributor_returned_pkts(d,
@@ -360,8 +355,8 @@ lcore_rx_and_distributor(struct lcore_params *p)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
-	/* set tx threads quit flag */
+	printf("\nCore %u exiting Rx task.\n", rte_lcore_id());
+	/* set Tx threads quit flag */
 	quit_signal = 1;
 	/* set worker threads quit flag */
 	quit_signal_work = 1;
@@ -448,7 +443,7 @@ lcore_distributor(struct lcore_params *p)
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
 	printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
-	/* set tx threads quit flag */
+	/* set Tx threads quit flag */
 	quit_signal = 1;
 	/* set worker threads quit flag */
 	quit_signal_work = 1;
@@ -478,7 +473,7 @@ lcore_tx(struct rte_ring *in_r)
 					"be optimal.\n", port);
 	}
 
-	printf("\nCore %u doing packet TX.\n", rte_lcore_id());
+	printf("\nCore %u doing packet Tx.\n", rte_lcore_id());
 	while (!quit_signal) {
 
 		RTE_ETH_FOREACH_DEV(port) {
@@ -524,7 +519,7 @@ lcore_tx(struct rte_ring *in_r)
 	}
 	if (power_lib_initialised)
 		rte_power_exit(rte_lcore_id());
-	printf("\nCore %u exiting tx task.\n", rte_lcore_id());
+	printf("\nCore %u exiting Tx task.\n", rte_lcore_id());
 	return 0;
 }
 
@@ -532,7 +527,7 @@ static void
 int_handler(int sig_num)
 {
 	printf("Exiting on signal %d\n", sig_num);
-	/* set quit flag for rx thread to exit */
+	/* set quit flag for Rx thread to exit */
 	quit_signal_rx = 1;
 }
 
@@ -698,7 +693,7 @@ print_usage(const char *prgname)
 {
 	printf("%s [EAL options] -- -p PORTMASK [-c]\n"
 			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n"
-			"  -c: Combines the RX core with the distribution core\n",
+			"  -c: Combines the Rx core with the distribution core\n",
 			prgname);
 }
 
@@ -798,11 +793,11 @@ main(int argc, char *argv[])
 		rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
 	if (enable_lcore_rx_distributor) {
-	/* RX and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
+	/* Rx and distributor combined, 3 fixed function cores (stat, Tx, at least 1 worker) */
 		min_cores = 4;
 		num_workers = rte_lcore_count() - 3;
 	} else {
-	/* separate RX and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
+	/* separate Rx and distributor, 3 fixed function cores (stat, Tx, at least 1 worker) */
 		min_cores = 5;
 		num_workers = rte_lcore_count() - 4;
 	}
@@ -811,8 +806,8 @@ main(int argc, char *argv[])
 		rte_exit(EXIT_FAILURE, "Error, This application needs at "
 				"least 4 logical cores to run:\n"
 				"1 lcore for stats (can be core 0)\n"
-				"1 or 2 lcore for packet RX and distribution\n"
-				"1 lcore for packet TX\n"
+				"1 or 2 lcore for packet Rx and distribution\n"
+				"1 lcore for packet Tx\n"
 				"and at least 1 lcore for worker threads\n");
 
 	if (init_power_library() == 0)
@@ -875,13 +870,13 @@ main(int argc, char *argv[])
 
 	if (power_lib_initialised) {
 		/*
-		 * Here we'll pre-assign lcore ids to the rx, tx and
+		 * 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.
+		 * distributor first, then Rx, then Tx.
 		 */
 		RTE_LCORE_FOREACH_WORKER(lcore_id) {
 
@@ -939,18 +934,18 @@ main(int argc, char *argv[])
 	}
 
 	if (enable_lcore_rx_distributor)
-		printf(" tx id %d, rx id %d\n",
+		printf(" Tx id %d, Rx id %d\n",
 			tx_core_id,
 			rx_core_id);
 	else
-		printf(" tx id %d, dist id %d, rx id %d\n",
+		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.
+	 * lcore_ids for Tx, Rx and distributor workloads.
 	 */
 	RTE_LCORE_FOREACH_WORKER(lcore_id) {
 		if (lcore_id == (unsigned int)distr_core_id ||
@@ -970,7 +965,7 @@ main(int argc, char *argv[])
 				p, lcore_id);
 	}
 
-	/* Start tx core */
+	/* Start Tx core */
 	rte_eal_remote_launch((lcore_function_t *)lcore_tx,
 			dist_tx_ring, tx_core_id);
 
@@ -978,7 +973,7 @@ main(int argc, char *argv[])
 	struct lcore_params *pd = NULL;
 	if (!enable_lcore_rx_distributor) {
 		pd = rte_malloc(NULL, sizeof(*pd), 0);
-		if (!pd)
+		if (pd == NULL)
 			rte_panic("malloc failure\n");
 		*pd = (struct lcore_params){worker_id++, d,
 			rx_dist_ring, dist_tx_ring, mbuf_pool};
@@ -986,7 +981,7 @@ main(int argc, char *argv[])
 				pd, distr_core_id);
 	}
 
-	/* Start rx core */
+	/* Start Rx core */
 	struct lcore_params *pr =
 		rte_malloc(NULL, sizeof(*pr), 0);
 	if (!pr)
-- 
2.27.0

[PATCH v2] examples/distributor: update dynamic configuration

[Abdullah Ömer Yamaç]

In this patch,
    * It is possible to switch the running mode of the distributor
using the command line argument.
    * With "-c" parameter, you can run RX and Distributor
on the same core.
    * Without "-c" parameter, you can run RX and Distributor
on the different core.
    * Syntax error of the single RX and distributor core is fixed.
    * Consecutive termination of the lcores fixed.
The termination order was wrong, and you couldn't terminate the
application while traffic was capturing. The current order is
RX -> Distributor -> TX -> Workers
    * When "-c" parameter is active, the wasted distributor core is
also deactivated in the main function.

Fixes: 4a7f40c0ff9a ("examples/distributor: add dedicated core")
Cc: stable@dpdk.org

Signed-off-by: Abdullah Ömer Yamaç <omer.yamac@ceng.metu.edu.tr>

---
Cc: david.hunt@intel.com
---
 doc/guides/sample_app_ug/dist_app.rst |   3 +-
 examples/distributor/main.c           | 222 ++++++++++++++++++--------
 2 files changed, 159 insertions(+), 66 deletions(-)

diff --git a/doc/guides/sample_app_ug/dist_app.rst b/doc/guides/sample_app_ug/dist_app.rst
index 3bd03905c3..5c80561187 100644
--- a/doc/guides/sample_app_ug/dist_app.rst
+++ b/doc/guides/sample_app_ug/dist_app.rst
@@ -42,11 +42,12 @@ Running the Application
 
    ..  code-block:: console
 
-       ./<build-dir>/examples/dpdk-distributor [EAL options] -- -p PORTMASK
+       ./<build-dir>/examples/dpdk-distributor [EAL options] -- -p PORTMASK [-c]
 
    where,
 
    *   -p PORTMASK: Hexadecimal bitmask of ports to configure
+   *   -c: Combines the RX core with distribution core
 
 #. To run the application in linux environment with 10 lcores, 4 ports,
    issue the command:
diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 8995806b4e..632d5e5554 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -39,6 +39,8 @@ volatile uint8_t quit_signal_rx;
 volatile uint8_t quit_signal_dist;
 volatile uint8_t quit_signal_work;
 unsigned int power_lib_initialised;
+bool enable_lcore_rx_distributor;
+unsigned int num_workers;
 
 static volatile struct app_stats {
 	struct {
@@ -256,14 +258,82 @@ lcore_rx(struct lcore_params *p)
 		}
 		app_stats.rx.rx_pkts += nb_rx;
 
-/*
- * 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(p->d, bufs, nb_rx);
-		const uint16_t nb_ret = rte_distributor_returned_pkts(p->d,
-			bufs, BURST_SIZE*2);
+		/*
+		 * 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_rx, NULL);
+
+		app_stats.rx.enqueued_pkts += sent;
+		if (unlikely(sent < nb_rx)) {
+			app_stats.rx.enqdrop_pkts +=  nb_rx - sent;
+			RTE_LOG_DP(DEBUG, DISTRAPP,
+				"%s:Packet loss due to full ring\n", __func__);
+			while (sent < nb_rx)
+				rte_pktmbuf_free(bufs[sent++]);
+		}
+		if (++port == nb_ports)
+			port = 0;
+	}
+	if (power_lib_initialised)
+		rte_power_exit(rte_lcore_id());
+	printf("\nCore %u exiting rx task.\n", rte_lcore_id());
+	/* set distributor threads quit flag */
+	quit_signal_dist = 1;
+	return 0;
+}
+
+static int
+lcore_rx_and_distributor(struct lcore_params *p)
+{
+	struct rte_distributor *d = p->d;
+	const uint16_t nb_ports = rte_eth_dev_count_avail();
+	const int socket_id = rte_socket_id();
+	uint16_t port;
+	struct rte_mbuf *bufs[BURST_SIZE*2];
+
+	RTE_ETH_FOREACH_DEV(port) {
+		/* skip ports that are not enabled */
+		if ((enabled_port_mask & (1 << port)) == 0)
+			continue;
+
+		if (rte_eth_dev_socket_id(port) > 0 &&
+				rte_eth_dev_socket_id(port) != socket_id)
+			printf("WARNING, port %u is on remote NUMA node to "
+					"RX thread.\n\tPerformance will not "
+					"be optimal.\n", port);
+	}
+
+	printf("\nCore %u doing packet RX and Distributor.\n", rte_lcore_id());
+	port = 0;
+	while (!quit_signal_rx) {
+
+		/* skip ports that are not enabled */
+		if ((enabled_port_mask & (1 << port)) == 0) {
+			if (++port == nb_ports)
+				port = 0;
+			continue;
+		}
+		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;
+
+		/*
+		 * Run the distributor on the rx core. Returned
+		 * packets are then send straight to the tx core.
+		 */
+		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)) {
@@ -275,18 +345,6 @@ 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, 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)) {
@@ -301,9 +359,14 @@ lcore_rx(struct lcore_params *p)
 	}
 	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());
+	/* set tx threads quit flag */
 	quit_signal = 1;
+	/* set worker threads quit flag */
+	quit_signal_work = 1;
+	rte_distributor_flush(d);
+	/* Unblock any returns so workers can exit */
+	rte_distributor_clear_returns(d);
 	return 0;
 }
 
@@ -381,14 +444,16 @@ 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());
+	printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
+	/* set tx threads quit flag */
+	quit_signal = 1;
+	/* set worker threads quit flag */
+	quit_signal_work = 1;
 	rte_distributor_flush(d);
 	/* Unblock any returns so workers can exit */
 	rte_distributor_clear_returns(d);
-	quit_signal_rx = 1;
 	return 0;
 }
 
@@ -467,7 +532,7 @@ int_handler(int sig_num)
 {
 	printf("Exiting on signal %d\n", sig_num);
 	/* set quit flag for rx thread to exit */
-	quit_signal_dist = 1;
+	quit_signal_rx = 1;
 }
 
 static void
@@ -475,7 +540,6 @@ print_stats(void)
 {
 	struct rte_eth_stats eth_stats;
 	unsigned int i, j;
-	const unsigned int num_workers = rte_lcore_count() - 4;
 
 	RTE_ETH_FOREACH_DEV(i) {
 		rte_eth_stats_get(i, &eth_stats);
@@ -504,20 +568,22 @@ print_stats(void)
 			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);
+	if (!enable_lcore_rx_distributor) {
+		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",
@@ -629,8 +695,9 @@ init_power_library(void)
 static void
 print_usage(const char *prgname)
 {
-	printf("%s [EAL options] -- -p PORTMASK\n"
-			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n",
+	printf("%s [EAL options] -- -p PORTMASK [-c]\n"
+			"  -p PORTMASK: hexadecimal bitmask of ports to configure\n"
+			"  -c: Combines the RX core with the distribution core\n",
 			prgname);
 }
 
@@ -661,8 +728,8 @@ parse_args(int argc, char **argv)
 	};
 
 	argvopt = argv;
-
-	while ((opt = getopt_long(argc, argvopt, "p:",
+	enable_lcore_rx_distributor = false;
+	while ((opt = getopt_long(argc, argvopt, "cp:",
 			lgopts, &option_index)) != EOF) {
 
 		switch (opt) {
@@ -676,6 +743,10 @@ parse_args(int argc, char **argv)
 			}
 			break;
 
+		case 'c':
+			enable_lcore_rx_distributor = true;
+			break;
+
 		default:
 			print_usage(prgname);
 			return -1;
@@ -705,6 +776,7 @@ main(int argc, char *argv[])
 	unsigned int lcore_id, worker_id = 0;
 	int distr_core_id = -1, rx_core_id = -1, tx_core_id = -1;
 	unsigned nb_ports;
+	unsigned int min_cores;
 	uint16_t portid;
 	uint16_t nb_ports_available;
 	uint64_t t, freq;
@@ -724,12 +796,21 @@ main(int argc, char *argv[])
 	if (ret < 0)
 		rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
 
-	if (rte_lcore_count() < 5)
+	if (enable_lcore_rx_distributor) {
+	/* RX and distributor combined, 3 fixed function cores (stat, TX, at least 1 worker) */
+		min_cores = 4;
+		num_workers = rte_lcore_count() - 3;
+	} else {
+	/* separate RX and distributor, 3 fixed function cores (stat, TX, at least 1 worker) */
+		min_cores = 5;
+		num_workers = rte_lcore_count() - 4;
+	}
+
+	if (rte_lcore_count() < min_cores)
 		rte_exit(EXIT_FAILURE, "Error, This application needs at "
-				"least 5 logical cores to run:\n"
+				"least 4 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 or 2 lcore for packet RX and distribution\n"
 				"1 lcore for packet TX\n"
 				"and at least 1 lcore for worker threads\n");
 
@@ -772,7 +853,7 @@ main(int argc, char *argv[])
 	}
 
 	d = rte_distributor_create("PKT_DIST", rte_socket_id(),
-			rte_lcore_count() - 4,
+			num_workers,
 			RTE_DIST_ALG_BURST);
 	if (d == NULL)
 		rte_exit(EXIT_FAILURE, "Cannot create distributor\n");
@@ -808,7 +889,7 @@ main(int argc, char *argv[])
 			if (lcore_cap.priority != 1)
 				continue;
 
-			if (distr_core_id < 0) {
+			if (distr_core_id < 0 && !enable_lcore_rx_distributor) {
 				distr_core_id = lcore_id;
 				printf("Distributor on priority core %d\n",
 					lcore_id);
@@ -839,7 +920,7 @@ main(int argc, char *argv[])
 				lcore_id == (unsigned int)rx_core_id ||
 				lcore_id == (unsigned int)tx_core_id)
 			continue;
-		if (distr_core_id < 0) {
+		if (distr_core_id < 0 && !enable_lcore_rx_distributor) {
 			distr_core_id = lcore_id;
 			printf("Distributor on core %d\n", lcore_id);
 			continue;
@@ -856,7 +937,12 @@ main(int argc, char *argv[])
 		}
 	}
 
-	printf(" tx id %d, dist id %d, rx id %d\n",
+	if (enable_lcore_rx_distributor)
+		printf(" tx id %d, rx id %d\n",
+			tx_core_id,
+			rx_core_id);
+	else
+		printf(" tx id %d, dist id %d, rx id %d\n",
 			tx_core_id,
 			distr_core_id,
 			rx_core_id);
@@ -888,15 +974,16 @@ main(int argc, char *argv[])
 			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);
+	struct lcore_params *pd = NULL;
+	if (!enable_lcore_rx_distributor) {
+		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 =
@@ -905,12 +992,16 @@ main(int argc, char *argv[])
 		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);
+	if (enable_lcore_rx_distributor)
+		rte_eal_remote_launch((lcore_function_t *)lcore_rx_and_distributor,
+				pr, rx_core_id);
+	else
+		rte_eal_remote_launch((lcore_function_t *)lcore_rx,
+				pr, rx_core_id);
 
 	freq = rte_get_timer_hz();
 	t = rte_rdtsc() + freq;
-	while (!quit_signal_dist) {
+	while (!quit_signal) {
 		if (t < rte_rdtsc()) {
 			print_stats();
 			t = rte_rdtsc() + freq;
@@ -925,7 +1016,8 @@ main(int argc, char *argv[])
 
 	print_stats();
 
-	rte_free(pd);
+	if (pd)
+		rte_free(pd);
 	rte_free(pr);
 
 	/* clean up the EAL */
-- 
2.27.0