int chat *int sizeof Ringbuffer性能优化方法

int chat *int sizeof Ringbuffer性能优化方法

对于Ola所提出的主要优化手段，本文已然展开了分析，并且于Arm服务器之上，将Ola之优化予以重现，以此来验证其实际所呈现的效果。博客之中所运用的代码，仅仅是示例代码罢了，其功能完备程度以及鲁棒性，均无法满足实际应用的需求。

介绍

一般是由一个存有数据的数组，以及分别指向队列头也就是 Head 和队尾也就是 Tail 的指针共同构成的，就如同图 1 左边所呈现的那样，指针仅仅能够依照唯一一个方向进行移动，在指针移动至数组最末端的时候，指针会返回到队列起始的位置 。

对于Head指针而言，它所表示的是接踵而至的、能够予以弹出动作的数据；而当数据从队列之中加以弹出（这里对此特定行为称作某一活动）之际，Head指针会进行加值操作，将自身数值增加1；至于Tail指针，它所能表示的是随后到来的、可以用来实施压入数据行为的单元；当数据被压入队列（此行为被叫做另一活动）之时，Tail指针会执行加值动作，把自身数值提升1；。

根据逻辑而言，其中的数组构成了一个环，此环如同一幅图里图1右边所展示的那般，在这个环上，用于标识起始的头指针以及指向末尾的尾指针，二者朝着相同的方向进行移动，这便是其名称的由来所在了。

图 1 示意图

将程序压入单元的被称作生产者（），把程序从弹出单元取出的堪称消费者（），情况如图 2 所示。一个 一般存在诸多生产者与消费者，每一个生产者以及消费者皆是各异的线程或者进程。这些线程或者进程会被安排到系统里不同的处理器核心之上，而 乃是不同处理器核心之间共同享有的数据。

图 2 生产者和消费者

基于 Mutex 的 实现

为确保共享数据的功能无误 ，一般会运用 mutex 加以保护 ，借此保证仅有一个生产者或者消费者能够实施访问或者操作 。

基于 Mutex 的 声明

的数据结构如代码块 1 所示。

// Entry in ring buffer
class BufferEntry
{
public:
空位指针类型的变量，名为mp_ptr，它是用于，指向数据项的指针 。
};
class RingBuffer
{
public:
定义一个整型变量m_mutex，它用于当作共享指针相互间的互斥对象，以此来保证共享指针在互斥的情况下互斥地进行操作，从而使其互斥地进行数据访问 。
无符号整型，m_size，其表示物理条目的数量 。  --->  无符号的、用于表示物理条目的数量的那一类型，被称作m_size 。  --->  被定义为无符号整型的m_size，是用来表示物理条目的数量的 。
这是无符号整数类型，名为m_tail，它是尾指针，用于表示特定位置或顺序中的最后一项，或用于表示某种数据结构中最后单元的变量或数据。具体来说，它指向构成某个。
指向物理条目的指针，是名为mp_entries的BufferEntry类型指针。，。
};

代码 1 基于 Mutex 的 声明

在数据数组里的每个单元呢，仅仅是提供一个指针，这个指针所指向的是实际上需要进入的那个数据结构。当处于某种情况以及另外一种情况的时候呀，只需求复制那个数据结构的指针就行啦，而并非要去复制数据，这样一来呢便能够减少数据拷贝的花费了。

基于 Mutex 的 的 和 函数

图3呈现出了，和的一种流程示意图案，图3的左边部分所示的是那种阻塞操作的流程示意图案，右边部分所呈示的则是非阻塞操作的流程示意图案，要是达成的是阻塞样式的操作情况，那么函数会持续不断重复去检查空满条件状况，一直持续到达到或满足操作所需要的空满条件状况，要是达成的是非阻塞样式的操作情况，那么函数能够在检查空满条件失败之后径直退出。

图 3 基于 Mutex 的 实现流程图

在本系列文章示例里，采用逻辑单元与物理单元的控制方式，这是循环队列空满控制的诸多方法里的一种，head和tail的范围并非局限于物理单元数量，而是能够持续累加，访问单元时，借由(head MOD size)或者(tail MOD size)获取对应的物理单元。

对于该操作而言，需要达成一种情况，那就是不为满的状态，也就是加起来的结果不可等于，这意味着head和tail之间的差值不能大于物理单元所具有的数量。

对于该操作而言，其要求被进行操作的对象不能处于为空的状态，也就是说，该对象必须满足不等于这样一种情况，即该对象不能是这样的状态，在这种状态下其头部与尾部指向相同的单元 。

基于 Mutex 的 实现如下：

// Blocking Enqueue Function
把一个指向环形缓冲区的指针，给到名为 “进行入队操作的函数”，再把一个指向要插入元素的指针，也给到这个函数，函数返回一个整型值，表示某个操作的结果，这个操作用于在环形缓冲区中插入元素，这个。
{
    // Lock mutex.
    lock(&ring_buffer->m_mutex);
存在环形缓冲区已满却无事可做的情况句号，在这种已满的状态下不采取任何行动句号。
存在这样一种情况，处于循环状态下，对于环形缓冲区而言，当环形缓冲区的头部位置数值加上环形缓冲区的大小数值等于环形缓冲区的尾部位置数值之时 。  }。
    // Enqueue entry.
