[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..6e98f78054 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;
+	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 (!enable_lcore_rx_distributor)
+		rte_free(pd);
 	rte_free(pr);
 
 	/* clean up the EAL */
-- 
2.27.0

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

[Hunt, David]

Hi Ömer,

I've a few comments:

On 21/06/2022 21:15, 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.
> 	* 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..6e98f78054 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;

This could be "num_workers = rte_lcore_count() - (3 + 
enable_lcore_rx_distributor)"

Also, if you don't like the magic number, you could add a #define 
FIXED_FUNCTION_CORES 3 at the top of the file and use here.


>   
>   	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;


I'd suggest removing this line and setting the initial value in the 
declaration above. I would also suggest defaulting it to true.


>   
>   	/* 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;
> +

This could be "num_workers = rte_lcore_count() - (4 - 
enable_lcore_rx_distributor)".


> +	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");
> +


the two checks above could be replaced with something like:

min_cores = 4 + enable_lcore_rx_distributor;
if (rte_lcore_count() < min_cores)
                 rte_exit(EXIT_FAILURE, "Error, This application needs at "
                                 "least %d 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",
                                 min_cores);

> +
>   	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;
> +	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 (!enable_lcore_rx_distributor)
> +		rte_free(pd);
>   	rte_free(pr);
>   
>   	/* clean up the EAL */


Thanks,

Dave.

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

[Omer Yamac]

Hi David,

Thank you for your review. I have two questions. The first one is about 
new release. As I remember new DPDK realize will be published in a short 
time and my previous fix in that release. Therefore, Should I wait for 
that release to submit patch?

The other question is below,

On 27.06.2022 18:51, Hunt, David wrote:
> Hi Ömer,
> 
> I've a few comments:
> 
> On 21/06/2022 21:15, 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.
>> 	* 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..6e98f78054 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;
> 
> This could be "num_workers = rte_lcore_count() - (3 +
> enable_lcore_rx_distributor)"
> 
> Also, if you don't like the magic number, you could add a #define
> FIXED_FUNCTION_CORES 3 at the top of the file and use here.
> 
> 
>>     	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;
> 
> 
> I'd suggest removing this line and setting the initial value in the
> declaration above. I would also suggest defaulting it to true.
> 
> 
>>     	/* 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;
>> +
> 
> This could be "num_workers = rte_lcore_count() - (4 -
> enable_lcore_rx_distributor)".
> 
> 
>> +	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");
>> +
> 
> 
> the two checks above could be replaced with something like:
> 
> min_cores = 4 + enable_lcore_rx_distributor;
> if (rte_lcore_count() < min_cores)
>                 rte_exit(EXIT_FAILURE, "Error, This application needs 
> at "
>                                 "least %d 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",
>                                 min_cores);
> 
Is it okay, if I change the error string such that:
"1 or 2 lcore for packet RX and distribution"

>> +
>>   	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;
>> +	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 (!enable_lcore_rx_distributor)
>> +		rte_free(pd);
>>   	rte_free(pr);
>>     	/* clean up the EAL */
> 
> 
> Thanks,
> 
> Dave.

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

[Hunt, David]

Hi Ömer,

On 27/06/2022 17:28, Omer Yamac wrote:
> Hi David,
>
> Thank you for your review. I have two questions. The first one is 
> about new release. As I remember new DPDK realize will be published in 
> a short time and my previous fix in that release. Therefore, Should I 
> wait for that release to submit patch?
>

Yes, You can wait, or you can submit to the mailing list now and mark 
the patch as "Deferred" in patchwork. Once 22.07 is released it will get 
marked as "New", and under consideration for 22.11.


> The other question is below,
>
> On 27.06.2022 18:51, Hunt, David wrote:
>> Hi Ömer,
>>
>> I've a few comments:
>>
>> On 21/06/2022 21:15, 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.
>>>     * 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>
>>>
>>> --- 


--snip--


>>>                   "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");
>>> +
>>
>>
>> the two checks above could be replaced with something like:
>>
>> min_cores = 4 + enable_lcore_rx_distributor;
>> if (rte_lcore_count() < min_cores)
>>                 rte_exit(EXIT_FAILURE, "Error, This application needs 
>> at "
>>                                 "least %d 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",
>>                                 min_cores);
>>
> Is it okay, if I change the error string such that:
> "1 or 2 lcore for packet RX and distribution"
>

