[PATCH] testpmd: add hairpin-map parameter

[PATCH] testpmd: add hairpin-map parameter

[Gregory Etelson]

Hairpin offloads packet forwarding.
Packet is expected on Rx port <rp>, Rx queue <rq> and is delivered
to Tx port <tp> Tx queue <tq>.

Testpmd implements a static hairpin configuration scheme.
The scheme implicitly matches next valid port for given <rp> or <tp>.
That approach can be used in a single or double port setups only.

The new parameter allows explicit selection of Rx and Tx ports and
queues in hairpin configuration.
The new `hairpin-map` parameter is provided with 5 parameters,
separated by `:`

`--hairpin-map=Rx port id:Rx queue:Tx port id:Tx queue:queues number`

Testpmd operator can provide several `hairpin-map` parameters for
different hairpin maps.
Example:

dpdk-testpmd <EAL params> -- \
  <testpmd params> \
  --rxq=2 --txq=2 --hairpinq=2 --hairpin-mode=0x12 \
  --hairpin-map=0:2:1:2:1 \ # [1]
  --hairpin-map=0:3:2:2:3   # [2]

Hairpin map [1] binds Rx port 0, queue 2 with Tx port 1, queue 2.
Hairpin map [2] binds
  Rx port 0, queue 3 with Tx port 2, queue 2,
  Rx port 0, queue 4 with Tx port 2, queue 3,
  Rx port 0, queue 5 with Tx port 2, queue 4.

The new `hairpin-map` parameter is optional.
If omitted, testpmd will create "default" hairpin maps.

Signed-off-by: Gregory Etelson <getelson@nvidia.com>
---
 app/test-pmd/parameters.c             |  63 ++++++++
 app/test-pmd/testpmd.c                | 212 ++++++++++++++++++--------
 app/test-pmd/testpmd.h                |  18 +++
 doc/guides/testpmd_app_ug/run_app.rst |   3 +
 4 files changed, 230 insertions(+), 66 deletions(-)

diff --git a/app/test-pmd/parameters.c b/app/test-pmd/parameters.c
index a9ca58339d..675de81861 100644
--- a/app/test-pmd/parameters.c
+++ b/app/test-pmd/parameters.c
@@ -206,6 +206,12 @@ usage(char* progname)
 	printf("  --hairpin-mode=0xXX: bitmask set the hairpin port mode.\n"
 	       "    0x10 - explicit Tx rule, 0x02 - hairpin ports paired\n"
 	       "    0x01 - hairpin ports loop, 0x00 - hairpin port self\n");
+	printf("  --hairpin-map=rxpi:rxq:txpi:txq:n: hairpin map.\n"
+	       "    rxpi - Rx port index.\n"
+	       "    rxq  - Rx queue.\n"
+	       "    txpi - Tx port index.\n"
+	       "    txq  - Tx queue.\n"
+	       "    n    - hairpin queues number.\n");
 }
 
 #ifdef RTE_LIB_CMDLINE
@@ -588,6 +594,55 @@ parse_link_speed(int n)
 	return speed;
 }
 
+static __rte_always_inline
+char *parse_hairpin_map_entry(char *input, char **next)
+{
+	char *tail = strchr(input, ':');
+
+	if (!tail)
+		return NULL;
+	tail[0] = '\0';
+	*next = tail + 1;
+	return input;
+}
+
+static int
+parse_hairpin_map(char *hpmap)
+{
+	/*
+	 * Testpmd hairpin map format:
+	 * <Rx port id:First Rx queue:Tx port id:First Tx queue:queues number>
+	 */
+	char *head, *next = hpmap;
+	struct hairpin_map *map = calloc(1, sizeof(*map));
+
+	if (!map)
+		return -ENOMEM;
+
+	head = parse_hairpin_map_entry(next, &next);
+	if (!head)
+		goto err;
+	map->rx_port = atoi(head);
+	head = parse_hairpin_map_entry(next, &next);
+	if (!head)
+		goto err;
+	map->rxq_head = atoi(head);
+	head = parse_hairpin_map_entry(next, &next);
+	if (!head)
+		goto err;
+	map->tx_port = atoi(head);
+	head = parse_hairpin_map_entry(next, &next);
+	if (!head)
+		goto err;
+	map->txq_head = atoi(head);
+	map->qnum = atoi(next);
+	hairpin_add_multiport_map(map);
+	return 0;
+err:
+	free(map);
+	return -EINVAL;
+}
+
 void
 launch_args_parse(int argc, char** argv)
 {
@@ -663,6 +718,7 @@ launch_args_parse(int argc, char** argv)
 		{ "txd",			1, 0, 0 },
 		{ "hairpinq",			1, 0, 0 },
 		{ "hairpin-mode",		1, 0, 0 },
+		{ "hairpin-map",                1, 0, 0 },
 		{ "burst",			1, 0, 0 },
 		{ "flowgen-clones",		1, 0, 0 },
 		{ "flowgen-flows",		1, 0, 0 },
@@ -1111,6 +1167,13 @@ launch_args_parse(int argc, char** argv)
 				else
 					hairpin_mode = (uint32_t)n;
 			}
