本文轉(zhuǎn)載于極術(shù)社區(qū)

作者：非典型程序員

Ringbuffer（循環(huán)緩存）是軟件中非常常用的數(shù)據(jù)結(jié)構(gòu)之一， 在互聯(lián)網(wǎng)應(yīng)用、數(shù)據(jù)庫應(yīng)用等中使用廣泛。處理器執(zhí)行 Ringbuffer 的效率與其存儲(chǔ)系統(tǒng)處理共享數(shù)據(jù)的性能息息相關(guān)。

本文旨在介紹 Ringbuffer 的性能優(yōu)化方法。本文介紹基于 mutex 的 Ringbuffer 實(shí)現(xiàn)、阻塞的無鎖實(shí)現(xiàn)以及非阻塞的無鎖實(shí)現(xiàn)，并且比較和分析這些不同的實(shí)現(xiàn)在 Arm 服務(wù)器平臺(tái)上的性能。

聲明：本文介紹的針對 Ringbuffer 實(shí)現(xiàn)的優(yōu)化方法來自于 Ola Liljedahl 的研究。Ola 現(xiàn)任 Arm 公司的架構(gòu)師，長期專注于網(wǎng)絡(luò)軟件的性能優(yōu)化設(shè)計(jì)，尤其是可擴(kuò)展的多線程共享存儲(chǔ)編程。針對 Arm A 系列處理的多線程性能，他開發(fā)了一個(gè)可擴(kuò)展的多線程并行庫progress64(github.com/ARM-software/progress64)。

本文分析了 Ola 提出的主要優(yōu)化手段，并且在 Arm服務(wù)器上重現(xiàn)了 Ola 的優(yōu)化以驗(yàn)證其實(shí)際效果。博客中使用的代碼只是示例代碼(https://github.com/wangeddie67/ringbuffer_opt_demo)，其功能完備性和魯棒性都不能滿足實(shí)際應(yīng)用的需求。

介紹

Ringbuffer 通常由一個(gè)包含數(shù)據(jù)的數(shù)組和指向隊(duì)列頭（Head）和隊(duì)尾（Tail）的指針構(gòu)成（如圖 1 左邊所示）。指針只能按照一個(gè)方向移動(dòng)。當(dāng)指針移動(dòng)到數(shù)組結(jié)尾時(shí)，指針會(huì)折回到隊(duì)列開始。

Head 指針表示下一個(gè)可以彈出的數(shù)據(jù)；當(dāng)數(shù)據(jù)彈出隊(duì)列（稱為 Dequeue）時(shí)，Head 指針加 1。Tail 指針表示下一個(gè)可以壓入數(shù)據(jù)的單元；當(dāng)數(shù)據(jù)壓入隊(duì)列（稱為 Enqueue）時(shí)，Tail 指針加 1；

從邏輯上看，Ringbuffer 中的數(shù)組形成了一個(gè)環(huán)（如圖 1 右邊所示）。頭和尾指針在這個(gè)環(huán)上沿著相同的方向移動(dòng)。這就是 Ringbuffer 名稱的由來。

圖 1 Ringbuffer 示意圖

向 Ringbuffer 壓入單元的程序稱為生產(chǎn)者（Producer），從 Ringbuffer 彈出單元的程序稱為消費(fèi)者（Consumer），如圖 2 所示。一個(gè) Ringbuffer 通常具有很多的生產(chǎn)者和消費(fèi)者，每個(gè)生產(chǎn)者和消費(fèi)者都是不同的線程或進(jìn)程。這些線程或進(jìn)程會(huì)被調(diào)度到系統(tǒng)中的不同處理器核心上，而 Ringbuffer 則是不同處理器核心之間共享的數(shù)據(jù)。

圖 2 生產(chǎn)者和消費(fèi)者

基于 Mutex 的 Ringbuffer 實(shí)現(xiàn)

為了保證共享數(shù)據(jù)的功能正確，通常使用 mutex 來進(jìn)行保護(hù) Ringbuffer，以保證只有一個(gè)生產(chǎn)者或消費(fèi)者能夠訪問或操作 Ringbuffer。

基于 Mutex 的 Ringbuffer 聲明

Ringbuffer 的數(shù)據(jù)結(jié)構(gòu)如代碼塊 1 所示。

// Entry in ring buffer
class BufferEntry
{
public:
    void *mp_ptr;    // Pointer to data entry.
};


class RingBuffer
{
public:
    int m_mutex;                // Mutex for shared pointers.
    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.
};

代碼 1 基于 Mutex 的 Ringbuffer 聲明

數(shù)據(jù)數(shù)組中每個(gè)單元BufferEntry只提供一個(gè)指針，指針指向?qū)嶋H需要進(jìn)入 Ringbuffer 的數(shù)據(jù)結(jié)構(gòu)。當(dāng) Enqueue 和 Dequeue 時(shí)，只需要復(fù)制數(shù)據(jù)結(jié)構(gòu)的指針即可，而不需要復(fù)制數(shù)據(jù)，從而減少數(shù)據(jù)拷貝的開銷。

基于 Mutex 的 Ringbuffer 的 Enqueue 和 Dequeue 函數(shù)

圖 3 展示了 Enqueue 和 Dequeue 的流程圖。圖 3 左邊展示的是阻塞操作的流程圖；右邊展示的是非阻塞操作的流程圖。如果實(shí)現(xiàn)的是阻塞式的操作，那么函數(shù)一直重復(fù)檢查空滿條件，直到 Ringbuffer 的滿足操作所需的空滿條件；如果實(shí)現(xiàn)的是非阻塞式的操作，那么函數(shù)可以在檢查空滿條件失敗后直接退出。

圖 3 基于 Mutex 的 Ringbuffer 實(shí)現(xiàn)流程圖

循環(huán)隊(duì)列的空滿控制有很多種方法。在本系列文章的示例中采用邏輯單元和物理單元的控制方法。head 和 tail 的范圍并不只限于物理單元的數(shù)量，而是可以一直累加。訪問單元時(shí)，將(head MOD size)或(tail MOD size)得到對應(yīng)的物理單元。

對于 Enqueue 操作，需要 Ringbuffer 不為滿，即m_head+m_size!=m_tail（head 和 tail 之差不大于物理單元的數(shù)量）。

對于 Dequeue 操作，需要Ringbuufer 不為空，即m_head!=m_tail （head 和 tail 沒有指向同一個(gè)單元）。

