examples/packet_ordering: fix Rx with reorder mode disabled

[ upstream commit 7ba49dc729937ea97642a615e9b08f33919b94f4 ]

The packet_ordering example works in two modes (opt via --disable-reorder):
  - When reorder is enabled: rx_thread - N*worker_thread - send_thread
  - When reorder is disabled: rx_thread - N*worker_thread - tx_thread
N parallel worker_thread(s) generate out-of-order packets.

When reorder is enabled, send_thread uses sequence number generated in
rx_thread (L459) to enforce packet ordering. Otherwise rx_thread just
sends any packet it receives.

rx_thread writes sequence number into a dynamic field, which is only
registered by calling rte_reorder_create() (Line 741) when reorder is
enabled. However, rx_thread marks sequence number onto each packet no
matter whether reorder is enabled, overwriting the leading bytes in packet
mbufs when reorder is disabled, resulting in segfaults when PMD tries to
DMA packets.

`if (!disable_reorder_flag) {...}` is added in rx_thread to fix the bug.
The test is inlined by the compiler to prevent any performance loss.

Signed-off-by: Qian Hao <[email protected]>
Acked-by: Volodymyr Fialko <[email protected]>

.mailmap

@@ -1088,6 +1088,7 @@ Przemyslaw Czesnowicz <[email protected]>
 Przemyslaw Patynowski <[email protected]>
 Przemyslaw Zegan <[email protected]>
 Pu Xu <[email protected]>
+Qian Hao <[email protected]>
 Qian Xu <[email protected]>
 Qiao Liu <[email protected]>
 Qi Fu <[email protected]>

examples/packet_ordering/main.c

@@ -4,6 +4,7 @@
 
 #include <signal.h>
 #include <getopt.h>
+#include <stdbool.h>
 
 #include <rte_eal.h>
 #include <rte_common.h>
@@ -426,8 +427,8 @@ int_handler(int sig_num)
  * The mbufs are then passed to the worker threads via the rx_to_workers
  * ring.
  */
-static int
-rx_thread(struct rte_ring *ring_out)
+static __rte_always_inline int
+rx_thread(struct rte_ring *ring_out, bool disable_reorder_flag)
 {
 	uint32_t seqn = 0;
 	uint16_t i, ret = 0;
@@ -453,9 +454,11 @@ rx_thread(struct rte_ring *ring_out)
 				}
 				app_stats.rx.rx_pkts += nb_rx_pkts;
 
-				/* mark sequence number */
-				for (i = 0; i < nb_rx_pkts; )
-					*rte_reorder_seqn(pkts[i++]) = seqn++;
+				/* mark sequence number if reorder is enabled */
+				if (!disable_reorder_flag) {
+					for (i = 0; i < nb_rx_pkts;)
+						*rte_reorder_seqn(pkts[i++]) = seqn++;
+				}
 
 				/* enqueue to rx_to_workers ring */
 				ret = rte_ring_enqueue_burst(ring_out,
@@ -472,6 +475,18 @@ rx_thread(struct rte_ring *ring_out)
 	return 0;
 }
 
+static __rte_noinline int
+rx_thread_reorder(struct rte_ring *ring_out)
+{
+	return rx_thread(ring_out, false);
+}
+
+static __rte_noinline int
+rx_thread_reorder_disabled(struct rte_ring *ring_out)
+{
+	return rx_thread(ring_out, true);
+}
+
 /**
  * This thread takes bursts of packets from the rx_to_workers ring and
  * Changes the input port value to output port value. And feds it to
@@ -771,8 +786,11 @@ main(int argc, char **argv)
 				(void *)&send_args, last_lcore_id);
 	}
 
-	/* Start rx_thread() on the main core */
-	rx_thread(rx_to_workers);
+	/* Start rx_thread_xxx() on the main core */
+	if (disable_reorder)
+		rx_thread_reorder_disabled(rx_to_workers);
+	else
+		rx_thread_reorder(rx_to_workers);
 
 	RTE_LCORE_FOREACH_WORKER(lcore_id) {
 		if (rte_eal_wait_lcore(lcore_id) < 0)