定义一个整型变量enq_ptr，其值为，环形缓冲区的尾部索引，对，环形缓冲区的大小，进行取模运算后的结果 。
将环形缓冲区的入口指针数组中，处于入队指针位置的元素的指针，指向入口，。
    // Increase tail pointer.
环形缓冲区里面的“m_tail”，等于它自己加上一 ，再赋值给那个“m_tail” 。
    // Unlock mutex.
对那个环形缓冲区所关联的互斥量进行解锁操作，此互斥量由ring_buffer里的m_mutex表示 。
    return 0;
}
// Blocking Dequeue Function
搞一个函数叫做int ，那个函数是dequeue_ringbuf哦，它的参数有RingBuffer *这种类型的ring_buffer ，还有void **这种类型的entry ，这个函数返回一个int数值 。
{
    // Lock mutex.
    lock(&ring_buffer->m_mutex);
当环形缓冲区的写指针等于环形缓冲区的读指针时，处于某种状态，这种状态下会执行特定操作，操作会持续进行，直到该式不成立，操作才会重新判断，重新判断以新一轮的这种状态为依据，然后该。
    // Dequeue entry.
在程序里，定义了一个 int 类型的变量，名为 deq_ptr。它的值是通过一种运算得到的，这种运算就是，取 ring_buffer 对象中的 m_head 属性值，用它除以 ring_buffer 对象里面的 m_size 属性值后所得到的余数 。
将“*entry”赋值为，“ring_buffer”的“mp_entries”数组中，索引为“deq_ptr”的元素的“mp_ptr”指针所指向的值，。
环形缓冲区指向的元素数组中，对应循环双端队列指针位置的元素所指向的指针，使其为空 。
    // Increase head pointer.
ring_buffer的m_head，等于，ring_buffer的m_head，加上1 ，句号。
    // Unlock mutex.
    unlock(&ring_buffer->m_mutex);
    return 0;
}

代码 2 基于 Mutex 的 实现

Lock 和

1. 对于进行操作，最多只允许有一个生产者或者一个消费者这一情况。 2. 而这一情况需要借助锁标志来予以控制。 3. 由于考虑到其中的mutex API所引入的系统调度开销实在是过大。 4. 所以在这里则采用原语来实现lock以及操作。

类型 __sync_val_compare_and_swap，类型指针为指针，类型旧值为旧值，类型新值为新值；。

同步功能为原语所提供，原子CAS操作这个功能也是原语所具备的，其功能核心是CASAL指令。原语会拿去ptr指向的那个变量来跟某值进行比较，倘若值是相同的，那便会把ptr指向的变量改作某个值，并且返回ptr指向变量的旧值；反过来，要是值不相同，那就不会去修改ptr指向的变量，而是返回ptr指向变量的值。与此同时，原语对load/store的同步起到了保证作用，也就是说，在指令序方面，先于原语的load/store已然完成，在指令序上后于原语的load/store不会去执行。

用原语实现的 lock 和 功能如下：

// Lock function
void lock(int *mutex)
{
当，通过__sync_val_compare_and_swap操作互斥锁，使其把0与1进行比较和交换，并最终结果不等于0时，。
    {
    }
}
// Unlock function
void unlock(int *mutex)
{
执行 __sync_val_compare_and_swap 操作于 mutex，将 1 与之比较，如相等于则替换为 0，。
}

代码 3 lock 和 函数的实现

如果mutex变量的值是0，这就表示该mutex处于释放状态；而要是变量的值为1，那就意味着mutex处于锁定状态。在lock函数之中，要是当前的mutex的值是0也就是处于释放状态的话，那么就会将其更新为1也就是锁定状态，与此同时lock函数会返回0；相反的情况要是mutex的值不为0，那么lock函数就会返回1。基于mutex所实现的完整源码请参见()。

测试框架

一个用于测试 的程序存在着这样的情况，里面哟一个起着把控测试流程进度作用的主线程，还有好些充任生产者以及消费者角色的子线程。主线程先是去进行创建以及初始化操作，之后又去创建规定数量的子线程，最后处于等待子线程结束的状态。子线程的数量是能够展开配置操作的。当中的每个子线程既是生产者同时又是消费者。每一次测试迭代情况是从队列里取出来一个单元，然后把这个单元（不进行额外的操作处理）放置到队尾位置，也就是一次放进操作和一次取出这样子的操作行径。整个测试流程所需要展开执行的放进操作和取出操作的数量是一样的。假如把一部分线程设定成生产者，而把另一部分线程设定成消费者，那么数据单元必定是从生产者朝着消费者移动的。这有可能在互联结构里出现带有方向性的数据流，进而强化测试结果跟线程调度的核心的物理位置之间的关系。为了确保子线程同步启动测试，子线程首先会陷入。当所有子程序都创建完毕后，子线程从退出。每个线程执行的测试迭代数量都是一样的。