基于 Mutex 的 Ringbuffer 實(shí)現(xiàn)如下：

// Blocking Enqueue Function
int enqueue_ringbuf(RingBuffer *ring_buffer, void *entry)
{
    // Lock mutex.
    lock(&ring_buffer->m_mutex);


    // Ringbuffer is full. Do nothing.
    while (ring_buffer->m_head + ring_buffer->m_size == ring_buffer->m_tail);


    // Enqueue entry.
    int enq_ptr = ring_buffer->m_tail % ring_buffer->m_size;
    ring_buffer->mp_entries[enq_ptr].mp_ptr = entry;


    // Increase tail pointer.
    ring_buffer->m_tail = ring_buffer->m_tail + 1;


    // Unlock mutex.
    unlock(&ring_buffer->m_mutex);


    return 0;
}


// Blocking Dequeue Function
int dequeue_ringbuf(RingBuffer *ring_buffer, void **entry)
{
    // Lock mutex.
    lock(&ring_buffer->m_mutex);


    // Ringbuffer is empty. Do nothing.
    while (ring_buffer->m_head == ring_buffer->m_tail);


    // Dequeue entry.
    int deq_ptr = ring_buffer->m_head % ring_buffer->m_size;
    *entry = ring_buffer->mp_entries[deq_ptr].mp_ptr;
    ring_buffer->mp_entries[deq_ptr].mp_ptr = NULL;


    // Increase head pointer.
    ring_buffer->m_head = ring_buffer->m_head + 1;


    // Unlock mutex.
    unlock(&ring_buffer->m_mutex);


    return 0;
}

代碼 2 基于 Mutex 的 Ringbuffer 實(shí)現(xiàn)

Lock 和 Unlock

Ringbuffer 最多允許一個(gè)生產(chǎn)者或一個(gè)消費(fèi)者對于 Ringbuffer 進(jìn)行操作。這需要通過鎖標(biāo)志m_mutex進(jìn)行控制。考慮到 pthread 中的 mutex API 引入的系統(tǒng)調(diào)度開銷過大，這里采用__sync_val_compare_and_swap原語實(shí)現(xiàn) lock 和 unlock 操作。

type __sync_val_compare_and_swap(type* ptr, type oldval, type newval);

__sync_val_compare_and_swap原語提供了同步和原子 CAS 操作兩個(gè)功能，其功能核心是 CASAL 指令。原語將ptr指向的變量與oldval比較，如果值相同則將ptr指向的變量修改為newval，并返回ptr指向變量的舊值；反之，則不修改ptr指向的變量，并返回ptr指向變量的值。同時(shí)，原語保證了 load/store 的同步，即在指令序上先于原語的 load/store 已經(jīng)完成，指令序上晚于原語的 load/store 不會(huì)執(zhí)行。
用__sync_val_compare_and_swap原語實(shí)現(xiàn)的 lock 和 unlock 功能如下：

// Lock function
void lock(int *mutex)
{
    while (__sync_val_compare_and_swap(mutex, 0, 1) != 0)
    {
    }
}


// Unlock function
void unlock(int *mutex)
{
    __sync_val_compare_and_swap(mutex, 1, 0);
}

代碼 3 lock 和 unlock 函數(shù)的實(shí)現(xiàn)

