X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=app%2Ftest-eventdev%2Ftest_pipeline_queue.c;h=7bebac34fce2774787a925adb7a2208a1290a513;hb=042f5c94fd3a49e4fe85e5ae2405a39ca65a2436;hp=773c3ecaa3ac9af864992798ce9fd9db06e6719b;hpb=d60b418503ef363e406b8c32bf5bcf81527dc395;p=dpdk.git

diff --git a/app/test-eventdev/test_pipeline_queue.c b/app/test-eventdev/test_pipeline_queue.c
index 773c3ecaa3..7bebac34fc 100644
--- a/app/test-eventdev/test_pipeline_queue.c
+++ b/app/test-eventdev/test_pipeline_queue.c
@@ -5,30 +5,306 @@
 
 #include "test_pipeline_common.h"
 
-/* See http://dpdk.org/doc/guides/tools/testeventdev.html for test details */
+/* See http://doc.dpdk.org/guides/tools/testeventdev.html for test details */
 
 static __rte_always_inline int
 pipeline_queue_nb_event_queues(struct evt_options *opt)
 {
-	uint16_t eth_count = rte_eth_dev_count();
+	uint16_t eth_count = rte_eth_dev_count_avail();
 
 	return (eth_count * opt->nb_stages) + eth_count;
 }
 
