[PATCH v7 1/2] ethdev: add Tx queue mapping of aggregated ports

[Jiawei Wang <jiaweiw@nvidia.com>]

When multiple ports are aggregated into a single DPDK port,
(example: Linux bonding, DPDK bonding, failsafe, etc.),
we want to know which port use for Tx via a queue.

This patch introduces the new ethdev API
rte_eth_dev_map_aggr_tx_affinity(), it's used to map a Tx queue
with an aggregated port of the DPDK port (specified with port_id),
The affinity is the number of the aggregated port.
Value 0 means no affinity and traffic could be routed to any
aggregated port, this is the default current behavior.

The maximum number of affinity is given by rte_eth_dev_count_aggr_ports().

Add the trace point for ethdev rte_eth_dev_count_aggr_ports()
and rte_eth_dev_map_aggr_tx_affinity() functions.

Add the testpmd command line:
testpmd> port config (port_id) txq (queue_id) affinity (value)

For example, there're two physical ports connected to
a single DPDK port (port id 0), and affinity 1 stood for
the first physical port and affinity 2 stood for the second
physical port.
Use the below commands to config tx phy affinity for per Tx Queue:
        port config 0 txq 0 affinity 1
        port config 0 txq 1 affinity 1
        port config 0 txq 2 affinity 2
        port config 0 txq 3 affinity 2

These commands config the Tx Queue index 0 and Tx Queue index 1 with
phy affinity 1, uses Tx Queue 0 or Tx Queue 1 send packets,
these packets will be sent from the first physical port, and similar
with the second physical port if sending packets with Tx Queue 2
or Tx Queue 3.

Signed-off-by: Jiawei Wang <jiaweiw@nvidia.com>
Acked-by: Thomas Monjalon <thomas@monjalon.net>
---
 app/test-pmd/cmdline.c                      | 92 +++++++++++++++++++++
 doc/guides/rel_notes/release_23_03.rst      |  7 ++
 doc/guides/testpmd_app_ug/testpmd_funcs.rst | 14 ++++
 lib/ethdev/ethdev_driver.h                  | 37 +++++++++
 lib/ethdev/ethdev_trace.h                   | 17 ++++
 lib/ethdev/ethdev_trace_points.c            |  6 ++
 lib/ethdev/rte_ethdev.c                     | 72 ++++++++++++++++
 lib/ethdev/rte_ethdev.h                     | 49 +++++++++++
 lib/ethdev/version.map                      |  2 +
 9 files changed, 296 insertions(+)

diff --git a/app/test-pmd/cmdline.c b/app/test-pmd/cmdline.c
index bb7ff2b449..151f356224 100644
--- a/app/test-pmd/cmdline.c
+++ b/app/test-pmd/cmdline.c
@@ -776,6 +776,10 @@ static void cmd_help_long_parsed(void *parsed_result,
 
 			"port cleanup (port_id) txq (queue_id) (free_cnt)\n"
 			"    Cleanup txq mbufs for a specific Tx queue\n\n"
+
+			"port config (port_id) txq (queue_id) affinity (value)\n"
+			"    Map a Tx queue with an aggregated port "
+			"of the DPDK port\n\n"
 		);
 	}
 
@@ -12636,6 +12640,93 @@ static cmdline_parse_inst_t cmd_show_port_flow_transfer_proxy = {
 	}
 };
 
