原文：http://dpdk.org/doc/guides/prog_guide/ring_lib.html
5.Ring Library
ring 管理队列，ring不是一个无限大小的链表，它具有以下属性：
*FIFO
*大小固定，指针存储在表中
*无锁实现
*多消费者或单消费者出队
*多生产者或单生产者入队
*批量出队 - 如果成功，将指定数量的对象出队; 否则失败
*批量入库 - 如果成功，将指定数量的对象入队; 否则失败
*爆发出队 - 如果指定数量的对象无法满足，则将最大可用数量的对象出队
*爆发入队 - 如果指定数量的对象无法满足，则将最大可用数量的对象入队

这种数据结构优于链表队列的
优点如下：
*更快：比较void *大小的数据，只需要执行单次Compare-And-Swap指令，而不需要执行2次Compare-And-Swap指令
*比完全无锁队列简单
*适用于批量入队/出队操作。因为指针存储在表中，多个对象出队并不会像链表队列那样产生大量的缓存未命中，此外，多个对象批量出队不会比单个对象出队开销大
CAS(Compare and Swap)是个原子操作

缺点如下：
*大小固定
*许多环在内存方面的成本比链表列表的成本更高。空环至少包含N个指针。
存储在数据结构中的consumer head，consumer tail，producer head，producer tail简单的表示了ring的逻辑结构

文章图片


5.1 FreeBSD中Ring实现的参考*
在FreeBSD 8.0中添加了以下代码，并在某些网络设备驱动程序中使用（至少在Intel驱动程序中）：
FreeBSD中的bufring.h
FreeBSD中的bufring.c
5.2 Linux中无锁Ring的实现*
以下是描述Linux无锁环缓冲设计的链接

5.3 附加特性
5.3.1 名字
一个Ring被唯一的名字识别，创建两个名字相同的Ring是不可能的（当尝试这样操作时，rte_ring_create()函数在第二次执行时返回NULL）。
5.4 用例
Ring库的用例包括：
*DPDK应用之间的信息交互
*内存池中的使用
5.5 剖析Ring缓存
本节介绍环形缓存的运作方式。ring结构由两对head，tail组成，一对被生产者使用，一对被消费者使用，在下面的图形中将他们称为prod_head，prod_tail,，cons_head 和 ons_tail。
每种图形代表了环形缓存ring的一个简单状态。局部变量在图形的上面表示，ring结构相关变量在图形的下面表示。
备注：在下面的图解中，红色方框链表代表ring，红色方框上面的变量代表代码实现的局部变量，下面的代表存储在ring结构体里的索引，注意区分哦
5.5.1 单生产者入队
本节介绍当单生产者添加一个对象到ring时发生了什么。在这个例子中，仅仅只有一个生产者，仅仅只有生产者的head和tail（prod_head and prod_tail）索引被修改了。
在初始状态， prod_head 和 prod_tail指向相同的位置。
\lib\librte_ring\rte_ring.h

/** * @internal Enqueue several objects on a ring (NOT multi-producers safe). * * @param r *A pointer to the ring structure. * @param obj_table *A pointer to a table of void * pointers (objects). * @param n *The number of objects to add in the ring from the obj_table. * @param behavior *RTE_RING_QUEUE_FIXED:Enqueue a fixed number of items from a ring *RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring * @return *Depend on the behavior value *if behavior = RTE_RING_QUEUE_FIXED *- 0: Success; objects enqueue. *- -EDQUOT: Quota exceeded. The objects have been enqueued, but the *high water mark is exceeded. *- -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued. *if behavior = RTE_RING_QUEUE_VARIABLE *- n: Actual number of objects enqueued. */ static inline int __attribute__((always_inline)) __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table, unsigned n, enum rte_ring_queue_behavior behavior) { uint32_t prod_head, cons_tail; uint32_t prod_next, free_entries; unsigned i; uint32_t mask = r->prod.mask; int ret; prod_head = r->prod.head; cons_tail = r->cons.tail; /* The subtraction is done between two unsigned 32bits value * (the result is always modulo 32 bits even if we have * prod_head > cons_tail). So 'free_entries' is always between 0 * and size(ring)-1. */ free_entries = mask + cons_tail - prod_head; /* check that we have enough room in ring */ if (unlikely(n > free_entries)) { if (behavior == RTE_RING_QUEUE_FIXED) { __RING_STAT_ADD(r, enq_fail, n); return -ENOBUFS; } else { /* No free entry available */ if (unlikely(free_entries == 0)) { __RING_STAT_ADD(r, enq_fail, n); return 0; }n = free_entries; } } prod_next = prod_head + n; r->prod.head = prod_next; /* write entries in ring */ ENQUEUE_PTRS(); rte_smp_wmb(); /* if we exceed the watermark */ if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) { ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT : (int)(n | RTE_RING_QUOT_EXCEED); __RING_STAT_ADD(r, enq_quota, n); } else { ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n; __RING_STAT_ADD(r, enq_success, n); } r->prod.tail = prod_next; return ret; }

