本文整理下之前的学习笔记，基于DPDK17.11版本源码，主要分析无锁队列ring的实现。

rte_ring_tailq保存rte_ring链表

创建ring后会将其插入共享内存链表rte_ring_tailq，以便主从进程都可以访问。

//定义队列头结构 struct rte_tailq_elem_head
TAILQ_HEAD(rte_tailq_elem_head, rte_tailq_elem);//声明全局变量rte_tailq_elem_head，类型为struct rte_tailq_elem_head，
//相当于是链表头，用来保存本进程注册的队列
/* local tailq list */
static struct rte_tailq_elem_head rte_tailq_elem_head =TAILQ_HEAD_INITIALIZER(rte_tailq_elem_head);//调用EAL_REGISTER_TAILQ在main函数前注册rte_ring_tailq到全局变量rte_tailq_elem_head。
#define RTE_TAILQ_RING_NAME "RTE_RING"
static struct rte_tailq_elem rte_ring_tailq = {.name = RTE_TAILQ_RING_NAME,
};
EAL_REGISTER_TAILQ(rte_ring_tailq)

调用rte_eal_tailq_update遍历链表rte_tailq_elem_head上的节点，将节点中的head指向 struct rte_mem_ring->tailq_head[]数组中的一个tailq_head，此head又作为另一个链表头。比如注册的rte_ring_tailq节点，其head专门用来保存创建的rte_ring(将rte_ring作为struct rte_tailq_entry的data，将struct rte_tailq_entry插入head)。前面说过struct rte_mem_ring->tailq_head存放在共享内存中，主从进程都可以访问，这样对于rte_ring来说，主从进程都可以创建/访问ring。

创建ring

调用函数rte_ring_create创建ring，它会申请一块memzone的内存，大小为struct rte_ring结构加上count个void类型指针，内存结构如下

然后将ring中生产者和消费者的头尾指向0，最后将ring作为struct rte_tailq_entry的data插入共享内存链表，这样主从进程都可以访问此ring。

/*** An RTE ring structure.** The producer and the consumer have a head and a tail index. The particularity* of these index is that they are not between 0 and size(ring). These indexes* are between 0 and 2^32, and we mask their value when we access the ring[]* field. Thanks to this assumption, we can do subtractions between 2 index* values in a modulo-32bit base: that's why the overflow of the indexes is not* a problem.*/
struct rte_ring {/** Note: this field kept the RTE_MEMZONE_NAMESIZE size due to ABI* compatibility requirements, it could be changed to RTE_RING_NAMESIZE* next time the ABI changes*/char name[RTE_MEMZONE_NAMESIZE] __rte_cache_aligned; /**< Name of the ring. *///flags有如下三个值://RING_F_SP_ENQ创建单生产者，//RING_F_SC_DEQ创建单消费者，//RING_F_EXACT_SZint flags;               /**< Flags supplied at creation. *///memzone内存管理的底层结构，用来分配内存const struct rte_memzone *memzone;/**< Memzone, if any, containing the rte_ring *///size为ring大小，值和RING_F_EXACT_SZ有关，如果指定了flag     //RING_F_EXACT_SZ，则size为rte_ring_create的参数count的//向上取2次方，比如count为15，则size就为16。如果没有指定//flag，则count必须是2的次方，此时size等于countuint32_t size;           /**< Size of ring. *///mask值为size-1uint32_t mask;           /**< Mask (size-1) of ring. *///capacity的值也和RING_F_EXACT_SZ有关，如果指定了，//则capacity为rte_ring_create的参数count，如果没指定，//则capacity为size-1uint32_t capacity;       /**< Usable size of ring *///生产者位置，包含head和tail，head代表着下一次生产时的起//始位置。tail代表消费者可以消费的位置界限，到达tail后就无  //法继续消费，通常情况下生产完成后tail = head，意味着刚生//产的元素皆可以被消费/** Ring producer status. */struct rte_ring_headtail prod __rte_aligned(PROD_ALIGN);// 消费者位置，也包含head和tail，head代表着下一次消费时的//起始位置。tail代表生产者可以生产的位置界限，到达tail后就//无法继续生产，通常情况下消费完成后，tail =head，意味着//刚消费的位置皆可以被生产/** Ring consumer status. */struct rte_ring_headtail cons __rte_aligned(CONS_ALIGN);
};