mutex變量為 0 表示mutex釋放；變量為 1 表示mutex鎖定。在 lock 函數(shù)中，如果當(dāng)前mutex為 0（釋放），則更新為 1（鎖定），并且 lock 函數(shù)返回 0；反之則 lock 函數(shù)返回 1。 基于 mutex 的 ringbuffer 實(shí)現(xiàn)的完整源碼請參見(https://github.com/wangeddie67/ringbuffer_opt_demo/blob/main/srcs/mutex_blkring.h)。

測試框架

測試程序包含一個(gè)控制測試過程的主線程和多個(gè)充當(dāng)生產(chǎn)者和消費(fèi)者的子線程。主線程首先創(chuàng)建并初始化 Ringbuffer，并且創(chuàng)建指定數(shù)量的子線程，最后等待子線程結(jié)束。子線程數(shù)量可以配置。 每個(gè)子線程即使生產(chǎn)者也是消費(fèi)者。每次測試迭代從隊(duì)列中讀取一個(gè)單元，再將這個(gè)單元（不做額外操作）插入隊(duì)尾，即一次 Dequeue 和一次 Enqueue 操作。整個(gè)測試過程需要執(zhí)行的 Enqueue 和 Dequeue 操作數(shù)量相同。如果將一部分線程設(shè)置為生產(chǎn)者，另一部分線程設(shè)置為消費(fèi)者，那么數(shù)據(jù)單元一定由生產(chǎn)者向消費(fèi)者移動(dòng)。這可能在互聯(lián)結(jié)構(gòu)中出現(xiàn)具有方向性的數(shù)據(jù)流，從而加強(qiáng)測試結(jié)果與線程調(diào)度的核心的物理位置的關(guān)系。 為了保證子線程同步啟動(dòng)測試，子線程首先會(huì)陷入 barrier。當(dāng)所有子程序都創(chuàng)建完成后，子線程從 barrier 退出。每個(gè)線程執(zhí)行的測試迭代數(shù)量都相同。

圖 4 測試程序流程圖 每個(gè)子線程獨(dú)立統(tǒng)計(jì)測試階段完成的時(shí)間，并且分別計(jì)算操作的吞吐率。這是為了避免統(tǒng)計(jì)過程引入新的共享數(shù)據(jù)。最后，主線程獲取每個(gè)子線程統(tǒng)計(jì)的操作吞吐率，并且求和得到總吞吐率。吞吐率用每秒執(zhí)行的 Enqueue 或 Dequeue 操作數(shù)量表示，即 op/s。 由于 Ringbuffer 的性能與存儲(chǔ)系統(tǒng)的性能有直接關(guān)聯(lián)，因此子線程調(diào)度到的核心對于具體實(shí)現(xiàn)具有很大的關(guān)系。為了保證測試的穩(wěn)定性，需要將子線程調(diào)度到指定的核心（綁核）。如果不做綁核，那么相同實(shí)現(xiàn)的相同配置的不同測試結(jié)果可能會(huì)大相徑庭，測試結(jié)果偏差極大。 測試框架的源碼請參見https://github.com/wangeddie67/ringbuffer_opt_demo/blob/main/testbench/testbench.cc)。

基于 Mutex 的 Ringbuffer 實(shí)現(xiàn)的測試結(jié)果

測試采用的 Arm 服務(wù)器具有 32 個(gè) N1 處理器，使用 CMN600 作為互聯(lián)結(jié)果。測試用的 Arm 服務(wù)器上并沒有區(qū)分 NUMA 結(jié)點(diǎn)，因此采用自然映射的關(guān)系，即子線程 0 調(diào)度到 CPU0；子線程 1 調(diào)度到 CPU1，以此類推。 圖 5 展示了基于 Mutex 的 Ringbuffer 的性能測試結(jié)果。顯然，隨著子線程數(shù)量的增加，Ringbuffer 的吞吐率顯著降低。當(dāng)使用 2 個(gè)子線程時(shí)，吞吐率為 3985230 op/s；當(dāng)使用全部 32 個(gè)子線程時(shí)，吞吐率為 323111 op/s。前者是后者的 12.3 倍。

圖 5 基于 Mutex 的 Ringbuffer 性能測試

圖 5 展示的性能測試結(jié)果是典型的以多線程競爭為主導(dǎo)的性能曲線。隨著參與競爭的子線程數(shù)增加，在競爭 mutex 上花費(fèi)的時(shí)間就更多，導(dǎo)致性能快速下降。 使用 perf 工具對于啟動(dòng) 16 個(gè)線程的測試程序進(jìn)行熱點(diǎn)分析，熱點(diǎn)分析結(jié)果如圖 6 所示。圖中橙色部分提示了lock函數(shù)占用了程序執(zhí)行時(shí)間的 91%。因此，對于 Ringbuffer 進(jìn)行優(yōu)化的首要步驟就是采用無鎖的實(shí)現(xiàn)方式。 圖 6 Perf 熱點(diǎn)分析結(jié)果

無鎖 Ringbuffer 實(shí)現(xiàn)

對于多線程共享程序，需要使用 mutex 來保護(hù)對于共享變量的 load-modify-store 代碼序列，從而保證功能正確性。但是 mutex 的競爭引入了非常明顯的性能開銷。因此，需要探索無鎖的 Ringbuffer 實(shí)現(xiàn)。

用原子操作替代鎖

除了 mutex，另一種可以保護(hù)共享變量的方法是原子操作。單個(gè)原子操作就是由 load-modify-store 三個(gè)部分構(gòu)成。原子操作保證，在完成 load-modify-store 序列之前，被訪問的地址不會(huì)被其他指令訪問。典型的原子操作包括 Compare-and-swap （CAS），以及 Arm 提供的 LDADD 等原子操作。如果有多個(gè) PE 針對同一個(gè)共享變量發(fā)起多個(gè)原子操作，那么這些操作只能按照某種順序串行，而不能并行。執(zhí)行順序取決于硬件實(shí)現(xiàn)。 CAS 保證了對于 mutex 的操作都是串行的。例如 PE1 執(zhí)行原子操作 CAS(x, 0, 1)，同時(shí) PE2 執(zhí)行原子操作 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 實(shí)現(xiàn)中，特別需要保護(hù)的就是被所有生產(chǎn)者和消費(fèi)者共享的 head 和 tail 指針?？梢杂胈_atomic_fetch_add原語來實(shí)現(xiàn)指針的原子指令并且實(shí)現(xiàn)指針的累加。 __atomic_fetch_add原語的聲明如下，其核心指令是 LDADD。原語讀取ptr指向的變量，然后將變量累加val后的新值會(huì)寫到原來的位置；同時(shí)，原語將ptr指令的變量的舊值返回。

type __atomic_fetch_add (type *ptr, type val, int memorder)

將__atomic_fetch_add原語應(yīng)用到 head 或 tail 指針上，則可以將存儲(chǔ)在內(nèi)存中的指針變量加 1，并且同時(shí)將加 1 前的舊值返回。這就相當(dāng)于將目前 head/tail 指針指向的單元分配給了當(dāng)前的 Enqueue 或 Dequeue 的生產(chǎn)者或消費(fèi)者，同時(shí)與其他生產(chǎn)者或消費(fèi)者同步了更新后的指針。

Enqueue 和 Dequeue

圖 7 展示了 Enqueue 和 Dequeue 的流程圖。

圖 7 無鎖 Ringbuffer 實(shí)現(xiàn)流程圖（使用原子操作替換 mutex） 相對于基于 mutex 的 Ringbuffer 實(shí)現(xiàn)，無鎖 Ringbuffer 實(shí)現(xiàn)從軟件和硬件兩方面的優(yōu)化。 從軟件的角度，無鎖 Ringbuffer 的關(guān)鍵區(qū)要小很多。在基于 mutex 的 Ringbuffer 實(shí)現(xiàn)（圖 8 左側(cè)）中，鎖保護(hù)了整個(gè) Enqueue 和 Dequeue 過程，每個(gè) Enqueue 和 Dequeue 過程都只能串行執(zhí)行。在無鎖 Ringbuffer 實(shí)現(xiàn)（圖 8 右側(cè)）中，原子指令只保護(hù)了對于 head 或 tail 指針的累加。除了__atomic_fetch_add必須串行，其他部分都是可以并行的，從而增加了程序的并行度。

圖 8 基于 mutex 的 Ringbuffer 實(shí)現(xiàn)和無鎖 Ringbuffer 實(shí)現(xiàn)的關(guān)鍵區(qū)（著色部分）對比 從硬件的角度，原子指令引入更少的 snoop 操作。使用常規(guī)的 load 指令讀取數(shù)據(jù)后，在 PE 側(cè)的緩存中的數(shù)據(jù)副本的狀態(tài)為 Share（共享）。所有訪問這個(gè)數(shù)據(jù)的 PE 的緩存都會(huì)具有這個(gè)共享數(shù)據(jù)的副本。當(dāng)共享數(shù)據(jù)被某個(gè) PE 修改后，需要大量的 snoop 去失效其他 PE 中的所有副本。 使用原子指令讀取數(shù)據(jù)之后，數(shù)據(jù)的緩存狀態(tài)為 Exclusive（獨(dú)有），即只有一個(gè) PE 具有這個(gè)共享數(shù)據(jù)的副本。當(dāng)另一個(gè) PE 也執(zhí)行原子操作后，只需要一個(gè) snoop 就可以將已經(jīng)存在副本清除。 從帶寬的角度，更少 snoop 數(shù)量意味著 snoop 通信占用的通信帶寬更少，提供給數(shù)據(jù)通信的帶寬就會(huì)更高。從延遲的角度，load/store 操作需要等到其引起的所有 snoop 都完成才能被標(biāo)識(shí)為完成。更少的 snoop 數(shù)量意味著等待 snoop 完成的時(shí)間更短。 無鎖 Ringbuffer 同時(shí)從這兩方面優(yōu)化中受惠，暫時(shí)沒有辦法定量分析出哪一個(gè)優(yōu)化的作用更大。一般來說，縮小關(guān)鍵區(qū)的作用更大。

邏輯單元序號(hào)

上述的無鎖 Ringbuffer 仍然可能導(dǎo)致功能錯(cuò)誤。例如一個(gè) Ringbuffer 有 16 個(gè)物理單元。首先為生產(chǎn)者 A 分配的邏輯單元是 16；隨后，為生產(chǎn)者 B 分配的單元是 32；這兩個(gè)單元映射到相同的物理單元 0。雖然單元分配和指針更新是串行的，但是對于數(shù)據(jù)單元的訪問是并行的，因此并不能保證生產(chǎn)者 B 訪問物理單元 0 時(shí)，生產(chǎn)者 A 一定完成了操作?？赡苌a(chǎn)者 A 和生產(chǎn)者 B 同時(shí)查詢到單元是空閑，進(jìn)而同時(shí)對這個(gè)物理單元進(jìn)行寫入。此時(shí)無法保證寫入單元的究竟是生產(chǎn)者 A 的數(shù)據(jù)還是生產(chǎn)者 B 的數(shù)據(jù)。 為了解決這樣的問題，Ola 提出，給每個(gè)單元增加一個(gè)域來表示其對應(yīng)的邏輯單元序號(hào)。只有當(dāng)邏輯單元序號(hào)與 head/tail 指針一致的時(shí)候才能進(jìn)行訪問。當(dāng) Dequeue 單元的時(shí)候，需要更新邏輯單元序號(hào)，將單元的序列號(hào)增加物理單元的總數(shù)。 還是上面的例子，當(dāng)生產(chǎn)者 A 和 B 同時(shí)查詢到內(nèi)存是空閑時(shí)，由于單元的序列號(hào)與生產(chǎn)者 A 分配的單元匹配，那么生產(chǎn)者 A 可以操作這個(gè)單元；生產(chǎn)者 B 則因?yàn)樾蛄刑?hào)不匹配而不能訪問，需要繼續(xù)等待。 增加邏輯單元序號(hào)后的 Enqueue 和 Dequeue 流程圖如圖 9。

