[PATCH 4/4] net/mlx5: sync stop/start of datapath with secondary process

[Yongseok Koh <yskoh@mellanox.com>]

Rx/Tx burst function pointers are stored in the rte_eth_dev structure,
which is local to a process. Even though primary process replaces the
function pointers, secondary will not run the new ones. With rte_mp APIs,
primary can easily broadcast a request to stop/start the datapath of
secondary processes.

Signed-off-by: Yongseok Koh <yskoh@mellanox.com>
---
 drivers/net/mlx5/mlx5.c         |   5 ++
 drivers/net/mlx5/mlx5.h         |   6 ++
 drivers/net/mlx5/mlx5_mp.c      | 144 ++++++++++++++++++++++++++++++++++++++++
 drivers/net/mlx5/mlx5_rxtx.c    |   2 +
 drivers/net/mlx5/mlx5_trigger.c |   5 ++
 5 files changed, 162 insertions(+)

diff --git a/drivers/net/mlx5/mlx5.c b/drivers/net/mlx5/mlx5.c
index ea8fd55ee6..dd9106296c 100644
--- a/drivers/net/mlx5/mlx5.c
+++ b/drivers/net/mlx5/mlx5.c
@@ -305,6 +305,9 @@ mlx5_dev_close(struct rte_eth_dev *dev)
 	/* Prevent crashes when queues are still in use. */
 	dev->rx_pkt_burst = removed_rx_burst;
 	dev->tx_pkt_burst = removed_tx_burst;
