example: classifier: create a dedicated pktout queue for each worker

Create a dedicated pktout queue for each worker. Print a message if
queues have to be shared because the pktout doesn't support that many
queues.

Signed-off-by: Jere Leppänen <jere.leppanen@nokia.com>
Reviewed-by: Tuomas Taipale <tuomas.taipale@nokia.com>

1 files changed, 24 insertions, 11 deletions

diff --git a/example/classifier/odp_classifier.c b/example/classifier/odp_classifier.c
index 8e077629c..623c2405e 100644
--- a/example/classifier/odp_classifier.c
+++ b/example/classifier/odp_classifier.c
@@ -80,6 +80,8 @@ typedef struct {
 } ci_pass_counters;
 
 typedef struct {
+	odp_pktout_queue_t pktout[MAX_WORKERS];
+	int num_pktout;
 	global_statistics stats[MAX_PMR_COUNT];
 	ci_pass_counters ci_pass_rules[MAX_PMR_COUNT];
 	int policy_count;	/**< global policy count */
@@ -250,6 +252,8 @@ static odp_pktio_t create_pktio(const char *dev, odp_pool_t pool)
 	odp_pktin_queue_param_t pktin_param;
 	odp_pktio_capability_t capa;
 	odp_pktio_config_t cfg;
+	odp_pktout_queue_param_t pktout_queue_param;
+	int num_tx;
 
 	odp_pktio_param_init(&pktio_param);
 	pktio_param.in_mode = ODP_PKTIN_MODE_SCHED;
@@ -277,11 +281,29 @@ static odp_pktio_t create_pktio(const char *dev, odp_pool_t pool)
 		exit(EXIT_FAILURE);
 	}
 
-	if (odp_pktout_queue_config(pktio, NULL)) {
+	num_tx = appl_args_gbl->cpu_count;
+
+	if (num_tx > (int)capa.max_output_queues) {
+		printf("Sharing %i output queues between %i workers\n",
+		       capa.max_output_queues, num_tx);
+		num_tx = capa.max_output_queues;
+	}
+
+	appl_args_gbl->num_pktout = num_tx;
+
+	odp_pktout_queue_param_init(&pktout_queue_param);
+	pktout_queue_param.num_queues = num_tx;
+
+	if (odp_pktout_queue_config(pktio, &pktout_queue_param)) {
 		ODPH_ERR("pktout queue config failed for %s\n", dev);
 		exit(EXIT_FAILURE);
 	}
 
+	if (odp_pktout_queue(pktio, appl_args_gbl->pktout, num_tx) != num_tx) {
+		ODPH_ERR("Pktout queue query failed: %s\n", dev);
+		exit(EXIT_FAILURE);
+	}
+
 	if (appl_args_gbl->promisc_mode && odp_pktio_promisc_mode(pktio) != 1) {
 		if (!capa.set_op.op.promisc_mode) {
 			ODPH_ERR("enabling promisc mode not supported %s\n", dev);
@@ -323,7 +345,6 @@ static odp_pktio_t create_pktio(const char *dev, odp_pool_t pool)
 static int pktio_receive_thread(void *arg)
 {
 	int thr;
-	odp_pktout_queue_t pktout;
 	odp_packet_t pkt;
 	odp_pool_t pool;
 	odp_event_t ev;
@@ -334,11 +355,10 @@ static int pktio_receive_thread(void *arg)
 	thr = odp_thread_id();
 	appl_args_t *appl = (appl_args_t *)arg;
 	uint64_t wait_time = odp_schedule_wait_time(100 * ODP_TIME_MSEC_IN_NS);
+	odp_pktout_queue_t pktout = appl_args_gbl->pktout[thr % appl_args_gbl->num_pktout];
 
 	/* Loop packets */
 	for (;;) {
-		odp_pktio_t pktio_tmp;
-
 		if (appl->shutdown)
 			break;
 
@@ -373,13 +393,6 @@ static int pktio_receive_thread(void *arg)
 			continue;
 		}
 
-		pktio_tmp = odp_packet_input(pkt);
-
-		if (odp_pktout_queue(pktio_tmp, &pktout, 1) != 1) {
-			ODPH_ERR("  [%02i] Error: no output queue\n", thr);
-			return -1;
-		}
-
 		pool = odp_packet_pool(pkt);
 
 		/* Swap Eth MACs and possibly IP-addrs before sending back */