src/dpdk/lib/librte_eventdev/rte_event_eth_rx_adapter.c
static inline uint32_t
rxa_intr_ring_dequeue(struct rte_event_eth_rx_adapter *rx_adapter)
{
	uint32_t n;
	uint32_t nb_rx = 0;
	int rxq_empty;
	struct rte_eth_event_enqueue_buffer *buf;
	rte_spinlock_t *ring_lock;
	uint8_t max_done = 0;

	if (rx_adapter->num_rx_intr == 0)
		return 0;

	if (rte_ring_count(rx_adapter->intr_ring) == 0
		&& !rx_adapter->qd_valid)
		return 0;

	buf = &rx_adapter->event_enqueue_buffer;
	ring_lock = &rx_adapter->intr_ring_lock;

	if (buf->count >= BATCH_SIZE)
		rxa_flush_event_buffer(rx_adapter);

	while (BATCH_SIZE <= (RTE_DIM(buf->events) - buf->count)) {
		struct eth_device_info *dev_info;
		uint16_t port;
		uint16_t queue;
		union queue_data qd  = rx_adapter->qd;
		int err;

		if (!rx_adapter->qd_valid) {
			struct eth_rx_queue_info *queue_info;

			rte_spinlock_lock(ring_lock);
			err = rte_ring_dequeue(rx_adapter->intr_ring, &qd.ptr);
			if (err) {
				rte_spinlock_unlock(ring_lock);
				break;
			}

			port = qd.port;
			queue = qd.queue;
			rx_adapter->qd = qd;
			rx_adapter->qd_valid = 1;
			dev_info = &rx_adapter->eth_devices[port];
			if (rxa_shared_intr(dev_info, queue))
				dev_info->shared_intr_enabled = 1;
			else {
				queue_info = &dev_info->rx_queue[queue];
				queue_info->intr_enabled = 1;
			}
			rte_eth_dev_rx_intr_enable(port, queue);
			rte_spinlock_unlock(ring_lock);
		} else {
			port = qd.port;
			queue = qd.queue;

			dev_info = &rx_adapter->eth_devices[port];
		}

		if (rxa_shared_intr(dev_info, queue)) {
			uint16_t i;
			uint16_t nb_queues;

			nb_queues = dev_info->dev->data->nb_rx_queues;
			n = 0;
			for (i = dev_info->next_q_idx; i < nb_queues; i++) {
				uint8_t enq_buffer_full;

				if (!rxa_intr_queue(dev_info, i))
					continue;
				n = rxa_eth_rx(rx_adapter, port, i, nb_rx,
					rx_adapter->max_nb_rx,
					&rxq_empty);
				nb_rx += n;

				enq_buffer_full = !rxq_empty && n == 0;
				max_done = nb_rx > rx_adapter->max_nb_rx;

				if (enq_buffer_full || max_done) {
					dev_info->next_q_idx = i;
					goto done;
				}
			}

			rx_adapter->qd_valid = 0;

			/* Reinitialize for next interrupt */
			dev_info->next_q_idx = dev_info->multi_intr_cap ?
						RTE_MAX_RXTX_INTR_VEC_ID - 1 :
						0;
		} else {
			n = rxa_eth_rx(rx_adapter, port, queue, nb_rx,
				rx_adapter->max_nb_rx,
				&rxq_empty);
			rx_adapter->qd_valid = !rxq_empty;
			nb_rx += n;
			if (nb_rx > rx_adapter->max_nb_rx)
				break;
		}
	}

done:
	rx_adapter->stats.rx_intr_packets += nb_rx;
	return nb_rx;
}
/* Dequeue <port, q> from interrupt ring and enqueue received
 * mbufs to eventdev
 */