下面看一下在函数rte_ring_create中ring是如何被创建的。

/* create the ring */
struct rte_ring *
rte_ring_create(const char *name, unsigned count, int socket_id, unsigned flags)
{char mz_name[RTE_MEMZONE_NAMESIZE];struct rte_ring *r;struct rte_tailq_entry *te;const struct rte_memzone *mz;ssize_t ring_size;int mz_flags = 0;struct rte_ring_list* ring_list = NULL;const unsigned int requested_count = count;int ret;//(tailq_entry)->tailq_head 的类型应该是 struct rte_tailq_entry_head，//但是返回的却是 struct rte_ring_list，因为 rte_tailq_entry_head 和 rte_ring_list 定义都是一样的，//可以认为是等同的。#define RTE_TAILQ_CAST(tailq_entry, struct_name) \(struct struct_name *)&(tailq_entry)->tailq_headring_list = RTE_TAILQ_CAST(rte_ring_tailq.head, rte_ring_list);/* for an exact size ring, round up from count to a power of two */if (flags & RING_F_EXACT_SZ)count = rte_align32pow2(count + 1);//获取需要的内存大小，包括结构体 struct rte_ring 和 count 个指针ring_size = rte_ring_get_memsize(count);ssize_t sz;sz = sizeof(struct rte_ring) + count * sizeof(void *);sz = RTE_ALIGN(sz, RTE_CACHE_LINE_SIZE);#define RTE_RING_MZ_PREFIX "RG_"snprintf(mz_name, sizeof(mz_name), "%s%s", RTE_RING_MZ_PREFIX, name);//分配 struct rte_tailq_entry，用来将申请的ring挂到共享链表ring_list中te = rte_zmalloc("RING_TAILQ_ENTRY", sizeof(*te), 0);rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);//申请memzone，/* reserve a memory zone for this ring. If we can't get rte_config or* we are secondary process, the memzone_reserve function will set* rte_errno for us appropriately - hence no check in this this function */mz = rte_memzone_reserve_aligned(mz_name, ring_size, socket_id, mz_flags, __alignof__(*r));if (mz != NULL) {//memzone的的addr指向分配的内存，ring也从此内存开始r = mz->addr;/* no need to check return value here, we already checked the* arguments above */rte_ring_init(r, name, requested_count, flags);//将ring保存到链表entry中te->data = (void *) r;r->memzone = mz;//将链表entry插入链表ring_listTAILQ_INSERT_TAIL(ring_list, te, next);} else {r = NULL;RTE_LOG(ERR, RING, "Cannot reserve memory\n");rte_free(te);}rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);return r;
}int
rte_ring_init(struct rte_ring *r, const char *name, unsigned count,unsigned flags)
{int ret;/* compilation-time checks */RTE_BUILD_BUG_ON((sizeof(struct rte_ring) &RTE_CACHE_LINE_MASK) != 0);RTE_BUILD_BUG_ON((offsetof(struct rte_ring, cons) &RTE_CACHE_LINE_MASK) != 0);RTE_BUILD_BUG_ON((offsetof(struct rte_ring, prod) &RTE_CACHE_LINE_MASK) != 0);/* init the ring structure */memset(r, 0, sizeof(*r));ret = snprintf(r->name, sizeof(r->name), "%s", name);if (ret < 0 || ret >= (int)sizeof(r->name))return -ENAMETOOLONG;r->flags = flags;r->prod.single = (flags & RING_F_SP_ENQ) ? __IS_SP : __IS_MP;r->cons.single = (flags & RING_F_SC_DEQ) ? __IS_SC : __IS_MC;if (flags & RING_F_EXACT_SZ) {r->size = rte_align32pow2(count + 1);r->mask = r->size - 1;r->capacity = count;} else {if ((!POWEROF2(count)) || (count > RTE_RING_SZ_MASK)) {RTE_LOG(ERR, RING,"Requested size is invalid, must be power of 2, and not exceed the size limit %u\n",RTE_RING_SZ_MASK);return -EINVAL;}r->size = count;r->mask = count - 1;r->capacity = r->mask;}//初始时，生产者和消费者的首尾都为0r->prod.head = r->cons.head = 0;r->prod.tail = r->cons.tail = 0;return 0;
}

