external_mbuf_pool_problem.cc

#include <inttypes.h>
#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 <sys/queue.h>
#include <vector>

#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 = {
	.rxmode = {
		.mq_mode = ETH_MQ_RX_RSS,
		.split_hdr_size = 0,
	},
	.txmode = {
		.mq_mode = ETH_MQ_TX_NONE,
	},
	.rx_adv_conf = {
		.rss_conf = {
			.rss_hf = ETH_RSS_IP | ETH_RSS_UDP | ETH_RSS_TCP,
		},
	},
};

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");
	rte_device *dev = dev_info.device;

	// 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);
		cerr << "mem_size: " << mem_size << endl;
		auto mem = aligned_alloc(4096, mem_size);
		mlock(mem, mem_size);
		// auto mem = mmap(0, mem_size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_HUGETLB, -1, 0);
		rte_pktmbuf_extmem ext_mem = {
			.buf_ptr = mem,
			.buf_iova = (uintptr_t)mem,
			.buf_len = mem_size,
			.elt_size = RTE_MBUF_DEFAULT_BUF_SIZE,
		};

		if (rte_extmem_register(ext_mem.buf_ptr, ext_mem.buf_len, nullptr, 0, 4096) != 0)
			throw runtime_error("Failed to register DPDK external memory");


		if (rte_dev_dma_map(dev, ext_mem.buf_ptr, ext_mem.buf_iova, ext_mem.buf_len) != 0)
			throw runtime_error("Failed to DMA map external memory");

		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_mem, 1);
		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;
}