图4，测试程序流程图，每个子线程，独立统计，测试阶段完成的时间，然后分别计算，操作的吞吐率 。这是为了预防统计的过程，引入新的共享数据。最后，主线程获取，每个子线程统计的操作吞吐率，继而求和，得到总的吞吐率。吞吐率，用每秒执行的，或操作数量来表示，也就是op/s 。由于性能与存储系统的性能，存在直接关联，所以子线程调度到的核心，对于具体实现，有着极大的关系。要确保测试具备稳定性，就得把子线程调度至指定的核心（进行绑核操作）。要是不开展绑核，那同种实现、相同配置的不同测试结果，极有可能差异巨大，测试结果偏差程度极大。测试框架的源码可参见)。

基于 Mutex 的 实现的测试结果

采用用于进行测试的 Arm 服务器，其具备 32 个 N1 处理器，将此设为互联的结果。在被用于测试的 Arm 服务器上，并未对 NUMA 结点作出区分，所以采用自然映射的关系，也就是子线程 0 被调度到 CPU0；子线程 1 被调度到 CPU1，依此类推。图 5 呈现出基于 Mutex 的性能测试所产生的结果。很明显，伴随子线程数量的增多，吞吐率显著下降。 )。运用 2 个子线程之际，吞吐率是 op/s；启用全部 32 个子线程的时候，吞吐率为 op/s，前段所述的情况是后面那种情形的 12.3 倍 。

图 5 基于 Mutex 的 性能测试

图5所展示出来的性能测试相关结果，是那种具备典型特征的、以多线程竞争作为主导因素的性能曲线来着。随着参与竞争的子线程数量不断增加，在对于mutex 进行 竞争这个行为上所花费的时间也就会变得更多起来，进而导致性能出现快速下降的情况。运用perf工具针对启动16个线程的测试程序展开热点分析，热点分析所得到的结果就如同图6那样呈现着。图当中橙色部分给出了提示，lock函数占用了程序执行时间的91%。所以呢，对于进行优化这件事情来讲，首要的步骤便是采用无锁的这种实现方式。

图 6 Perf 热点分析结果

无锁 实现

针对于多线程共享的程序而言，得运用 mutex 去守护对于共享变量的 load--store 代码序列，以此来保障功能的正确性。可是呢，mutex 的竞争带来了极为显著的性能开销。所以呀，要着手探索无锁的实现方式 。

用原子操作替代锁

另一种可保障共享变量的办法，并非 mutex，而是原子操作呀。单个原子操作呢，是由 load，store，再加上另外一个构成部分形成的。原子操作确保，于完成 load--store 序列以前呀，被访问的那个地址，不会遭受其他指令去访问。有代表性的原子操作包含 -and-swap （也就是 CAS），以及 Arm 所提供的 LDADD 这类原子操作呢。倘若存在多个针对同一共享变量发起多个原子操作的PE，那么这些操作仅能依照某种顺序串行进行，而非并行。其执行顺序取决于硬件实现。CAS确保了针对mutex的操作全都是串行的。比如，有PE1执行原子操作CAS(x, 0, 1)，与此同时，又有PE2去执行原子操作CAS(x, 0, 2) 。

假设，是PE1率先读到了x的值，如此一来，PE1的CAS操作便成功了；而PE2所读取到的x的值，乃是PE1更新之后的值，这个值等于1，所以PE2的CAS操作失败了。最终，PE1里返回的值为0；PE2里返回的值为1；x的值是1 。

若PE2率先读取到x的值，那么PE2的CAS操作达成成功状态；PE1所读取到的x的值为PE2更新以后的值等于2，于是CAS操作失败。最后PE1的返回数值是2；PE2的返回数值是0；x的值是2。

在具体的实现过程当中，尤其需要加以保护的便是，那些处于被所有生产者以及消费者共同分享状态的head和tail指针。能够借助原语去达成指针的原子指令，并且实现指针的累加操作。原语的声明情况是这样的，它的核心指令是LDADD。原语会读取ptr所指向的变量，接着，把变量累加val之后所形成的新值，会被写到原本的位置上；与此同时，原语会将ptr指令的变量的旧值返回出来。

以类型指针指向的位置为操作对象，将该位置存储的值加上给定的值，操作的内存顺序受到指定的内存顺序参数影响，返回该位置原来存储的值，有一种叫做“type __atomic_fetch_add”的操作，它的操作对象是类型指针所指向。

把原语运用到head指针上时，能让内存里存储的指针变量加1，还会把加1之前的旧值 ，这就如同把当前head指针所指向的单元分配给了此刻的生产者或者消费者一样，与此同时跟其他生产者或者消费者同步了更新后的指针；将原语应用到tail指针上时，能让内存中存储的指针变量加1，还会把加1之前的旧值返回，这就如同把当前tail指针所指向的单元分配给了此刻的生产者或者消费者一般，与此同时跟其他生产者或者消费者同步了更新后的指针。

