use_PA_to_store_packets.cc

#include <inttypes.h>
#include <ios>
#include <rte_common.h>
#include <rte_cycles.h>
#include <rte_eal.h>
#include <rte_ethdev.h>
#include <rte_log.h>
#include <rte_malloc.h>
#include <rte_hexdump.h>
#include <sys/queue.h>
#include <vector>

#include <stdexcept>

#include <sys/mman.h>

#include <iostream>

using namespace std;

constexpr size_t MBUFS_NUM = 8192;
constexpr size_t QUEUE_NUM = 1;

constexpr size_t RX_DESC_NUM = 1024;
constexpr size_t TX_DESC_NUM = 1024;

constexpr size_t MAX_PKT_BURST = 32;

constexpr size_t STATS_REFRESH_INTERVAL = 1;

rte_eth_conf port_conf;

int main(int argc, char **argv) {
	int dpdk_argc = rte_eal_init(argc, argv);
	if (dpdk_argc < 0)
		throw runtime_error("Failed to initialize EAL");
	argc -= dpdk_argc;
	argv += dpdk_argc;

	if (rte_eth_dev_count_avail() == 0)
		throw runtime_error("Invalid number of eth devices");

	rte_eth_dev_info dev_info;
	if (rte_eth_dev_info_get(0, &dev_info) != 0)
		throw runtime_error("Failed to get device info");

	// Configure mbuf pool
	auto external_mbuf_pool = true;
	rte_mempool *mp;
	if (external_mbuf_pool) {
		size_t mem_size = RTE_ALIGN_CEIL(MBUFS_NUM * QUEUE_NUM * RTE_MBUF_DEFAULT_BUF_SIZE, 4096);
		auto mem = mmap(0, mem_size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
		mlock(mem, mem_size);
		if (rte_extmem_register(mem, mem_size, nullptr, 0, 4096) != 0)
			throw runtime_error("Failed to register DPDK external memory");

		vector<rte_pktmbuf_extmem> ext_mems;
		for (size_t i = 0; i < mem_size; i += 4096) {
			((uint8_t *)mem)[i] = 0;
			auto cur_vaddr = (uintptr_t)mem + i;
			auto cur_paddr = rte_mem_virt2phy((void *)cur_vaddr);
			if (cur_paddr == RTE_BAD_IOVA)
				throw runtime_error("Failed to read memory mapping");
			cerr << hex << showbase << cur_vaddr << "->" << cur_paddr << dec << noshowbase << endl;
			rte_pktmbuf_extmem ext_mem = {
				.buf_ptr = (void *)cur_vaddr,
				.buf_iova = cur_paddr,
				.buf_len = 4096,
				.elt_size = RTE_MBUF_DEFAULT_BUF_SIZE,
			};
			ext_mems.emplace_back(ext_mem);
		}

		mp = rte_pktmbuf_pool_create_extbuf("ext_mbuf_pool", MBUFS_NUM * QUEUE_NUM,
			0, 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_eth_dev_socket_id(0), ext_mems.data(), ext_mems.size());
		if (mp == nullptr)
			throw runtime_error("Failed to create external mbuf pool");
	} else {
		mp = rte_pktmbuf_pool_create("mbuf_pool", MBUFS_NUM * QUEUE_NUM,
			0, 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_eth_dev_socket_id(0));
		if (mp == nullptr)
			throw runtime_error("Failed to create external mbuf pool");
	}


	// Configure and start device
	auto rss_hf = port_conf.rx_adv_conf.rss_conf.rss_hf;
	if ((rss_hf & dev_info.flow_type_rss_offloads) != rss_hf) {
		port_conf.rx_adv_conf.rss_conf.rss_hf &= dev_info.flow_type_rss_offloads;
		cerr << "Requested not supported RSS hash functions. Changing "
			<< hex << showbase << rss_hf << " to " << port_conf.rx_adv_conf.rss_conf.rss_hf
			<< noshowbase << dec << endl;
	}

	size_t port_id;
	RTE_ETH_FOREACH_DEV(port_id) {
		if (rte_eth_dev_configure(port_id, QUEUE_NUM, QUEUE_NUM, &port_conf) != 0)
			throw runtime_error("Failed to configure device");
		uint16_t rxd = RX_DESC_NUM, txd = TX_DESC_NUM;
		if (rte_eth_dev_adjust_nb_rx_tx_desc(port_id, &rxd, &txd) != 0)
			throw runtime_error("Failed to adjust number of rx/tx descriptors");

		rte_eth_rxconf rxq_conf = dev_info.default_rxconf;
		rxq_conf.offloads = port_conf.rxmode.offloads;

		rte_eth_txconf txq_conf = dev_info.default_txconf;
		txq_conf.offloads = port_conf.txmode.offloads;

		for (size_t q = 0; q < QUEUE_NUM; ++q)
			if (rte_eth_rx_queue_setup(port_id, q, rxd,
					rte_eth_dev_socket_id(0),
					&rxq_conf,
					mp) != 0)
				throw runtime_error("Failed to setup RX queue");

		for (size_t q = 0; q < QUEUE_NUM; ++q)
			if (rte_eth_tx_queue_setup(port_id, q, txd,
					rte_eth_dev_socket_id(0),
					&txq_conf) != 0)
				throw runtime_error("Failed to setup TX queue");

		if (rte_eth_dev_set_ptypes(port_id, RTE_PTYPE_UNKNOWN, nullptr, 0) != 0)
			throw runtime_error("Failed to set ptypes");

		if (rte_eth_promiscuous_enable(port_id) != 0)
			throw runtime_error("Failed to enable promiscuous mode");

		if (rte_eth_dev_start(port_id) != 0)
			throw runtime_error("Failed to start device");
	}


	// Start receiving loop
	struct stats_struct {
		size_t rx_burst_cnt = 0;
		size_t non_zero_rx_burst_cnt = 0;
		size_t rx_pkts = 0;
		size_t tx_pkts = 0;
	};
	vector<stats_struct> stats(QUEUE_NUM);

	vector<rte_mbuf *> mbufs(MAX_PKT_BURST);

	auto refresh_interval = STATS_REFRESH_INTERVAL * rte_get_timer_hz();
	auto prev_tsc = rte_rdtsc();
	size_t stats_cnt = 0;
	while (true) {
		auto cur_tsc = rte_rdtsc();
		auto diff_tsc = cur_tsc - prev_tsc;
		if (diff_tsc > refresh_interval) {
			++stats_cnt;
			for (size_t queue = 0; queue < QUEUE_NUM; ++queue) {
				auto &stat = stats[queue];
				cerr << "Queue: " << queue << endl;
				cerr << "Number of all rx burst calls: " << stat.rx_burst_cnt << endl;
				cerr << "Number of non-zero rx burst calls: " << stat.non_zero_rx_burst_cnt << endl;
				cerr << "Avg pkt nb received per rx burst: " << (double)stat.rx_pkts / stat.rx_burst_cnt << endl;
				cerr << "All received pkts: " << stat.rx_pkts << endl;
				cerr << "All sent pkts: " << stat.tx_pkts << endl;
				cerr << "All dropped pkts: " << stat.rx_pkts - stat.tx_pkts << endl;
				cerr << endl;
				stat = stats_struct {};
				prev_tsc = cur_tsc;
			}
		}

		for (size_t queue = 0; queue < QUEUE_NUM; ++queue) {
			auto &stat = stats[queue];

			++stat.rx_burst_cnt;

			auto rx_num = rte_eth_rx_burst(0, queue, mbufs.data(), MAX_PKT_BURST);
			if (!rx_num)
				continue;

			stat.rx_pkts += rx_num;
			++stat.non_zero_rx_burst_cnt;

			auto tx_num = rte_eth_tx_burst(1, queue, mbufs.data(), rx_num);
			stat.tx_pkts += tx_num;
			rte_pktmbuf_free_bulk(mbufs.data() + tx_num, rx_num - tx_num);
		}
	}

	return 0;
}