5.5.1.1 入队第一步
首先，局部变量保存ring->prod_head 和 ring->cons_tail。 prod_next局部变量指向prod_head的下一个元素，若是批量入队就指prod_head的下几个元素。
假如ring里没有足够的空间（检查cons_tail获知），入队函数将返回error

文章图片


5.5.1.2.入队第二步
第二步是修改ring结构体里的ring->prod_head 索引，将它指向上面提到的局部变量prod_next指向的位置。
接下来就是将添加对象(下图ring中的obj4)的索引复制到ring的里面。

文章图片


5.5.1.3. 入队最后一步
一旦添加对象被复制到ring后，ring结构体里的 ring->prod_tail索引将指向 ring->prod_head的位置，入队操作完成。

文章图片


5.5.2. 单消费者出队
本节介绍单消费者从ring取出一个对象时发生了什么。在这个例子中，仅仅只有一个消费者，仅仅只有消费者的head和tail（cons_head and cons_tail）索引被修改了。
在初始状态， cons_head 和 cons_tail指向相同的位置。

/** * @internal Dequeue several objects from a ring (NOT multi-consumers safe). * When the request objects are more than the available objects, only dequeue * the actual number of objects * * @param r *A pointer to the ring structure. * @param obj_table *A pointer to a table of void * pointers (objects) that will be filled. * @param n *The number of objects to dequeue from the ring to the obj_table. * @param behavior *RTE_RING_QUEUE_FIXED:Dequeue a fixed number of items from a ring *RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring * @return *Depend on the behavior value *if behavior = RTE_RING_QUEUE_FIXED *- 0: Success; objects dequeued. *- -ENOENT: Not enough entries in the ring to dequeue; no object is *dequeued. *if behavior = RTE_RING_QUEUE_VARIABLE *- n: Actual number of objects dequeued. */ static inline int __attribute__((always_inline)) __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table, unsigned n, enum rte_ring_queue_behavior behavior) { uint32_t cons_head, prod_tail; uint32_t cons_next, entries; unsigned i; uint32_t mask = r->prod.mask; cons_head = r->cons.head; prod_tail = r->prod.tail; /* The subtraction is done between two unsigned 32bits value * (the result is always modulo 32 bits even if we have * cons_head > prod_tail). So 'entries' is always between 0 * and size(ring)-1. */ entries = prod_tail - cons_head; if (n > entries) { if (behavior == RTE_RING_QUEUE_FIXED) { __RING_STAT_ADD(r, deq_fail, n); return -ENOENT; } else { if (unlikely(entries == 0)){ __RING_STAT_ADD(r, deq_fail, n); return 0; }n = entries; } } cons_next = cons_head + n; r->cons.head = cons_next; /* copy in table */ DEQUEUE_PTRS(); rte_smp_rmb(); __RING_STAT_ADD(r, deq_success, n); r->cons.tail = cons_next; return behavior == RTE_RING_QUEUE_FIXED ? 0 : n; }

5.5.2.1 出队第一步

首先，局部变量保存ring->cons_head 和 ring->prod_tail。 cons_next局部变量指向cons_head的下一个元素，若是批量出队就指向cons_head的下几个元素。
假如ring里没有足够的空间（prod_tail检查获知），出队函数将返回error

文章图片


5.5.2.2. 出队第二步

