俊昌, ,

(1.南京郵電大學 計算機學院,南京 210023; 2.江蘇省大數(shù)據(jù)安全與智能處理重點實驗室,南京 210023)

0 概述

近年來,隨著多核處理器的廣泛應用,高速應用程序的并行化飛速發(fā)展[1-4]。其中一種可行的并行化方法為流水線并行化,即多核處理器中的一個處理器核心完成較大任務中的第一階段子任務,然后將該任務傳遞給另一個核心,該核心負責完成下一階段的子任務處理。

由于現(xiàn)代商用多核處理器中普遍缺乏硬件支持的核間通信機制[5],因此當前研究大多采用基于軟件實現(xiàn)的無鎖隊列[1-2,6-7]作為流水線的核間通信機制。然而,現(xiàn)有無鎖隊列算法在設計時均假設數(shù)據(jù)的到達速率處于穩(wěn)定狀態(tài),如果輸入數(shù)據(jù)到達速率波動較大,隊列將頻繁地趨于空或趨于滿,從而導致無鎖隊列性能急劇下降。因此,在真實應用場景中,采用流水線并行技術的軟件性能往往急劇下降。

針對數(shù)據(jù)到達速率不穩(wěn)定的實際應用場景,本文提出一種新的無鎖隊列TinyQueue。當數(shù)據(jù)到達速率較低時,自適應地減小隊列大小,以減少內存占用;而當數(shù)據(jù)到達速率較高時,則自適應地增加隊列大小,以避免數(shù)據(jù)丟失。

1 相關工作

無鎖隊列被廣泛地應用于多線程程序中,并且也有大量關于并發(fā)無鎖隊列的研究[8-11],其中大部分研究都集中在多生產(chǎn)者多消費者隊列。然而,由于需要避免ABA問題,因此這些無鎖隊列性能都不甚理想。

作為一種特殊情況,單生產(chǎn)者單消費者隊列是實現(xiàn)高速核間通信很好的解決方案[1-2,6-7,12],因為它從根本上避免了ABA問題。此外,這些隊列的實現(xiàn)通常采用批處理來利用CPU高速緩存提高隊列的吞吐量。Lynx[13]是一個SPSC隊列,其降低了入隊操作和出隊操作的檢查開銷。但Lynx的主要設計目標是增加隊列的吞吐量,相比之下,TinyQueue更適用于實際應用場景中數(shù)據(jù)到達速率波動較大的情況。

另一種通信的方法是基于硬件隊列[14-15],它們從硬件層面上提供入隊操作和出隊操作的指令,以減少基于軟件隊列的開銷。然而,因為硬件隊列的狀態(tài)必須通過上下文切換來保留,所以需要對處理器和操作系統(tǒng)進行修改。目前它們也主要存在于仿真器中,對于商用處理器而言還不支持這種功能。

2 TinyQueue基礎算法

TinyQueue基于以下基礎算法(算法1和算法2),其中整個隊列是一個無限數(shù)組,使用head和tail來標識包含有效數(shù)據(jù)的部分。最初,每個單元格data[i]為空,初始值為ELEMENT_ZERO。

算法1基礎算法入隊操作

變量:value (將被插入隊列的數(shù)據(jù))

1.if(data[head]!= ELEMENT_ZERO ) {

2.return BUFFER_ FULL;

3.}

4.data[head]= value;

5.head ++;

6.if(head > QUEUE_MAX_SIZE ) {

7.head = 0;

8.}

9.return SUCCESS

算法2基礎算法出隊操作

1.if(data[tail] == ELEMENT_ZERO ) {

2.return BUFFER_EMPTY;

3.}

4.temp = data[tail];

5.data[tail] = ELEMENT_ZERO;

6.tail ++;

7.if(tail > QUEUE_MAX_SIZE) {

8.tail = 0;

9.}

10.return temp;

算法1是入隊操作的偽代碼。入隊操作首先讀取head指向的單元格的值(第1行)來檢查隊列是否已滿。如果該值等于ELEMENT_ZERO,即該單元格不包含有用的數(shù)據(jù),則可以插入數(shù)據(jù)。此時若該單元格已經(jīng)準備好接收數(shù)據(jù),入隊線程會將數(shù)據(jù)插入到該單元格中(第4行)且head遞增。若head超出了隊列長度,則將head置為0(第6行～第8行)。