入队操作

DPDK提供了如下几个api用来执行入队操作，它们最终都会调用__rte_ring_do_enqueue来实现，所以重点分析函数__rte_ring_do_enqueue。

//多生产者批量入队。入队个数n必须全部成功，否则入队失败。调用者明确知道是多生产者
rte_ring_mp_enqueue_bulk
//单生产者批量入队。入队个数n必须全部成功，否则入队失败。调用者明确知道是单生产者
rte_ring_sp_enqueue_bulk
//批量入队。入队个数n必须全部成功，否则入队失败。调用者不用关心是不是单生产者
rte_ring_enqueue_bulk
//多生产者批量入队。入队个数n不一定全部成功。调用者明确知道是多生产者
rte_ring_mp_enqueue_burst
//单生产者批量入队。入队个数n不一定全部成功。调用者明确知道是单生产者
rte_ring_sp_enqueue_burst
//批量入队。入队个数n不一定全部成功。调用者不用关心是不是单生产者
rte_ring_enqueue_burst

__rte_ring_do_enqueue主要做了三个事情:
a. 移动生产者head，此处在多生产者下可能会有冲突，需要使用cas操作循环检测，只有自己能移动head时才行。
b. 执行入队操作，将obj插入ring，从老的head开始，直到新head结束。
c. 更新生产者tail，只有这样消费者才能看到最新的消费对象。

其参数r指定了目标ring。
参数obj_table指定了入队对象。
参数n指定了入队对象个数。
参数behavior指定了入队行为，有两个值RTE_RING_QUEUE_FIXED和RTE_RING_QUEUE_VARIABLE，前者表示入队对象必须一次性全部成功，后者表示尽可能多的入队。
参数is_sp指定了是否为单生产者模式，默认为多生产者模式。

static __rte_always_inline unsigned int
__rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,unsigned int n, enum rte_ring_queue_behavior behavior,int is_sp, unsigned int *free_space)
{uint32_t prod_head, prod_next;uint32_t free_entries;//先移动生产者的头指针，prod_head保存移动前的head，prod_next保存移动后的headn = __rte_ring_move_prod_head(r, is_sp, n, behavior,&prod_head, &prod_next, &free_entries);if (n == 0)goto end;//&r[1]指向存放对象的内存。//从prod_head开始，将n个对象obj_table插入ring的prod_head位置ENQUEUE_PTRS(r, &r[1], prod_head, obj_table, n, void *);rte_smp_wmb();//更新生产者tailupdate_tail(&r->prod, prod_head, prod_next, is_sp);
end:if (free_space != NULL)*free_space = free_entries - n;return n;
}

__rte_ring_move_prod_head用来使用cas操作更新生产者head。