+static __rte_noinline int
+pipeline_queue_worker_single_stage_tx(void *arg)
+{
+	PIPELINE_WORKER_SINGLE_STAGE_INIT;
+
+	while (t->done == false) {
+		uint16_t event = rte_event_dequeue_burst(dev, port, &ev, 1, 0);
+
+		if (!event) {
+			rte_pause();
+			continue;
+		}
+
+		if (ev.sched_type == RTE_SCHED_TYPE_ATOMIC) {
+			pipeline_event_tx(dev, port, &ev);
+			w->processed_pkts++;
+		} else {
+			ev.queue_id++;
+			pipeline_fwd_event(&ev, RTE_SCHED_TYPE_ATOMIC);
+			pipeline_event_enqueue(dev, port, &ev);
+		}
+	}
+
+	return 0;
+}
+
+static __rte_noinline int
+pipeline_queue_worker_single_stage_fwd(void *arg)
+{
+	PIPELINE_WORKER_SINGLE_STAGE_INIT;
+	const uint8_t *tx_queue = t->tx_evqueue_id;
+
+	while (t->done == false) {
+		uint16_t event = rte_event_dequeue_burst(dev, port, &ev, 1, 0);
+
+		if (!event) {
+			rte_pause();
+			continue;
+		}
+
+		ev.queue_id = tx_queue[ev.mbuf->port];
+		rte_event_eth_tx_adapter_txq_set(ev.mbuf, 0);
+		pipeline_fwd_event(&ev, RTE_SCHED_TYPE_ATOMIC);
+		pipeline_event_enqueue(dev, port, &ev);
+		w->processed_pkts++;
+	}
+
+	return 0;
+}
+
+static __rte_noinline int
+pipeline_queue_worker_single_stage_burst_tx(void *arg)
+{
+	PIPELINE_WORKER_SINGLE_STAGE_BURST_INIT;
+
+	while (t->done == false) {
+		uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev,
+				BURST_SIZE, 0);
+
+		if (!nb_rx) {
+			rte_pause();
+			continue;
+		}
+
+		for (i = 0; i < nb_rx; i++) {
+			rte_prefetch0(ev[i + 1].mbuf);
+			if (ev[i].sched_type == RTE_SCHED_TYPE_ATOMIC) {
+				pipeline_event_tx(dev, port, &ev[i]);
+				ev[i].op = RTE_EVENT_OP_RELEASE;
+				w->processed_pkts++;
+			} else {
+				ev[i].queue_id++;
+				pipeline_fwd_event(&ev[i],
+						RTE_SCHED_TYPE_ATOMIC);
+			}
+		}
+
+		pipeline_event_enqueue_burst(dev, port, ev, nb_rx);
+	}
+
+	return 0;
+}
+
+static __rte_noinline int
+pipeline_queue_worker_single_stage_burst_fwd(void *arg)
+{
+	PIPELINE_WORKER_SINGLE_STAGE_BURST_INIT;
+	const uint8_t *tx_queue = t->tx_evqueue_id;
+
+	while (t->done == false) {
+		uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev,
+				BURST_SIZE, 0);
+
+		if (!nb_rx) {
+			rte_pause();
+			continue;
+		}
+
+		for (i = 0; i < nb_rx; i++) {
+			rte_prefetch0(ev[i + 1].mbuf);
+			ev[i].queue_id = tx_queue[ev[i].mbuf->port];
+			rte_event_eth_tx_adapter_txq_set(ev[i].mbuf, 0);
+			pipeline_fwd_event(&ev[i], RTE_SCHED_TYPE_ATOMIC);
+		}
+
+		pipeline_event_enqueue_burst(dev, port, ev, nb_rx);
+		w->processed_pkts += nb_rx;
+	}
+
+	return 0;
+}
+
+
+static __rte_noinline int
+pipeline_queue_worker_multi_stage_tx(void *arg)
+{
+	PIPELINE_WORKER_MULTI_STAGE_INIT;
+	const uint8_t *tx_queue = t->tx_evqueue_id;
+
+	while (t->done == false) {
+		uint16_t event = rte_event_dequeue_burst(dev, port, &ev, 1, 0);
+
+		if (!event) {
+			rte_pause();
+			continue;
+		}
+
+		cq_id = ev.queue_id % nb_stages;
+
+		if (ev.queue_id == tx_queue[ev.mbuf->port]) {
+			pipeline_event_tx(dev, port, &ev);
+			w->processed_pkts++;
+			continue;
+		}
+
+		ev.queue_id++;
+		pipeline_fwd_event(&ev, cq_id != last_queue ?
+				sched_type_list[cq_id] :
+				RTE_SCHED_TYPE_ATOMIC);
+		pipeline_event_enqueue(dev, port, &ev);
+	}
+
+	return 0;
+}
+
+static __rte_noinline int
+pipeline_queue_worker_multi_stage_fwd(void *arg)
+{
+	PIPELINE_WORKER_MULTI_STAGE_INIT;
+	const uint8_t *tx_queue = t->tx_evqueue_id;
+
+	while (t->done == false) {
+		uint16_t event = rte_event_dequeue_burst(dev, port, &ev, 1, 0);
+
+		if (!event) {
+			rte_pause();
+			continue;
+		}
+
+		cq_id = ev.queue_id % nb_stages;
+
+		if (cq_id == last_queue) {
+			ev.queue_id = tx_queue[ev.mbuf->port];
+			rte_event_eth_tx_adapter_txq_set(ev.mbuf, 0);
+			pipeline_fwd_event(&ev, RTE_SCHED_TYPE_ATOMIC);
+			w->processed_pkts++;
+		} else {
+			ev.queue_id++;
+			pipeline_fwd_event(&ev, sched_type_list[cq_id]);
+		}
+
+		pipeline_event_enqueue(dev, port, &ev);
+	}
+
+	return 0;
+}
+
+static __rte_noinline int
+pipeline_queue_worker_multi_stage_burst_tx(void *arg)
+{
+	PIPELINE_WORKER_MULTI_STAGE_BURST_INIT;
+	const uint8_t *tx_queue = t->tx_evqueue_id;
+
+	while (t->done == false) {
+		uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev,
+				BURST_SIZE, 0);
+
+		if (!nb_rx) {
+			rte_pause();
+			continue;
+		}
+
+		for (i = 0; i < nb_rx; i++) {
+			rte_prefetch0(ev[i + 1].mbuf);
+			cq_id = ev[i].queue_id % nb_stages;
+
+			if (ev[i].queue_id == tx_queue[ev[i].mbuf->port]) {
+				pipeline_event_tx(dev, port, &ev[i]);
+				ev[i].op = RTE_EVENT_OP_RELEASE;
+				w->processed_pkts++;
+				continue;
+			}
+
+			ev[i].queue_id++;
+			pipeline_fwd_event(&ev[i], cq_id != last_queue ?
+					sched_type_list[cq_id] :
+					RTE_SCHED_TYPE_ATOMIC);
+		}
+
+		pipeline_event_enqueue_burst(dev, port, ev, nb_rx);
+	}
+
+	return 0;
+}
+
+static __rte_noinline int
+pipeline_queue_worker_multi_stage_burst_fwd(void *arg)
+{
+	PIPELINE_WORKER_MULTI_STAGE_BURST_INIT;
+	const uint8_t *tx_queue = t->tx_evqueue_id;
+
+	while (t->done == false) {
+		uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev,
+				BURST_SIZE, 0);
+
+		if (!nb_rx) {
+			rte_pause();
+			continue;
+		}
+
+		for (i = 0; i < nb_rx; i++) {
+			rte_prefetch0(ev[i + 1].mbuf);
+			cq_id = ev[i].queue_id % nb_stages;
+
+			if (cq_id == last_queue) {
+				ev[i].queue_id = tx_queue[ev[i].mbuf->port];
+				rte_event_eth_tx_adapter_txq_set(ev[i].mbuf, 0);
+				pipeline_fwd_event(&ev[i],
+						RTE_SCHED_TYPE_ATOMIC);
+				w->processed_pkts++;
+			} else {
+				ev[i].queue_id++;
+				pipeline_fwd_event(&ev[i],
+						sched_type_list[cq_id]);
+			}
+		}
+
+		pipeline_event_enqueue_burst(dev, port, ev, nb_rx);
+	}
+
+	return 0;
+}
+
 static int
 worker_wrapper(void *arg)
 {
-	RTE_SET_USED(arg);
+	struct worker_data *w  = arg;
+	struct evt_options *opt = w->t->opt;
+	const bool burst = evt_has_burst_mode(w->dev_id);
+	const bool internal_port = w->t->internal_port;
+	const uint8_t nb_stages = opt->nb_stages;
+	RTE_SET_USED(opt);
+
+	if (nb_stages == 1) {
+		if (!burst && internal_port)
+			return pipeline_queue_worker_single_stage_tx(arg);
+		else if (!burst && !internal_port)
+			return pipeline_queue_worker_single_stage_fwd(arg);
+		else if (burst && internal_port)
+			return pipeline_queue_worker_single_stage_burst_tx(arg);
+		else if (burst && !internal_port)
+			return pipeline_queue_worker_single_stage_burst_fwd(
+					arg);
+	} else {
+		if (!burst && internal_port)
+			return pipeline_queue_worker_multi_stage_tx(arg);
+		else if (!burst && !internal_port)
+			return pipeline_queue_worker_multi_stage_fwd(arg);
+		else if (burst && internal_port)
+			return pipeline_queue_worker_multi_stage_burst_tx(arg);
+		else if (burst && !internal_port)
+			return pipeline_queue_worker_multi_stage_burst_fwd(arg);
+
+	}
 	rte_panic("invalid worker\n");
 }
 
 static int
 pipeline_queue_launch_lcores(struct evt_test *test, struct evt_options *opt)
 {
-	struct test_pipeline *t = evt_test_priv(test);
-
-	if (t->mt_unsafe)
-		rte_service_component_runstate_set(t->tx_service.service_id, 1);
 	return pipeline_launch_lcores(test, opt, worker_wrapper);
 }
 