算法2是出隊操作的偽代碼。出隊操作首先讀取tail指向的單元格的值并與ELEMENT_ZERO進行比較,檢查隊列是否為空(第1行)。如果tail指向的單元格的值等于ELEMENT_ZERO,表示數(shù)據(jù)不能出隊,返回BUFFER_EMPTY。否則,出隊操作讀取tail指向的單元格的值(第4行),然后將ELEMENT_ZERO寫入該單元格中,表示數(shù)據(jù)成功出隊(第5行)且tail遞增,如果tail的大小超出了隊列長度,則將tail置為0(第7行～第9行)。

值得注意的是,該基礎算法僅適用于單生產(chǎn)者單消費者循環(huán)隊列,所以,即使在沒有Compare-and-Swap(CAS)原語保護的情況下,將數(shù)據(jù)插入head指向的單元格中也是安全的。

3 TinyQueue核心算法

TinyQueue基于算法1和算法2,其設計思想如圖1所示。當數(shù)據(jù)到達速率較低時,隊列趨于空,如果此時生產(chǎn)者位于消費者后面且兩者都不會訪問隊列的后半部分,TinyQueue會自適應地將隊列大小減半,以減少其內存占用,提高緩存命中率;當數(shù)據(jù)到達速率較高時,隊列趨于滿,如果此時生產(chǎn)者位于消費者前面,TinyQueue會自適應地將隊列大小增加一倍,以避免數(shù)據(jù)丟失。

圖1 TinyQueue算法原理

3.1 數(shù)據(jù)結構和全局變量

算法3是TinyQueue的主要數(shù)據(jù)結構。每個隊列都會創(chuàng)建一個TinyQueue實例,其中:data是指向預分配的循環(huán)數(shù)組的指針;traffic是一個有符號整數(shù),用于計算隊列full/empty單元格數(shù),每當入隊操作因為隊列滿失敗時,traffic減1,每當出隊操作因隊列空失敗時,traffic加1。因此,traffic的大小可以反映數(shù)據(jù)到達頻率的波動情況,即可以用它來決定是否需要增加或減小隊列的大小;變量size用來表示隊列的大小,TinyQueue會預先分配一個MAX_SIZE大小的數(shù)組,其中MAX_SIZE等于100 000,但TinyQueue會根據(jù)traffic自動選擇預分配數(shù)組的第一部分;變量head和tail分別指向入隊線程將要插入數(shù)據(jù)的單元格和出隊線程將要提取數(shù)據(jù)的單元格。值得注意的是,要減小隊列大小時,變量head和size必須同時更新,所以將其封裝在結構體cas_info中。由于每個變量長度都是32位,因此cas_info是一個64位的字,在64位對齊的服務器上,其更新操作是原子的。

算法3TinyQueue數(shù)據(jù)結構

1.struct cas info {

2.head:Indicator of enqueue (32-bits integer);

3.size:Size of the cyclic array (32-bits integer);

4.};

5.struct tinyqueue {

6.info:instance of struct cas_info;

7.//緩存行(cache line)對齊

8.tail:Indicator of dequeue (32-bits integer);

9.//緩存行(cache line)對齊

10.traffic:Statistical counter of empty and full cases;

11.//緩存行(cache line)對齊

12.data:pointer to cyclic array;

13.};

3.2 TinyQueue入隊操作

TinyQueue入隊操作(算法4)基于算法1。入隊操作首先檢查隊列是否已滿(第3行)。如果隊列已滿,則traffic加1(第4行)并返回BUFFER_FULL;否則,入隊操作將head賦值給局部變量temp(第7行)且head遞增。如果head大于或等于size(第9行),表示head指向TinyQueue循環(huán)數(shù)組的尾部,入隊操作將比較traffic與預定義閾值ENLARGE_THRESHOLD的大小來判斷是否需要增加隊列的大小(第10行)。若traffic較大,入隊操作將隊列大小增加1倍即變量size乘以2(第11行),然后將traffic置為0;若traffic較小,則說明不需要調整隊列大小,入隊操作將head置為0,使其指向隊列的第一個單元格(第15行),然后保存數(shù)據(jù)并返回SUCCESS。

算法4TinyQueue入隊操作

變量:value (將被插入隊列的數(shù)據(jù))

1.//局部變量

2.temp:local variable to buffer value of info.head

3.if(data[head] != ELEMENT_ZERO ) {

4.traffic++;

5.return BUFFER_FULL;

6.}

7.temp = info.head;

8.info.head ++;

9.if(info.head > info.size ) {

10.if( traffic > ENLARGE_THRESHOLD ) {

11.info.size = info.size * 2;

12.traffic = 0;

13.}

14.else {

15.info.head = 0;

16.}

17.}

18.data[temp] = value;

19.return SUCCESS;

3.3 TinyQueue出隊操作

