example/classifier/odp_classifier.c
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;
	odp_queue_t queue;
	int i;
	global_statistics *stats;
	unsigned long err_cnt = 0;
	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);

	/* Loop packets */
	for (;;) {
		odp_pktio_t pktio_tmp;

		if (appl->shutdown)
			break;

		/* Use schedule to get buf from any input queue */
		ev = odp_schedule(&queue, wait_time);

		/* Loop back to receive packets incase of invalid event */
		if (odp_unlikely(ev == ODP_EVENT_INVALID))
			continue;

		pkt = odp_packet_from_event(ev);

		if (appl->verbose) {
			odp_queue_info_t info;
			uint32_t len = odp_packet_len(pkt);

			if (odp_queue_info(queue, &info) == 0)
				printf("Queue: %s\n", info.name);

			if (len > 96)
				len = 96;

			odp_packet_print_data(pkt, 0, len);
		}

		/* Total packets received */
		odp_atomic_inc_u64(&appl->total_packets);

		/* Drop packets with errors */
		if (odp_unlikely(drop_err_pkts(&pkt, 1) == 0)) {
			ODPH_ERR("Drop frame - err_cnt:%lu\n", ++err_cnt);
			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 */
		swap_pkt_addrs(&pkt, 1);
		for (i = 0; i <  MAX_PMR_COUNT; i++) {
			stats = &appl->stats[i];
			if (queue == stats->queue)
				odp_atomic_inc_u64(&stats->queue_pkt_count);
			if (pool == stats->pool)
				odp_atomic_inc_u64(&stats->pool_pkt_count);
		}

		if (appl->appl_mode == APPL_MODE_DROP) {
			odp_packet_free(pkt);
			continue;
		}

		if (odp_pktout_send(pktout, &pkt, 1) < 1) {
			ODPH_ERR("  [%i] Packet send failed\n", thr);
			odp_packet_free(pkt);
		}
	}

	return 0;
}
/**
 * Worker threads to receive the packet
 *
 */
Worker threads to receive the packet
Worker threads to receive the packet
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;
	odp_queue_t queue;
	int i;
	global_statistics *stats;
	unsigned long err_cnt = 0;
	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);
	for (;;) {
		odp_pktio_t pktio_tmp;
		if (appl->shutdown)
			break;
		ev = odp_schedule(&queue, wait_time);
		if (odp_unlikely(ev == ODP_EVENT_INVALID))
			continue;
		pkt = odp_packet_from_event(ev);
		if (appl->verbose) {
			odp_queue_info_t info;
			uint32_t len = odp_packet_len(pkt);
			if (odp_queue_info(queue, &info) == 0)
				printf("Queue: %s\n", info.name);
			if (len > 96)
				len = 96;
			odp_packet_print_data(pkt, 0, len);
		}
		odp_atomic_inc_u64(&appl->total_packets);
		if (odp_unlikely(drop_err_pkts(&pkt, 1) == 0)) {
			ODPH_ERR("Drop frame - err_cnt:%lu\n", ++err_cnt);
			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_pkt_addrs(&pkt, 1);
		for (i = 0; i <  MAX_PMR_COUNT; i++) {
			stats = &appl->stats[i];
			if (queue == stats->queue)
				odp_atomic_inc_u64(&stats->queue_pkt_count);
			if (pool == stats->pool)
				odp_atomic_inc_u64(&stats->pool_pkt_count);
		}
		if (appl->appl_mode == APPL_MODE_DROP) {
			odp_packet_free(pkt);
			continue;
		}
		if (odp_pktout_send(pktout, &pkt, 1) < 1) {
			ODPH_ERR("  [%i] Packet send failed\n", thr);
			odp_packet_free(pkt);
		}
	}
	return 0;
}
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 ; odp_queue_t queue ; int i ; global_statistics * stats ; unsigned long err_cnt = 0 ; 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 ) ; for ( ; ; ) { odp_pktio_t pktio_tmp ; if ( appl -> shutdown ) break ; ev = odp_schedule ( & queue , wait_time ) ; if ( odp_unlikely ( ev == ODP_EVENT_INVALID ) ) continue ; pkt = odp_packet_from_event ( ev ) ; if ( appl -> verbose ) { odp_queue_info_t info ; uint32_t len = odp_packet_len ( pkt ) ; if ( odp_queue_info ( queue , & info ) == 0 ) printf ( " \n " , info . name ) ; if ( len > 96 ) len = 96 ; odp_packet_print_data ( pkt , 0 , len ) ; } odp_atomic_inc_u64 ( & appl -> total_packets ) ; if ( odp_unlikely ( drop_err_pkts ( & pkt , 1 ) == 0 ) ) { ODPH_ERR ( " \n " , ++ err_cnt ) ; continue ; } pktio_tmp = odp_packet_input ( pkt ) ; if ( odp_pktout_queue ( pktio_tmp , & pktout , 1 ) != 1 ) { ODPH_ERR ( " \n " , thr ) ; return -1 ; } pool = odp_packet_pool ( pkt ) ; swap_pkt_addrs ( & pkt , 1 ) ; for ( i = 0 ; i < MAX_PMR_COUNT ; i ++ ) { stats = & appl -> stats [ i ] ; if ( queue == stats -> queue ) odp_atomic_inc_u64 ( & stats -> queue_pkt_count ) ; if ( pool == stats -> pool ) odp_atomic_inc_u64 ( & stats -> pool_pkt_count ) ; } if ( appl -> appl_mode == APPL_MODE_DROP ) { odp_packet_free ( pkt ) ; continue ; } if ( odp_pktout_send ( pktout , & pkt , 1 ) < 1 ) { ODPH_ERR ( " \n " , thr ) ; odp_packet_free ( pkt ) ; } } return 0 ; }
Worker threads to receive the packet
Worker threads to receive the packet
/* Loop packets */ /* Use schedule to get buf from any input queue */ /* Loop back to receive packets incase of invalid event */ /* Total packets received */ /* Drop packets with errors */ /* Swap Eth MACs and possibly IP-addrs before sending back */