0x01 缘由 要学习一个框架原理，就得首先学会如何使用。此库实现一个环形队列，在多核编程场景下，做了许多性能优化，下面简单的从编程手册中摘录一些，做为学习的基础。 0x02 概述 该环允许队列做管理。 代替用链表实现的无大小限制的环形队列，rte_ring具有以下属性： ?FIFO（先进先出队列） ?最大大小固定，指针存储在表格中（队列中存元素的指针） ?最小化锁实现（单生产者单消费者模式无锁实现、多生产者多消费者实际上利用CAS等操作） ?多消费或单消费者出队 ?多生产者或单生产者入队 ?批量出队 - 如果成功，将指定的出队对象数量;
否则失败 ?批量入队 - 如果成功，将指定的入队对象数量;
否则失败 ?突发出队 - 如果指定的计数不能满足计数，将最大数量可用对象出队 实现 ?突发入队 - 如果指定的计数不能满足计数 ，将最大数量可用对象入队 相比用链表实现的环形队列，这个队列的优势是： ? 更快;
只需要一个比较和交换指令sizeof（void *）而不是几个双重比较交换指令。 ? 比完全无锁队列简单。 ? 适应批量入队/出队操作。由于指针存储在表中，因此几个对象的出队将不会产生与链接队列中一样多的高速缓存未命中。 此外，许多对象的批量出库并不比简单的出队成本高。 目的。 缺点： ?尺寸固定（在使用前的值比较难估计，大部分只能依赖测试值和经验值） ?内存占用要比链表实现高。一个空的环包含至少N个指针。 下面一个简单的图示： 
文章图片
0x02 API描述 1.创建一个ring对象。 接口： /*Create a new ring named *name* in memory.*/ struct rte_ring *rte_ring_create(const char *name, unsigned count, int socket_id, unsigned flags);
2.出入队 有不同的出入队方式（单、bulk、burst）都在rte_ring.h中 static inline int __attribute__((always_inline)) rte_ring_enqueue(struct rte_ring *r, void *obj) static inline int __attribute__((always_inline)) rte_ring_dequeue(struct rte_ring *r, void **obj_p) 还有一些特性出入队API. 0x03 使用场景 用于多进程或多线程之间通信。 0x04 简单使用 具体原理不做深究，主要研究其特性，然后如何使用。转载于https://www.qcloud.com/community/article/542268 F-Stack是多进程架构，去除了进程间资源共享来达到更高的性能，但还是有部分控制信息需要在进程间同步，使用rte_ring让多个进程间的通信变得十分简单。 rte_ring在F-Stack中主要用于: 1.各个进程间ARP包的广播 2.KNI的转发 3.与工具(sysctl等)进行通信。 rte_ring是一个用CAS实现的无锁FIFO环形队列，支持多消费者/生产者同时出入队列，常用于多线程/多进程之间的通信。具体原理可以查看DPDK官方文档或者阅读源码，本文主要是介绍如何使用rte_ring来进行多进程间通信。 rte_ring需要与rte_mempool配合使用，通过rte_mempool来共享内存。 首先primary进程创建ring和mempool，secondary进程在primary进程启动后，通过rte_ring_lookup和rte_mempool_lookup来获取ring和mempool的地址。 primary:
// flags:标识是单消费者/生产者或者多消费者/生产者struct rte_ring *ring = rte_ring_create("message_ring",
ring_size, rte_socket_id(), flags);
struct rte_mempool *message_pool = rte_mempool_create(
"message_pool", pool_size,
string_size, pool_cache, 0,
NULL, NULL, NULL, NULL,
rte_socket_id(), flags);
secondary:
struct rte_ring *ring = rte_ring_lookup("message_ring");
struct rte_mempool *message_pool = rte_mempool_lookup(
"message_pool");
使用时，rte_mempool_get从mempool中获取一个对象，然后使用rte_ring_enqueue入队列，另一个进程通过rte_ring_dequeue来出队列,使用完成后需要rte_mempool_put将对象放回mempool。 sender:
void *msg = NULL;
if (rte_mempool_get(message_pool, &msg) < 0)
pannic();
snprintf((char *)msg, string_size, "%s", "helloworld");
if (rte_ring_enqueue(ring, msg) < 0) {
rte_mempool_put(message_pool, msg);
}
receiver:while (!quit){
void *msg;
if (rte_ring_dequeue(recv_ring, &msg) < 0){
usleep(5);
continue;
}printf("Received: '%s'\n", (char *)msg);
rte_mempool_put(message_pool, msg);
}实际代码可以参考dpdk example/multi_process/simple_mp 或者F-Stack lib/ff_dpdk_if.c和tools/ipc，非常简单易用。
0x05 多核编程 操作原子性：在整个系统可见范围内，一个操作要不就没有发生，要不就执行完毕，没有中间状态出现。 常见的 X86 CPU 来说，根据 Intel 的参考手册，它基于以下三种机制保证了操作原子性： （1）Guaranteed atomic operations （2）Bus locking, using the LOCK# signal and the LOCK instruction prefix （3）Cache coherency protocols that ensure that atomic operations can be carried out on cached data structures (cache lock); this mechanism is present in the Pentium 4, Intel Xeon, and P6 family processors