TinyQueue出隊操作(算法5)基于算法2實現(xiàn)。出隊操作首先檢查隊列是否為空(第5行)。如果隊列為空,traffic減1并返回EMPTY(第6行);如果隊列不為空,出隊操作將tail賦值給局部變量temp(第9行),tail遞增(第10行)。若tail等于size(第11行),則說明tail指向TinyQueue循環(huán)數(shù)組的尾部,出隊操作將比較traffic與預定義閾值SHRINK_THRESHOLD的大小來判斷是否需要減小隊列大小(第14行)。SHRINK_THRESHOLD是一個負整數(shù)(例如-50),用來表示隊列EMPTY的情況。若traffic較小,則出隊操作將局部變量size除以2(第19行)使隊列大小減半,然后更新全局共享變量size(第25行),強制入隊操作和出隊操作使用新的隊列大小,然而,減小隊列大小仍然存在一些易錯點;若traffic的值較大,則說明不需要調整隊列的大小,出隊操作將tail置為0,使其指向隊列的第一個單元格。最后,出隊操作從單元格中提取數(shù)據(jù)(第37行),將ELEMENT_ZERO寫入單元格中(第38行)并返回。

算法5TinyQueue出隊操作

1.//局部變量

2.temp:local variable to buffer value of tail

3.temp_value:local variable to buffer value of data pointed by tail

4.info_tmp,info_tmp_new:local instances of structure info

5.if(data[tail] == ELEMENT ZERO ) {

6.traffic--;

7.return BUFFER_EMPTY;

8.}

9.temp = tail;

10.tail++;

11.if(tail >= info.size ) {

12.count = 0;

13.tag1;

14.if(traffic <= SHRINK_THRESHOLD) {

15.info_tmp = info_tmp_new = info;//原子的讀取

16.if(count LOOP THRESHOLD ) {

17.goto 34;

18.}

19.if(info_tmp.head <= info_tmp.size / 2 ) {

20.info_tmp_new.size /= 2;

21.}

22.else {

23.goto 34;

24.}

25.if(CAS(&info,info_tmp,info_tmp_new)) {

26.traffic = 0;

27.goto 34;

28.}

29.else {

30.count++;

31.goto 13;

32.}

33.}

34.tag2;

35.tail = 0;

36.}

37.temp_value = data[temp];

38.data[temp] = ELEMENT_ZERO;

39.return temp_value;

3.4 TinyQueue增加隊列大小操作

由算法4和算法5可知,入隊操作和出隊操作每次都需要先讀取size的值來檢查它們是否到達TinyQueue循環(huán)數(shù)組的尾部。因此,TinyQueue增加隊列大小的基本思路是:如果size乘以2,那么入隊操作和出隊操作可以使用的循環(huán)數(shù)組的大小也會增加1倍。但是在算法設計中應保證其正確性和無鎖性。圖2展示了增加隊列大小時可能發(fā)生的2種情況:

1)生產(chǎn)者在消費者后面(圖2標為不安全)。在這種情況下,通過改變size的值來改變隊列的大小是不安全的。因為消費者線程會讀取新的size,然后使用未插入數(shù)據(jù)的單元格,進而產(chǎn)生錯誤。

2)生產(chǎn)者在消費者前面(圖2標為安全)。在這種情況下,當生產(chǎn)者到達當前循環(huán)數(shù)組的尾部時,生產(chǎn)者通過改變size來增加隊列大小并使用新分配的單元格是安全的。因為在生產(chǎn)者向新的單元格插入數(shù)據(jù)前,消費者不會超過生產(chǎn)者使用新分配的單元格。針對此種情況的具體實現(xiàn)在算法4中已給出(第9行～第17行)。

圖2 增加隊列大小的情況

3.5 TinyQueue減小隊列大小操作

由于TinyQueue是一個高速無鎖隊列,因此很難通過減小size來減小隊列大小。圖3展示了收縮隊列時可能遇到的2種情況:

1)生產(chǎn)者位于消費者前面(圖3標為不安全)。在這種情況下減小隊列大小是不安全的。因為生產(chǎn)者可能會將數(shù)據(jù)插入到TinyQueue不會使用的隊列后半部分。