+	rte_wmb();
+	/* Disable datapath on secondary process. */
+	mlx5_mp_req_stop_rxtx(dev);
 	if (priv->rxqs != NULL) {
 		/* XXX race condition if mlx5_rx_burst() is still running. */
 		usleep(1000);
@@ -803,6 +806,7 @@ mlx5_init_once(void)
 	case RTE_PROC_SECONDARY:
 		if (ld->init_done)
 			break;
+		mlx5_mp_init_secondary();
 		ret = mlx5_uar_init_secondary();
 		if (ret)
 			goto error;
@@ -823,6 +827,7 @@ mlx5_init_once(void)
 		break;
 	case RTE_PROC_SECONDARY:
 		mlx5_uar_uninit_secondary();
+		mlx5_mp_uninit_secondary();
 		break;
 	default:
 		break;
diff --git a/drivers/net/mlx5/mlx5.h b/drivers/net/mlx5/mlx5.h
index 4b606190f0..9c042bc3da 100644
--- a/drivers/net/mlx5/mlx5.h
+++ b/drivers/net/mlx5/mlx5.h
@@ -59,6 +59,8 @@ enum {
 /* Request types for IPC. */
 enum mlx5_mp_req_type {
 	MLX5_MP_REQ_VERBS_CMD_FD = 1,
+	MLX5_MP_REQ_START_RXTX,
+	MLX5_MP_REQ_STOP_RXTX,
 };
 
 /* Pameters for IPC. */
@@ -425,9 +427,13 @@ int mlx5_flow_create_drop_queue(struct rte_eth_dev *dev);
 void mlx5_flow_delete_drop_queue(struct rte_eth_dev *dev);
 
 /* mlx5_mp.c */
+void mlx5_mp_req_start_rxtx(struct rte_eth_dev *dev);
+void mlx5_mp_req_stop_rxtx(struct rte_eth_dev *dev);
 int mlx5_mp_req_verbs_cmd_fd(struct rte_eth_dev *dev);
 void mlx5_mp_init_primary(void);
 void mlx5_mp_uninit_primary(void);
+void mlx5_mp_init_secondary(void);
+void mlx5_mp_uninit_secondary(void);
 
 /* mlx5_nl.c */
 
diff --git a/drivers/net/mlx5/mlx5_mp.c b/drivers/net/mlx5/mlx5_mp.c
index 5f18c8f5c1..623dfb8097 100644
--- a/drivers/net/mlx5/mlx5_mp.c
+++ b/drivers/net/mlx5/mlx5_mp.c
@@ -12,6 +12,7 @@
 #include <rte_string_fns.h>
 
 #include "mlx5.h"
+#include "mlx5_rxtx.h"
 #include "mlx5_utils.h"
 
 /**
@@ -78,6 +79,129 @@ mp_primary_handle(const struct rte_mp_msg *mp_msg, const void *peer)
 }
 
 /**
+ * IPC message handler of a secondary process.
+ *
+ * @param[in] dev
+ *   Pointer to Ethernet structure.
+ * @param[in] peer
+ *   Pointer to the peer socket path.
+ *
+ * @return
+ *   0 on success, a negative errno value otherwise and rte_errno is set.
+ */
+static int
+mp_secondary_handle(const struct rte_mp_msg *mp_msg, const void *peer)
+{
+	struct rte_mp_msg mp_res;
+	struct mlx5_mp_param *res = (struct mlx5_mp_param *)mp_res.param;
+	const struct mlx5_mp_param *param =
+		(const struct mlx5_mp_param *)mp_msg->param;
+	struct rte_eth_dev *dev = &rte_eth_devices[param->port_id];
+	int ret = 0;
+
+	assert(rte_eal_process_type() == RTE_PROC_SECONDARY);
+	switch (param->type) {
+	case MLX5_MP_REQ_START_RXTX:
+		DRV_LOG(INFO, "port %u starting datapath", dev->data->port_id);
+		rte_mb();
+		dev->rx_pkt_burst = mlx5_select_rx_function(dev);
+		dev->tx_pkt_burst = mlx5_select_tx_function(dev);
+		mp_init_msg(dev, &mp_res, param->type);
+		res->result = 0;
+		ret = rte_mp_reply(&mp_res, peer);
+		break;
+	case MLX5_MP_REQ_STOP_RXTX:
+		DRV_LOG(INFO, "port %u stopping datapath", dev->data->port_id);
+		dev->rx_pkt_burst = removed_rx_burst;
+		dev->tx_pkt_burst = removed_tx_burst;
+		rte_mb();
+		mp_init_msg(dev, &mp_res, param->type);
+		res->result = 0;
+		ret = rte_mp_reply(&mp_res, peer);
+		break;
+	default:
+		rte_errno = EINVAL;
+		DRV_LOG(ERR, "port %u invalid mp request type",
+			dev->data->port_id);
+		return -rte_errno;
+	}
+	return ret;
+}
+
+/**
+ * Broadcast request of stopping/starting data-path to secondary processes.
+ *
+ * @param[in] dev
+ *   Pointer to Ethernet structure.
+ * @param[in] type
+ *   Request type.
+ */
+static void
+mp_req_on_rxtx(struct rte_eth_dev *dev, enum mlx5_mp_req_type type)
+{
+	struct rte_mp_msg mp_req;
+	struct rte_mp_msg *mp_res;
+	struct rte_mp_reply mp_rep;
+	struct mlx5_mp_param *res __rte_unused;
+	struct timespec ts = {.tv_sec = 5, .tv_nsec = 0};
+	int ret;
+
+	assert(rte_eal_process_type() == RTE_PROC_PRIMARY);
+	if (!mlx5_shared_data->secondary_cnt)
+		return;
+	if (type != MLX5_MP_REQ_START_RXTX && type != MLX5_MP_REQ_STOP_RXTX) {
+		DRV_LOG(ERR, "port %u unknown request (req_type %d)",
+			dev->data->port_id, type);
+		return;
+	}
+	mp_init_msg(dev, &mp_req, type);
+	ret = rte_mp_request_sync(&mp_req, &mp_rep, &ts);
+	if (ret) {
+		DRV_LOG(ERR,
+			"port %u failed to request stop/start Rx/Tx (%d)",
+			dev->data->port_id, type);
+		goto exit;
+	}
+	if (mp_rep.nb_sent != mp_rep.nb_received) {
+		DRV_LOG(ERR,
+			"port %u not all secondaries responded (req_type %d)",
+			dev->data->port_id, type);
+		goto exit;
+	}
+	mp_res = &mp_rep.msgs[0];
+	res = (struct mlx5_mp_param *)mp_res->param;
+	assert(!res->result);
+exit:
+	free(mp_rep.msgs);
+}
+
+/**
+ * Broadcast request of starting data-path to secondary processes. The request
+ * is synchronous.
+ *
+ * @param[in] dev
+ *   Pointer to Ethernet structure.
+ */
+void
+mlx5_mp_req_start_rxtx(struct rte_eth_dev *dev)
+{
+	mp_req_on_rxtx(dev, MLX5_MP_REQ_START_RXTX);
+}
+
+/**
+ * Broadcast request of stopping data-path to secondary processes. The request
+ * is synchronous.
+ *
+ * @param[in] dev
+ *   Pointer to Ethernet structure.
+ */
+void
+mlx5_mp_req_stop_rxtx(struct rte_eth_dev *dev)
+{
+	mp_req_on_rxtx(dev, MLX5_MP_REQ_STOP_RXTX);
+}
+
+/**
  * Request Verbs command file descriptor for mmap to the primary process.
  *
  * @param[in] dev
@@ -137,3 +261,23 @@ mlx5_mp_uninit_primary(void)
 	assert(rte_eal_process_type() == RTE_PROC_PRIMARY);
 	rte_mp_action_unregister(MLX5_MP_NAME);
 }
+
+/**
+ * Initialize by secondary process.
+ */
+void
+mlx5_mp_init_secondary(void)
+{
+	assert(rte_eal_process_type() == RTE_PROC_SECONDARY);
+	rte_mp_action_register(MLX5_MP_NAME, mp_secondary_handle);
+}
+
+/**
+ * Un-initialize by secondary process.
+ */
+void
+mlx5_mp_uninit_secondary(void)
+{
+	assert(rte_eal_process_type() == RTE_PROC_SECONDARY);
+	rte_mp_action_unregister(MLX5_MP_NAME);
+}
diff --git a/drivers/net/mlx5/mlx5_rxtx.c b/drivers/net/mlx5/mlx5_rxtx.c
index baa4079c14..b6946eff4a 100644
--- a/drivers/net/mlx5/mlx5_rxtx.c
+++ b/drivers/net/mlx5/mlx5_rxtx.c
@@ -2372,6 +2372,7 @@ removed_tx_burst(void *dpdk_txq __rte_unused,
 		 struct rte_mbuf **pkts __rte_unused,
 		 uint16_t pkts_n __rte_unused)
 {
+	rte_mb();
 	return 0;
 }
 
@@ -2396,6 +2397,7 @@ removed_rx_burst(void *dpdk_txq __rte_unused,
 		 struct rte_mbuf **pkts __rte_unused,
 		 uint16_t pkts_n __rte_unused)
 {
+	rte_mb();
 	return 0;
 }
 
diff --git a/drivers/net/mlx5/mlx5_trigger.c b/drivers/net/mlx5/mlx5_trigger.c
index 2137bdc461..7c9ff921ab 100644
--- a/drivers/net/mlx5/mlx5_trigger.c
+++ b/drivers/net/mlx5/mlx5_trigger.c
@@ -194,8 +194,11 @@ mlx5_dev_start(struct rte_eth_dev *dev)
 			dev->data->port_id);
 		goto error;
 	}
+	rte_wmb();
 	dev->tx_pkt_burst = mlx5_select_tx_function(dev);
 	dev->rx_pkt_burst = mlx5_select_rx_function(dev);
+	/* Enable datapath on secondary process. */
+	mlx5_mp_req_start_rxtx(dev);
 	mlx5_dev_interrupt_handler_install(dev);
 	return 0;
 error:
@@ -228,6 +231,8 @@ mlx5_dev_stop(struct rte_eth_dev *dev)
 	dev->rx_pkt_burst = removed_rx_burst;
 	dev->tx_pkt_burst = removed_tx_burst;
 	rte_wmb();
+	/* Disable datapath on secondary process. */
+	mlx5_mp_req_stop_rxtx(dev);
 	usleep(1000 * priv->rxqs_n);
 	DRV_LOG(DEBUG, "port %u stopping device", dev->data->port_id);
 	mlx5_flow_stop(dev, &priv->flows);
-- 
2.11.0