和

图 7 展示了 和 的流程图。

图7所呈现的无锁实现流程图（此流程图是通过使用原子操作来替换mutex达成的），同基于mutex的实现相比较而言，针对这种无锁实现做出了从软件以及硬件这两个方面的优化。就软件层面来说，无锁实现的关键区显著小了许多。在基于mutex的实现当中（也就是图8左侧所展示的那种情况），锁对整个相关过程起到了保护作用，每一个相关过程均只能以串行方式来执行。而在无锁实现里面（即图8右侧呈现的情况），原子指令仅仅对涉及head或者tail指针的累加操作起到了保护作用。除了那非得串行的情况之外，别的部分通通都是能够并行的，借此增加程序的并行程度，。

图8，针对基于mutex的实现与无锁实现的关键区（也就是着色部分）进行对比，从硬件角度来看，原子指令所引入的snoop操作更少，在使用常规的load指令读取数据之后，于PE侧的缓存里的数据副本状态呈现为Share（共享），所有对这个数据进行访问的PE的缓存都会拥有这个共享数据的副本，当共享数据被某个PE修改后，就需要大量的snoop去使其他PE中的所有副本失效。于使用原子指令读取好数据之后，数据的缓存状态呈现为（独有），也就是说仅仅只有一个PE持有着这个共享数据的副本。在另一个PE同样执行原子操作之后，单单只需一个snoop便能够将已然存在的副本给清除掉。从带宽所处的角度来讲，更少的snoop数量所寓意的是snoop通信占用的通信带宽更少，如此一来提供给数据通信的带宽便会更高。在从延迟的角度来看时，load/store操作得要等到其引发的所有snoop都完成了方可被标识为完成。意味着花费在等待，那种被称作snoop完成的时间，会因更低数量的snoop而缩短，无锁，同时从这两个方面存在的优化里获取益处，暂时没办法凭借定量分析，得出哪一个优化发挥的作用更为显著，通常来讲，缩小关键区所起到的作用更大。

逻辑单元序号

上述那种无处于加锁状态，依然存在致使功能出现差错的可能性。比如说存在这样一种情况，有十五六介 作为实物存在的不同单元。最开场 为那称作生产者 A 的赋予的具备逻辑性质的单元是十六一；紧接着，给被叫做生产者 B 的分配的单元是三十二一；这两组单元都影射到相同的作为实物存在的单元零。尽管单元开展赋值动作以及指针进行更新这两个行为是依照顺序依次进行的，然而针对数据单元施行的访问却是以并行方式进行的，所以无法对生产者 B 访问作为实物存在的单元零这个行为，确保生产者 A 必然已完成相应操作这一点作出保证。 返回。也许生产者 A 会查询到单元呈现空闲状态，生产者 B 同样会查询到单元处于空闲状况，紧接着两者会同步对该物理单元施行写入操作。这会儿难以确保写入单元而落入的究竟是不是生产者 A 的数据，又或者是不是生产者 B 的数据呢。为了去处理此类状况而引发的问题，Ola 给出主张，要给每一个单元添设一个域，以此去表明其相应的逻辑单元序号呢。唯独当逻辑单元序号跟 head/tail 指针吻合之际，方可展开访问之类动作。当涉及到单元情形时，得去更新逻辑单元序号，把单元的序列号增添物理单元的总数目呢。仍是上面所提的例子，当生产者A查询到内存处于空闲状态时，当生产者B也查询到内存处于空闲状态时，因单元序列号跟生产者A所分配的单元相匹配，所以生产者A能够对这个单元进行操作，因序列号不匹配，生产者B无法进行访问，生产者B须持续等待，增加逻辑单元序号后市，增加逻辑单元序号后和流程图如图9 。

图 9 无锁 实现流程图（使用原子操作替换 mutex）

无锁 声明

无锁 的声明如下。 不需要 mutex 进行保护。

// Entry in ring buffer
class BufferEntry
{
public:
无符号整型，m_sn，即：序列号 ， 需要注意的是，你提供的原始内容并不是一个完整的可执行语句，只是一个变量声明。如果要在实际代码中使用，。
空指针类型的变量 ，名为mp_ptr ，用于存放数据项的指针 ， 。
};
class RingBuffer
{
public:
    unsigned int m_size;        // The number of physical entries.
 
无符号整型，有着名为m_head的，是头部指针，这样一种情况。
未标明符号的整数类型，名为m_tail，那是尾指针，有此定义 。
 
mp_entries是指向物理条目的指针，这个指针是BufferEntry类型的，它被命名为mp_entries 。
}

代码 3 无锁 声明

无锁 的 和 函数

和 函数的实现如下。

