企业官网建设流程全解析

仓库：wait/1010-dpdk

之前的代码都是写在main中的while里面，接收数据，解析协议帧，打包协议帧，发送数据，定时器发送arp请求，都是在main loop中。

这种写法不符合实际的开发规范，现在对整体的代码架构进行优化，对接收的数据异步解耦。

收发数据 与 协议帧的解析打包 实现分离。异步解耦。

创建两个队列，一个in ring 一个out ring。

接收到的数据都放在in ring中，发送数据时就从out ring中获取数据。

main loop只实现俩三个功能：

1. 接收数据，放入in ring
2. 发送数据，从out ring中取数据
3. 定时器发送arp请求

再开辟一个线程用来pkt_process专门处理协议帧的解析和打包，打包好的协议帧放入out ring中。

实现收发与解析相互分离，只通过两个队列实现交互。

两个队列的设计使用的是dpdk提供的环形队列。

队列中的元素是rte_mbuf。

使用单例模式，创建一个全局变量，用来管理in ring和out ring。

struct inout_ring { struct rte_ring *in; struct rte_ring *out; }; extern struct inout_ring *rInst;

先进行初始化

struct inout_ring *ringInit(void) { if(rInst == NULL) { rInst = rte_malloc("in_out_ring",sizeof(struct inout_ring),0); memset(rInst,0,sizeof(struct inout_ring)); } return rInst; } struct inout_ring *ring = ringInit(); if(ring == NULL){ rte_exit(EXIT_FAILURE,"ring init fail\n"); } if(ring->in == NULL){ ring->in = rte_ring_create("in_ring",RING_SIZE,rte_socket_id(),RING_F_SP_ENQ | RING_F_SC_DEQ); } if(ring->out == NULL){ ring->out = rte_ring_create("out_ring",RING_SIZE,rte_socket_id(),RING_F_SP_ENQ | RING_F_SC_DEQ); }

通过ringInit确保只有一个全局变量rInst，然后对rInst中的in ring 和 out ring申请内存。

rte_ring_create(“out_ring”,RING_SIZE,rte_socket_id(),RING_F_SP_ENQ | RING_F_SC_DEQ);

out_ring: 名字
RING_SIZE: ring的容量，最多能够存放多少rte_mbuf 指针
rte_socket_id(): 获取当前的NUMA结点
RING_F_SP_ENQ: 单生产者入队
RING_F_SC_DEQ: 单消费者出队

NUMA：CPU临近的内存块，CPU访问其的速度快于访问其他内存的速度。一个 NUMA 节点可能包含多个 CPU 核。

S：single 单个

P：producer 生产者 入队，写入队列

C：consumer 消费者 出队，从队列中获取数据

ENQ：enqueue 入队

DEQ：dequeue 出队

对于入队出队操作，dpdk提供了很多种接口。

// 多生产者入队 rte_ring_mp_enqueue_burst(); // 单生产者入队 rte_ring_sp_enqueue_burst(); // 都可以 rte_ring_enqueue_burst(); // 多消费者 出队 rte_ring_mc_dequeue_burst(); // 单消费者 出队 rte_ring_sc_dequeue_burst(); // 都可以 rte_ring_dequeue_burst();

对于我们的架构，有两个线程，mian线程和，pkt process线程。

mian线程：in ring入队。out ring 出队。

pkt process线程：in ring出队。out ring入队。

对于in ring和out ring的入队出队操作都只有一个线程，所以使用单生产者入队，单消费者出队即可。

对于in ring和out ring的释放。

in ring的释放在pkt process线程中释放。

out ring在main线程中释放。

通常对线程是把任务丢给这个线程去做，做完后需要返回。但是dpdk的多线程不需要返回，线程之间通过队列进行通信，而不是return。一个dpdk的线程对应一个 lcore ，logical core，DPDK的逻辑核。通常一个逻辑核对应一个cpu核心。

每个线程对对应一个lcore，这个lcore长期存在的这个任务函数。

/****************************** in ring******************************/ struct rte_mbuf *rx[BURST_SIZE]; unsigned num_recvd = rte_eth_rx_burst(gDpdkPortId,0,rx,BURST_SIZE); if(num_recvd > BURST_SIZE) { rte_exit(EXIT_FAILURE,"rx burst get fail\n"); }else if( num_recvd > 0){ printf("in ring \n"); // tx,rx 是一个数组,都指向第一个元素(void **)rx 与 (void **)&rx 效果完全一致 // 而pkt_process_w中的指针,所以必须加& rte_ring_sp_enqueue_burst(ring->in,(void **)&rx,num_recvd,NULL); } /****************************** out ring ******************************/ struct rte_mbuf *tx[BURST_SIZE]; unsigned num_tx = rte_ring_sc_dequeue_burst(ring->out,(void **)&tx,BURST_SIZE,NULL); if( num_tx > 0) { printf("out ring \n"); rte_eth_tx_burst(gDpdkPortId,0,tx,num_tx); unsigned i = 0; for(i = 0;i<num_tx;i++){ rte_pktmbuf_free(tx[i]); } }

文章参考<零声教育>的C/C++linux服务期高级架构系统教程学习:0voice · GitHub