[PATCH v4-S3 61/65] event/dpaa2: add ordered schedule type support

[Prashant Gupta]

From: Jun Yang <jun.yang at nxp.com>

The dpaa2 eventdev supported only PARALLEL and ATOMIC schedule types.
Each schedule type uses its own DPCI Tx queue (dpci_txqs[] indexed by
event->sched_type) attached with a dedicated dequeue callback, and the
ORDERED slot was left unimplemented (marked TODO).

Add RTE_SCHED_TYPE_ORDERED support:

  - Attach the ordered DPCI queue (dpci_txqs[RTE_SCHED_TYPE_ORDERED])
    with a new dequeue callback dpaa2_eventdev_process_order() that
    saves the ORP context via dpaa2_dq_order_save().
  - Advertise RTE_EVENT_DEV_CAP_NONSEQ_MODE and accept
    RTE_SCHED_TYPE_ORDERED in queue setup.
  - Handle the ORP flag in the enqueue path via
    dpaa2_eq_set_order_desc().

Also optimize the Tx adapter enqueue path to use rte_eth_tx_burst()
directly when all packets target the same queue and no ordered/atomic
(ORP or DCA) handling is required, falling back to
dpaa2_dev_tx_multi_txq_ordered() otherwise. The DPCI attach error logs
are reworded from "RX" to "qp" for clarity.

Signed-off-by: Jun Yang <jun.yang at nxp.com>
---
 drivers/event/dpaa2/dpaa2_eventdev.c | 84 +++++++++++++++++++++-------
 1 file changed, 65 insertions(+), 19 deletions(-)

