From: Jun Yang <[email protected]> The "same destination" Tx adapter enqueue path assumed every event in a burst targeted the same ethdev port and Tx queue, reading the port and queue id only from the first mbuf (ev[0]) and pushing the whole burst with rte_eth_tx_burst() to that single destination.
This assumption does not always hold: an event burst may carry mbufs bound for different ports or queues. In that case, packets destined for other ports/queues were sent to the destination of the first mbuf, causing misdirected transmission. Make the generic enqueue handler, which resolves the port and Tx queue per mbuf and transmits through dpaa2_dev_tx_multi_txq_ordered(), the sole implementation. The same-destination variant now simply forwards to it, so both callbacks handle mixed destinations correctly. Signed-off-by: Jun Yang <[email protected]> --- drivers/event/dpaa2/dpaa2_eventdev.c | 45 ++++++++-------------------- 1 file changed, 12 insertions(+), 33 deletions(-) diff --git a/drivers/event/dpaa2/dpaa2_eventdev.c b/drivers/event/dpaa2/dpaa2_eventdev.c index 03016d2a12..78a069b456 100644 --- a/drivers/event/dpaa2/dpaa2_eventdev.c +++ b/drivers/event/dpaa2/dpaa2_eventdev.c @@ -983,28 +983,28 @@ dpaa2_eventdev_tx_adapter_caps(const struct rte_eventdev *dev, } static uint16_t -dpaa2_eventdev_txa_enqueue_same_dest(void *port, +dpaa2_eventdev_txa_enqueue(void *port, struct rte_event ev[], uint16_t nb_events) { - struct rte_mbuf *m[MAX_TX_RING_SLOTS], *m0; - uint16_t sent = 0, burst, i, port_id; - uint8_t qid; + void *txq[MAX_TX_RING_SLOTS]; + struct rte_mbuf *m[MAX_TX_RING_SLOTS]; + uint16_t sent = 0, burst, i, port_id, qid; RTE_SET_USED(port); - m0 = ev[0].mbuf; - qid = rte_event_eth_tx_adapter_txq_get(m0); - port_id = m0->port; - /* The advertised enqueue depth is the EQCR size, which may exceed the * on-stack burst array, so transmit in ring-sized chunks. */ while (sent < nb_events) { burst = RTE_MIN(nb_events - sent, (uint16_t)RTE_DIM(m)); - for (i = 0; i < burst; i++) + 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]; + } - i = rte_eth_tx_burst(port_id, qid, m, burst); + i = dpaa2_dev_tx_multi_txq_ordered(txq, m, burst); sent += i; if (i < burst) break; @@ -1014,31 +1014,10 @@ dpaa2_eventdev_txa_enqueue_same_dest(void *port, } static uint16_t -dpaa2_eventdev_txa_enqueue(void *port, +dpaa2_eventdev_txa_enqueue_same_dest(void *port, struct rte_event ev[], uint16_t nb_events) { - void *txq[MAX_TX_RING_SLOTS]; - struct rte_mbuf *m[MAX_TX_RING_SLOTS]; - uint16_t sent = 0, burst, i; - uint8_t qid; - - RTE_SET_USED(port); - - while (sent < nb_events) { - burst = RTE_MIN(nb_events - sent, (uint16_t)RTE_DIM(m)); - for (i = 0; i < burst; i++) { - m[i] = ev[sent + i].mbuf; - qid = rte_event_eth_tx_adapter_txq_get(m[i]); - txq[i] = rte_eth_devices[m[i]->port].data->tx_queues[qid]; - } - - i = dpaa2_dev_tx_multi_txq_ordered(txq, m, burst); - sent += i; - if (i < burst) - break; - } - - return sent; + return dpaa2_eventdev_txa_enqueue(port, ev, nb_events); } static struct eventdev_ops dpaa2_eventdev_ops = { -- 2.43.0