圖 9 無鎖 Ringbuffer 實(shí)現(xiàn)流程圖（使用原子操作替換 mutex）

無鎖 Ringbuffer 聲明

無鎖 Ringbuffer 的聲明如下。Ringbuffer 不需要 mutex 進(jìn)行保護(hù)。

// Entry in ring buffer
class 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 函數(shù)

Enqueue 和 Dequeue 函數(shù)的實(shí)現(xiàn)如下。

// Blocking Enqueue Function
int 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 Function
int 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
    {
        // Replace load by atomic load.
        // 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.
    // Replace store by atomic store
    // __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 函數(shù)

Enqueue 和 Dequeue 函數(shù)中對于數(shù)據(jù)單元的 load/store 操作可以進(jìn)一步被原子操作__atomic_load_n和__atomic_store_n取代。代碼 4 中的注釋給出了可以替換的原子操作。

無鎖 ringbuffer 實(shí)現(xiàn)的完整源碼請參見(https://github.com/wangeddie67/ringbuffer_opt_demo/blob/main/srcs/lockfree_blkring.h)。

使用原子操作數(shù)據(jù)單元的無鎖 Ringbuffer 實(shí)現(xiàn)的完整源碼請參見(https://github.com/wangeddie67/ringbuffer_opt_demo/blob/main/srcs/atomic_blkring.h)。

無鎖 Ringbuffer 性能測試結(jié)果

無鎖 Ringbuffer 的性能如圖 10 所示。曲線 mutex 表示基于 mutex 的 Ringbuffer 實(shí)現(xiàn)；lockfree 和 atomic 表示無鎖 Ringbuffer 實(shí)現(xiàn)（lockfree 使用普通 load/store 操作數(shù)據(jù)單元；atomic 使用原子 load/store 操作數(shù)據(jù)單元）。

圖 10 無鎖 Ringbuffer 性能測試

從圖 10 中可以發(fā)現(xiàn)，相對于基于 mutex 的 Ringbuffer 實(shí)現(xiàn)，無鎖 Ringbuffer 實(shí)現(xiàn)的性能得到了很大的提升。用 atomic 曲線和 mutex 曲線進(jìn)行比較。當(dāng)只有 2 個(gè)線程時(shí)，無鎖 Ringbuffer 實(shí)現(xiàn)的吞吐率是基于 mutex 的 Ringbuffer 實(shí)現(xiàn)的吞吐率的 1.86 倍；當(dāng)使用 32 個(gè)線程時(shí)，無鎖 Ringbuffer 實(shí)現(xiàn)的吞吐率是基于 mutex 的 Ringbuffer 實(shí)現(xiàn)的吞吐率的 18 倍。

從圖 10 中可以發(fā)現(xiàn)，對于數(shù)據(jù)單元的訪問是否使用原子操作對于性能的影響并不明顯。兩條曲線基本上是重合的。這是因?yàn)闇y試的 ringbuffer 提供了足夠的單元，多個(gè)線程共享 Ringbuffer 單元的競爭并不強(qiáng)烈。

無鎖 Ringbuffer 實(shí)現(xiàn)的吞吐率還是會(huì)隨著子線程數(shù)增加而降低，但是并不如基于 mutex 的 Ringbuffer 強(qiáng)烈。當(dāng)使用 2 個(gè)子線程時(shí)，吞吐率為 7394830 op/s；當(dāng)使用全部 32 個(gè)子線程時(shí)，吞吐率為 5800350 op/s。前者是后者的 1.27 倍。

利用 Perf 分析程序熱點(diǎn)只能定位到特點(diǎn)在 enqueue 和 dequeue 函數(shù)，無法提供進(jìn)一步的信息。因此使用 Arm 的 top-down 工具對于微架構(gòu)執(zhí)行情況進(jìn)行詳細(xì)分析，得到的結(jié)果如圖 11。

圖 11 利用 Top-down 對無鎖 Ringbuffer 進(jìn)行性能分析

圖 11 中用紅圈標(biāo)注了主要注意的指標(biāo)。首先，程序明確屬于后端瓶頸的程序，后端阻塞周期高達(dá) 97%。其次，L1、L2 和 LC（Last-level cache）明顯高于其他異常情況，這表示程序中存在大量的數(shù)據(jù)競爭。最后 L1、L2 和 LC 的數(shù)據(jù)缺失率屬于同一量級，這表示數(shù)據(jù)競爭是全局性的，需要通過系統(tǒng)的最后一級緩存才能得到解決。top-down 的分析結(jié)果提示，無鎖的 ringbuffer 實(shí)現(xiàn)仍然導(dǎo)致了大量的全局性的數(shù)據(jù)沖突。

無鎖 Ringbuffer 的對齊改進(jìn)

通過對于程序的分析，符合上述特征的共享數(shù)據(jù)只有 head 和 tail 指針。因?yàn)?head 和 tail 指針處于相同的緩存行。這就導(dǎo)致生產(chǎn)者和消費(fèi)者仍然在競爭同一個(gè)緩存行?？梢酝ㄟ^對與 Ringbuffer 的數(shù)據(jù)結(jié)構(gòu)進(jìn)行緩存行對齊，減少不同生產(chǎn)者和消費(fèi)者競爭同一個(gè)緩存行的情況。

Ringbuffer 實(shí)現(xiàn)的緩存行對齊包含兩個(gè)部分：

head/tail 指針的對齊。將 head 指針和 tail 指針分別放置到不同的緩存行。這可以通過 C 語言原語 alignas 實(shí)現(xiàn)。

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 的聲明

這樣只有生產(chǎn)者競爭 tail 指針?biāo)诘木彺嫘?；消費(fèi)者競爭 head 指針?biāo)诘木彺嫘小?br />數(shù)據(jù)單元的對齊。這是為了避免訪問不同單元的線程因?yàn)閱卧幱谙嗤彺嫘卸a(chǎn)生競爭。比如訪問物理單元 0-3 的線程實(shí)際上在競爭同一個(gè)緩存行。
這一部分可以通過空間換效率的方式實(shí)現(xiàn)。一個(gè)緩存行（64B）中可以放置 4 個(gè)單元（每個(gè)單元 16B，8B 指針和 8B 序列號(hào)）。在初始化時(shí)分配 4 倍的單元，在 Enqueue 和 Dequeue 每次都只使用 4 對齊的序號(hào)。邏輯單元 0 對應(yīng)到物理單元 0；邏輯單元 1 對應(yīng)到物理單元 4；依次類推。
物理單元序號(hào)計(jì)算方法如下：