Dequeue  from interrupt ring and enqueue received
mbufs to eventdev
Dequeue from interrupt ring and enqueue received mbufs to eventdev
static inline uint32_t
rxa_intr_ring_dequeue(struct rte_event_eth_rx_adapter *rx_adapter)
{
	uint32_t n;
	uint32_t nb_rx = 0;
	int rxq_empty;
	struct rte_eth_event_enqueue_buffer *buf;
	rte_spinlock_t *ring_lock;
	uint8_t max_done = 0;
	if (rx_adapter->num_rx_intr == 0)
		return 0;
	if (rte_ring_count(rx_adapter->intr_ring) == 0
		&& !rx_adapter->qd_valid)
		return 0;
	buf = &rx_adapter->event_enqueue_buffer;
	ring_lock = &rx_adapter->intr_ring_lock;
	if (buf->count >= BATCH_SIZE)
		rxa_flush_event_buffer(rx_adapter);
	while (BATCH_SIZE <= (RTE_DIM(buf->events) - buf->count)) {
		struct eth_device_info *dev_info;
		uint16_t port;
		uint16_t queue;
		union queue_data qd  = rx_adapter->qd;
		int err;
		if (!rx_adapter->qd_valid) {
			struct eth_rx_queue_info *queue_info;
			rte_spinlock_lock(ring_lock);
			err = rte_ring_dequeue(rx_adapter->intr_ring, &qd.ptr);
			if (err) {
				rte_spinlock_unlock(ring_lock);
				break;
			}
			port = qd.port;
			queue = qd.queue;
			rx_adapter->qd = qd;
			rx_adapter->qd_valid = 1;
			dev_info = &rx_adapter->eth_devices[port];
			if (rxa_shared_intr(dev_info, queue))
				dev_info->shared_intr_enabled = 1;
			else {
				queue_info = &dev_info->rx_queue[queue];
				queue_info->intr_enabled = 1;
			}
			rte_eth_dev_rx_intr_enable(port, queue);
			rte_spinlock_unlock(ring_lock);
		} else {
			port = qd.port;
			queue = qd.queue;
			dev_info = &rx_adapter->eth_devices[port];
		}
		if (rxa_shared_intr(dev_info, queue)) {
			uint16_t i;
			uint16_t nb_queues;
			nb_queues = dev_info->dev->data->nb_rx_queues;
			n = 0;
			for (i = dev_info->next_q_idx; i < nb_queues; i++) {
				uint8_t enq_buffer_full;
				if (!rxa_intr_queue(dev_info, i))
					continue;
				n = rxa_eth_rx(rx_adapter, port, i, nb_rx,
					rx_adapter->max_nb_rx,
					&rxq_empty);
				nb_rx += n;
				enq_buffer_full = !rxq_empty && n == 0;
				max_done = nb_rx > rx_adapter->max_nb_rx;
				if (enq_buffer_full || max_done) {
					dev_info->next_q_idx = i;
					goto done;
				}
			}
			rx_adapter->qd_valid = 0;
			dev_info->next_q_idx = dev_info->multi_intr_cap ?
						RTE_MAX_RXTX_INTR_VEC_ID - 1 :
						0;
		} else {
			n = rxa_eth_rx(rx_adapter, port, queue, nb_rx,
				rx_adapter->max_nb_rx,
				&rxq_empty);
			rx_adapter->qd_valid = !rxq_empty;
			nb_rx += n;
			if (nb_rx > rx_adapter->max_nb_rx)
				break;
		}
	}
done:
	rx_adapter->stats.rx_intr_packets += nb_rx;
	return nb_rx;
}
static inline uint32_t rxa_intr_ring_dequeue ( struct rte_event_eth_rx_adapter * rx_adapter ) { uint32_t n ; uint32_t nb_rx = 0 ; int rxq_empty ; struct rte_eth_event_enqueue_buffer * buf ; rte_spinlock_t * ring_lock ; uint8_t max_done = 0 ; if ( rx_adapter -> num_rx_intr == 0 ) return 0 ; if ( rte_ring_count ( rx_adapter -> intr_ring ) == 0 && ! rx_adapter -> qd_valid ) return 0 ; buf = & rx_adapter -> event_enqueue_buffer ; ring_lock = & rx_adapter -> intr_ring_lock ; if ( buf -> count >= BATCH_SIZE ) rxa_flush_event_buffer ( rx_adapter ) ; while ( BATCH_SIZE <= ( RTE_DIM ( buf -> events ) - buf -> count ) ) { struct eth_device_info * dev_info ; uint16_t port ; uint16_t queue ; union queue_data qd = rx_adapter -> qd ; int err ; if ( ! rx_adapter -> qd_valid ) { struct eth_rx_queue_info * queue_info ; rte_spinlock_lock ( ring_lock ) ; err = rte_ring_dequeue ( rx_adapter -> intr_ring , & qd . ptr ) ; if ( err ) { rte_spinlock_unlock ( ring_lock ) ; break ; } port = qd . port ; queue = qd . queue ; rx_adapter -> qd = qd ; rx_adapter -> qd_valid = 1 ; dev_info = & rx_adapter -> eth_devices [ port ] ; if ( rxa_shared_intr ( dev_info , queue ) ) dev_info -> shared_intr_enabled = 1 ; else { queue_info = & dev_info -> rx_queue [ queue ] ; queue_info -> intr_enabled = 1 ; } rte_eth_dev_rx_intr_enable ( port , queue ) ; rte_spinlock_unlock ( ring_lock ) ; } else { port = qd . port ; queue = qd . queue ; dev_info = & rx_adapter -> eth_devices [ port ] ; } if ( rxa_shared_intr ( dev_info , queue ) ) { uint16_t i ; uint16_t nb_queues ; nb_queues = dev_info -> dev -> data -> nb_rx_queues ; n = 0 ; for ( i = dev_info -> next_q_idx ; i < nb_queues ; i ++ ) { uint8_t enq_buffer_full ; if ( ! rxa_intr_queue ( dev_info , i ) ) continue ; n = rxa_eth_rx ( rx_adapter , port , i , nb_rx , rx_adapter -> max_nb_rx , & rxq_empty ) ; nb_rx += n ; enq_buffer_full = ! rxq_empty && n == 0 ; max_done = nb_rx > rx_adapter -> max_nb_rx ; if ( enq_buffer_full || max_done ) { dev_info -> next_q_idx = i ; goto done ; } } rx_adapter -> qd_valid = 0 ; dev_info -> next_q_idx = dev_info -> multi_intr_cap ? RTE_MAX_RXTX_INTR_VEC_ID - 1 : 0 ; } else { n = rxa_eth_rx ( rx_adapter , port , queue , nb_rx , rx_adapter -> max_nb_rx , & rxq_empty ) ; rx_adapter -> qd_valid = ! rxq_empty ; nb_rx += n ; if ( nb_rx > rx_adapter -> max_nb_rx ) break ; } } done : rx_adapter -> stats . rx_intr_packets += nb_rx ; return nb_rx ; }
Dequeue <port, q> from interrupt ring and enqueue received
 mbufs to eventdev
Dequeue <port q > from interrupt ring and enqueue received mbufs to eventdev
/* Reinitialize for next interrupt */