// Blocking Enqueue Function
int enqueue_ringbuf(RingBuffer *ring_buffer, void *entry)
{
    // Update the tail pointer.
将变量 `sn` 赋值为，通过以 `__ATOMIC_RELAXED` 内存顺序，对 `ring_buffer` 的成员 `m_tail` 进行原子_fetch_add操作加上1的结果，其中 `ring_buffer` 为所给变量，之后的逗号及1为操作数及参数，最终通过该操作获得的值。
这里有代码片段，它有一个变量，这个变量被命名为enq_ptr ，这个变量被赋值了，其值是sn对ring_buffer的m_size的取余运算结果，这里。
 
反复核查数据录入情况，一直核对到其变成空的状态止，。
    unsigned int entry_sn;
    void *entry_ptr;
    do
    {
        // Replace load by atomic load.
// 那进入的序号等于以原子方式加载非对准的并带有宽松内存序的环形缓冲区中对应入队指针位置的条目的序号，这里的对应条目是指环形缓冲区里的条目，而这个条目的序号会被加载。
// 首先，针对，获取，环形缓冲区，其，mp_entries，数组，在，enq_ptr，位置，所，对应的，mp_ptr，指针，进行，读取，操作，采用，__atomic_load_n，函数，遵循，__ATOMIC_RELAXED，内存序，将，读取，结果，赋值给，entry_ptr。
enq_ptr所指的ring_buffer对象的mp_entries数组元素中的m_sn，被赋值给entry_sn。
取进入指针，指向环形缓冲区的对应的入口元件位置，从该位置选取记录指针，最后返回取用元件所指的包含有指向入口指针的指针，其中入口指针初始值为环形缓冲区入口指针数组中由入队指针所指的那个元素处的指针。
} 当入口序列号不等于序列号，或者入口指针不为空的时候，持续循环 。
 
    // Write entry.
// 用原子存储替换存储装置，通过原子存储来进行替换，将存储装置替换成原子存储，用原子存储把存储装置加以替换 。
取出环形缓冲区中位于入列指针处的元素中的指针并替换为指定的某个元素，采用原子存储且具备释放语义，具体操作为对从环形缓冲区的元素数组中依据当前缓冲区处理所用且在入列、出列循环中不断变化的入列指针所指向的元素里进一步获取的指针进行原子设置 ，设置为何种元素呢 ，将其设置为给定的那个元素 ，。
    ring_buffer->mp_entries[enq_ptr].mp_ptr = entry;
 
    return 0;
}
 
// Blocking Dequeue Function
int dequeue_ringbuf(RingBuffer *ring_buffer, void **entry)
{
    // Update the head pointer.
将变量 sn 赋值为，通过以宽松内存顺序进行原子操作，对指向环形缓冲区的 m_head 指针所指向的值，增加 1 之后的结果。这样就取回了增加 1 之后的值，并存入 sn 中。
整数类型的变量deq_ptr被赋值为，sn对ring_buffer所指向结构体的m_size成员进行取模运算的结果，。
 
    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等于，以deq_ptr下标，指向ring_buffer下mp_entries数组的元素，该元素为mp_ptr 。
} while (entry_ptr等于NULL或者entry_sn不等于sn) ;。
 
    // Read entry.
    *entry = entry_ptr;
把存储替换成原子存储，通过原子存储来进行替换，以原子存储替代存储 。
// 将 0 以可放宽原子访问的方式原子性存储到 ring_buffer 的 mp_entries 数组中 deq_ptr 下标的 mp_ptr 成员处 。
戒指缓冲器指向的那个，其条目编号为出队指针的，所对应的指向指针被设定成零 。
    // Update sequence number.
    // Replace store by atomic store
把环缓冲区中对应双端队列指针处的条目的序列号，设置为序列号加上环缓冲区的大小，以原子方式并且带着发布内存顺序进行存储，这里所指的环缓冲区的部分是即包含元素的那个数组中的具体位置，而双端队列指针是用于指示当前操作位置；原子特性的存储操作通过特定指针指向特定数组位置的元素部分来设置序列号，使用的是原子存储到指定值的函数并且内存顺序带有发布性质 。
ring_buffer所指向的mp_entries数组当中，处于deq_ptr位置的元素的m_sn成员，被赋值为sn加上ring_buffer的m_size属性的值。
 
    return 0;
}

代码 4 无锁 的 和 函数

函数里针对数据单元展开的load/store操作，能够进一步被原子操作予以取代，代码4当中的注释，给出了能够加以替换的原子操作。

无锁 实现的完整源码请参见()。

使用原子操作数据单元的无锁 实现的完整源码请参见()。

无锁 性能测试结果

图 10 展现出无锁在性能方面的情况。依据 mutex 的那一条曲线呈现的乃是基于 mutex 的实现方式；而另外两条曲线所代表的则都是无锁实现方式（其一是用到普通 load/store 操作数据单元来实现无锁的；其二是借助原子 load/store 操作数据单元来达成无锁的）。

图 10 无锁 性能测试