int enq_ptr = (sn % ring_buffer->m_size) * 4;

代碼 6 對齊的無鎖 Ringbuffer 映射邏輯單元到物理單元

對齊后的無鎖 ringbuffer 的內(nèi)存分布如下。其中紅色部分是實(shí)際訪問的數(shù)據(jù)單元。綠色部分是沒有使用到的數(shù)據(jù)單元。

圖 12 對齊后的無鎖 Ringbuffer 的內(nèi)存分布

需要說明的是，當(dāng)物理數(shù)據(jù)單元數(shù)量遠(yuǎn)大于生產(chǎn)者和消費(fèi)者數(shù)量時(shí)，數(shù)據(jù)單元的對齊對于性能的影響非常微小。因?yàn)楣蚕硪粋€(gè)緩存行的線程數(shù)量很小，最多只有 4 個(gè)。

緩存行對齊的無鎖 ringbuffer 實(shí)現(xiàn)的完整源碼請參見(https://github.com/wangeddie67/ringbuffer_opt_demo/blob/main/srcs/align_blkring.h)。

對齊的無鎖 Ringbuffer 性能測試結(jié)果

緩存行對齊的無鎖 Ringbuffer 的性能如圖 13 所示。曲線 mutex 表示基于 mutex 的 Ringbuffer 實(shí)現(xiàn)；lockfree 和 atomic 表示無鎖 Ringbuffer 實(shí)現(xiàn)（lockfree 使用普通 load/store 操作數(shù)據(jù)單元；atomic 使用原子 load/store 操作數(shù)據(jù)單元）；align 表示緩存行對齊后的無鎖 Ringbuffer 實(shí)現(xiàn)。

圖 13 對齊的無鎖 Ringbuffer 性能測試

緩存行對齊的無鎖 Ringbuffer 實(shí)現(xiàn)的性能曲線隨著子線程數(shù)量的增加呈現(xiàn)增長的趨勢。具體來說，

首先，當(dāng)子線程數(shù)小于 16 時(shí)，吞吐率隨著子線程數(shù)增加而趨向于一個(gè)穩(wěn)定的值。這個(gè)穩(wěn)定值受限于是互聯(lián)網(wǎng)絡(luò)能夠提供的最大帶寬。

其次，當(dāng)子線程數(shù)大于 16 時(shí)，吞吐率隨著子線程數(shù)增加而增加。這與互聯(lián)結(jié)構(gòu)的拓?fù)溆嘘P(guān)系。當(dāng)子線程數(shù)大于 32 時(shí)，新的子線程被分配到互聯(lián)結(jié)構(gòu)的另一半部分，從而使得能夠利用的帶寬增加。

用 align 曲線和 mutex 曲線進(jìn)行比較。當(dāng)只有 2 個(gè)線程時(shí)，無鎖 Ringbuffer 實(shí)現(xiàn)的吞吐率是基于 mutex 的 Ringbuffer 實(shí)現(xiàn)的吞吐率的 3.31 倍；當(dāng)使用 16 個(gè)線程時(shí)，無鎖 Ringbuffer 實(shí)現(xiàn)的吞吐率是基于 mutex 的 Ringbuffer 實(shí)現(xiàn)的吞吐率的 33.8 倍。當(dāng)使用 32 個(gè)線程時(shí)，無鎖 Ringbuffer 實(shí)現(xiàn)的吞吐率是基于 mutex 的 Ringbuffer 實(shí)現(xiàn)的吞吐率的 110 倍。

緩存行對齊的無鎖 Ringbuffer 實(shí)現(xiàn)的性能曲線與基于 mutex 的 Ringbuffer 實(shí)現(xiàn)和無鎖 Ringbuffer 的實(shí)現(xiàn)有本質(zhì)上的不同，呈現(xiàn)了一種互聯(lián)帶寬測試的規(guī)律。這表示緩存行對齊的無鎖 Ringbuffer 實(shí)現(xiàn)能夠充分利用互聯(lián)網(wǎng)絡(luò)提供的通信性能，是一個(gè)非常有利于系統(tǒng)規(guī)模擴(kuò)展的 Ringbuffer 實(shí)現(xiàn)。