@@ -40,36 +316,25 @@ pipeline_queue_eventdev_setup(struct evt_test *test, struct evt_options *opt)
 	int nb_queues;
 	int nb_stages = opt->nb_stages;
 	uint8_t queue;
-	struct rte_event_dev_info info;
-	struct test_pipeline *t = evt_test_priv(test);
-	uint8_t tx_evqueue_id = 0;
+	uint8_t tx_evport_id = 0;
+	uint8_t tx_evqueue_id[RTE_MAX_ETHPORTS];
 	uint8_t queue_arr[RTE_EVENT_MAX_QUEUES_PER_DEV];
 	uint8_t nb_worker_queues = 0;
+	uint16_t prod = 0;
+	struct rte_event_dev_info info;
+	struct test_pipeline *t = evt_test_priv(test);
 
 	nb_ports = evt_nr_active_lcores(opt->wlcores);
-	nb_queues = rte_eth_dev_count() * (nb_stages);
+	nb_queues = rte_eth_dev_count_avail() * (nb_stages);
 
-	/* Extra port for Tx service. */
-	if (t->mt_unsafe) {
-		tx_evqueue_id = nb_queues;
-		nb_ports++;
-		nb_queues++;
-	} else
-		nb_queues += rte_eth_dev_count();
+	/* One queue for Tx adapter per port */
+	nb_queues += rte_eth_dev_count_avail();
 
