[dpdk-dev] [PATCH v3 8/8] app/testpmd: support shared Rx queue forwarding

[Xueming Li <xuemingl@nvidia.com>]

By enabling shared Rx queue, received packets come from all member ports
in same shared Rx queue.

This patch adds a common forwarding function for shared Rx queue, groups
source forwarding stream by looking into local streams on current lcore
with packet source port(mbuf->port) and queue, then invokes callback to
handle received packets for each source stream.

Signed-off-by: Xueming Li <xuemingl@nvidia.com>
---
 app/test-pmd/ieee1588fwd.c | 30 +++++++++++------
 app/test-pmd/testpmd.c     | 69 ++++++++++++++++++++++++++++++++++++++
 app/test-pmd/testpmd.h     |  9 ++++-
 3 files changed, 97 insertions(+), 11 deletions(-)

diff --git a/app/test-pmd/ieee1588fwd.c b/app/test-pmd/ieee1588fwd.c
index 034f238c34..0151d6de74 100644
--- a/app/test-pmd/ieee1588fwd.c
+++ b/app/test-pmd/ieee1588fwd.c
@@ -90,23 +90,17 @@ port_ieee1588_tx_timestamp_check(portid_t pi)
 }
 
 static void
-ieee1588_packet_fwd(struct fwd_stream *fs)
+ieee1588_fwd_stream(struct fwd_stream *fs, uint16_t nb_rx,
+		struct rte_mbuf **pkt)
 {
-	struct rte_mbuf  *mb;
+	struct rte_mbuf *mb = (*pkt);
 	struct rte_ether_hdr *eth_hdr;
 	struct rte_ether_addr addr;
 	struct ptpv2_msg *ptp_hdr;
 	uint16_t eth_type;
 	uint32_t timesync_index;
 
-	/*
-	 * Receive 1 packet at a time.
-	 */
-	if (rte_eth_rx_burst(fs->rx_port, fs->rx_queue, &mb, 1) == 0)
-		return;
-
-	fs->rx_packets += 1;
-
+	RTE_SET_USED(nb_rx);
 	/*
 	 * Check that the received packet is a PTP packet that was detected
 	 * by the hardware.
@@ -198,6 +192,22 @@ ieee1588_packet_fwd(struct fwd_stream *fs)
 	port_ieee1588_tx_timestamp_check(fs->rx_port);
 }
 
+/*
+ * Wrapper of real fwd ingine.
+ */
+static void
+ieee1588_packet_fwd(struct fwd_stream *fs)
+{
+	struct rte_mbuf *mb;
+
+	if (rte_eth_rx_burst(fs->rx_port, fs->rx_queue, &mb, 1) == 0)
+		return;
+	if (unlikely(fs->rxq_share > 0))
+		forward_shared_rxq(fs, 1, &mb, ieee1588_fwd_stream);
+	else
+		ieee1588_fwd_stream(fs, 1, &mb);
+}
+
 static void
 port_ieee1588_fwd_begin(portid_t pi)
 {
diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c
index cab4b36b04..1d82397831 100644
--- a/app/test-pmd/testpmd.c
+++ b/app/test-pmd/testpmd.c
@@ -2106,6 +2106,75 @@ flush_fwd_rx_queues(void)
 	}
 }
 
+/**
+ * Get packet source stream by source port and queue.
+ * All streams of same shared Rx queue locates on same core.
+ */
+static struct fwd_stream *
+forward_stream_get(struct fwd_stream *fs, uint16_t port)
+{
+	streamid_t sm_id;
+	struct fwd_lcore *fc;
+	struct fwd_stream **fsm;
+	streamid_t nb_fs;
+
+	fc = fs->lcore;
+	fsm = &fwd_streams[fc->stream_idx];
+	nb_fs = fc->stream_nb;
+	for (sm_id = 0; sm_id < nb_fs; sm_id++) {
+		if (fsm[sm_id]->rx_port == port &&
+		    fsm[sm_id]->rx_queue == fs->rx_queue)
+			return fsm[sm_id];
+	}
+	return NULL;
+}
+
+/**
+ * Forward packet by source port and queue.
+ */
+static void
+forward_by_port(struct fwd_stream *src_fs, uint16_t port, uint16_t nb_rx,
+		struct rte_mbuf **pkts, packet_fwd_cb fwd)
+{
+	struct fwd_stream *fs = forward_stream_get(src_fs, port);
+
+	if (fs != NULL) {
+		fs->rx_packets += nb_rx;
+		fwd(fs, nb_rx, pkts);
+	} else {
+		/* Source stream not found, drop all packets. */
+		src_fs->fwd_dropped += nb_rx;
+		while (nb_rx > 0)
+			rte_pktmbuf_free(pkts[--nb_rx]);
+	}
+}
+
+/**
+ * Forward packets from shared Rx queue.
+ *
+ * Source port of packets are identified by mbuf->port.
+ */
+void
+forward_shared_rxq(struct fwd_stream *fs, uint16_t nb_rx,
+		   struct rte_mbuf **pkts_burst, packet_fwd_cb fwd)
+{
+	uint16_t i, nb_fs_rx = 1, port;
+
+	/* Locate real source fs according to mbuf->port. */
+	for (i = 0; i < nb_rx; ++i) {
+		rte_prefetch0(pkts_burst[i + 1]);
+		port = pkts_burst[i]->port;
+		if (i + 1 == nb_rx || pkts_burst[i + 1]->port != port) {
+			/* Forward packets with same source port. */
+			forward_by_port(fs, port, nb_fs_rx,
+					&pkts_burst[i + 1 - nb_fs_rx], fwd);
+			nb_fs_rx = 1;
+		} else {
+			nb_fs_rx++;
+		}
+	}
+}
+
 static void
 run_pkt_fwd_on_lcore(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
 {
diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h
index 3b8796a7a5..7869f61f74 100644
--- a/app/test-pmd/testpmd.h
+++ b/app/test-pmd/testpmd.h
@@ -276,6 +276,8 @@ struct fwd_lcore {
 typedef void (*port_fwd_begin_t)(portid_t pi);
 typedef void (*port_fwd_end_t)(portid_t pi);
 typedef void (*packet_fwd_t)(struct fwd_stream *fs);
+typedef void (*packet_fwd_cb)(struct fwd_stream *fs, uint16_t nb_rx,
+			      struct rte_mbuf **pkts);
 
 struct fwd_engine {
 	const char       *fwd_mode_name; /**< Forwarding mode name. */
@@ -910,6 +912,8 @@ char *list_pkt_forwarding_modes(void);
 char *list_pkt_forwarding_retry_modes(void);
 void set_pkt_forwarding_mode(const char *fwd_mode);
 void start_packet_forwarding(int with_tx_first);
+void forward_shared_rxq(struct fwd_stream *fs, uint16_t nb_rx,
+			struct rte_mbuf **pkts_burst, packet_fwd_cb fwd);
 void fwd_stats_display(void);
 void fwd_stats_reset(void);
 void stop_packet_forwarding(void);
@@ -1046,7 +1050,10 @@ pkt_burst_fwd(struct fwd_stream *fs)                            \
 	if (unlikely(nb_rx == 0))                               \
 		return;                                         \
 	fs->rx_packets += nb_rx;                                \
-	cb(fs, nb_rx, pkts_burst);                              \
+	if (fs->rxq_share)                                      \
+		forward_shared_rxq(fs, nb_rx, pkts_burst, cb);  \
+	else                                                    \
+		cb(fs, nb_rx, pkts_burst);                      \
 	get_end_cycles(fs, start_tsc);                          \
 }
 
-- 
2.33.0