非阻塞 Ringbuffer 實(shí)現(xiàn)

前文介紹的無鎖 Ringbuffer 實(shí)現(xiàn)存在一個(gè)重大的功能缺陷，就是上述實(shí)現(xiàn)只能構(gòu)建阻塞式的 Ringbuffer。當(dāng)生產(chǎn)者或消費(fèi)者進(jìn)入 Enqueue 或 Dequeue 函數(shù)后，生產(chǎn)者或消費(fèi)者的程序會(huì)一直嘗試操作，直到操作成功才能推出 Enqueue 或 Dequeue 函數(shù)。

在實(shí)際軟件更加傾向于實(shí)現(xiàn)非阻塞的 Ringbuffer。因?yàn)楫?dāng) Enqueue 或 Dequeue 操作不成功時(shí)，軟件可以主動(dòng)進(jìn)行線程調(diào)度或休眠，將處理器資源讓渡給可以進(jìn)行操作的線程或任務(wù)。這有利于提高資源利用率。

指針保護(hù)

這里分三個(gè)步驟來推演非阻塞 Ringbuffer 的實(shí)現(xiàn)。

第一步。為了能夠?qū)崿F(xiàn)非阻塞的 Ringbuffer，那么仍然需要在 Enqueue 和 Dequeue 函數(shù)的開始的時(shí)候檢查隊(duì)列的空滿情況。如果 Ringbuffer 不滿足進(jìn)行 Enqueue 或 Dequeue 的條件，則退出函數(shù)并且返回錯(cuò)誤代碼。

考慮到這一點(diǎn)，非阻塞 Ringbuffer 的 Enqueue 和 Dequeue 的流程應(yīng)當(dāng)如下圖：

圖 14 非阻塞無鎖 Ringbuffer 實(shí)現(xiàn)的流程圖（第一步）

第二步。非阻塞實(shí)現(xiàn)要使用原子指令替代 mutex 保護(hù)指針。但是原子指令可以保護(hù)對于一個(gè)共享變量的 load-modify-store 序列，但是不能保證對于多個(gè)共享變量的 load-modify-store 之間的原子性。因此，Enqueue 中只能用原子指令保護(hù) tail 指針；Dequeue 中只能用原子指令保護(hù) head 指針。

考慮到這一點(diǎn)，非阻塞 Ringbuffer 的 Enqueue 和 Dequeue 的流程應(yīng)當(dāng)如下圖：

圖 15 非阻塞無鎖 Ringbuffer 實(shí)現(xiàn)的流程圖（第二步）

以 enqueue 為例，函數(shù)首先讀取指針判斷空滿條件。如果 Ringbuffer 不滿，則執(zhí)行 CAS(tail_addr, tail, tail+1)。

如果 CAS 成功，則表示從判斷空滿條件到更新 tail 這段時(shí)間，沒有其他線程修改了 tail 指針，那么可以將 tail 指向的單元分配給當(dāng)前線程，并且更新 CAS。

如果 CAS 失敗，則表示從判斷空間條件到更新 tail 這段時(shí)間，其他線程修改了 tail 指針。此時(shí)，需要返回函數(shù)開始的地方重新判斷空滿條件。

這種方法可以保證 tail 指針的更新是原子的，每個(gè)生產(chǎn)者都是在最新的 tail 值基礎(chǔ)上進(jìn)行更新。

這里使用的是__atomic_compare_exchange_n原語。

bool __atomic_compare_exchange_n (type *ptr, type *expected, type desired, bool weak, int success_memorder, int failure_memorder）

__atomic_compare_exchange_n讀取ptr指向的變量，與expected指向的變量比較。如果兩者數(shù)值相同，則更新ptr指向的變量為desired，并且返回真。如果數(shù)值不同，則將讀取的數(shù)值賦值給expected指向的變量，同時(shí)返回假。

第三步。從生產(chǎn)者的角度，Enqueue 函數(shù)需要在檢查空滿條件后分配一個(gè)單元給當(dāng)前線程，并且讓其他生產(chǎn)者知道更新后的 head 或 tail 值。第二步中引入的 CAS 操作保證了這一點(diǎn)。另一方面，消費(fèi)者需要等到分配的單元被填充后才能知道指針的更新。也就是說，生產(chǎn)者和消費(fèi)者看到指針更新的時(shí)間不同。

因此，非阻塞無鎖 Ringbuffer 引入了兩組指針分別針對生產(chǎn)者和消費(fèi)者，即 prod_head/prod_tail，和 cons_head/cons_tail，如圖 15 所示。

圖 16 生產(chǎn)者和消費(fèi)者指針示意圖

對于生產(chǎn)者來說，其 prod_tail 指針一定是準(zhǔn)確的，但是 prod_head 指針并不是最新的。這會(huì)導(dǎo)致生產(chǎn)者認(rèn)為 Ringbuffer 中被占用的單元數(shù)多于實(shí)際被占用的單元數(shù)量，從而導(dǎo)致生產(chǎn)者錯(cuò)誤地認(rèn)為隊(duì)列是滿的，而放棄 Enqueue 操作。類似地，對于消費(fèi)者來說，其 cons_head 指針一定是準(zhǔn)確的，但是 cons_tail 指針并不是最新的。那么消費(fèi)者可能認(rèn)為空閑單元少于實(shí)際的空閑單元，從而導(dǎo)致消費(fèi)者錯(cuò)誤地認(rèn)為隊(duì)列是滿的，而放棄 Dequeue 操作。這樣的誤導(dǎo)判斷雖然會(huì)導(dǎo)致生產(chǎn)者和消費(fèi)者對于 Ringbuffer 的競爭比較保守，結(jié)果是 Ringbuffer 效率降低，但是不會(huì)引起功能錯(cuò)誤。

考慮到這一點(diǎn)，非阻塞 Ringbuffer 的 Enqueue 和 Dequeue 的流程應(yīng)當(dāng)如下圖：

圖 17 非阻塞無鎖 Ringbuffer 實(shí)現(xiàn)的流程圖（第三步）

生產(chǎn)者和消費(fèi)者的指針同步

當(dāng) Enqueue 操作完成時(shí)，需要將生產(chǎn)者視角的 prod_tail 指針同步到消費(fèi)者視角的 cons_tail 指針，告知消費(fèi)者這些單元已經(jīng)完成了 Enqueue 操作；當(dāng) Dequeue 操作完成時(shí)，需要將消費(fèi)者的 cons_head 指針同步到消費(fèi)者的 prod_head 指針，告知生產(chǎn)者這些單元已經(jīng)完成了 Dequeue 操作。

但是這并不能夠天然保證的。因?yàn)槌酸槍?prod_tail 和 cons_head 指針的 CAS 操作，Enqueue 和 Dequeue 函數(shù)的其他部分都是并行，如圖 18 所示。因此，某個(gè)線程的分配單元完成 Enqueue 或 Dequeue 操作時(shí)，完全不能推測這個(gè)單元之前的其他單元是否完成了 Enqueue 或 Dequeue 的操作。

