5.DPDK Ring Library-白红宇

5.DPDK Ring Library

阅读量：4057 次

发布时间：2019-05-25

本文共 11030 字，大约阅读时间需要 36 分钟。

原文：

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驱动程序中）：

5.2 Linux中无锁Ring的实现*

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位的整数。