在图10里能够观察出，相较于基于mutex的那种实现，无锁的实现其性能有了极大的提升，将曲线与mutex曲线予以比较，在仅有2个线程的情况下，无锁的实现的吞吐率是基于mutex的实现的吞吐率的1.86倍，在使用32个线程时，无锁的实现的吞吐率是基于mutex的实现的吞吐率的18倍。

能够看得出，在图10当中，数据单元访问时，使用原子操作与否，对于性能所起到的影响并非显著。两条曲线大体上是重合的状态。原因在于，测试所提供的单元数量充足，多个线程共同分享单元所产生的竞争并不剧烈 。

不需要加锁方式实现的吞吐率仍旧会跟着子线程数值变多而有所下降，只不过并没有像基于mutex实现那样明显降低。于其使用2个子线程状况下，吞吐率是op/s；而当采用全部32个子线程的时候，吞吐率则是op/s。前面那种情形里吞吐率是后边那种情形下的1.27倍。

凭借Perf剖析程序热点，仅能定位至特定于其他某些函数，没办法给出更深层次的信息，所以运用Arm的top-down工具针对微架构执行状况施以详尽剖析，所获结果如图11 。

图 11 利用 Top-down 对无锁 进行性能分析

主要注意的指标，在图11中是用红圈进行标注的。首先，程序明确是属于后端瓶颈的那种程序，而后端阻塞周期高达97%。其次，L1、L2和LC（Last-level cache）明显高于其他异常情况，这意味着程序当中存在大量的数据竞争。最后，L1、L2和LC的数据缺失率属于同一量级，这意味着数据竞争是全局性的，要经系统的最后一级缓存才能得以解决。分析结果自上至下给出提示，无锁的那种实现居然仍旧致使了数量众多的全局性的数据冲突发生 。

无锁 的对齐改进

对程序予以分析后发现，符合所述特征的共享数据仅有head和tail指针，因为head和tail指针处在相同的缓存行之内，这致使生产者与消费者依旧在竞争同一缓存行，可借助对相关的数据结构开展缓存行对齐，来将不同生产者和消费者竞争同一缓存行的情形予以减少。

实现的缓存行对齐包含两个部分：

对头指针与尾指针进行对齐。把head指针，放置到不同的缓存行。把tail指针，也放置到不同的缓存行。这能够借助C语言原语来达成。

class RingBuffer
{
public:
使得对齐数量为（64）的，无符号整型的“m_size” ，。
将无符号整型的 m_head 进行对齐，对齐方式为 64 位，这是一种规定，用于确保该无符号整型在内存中的存储位置。
声明一个无符号整型变量m_tail，使其按照64字节进行对齐，。
以64字节对齐方式的，缓冲条目对象指针，被命名为mp_entries 。
}

代码 5 对齐的无锁 的声明

是这样的，仅有生产者去竞争处于tail指针所在之处的缓存行，而消费者会竞争位于head指针所在之地的缓存行。

为了避免访问不同单元的线程，因单元处于相同缓存行，从而产生竞争来进行数据单元的对齐，例如访问物理单元0 - 3的线程，实际上就在竞争同一缓存行 。

这一部分有望借助空间换效率的途径达成，一个缓存行六十四个字节里面能够安放四个单元，每个单元有十六个字节，包含八个字节的指针以及八个字节的序列号，在初始化阶段分配四倍数量的单元，在后续每次都仅仅使用四对齐的序号，逻辑单元零精确对应到物理单元零，逻辑单元一精准对应到物理单元四，依此类推 。

物理单元序号计算方法如下：

定义一个整型变量enq_ptr，其值通过将sn对ring_buffer的m_size取模后，再乘以4得到 。

代码 6 对齐的无锁 映射逻辑单元到物理单元

经对齐之后的，不存在自锁情况的内存分布情形如下，当中红色的那一部分，是确实被访问到的数据单元，而绿色的那一部分，则是没有被用到的数据单元 。

图 12 对齐后的无锁 的内存分布

需要加以说明的是，当出现物理数据单元的数量远远大于生产者以及消费者数量的情况时，数据单元的对齐针对性能所产生的影响表现得极为微小。由于共享一个缓存行的线程数量呈现出很小的态势，总共最多仅有4个 。

缓存行对齐的无锁 实现的完整源码请参见()。

对齐的无锁 性能测试结果

无锁且缓存行对齐的性能情况如图13所示，其中表述为，曲线mutex代表基于mutex的实现方式，还有两种表示无锁实现的情况，一种是使用普通load/store操作数据单元的无锁实现，另一种是使用原子load/store操作数据单元的无锁实现，而align表示缓存行对齐之后的无锁实现。

图 13 对齐的无锁 性能测试

缓存行对齐的那种无锁实现，其性能曲线，会随着子线程数量的增多，呈现出增长的态势。具体而言，。

首先，在子线程数小于16的这种情况下，吞吐率依据子线程数增添变得接近一个相对稳定的数值。而这个稳定的数值受到互联网络所能给予的最大带宽的限制。