第二步是修改ring结构体里的ring->cons_head 索引，将它指向上面提到的局部变量cons_next指向的位置。
接下来就是将对象(下图ring中的obj1)的指针复制到用户传进来的指针中。

文章图片


5.5.2.3. 出队最后一步
最后，ring结构体中的 ring->cons_tail索引指向和 ring->cons_head，局部变量cons_next相同的位置（obj2的位置）。出队操作完成。

文章图片

5.5.3 多生产者入队
本节介绍当两个生产者同时添加对象到ring时发生了什么。在这个例子中，仅仅只有生产者的head和tail（prod_head and prod_tail）索引被修改了。
在初始状态， prod_head 和 prod_tail指向相同的位置。

/** * @internal Enqueue several objects on the ring (multi-producers safe). * * This function uses a "compare and set" instruction to move the * producer index atomically. * * @param r *A pointer to the ring structure. * @param obj_table *A pointer to a table of void * pointers (objects). * @param n *The number of objects to add in the ring from the obj_table. * @param behavior *RTE_RING_QUEUE_FIXED:Enqueue a fixed number of items from a ring *RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring * @return *Depend on the behavior value *if behavior = RTE_RING_QUEUE_FIXED *- 0: Success; objects enqueue. *- -EDQUOT: Quota exceeded. The objects have been enqueued, but the *high water mark is exceeded. *- -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued. *if behavior = RTE_RING_QUEUE_VARIABLE *- n: Actual number of objects enqueued. */ static inline int __attribute__((always_inline)) __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table, unsigned n, enum rte_ring_queue_behavior behavior) { uint32_t prod_head, prod_next; uint32_t cons_tail, free_entries; const unsigned max = n; int success; unsigned i, rep = 0; uint32_t mask = r->prod.mask; int ret; /* Avoid the unnecessary cmpset operation below, which is also * potentially harmful when n equals 0. */ if (n == 0) return 0; /* move prod.head atomically */ do { /* Reset n to the initial burst count */ n = max; prod_head = r->prod.head; cons_tail = r->cons.tail; /* The subtraction is done between two unsigned 32bits value * (the result is always modulo 32 bits even if we have * prod_head > cons_tail). So 'free_entries' is always between 0 * and size(ring)-1. */ free_entries = (mask + cons_tail - prod_head); /* check that we have enough room in ring */ if (unlikely(n > free_entries)) { if (behavior == RTE_RING_QUEUE_FIXED) { __RING_STAT_ADD(r, enq_fail, n); return -ENOBUFS; } else { /* No free entry available */ if (unlikely(free_entries == 0)) { __RING_STAT_ADD(r, enq_fail, n); return 0; }n = free_entries; } }prod_next = prod_head + n; success = rte_atomic32_cmpset(&r->prod.head, prod_head, prod_next); } while (unlikely(success == 0)); /* write entries in ring */ ENQUEUE_PTRS(); rte_smp_wmb(); /* if we exceed the watermark */ if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) { ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT : (int)(n | RTE_RING_QUOT_EXCEED); __RING_STAT_ADD(r, enq_quota, n); } else { ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n; __RING_STAT_ADD(r, enq_success, n); } /* * If there are other enqueues in progress that preceded us, * we need to wait for them to complete */ while (unlikely(r->prod.tail != prod_head)) { rte_pause(); /* Set RTE_RING_PAUSE_REP_COUNT to avoid spin too long waiting * for other thread finish. It gives pre-empted thread a chance * to proceed and finish with ring dequeue operation. */ if (RTE_RING_PAUSE_REP_COUNT && ++rep == RTE_RING_PAUSE_REP_COUNT) { rep = 0; sched_yield(); } } r->prod.tail = prod_next; return ret; }

5.5.3.1 多生产者入队第一步
在两个生产者core中（这个core可以理解成同时运行的线程或进程），各自的局部变量都保存ring->prod_head 和 ring->cons_tail。 各自的局部变量prod_next索引指向ring->prod_head的下一个元素，如果是批量入队，指向下几个元素。
假如ring里没有足够的空间（检查cons_tail获知），入队函数将返回error


文章图片


5.5.3.2. 多生产者入队第二步