static __rte_always_inline unsigned int
__rte_ring_move_prod_head(struct rte_ring *r, int is_sp,unsigned int n, enum rte_ring_queue_behavior behavior,uint32_t *old_head, uint32_t *new_head,uint32_t *free_entries)
{const uint32_t capacity = r->capacity;unsigned int max = n;int success;do {/* Reset n to the initial burst count */n = max;//获取生产者当前的head位置*old_head = r->prod.head;/* add rmb barrier to avoid load/load reorder in weak* memory model. It is noop on x86*/rte_smp_rmb();const uint32_t 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* *old_head > cons_tail). So 'free_entries' is always between 0* and capacity (which is < size).*///获取空闲 entry 个数*free_entries = (capacity + cons_tail - *old_head);//如果入队的对象个数大于空闲entry个数，则如果入队要求固定大小，则入队失败，返回0，否则//只入队空闲entry个数的对象/* check that we have enough room in ring */if (unlikely(n > *free_entries))n = (behavior == RTE_RING_QUEUE_FIXED) ?0 : *free_entries;if (n == 0)return 0;//当前head位置加上入队对象个数获取新的生产者head*new_head = *old_head + n;//如果是单生产者，直接更新生产者head，并返回1if (is_sp)r->prod.head = *new_head, success = 1;else //如果是多生产者，需要借助函数rte_atomic32_cmpset，比较old_head和r->prod.head是否相同，//如果相同，则将r->prod.head更新为new_head，并返回1，退出循环，//如果不相同说明有其他生产者更新head了，返回0，继续循环。success = rte_atomic32_cmpset(&r->prod.head,*old_head, *new_head);} while (unlikely(success == 0));return n;
}

ENQUEUE_PTRS定义了入队操作。

/* the actual enqueue of pointers on the ring.* Placed here since identical code needed in both* single and multi producer enqueue functions */
#define ENQUEUE_PTRS(r, ring_start, prod_head, obj_table, n, obj_type) do { \unsigned int i; \const uint32_t size = (r)->size; \uint32_t idx = prod_head & (r)->mask; \obj_type *ring = (obj_type *)ring_start; \//idx+n 大于 size，说明入队n个对象后，ring还没满，还没翻转if (likely(idx + n < size)) { \//一次循环入队四个对象for (i = 0; i < (n & ((~(unsigned)0x3))); i+=4, idx+=4) { \ring[idx] = obj_table[i]; \ring[idx+1] = obj_table[i+1]; \ring[idx+2] = obj_table[i+2]; \ring[idx+3] = obj_table[i+3]; \} \//还有剩余不满四个对象，则在switch里入队switch (n & 0x3) { \case 3: \ring[idx++] = obj_table[i++]; /* fallthrough */ \case 2: \ring[idx++] = obj_table[i++]; /* fallthrough */ \case 1: \ring[idx++] = obj_table[i++]; \} \} else { \//入队n个对象，会导致ring满，发生翻转，//则先入队idx到size的位置，for (i = 0; idx < size; i++, idx++)\ring[idx] = obj_table[i]; \//再翻转回到ring起始位置，入队剩余的对象for (idx = 0; i < n; i++, idx++) \ring[idx] = obj_table[i]; \} \
} while (0)

最后更新生产者tail。

static __rte_always_inline void
update_tail(struct rte_ring_headtail *ht, uint32_t old_val, uint32_t new_val,uint32_t single)
{/** If there are other enqueues/dequeues in progress that preceded us,* we need to wait for them to complete*/if (!single)//多生产者时，必须等到其他生产者入队成功，再更新自己的tailwhile (unlikely(ht->tail != old_val))rte_pause();ht->tail = new_val;
}

出队操作

DPDK提供了如下几个api用来执行出队操作，它们最终都会调用__rte_ring_do_dequeue来实现，所以重点分析函数__rte_ring_do_dequeue。

//多消费者批量出队。出队个数n必须全部成功，否则出队失败。调用者明确知道是多消费者
rte_ring_mc_dequeue_bulk
//单消费者批量出队。出队个数n必须全部成功，否则出队失败。调用者明确知道是单消费者
rte_ring_sc_dequeue_bulk
//批量出队。出队个数n必须全部成功，否则出队失败。调用者不用关心是不是单消费者
rte_ring_dequeue_bulk
//多消费者批量出队。出队个数n不一定全部成功。调用者明确知道是多消费者
rte_ring_mc_dequeue_burst
//单消费者批量出队。出队个数n不一定全部成功。调用者明确知道是单消费者
rte_ring_sc_dequeue_burst
//批量出队。出队个数n不一定全部成功。调用者不用关心是不是单消费者
rte_ring_dequeue_burst