圖 18 非阻塞無鎖 Ringbuffer 的執(zhí)行時(shí)序示意圖

指針同步方式有很多種實(shí)現(xiàn)方法，主要分為順序釋放和亂序釋放兩種實(shí)現(xiàn)方法。

順序釋放。要求同步到另一個(gè)視角的單元必須完成了 Enqueue 或 Dequeue 操作。這就需要在更新指針前對于單元進(jìn)行檢查，只能將操作完成的單元可以同步給另一個(gè)視角。

比如，某個(gè)線程的 Enqueue 函數(shù)分配的單元為 16，此時(shí) cons_tail 的值是 10，則 Enqueue 函數(shù)不能更新 cons_tail。只有 cons_tail 的值是 15 時(shí)才能更新 cons_tail。這樣才能保證之前的物理單元都已經(jīng)完成了 Enqueue 操作。

為了保證指針操作的原子性，需要使用__atomic_compare_exchange_n原語更新另一個(gè)視角的指針。

亂序釋放。在同步不同視角的指針時(shí)，不檢查單元是否已經(jīng)完全寫入完成，而是直接將指針加 1。這就會(huì)導(dǎo)致在后續(xù) Enqueue 或 Dequeue 種分配的單元不能滿足操作的要求。

比如，某個(gè)線程的 Enqueue 函數(shù)分配的單元為 16，此時(shí) cons_tail 的值是 10。當(dāng)這個(gè) Enqueue 函數(shù)完成時(shí)將 cons_tail 更新為 11，但是此時(shí)對于物理單元 11 的操作狀態(tài)未知。物理單元 11 有可能已經(jīng)完成了 Enqueue 操作，也有可能物理單元 11 還是空的。

為了避免功能錯(cuò)誤，則需要在操作分配的單元時(shí)進(jìn)行檢查。對于 Enqueue，如果分配的單元不為空，則一直等待；類似地，對于 Dequeue，如果分配的單元為空，則也要一直等待。

為了保證指針操作的原子性，需要使用__atomic_fetch_add原語更新另一個(gè)視角的指針。

至此，我們推導(dǎo)出了，非阻塞 Ringbuffer 的 Enqueue 和 Dequeue 的完整流程，如圖 19 所示。

圖 19 非阻塞無鎖 Ringbuffer 實(shí)現(xiàn)的流程圖

Ringbuffer 聲明

非阻塞的無鎖 Ringbuffer 不需要序列號(hào)。每個(gè)數(shù)據(jù)單元中只有一個(gè)指向?qū)嶋H數(shù)據(jù)結(jié)構(gòu)的指針。

// Entry in ring buffer
class BufferEntry
{
public:
    void *mp_ptr; // Pointer to data entry.
};


class RingBuffer
{
public:
    alignas(64) unsigned int m_size;
 
    // Pointers in the view of producer.
    alignas(64) unsigned int m_prod_head; // read pointer
    alignas(64) unsigned int m_prod_tail; // write pointer
 
    // Pointers in the view of consumer.
    alignas(64) unsigned int m_cons_head; // write pointer
    alignas(64) unsigned int m_cons_tail; // read pointer
 
    alignas(64) BufferEntry *mp_entries;
};

代碼 7 非阻塞無鎖 Ringbuffer 的聲明

Enqueue 和 Dequeue

本文測試的非阻塞 Ringbuffer 實(shí)現(xiàn)采用了亂序釋放的方法。

 int enqueue_ringbuf(RingBuffer *ring_buffer, void *entry)
{
    // Step 1: acquire slots
    unsigned int tail = __atomic_load_n(&ring_buffer->m_prod_tail, __ATOMIC_RELAXED);
    do
    {
        unsigned int head = __atomic_load_n(
            &ring_buffer->m_prod_head, __ATOMIC_ACQUIRE);
        if (ring_buffer->m_size + head == tail)
        {
            return 1;
        }
    } while (!__atomic_compare_exchange_n(&ring_buffer->m_prod_tail,
                                          &tail, // Updated on failure
                                          tail + 1,
                                          /*weak=*/true,
                                          __ATOMIC_RELAXED,
                                          __ATOMIC_RELAXED));
 
    // Step 2: Write slots
    int enq_ptr = (tail % ring_buffer->m_size) * 8;
    void *entry_ptr;
    do
    {
        entry_ptr = __atomic_load_n(&ring_buffer->mp_entries[enq_ptr].mp_ptr, __ATOMIC_RELAXED);
    } while (entry_ptr != NULL);
 
    __atomic_store_n(&ring_buffer->mp_entries[enq_ptr].mp_ptr, entry, __ATOMIC_RELEASE);
 
    // Finally make all released slots available for new acquisitions
    __atomic_fetch_add(&ring_buffer->m_cons_tail, 1, __ATOMIC_RELEASE);
    return 0;
}
 
