前面我们已经了解了如何使用DPDK创建线程并绑定核心，以及如何申请内存池并创建 RX/TX 队列。
接下来我们将了解DPDK的核心内容之一：以轮询的方式从网卡中收取报文。
下面直接给出一个实例，此实例使用核心1及核心2创建了两个线程用于报文处理，其中在核心1上运行的线程接收网卡0的消息，而在核心2上运行的线程接收网卡1的消息。在这里，从网卡上获取报文的方式为轮询，具体的接口为 rte_eth_rx_burst，而所谓的报文处理，就是简单地打印报文消息。
#include <arpa/inet.h>#include <rte_eal.h>
#include <rte_mbuf.h>
#include <rte_ethdev.h>#include<rte_errno.h>#define NB_SOCKETS 10
#define MEMPOOL_CACHE_SIZE 250static unsigned nb_mbuf = 512;
static int numa_on = 0;
static struct rte_mempool * pktmbuf_pool[NB_SOCKETS];static int
init_mem(unsigned nb_mbuf){unsigned lcore_id;int socketid;char mbuf_pool_name[64];for(lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++){if (rte_lcore_is_enabled(lcore_id) == 0){continue;}if(numa_on){socketid = rte_lcore_to_socket_id(lcore_id);}else{socketid = 0;}if(socketid >= NB_SOCKETS){rte_exit(EXIT_FAILURE, "Socket %d of lcore %u is out of range %d\n",socketid, lcore_id, NB_SOCKETS);}if(pktmbuf_pool[socketid] == NULL){snprintf(mbuf_pool_name, sizeof(mbuf_pool_name), "mbuf_pool_%d", socketid);pktmbuf_pool[socketid] = rte_pktmbuf_pool_create(mbuf_pool_name, nb_mbuf,MEMPOOL_CACHE_SIZE, 0, RTE_MBUF_DEFAULT_BUF_SIZE, socketid);if(pktmbuf_pool[socketid] == NULL){rte_exit(EXIT_FAILURE, "Cannot init mbuf pool on socket %d\n", socketid);}else{printf("Allocated mbuf pool on socket %d\n", socketid);}}}return 0;
}#define MAX_RX_QUEUE_PER_PORT 1
#define MAX_TX_QUEUE_PER_PORT 1
#define RX_RING_SIZE 128
#define TX_RING_SIZE 512
#define BURST_SIZE 128static const struct rte_eth_conf dev_conf_default = {.rxmode = {.max_rx_pkt_len = ETHER_MAX_LEN}
};static void
init_port(void)
{int nb_port = 0;int portid = 0;int ret = 0;nb_port = rte_eth_dev_count_avail();if(!nb_port){rte_exit(EXIT_FAILURE, "No support eth found\n");}printf("nb_port = %d\n", nb_port);for(portid=0;portid < nb_port;portid++){ret = rte_eth_dev_configure(portid, MAX_RX_QUEUE_PER_PORT, MAX_TX_QUEUE_PER_PORT, &dev_conf_default);if (ret < 0) {rte_exit(EXIT_FAILURE, "Cannot configure device: err=%d, port=%u\n", ret, portid);}ret = rte_eth_rx_queue_setup(portid, 0, RX_RING_SIZE, rte_eth_dev_socket_id(portid), NULL, pktmbuf_pool[0]);if (ret < 0) {rte_exit(EXIT_FAILURE, "rte_eth_rx_queue_setup:err=%d, port=%u\n", ret, portid);}ret = rte_eth_tx_queue_setup(portid, 0, TX_RING_SIZE, rte_eth_dev_socket_id(portid), NULL);if (ret < 0) {rte_exit(EXIT_FAILURE, "rte_eth_tx_queue_setup:err=%d, port=%u\n", ret, portid);}ret = rte_eth_dev_start(portid);if (ret < 0) {rte_exit(EXIT_FAILURE, "rte_eth_dev_start:err=%d, port=%u\n", ret, portid);}rte_eth_promiscuous_enable(portid);}
}static void
pkt_process(struct rte_mbuf *mbuf){struct ether_hdr *eth = rte_pktmbuf_mtod(mbuf, struct ether_hdr *);if(eth->ether_type == rte_cpu_to_be_16(ETHER_TYPE_IPv4)){struct ipv4_hdr *iphdr = rte_pktmbuf_mtod_offset(mbuf, struct ipv4_hdr *,sizeof(struct ether_hdr));//udpif(iphdr->next_proto_id == IPPROTO_UDP){struct udp_hdr *udphdr = (struct udp_hdr *)(iphdr + 1);uint16_t length = ntohs(udphdr->dgram_len);*(char*)(udphdr + length) = '\0';//这个操作有必要吗?struct in_addr addr;addr.s_addr = iphdr->src_addr;printf("src:%s:%d,",inet_ntoa(addr),ntohs(udphdr->src_port));addr.s_addr = iphdr->dst_addr;printf("dst:%s:%d, %s\n",inet_ntoa(addr),ntohs(udphdr->dst_port),(char*)(udphdr + 1));}}rte_pktmbuf_free(mbuf);
}static int
main_loop(__attribute__((unused)) void *dummy)
{struct rte_mbuf *mbufs[BURST_SIZE];int num_recvd = 0;unsigned target_lcore_id_1 = 1;unsigned target_lcore_id_2 = 2;int portid = 0;unsigned lcore_id = rte_lcore_id();int i;if(lcore_id != target_lcore_id_1 && lcore_id != target_lcore_id_2){return 0;}//核1接收port0的消息//核2接收port1的消息if(lcore_id == target_lcore_id_1){portid = 0;}else if(lcore_id == target_lcore_id_2){portid = 1;}else{return 0;}while(1){num_recvd = rte_eth_rx_burst(portid, 0, mbufs, BURST_SIZE);if(num_recvd > BURST_SIZE){rte_exit(EXIT_FAILURE, "rte_eth_rx_burst failed\n");}for(i = 0; i < num_recvd; i++){pkt_process(mbufs[i]);}}return 0;
}int main(int argc, char *argv[]) {unsigned lcore_id;// init ealif(rte_eal_init(argc, argv) < 0){rte_exit(EXIT_FAILURE, "Error with eal init\n");}// alloc memoryinit_mem(nb_mbuf);// create tx queue and rx queueinit_port();// main_looprte_eal_mp_remote_launch(main_loop, NULL, CALL_MASTER);RTE_LCORE_FOREACH_SLAVE(lcore_id) {if (rte_eal_wait_lcore(lcore_id) < 0){return -1;}}// freefor(int i = 0; i < NB_SOCKETS; i++){rte_mempool_free(pktmbuf_pool[i]);}return 0;
}
