今天我们看一个ring环在项目中的使用。我们选的例子是dpdk官方提供的使用实例。

一个ring环的例子

例子的位置在

dpdk-stable-18.11.2/examples/packet_ordering

我们下载的dpdk版本是dpdk-stable-18.11.2

例子packet_ordering是展示packet ordering的，里面需要多线程之间传递报文，使用的是ring。

如上图，这是代码实现的线程、ring环组成的报文处理流程

1、 接收线程rx_thread（1个线程）

2、 报文处理线程worker_thread（若干个）

3、 发送线程send_thread（1个）

报文由rx_thread接收，经过ring环（rx_to_workers），发送给worker_thread线程处理报文，处理后通过ring环（workers_to_tx）发送给发送线程send_thread。

代码中看创建ring

1、 创建了两个ring环；

2、 分别是rx_to_workers 和workers_to_tx。这两个名称是标识ring的位置标志，不会重复，并且每个ring唯一一个名称。

3、 rte_socket_id，ring必须归属于一个唯一的socket，即现有的服务器框架都是多numa框架，标识这个ring是属于哪一个numa的。

4、 rx_to_workers这个ring的参数RING_F_SP_ENQ，SP_ENQ，意思是single product enquence，即单个消息产出并入ring的环，说明这个ring环是一个线程进行消息入环。

5、 workers_to_tx这个ring的参数RING_F_SC_DEQ，同理，SC DEQ 意思是single consumer dequence，即出ring的线程是单个线程。

线程入ring和出ring代码

rx_thread入ring代码

/* enqueue to rx_to_workers ring */

ret = rte_ring_enqueue_burst(ring_out,(void *)pkts, nb_rx_pkts, NULL);

worker_thread线程读ring的代码

/* dequeue the mbufs from rx_to_workers ring */

burst_size = rte_ring_dequeue_burst(ring_in,(void *)burst_buffer, MAX_PKTS_BURST, NULL);

worker_thread线程入ring的代码

/* enqueue the modified mbufs to workers_to_tx ring */

ret = rte_ring_enqueue_burst(ring_out, (void *)burst_buffer,burst_size, NULL);

send_thread读ring的代码

/* deque the mbufs from workers_to_tx ring */

nb_dq_mbufs = rte_ring_dequeue_burst(args->ring_in,(void *)mbufs, MAX_PKTS_BURST, NULL);