__rte_ring_do_dequeue主要做了三个事情:
a. 移动消费者head，此处在多消费者下可能会有冲突，需要使用cas操作循环检测，只有自己能移动head时才行。
b. 执行出队操作，将ring中的obj插入obj_table，从老的head开始，直到新head结束。
c. 更新消费者tail，只有这样生成者才能进行生产。

其参数r指定了目标ring。
参数obj_table指定了出队对象出队后存放位置。
参数n指定了入队对象个数。
参数behavior指定了出队行为，有两个值RTE_RING_QUEUE_FIXED和RTE_RING_QUEUE_VARIABLE，前者表示出队对象必须一次性全部成功，后者表示尽可能多的出队。
参数is_sp指定了是否为单消费者模式，默认为多消费者模式。

static __rte_always_inline unsigned int
__rte_ring_do_dequeue(struct rte_ring *r, void **obj_table,unsigned int n, enum rte_ring_queue_behavior behavior,int is_sc, unsigned int *available)
{uint32_t cons_head, cons_next;uint32_t entries;//先移动消费者head，成功后，cons_head为老的head，cons_next为新的head，//两者之间的部分为此次可消费的对象n = __rte_ring_move_cons_head(r, is_sc, n, behavior,&cons_head, &cons_next, &entries);if (n == 0)goto end;//执行出队操作，从老的cons_head开始出队n个对象DEQUEUE_PTRS(r, &r[1], cons_head, obj_table, n, void *);rte_smp_rmb();//更新消费者tail，和前面更新生产者head代码相同update_tail(&r->cons, cons_head, cons_next, is_sc);end:if (available != NULL)*available = entries - n;return n;
}

__rte_ring_move_cons_head用来使用cas操作更新消费者head。