-	rte_event_dev_info_get(opt->dev_id, &info);
+	memset(tx_evqueue_id, 0, sizeof(uint8_t) * RTE_MAX_ETHPORTS);
+	memset(queue_arr, 0, sizeof(uint8_t) * RTE_EVENT_MAX_QUEUES_PER_DEV);
 
-	const struct rte_event_dev_config config = {
-			.nb_event_queues = nb_queues,
-			.nb_event_ports = nb_ports,
-			.nb_events_limit  = info.max_num_events,
-			.nb_event_queue_flows = opt->nb_flows,
-			.nb_event_port_dequeue_depth =
-				info.max_event_port_dequeue_depth,
-			.nb_event_port_enqueue_depth =
-				info.max_event_port_enqueue_depth,
-	};
-	ret = rte_event_dev_configure(opt->dev_id, &config);
+	rte_event_dev_info_get(opt->dev_id, &info);
+	ret = evt_configure_eventdev(opt, nb_queues, nb_ports);
 	if (ret) {
 		evt_err("failed to configure eventdev %d", opt->dev_id);
 		return ret;
@@ -84,24 +349,19 @@ pipeline_queue_eventdev_setup(struct evt_test *test, struct evt_options *opt)
 	for (queue = 0; queue < nb_queues; queue++) {
 		uint8_t slot;
 
-		if (!t->mt_unsafe) {
-			slot = queue % (nb_stages + 1);
-			q_conf.schedule_type = slot == nb_stages ?
-				RTE_SCHED_TYPE_ATOMIC :
-				opt->sched_type_list[slot];
-		} else {
-			slot = queue % nb_stages;
-
-			if (queue == tx_evqueue_id) {
-				q_conf.schedule_type = RTE_SCHED_TYPE_ATOMIC;
+		q_conf.event_queue_cfg = 0;
+		slot = queue % (nb_stages + 1);
+		if (slot == nb_stages) {
+			q_conf.schedule_type = RTE_SCHED_TYPE_ATOMIC;
+			if (!t->internal_port) {
 				q_conf.event_queue_cfg =
 					RTE_EVENT_QUEUE_CFG_SINGLE_LINK;
-			} else {
-				q_conf.schedule_type =
-					opt->sched_type_list[slot];
-				queue_arr[nb_worker_queues] = queue;
-				nb_worker_queues++;
 			}
+			tx_evqueue_id[prod++] = queue;
+		} else {
+			q_conf.schedule_type = opt->sched_type_list[slot];
+			queue_arr[nb_worker_queues] = queue;
+			nb_worker_queues++;
 		}
 
 		ret = rte_event_queue_setup(opt->dev_id, queue, &q_conf);
@@ -111,6 +371,9 @@ pipeline_queue_eventdev_setup(struct evt_test *test, struct evt_options *opt)
 		}
 	}
 
+	if (opt->wkr_deq_dep > info.max_event_port_dequeue_depth)
+		opt->wkr_deq_dep = info.max_event_port_dequeue_depth;
+
 	/* port configuration */
 	const struct rte_event_port_conf p_conf = {
 			.dequeue_depth = opt->wkr_deq_dep,
@@ -118,19 +381,11 @@ pipeline_queue_eventdev_setup(struct evt_test *test, struct evt_options *opt)
 			.new_event_threshold = info.max_num_events,
 	};
 
-	/*
-	 * If tx is multi thread safe then allow workers to do Tx else use Tx
-	 * service to Tx packets.
-	 */
-	if (t->mt_unsafe) {
+	if (!t->internal_port) {
 		ret = pipeline_event_port_setup(test, opt, queue_arr,
 				nb_worker_queues, p_conf);
 		if (ret)
 			return ret;
-
-		ret = pipeline_event_tx_service_setup(test, opt, tx_evqueue_id,
-				nb_ports - 1, p_conf);
-
 	} else
 		ret = pipeline_event_port_setup(test, opt, NULL, nb_queues,
 				p_conf);
@@ -142,7 +397,6 @@ pipeline_queue_eventdev_setup(struct evt_test *test, struct evt_options *opt)
 	 *
 	 * eth_dev_count = 2, nb_stages = 2.
 	 *
-	 * Multi thread safe :
 	 *	queues = 6
 	 *	stride = 3
 	 *
@@ -150,21 +404,14 @@ pipeline_queue_eventdev_setup(struct evt_test *test, struct evt_options *opt)
 	 *	eth0 -> q0 -> q1 -> (q2->tx)
 	 *	eth1 -> q3 -> q4 -> (q5->tx)
 	 *
-	 *	q2, q5 configured as ATOMIC
+	 *	q2, q5 configured as ATOMIC | SINGLE_LINK
 	 *
-	 * Multi thread unsafe :
-	 *	queues = 5
-	 *	stride = 2
-	 *
-	 *	event queue pipelines:
-	 *	eth0 -> q0 -> q1
-	 *			} (q4->tx) Tx service
-	 *	eth1 -> q2 -> q3
-	 *
-	 *	q4 configured as SINGLE_LINK|ATOMIC
 	 */
-	ret = pipeline_event_rx_adapter_setup(opt,
-			t->mt_unsafe ? nb_stages : nb_stages + 1, p_conf);
+	ret = pipeline_event_rx_adapter_setup(opt, nb_stages + 1, p_conf);
+	if (ret)
+		return ret;
+
+	ret = pipeline_event_tx_adapter_setup(opt, p_conf);
 	if (ret)
 		return ret;
 
@@ -178,12 +425,61 @@ pipeline_queue_eventdev_setup(struct evt_test *test, struct evt_options *opt)
 		}
 	}
 