int dequeue_ringbuf(RingBuffer *ring_buffer, void **entry)
{
    // Step 1: acquire slots
    unsigned int tail = __atomic_load_n(&ring_buffer->m_cons_head, __ATOMIC_RELAXED);
    do
    {
        unsigned int head = __atomic_load_n(
            &ring_buffer->m_cons_tail, __ATOMIC_ACQUIRE);
        if (head == tail)
        {
            return 1;
        }
    } while (!__atomic_compare_exchange_n(&ring_buffer->m_cons_head,
                                          &tail, // Updated on failure
                                          tail + 1,
                                          /*weak=*/true,
                                          __ATOMIC_RELAXED,
                                          __ATOMIC_RELAXED));
 
    // Step 2: Write slots (write NIL for dequeue)
    int deq_ptr = (tail % ring_buffer->m_size) * 8;
    unsigned int entry_sn;
    void *entry_ptr;
    do
    {
        entry_ptr = __atomic_load_n(&ring_buffer->mp_entries[deq_ptr].mp_ptr, __ATOMIC_ACQUIRE);
    } while (entry_ptr == NULL);
 
    *entry = entry_ptr;
    __atomic_store_n(&ring_buffer->mp_entries[deq_ptr].mp_ptr, 0, __ATOMIC_RELAXED);
 
    // Finally make all released slots available for new acquisitions
    __atomic_fetch_add(&ring_buffer->m_prod_head, 1, __ATOMIC_RELEASE);
    return 0;
}

代碼 8 非阻塞無鎖 Ringbuffer 的實(shí)現(xiàn)

非阻塞的無鎖 ringbuffer 實(shí)現(xiàn)的完整源碼請參見(https://github.com/wangeddie67/ringbuffer_opt_demo/blob/main/srcs/buck_blkring.h)。

非阻塞 Ringbuffer 性能測試結(jié)果

非阻塞的無鎖 Ringbuffer 的性能如圖 13 所示。曲線 mutex 表示基于 mutex 的 Ringbuffer 實(shí)現(xiàn)；lockfree 和 atomic 表示無鎖 Ringbuffer 實(shí)現(xiàn)（lockfree 使用普通 load/store 操作數(shù)據(jù)單元；atomic 使用原子 load/store 操作數(shù)據(jù)單元）；align 表示緩存行對齊后的無鎖 Ringbuffer 實(shí)現(xiàn)；buck 表示的非阻塞無鎖 Ringbuffer 實(shí)現(xiàn)。

圖 20 非阻塞無鎖 Ringbuffer 性能測試

非阻塞的無鎖 Ringbuffer 的性能曲線呈現(xiàn)兩段趨勢：

從 2 個(gè)線程到 16 個(gè)線程左右，吞吐率隨著線程數(shù)增加而緩慢增加。

從 16 個(gè)線程左右到 32 個(gè)線程，吞吐率隨著線程數(shù)增加而緩慢降低。

與基于 mutex 的 Ringbuffer 相比，非阻塞 Ringbuffer 的性能要遠(yuǎn)好于基于 mutex 的 Ringbuffer 的性能，并且在使用 16-20 個(gè)線程時(shí)達(dá)到最高。

另一方面，與緩存行對齊的無鎖 Ringbuffer 實(shí)現(xiàn)相比，非阻塞 Ringbuffer 的性能差強(qiáng)人意。這是因?yàn)榉亲枞?Ringbuffer 引入了更多的指針競爭。阻塞的無鎖 Ringbuffer 實(shí)現(xiàn)的 Enqueue 和 Dequeue 中只關(guān)注一個(gè)共享指針。但是非阻塞 Ringbuffer 實(shí)現(xiàn)的 Enqueue 和 Dequeue 則需要關(guān)注三個(gè)共享指針。這就導(dǎo)致出現(xiàn)了更多的共享變量的競爭，從而引起了性能下降。

取 2 個(gè)線程、16 個(gè)線程和 32 個(gè)線程三個(gè)位置進(jìn)行定量比較，得到下表。表中第一行是吞吐率數(shù)據(jù)（單元 ops/s），第二行是相對于基準(zhǔn)（基于 mutex 的 Ringbuffer 實(shí)現(xiàn)）的加速比。

相對于基于 mutex 的 Ringbuffer，非阻塞的無鎖 Ringbuffer 可以取得超過 10 倍的吞吐率提升。這表示非阻塞的無鎖 Ringbuffer 實(shí)現(xiàn)仍然是非常有效的，值得在實(shí)際應(yīng)用中嘗試。

與 program64 的性能比較

在文系列完整的開篇處已經(jīng)聲明，本系列文章介紹的所有優(yōu)化來自于 Ola 的研究。Ola 開發(fā)的 program64 庫也是示例的 Ringbuffer 實(shí)現(xiàn)的基礎(chǔ)。

無鎖 Ringbuffer 實(shí)現(xiàn)對應(yīng)于 program64 庫中的 p64_blkring 實(shí)現(xiàn)。

非阻塞 Ringbuffer 實(shí)現(xiàn)對應(yīng)于 program64 庫中的 p64_buckring 實(shí)現(xiàn)。

這里將 program64 庫的實(shí)現(xiàn)集成到我們的測試程序中，使用相同的測試環(huán)境和配置進(jìn)行了測試。align 和 p64_blkring 是一個(gè)對照組；buck 和 p64_buckring 是一個(gè)對照組。

圖 21 在 Arm 平臺(tái)上與 Program64 庫對比性能

我們的示例代碼的性能與 program64 的性能在趨勢上是一致，并且在數(shù)值范圍上基本重合。兩者之間的差距則是由于實(shí)現(xiàn)的細(xì)節(jié)引入的。比如 program64 的 Enqueue 和 Dequeue 可以操作多個(gè)單元，而示例代碼只能操作一個(gè)單元。這些差距并不影響關(guān)于 Ringbuffer 優(yōu)化的結(jié)論。

結(jié)論

本文介紹了多種 Ringbuffer 實(shí)現(xiàn)，并且比較和分析了不同 Ringbuffer 實(shí)現(xiàn)的性能表現(xiàn)：

基于 mutex 的 Ringbuffer 性能主要受到多線程相互競爭 mutex 的影響。參與競爭的線程越多，Ringbuffer 的吞吐率越低，非常不利于系統(tǒng)規(guī)模的擴(kuò)展。

阻塞式的無鎖 Ringbuffer 能夠從原子操作和緩存行對齊兩個(gè)優(yōu)化中獲得非常明顯的收益，使得實(shí)現(xiàn)能夠充分利用系統(tǒng)中互聯(lián)系統(tǒng)提供的通信容量，是一種非常有利于系統(tǒng)規(guī)模擴(kuò)展的 Ringbuffer 實(shí)現(xiàn)。

非阻塞 Ringbuffer 實(shí)現(xiàn)豐富并完善了 Ringbuffer 的功能，并且同樣可以取得很好的性能優(yōu)化。但是由于引入了更多的共享變量，非阻塞 Ringbuffer 實(shí)現(xiàn)的性能要弱于緩存行對齊的無鎖 Ringbuffer 實(shí)現(xiàn)。