[dpdk-dev,RFC,5/7] testpmd: support multi-pthread mode

diff --git a/app/test-pmd/cmdline.c b/app/test-pmd/cmdline.c
index 882a5a2..9c2322c 100644
--- a/app/test-pmd/cmdline.c
+++ b/app/test-pmd/cmdline.c
@@ -8697,6 +8697,45 @@  cmdline_parse_inst_t cmd_set_flow_director_flex_payload = {
 	},
 };
 
+/* *** SET SP/MP *** */
+struct cmd_set_mp_result {
+	cmdline_fixed_string_t set;
+	cmdline_fixed_string_t mp;
+	cmdline_fixed_string_t mode;
+};
+
+static void cmd_set_mp_parsed(void *parsed_result,
+			      __attribute__((unused)) struct cmdline *cl,
+			      __attribute__((unused)) void *data)
+{
+	struct cmd_set_mp_result *res = parsed_result;
+
+	if (!strcmp(res->mode, "on"))
+		set_multi_thread(1);
+	else
+		set_multi_thread(0);
+}
+
+cmdline_parse_token_string_t cmd_setmp_set =
+	TOKEN_STRING_INITIALIZER(struct cmd_set_mp_result, set, "set");
+cmdline_parse_token_string_t cmd_setmp_mp =
+	TOKEN_STRING_INITIALIZER(struct cmd_set_mp_result, mp, "mp");
+cmdline_parse_token_string_t cmd_setmp_mode =
+	TOKEN_STRING_INITIALIZER(struct cmd_set_mp_result, mode,
+				 "on#off");
+
+cmdline_parse_inst_t cmd_set_mp = {
+	.f = cmd_set_mp_parsed,
+	.data = (void *)1,
+	.help_str = "set mp on|off: turn on/off multi-thread per lcore",
+	.tokens = {
+		(void *)&cmd_setmp_set,
+		(void *)&cmd_setmp_mp,
+		(void *)&cmd_setmp_mode,
+		NULL,
+	},
+};
+
 /* ******************************************************************************** */
 
 /* list of instructions */
@@ -8836,6 +8875,7 @@  cmdline_parse_ctx_t main_ctx[] = {
 	(cmdline_parse_inst_t *)&cmd_flush_flow_director,
 	(cmdline_parse_inst_t *)&cmd_set_flow_director_flex_mask,
 	(cmdline_parse_inst_t *)&cmd_set_flow_director_flex_payload,
+	(cmdline_parse_inst_t *)&cmd_set_mp,
 	NULL,
 };
 
@@ -8906,3 +8946,4 @@  bypass_is_supported(portid_t port_id)
 	}
 }
 #endif
+
diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c
index 8c69756..7ff9d0c 100644
--- a/app/test-pmd/testpmd.c
+++ b/app/test-pmd/testpmd.c
@@ -943,7 +943,7 @@  flush_fwd_rx_queues(void)
 }
 
 static void
-run_pkt_fwd_on_lcore(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
+run_pkt_fwd_on_lcore_sp(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
 {
 	struct fwd_stream **fsm;
 	streamid_t nb_fs;
@@ -957,6 +957,70 @@  run_pkt_fwd_on_lcore(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
 	} while (! fc->stopped);
 }
 
+struct work_arg {
+	struct fwd_lcore *fc;
+	struct fwd_stream *fs;
+	packet_fwd_t pkt_fwd;
+};
+
+static void* work(void *arg)
+{
+	struct work_arg *warg = (struct work_arg *)arg;
+	struct fwd_stream *fs = warg->fs;
+	struct fwd_lcore *fc = warg->fc;
+	packet_fwd_t pkt_fwd = warg->pkt_fwd;
+
+	do {
+		(*pkt_fwd)(fs);
+	} while (! fc->stopped);
+
+	return NULL;
+}
+
+static void
+run_pkt_fwd_on_lcore_mp(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
+{
+	struct fwd_stream **fsm;
+	streamid_t nb_fs;
+	streamid_t sm_id;
+	streamid_t i;
+	struct work_arg *work_arg = NULL;
+	pthread_t *tids = NULL;
+
+	fsm = &fwd_streams[fc->stream_idx];
+	nb_fs = fc->stream_nb;
+	tids = calloc(nb_fs, nb_fs * sizeof(*tids));
+	if (!tids)
+		goto exit;
+
+	work_arg = calloc(nb_fs, nb_fs * sizeof(*work_arg));
+	if (!work_arg)
+		goto exit;
+
+	for (sm_id = 0; sm_id < nb_fs; sm_id++) {
+		work_arg[sm_id].pkt_fwd = pkt_fwd;
+		work_arg[sm_id].fc = fc;
+		work_arg[sm_id].fs = fsm[sm_id];
+		if (rte_pthread_create(&tids[sm_id], work,
+				       &work_arg[sm_id]) < 0) {
+			printf("create pthread fail on stream %u\n",
+				sm_id);
+			break;
+		}
+	}
+
+	for (i = 0; i < sm_id; i++)
+		(void)pthread_join(tids[i], NULL);
+
+exit:
+	free(tids);
+	free(work_arg);
+}
+
+void
+(*run_pkt_fwd_on_lcore)(struct fwd_lcore *fc, packet_fwd_t pkt_fwd) = 
+	run_pkt_fwd_on_lcore_sp;
+
 static int
 start_pkt_forward_on_core(void *fwd_arg)
 {
@@ -965,6 +1029,24 @@  start_pkt_forward_on_core(void *fwd_arg)
 	return 0;
 }
 
+int set_multi_thread(int on)
+{
+	unsigned int lc_id;
+
+	for (lc_id = 0; lc_id < cur_fwd_config.nb_fwd_lcores; lc_id++)
+		if (fwd_lcores[lc_id]->stopped != 1) {
+			printf("Make sure stop forwarding first\n");
+			return -1;
+		}
+
+	if (on)
+		run_pkt_fwd_on_lcore = run_pkt_fwd_on_lcore_mp;
+	else
+		run_pkt_fwd_on_lcore = run_pkt_fwd_on_lcore_sp;
+
+	return 0;
+}
+
 /*
  * Run the TXONLY packet forwarding engine to send a single burst of packets.
  * Used to start communication flows in network loopback test configurations.
diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h
index f8b0740..3658e0f 100644
--- a/app/test-pmd/testpmd.h
+++ b/app/test-pmd/testpmd.h
@@ -558,6 +558,7 @@  void get_flex_filter(uint8_t port_id, uint16_t index);
 int port_id_is_invalid(portid_t port_id);
 int rx_queue_id_is_invalid(queueid_t rxq_id);
 int tx_queue_id_is_invalid(queueid_t txq_id);
+int set_multi_thread(int on);
 
 /*
  * Work-around of a compilation error with ICC on invocations of the