+	/* Connect the tx_evqueue_id to the Tx adapter port */
+	if (!t->internal_port) {
+		RTE_ETH_FOREACH_DEV(prod) {
+			ret = rte_event_eth_tx_adapter_event_port_get(prod,
+					&tx_evport_id);
+			if (ret) {
+				evt_err("Unable to get Tx adptr[%d] evprt[%d]",
+						prod, tx_evport_id);
+				return ret;
+			}
+
+			if (rte_event_port_link(opt->dev_id, tx_evport_id,
+						&tx_evqueue_id[prod],
+						NULL, 1) != 1) {
+				evt_err("Unable to link Tx adptr[%d] evprt[%d]",
+						prod, tx_evport_id);
+				return ret;
+			}
+		}
+	}
+
 	ret = rte_event_dev_start(opt->dev_id);
 	if (ret) {
 		evt_err("failed to start eventdev %d", opt->dev_id);
 		return ret;
 	}
 
+
+	RTE_ETH_FOREACH_DEV(prod) {
+		ret = rte_eth_dev_start(prod);
+		if (ret) {
+			evt_err("Ethernet dev [%d] failed to start."
+					" Using synthetic producer", prod);
+			return ret;
+		}
+
+	}
+
+	RTE_ETH_FOREACH_DEV(prod) {
+		ret = rte_event_eth_rx_adapter_start(prod);
+		if (ret) {
+			evt_err("Rx adapter[%d] start failed", prod);
+			return ret;
+		}
+
+		ret = rte_event_eth_tx_adapter_start(prod);
+		if (ret) {
+			evt_err("Tx adapter[%d] start failed", prod);
+			return ret;
+		}
+	}
+
+	memcpy(t->tx_evqueue_id, tx_evqueue_id, sizeof(uint8_t) *
+			RTE_MAX_ETHPORTS);
+
 	return 0;
 }