static __rte_always_inline unsigned int
__rte_ring_move_cons_head(struct rte_ring *r, int is_sc,unsigned int n, enum rte_ring_queue_behavior behavior,uint32_t *old_head, uint32_t *new_head,uint32_t *entries)
{unsigned int max = n;int success;/* move cons.head atomically */do {/* Restore n as it may change every loop */n = max;//取出当前head位置*old_head = r->cons.head;/* add rmb barrier to avoid load/load reorder in weak* memory model. It is noop on x86*/rte_smp_rmb();//生产者tail减去消费者head为可消费的对象个数。//因为head和tail都是无符号32位类型，即使生产者tail比消费者head//小，也能正确得出结果，不用担心溢出。const uint32_t 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 - *old_head);//要求出队对象个数大于实际可消费对象个数/* Set the actual entries for dequeue */if (n > *entries)//此时如果behavior为RTE_RING_QUEUE_FIXED，表示必须满足n，满足不了就一个都不出队，返回0，//如果不为RTE_RING_QUEUE_FIXED，则尽可能多的出队n = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;if (unlikely(n == 0))return 0;//当前head加上n即为新的消费者head*new_head = *old_head + n;if (is_sc)//如果单消费者，直接更新head即可，返回1r->cons.head = *new_head, success = 1;else//多消费者，需要借用rte_atomic32_cmpset更新headsuccess = rte_atomic32_cmpset(&r->cons.head, *old_head,*new_head);} while (unlikely(success == 0));return n;
}

ring是否满或者是否为空

函数rte_ring_full用来判断ring是否满
static inline int
rte_ring_full(const struct rte_ring *r)
{return rte_ring_free_count(r) == 0;
}static inline unsigned
rte_ring_free_count(const struct rte_ring *r)
{return r->capacity - rte_ring_count(r);
}

函数rte_ring_empty用来判断ring是否为空

static inline int
rte_ring_empty(const struct rte_ring *r)
{return rte_ring_count(r) == 0;
}

判断ring是否为空或者是否满都需要调用rte_ring_count获取当前ring中已使用的个数。

static inline unsigned
rte_ring_count(const struct rte_ring *r)
{uint32_t prod_tail = r->prod.tail;uint32_t cons_tail = r->cons.tail;uint32_t count = (prod_tail - cons_tail) & r->mask;return (count > r->capacity) ? r->capacity : count;
}

DPDK 无锁ring，详解相关推荐

java一个方法排他调用_Java编程实现排他锁代码详解
一 .前言某年某月某天,同事说需要一个文件排他锁功能,需求如下: (1)写操作是排他属性 (2)适用于同一进程的多线程/也适用于多进程的排他操作 (3)容错性:获得锁的进程若Crash,不影响到后续 ...
高效并发：Synchornized的锁优化详解
高效并发:Synchornized的锁优化详解 1. HotSpot虚拟机的对象头的内存布局 2. 偏向锁举一反三:当锁进入偏向状态时,存储hash码的位置被覆盖了,那对象的hash码存储到哪儿的? ...
mysql悲观锁会有脏数据吗_mysql悲观锁原理详解
mysql中的锁概念 mysql已经成为大家日常数据存储的最常用平台,但随着业务量和访问量的上涨,会出现并发访问等场景,如果处理不好并发问题的话会带来严重困扰.下面介绍一下如何通过mysql的悲观锁来 ...
mysql innodb 的锁机制_Mysql之Innodb锁机制详解
InnoDB与MyISAM的最大不同有两点:一是支持事务(TRANSACTION):二是采用了行级锁.关于事务我们之前有专题介绍,这里就着重介绍下它的锁机制. 总的来说,InnoDB按照不同的分类共有 ...
php使用redis分布式锁,php基于redis的分布式锁实例详解
在使用分布式锁进行互斥资源访问时候,我们很多方案是采用redis的实现. 固然,redis的单节点锁在极端情况也是有问题的,假设你的业务允许偶尔的失效,使用单节点的redis锁方案就足够了,简单而且效 ...
ajax 页面无刷新,Ajax的页面无刷新实现详解（附代码）
这次给大家带来Ajax的页面无刷新实现详解(附代码),Ajax页面无刷新实现的注意事项有哪些,下面就是实战案例,一起来看一下. ajax (ajax开发) AJAX即"Asynchronou ...
DPDK无锁队列rte_ring相关代码及示例程序（rte_ring.h，rte_ring.c，main.c，makefile）
目录 rte_ring.h rte_ring.c main.c makefile 推荐阅读: [共享内存]基于共享内存的无锁消息队列设计:https://rtoax.blog.csdn.net/art ...
mysql 事务 for update,mysql事务锁_详解mysql 锁表 for update
摘要腾兴网为您分享:详解mysql 锁表 for update,智慧农业,真还赚,悦读小说,学习帮等软件知识,以及电池管家,三国群英传3,userland,运满满货主版,王者荣耀,简单3d动画,嘉丽 ...
深入并发线程、进程、纤程、协程、管程与死锁、活锁、锁饥饿详解
一.进程.线程.纤程.协程.管程概念理解在现在你可能会经常看到进程.线程.纤程.协程.管程.微线程.绿色线程....一大堆xx程的概念,其实这些本质上都是为了满足并行执行.异步执行而出现的一些概念. ...
java分类锁_【基本功】java锁分类详解
[基础课]--[锁]--锁分类mp.weixin.qq.com 前言 Java提供了种类丰富的锁,每种锁因其特性的不同,在适当的场景下能够展现出非常高的效率.本文旨在对锁相关源码(本文中的源码来自J ...

DPDK 无锁ring，详解

rte_ring_tailq保存rte_ring链表

创建ring

入队操作

出队操作

ring是否满或者是否为空

DPDK 无锁ring，详解相关推荐

最新文章

热门文章

DPDK 无锁ring， 详解

rte_ring_tailq保存rte_ring链表

创建ring

入队操作

出队操作

ring是否满或者是否为空

DPDK 无锁ring， 详解相关推荐

最新文章

热门文章

DPDK 无锁ring，详解

DPDK 无锁ring，详解相关推荐