+/* *** configure port txq affinity value *** */
+struct cmd_config_tx_affinity_map {
+	cmdline_fixed_string_t port;
+	cmdline_fixed_string_t config;
+	portid_t portid;
+	cmdline_fixed_string_t txq;
+	uint16_t qid;
+	cmdline_fixed_string_t affinity;
+	uint8_t value;
+};
+
+static void
+cmd_config_tx_affinity_map_parsed(void *parsed_result,
+				  __rte_unused struct cmdline *cl,
+				  __rte_unused void *data)
+{
+	struct cmd_config_tx_affinity_map *res = parsed_result;
+	int ret;
+
+	if (port_id_is_invalid(res->portid, ENABLED_WARN))
+		return;
+
+	if (res->portid == (portid_t)RTE_PORT_ALL) {
+		printf("Invalid port id\n");
+		return;
+	}
+
+	if (strcmp(res->txq, "txq")) {
+		printf("Unknown parameter\n");
+		return;
+	}
+	if (tx_queue_id_is_invalid(res->qid))
+		return;
+
+	ret = rte_eth_dev_count_aggr_ports(res->portid);
+	if (ret < 0) {
+		printf("Failed to count the aggregated ports: (%s)\n",
+			strerror(-ret));
+		return;
+	}
+
+	ret = rte_eth_dev_map_aggr_tx_affinity(res->portid, res->qid, res->value);
+	if (ret != 0) {
+		printf("Failed to map tx queue with an aggregated port: %s\n",
+			rte_strerror(-ret));
+		return;
+	}
+}
+
+cmdline_parse_token_string_t cmd_config_tx_affinity_map_port =
+	TOKEN_STRING_INITIALIZER(struct cmd_config_tx_affinity_map,
+				 port, "port");
+cmdline_parse_token_string_t cmd_config_tx_affinity_map_config =
+	TOKEN_STRING_INITIALIZER(struct cmd_config_tx_affinity_map,
+				 config, "config");
+cmdline_parse_token_num_t cmd_config_tx_affinity_map_portid =
+	TOKEN_NUM_INITIALIZER(struct cmd_config_tx_affinity_map,
+				 portid, RTE_UINT16);
+cmdline_parse_token_string_t cmd_config_tx_affinity_map_txq =
+	TOKEN_STRING_INITIALIZER(struct cmd_config_tx_affinity_map,
+				 txq, "txq");
+cmdline_parse_token_num_t cmd_config_tx_affinity_map_qid =
+	TOKEN_NUM_INITIALIZER(struct cmd_config_tx_affinity_map,
+			      qid, RTE_UINT16);
+cmdline_parse_token_string_t cmd_config_tx_affinity_map_affinity =
+	TOKEN_STRING_INITIALIZER(struct cmd_config_tx_affinity_map,
+				 affinity, "affinity");
+cmdline_parse_token_num_t cmd_config_tx_affinity_map_value =
+	TOKEN_NUM_INITIALIZER(struct cmd_config_tx_affinity_map,
+			      value, RTE_UINT8);
+
+static cmdline_parse_inst_t cmd_config_tx_affinity_map = {
+	.f = cmd_config_tx_affinity_map_parsed,
+	.data = (void *)0,
+	.help_str = "port config <port_id> txq <queue_id> affinity <value>",
+	.tokens = {
+		(void *)&cmd_config_tx_affinity_map_port,
+		(void *)&cmd_config_tx_affinity_map_config,
+		(void *)&cmd_config_tx_affinity_map_portid,
+		(void *)&cmd_config_tx_affinity_map_txq,
+		(void *)&cmd_config_tx_affinity_map_qid,
+		(void *)&cmd_config_tx_affinity_map_affinity,
+		(void *)&cmd_config_tx_affinity_map_value,
+		NULL,
+	},
+};
+
 /* ******************************************************************************** */
 
 /* list of instructions */
@@ -12869,6 +12960,7 @@ static cmdline_parse_ctx_t builtin_ctx[] = {
 	(cmdline_parse_inst_t *)&cmd_show_port_cman_capa,
 	(cmdline_parse_inst_t *)&cmd_show_port_cman_config,
 	(cmdline_parse_inst_t *)&cmd_set_port_cman_config,
+	(cmdline_parse_inst_t *)&cmd_config_tx_affinity_map,
 	NULL,
 };
 
diff --git a/doc/guides/rel_notes/release_23_03.rst b/doc/guides/rel_notes/release_23_03.rst
index 6ecc22db3a..97a9a56bc7 100644
--- a/doc/guides/rel_notes/release_23_03.rst
+++ b/doc/guides/rel_notes/release_23_03.rst
@@ -68,6 +68,13 @@ New Features
   * Applications can register a callback at startup via
     ``rte_lcore_register_usage_cb()`` to provide lcore usage information.
 