Sure, that's fine.



--snip--


Thanks,

Dave.

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

[Omer Yamac]

Hi David,

I have one more question. When I was working on new patch, I just want 
to make sure what we are doing.
On 27.06.2022 18:51, Hunt, David wrote:
> Hi Ömer,
> 
> I've a few comments:
> 
> On 21/06/2022 21:15, Abdullah Ömer Yamaç wrote:
--clipped--
>> @@ -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 {
--clipped--
>> @@ -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;
>> +
> 
> This could be "num_workers = rte_lcore_count() - (4 -
> enable_lcore_rx_distributor)".
> 
For the "if-else" case of enable_lcore_rx_distributor, we will reduce 
the line of codes; but I am not sure about that change. Because the type 
of the variable is bool and we are using arithmetic operation on that 
variable. I think it is a little bit harder for people to understand 
operation. Am I right? I can suggest one more solution. We may change 
the data type to "unsigned int" or Is it okay to leave as before?

--clipped--

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

[Hunt, David]

On 28/06/2022 12:06, Omer Yamac wrote:
> Hi David,
>
> I have one more question. When I was working on new patch, I just want 
> to make sure what we are doing.
> On 27.06.2022 18:51, Hunt, David wrote:
>> Hi Ömer,
>>
>> I've a few comments:
>>
>> On 21/06/2022 21:15, Abdullah Ömer Yamaç wrote:
> --clipped--
>>> @@ -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 {
> --clipped--
>>> @@ -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;
>>> +
>>
>> This could be "num_workers = rte_lcore_count() - (4 -
>> enable_lcore_rx_distributor)".
>>
> For the "if-else" case of enable_lcore_rx_distributor, we will reduce 
> the line of codes; but I am not sure about that change. Because the 
> type of the variable is bool and we are using arithmetic operation on 
> that variable. I think it is a little bit harder for people to 
> understand operation. Am I right? I can suggest one more solution. We 
> may change the data type to "unsigned int" or Is it okay to leave as 
> before?
>
> --clipped--


Hi Ömer,

    You raise a good point about readability. Let's leave it as you had 
it originally.  Maybe just add a couple of one-line comments? "rx and 
distributor combined, 3 fixed function cores" and "separate rx and 
distributor, 4 fixed function cores?

Rgds,
Dave.

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

[Omer Yamac]

Hi,
Here is the final version. If it is ok, I will test the code and 
publish.

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;
}

On 28.06.2022 14:25, Hunt, David wrote:
> On 28/06/2022 12:06, Omer Yamac wrote:
>> Hi David,
>> 
>> I have one more question. When I was working on new patch, I just want 
>> to make sure what we are doing.
>> On 27.06.2022 18:51, Hunt, David wrote:
>>> Hi Ömer,
>>> 
>>> I've a few comments:
>>> 
>>> On 21/06/2022 21:15, Abdullah Ömer Yamaç wrote:
>> --clipped--
>>>> @@ -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 {
>> --clipped--
>>>> @@ -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;
>>>> +
>>> 
>>> This could be "num_workers = rte_lcore_count() - (4 -
>>> enable_lcore_rx_distributor)".
>>> 
>> For the "if-else" case of enable_lcore_rx_distributor, we will reduce 
>> the line of codes; but I am not sure about that change. Because the 
>> type of the variable is bool and we are using arithmetic operation on 
>> that variable. I think it is a little bit harder for people to 
>> understand operation. Am I right? I can suggest one more solution. We 
>> may change the data type to "unsigned int" or Is it okay to leave as 
>> before?
>> 
>> --clipped--
> 
> 
> Hi Ömer,
> 
>    You raise a good point about readability. Let's leave it as you had
> it originally.  Maybe just add a couple of one-line comments? "rx and
> distributor combined, 3 fixed function cores" and "separate rx and
> distributor, 4 fixed function cores?
> 
> Rgds,
> Dave.

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

[Hunt, David]

On 28/06/2022 13:06, Omer Yamac wrote:
> Hi,
> Here is the final version. If it is ok, I will test the code and publish.
>
> 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;
> }
>

Hi Ömer,

Please use standard comment formatting (look elsewhere for examples).

I'm not sure you need min_cores any more.

Rgds,

Dave.


--snip--

[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