在这之后，当子线程的数量超过 16 这个数值的时候，吞吐率会随着子线程数量的进一步增加而出现增加这样的情况。这样的一种现象和互联结构所具备的拓扑是有着相应关系的。当子线程的数量大于 32 这个数字的时候，新的子线程会被分配到互联结构的另一半部分当中，进而使得能够被利用的带宽呈现出增加的态势。

将 align 曲线与 mutex 曲线予以比较。在仅有2个线程的情形下，无锁实现的吞吐率，是基于 mutex 的实现的吞吐率的3.31倍；于使用16个线程时，无锁实现的吞吐率，则是基于 mutex 的实现的吞吐率的33.8倍。当采用32个线程时，无锁实现的吞吐率，为基于 mutex 的实现的吞吐率的110倍。

缓存行对齐的无锁实现，其性能曲线，和基于mutex的实现以及无锁的实现，有着本质上的不同，呈现出一种互联带宽测试的规律。这意味着，缓存行对齐的无锁实现，能够充分利用互联网络所提供的通信性能，是一个对系统规模扩展非常有利的实现。

非阻塞 实现

前文所介绍的无锁实现之中，存在着一个重大的功能方面的缺陷，即以上所提及的实现，仅仅能够构建而成阻塞式的 。在生产者或者消费者进入 要不然便是 函数之后，生产者或者消费者的程序，将会持续不断地尝试进行操作，直至相关操作获取成功，才得以推出 又或者是 函数呀。

在实际情况中 ，相关软件越发倾向于达成非阻塞性的 ，鉴于当某一操作或者另一操作未能成功施行时 ，该软件能够主动实施线程调度或者进入休眠状态 ，进而把处理器资源出让给能够开展操作的线程或者任务 ，这对于提升资源利用率具备积极作用 。

指针保护

这里分三个步骤来推演非阻塞 的实现。

首先，要达成非阻塞的目的来说，于函数开始之际，仍得去查验队列的空满情形。要是不具进行状态对应的必备条件，那就退出函数，最后返回错误代码。

考虑到这一点，非阻塞 的 和 的流程应当如下图：

图 14 非阻塞无锁 实现的流程图（第一步）

然后是步骤二，非阻塞状态下的实现方式的话，需要运用原子指令去替换 mutex 从而保护住指针。然而呢，虽说原子指令能够对一个共享变量的 load--store 序列予以保护，可它没办法保证针对多个共享变量的 load--store 之间具备原子性。所以呀，在一种处境里仅仅能够依靠原子指令来保护 tail 指针；在另一种处境中仅仅能够依靠原子指令来保护 head 指针。

考虑到这一点，非阻塞 的 和 的流程应当如下图：

图 15 非阻塞无锁 实现的流程图（第二步）

以某情况为例，函数先是读取指针，以此来判断空满条件。要是该情况不满，那就执行CAS(, tail, tail + 1)。

若 CAS 达成成功状态，那就意味，从对空满条件予以判断开始，直至更新 tail 为止的这段时长内，不存在其他线程对 tail 指针作出修改这一情况，如此一来，能够把 tail 所指向的单元分配给当前线程，并且对 CAS 进行更新。

倘若 CAS 遭遇失败，那就意味着在从判定空间条件起直至更新 tail 的这段时间当中，别的线程对 tail 指针作出了修改，这时，就得回到函数起始之处再度去判断空满条件。

tail指针如此这般更新能保障其更新于原子层面无误，每一个生产者均是在当下最新的tail值的基础之上开展更新作业着。

这里使用的是原语。

布尔型的，名为__atomic_compare_exchange_n的函数，其参数有指向某种类型的指针ptr，指向某种类型的指针expected，某种类型的desired，布尔型的weak，整数值的success_memorder，整数值的failure_memorder 。

读取，由ptr所指向的变量，去与这个指向的变量进行比较。要是两者的数值是相同的，那么就更新，ptr所指向的变量变为，并且返回真。倘若数值是不同的，那就把读取的数值，赋值给指向的变量，与此同时返回假。

换个角度从生产者来看，第三步里函数要做的是，先检查空满条件，之后给当前线程分配一个单元，还要让其他生产者清楚更新后的head或者tail的值，第二步引入的CAS操作把这点给保证啦。相对的，消费者得等分配的单元被填满了，才能知晓指针的更新情况。这表明，生产者和消费者看到指针更新的时间是不一样的。

因此，非阻塞无锁引入了两组指针，这两组指针分别针对生产者和消费者，也就是 /，和 /，就如同图 15 所展示的那样。

图 16 生产者和消费者指针示意图

就生产者而言，其那指针肯定是精确无误的，然而，那指针并非是最新的情况。这会致使生产者觉得其中已被占用的单元数量比实际上的数字还要多，进而致使生产者出现错误认知，觉得队列已满，随即便放弃了某操作。相似情形下，就消费者而言，其相应指针必定准确，可不巧的是，那指针并非处于最新状态。这般一来，消费有可能意识到空闲单元数量比实际数量还少，进而使得消费者错误地觉得队列满了，最终放弃了那操作。虽如此种误导判断会致使生产者以及消费者针对的竞争颇为保守，其结果乃效率降低，然而却不会引发功能错误。