2)生產(chǎn)者位于消費者之后(圖3標為安全)。在這種情況下,當消費者到達隊列尾部時,如果生產(chǎn)者位于隊列前半部分,消費者通過改變size來減小隊列大小是安全的,但減小size和檢查生產(chǎn)者是否在隊列前半部分必須是同時的原子操作。為了實現(xiàn)這一點,本文將變量size和head封裝為一個64位結構體(算法3中的cas_info),其可以通過CAS指令原子的更新。因此,本文做了一個符合現(xiàn)實的假設,即TinyQueue隊列大小最大不超過232。但當消費者嘗試減小隊列大小時,若生產(chǎn)者一直位于隊列前半部分,會使消費者一直不斷嘗試減小隊列大小。為了避免這種潛在的活鎖問題,本文設置了一個閾值LOOP_THREASHOLD,當消費者嘗試次數(shù)大于LOOP_THREASHOLD時,消費者就會放棄。

圖3 減小隊列大小的情況

在算法5中,出隊操作首先讀取結構體info(第15行),然后檢查其是否已經(jīng)多次嘗試減小隊列大小。如果嘗試次數(shù)已經(jīng)大于LOOP_THREASHOLD,那么將會放棄出隊(第16行)以避免活鎖問題。否則,檢查head是否小于隊列大小的一半(第19行)來判斷生產(chǎn)者是否位于隊列前半部分。如果生產(chǎn)者位于隊列前半部分,入隊操作引入新的變量info_tmp_new,并將info_tmp_new.size的值減半(第20行)。否則,減小隊列大小和出隊操作都是不安全的。如果入隊操作可以減小隊列大小,那么它將嘗試使用CAS指令來更新size(第25行),從而保證以下2個操作都是原子的:1)確保生產(chǎn)者沒有移動到將要移除的數(shù)組的下半部分;2)size減半。如果CAS指令成功,則出列操作將traffic置為0并返回(第26行)。否則,count遞增,并返回到開始位置再次嘗試(第30行)。

4 性能評估

本節(jié)將會對TinyQueue的性能與現(xiàn)有的最佳解決方案FastForward、MCRingBuffer和B-Queue的性能進行對比。

4.1 實驗設置

在Dell R730服務器上進行性能評估。該服務器配有2個Intel Xeon E5-2609處理器,每個處理器包含8個CPU內核,最高頻率可達1.7 GHz。每個CPU內核有32 KB的一級數(shù)據(jù)緩存,32 KB的一級指令緩存和256 KB的二級緩存。同一處理器上的8個CPU內核共享20 MB的三級緩存。2個處理器間通過2條6.4 GT/s的QPI鏈路相連。該服務器使用128 GB DDR4內存,運行64位的Ubuntu 16.04,其內核版本為Linux kernel version 4.4.0。TinyQueue使用GCC 5.4.0編譯。

實驗中,生產(chǎn)者將1013條數(shù)據(jù)插入到隊列中,消費者將按順序讀取這些數(shù)據(jù)。每個線程都將綁定到一個專用的CPU上,并且生產(chǎn)者線程和消費者線程分別運行在不同的CPU上。為了獲取入隊操作和出隊操作的執(zhí)行時間,使用X86中基于硬件的時間戳計數(shù)器來記錄RDTSC寄存器執(zhí)行1013次入隊操作和出隊操作所需的CPU周期。單個入隊或出隊操作的時間可通過總CPU周期數(shù)除以1013減去工作負載時間求得。值得注意的是,讀取RDTSC寄存器并不是一個序列化指令,這意味著總CPU周期數(shù)會存在一些偏差。盡管如此,因為總CPU周期數(shù)需要除以1013,所以筆者認為單個入隊操作或出隊操作的時間偏差可忽略不計。對于其他的統(tǒng)計信息(如緩存未命中率),本文使用Linux Perf 獲得。每條數(shù)據(jù)通過30次實驗獲得。

4.2 理想平臺性能測試

在理想測試平臺上對4種無鎖隊列進行了性能測試。在實驗中,生產(chǎn)者線程執(zhí)行while循環(huán),并且在每次迭代中都會盡可能快地將數(shù)據(jù)插入隊列中。當隊列滿時,生產(chǎn)者線程將休眠1 μs避免和消費者線程沖突。隊列大小設置為2 048。

表1列出了在同一CPU(2個線程運行在同一CPU上)和跨CPU(2個線程運行在不同的CPU上)每個操作所需的CPU周期。實驗中使用Perf分析無鎖隊列的緩存行為(選擇一級數(shù)據(jù)緩存命中率進行分析)。

表1 理想測試平臺上的最高性能

表1的數(shù)據(jù)是4種隊列在理想測試臺上達到的最佳性能,每個入隊或出隊操作最少只需要十幾個CPU周期。為了進一步研究不同的無鎖隊列在理想測試平臺上的性能,本文統(tǒng)計了隊列的緩存行為。一級緩存缺失率一列數(shù)據(jù)表明4種無鎖隊列都具有較高的一級緩存命中率。造成這種情況的原因在于在理想測試平臺上,無鎖隊列工作集很小,幾乎所有的內存訪問都可以通過一級緩存進行。