diff --git a/drivers/event/dpaa2/dpaa2_eventdev.c b/drivers/event/dpaa2/dpaa2_eventdev.c
index 8ad9fe7d30..a03b04a057 100644
--- a/drivers/event/dpaa2/dpaa2_eventdev.c
+++ b/drivers/event/dpaa2/dpaa2_eventdev.c
@@ -178,18 +178,29 @@ dpaa2_eventdev_enqueue_burst(void *port, const struct rte_event ev[],
 
 			/* Prepare enqueue descriptor */
 			qbman_eq_desc_clear(&eqdesc[loop]);
-			qbman_eq_desc_set_fq(&eqdesc[loop], dpci_txq[loop]->fqid);
-			qbman_eq_desc_set_no_orp(&eqdesc[loop], 0);
-			qbman_eq_desc_set_response(&eqdesc[loop], 0, 0);
 			flags[loop] = 0;
 			dqrr_held[loop] = 0;
 			dqrr_size[loop] = 0;
 
-			if (event->sched_type == RTE_SCHED_TYPE_ATOMIC &&
-				dpaa2_mbuf_seq_is_flag_set(event->mbuf, QBMAN_ENQUEUE_FLAG_DCA)) {
-				dpaa2_eq_atomic_consume(event->mbuf, dqrr_size[loop],
-					dqrr_held[loop], dpio_dev, &flags[loop],
-					&eqdesc[loop], atomic);
+			if (event->sched_type == RTE_SCHED_TYPE_ORDERED &&
+				dpaa2_mbuf_seq_is_flag_set(event->mbuf, DPAA2_ENQUEUE_FLAG_ORP)) {
+				ret = dpaa2_eq_set_order_desc(dpci_txq[loop], event->mbuf,
+					&eqdesc[loop], true);
+				if (ret) {
+					rte_mempool_put(dpci_txq[loop]->env_pool, ev_tx[loop]);
+					goto send_partial;
+				}
+			} else {
+				if (event->sched_type == RTE_SCHED_TYPE_ATOMIC &&
+					dpaa2_mbuf_seq_is_flag_set(event->mbuf,
+					QBMAN_ENQUEUE_FLAG_DCA)) {
+					dpaa2_eq_atomic_consume(event->mbuf, dqrr_size[loop],
+						dqrr_held[loop], dpio_dev, &flags[loop],
+						&eqdesc[loop], atomic);
+				}
+				qbman_eq_desc_set_fq(&eqdesc[loop], dpci_txq[loop]->fqid);
+				qbman_eq_desc_set_no_orp(&eqdesc[loop], 0);
+				qbman_eq_desc_set_response(&eqdesc[loop], 0, 0);
 			}
 
 			memset(&fd_arr[loop], 0, sizeof(struct qbman_fd));
@@ -258,16 +269,23 @@ static void dpaa2_eventdev_dequeue_wait(struct dpaa2_dpio_dev *dpio_dev,
 	DPAA2_PMD_DP_DEBUG("%s: poll return(%d)", __func__, ret);
 }
 
-static void dpaa2_eventdev_process_parallel(struct dpaa2_dpio_dev *dpio_dev,
-	const struct qbman_fd *fd, const struct qbman_result *dq,
+static inline void
+dpaa2_eventdev_rx_get_event(const struct qbman_fd *fd,
 	struct dpaa2_queue *rxq, struct rte_event *ev)
 {
 	struct rte_event *rx_ev = (void *)DPAA2_GET_FD_ADDR(fd);
-	struct qbman_swp *swp = dpio_dev->sw_portal;
 
 	rte_memcpy(ev, rx_ev, sizeof(struct rte_event));
 	rte_mempool_put(rxq->env_pool, rx_ev);
+}
 
+static void dpaa2_eventdev_process_parallel(struct dpaa2_dpio_dev *dpio_dev,
+	const struct qbman_fd *fd, const struct qbman_result *dq,
+	struct dpaa2_queue *rxq, struct rte_event *ev)
+{
+	struct qbman_swp *swp = dpio_dev->sw_portal;
+
+	dpaa2_eventdev_rx_get_event(fd, rxq, ev);
 	qbman_swp_dqrr_consume(swp, dq);
 }
 
@@ -275,13 +293,21 @@ static void dpaa2_eventdev_process_atomic(struct dpaa2_dpio_dev *dpio_dev,
 	const struct qbman_fd *fd, const struct qbman_result *dq,
 	struct dpaa2_queue *rxq, struct rte_event *ev)
 {
-	struct rte_event *rx_ev = (void *)DPAA2_GET_FD_ADDR(fd);
-
-	rte_memcpy(ev, rx_ev, sizeof(struct rte_event));
-	rte_mempool_put(rxq->env_pool, rx_ev);
+	dpaa2_eventdev_rx_get_event(fd, rxq, ev);
 	dpaa2_dq_atomic_save(dq, ev->mbuf, dpio_dev);
 }
 
+static void dpaa2_eventdev_process_order(struct dpaa2_dpio_dev *dpio_dev,
+	const struct qbman_fd *fd, const struct qbman_result *dq,
+	struct dpaa2_queue *rxq, struct rte_event *ev)
+{
+	struct qbman_swp *swp = dpio_dev->sw_portal;
+
+	dpaa2_eventdev_rx_get_event(fd, rxq, ev);
+	dpaa2_dq_order_save(dq, ev->mbuf);
+	qbman_swp_dqrr_consume(swp, dq);
+}
+
 static uint16_t
 dpaa2_eventdev_dequeue_burst(void *port, struct rte_event ev[],
 	uint16_t nb_events, uint64_t timeout_ticks)
@@ -439,6 +465,7 @@ dpaa2_eventdev_info_get(struct rte_eventdev *dev,
 		RTE_EVENT_DEV_CAP_RUNTIME_PORT_LINK |
 		RTE_EVENT_DEV_CAP_MULTIPLE_QUEUE_PORT |
 		RTE_EVENT_DEV_CAP_QUEUE_ALL_TYPES |
+		RTE_EVENT_DEV_CAP_NONSEQ_MODE |
 		RTE_EVENT_DEV_CAP_CARRY_FLOW_ID |
 		RTE_EVENT_DEV_CAP_MAINTENANCE_FREE;
 	dev_info->max_profiles_per_port = 1;
@@ -653,6 +680,7 @@ dpaa2_eventdev_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
 	switch (queue_conf->schedule_type) {
 	case RTE_SCHED_TYPE_PARALLEL:
 	case RTE_SCHED_TYPE_ATOMIC:
+	case RTE_SCHED_TYPE_ORDERED:
 		break;
 	default:
 		DPAA2_EVENTDEV_ERR("Schedule type(%d) is not supported.",
@@ -1169,6 +1197,7 @@ dpaa2_eventdev_txa_enqueue(void *port,
 	struct dpaa2_queue *txq[MAX_TX_RING_SLOTS];
 	struct rte_mbuf *m[MAX_TX_RING_SLOTS];
 	uint16_t sent = 0, burst, i, port_id, qid;
+	int same_q, order;
 
 	RTE_SET_USED(port);
 
@@ -1176,15 +1205,25 @@ dpaa2_eventdev_txa_enqueue(void *port,
 	 * on-stack burst array, so transmit in ring-sized chunks.
 	 */
 	while (sent < nb_events) {
+		same_q = true;
+		order = false;
 		burst = RTE_MIN(nb_events - sent, (uint16_t)RTE_DIM(m));
 		for (i = 0; i < burst; i++) {
 			m[i] = ev[sent + i].mbuf;
 			port_id = m[i]->port;
 			qid = rte_event_eth_tx_adapter_txq_get(m[i]);
 			txq[i] = rte_eth_devices[port_id].data->tx_queues[qid];
+			if (same_q && i > 0 && txq[i] != txq[i - 1])
+				same_q = false;
+			if (!order && dpaa2_mbuf_seq_is_flag_set(m[i],
+				(DPAA2_ENQUEUE_FLAG_ORP | QBMAN_ENQUEUE_FLAG_DCA)))
+				order = true;
 		}
 
-		i = dpaa2_dev_tx_multi_txq_ordered(txq, m, burst);
+		if (same_q && !order && burst > 0)
+			i = rte_eth_tx_burst(port_id, qid, m, burst);
+		else
+			i = dpaa2_dev_tx_multi_txq_ordered(txq, m, burst);
 		sent += i;
 		if (i < burst)
 			break;
@@ -1283,7 +1322,7 @@ dpaa2_eventdev_create(const char *name, struct rte_vdev_device *vdev)
 			dpaa2_eventdev_process_parallel,
 			&evq_info->dpci_txqs[RTE_SCHED_TYPE_PARALLEL]);
 		if (ret) {
-			DPAA2_EVENTDEV_ERR("DPCI attach parallel RX failed: err(%d)", ret);
+			DPAA2_EVENTDEV_ERR("DPCI attach parallel qp failed: err(%d)", ret);
 			goto err_break;
 		}
 
@@ -1292,11 +1331,18 @@ dpaa2_eventdev_create(const char *name, struct rte_vdev_device *vdev)
 			dpaa2_eventdev_process_atomic,
 			&evq_info->dpci_txqs[RTE_SCHED_TYPE_ATOMIC]);
 		if (ret) {
-			DPAA2_EVENTDEV_ERR("DPCI attach atomic RX failed: err(%d)", ret);
+			DPAA2_EVENTDEV_ERR("DPCI attach atomic qp failed: err(%d)", ret);
 			goto err_break;
 		}
 
-		/** TO DO: order schedule.*/
+		ret = rte_dpaa2_dpci_link_attach(dpci_dev, DPCI_DEST_DPCON,
+			dpcon_dev->dpcon_id, 0,
+			dpaa2_eventdev_process_order,
+			&evq_info->dpci_txqs[RTE_SCHED_TYPE_ORDERED]);
+		if (ret) {
+			DPAA2_EVENTDEV_ERR("DPCI attach order qp failed: err(%d)", ret);
+			goto err_break;
+		}
 
 		evq_info->valid = true;
 		evq_info->event_queue_id = priv->max_event_queues;
-- 
2.43.0