考虑到这一点，非阻塞 的 和 的流程应当如下图：

图 17 非阻塞无锁 实现的流程图（第三步）

生产者和消费者的指针同步

当操作完结之际，得把处于生产者角度的指针，同步至处于消费者角度的指针，去告知消费者那些单元已然完成了操作；当操作完成之时，要将于消费者处的指针，同步到消费者的指针，来告知生产者这些单元已经完成了操作。

此情况显然无法天然予以保证，原因在于除针对特定指针的CAS操作外，相关函数其他各部处于并行运转状态，恰似图18所呈现之状况，如此这般，当某线程的分配单元完成特定数值或特定运算操作之际，并全然无从推测该单元之前的其他单元是否有已然完成相应数值或对应运算操作的情形发生。

图 18 非阻塞无锁 的执行时序示意图

存在着好些种用以实现指针同步方式的方法，就主要被划分成了顺序释放以及乱序释放这两类实现方法。

逐一依次地去进行释放，提出一种要求，即那些需要同步到另外一个视角的单元，必须已经完成了或者完成了相应的操作，这就使得在更新指针之前，必须针对单元展开检查，只有那些操作已然完成的单元以及能够按照要求完成操作的单元，才能够被同步给予至另外一个视角。

比如说，有一个线程的函数所分配的单元是16，在这个时候其值为10，那么这个函数就不可以进行更新，唯一能进行更新的是当值为15的时候开始更新，只有这样做才能够确保之前所存在的物理单元已然全部完成了操作。

要确保指针操作具备原子性，就得运用原语去更新处于另一个视角下的指针，。

指针所指向单元的释放顺序是混乱无序的，在对不同视角下的指针开展必要同步操作期间，并未针对单元是否已然彻底完成写入情况进行仔细检查，反而是径直将指针的数值增加了1，如此这般便会致使在后续时段里，或者是在后续的相关特定情境下，那种被分配的单元无法充分满足操作所提出的一系列要求。

比如，有某个线程，其函数分配的单元是16，这时的值是10。当这个函数完成之际，会将更新成11，然而此时针对物理单元11的操作状态并不知晓。物理单元11存在两种可能，有可能已经完成了操作，也有可能物理单元11当前依旧为空的。

若要防止功能出现错误，那么在对分配的单元予以操作之际就得开展检查。进而而言，针对第一种情况，要是分配的单元并非处于空闲状态，那就持续保持等待；与之相似的是，对于第二种情况，倘若分配的单元是空闲的，同样也要持续等待。

为了保证指针操作的原子性，需要使用原语更新另一个视角的指针。

截至此时，我们已经推导得出了，并非阻塞状态的，以及的完整流程，就如同图19所展示的那样。

图 19 非阻塞无锁 实现的流程图

声明

既非阻塞，又无锁，且不需要序列号，每个数据单元里仅有一个指向实际数据结构的指针 。

// Entry in ring buffer
class BufferEntry
{
public:
空指针类型的变量来存放指针 ，即mp_ptr ，它是用于指向数据项 。
};
class RingBuffer
{
public:
    alignas(64) unsigned int m_size;
 
从生产者的角度来看指针，指针涉及在其代码结构里的特定指向，这是生产行为开展过程里相关的一种存储地址的指示方式，它与生产所涉及的数据。
以64字节对齐，无符号整数，m_prod_head，此为读指针（处声明），。
对齐为64位，无符号整型，m_prod_tail是写指针，用于指示写入位置，是产品尾部相关的变量的声明 。
 
消费者视角下的指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，指针，。
被对齐为六十四的无符号整型的，称作m_cons_tail的，这么一个数值，它是用来作为 read pointer的 。
 
    alignas(64) BufferEntry *mp_entries;
};

代码 7 非阻塞无锁 的声明

和

本文测试的非阻塞 实现采用了乱序释放的方法。

要进入环形缓冲区，将指针指向环形缓冲区结构体，存入数据项，返回一个整数值，这个整数值代表存入操作的结果，有这样一个函数，请问哪个字没有按照原有词句的顺序出现呢 。
{
    // Step 1: acquire slots
无符号整型的“tail”，它等于通过“__atomic_load_n”函数，以非严格原子内存顺序加载“ring_buffer”的“m_prod_tail”成员的地址及值，加载之后的值赋给“tail”，并且加载操作之后有标点符号。
    do
    {
无符号整型的头，等于，原子加载数值，通过，__atomic_load_n(。
&ring_buffer 的 m_prod_head，以 __ATOMIC_ACQUIRE 的方式 。
假如，环形缓冲区的，那个大小，再加上，头部的值，等于，尾部的值 ，怎么办