+			if (!strcmp(lgopts[opt_idx].name, "hairpin-map")) {
+				hairpin_multiport_mode = true;
+				ret = parse_hairpin_map(optarg);
+				if (ret)
+					rte_exit(EXIT_FAILURE, "invalid hairpin map\n");
+
+			}
 			if (!strcmp(lgopts[opt_idx].name, "burst")) {
 				n = atoi(optarg);
 				if (n == 0) {
diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c
index 938ca035d4..2c6975f22d 100644
--- a/app/test-pmd/testpmd.c
+++ b/app/test-pmd/testpmd.c
@@ -434,6 +434,16 @@ uint8_t clear_ptypes = true;
 /* Hairpin ports configuration mode. */
 uint32_t hairpin_mode;
 
+bool hairpin_multiport_mode = false;
+
+static LIST_HEAD(, hairpin_map) hairpin_map_head = LIST_HEAD_INITIALIZER();
+
+void
+hairpin_add_multiport_map(struct hairpin_map *map)
+{
+	LIST_INSERT_HEAD(&hairpin_map_head, map, entry);
+}
+
 /* Pretty printing of ethdev events */
 static const char * const eth_event_desc[] = {
 	[RTE_ETH_EVENT_UNKNOWN] = "unknown",
@@ -2677,28 +2687,107 @@ port_is_started(portid_t port_id)
 #define HAIRPIN_MODE_TX_LOCKED_MEMORY RTE_BIT32(16)
 #define HAIRPIN_MODE_TX_RTE_MEMORY RTE_BIT32(17)
 
-
-/* Configure the Rx and Tx hairpin queues for the selected port. */
 static int
-setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
+port_config_hairpin_rxq(portid_t pi, uint16_t peer_tx_port,
+			queueid_t rxq_head, queueid_t txq_head,
+			uint16_t qcount, uint32_t manual_bind)
 {
-	queueid_t qi;
+	int diag;
+	queueid_t i, qi;
+	uint32_t tx_explicit = !!(hairpin_mode & 0x10);
+	uint32_t force_mem = !!(hairpin_mode & HAIRPIN_MODE_RX_FORCE_MEMORY);
+	uint32_t locked_mem = !!(hairpin_mode & HAIRPIN_MODE_RX_LOCKED_MEMORY);
+	uint32_t rte_mem = !!(hairpin_mode & HAIRPIN_MODE_RX_RTE_MEMORY);
+	struct rte_port *port = &ports[pi];
 	struct rte_eth_hairpin_conf hairpin_conf = {
 		.peer_count = 1,
 	};
-	int i;
+
+	for (qi = rxq_head, i = 0; qi < rxq_head + qcount; qi++) {
+		hairpin_conf.peers[0].port = peer_tx_port;
+		hairpin_conf.peers[0].queue = i + txq_head;
+		hairpin_conf.manual_bind = manual_bind;
+		hairpin_conf.tx_explicit = tx_explicit;
+		hairpin_conf.force_memory = force_mem;
+		hairpin_conf.use_locked_device_memory = locked_mem;
+		hairpin_conf.use_rte_memory = rte_mem;
+		diag = rte_eth_rx_hairpin_queue_setup
+			(pi, qi, nb_rxd, &hairpin_conf);
+		i++;
+		if (diag == 0)
+			continue;
+
+		/* Fail to setup rx queue, return */
+		if (port->port_status == RTE_PORT_HANDLING)
+			port->port_status = RTE_PORT_STOPPED;
+		else
+			fprintf(stderr,
+				"Port %d can not be set back to stopped\n", pi);
+		fprintf(stderr,
+			"Port %d failed to configure hairpin on rxq %u.\n"
+			"Peer port: %u peer txq: %u\n",
+			pi, qi, peer_tx_port, i);
+		/* try to reconfigure queues next time */
+		port->need_reconfig_queues = 1;
+		return -1;
+	}
+	return 0;
+}
+
+static int
+port_config_hairpin_txq(portid_t pi, uint16_t peer_rx_port,
+			queueid_t rxq_head, queueid_t txq_head,
+			uint16_t qcount, uint32_t manual_bind)
+{
 	int diag;
+	queueid_t i, qi;
+	uint32_t tx_explicit = !!(hairpin_mode & 0x10);
+	uint32_t force_mem = !!(hairpin_mode & HAIRPIN_MODE_TX_FORCE_MEMORY);
+	uint32_t locked_mem = !!(hairpin_mode & HAIRPIN_MODE_TX_LOCKED_MEMORY);
+	uint32_t rte_mem = !!(hairpin_mode & HAIRPIN_MODE_TX_RTE_MEMORY);
 	struct rte_port *port = &ports[pi];
+	struct rte_eth_hairpin_conf hairpin_conf = {
+		.peer_count = 1,
+	};
+
+	for (qi = txq_head, i = 0; qi < txq_head + qcount; qi++) {
+		hairpin_conf.peers[0].port = peer_rx_port;
+		hairpin_conf.peers[0].queue = i + rxq_head;
+		hairpin_conf.manual_bind = manual_bind;
+		hairpin_conf.tx_explicit = tx_explicit;
+		hairpin_conf.force_memory = force_mem;
+		hairpin_conf.use_locked_device_memory = locked_mem;
+		hairpin_conf.use_rte_memory = rte_mem;
+		diag = rte_eth_tx_hairpin_queue_setup
+			(pi, qi, nb_txd, &hairpin_conf);
+		i++;
+		if (diag == 0)
+			continue;
+
+		/* Fail to setup rx queue, return */
+		if (port->port_status == RTE_PORT_HANDLING)
+			port->port_status = RTE_PORT_STOPPED;
+		else
+			fprintf(stderr,
+				"Port %d can not be set back to stopped\n", pi);
+		fprintf(stderr,
+			"Port %d failed to configure hairpin on txq %u.\n"
+			"Peer port: %u peer rxq: %u\n",
+			pi, qi, peer_rx_port, i);
+		/* try to reconfigure queues next time */
+		port->need_reconfig_queues = 1;
+		return -1;
+	}
+	return 0;
+}
+
+static int
+setup_legacy_hairpin_queus(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
+{
+	int diag;
 	uint16_t peer_rx_port = pi;
 	uint16_t peer_tx_port = pi;
 	uint32_t manual = 1;
-	uint32_t tx_exp = hairpin_mode & 0x10;
-	uint32_t rx_force_memory = hairpin_mode & HAIRPIN_MODE_RX_FORCE_MEMORY;
-	uint32_t rx_locked_memory = hairpin_mode & HAIRPIN_MODE_RX_LOCKED_MEMORY;
-	uint32_t rx_rte_memory = hairpin_mode & HAIRPIN_MODE_RX_RTE_MEMORY;
-	uint32_t tx_force_memory = hairpin_mode & HAIRPIN_MODE_TX_FORCE_MEMORY;
-	uint32_t tx_locked_memory = hairpin_mode & HAIRPIN_MODE_TX_LOCKED_MEMORY;
-	uint32_t tx_rte_memory = hairpin_mode & HAIRPIN_MODE_TX_RTE_MEMORY;
 
 	if (!(hairpin_mode & 0xf)) {
 		peer_rx_port = pi;
@@ -2706,10 +2795,10 @@ setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
 		manual = 0;
 	} else if (hairpin_mode & 0x1) {
 		peer_tx_port = rte_eth_find_next_owned_by(pi + 1,
-						       RTE_ETH_DEV_NO_OWNER);
+							  RTE_ETH_DEV_NO_OWNER);
 		if (peer_tx_port >= RTE_MAX_ETHPORTS)
 			peer_tx_port = rte_eth_find_next_owned_by(0,
-						RTE_ETH_DEV_NO_OWNER);
+								  RTE_ETH_DEV_NO_OWNER);
 		if (p_pi != RTE_MAX_ETHPORTS) {
 			peer_rx_port = p_pi;
 		} else {
@@ -2725,69 +2814,60 @@ setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
 			peer_rx_port = p_pi;
 		} else {
 			peer_rx_port = rte_eth_find_next_owned_by(pi + 1,
-						RTE_ETH_DEV_NO_OWNER);
+								  RTE_ETH_DEV_NO_OWNER);
 			if (peer_rx_port >= RTE_MAX_ETHPORTS)
 				peer_rx_port = pi;
 		}
 		peer_tx_port = peer_rx_port;
 		manual = 1;
 	}
+	diag = port_config_hairpin_txq(pi, peer_rx_port, nb_rxq, nb_txq,
+				       nb_hairpinq, manual);
+	if (diag)
+		return diag;
+	diag = port_config_hairpin_rxq(pi, peer_tx_port, nb_rxq, nb_txq,
+				       nb_hairpinq, manual);
+	if (diag)
+		return diag;
+	return 0;
+}
 
-	for (qi = nb_txq, i = 0; qi < nb_hairpinq + nb_txq; qi++) {
-		hairpin_conf.peers[0].port = peer_rx_port;
-		hairpin_conf.peers[0].queue = i + nb_rxq;
-		hairpin_conf.manual_bind = !!manual;
-		hairpin_conf.tx_explicit = !!tx_exp;
-		hairpin_conf.force_memory = !!tx_force_memory;
-		hairpin_conf.use_locked_device_memory = !!tx_locked_memory;
-		hairpin_conf.use_rte_memory = !!tx_rte_memory;
-		diag = rte_eth_tx_hairpin_queue_setup
-			(pi, qi, nb_txd, &hairpin_conf);
-		i++;
-		if (diag == 0)
-			continue;
-
-		/* Fail to setup rx queue, return */
-		if (port->port_status == RTE_PORT_HANDLING)
-			port->port_status = RTE_PORT_STOPPED;
-		else
-			fprintf(stderr,
-				"Port %d can not be set back to stopped\n", pi);
-		fprintf(stderr, "Fail to configure port %d hairpin queues\n",
-			pi);
-		/* try to reconfigure queues next time */
-		port->need_reconfig_queues = 1;
-		return -1;
-	}
-	for (qi = nb_rxq, i = 0; qi < nb_hairpinq + nb_rxq; qi++) {
-		hairpin_conf.peers[0].port = peer_tx_port;
-		hairpin_conf.peers[0].queue = i + nb_txq;
-		hairpin_conf.manual_bind = !!manual;
-		hairpin_conf.tx_explicit = !!tx_exp;
-		hairpin_conf.force_memory = !!rx_force_memory;
-		hairpin_conf.use_locked_device_memory = !!rx_locked_memory;
-		hairpin_conf.use_rte_memory = !!rx_rte_memory;
-		diag = rte_eth_rx_hairpin_queue_setup
-			(pi, qi, nb_rxd, &hairpin_conf);
-		i++;
-		if (diag == 0)
-			continue;
-
-		/* Fail to setup rx queue, return */
-		if (port->port_status == RTE_PORT_HANDLING)
-			port->port_status = RTE_PORT_STOPPED;
-		else
-			fprintf(stderr,
-				"Port %d can not be set back to stopped\n", pi);
-		fprintf(stderr, "Fail to configure port %d hairpin queues\n",
-			pi);
-		/* try to reconfigure queues next time */
-		port->need_reconfig_queues = 1;
-		return -1;
+static int
+setup_mapped_harpin_queues(portid_t pi)
+{
+	int ret = 0;
+	struct hairpin_map *map;
+
+	LIST_FOREACH(map, &hairpin_map_head, entry) {
+		if (map->rx_port == pi) {
+			ret = port_config_hairpin_rxq(pi, map->tx_port,
+						      map->rxq_head,
+						      map->txq_head,
+						      map->qnum, true);
+			if (ret)
+				return ret;
+		}
+		if (map->tx_port == pi) {
+			ret = port_config_hairpin_txq(pi, map->rx_port,
+						      map->rxq_head,
+						      map->txq_head,
+						      map->qnum, true);
+			if (ret)
+				return ret;
+		}
 	}
 	return 0;
 }
 
+/* Configure the Rx and Tx hairpin queues for the selected port. */
+static int
+setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
+{
+	return !hairpin_multiport_mode ?
+	       setup_legacy_hairpin_queus(pi, p_pi, cnt_pi) :
+	       setup_mapped_harpin_queues(pi);
+}
+
 /* Configure the Rx with optional split. */
 int
 rx_queue_setup(uint16_t port_id, uint16_t rx_queue_id,
diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h
index f1df6a8faf..208e8e9514 100644
--- a/app/test-pmd/testpmd.h
+++ b/app/test-pmd/testpmd.h
@@ -125,6 +125,24 @@ enum noisy_fwd_mode {
 	NOISY_FWD_MODE_MAX,
 };
 
+struct hairpin_map {
+	LIST_ENTRY(hairpin_map) entry; /**< List entry. */
+	portid_t rx_port; /**< Hairpin Rx port ID. */
+	portid_t tx_port; /**< Hairpin Tx port ID. */
+	uint16_t rxq_head; /**< Hairpin Rx queue head. */
+	uint16_t txq_head; /**< Hairpin Tx queue head. */
+	uint16_t qnum; /**< Hairpin queues number. */
+};
+
+/**
+ * Command line arguments parser sets `hairpin_multiport_mode` to True
+ * if explicit hairpin map configuration mode was used.
+ */
+extern bool hairpin_multiport_mode;
+
+/** Hairpin maps list. */
+extern void hairpin_add_multiport_map(struct hairpin_map *map);
+
 /**
  * The data structure associated with RX and TX packet burst statistics
  * that are recorded for each forwarding stream.
diff --git a/doc/guides/testpmd_app_ug/run_app.rst b/doc/guides/testpmd_app_ug/run_app.rst
index 6e9c552e76..a202c98b4c 100644
--- a/doc/guides/testpmd_app_ug/run_app.rst
+++ b/doc/guides/testpmd_app_ug/run_app.rst
@@ -566,6 +566,9 @@ The command line options are:
 
     The default value is 0. Hairpin will use single port mode and implicit Tx flow mode.
 
+*   ``--hairpin-map=Rx port id:Rx queue:Tx port id:Tx queue:queues number``
+
+    Set explicit hairpin configuration.
 
 Testpmd Multi-Process Command-line Options
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-- 
2.39.2

Re: [PATCH] testpmd: add hairpin-map parameter

[Stephen Hemminger]

On Thu, 28 Sep 2023 16:39:06 +0300
Gregory Etelson <getelson@nvidia.com> wrote:

> +static __rte_always_inline
> +char *parse_hairpin_map_entry(char *input, char **next)
> +{
> +	char *tail = strchr(input, ':');
> +
> +	if (!tail)
> +		return NULL;
> +	tail[0] = '\0';
> +	*next = tail + 1;
> +	return input;
> +}
> +

There is no reason to mark this as inline. It is not in fast path.
Let compiler decide.


> +	head = parse_hairpin_map_entry(next, &next);
> +	if (!head)
> +		goto err;
> +	map->rx_port = atoi(head);

Use strtoul() to allow checking for invalid number.
Seems like the parsing here is following the parsing code pattern
that could use strtok_r.

> +		/* Fail to setup rx queue, return */
> +		if (port->port_status == RTE_PORT_HANDLING)
> +			port->port_status = RTE_PORT_STOPPED;
> +		else
> +			fprintf(stderr,
> +				"Port %d can not be set back to stopped\n", pi);
> +		fprintf(stderr,
> +			"Port %d failed to configure hairpin on rxq %u.\n"
> +			"Peer port: %u peer txq: %u\n",
> +			pi, qi, peer_tx_port, i);

Minor nit, port should be printed with %u since it is unsigned.
Lots of testpmd code gets this wrong, and compiler doesn't care.


> +/* Configure the Rx and Tx hairpin queues for the selected port. */
> +static int
> +setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
> +{
> +	return !hairpin_multiport_mode ?
> +	       setup_legacy_hairpin_queus(pi, p_pi, cnt_pi) :
> +	       setup_mapped_harpin_queues(pi);
> +}

It is clearer to use an if statement in this case.
Also, better to write positive rather than negative logic.

	if (hairpin_multiport_mode)
		return setup_mapped_hairpin_queues(pi);
	else
		return setup_legacy_hairpin_queus(pi, p_pi, cnt_pi);

Re: [PATCH] testpmd: add hairpin-map parameter

[Etelson, Gregory]

Hello Stephen,

>> +static __rte_always_inline
>> +char *parse_hairpin_map_entry(char *input, char **next)
>> +{
>> +     char *tail = strchr(input, ':');
>> +
>> +     if (!tail)
>> +             return NULL;
>> +     tail[0] = '\0';
>> +     *next = tail + 1;
>> +     return input;
>> +}
>> +
>
> There is no reason to mark this as inline. It is not in fast path.
> Let compiler decide.
>

There will be no parse_hairpin_map_entry() in the next patch version.

>
>> +     head = parse_hairpin_map_entry(next, &next);
>> +     if (!head)
>> +             goto err;
>> +     map->rx_port = atoi(head);
>
> Use strtoul() to allow checking for invalid number.
> Seems like the parsing here is following the parsing code pattern
> that could use strtok_r.

I switched to sscanf().

>
>> +             /* Fail to setup rx queue, return */
>> +             if (port->port_status == RTE_PORT_HANDLING)
>> +                     port->port_status = RTE_PORT_STOPPED;
>> +             else
>> +                     fprintf(stderr,
>> +                             "Port %d can not be set back to stopped\n", pi);
>> +             fprintf(stderr,
>> +                     "Port %d failed to configure hairpin on rxq %u.\n"
>> +                     "Peer port: %u peer txq: %u\n",
>> +                     pi, qi, peer_tx_port, i);
>
> Minor nit, port should be printed with %u since it is unsigned.
> Lots of testpmd code gets this wrong, and compiler doesn't care.
>

fixed.

>
>> +/* Configure the Rx and Tx hairpin queues for the selected port. */
>> +static int
>> +setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
>> +{
>> +     return !hairpin_multiport_mode ?
>> +            setup_legacy_hairpin_queus(pi, p_pi, cnt_pi) :
>> +            setup_mapped_harpin_queues(pi);
>> +}
>
> It is clearer to use an if statement in this case.
> Also, better to write positive rather than negative logic.
>
>        if (hairpin_multiport_mode)
>                return setup_mapped_hairpin_queues(pi);
>        else
>                return setup_legacy_hairpin_queus(pi, p_pi, cnt_pi);

fixed.

I'll post fixes in the next patch version.

Regards,
Gregory

[PATCH] testpmd: add hairpin-map parameter

[Gregory Etelson]

Testpmd hairpin implementation always sets the next valid port to
complete hairpin binding. That limits hairpin configuration options.

The new parameter allows explicit selection of Rx and Tx ports and
queues in hairpin configuration.
The new `hairpin-map` parameter is provided with 5 parameters,
separated by `:`

`--hairpin-map=Rx port id:Rx queue:Tx port id:Tx queue:queues number`

Testpmd operator can provide several `hairpin-map` parameters for
different hairpin maps.
Example:

dpdk-testpmd <EAL params> -- \
  <testpmd params> \
  --rxq=2 --txq=2 --hairpinq=2 --hairpin-mode=0x12 \
  --hairpin-map=0:2:1:2:1 \ # [1]
  --hairpin-map=0:3:2:2:3   # [2]

Hairpin map [1] binds Rx port 0, queue 2 with Tx port 1, queue 2.
Hairpin map [2] binds
  Rx port 0, queue 3 with Tx port 2, queue 2,
  Rx port 0, queue 4 with Tx port 2, queue 3,
  Rx port 0, queue 5 with Tx port 2, queue 4.

The new `hairpin-map` parameter is optional.
If omitted, testpmd will create "default" hairpin maps.

Signed-off-by: Gregory Etelson <getelson@nvidia.com>
---
 app/test-pmd/parameters.c             |  63 ++++++++
 app/test-pmd/testpmd.c                | 212 ++++++++++++++++++--------
 app/test-pmd/testpmd.h                |  18 +++
 doc/guides/testpmd_app_ug/run_app.rst |   3 +
 4 files changed, 230 insertions(+), 66 deletions(-)

diff --git a/app/test-pmd/parameters.c b/app/test-pmd/parameters.c
index a9ca58339d..675de81861 100644
--- a/app/test-pmd/parameters.c
+++ b/app/test-pmd/parameters.c
@@ -206,6 +206,12 @@ usage(char* progname)
 	printf("  --hairpin-mode=0xXX: bitmask set the hairpin port mode.\n"
 	       "    0x10 - explicit Tx rule, 0x02 - hairpin ports paired\n"
 	       "    0x01 - hairpin ports loop, 0x00 - hairpin port self\n");
+	printf("  --hairpin-map=rxpi:rxq:txpi:txq:n: hairpin map.\n"
+	       "    rxpi - Rx port index.\n"
+	       "    rxq  - Rx queue.\n"
+	       "    txpi - Tx port index.\n"
+	       "    txq  - Tx queue.\n"
+	       "    n    - hairpin queues number.\n");
 }
 
 #ifdef RTE_LIB_CMDLINE
@@ -588,6 +594,55 @@ parse_link_speed(int n)
 	return speed;
 }
 
+static __rte_always_inline
+char *parse_hairpin_map_entry(char *input, char **next)
+{
+	char *tail = strchr(input, ':');
+
+	if (!tail)
+		return NULL;
+	tail[0] = '\0';
+	*next = tail + 1;
+	return input;
+}
+
+static int
+parse_hairpin_map(char *hpmap)
+{
+	/*
+	 * Testpmd hairpin map format:
+	 * <Rx port id:First Rx queue:Tx port id:First Tx queue:queues number>
+	 */
+	char *head, *next = hpmap;
+	struct hairpin_map *map = calloc(1, sizeof(*map));
+
+	if (!map)
+		return -ENOMEM;
+
+	head = parse_hairpin_map_entry(next, &next);
+	if (!head)
+		goto err;
+	map->rx_port = atoi(head);
+	head = parse_hairpin_map_entry(next, &next);
+	if (!head)
+		goto err;
+	map->rxq_head = atoi(head);
+	head = parse_hairpin_map_entry(next, &next);
+	if (!head)
+		goto err;
+	map->tx_port = atoi(head);
+	head = parse_hairpin_map_entry(next, &next);
+	if (!head)
+		goto err;
+	map->txq_head = atoi(head);
+	map->qnum = atoi(next);
+	hairpin_add_multiport_map(map);
+	return 0;
+err:
+	free(map);
+	return -EINVAL;
+}
+
 void
 launch_args_parse(int argc, char** argv)
 {
@@ -663,6 +718,7 @@ launch_args_parse(int argc, char** argv)
 		{ "txd",			1, 0, 0 },
 		{ "hairpinq",			1, 0, 0 },
 		{ "hairpin-mode",		1, 0, 0 },
+		{ "hairpin-map",                1, 0, 0 },
 		{ "burst",			1, 0, 0 },
 		{ "flowgen-clones",		1, 0, 0 },
 		{ "flowgen-flows",		1, 0, 0 },
@@ -1111,6 +1167,13 @@ launch_args_parse(int argc, char** argv)
 				else
 					hairpin_mode = (uint32_t)n;
 			}
+			if (!strcmp(lgopts[opt_idx].name, "hairpin-map")) {
+				hairpin_multiport_mode = true;
+				ret = parse_hairpin_map(optarg);
+				if (ret)
+					rte_exit(EXIT_FAILURE, "invalid hairpin map\n");
+
+			}
 			if (!strcmp(lgopts[opt_idx].name, "burst")) {
 				n = atoi(optarg);
 				if (n == 0) {
diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c
index 938ca035d4..2ba1727c51 100644
--- a/app/test-pmd/testpmd.c
+++ b/app/test-pmd/testpmd.c
@@ -434,6 +434,16 @@ uint8_t clear_ptypes = true;
 /* Hairpin ports configuration mode. */
 uint32_t hairpin_mode;
 
+bool hairpin_multiport_mode = false;
+
+static LIST_HEAD(,hairpin_map) hairpin_map_head = LIST_HEAD_INITIALIZER();
+
+void
+hairpin_add_multiport_map(struct hairpin_map *map)
+{
+	LIST_INSERT_HEAD(&hairpin_map_head, map, entry);
+}
+
 /* Pretty printing of ethdev events */
 static const char * const eth_event_desc[] = {
 	[RTE_ETH_EVENT_UNKNOWN] = "unknown",
@@ -2677,28 +2687,107 @@ port_is_started(portid_t port_id)
 #define HAIRPIN_MODE_TX_LOCKED_MEMORY RTE_BIT32(16)
 #define HAIRPIN_MODE_TX_RTE_MEMORY RTE_BIT32(17)
 
-
-/* Configure the Rx and Tx hairpin queues for the selected port. */
 static int
-setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
+port_config_hairpin_rxq(portid_t pi, uint16_t peer_tx_port,
+			queueid_t rxq_head, queueid_t txq_head,
+			uint16_t qcount, uint32_t manual_bind)
 {
-	queueid_t qi;
+	int diag;
+	queueid_t i, qi;
+	uint32_t tx_explicit = !!(hairpin_mode & 0x10);
+	uint32_t force_mem = !!(hairpin_mode & HAIRPIN_MODE_RX_FORCE_MEMORY);
+	uint32_t locked_mem = !!(hairpin_mode & HAIRPIN_MODE_RX_LOCKED_MEMORY);
+	uint32_t rte_mem = !!(hairpin_mode & HAIRPIN_MODE_RX_RTE_MEMORY);
+	struct rte_port *port = &ports[pi];
 	struct rte_eth_hairpin_conf hairpin_conf = {
 		.peer_count = 1,
 	};
-	int i;
+
+	for (qi = rxq_head, i = 0; qi < rxq_head + qcount; qi++) {
+		hairpin_conf.peers[0].port = peer_tx_port;
+		hairpin_conf.peers[0].queue = i + txq_head;
+		hairpin_conf.manual_bind = manual_bind;
+		hairpin_conf.tx_explicit = tx_explicit;
+		hairpin_conf.force_memory = force_mem;
+		hairpin_conf.use_locked_device_memory = locked_mem;
+		hairpin_conf.use_rte_memory = rte_mem;
+		diag = rte_eth_rx_hairpin_queue_setup
+			(pi, qi, nb_rxd, &hairpin_conf);
+		i++;
+		if (diag == 0)
+			continue;
+
+		/* Fail to setup rx queue, return */
+		if (port->port_status == RTE_PORT_HANDLING)
+			port->port_status = RTE_PORT_STOPPED;
+		else
+			fprintf(stderr,
+				"Port %d can not be set back to stopped\n", pi);
+		fprintf(stderr,
+			"Port %d failed to configure hairpin on rxq %u.\n"
+			"Peer port: %u peer txq: %u\n",
+			pi, qi, peer_tx_port, i);
+		/* try to reconfigure queues next time */
+		port->need_reconfig_queues = 1;
+		return -1;
+	}
+	return 0;
+}
+
+static int
+port_config_hairpin_txq(portid_t pi, uint16_t peer_rx_port,
+			queueid_t rxq_head, queueid_t txq_head,
+			uint16_t qcount, uint32_t manual_bind)
+{
 	int diag;
+	queueid_t i, qi;
+	uint32_t tx_explicit = !!(hairpin_mode & 0x10);
+	uint32_t force_mem = !!(hairpin_mode & HAIRPIN_MODE_TX_FORCE_MEMORY);
+	uint32_t locked_mem = !!(hairpin_mode & HAIRPIN_MODE_TX_LOCKED_MEMORY);
+	uint32_t rte_mem = !!(hairpin_mode & HAIRPIN_MODE_TX_RTE_MEMORY);
 	struct rte_port *port = &ports[pi];
+	struct rte_eth_hairpin_conf hairpin_conf = {
+		.peer_count = 1,
+	};
+
+	for (qi = txq_head, i = 0; qi < txq_head + qcount; qi++) {
+		hairpin_conf.peers[0].port = peer_rx_port;
+		hairpin_conf.peers[0].queue = i + rxq_head;
+		hairpin_conf.manual_bind = manual_bind;
+		hairpin_conf.tx_explicit = tx_explicit;
+		hairpin_conf.force_memory = force_mem;
+		hairpin_conf.use_locked_device_memory = locked_mem;
+		hairpin_conf.use_rte_memory = rte_mem;
+		diag = rte_eth_tx_hairpin_queue_setup
+			(pi, qi, nb_txd, &hairpin_conf);
+		i++;
+		if (diag == 0)
+			continue;
+
+		/* Fail to setup rx queue, return */
+		if (port->port_status == RTE_PORT_HANDLING)
+			port->port_status = RTE_PORT_STOPPED;
+		else
+			fprintf(stderr,
+				"Port %d can not be set back to stopped\n", pi);
+		fprintf(stderr,
+			"Port %d failed to configure hairpin on txq %u.\n"
+			"Peer port: %u peer rxq: %u\n",
+			pi, qi, peer_rx_port, i);
+		/* try to reconfigure queues next time */
+		port->need_reconfig_queues = 1;
+		return -1;
+	}
+	return 0;
+}
+
+static int
+setup_legacy_hairpin_queus(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
+{
+	int diag;
 	uint16_t peer_rx_port = pi;
 	uint16_t peer_tx_port = pi;
 	uint32_t manual = 1;
-	uint32_t tx_exp = hairpin_mode & 0x10;
-	uint32_t rx_force_memory = hairpin_mode & HAIRPIN_MODE_RX_FORCE_MEMORY;
-	uint32_t rx_locked_memory = hairpin_mode & HAIRPIN_MODE_RX_LOCKED_MEMORY;
-	uint32_t rx_rte_memory = hairpin_mode & HAIRPIN_MODE_RX_RTE_MEMORY;
-	uint32_t tx_force_memory = hairpin_mode & HAIRPIN_MODE_TX_FORCE_MEMORY;
-	uint32_t tx_locked_memory = hairpin_mode & HAIRPIN_MODE_TX_LOCKED_MEMORY;
-	uint32_t tx_rte_memory = hairpin_mode & HAIRPIN_MODE_TX_RTE_MEMORY;
 
 	if (!(hairpin_mode & 0xf)) {
 		peer_rx_port = pi;
@@ -2706,10 +2795,10 @@ setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
 		manual = 0;
 	} else if (hairpin_mode & 0x1) {
 		peer_tx_port = rte_eth_find_next_owned_by(pi + 1,
-						       RTE_ETH_DEV_NO_OWNER);
+							  RTE_ETH_DEV_NO_OWNER);
 		if (peer_tx_port >= RTE_MAX_ETHPORTS)
 			peer_tx_port = rte_eth_find_next_owned_by(0,
-						RTE_ETH_DEV_NO_OWNER);
+								  RTE_ETH_DEV_NO_OWNER);
 		if (p_pi != RTE_MAX_ETHPORTS) {
 			peer_rx_port = p_pi;
 		} else {
@@ -2725,69 +2814,60 @@ setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
 			peer_rx_port = p_pi;
 		} else {
 			peer_rx_port = rte_eth_find_next_owned_by(pi + 1,
-						RTE_ETH_DEV_NO_OWNER);
+								  RTE_ETH_DEV_NO_OWNER);
 			if (peer_rx_port >= RTE_MAX_ETHPORTS)
 				peer_rx_port = pi;
 		}
 		peer_tx_port = peer_rx_port;
 		manual = 1;
 	}
+	diag = port_config_hairpin_txq(pi, peer_rx_port, nb_rxq, nb_txq,
+				       nb_hairpinq, manual);
+	if (diag)
+		return diag;
+	diag = port_config_hairpin_rxq(pi, peer_tx_port, nb_rxq, nb_txq,
+				       nb_hairpinq, manual);
+	if (diag)
+		return diag;
+	return 0;
+}
 
-	for (qi = nb_txq, i = 0; qi < nb_hairpinq + nb_txq; qi++) {
-		hairpin_conf.peers[0].port = peer_rx_port;
-		hairpin_conf.peers[0].queue = i + nb_rxq;
-		hairpin_conf.manual_bind = !!manual;
-		hairpin_conf.tx_explicit = !!tx_exp;
-		hairpin_conf.force_memory = !!tx_force_memory;
-		hairpin_conf.use_locked_device_memory = !!tx_locked_memory;
-		hairpin_conf.use_rte_memory = !!tx_rte_memory;
-		diag = rte_eth_tx_hairpin_queue_setup
-			(pi, qi, nb_txd, &hairpin_conf);
-		i++;
-		if (diag == 0)
-			continue;
-
-		/* Fail to setup rx queue, return */
-		if (port->port_status == RTE_PORT_HANDLING)
-			port->port_status = RTE_PORT_STOPPED;
-		else
-			fprintf(stderr,
-				"Port %d can not be set back to stopped\n", pi);
-		fprintf(stderr, "Fail to configure port %d hairpin queues\n",
-			pi);
-		/* try to reconfigure queues next time */
-		port->need_reconfig_queues = 1;
-		return -1;
-	}
-	for (qi = nb_rxq, i = 0; qi < nb_hairpinq + nb_rxq; qi++) {
-		hairpin_conf.peers[0].port = peer_tx_port;
-		hairpin_conf.peers[0].queue = i + nb_txq;
-		hairpin_conf.manual_bind = !!manual;
-		hairpin_conf.tx_explicit = !!tx_exp;
-		hairpin_conf.force_memory = !!rx_force_memory;
-		hairpin_conf.use_locked_device_memory = !!rx_locked_memory;
-		hairpin_conf.use_rte_memory = !!rx_rte_memory;
-		diag = rte_eth_rx_hairpin_queue_setup
-			(pi, qi, nb_rxd, &hairpin_conf);
-		i++;
-		if (diag == 0)
-			continue;
-
-		/* Fail to setup rx queue, return */
-		if (port->port_status == RTE_PORT_HANDLING)
-			port->port_status = RTE_PORT_STOPPED;
-		else
-			fprintf(stderr,
-				"Port %d can not be set back to stopped\n", pi);
-		fprintf(stderr, "Fail to configure port %d hairpin queues\n",
-			pi);
-		/* try to reconfigure queues next time */
-		port->need_reconfig_queues = 1;
-		return -1;
+static int
+setup_mapped_harpin_queues(portid_t pi)
+{
+	int ret = 0;
+	struct hairpin_map *map;
+
+	LIST_FOREACH(map, &hairpin_map_head, entry) {
+		if (map->rx_port == pi) {
+			ret = port_config_hairpin_rxq(pi, map->tx_port,
+						      map->rxq_head,
+						      map->txq_head,
+						      map->qnum, true);
+			if (ret)
+				return ret;
+		}
+		if (map->tx_port == pi) {
+			ret = port_config_hairpin_txq(pi, map->rx_port,
+						      map->rxq_head,
+						      map->txq_head,
+						      map->qnum, true);
+			if (ret)
+				return ret;
+		}
 	}
 	return 0;
 }
 
+/* Configure the Rx and Tx hairpin queues for the selected port. */
+static int
+setup_hairpin_queues(portid_t pi, portid_t p_pi, uint16_t cnt_pi)
+{
+	return !hairpin_multiport_mode ?
+	       setup_legacy_hairpin_queus(pi, p_pi, cnt_pi) :
+	       setup_mapped_harpin_queues(pi);
+}
+
 /* Configure the Rx with optional split. */
 int
 rx_queue_setup(uint16_t port_id, uint16_t rx_queue_id,
diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h
index f1df6a8faf..208e8e9514 100644
--- a/app/test-pmd/testpmd.h
+++ b/app/test-pmd/testpmd.h
@@ -125,6 +125,24 @@ enum noisy_fwd_mode {
 	NOISY_FWD_MODE_MAX,
 };
 
+struct hairpin_map {
+	LIST_ENTRY(hairpin_map) entry; /**< List entry. */
+	portid_t rx_port; /**< Hairpin Rx port ID. */
+	portid_t tx_port; /**< Hairpin Tx port ID. */
+	uint16_t rxq_head; /**< Hairpin Rx queue head. */
+	uint16_t txq_head; /**< Hairpin Tx queue head. */
+	uint16_t qnum; /**< Hairpin queues number. */
+};
+
+/**
+ * Command line arguments parser sets `hairpin_multiport_mode` to True
+ * if explicit hairpin map configuration mode was used.
+ */
+extern bool hairpin_multiport_mode;
+
+/** Hairpin maps list. */
+extern void hairpin_add_multiport_map(struct hairpin_map *map);
+
 /**
  * The data structure associated with RX and TX packet burst statistics
  * that are recorded for each forwarding stream.
diff --git a/doc/guides/testpmd_app_ug/run_app.rst b/doc/guides/testpmd_app_ug/run_app.rst
index 6e9c552e76..a202c98b4c 100644
--- a/doc/guides/testpmd_app_ug/run_app.rst
+++ b/doc/guides/testpmd_app_ug/run_app.rst
@@ -566,6 +566,9 @@ The command line options are:
 
     The default value is 0. Hairpin will use single port mode and implicit Tx flow mode.
 
+*   ``--hairpin-map=Rx port id:Rx queue:Tx port id:Tx queue:queues number``
+
+    Set explicit hairpin configuration.
 
 Testpmd Multi-Process Command-line Options
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-- 
2.39.2