第二步是修改ring结构体里的ring->prod_head 索引，将它指向上面提到的局部变量prod_next指向的位置。这个操作是通过使用 Compare And Swap (CAS)执行完成的， Compare And Swap (CAS)包含以下原子操作：
*如果ring->prod_head索引和局部变量prod_head索引不相等，CAS操作失败，代码将从新从第一步开始执行。
*否则，将ring->prod_head索引指向局部变量prod_next的位置，CAS操作成功，继续下一步处理。
在下图中，生产者core1执行成功，生产者core2重新运行。

文章图片


5.5.3.3. 多生产者入队第三步
生产者core2中CAS指令重试成功
生产者core1更新对象obj4到ring中，生产者core2更新对象obj5到ring中（CAS指令重试后执行成功的）。

文章图片

5.5.3.4. 多生产者入队第四步
现在每个生产者core都想更新 ring->prod_tail索引。生产者core代码中，只有ring->prod_tail等于自己局部变量prod_head才能被更新，显然从上图中可知，只有生产者core1才能满足，生产者core1完成了入队操作。

文章图片


5.5.3.5. 多生产者入队最后一步
一旦生产者core1更新了ring->prod_tail后，生产者core2也可以更新ring->prod_tail了。生产者core2也完成了入队操作

文章图片


5.5.4 32位取模索引
在前面的图例中，prod_head, prod_tail, cons_head 和 cons_tail 都是用箭头表示的。但在实际的代码实现中，他们的值并不是0和ring大小减一之间的数值。索引的大小范围是0----2^32-1，当访问ring中的数据时，真正的索引等于ring中索引值和掩码与之后的值。32 bit取模的意思是如果索引操作（加减）的结果的值超出了32 bit数据的范围，溢出的值忽略，只看省下的位组成的数。
下面的两个例子帮助解释索引在ring中如何使用的
！Note
为了简便，例子操作的是16位而不是32位。另外，关键的四个索引也被定义成16位的整数，现实代码实现是用得32位的整数。

文章图片


ring包含了11000条目

文章图片


ring包含了12536个条目
！Note
为了便于理解，我们在上面的例子中使用模65536的操作。在真实代码实现中，这是冗长低效的，但是当结果溢出时是自动完成的。
代码实现总是将producer 和 consumer保持0----ring大小减一的距离。 这个特性的好处是我们能在两个32位索引值之间做减法，且差值永远在0----ring大小减一范围内：这也是为什么结果溢出不是什么大问题。
在任何时候，已经使用的条目和空闲的条目永远在0----ring大小减一之间。
uint32_t entries = (prod_tail - cons_head);
uint32_t free_entries = (mask + cons_tail -prod_head);

5.6 参考索引
【5.DPDK Ring Library】bufring.h in FreeBSD (version 8)
bufring.c in FreeBSD (version 8)
Linux Lockless Ring Buffer Design

推荐阅读

• 

  玉米面条怎么做好吃简单的做法
• 

  吃完洋葱怎么去味
• 

  商标注册的条件 哪些企业需要申请商标，怎么申请商标
• 

  云虚拟主机教学，云虚拟主机使用教程
• 

  vue|vue 如何实现表单校验
• 

  林蛙养殖技术大全视频 林蛙养殖技术
• 

  守夜是什么意思
• 

  如何去除保温杯硅胶圈异味 去除保温杯硅胶圈异味的方法
• 

  苹果手机迅雷一直闪退怎么办？iPhone迅雷闪退解决方法
• 

  安卓vr开发,安卓手机下载utovr软件点击浏览
• 

  吸引眼球好听的支付宝名字 唯美吸引眼球好听的支付宝名字
• 

  spss 生存分析,SPSS生存分析
• 

  去海边沙滩玩戴什么太阳镜最好
• 

  有哪些三维建模软件？三维绘图软件有哪些?
• 

  离合踏板是什么意思? 离合踏板啥意思
• 

  mysqladmin ping mysqlping方法
• 

  慢性病人群可以接种新冠病毒疫苗吗
• 

  csf 关键成功要素法分析的步骤
• 

  里约奥运会中国代表队旗手 2016里约奥运会中国队旗手会是谁
• 

  新冠疫苗加强针接种的注意事项

• DPDK中文-DPDK的ring
• 【DPDK】虚拟机开发环境配置