+* **Added support for Tx queue mapping with an aggregated port.**
+
+  * Introduced new function ``rte_eth_dev_count_aggr_ports()``
+    to get the number of aggregated ports.
+  * Introduced new function ``rte_eth_dev_map_aggr_tx_affinity()``
+    to map a Tx queue with an aggregated port of the DPDK port.
+
 * **Added flow matching of IPv6 routing extension.**
 
   Added ``RTE_FLOW_ITEM_TYPE_IPV6_ROUTING_EXT``
diff --git a/doc/guides/testpmd_app_ug/testpmd_funcs.rst b/doc/guides/testpmd_app_ug/testpmd_funcs.rst
index 357adb09d7..1d3c372601 100644
--- a/doc/guides/testpmd_app_ug/testpmd_funcs.rst
+++ b/doc/guides/testpmd_app_ug/testpmd_funcs.rst
@@ -1612,6 +1612,20 @@ Enable or disable a per queue Tx offloading only on a specific Tx queue::
 
 This command should be run when the port is stopped, or else it will fail.
 
+config per queue Tx affinity mapping
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Map a Tx queue with an aggregated port of the DPDK port (specified with port_id)::
+
+   testpmd> port (port_id) txq (queue_id) affinity (value)
+
+* ``affinity``: the number of the aggregated port.
+                When multiple ports are aggregated into a single one,
+                it allows to choose which port to use for Tx via a queue.
+
+This command should be run when the port is stopped, otherwise it fails.
+
+
 Config VXLAN Encap outer layers
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
diff --git a/lib/ethdev/ethdev_driver.h b/lib/ethdev/ethdev_driver.h
index 6a550cfc83..2c9d615fb5 100644
--- a/lib/ethdev/ethdev_driver.h
+++ b/lib/ethdev/ethdev_driver.h
@@ -1171,6 +1171,38 @@ typedef int (*eth_tx_descriptor_dump_t)(const struct rte_eth_dev *dev,
 					uint16_t queue_id, uint16_t offset,
 					uint16_t num, FILE *file);
 
+/**
+ * @internal
+ * Get the number of aggregated ports.
+ *
+ * @param dev
+ *   Port (ethdev) handle.
+ *
+ * @return
+ *   Negative errno value on error, 0 or positive on success.
+ *
+ * @retval >=0
+ *   The number of aggregated port if success.
+ */
+typedef int (*eth_count_aggr_ports_t)(struct rte_eth_dev *dev);
+
+/**
+ * @internal
+ * Map a Tx queue with an aggregated port of the DPDK port.
+ *
+ * @param dev
+ *   Port (ethdev) handle.
+ * @param tx_queue_id
+ *   The index of the transmit queue used in rte_eth_tx_burst().
+ * @param affinity
+ *   The number of the aggregated port.
+ *
+ * @return
+ *   Negative on error, 0 on success.
+ */
+typedef int (*eth_map_aggr_tx_affinity_t)(struct rte_eth_dev *dev, uint16_t tx_queue_id,
+					  uint8_t affinity);
+
 /**
  * @internal A structure containing the functions exported by an Ethernet driver.
  */
