[dpdk-dev] [PATCH] Fix librte_pmd_ring: connect primary and secondary ring with correct port.

[Masaru OKI <m-oki@stratosphere.co.jp>]

librte_pmd_ring provides created port and attached port.
Packet is received from attached port if packet is sent to created port.
So, packet is received from created port if packet is sent to attached port.
It must be need two rings such as "create to attach" and "attach to create".
But current implementation uses only one ring for rx/tx.
It causes incorrect result.
Fixed:
- Make ring both rx and tx
- Connect created (primary) ring and attached (secondary) ring
- Correct m->port like librte_pmd_pcap

Signed-off-by: Masaru OKI <m-oki@stratosphere.co.jp>
---
 lib/librte_pmd_ring/rte_eth_ring.c | 36 +++++++++++++++++++++++++++---------
 1 file changed, 27 insertions(+), 9 deletions(-)

diff --git a/lib/librte_pmd_ring/rte_eth_ring.c b/lib/librte_pmd_ring/rte_eth_ring.c
index 4f1b6ed..d926d00 100644
--- a/lib/librte_pmd_ring/rte_eth_ring.c
+++ b/lib/librte_pmd_ring/rte_eth_ring.c
@@ -54,6 +54,7 @@ struct ring_queue {
 	rte_atomic64_t rx_pkts;
 	rte_atomic64_t tx_pkts;
 	rte_atomic64_t err_pkts;
+	uint8_t in_port;
 };
 
 struct pmd_internals {
@@ -80,10 +81,14 @@ eth_ring_rx(void *q, struct rte_mbuf **bufs, uint16_t nb_bufs)
 	struct ring_queue *r = q;
 	const uint16_t nb_rx = (uint16_t)rte_ring_dequeue_burst(r->rng,
 			ptrs, nb_bufs);
+        uint16_t cnt;
 	if (r->rng->flags & RING_F_SC_DEQ)
 		r->rx_pkts.cnt += nb_rx;
 	else
 		rte_atomic64_add(&(r->rx_pkts), nb_rx);
+        for (cnt = 0; cnt < nb_rx; cnt++) {
+          bufs[cnt]->port = r->in_port;
+        }
 	return nb_rx;
 }
 
@@ -129,6 +134,8 @@ eth_rx_queue_setup(struct rte_eth_dev *dev,uint16_t rx_queue_id,
 {
 	struct pmd_internals *internals = dev->data->dev_private;
 	dev->data->rx_queues[rx_queue_id] = &internals->rx_ring_queues[rx_queue_id];
+	internals->rx_ring_queues[rx_queue_id].in_port = dev->data->port_id;
+
 	return 0;
 }
 
@@ -319,23 +326,34 @@ eth_dev_ring_create(const char *name, const unsigned numa_node,
 	/* rx and tx are so-called from point of view of first port.
 	 * They are inverted from the point of view of second port
 	 */
-	struct rte_ring *rxtx[RTE_PMD_RING_MAX_RX_RINGS];
+	struct rte_ring *rx[RTE_PMD_RING_MAX_RX_RINGS];
+	struct rte_ring *tx[RTE_PMD_RING_MAX_TX_RINGS];
 	unsigned i;
-	char rng_name[RTE_RING_NAMESIZE];
+	char rng_rxname[RTE_RING_NAMESIZE];
+	char rng_txname[RTE_RING_NAMESIZE];
 	unsigned num_rings = RTE_MIN(RTE_PMD_RING_MAX_RX_RINGS,
 			RTE_PMD_RING_MAX_TX_RINGS);
 
 	for (i = 0; i < num_rings; i++) {
-		snprintf(rng_name, sizeof(rng_name), "ETH_RXTX%u_%s", i, name);
-		rxtx[i] = (action == DEV_CREATE) ?
-				rte_ring_create(rng_name, 1024, numa_node,
-						RING_F_SP_ENQ|RING_F_SC_DEQ) :
-				rte_ring_lookup(rng_name);
-		if (rxtx[i] == NULL)
+		snprintf(rng_rxname, sizeof(rng_rxname),
+                         "ETH_RX%u_%s", i, name);
+		snprintf(rng_txname, sizeof(rng_txname),
+                         "ETH_TX%u_%s", i, name);
+		rx[i] = (action == DEV_CREATE) ?
+                        rte_ring_create(rng_rxname, 1024, numa_node,
+                                        RING_F_SP_ENQ|RING_F_SC_DEQ) :
+                        rte_ring_lookup(rng_txname);
+		if (rx[i] == NULL)
+			return -1;
+		tx[i] = (action == DEV_CREATE) ?
+                        rte_ring_create(rng_txname, 1024, numa_node,
+                                        RING_F_SP_ENQ|RING_F_SC_DEQ) :
+                        rte_ring_lookup(rng_rxname);
+		if (tx[i] == NULL)
 			return -1;
 	}
 
-	if (rte_eth_from_rings(name, rxtx, num_rings, rxtx, num_rings, numa_node))
+	if (rte_eth_from_rings(name, rx, num_rings, tx, num_rings, numa_node))
 		return -1;
 
 	return 0;
-- 
1.9.1