然而,在真正的應用程序中,無鎖隊列的工作集通常會很大,數(shù)據(jù)傳入速率也會大幅波動。在這種情況下,無鎖隊列的工作集就不能全都保存在一級緩存中,從而會導致容量緩存未命中和一致性緩存未命中。例如,對于本文使用的E5-2609處理器,一次一級緩存命中需要2個CPU周期,一次二級緩存命中需要10個CPU周期,一次三級緩存命中需要40個CPU周期,一次存儲區(qū)訪問則需要200個CPU周期。顯然,單獨的三級緩存命中會妨礙無鎖隊列實現(xiàn)最高性能。也就是說,在理想測試平臺上性能優(yōu)異的無鎖隊列在實際應用中的性能可能會有所下降。這一點筆者將在下文中進行論述。

4.3 真實平臺性能測試

為了評估TinyQueue在實際應用中處理傳入數(shù)據(jù)突發(fā)性的能力,在真實測試平臺上對其進行了性能測試。在該測試平臺中,用戶每次從隊列中檢索一條數(shù)據(jù)并等待85個CPU周期(一條10 Gb/s的鏈路處理一個數(shù)據(jù)包的時間)來模擬工作負載。為了模擬實際應用中的突發(fā)事件,生產(chǎn)者線程在每次迭代中都會執(zhí)行以下操作:1)休眠一段時間(處理單條數(shù)據(jù)的時間乘以突發(fā)數(shù)據(jù)的大小)以模擬批處理過程中網(wǎng)絡設備中斷;2)盡可能快地插入數(shù)據(jù)。

圖4顯示了入隊或出隊操作所需的平均CPU周期數(shù)?？梢钥闯?當突發(fā)數(shù)據(jù)較小時(例如小于2 048),不同的無鎖隊列的性能是相似的。但是隨著數(shù)據(jù)大小的增加,FastForward、MCRingBuffer和BQueue的性能會急劇下降,但TinyQueue的性能幾乎不變。

圖4 真實平臺上不同無鎖隊列的性能表現(xiàn)

為探究TinyQueue在真實測試平臺中表現(xiàn)更好的原因,將突發(fā)數(shù)據(jù)的大小設置為4 096。表2列出了不同無鎖隊列的性能。由同一CPU列和跨CPU列的數(shù)據(jù)可知,TinyQueue的性能明顯優(yōu)于其他3種隊列。其中最小長度列和最大長度列的數(shù)據(jù)分別表示系統(tǒng)運行時隊列的最小長度和最大長度(這里只列出了TinyQueue的隊列長度,因為其他隊列不能隨時改變它們的隊列大小)。由實驗數(shù)據(jù)可知,當數(shù)據(jù)到達速率較高時,TinyQueue可以將隊列增加到65 536個單元,以處理數(shù)據(jù)的突發(fā)性,而當數(shù)據(jù)到達速率較低時,TinyQueue也可以將隊列減小到256個單元。真實平臺上的性能測試結果表明,由于TinyQueue可以自適應地調整隊列大小,因此其具有最高的一級緩存命中率(99.20%)和最低的三級緩存未命中率(0.07%)。通過減小隊列大小,TinyQueue可以大大減小緩存缺失率,從而提高隊列的整體性能。

表2 真實測試平臺中各無鎖隊列的性能

值得注意的是,在該真實測試平臺上,所有隊列的性能都比在文獻[1,6]中所采用的理想測試平臺中的性能要差(在理想測試平臺中,一次入隊或出隊操作通常需要10 ns),其原因在于在該真實測試平臺中,本文更真實地還原了實際應用中數(shù)據(jù)的接收處理場景。正如文獻[1]指出的,在這樣的測試平臺上測試一個隊列的性能比在理想測試平臺上測試更有價值。

5 結束語

本文提出一種適用于多核處理器中流水線并行化的高效無鎖隊列TinyQueue。當數(shù)據(jù)到達速率較低時,TinyQueue可以自適應地減小隊列大小,保持較小的內存占用,提高緩存命中率;當數(shù)據(jù)速率較高時,TinyQueue可以自適應地增加隊列大小,以避免數(shù)據(jù)丟失。因此,即使在輸入數(shù)據(jù)頻率波動較大的實際應用中,TinyQueue仍能在較少的CPU周期內完成一次入隊或出隊操作,保持較好的性能。由于TinyQueue的性能可以通過實時系統(tǒng)調度器進一步優(yōu)化,因此下一步將對此進行研究。