+ /* allocate bufs for fill queue replenishment after rx */
+ if (rte_pktmbuf_alloc_bulk(umem->mb_pool, fq_bufs, nb_pkts)) {
+ AF_XDP_LOG(DEBUG,
+ "Failed to get enough buffers for fq.\n");
+ return -1;
+ }
+
+ rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
+
+ if (rcvd == 0) {
+#if defined(XDP_USE_NEED_WAKEUP)
+ if (xsk_ring_prod__needs_wakeup(&umem->fq))
+ (void)poll(rxq->fds, 1, 1000);
+#endif
+
+ goto out;
+ }
+
+ for (i = 0; i < rcvd; i++) {
+ const struct xdp_desc *desc;
+ uint64_t addr;
+ uint32_t len;
+ uint64_t offset;
+
+ desc = xsk_ring_cons__rx_desc(rx, idx_rx++);
+ addr = desc->addr;
+ len = desc->len;
+
+ offset = xsk_umem__extract_offset(addr);
+ addr = xsk_umem__extract_addr(addr);
+
+ bufs[i] = (struct rte_mbuf *)
+ xsk_umem__get_data(umem->buffer, addr +
+ umem->mb_pool->header_size);
+ bufs[i]->data_off = offset - sizeof(struct rte_mbuf) -
+ rte_pktmbuf_priv_size(umem->mb_pool) -
+ umem->mb_pool->header_size;
+
+ rte_pktmbuf_pkt_len(bufs[i]) = len;
+ rte_pktmbuf_data_len(bufs[i]) = len;
+ rx_bytes += len;
+ }
+
+ xsk_ring_cons__release(rx, rcvd);
+
+ (void)reserve_fill_queue(umem, rcvd, fq_bufs);
+
+ /* statistics */
+ rxq->stats.rx_pkts += rcvd;
+ rxq->stats.rx_bytes += rx_bytes;
+
+out:
+ if (rcvd != nb_pkts)
+ rte_mempool_put_bulk(umem->mb_pool, (void **)&fq_bufs[rcvd],
+ nb_pkts - rcvd);
+
+ return rcvd;
+}
+#else
+static uint16_t
+af_xdp_rx_cp(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
+{
+ struct pkt_rx_queue *rxq = queue;
+ struct xsk_ring_cons *rx = &rxq->rx;
+ struct xsk_umem_info *umem = rxq->umem;
+ struct xsk_ring_prod *fq = &umem->fq;
+ uint32_t idx_rx = 0;
+ unsigned long rx_bytes = 0;
+ int rcvd, i;
+ uint32_t free_thresh = fq->size >> 1;
+ struct rte_mbuf *mbufs[ETH_AF_XDP_RX_BATCH_SIZE];