@@ -1403,6 +1435,11 @@ struct eth_dev_ops {
 	eth_cman_config_set_t cman_config_set;
 	/** Retrieve congestion management configuration */
 	eth_cman_config_get_t cman_config_get;
+
+	/** Get the number of aggregated ports */
+	eth_count_aggr_ports_t count_aggr_ports;
+	/** Map a Tx queue with an aggregated port of the DPDK port */
+	eth_map_aggr_tx_affinity_t map_aggr_tx_affinity;
 };
 
 /**
diff --git a/lib/ethdev/ethdev_trace.h b/lib/ethdev/ethdev_trace.h
index 8932ac33e0..53d1a71ff0 100644
--- a/lib/ethdev/ethdev_trace.h
+++ b/lib/ethdev/ethdev_trace.h
@@ -1385,6 +1385,23 @@ RTE_TRACE_POINT(
 	rte_trace_point_emit_int(ret);
 )
 
+RTE_TRACE_POINT(
+	rte_eth_trace_count_aggr_ports,
+	RTE_TRACE_POINT_ARGS(uint16_t port_id, int ret),
+	rte_trace_point_emit_u16(port_id);
+	rte_trace_point_emit_int(ret);
+)
+
+RTE_TRACE_POINT(
+	rte_eth_trace_map_aggr_tx_affinity,
+	RTE_TRACE_POINT_ARGS(uint16_t port_id, uint16_t tx_queue_id,
+			     uint8_t affinity, int ret),
+	rte_trace_point_emit_u16(port_id);
+	rte_trace_point_emit_u16(tx_queue_id);
+	rte_trace_point_emit_u8(affinity);
+	rte_trace_point_emit_int(ret);
+)
+
 RTE_TRACE_POINT(
 	rte_flow_trace_dynf_metadata_register,
 	RTE_TRACE_POINT_ARGS(int offset, uint64_t flag),
diff --git a/lib/ethdev/ethdev_trace_points.c b/lib/ethdev/ethdev_trace_points.c
index 34d12e2859..61010cae56 100644
--- a/lib/ethdev/ethdev_trace_points.c
+++ b/lib/ethdev/ethdev_trace_points.c
@@ -475,6 +475,12 @@ RTE_TRACE_POINT_REGISTER(rte_eth_trace_cman_config_set,
 RTE_TRACE_POINT_REGISTER(rte_eth_trace_cman_config_get,
 	lib.ethdev.cman_config_get)
 
+RTE_TRACE_POINT_REGISTER(rte_eth_trace_count_aggr_ports,
+	lib.ethdev.count_aggr_ports)
+
+RTE_TRACE_POINT_REGISTER(rte_eth_trace_map_aggr_tx_affinity,
+	lib.ethdev.map_aggr_tx_affinity)
+
 RTE_TRACE_POINT_REGISTER(rte_flow_trace_copy,
 	lib.ethdev.flow.copy)
 
diff --git a/lib/ethdev/rte_ethdev.c b/lib/ethdev/rte_ethdev.c
index 055c46082b..36f4222fe8 100644
--- a/lib/ethdev/rte_ethdev.c
+++ b/lib/ethdev/rte_ethdev.c
@@ -6946,6 +6946,78 @@ rte_eth_buffer_split_get_supported_hdr_ptypes(uint16_t port_id, uint32_t *ptypes
 	return j;
 }
 
+int rte_eth_dev_count_aggr_ports(uint16_t port_id)
+{
+	struct rte_eth_dev *dev;
+	int ret;
+
+	RTE_ETH_VALID_PORTID_OR_ERR_RET(port_id, -ENODEV);
+	dev = &rte_eth_devices[port_id];
+
+	if (*dev->dev_ops->count_aggr_ports == NULL)
+		return 0;
+	ret = eth_err(port_id, (*dev->dev_ops->count_aggr_ports)(dev));
+
+	rte_eth_trace_count_aggr_ports(port_id, ret);
+
+	return ret;
+}
+
+int rte_eth_dev_map_aggr_tx_affinity(uint16_t port_id, uint16_t tx_queue_id,
+				     uint8_t affinity)
+{
+	struct rte_eth_dev *dev;
+	int aggr_ports;
+	int ret;
+
+	RTE_ETH_VALID_PORTID_OR_ERR_RET(port_id, -ENODEV);
+	dev = &rte_eth_devices[port_id];
+
+	if (tx_queue_id >= dev->data->nb_tx_queues) {
+		RTE_ETHDEV_LOG(ERR, "Invalid Tx queue_id=%u\n", tx_queue_id);
+		return -EINVAL;
+	}
+
+	if (*dev->dev_ops->map_aggr_tx_affinity == NULL)
+		return -ENOTSUP;
+
+	if (dev->data->dev_configured == 0) {
+		RTE_ETHDEV_LOG(ERR,
+			"Port %u must be configured before Tx affinity mapping\n",
+			port_id);
+		return -EINVAL;
+	}
+
+	if (dev->data->dev_started) {
+		RTE_ETHDEV_LOG(ERR,
+			"Port %u must be stopped to allow configuration\n",
+			port_id);
+		return -EBUSY;
+	}
+
+	aggr_ports = rte_eth_dev_count_aggr_ports(port_id);
+	if (aggr_ports == 0) {
+		RTE_ETHDEV_LOG(ERR,
+			"Port %u number of aggregated ports is 0 which is invalid\n",
+			port_id);
+		return -ENOTSUP;
+	}
+
+	if (affinity > aggr_ports) {
+		RTE_ETHDEV_LOG(ERR,
+			"Port %u map invalid affinity %u exceeds the maximum number %u\n",
+			port_id, affinity, aggr_ports);
+		return -EINVAL;
+	}
+
+	ret = eth_err(port_id, (*dev->dev_ops->map_aggr_tx_affinity)(dev,
+				tx_queue_id, affinity));
+
+	rte_eth_trace_map_aggr_tx_affinity(port_id, tx_queue_id, affinity, ret);
+
+	return ret;
+}
+
 RTE_LOG_REGISTER_DEFAULT(rte_eth_dev_logtype, INFO);
 
 RTE_INIT(ethdev_init_telemetry)
diff --git a/lib/ethdev/rte_ethdev.h b/lib/ethdev/rte_ethdev.h
index c129ca1eaf..049641d57c 100644
--- a/lib/ethdev/rte_ethdev.h
+++ b/lib/ethdev/rte_ethdev.h
@@ -2589,6 +2589,55 @@ int rte_eth_hairpin_bind(uint16_t tx_port, uint16_t rx_port);
 __rte_experimental
 int rte_eth_hairpin_unbind(uint16_t tx_port, uint16_t rx_port);
 
+/**
+ * @warning
+ * @b EXPERIMENTAL: this API may change without prior notice.
+ *
+ *  Get the number of aggregated ports of the DPDK port (specified with port_id).
+ *  It is used when multiple ports are aggregated into a single one.
+ *
+ *  For the regular physical port doesn't have aggregated ports,
+ *  the number of aggregated ports is reported as 0.
+ *
+ * @param port_id
+ *   The port identifier of the Ethernet device.
+ * @return
+ *   - (>=0) the number of aggregated port if success.
+ */
+__rte_experimental
+int rte_eth_dev_count_aggr_ports(uint16_t port_id);
+
+/**
+ * @warning
+ * @b EXPERIMENTAL: this API may change without prior notice.
+ *
+ *  Map a Tx queue with an aggregated port of the DPDK port (specified with port_id).
+ *  When multiple ports are aggregated into a single one,
+ *  it allows to choose which port to use for Tx via a queue.
+ *
+ *  The application should use rte_eth_dev_map_aggr_tx_affinity()
+ *  after rte_eth_dev_configure(), rte_eth_tx_queue_setup(), and
+ *  before rte_eth_dev_start().
+ *
+ * @param port_id
+ *   The identifier of the port used in rte_eth_tx_burst().
+ * @param tx_queue_id
+ *   The index of the transmit queue used in rte_eth_tx_burst().
+ *   The value must be in the range [0, nb_tx_queue - 1] previously supplied
+ *   to rte_eth_dev_configure().
+ * @param affinity
+ *   The number of the aggregated port.
+ *   Value 0 means no affinity and traffic could be routed to any aggregated port.
+ *   The first aggregated port is number 1 and so on.
+ *   The maximum number is given by rte_eth_dev_count_aggr_ports().
+ *
+ * @return
+ *   Zero if successful. Non-zero otherwise.
+ */
+__rte_experimental
+int rte_eth_dev_map_aggr_tx_affinity(uint16_t port_id, uint16_t tx_queue_id,
+				     uint8_t affinity);
+
 /**
  * Return the NUMA socket to which an Ethernet device is connected
  *
diff --git a/lib/ethdev/version.map b/lib/ethdev/version.map
index 40d43035eb..72c443e996 100644
--- a/lib/ethdev/version.map
+++ b/lib/ethdev/version.map
@@ -300,6 +300,8 @@ EXPERIMENTAL {
 	rte_mtr_meter_profile_get;
 
 	# added in 23.03
+	rte_eth_dev_count_aggr_ports;
+	rte_eth_dev_map_aggr_tx_affinity;
 	rte_flow_action_handle_query_update;
 	rte_flow_async_action_handle_query_update;
 	rte_flow_async_create_by_index;
-- 
2.18.1