摘要：Ringbuffer（循环缓存）是软件中非常常用的数据结构之一， 在互联网应用、数据库应用等中使用广泛。处理器执行Ringbuffer的效率与其存储系统处理共享数据的性能息息相关。
Ringbuffer（循环缓存）是软件中非常常用的数据结构之一， 在互联网应用、数据库应用等中使用广泛。处理器执行Ringbuffer的效率与其存储系统处理共享数据的性能息息相关。
作为本系列文章的第二篇，本文将介绍无锁Ringbuffer的实现，并且在Arm服务器平台上验证这些优化的实际效果。
声明
本系列文章介绍的针对Ringbuffer实现的优化方法来自于Ola Liljedahl的研究。Ola现任ARM公司的架构师，长期专注于网络软件的性能优化设计，尤其是可扩展的多线程共享存储编程。针对ARM A系列处理的多线程性能，他开发了一个可扩展的多线程并行库progress64。
progress64本系列文章分析了Ola提出的主要优化手段，并且在ARM服务器上重现了Ola的优化以验证其实际效果。博客中使用的代码只是示例代码（GitHub - wangeddie67/ringbuffer_opt_demo），其功能完备性和鲁棒性都不能满足实际应用的需求。
无锁Ringbuffer实现对于多线程共享程序，需要使用mutex来保护对于共享变量的load-modify-store代码序列，从而保证功能正确性。但是mutex的竞争引入了非常明显的性能开销。因此，需要探索无锁的Ringbuffer实现。
用原子操作替代锁
除了mutex，另一种可以保护共享变量的方法是原子操作。单个原子操作就是由load-modify-store三个部分构成。原子操作保证，在完成load-modify-store序列之前，被访问的地址不会被其他指令访问。典型的原子操作包括Compare-and-swap （CAS），以及ARM提供的LDADD等原子操作。如果有多个PE针对同一个共享变量发起多个原子操作，那么这些操作只能按照某种顺序串行，而不能并行。执行顺序取决于硬件实现。
一篇完整中介绍lock/unlock函数实现时已经介绍了CAS(ptr, oldval, newval)的语义。CAS保证了对于mutex的操作都是串行的。例如PE1执行原子操作CAS(x, 0, 1)，同时PE2执行原子操作CAS(x, 0, 2)：
如果PE1先读取到x的值，那么PE1的CAS操作成功；PE2读取到的x的值是PE1更新后的值=1，CAS操作失败。最后PE1中返回值是0；PE2中返回值是1；x的值是1。如果PE2先读取到x的值，那么PE2的CAS操作成功；PE1读取到的x的值是PE2更新后的值=2，CAS操作失败。最后PE1的返回值是2；PE2的返回值是0；x的值是2。在Ringbuffer实现中，特别需要保护的就是被所有生产者和消费者共享的head和Tail指针。可以用__atomic_fetch_add原语的声明如下，其核心指令是LDADD。原语读取ptr指向的变量，然后将变量累加val后的新值会写到原来的位置；同时，原语将ptr指令的变量的旧值返回。type __atomic_fetch_add (type *ptr, type val, int memorder)将原语应用到head或tail指针上，则可以将存储在内存中的指针变量加1，并且同时将加1前的旧值返回。这就相当于将目前head/tail指针指向的单元分配给了当前的Enqueue或Dequeue的生产者或消费者，同时与其他生产者或消费者同步了更新后的指针。Enqueue和dequeue
图7展示了Enqueue和Dequeue的流程图。
相对于基于mutex的Ringbuffer实现，无锁Ringbuffer实现从软件和硬件两方面的优化。
从软件的角度，无锁Ringbuffer的关键区要小很多。在基于mutex的Ringbuffer实现（图8左侧）中，锁保护了整个Enqueue和Dequeue过程，每个Enqueue和Dequeue过程都只能串行执行。在无锁Ringbuffer实现（图8右侧）中，原子指令只保护了对于head或tail指针的累加。除了必须串行，其他部分都是可以并行的，从而增加了程序的并行度。从硬件的角度，原子指令引入更少的snoop操作。使用常规的load指令读取数据后，在PE侧的缓存中的数据副本的状态为Share（共享）。所有访问这个数据的PE的缓存都会具有这个共享数据的副本。当共享数据被某个PE修改后，需要大量的snoop去失效其他PE中的所有副本。
使用原子指令读取数据之后，数据的缓存状态为Exclusive（独有），即只有一个PE具有这个共享数据的副本。当另一个PE也执行原子操作后，只需要一个snoop就可以将已经存在副本清除。
从带宽的角度，更少snoop数量意味着snoop通信占用的通信带宽更少，提供给数据通信的带宽就会更高。从延迟的角度，load/store操作需要等到其引起的所有snoop都完成才能被标识为完成。更少的snoop数量意味着等待snoop完成的时间更短。
无锁Ringbuffer同时从这两方面优化中受惠，暂时没有办法定量分析出哪一个优化的作用更大。一般来说，缩小关键区的作用更大。
逻辑单元序号
上述的无锁Ringbuffer仍然可能导致功能错误。例如一个Ringbuffer有16个物理单元。首先为生产者A分配的逻辑单元是16；随后，为生产者B分配的单元是32；这两个单元映射到相同的物理单元0。虽然单元分配和指针更新是串行的，但是对于数据单元的访问是并行的，因此并不能保证生产者B访问物理单元0时，生产者A一定完成了操作。可能生产者A和生产者B同时查询到单元是空闲，进而同时对这个物理单元进行写入。此时无法保证写入单元的究竟是生产者A的数据还是生产者B的数据。
为了解决这样的问题，Ola提出，给每个单元增加一个域来表示其对应的逻辑单元序号。只有当逻辑单元序号与head/tail指针一致的时候才能进行访问。当Dequeue单元的时候，需要更新逻辑单元序号，将单元的序列号增加物理单元的总数。
还是上面的例子，当生产者A和B同时查询到内存是空闲时，由于单元的序列号与生产者A分配的单元匹配，那么生产者A可以操作这个单元；生产者B则因为序列号不匹配而不能访问，需要继续等待。
增加逻辑单元序号后的Enqueue和Dequeue流程图如图9。
无锁Ringbuffer声明
无锁Ringbuffer的声明如下。Ringbuffer不需要mutex进行保护。
// Entry in ring bufferclass BufferEntry{public: unsigned int m_sn; // Sequential number.void *mp_ptr; // Pointer to data entry.};class RingBuffer{public: unsigned int m_size; // The number of physical entries.unsigned int m_head; // Head pointer.unsigned int m_tail; // tail pointer.BufferEntry *mp_entries; // Pointer to physical entries.}代码3 无锁Ringbuffer声明
无锁Ringbuffer的Enqueue和Dequeue函数
Enqueue和Dequeue函数的实现如下。
// Blocking Enqueue Functionint enqueue_ringbuf(RingBuffer *ring_buffer, void *entry){ // Update the tail pointer.int sn = __atomic_fetch_add(&ring_buffer->m_tail, 1, __ATOMIC_RELAXED); int enq_ptr = sn % ring_buffer->m_size; // Check the data entry until empty.unsigned int entry_sn; void *entry_ptr; do { // Replace load by atomic load.// entry_sn = __atomic_load_n(&ring_buffer->mp_entries[enq_ptr].m_sn, __ATOMIC_RELAXED);// entry_ptr = __atomic_load_n(&ring_buffer->mp_entries[enq_ptr].mp_ptr, __ATOMIC_RELAXED);entry_sn = ring_buffer->mp_entries[enq_ptr].m_sn; entry_ptr = ring_buffer->mp_entries[enq_ptr].mp_ptr; } while (entry_sn != sn || entry_ptr != NULL); // Write entry.// Replace store by atomic store.// __atomic_store_n(&ring_buffer->mp_entries[enq_ptr].mp_ptr, entry, __ATOMIC_RELEASE);ring_buffer->mp_entries[enq_ptr].mp_ptr = entry; return 0;} // Blocking Dequeue Functionint dequeue_ringbuf(RingBuffer *ring_buffer, void **entry){ // Update the head pointer.int sn = __atomic_fetch_add(&ring_buffer->m_head, 1, __ATOMIC_RELAXED); int deq_ptr = sn % ring_buffer->m_size; // Check the data entry until not empty.unsigned int entry_sn; void *entry_ptr; do { // entry_sn = __atomic_load_n(&ring_buffer->mp_entries[deq_ptr].m_sn, __ATOMIC_RELAXED);// entry_ptr = __atomic_load_n(&ring_buffer->mp_entries[deq_ptr].mp_ptr, __ATOMIC_ACQUIRE);entry_sn = ring_buffer->mp_entries[deq_ptr].m_sn; entry_ptr = ring_buffer->mp_entries[deq_ptr].mp_ptr; } while (entry_ptr == NULL || entry_sn != sn); // Read entry.*entry = entry_ptr; // Replace store by atomic store// __atomic_store_n(&ring_buffer->mp_entries[deq_ptr].mp_ptr, 0, __ATOMIC_RELAXED);ring_buffer->mp_entries[deq_ptr].mp_ptr = 0; // Update sequence number.// __atomic_store_n(&ring_buffer->mp_entries[deq_ptr].m_sn, sn + ring_buffer->m_size, __ATOMIC_RELEASE);ring_buffer->mp_entries[deq_ptr].m_sn = sn + ring_buffer->m_size; return 0;}代码4 无锁Ringbuffer的Enqueue和Dequeue函数
Enqueue和Dequeue函数中对于数据单元的load/store操作可以进一步被原子操作__atomic_load_n和__atomic_store_n取代。代码4中的注释给出了可以替换的原子操作。无锁ringbuffer实现的完整源码请参见
ringbuffer_opt_demo/srcs/lockfree_blkring.h at main · wangeddie67/ringbuffer_opt_demo · GitHub使用原子操作数据单元的无锁Ringbuffer实现的完整源码请参见
ringbuffer_opt_demo/srcs/atomic_blkring.h at main · wangeddie67/ringbuffer_opt_demo · GitHub无锁Ringbuffer性能测试结果
无锁Ringbuffer的性能如图10所示。曲线mutex表示基于mutex的Ringbuffer实现；lockfree和atomic表示无锁Ringbuffer实现（lockfree使用普通load/store操作数据单元；atomic使用原子load/store操作数据单元）。
从图10中可以发现，相对于基于mutex的Ringbuffer实现，无锁Ringbuffer实现的性能得到了很大的提升。用atomic曲线和mutex曲线进行比较。当只有2个线程时，无锁Ringbuffer实现的吞吐率是基于mutex的Ringbuffer实现的吞吐率的1.86倍；当使用32个线程时，无锁Ringbuffer实现的吞吐率是基于mutex的Ringbuffer实现的吞吐率的18倍。
从图10中可以发现，对于数据单元的访问是否使用原子操作对于性能的影响并不明显。两条曲线基本上是重合的。这是因为测试的ringbuffer提供了足够的单元，多个线程共享Ringbuffer单元的竞争并不强烈。
无锁Ringbuffer实现的吞吐率还是会随着子线程数增加而降低，但是并不如基于mutex的Ringbuffer强烈。当使用2个子线程时，吞吐率为7394830 op/s；当使用全部32个子线程时，吞吐率为5800350 op/s。前者是后者的1.27倍。
利用Perf分析程序热点只能定位到特点在enqueue和dequeue函数，无法提供进一步的信息。因此使用Arm的top-down工具对于微架构执行情况进行详细分析，得到的结果如图11。
图11中用红圈标注了主要注意的指标。首先，程序明确属于后端瓶颈的程序，后端阻塞周期高达97%。其次，L1、L2和LC（Last-level cache）明显高于其他异常情况，这表示程序中存在大量的数据竞争。最后L1、L2和LC的数据缺失率属于同一量级，这表示数据竞争是全局性的，需要通过系统的最后一级缓存才能得到解决。top-down的分析结果提示，无锁的ringbuffer实现仍然导致了大量的全局性的数据冲突。
无锁Ringbuffer的对齐改进通过对于程序的分析，符合上述特征的共享数据只有head和tail指针。因为head和tail指针处于相同的缓存行。这就导致生产者和消费者仍然在竞争同一个缓存行。可以通过对与Ringbuffer的数据结构进行缓存行对齐，减少不同生产者和消费者竞争同一个缓存行的情况。
Ringbuffer实现的缓存行对齐包含两个部分：
head/tail指针的对齐。将head指针和tail指针分别放置到不同的缓存行。这可以通过C语言原语alignas实现。
class RingBuffer{public: alignas(64) unsigned int m_size; alignas(64) unsigned int m_head; alignas(64) unsigned int m_tail; alignas(64) BufferEntry *mp_entries;}代码5 对齐的无锁Ringbuffer的声明
这样只有生产者竞争tail指针所在的缓存行；消费者竞争head指针所在的缓存行。
数据单元的对齐。这是为了避免访问不同单元的线程因为单元处于相同缓存行而产生竞争。比如访问物理单元0-3的线程实际上在竞争同一个缓存行。
这一部分可以通过空间换效率的方式实现。一个缓存行（64B）中可以放置4个单元（每个单元16B，8B指针和8B序列号）。在初始化时分配4倍的单元，在Enqueue和Dequeue每次都只使用4对齐的序号。逻辑单元0对应到物理单元0；逻辑单元1对应到物理单元4；依次类推。
物理单元序号计算方法如下：
int enq_ptr = (sn % ring_buffer->m_size) * 4;代码6 对齐的无锁Ringbuffer映射逻辑单元到物理单元
对齐后的无锁ringbuffer的内存分布如下。其中红色部分是实际访问的数据单元。绿色部分是没有使用到的数据单元。
需要说明的是，当物理数据单元数量远大于生产者和消费者数量时，数据单元的对齐对于性能的影响非常微小。因为共享一个缓存行的线程数量很小，最多只有4个。
缓存行对齐的无锁ringbuffer实现的完整源码请参见
ringbuffer_opt_demo/srcs/align_blkring.h at main · wangeddie67/ringbuffer_opt_demo · GitHub对齐的无锁Ringbuffer性能测试结果
缓存行对齐的无锁Ringbuffer的性能如图13所示。曲线mutex表示基于mutex的Ringbuffer实现；lockfree和atomic表示无锁Ringbuffer实现（lockfree使用普通load/store操作数据单元；atomic使用原子load/store操作数据单元）；align表示缓存行对齐后的无锁Ringbuffer实现。
缓存行对齐的无锁Ringbuffer实现的性能曲线随着子线程数量的增加呈现增长的趋势。具体来说，
首先，当子线程数小于16时，吞吐率随着子线程数增加而趋向于一个稳定的值。这个稳定值受限于是互联网络能够提供的最大带宽。其次，当子线程数大于16时，吞吐率随着子线程数增加而增加。这与互联结构的拓扑有关系。当子线程数大于32时，新的子线程被分配到互联结构的另一半部分，从而使得能够利用的带宽增加。用align曲线和mutex曲线进行比较。当只有2个线程时，无锁Ringbuffer实现的吞吐率是基于mutex的Ringbuffer实现的吞吐率的3.31倍；当使用16个线程时，无锁Ringbuffer实现的吞吐率是基于mutex的Ringbuffer实现的吞吐率的33.8倍。当使用32个线程时，无锁Ringbuffer实现的吞吐率是基于mutex的Ringbuffer实现的吞吐率的110倍。
缓存行对齐的无锁Ringbuffer实现的性能曲线与基于mutex的Ringbuffer实现和无锁Ringbuffer的实现有本质上的不同，呈现了一种互联带宽测试的规律。这表示缓存行对齐的无锁Ringbuffer实现能够充分利用互联网络提供的通信性能，是一个非常有利于系统规模扩展的Ringbuffer实现。
结论本文介绍了无锁Ringbuffer的实现方法，并且对基于mutex的Ringbuffer实现和无锁Ringbuffer实现进行了性能对比和分析。无锁Ringbuffer以及缓存行对齐这两个优化都可以显著提高Ringbuffer的吞吐率。缓存行对齐的无锁Ringbuffer能够充分利用系统中互联系统提供的通信容量，是一个非常有利于系统规模扩展的Ringbuffer实现。
来源：小康科技园地
