多核分布式隊(duì)列的實(shí)現(xiàn)：“偷”與“自私”的運(yùn)用
在討論本文的正題前，不得不先說(shuō)一些閑話，嫌哆嗦者可以跳過(guò)“前言”部分不讀。
1.前言
在發(fā)表了“老子是偉大的多核計(jì)算科學(xué)家” （鏈接：http://blog.csdn.net/drzhouweiming/archive/2008/11/07/3246254.aspx，為敘述方便，后面將這篇文章簡(jiǎn)稱為“老子”）一文后，褒揚(yáng)者有許多，但是也引來(lái)了許多板磚。當(dāng)然大部分板磚都只是泛泛的批評(píng)，沒(méi)有任何內(nèi)容。不過(guò)有些人覺(jué)得似乎有些牽強(qiáng)附會(huì)，倒是引起了我的注意，確實(shí)這類文章可能確實(shí)容易給人牽強(qiáng)附會(huì)的感覺(jué)。
需要說(shuō)明的是，本人并沒(méi)有覺(jué)得它是牽強(qiáng)附會(huì)的。首先申明一下，我并不是研究哲學(xué)的，也沒(méi)有詳細(xì)研究過(guò)老子的《道德經(jīng)》，但是我在設(shè)計(jì)多核算法時(shí)，確實(shí)受到了《道德經(jīng)》中的思想啟發(fā)。舉兩個(gè)例子如下：
第一個(gè)例子是在設(shè)計(jì)多核查找算法（鏈接：http://blog.csdn.net/drzhouweiming/archive/2008/10/27/3159501.aspx）時(shí)，最初我是用AVL樹(shù)作為多級(jí)查找結(jié)構(gòu)的子查找結(jié)構(gòu)的，當(dāng)時(shí)覺(jué)得AVL樹(shù)肯定會(huì)比數(shù)組更好，因?yàn)閷?duì)稍微大一點(diǎn)的數(shù)組進(jìn)行插入刪除的效率非常低，只能用在很少數(shù)據(jù)的表上，不能對(duì)大量數(shù)據(jù)的表進(jìn)行管理。記得有一天看電視時(shí)，湊巧看到在講老子的小國(guó)寡民思想，談到了結(jié)繩而治的問(wèn)題，受此啟發(fā)，對(duì)AVL樹(shù)比數(shù)組更好的想法產(chǎn)生了懷疑，于是試著將查找子結(jié)構(gòu)改為用最原始的數(shù)組來(lái)實(shí)現(xiàn)，結(jié)果發(fā)現(xiàn)即使對(duì)上百萬(wàn)個(gè)規(guī)模的數(shù)據(jù)的表進(jìn)行處理，綜合性能也比用AVL樹(shù)更好。
第 二個(gè)例子是在設(shè)計(jì)多核分布式內(nèi)存管理算法時(shí)，采用了“搶”的方法，使得分配和釋放內(nèi)存不需要使用鎖。這也是受《道德經(jīng)》中的“無(wú)為”及“大道自然”的思想 影響，因?yàn)橹耙呀?jīng)發(fā)現(xiàn)“貪心”、“自私”、“偷”這幾種人性的本能在算法中得到廣泛使用，既然連“偷”都在多核算法中得到使用，那么它的孿生兄弟“搶” 應(yīng)該也可以在多核算法中得到使用，本著此思想，后來(lái)終于發(fā)現(xiàn)可以將“搶”的思想用在多核分布式內(nèi)存管理算法中，大大提高共享內(nèi)存分配和釋放的效率。
對(duì)老子《道德經(jīng)》的解釋，歷來(lái)有各種不同的解釋。既然有些人只是在理論層面都可以進(jìn)行解釋，我現(xiàn)在把它的部分思想用到了具體的多核算法中，變成了在計(jì)算機(jī)里可以實(shí)際運(yùn)行的程序，對(duì)它解釋一下就變成了牽強(qiáng)附會(huì)的話，那么這種牽強(qiáng)附會(huì)我想越多越好。
閑話少敘，言歸正傳，下面就來(lái)談一個(gè)使用“偷”與“自私”的方法實(shí)現(xiàn)的多核分布式隊(duì)列的詳細(xì)實(shí)例，以看看如何將看似泛泛而談的思想變成可以運(yùn)行的程序的。
2.分布式隊(duì)列的基本概念
在“多核編程中的條件同步模式”（鏈接： http://softwareblogs-zho.intel.com/2009/01/14/845/）這篇文章中，講到了如何減少共享隊(duì)列中的鎖的使用次數(shù)的具體方法，在它的基礎(chǔ)上，可以構(gòu)造出一個(gè)高效的隊(duì)列池。
如果采用線程分組競(jìng)爭(zhēng)模式（參見(jiàn)“多核編程中的線程分組競(jìng)爭(zhēng)模式，鏈接：http://blog.csdn.net/drzhouweiming/archive/2007/07/10/1684753.aspx）來(lái)實(shí)現(xiàn)隊(duì)列池，那么每組線程對(duì)應(yīng)于隊(duì)列池中的一個(gè)子隊(duì)列，當(dāng)某個(gè)線程在操作自己所屬的子隊(duì)列時(shí)，如果子隊(duì)列為空卻進(jìn)行出隊(duì)操作，那么此時(shí)可以從其他組線程所屬的子隊(duì)列中進(jìn)行出隊(duì)操作，這也就是“老子”一文中所說(shuō)的“偷”的方法的使用。
有沒(méi)有更好的方法進(jìn)一步減少同步或者鎖的使用呢？答案是有的。偷別人的東西總不如掏自己口袋里的東西來(lái)得方便，之所以需要“偷”，乃是因?yàn)樽约嚎诖锟湛?。如果大家都富裕了，口袋都鼓鼓的了，自然不需要?ldquo;偷”別人的了。
當(dāng)然在計(jì)算機(jī)中，“富裕”的辦法就是給每個(gè)線程賦予一個(gè)私有隊(duì)列，這樣每個(gè)線程可以大部分時(shí)間都操作自己私有隊(duì)列，不需要同步操作，大大提高效率，這也就是“老子”一文中所說(shuō)的“自私”方法的使用。
基于“偷”和“自私”兩種方法，就可以設(shè)計(jì)出一個(gè)適應(yīng)多核環(huán)境的分布式隊(duì)列。在分布式隊(duì)列中，每個(gè)操作隊(duì)列的線程都有一個(gè)私有隊(duì)列，另外為了解決私有隊(duì)列間的負(fù)載均衡問(wèn)題，還需要一個(gè)隊(duì)列池來(lái)維護(hù)數(shù)據(jù)的負(fù)載均衡。
分布式隊(duì)列的數(shù)據(jù)結(jié)構(gòu)示意圖如下：
圖1：分布式隊(duì)列數(shù)據(jù)結(jié)構(gòu)示意圖
有了上面的數(shù)據(jù)結(jié)構(gòu)圖，具體來(lái)實(shí)現(xiàn)就可以分為兩個(gè)步驟：
1、 實(shí)現(xiàn)一個(gè)隊(duì)列池
2、 給每個(gè)線程賦予一個(gè)私有隊(duì)列
隊(duì)列池的實(shí)現(xiàn)可以采用前面講過(guò)方法實(shí)現(xiàn)，這里就不詳述了，下面主要談?wù)勅绾谓o每個(gè)線程賦予一個(gè)私有隊(duì)列（也稱作本地化隊(duì)列）的詳細(xì)實(shí)現(xiàn)方法。
3.本地化隊(duì)列的實(shí)現(xiàn)思路
要給線程指定一個(gè)本地化隊(duì)列，通常的做法是先將創(chuàng)建好的隊(duì)列放入一個(gè)數(shù)組中，然后給線程編號(hào)，從0開(kāi)始進(jìn)行編號(hào)，編號(hào)為0的線程對(duì)應(yīng)于數(shù)組下標(biāo)為0位置上存放的隊(duì)列，編號(hào)為1的線程對(duì)應(yīng)于數(shù)組下標(biāo)為1位置上存放的隊(duì)列，…。
每個(gè)線程要獲取自己的本地化隊(duì)列時(shí)，只需要先獲取線程編號(hào)，然后就可以通過(guò)線程編號(hào)去訪問(wèn)對(duì)應(yīng)的隊(duì)列，由于每個(gè)線程的編號(hào)都不相同，因此每個(gè)線程訪問(wèn)的隊(duì)列都不相同，即每個(gè)隊(duì)列只有一個(gè)線程訪問(wèn)它，這樣就可以實(shí)現(xiàn)每個(gè)線程的本地化隊(duì)列。
那么如何給線程編號(hào)從0開(kāi)始編號(hào)呢，操作系統(tǒng)并沒(méi)有直接提供這種功能。即使操作系統(tǒng)提供了線程從0開(kāi)始編號(hào)的功能也沒(méi)有用，因?yàn)椴⒉灰欢ㄋ械木€程都會(huì)訪問(wèn)分布式隊(duì)列。例如有8個(gè)線程，其中編號(hào)為0，3，5，7的線程會(huì)訪問(wèn)分布式隊(duì)列，那么在創(chuàng)建分布式隊(duì)列時(shí)，就需要?jiǎng)?chuàng)建8個(gè)本地隊(duì)列，否則線程編號(hào)將無(wú)法和存放隊(duì)列的數(shù)組下標(biāo)對(duì)應(yīng)起來(lái)。
看到這里，目標(biāo)已經(jīng)很明確了，那就是要給所有訪問(wèn)分布式隊(duì)列的線程從0開(kāi)始依次編號(hào)。比如有N個(gè)線程要訪問(wèn)分布式隊(duì)列，那么需要給這N個(gè)線程依次編號(hào)為0，1，…N-1。下面就來(lái)討論如何給線程編號(hào)的問(wèn)題。
#p#
4.給線程編號(hào)的方法
在操作系統(tǒng)中，通常提供了線程本地存儲(chǔ)的API，通過(guò)API可以給每個(gè)線程設(shè)定一個(gè)數(shù)據(jù)（可以是指針，也可以是一個(gè)整數(shù)），同時(shí)也可以通過(guò)API來(lái)取出當(dāng)前線程設(shè)置的那個(gè)數(shù)據(jù)。比如給一個(gè)線程A設(shè)定一個(gè)整數(shù)0，那么線程A執(zhí)行的任何地方都可以調(diào)用相應(yīng)的API獲取到整數(shù)0，這樣就可以在程序的任何地獲取到線程A的編號(hào)為0。
在Windows系列操作系統(tǒng)中，提供了Tls_Alloc()，Tls_SetValue()，Tls_GetValue()，Tls_Free()這幾個(gè)函數(shù)來(lái)實(shí)現(xiàn)線程本地存儲(chǔ)操作。
pthread中，可以通過(guò)pthread_key_create(), pthread_setspecific(), pthread_getspecific()等函數(shù)來(lái)實(shí)現(xiàn)線程本地存儲(chǔ)操作，其中pthread_create_key()和Tls_Alloc()功能相同，只是參數(shù)有所不同，Tls_SetValue()和pthread_setspecific()功能等價(jià)，Tls_GetValue()和pthread_getspecific()功能等價(jià)。
下面演示一下TlsAlloc()，Tls_SetValue()，Tls_GetValue()，Tls_Free()這幾個(gè)函數(shù)的基本用法。
- DWORD g_dwTlsIndex;
- LONG volatile g_dwThreadId = 0;
- int GetId()
- {
- //獲取當(dāng)前執(zhí)行線程的由TlsSetValue()設(shè)置的值
- int nId = (int)TlsGetValue(g_dwTlsIndex);
- return (nId-1);
- }
- void ThreadFunc(void *args)
- {
- LONG Id = AtomicIncrement (&g_dwThreadId); //對(duì)g_dwThreadId進(jìn)行原子加1操作
- TlsSetValue(g_dwTlsIndex, (void *)Id); //給當(dāng)前執(zhí)行的線程設(shè)置一個(gè)值
- printf("ThreadFunc2: Thread Id = %ld/n", GetId());
- }
- int main(int argc, char* argv[])
- {
- g_dwTlsIndex = TlsAlloc(); //分配一個(gè)線程本地存儲(chǔ)索引，需要在創(chuàng)建線程前執(zhí)行
- _beginthread(ThreadFunc, 0, NULL);
- _beginthread(ThreadFunc, 0, NULL);
- Sleep(100); //延時(shí)等待上面兩個(gè)線程執(zhí)行完
- TlsFree(g_dwTlsIndex);
- return 0;
- }
要說(shuō)明一下，在ThreadFunc()函數(shù)中，使用了一個(gè)AtomicIncrement()函數(shù)，這個(gè)函數(shù)對(duì)應(yīng)于Windows操作系統(tǒng)中的InterlockedIncrement()函數(shù)。在Widnows系統(tǒng)中，可以使用以下宏定義來(lái)實(shí)現(xiàn)AtomicIncrement()函數(shù)：
#define AtomicIncrement(x) InterlockedIncrement(x)
上面程序在運(yùn)行后，會(huì)打印出以下結(jié)果：
ThreadFunc: Thread Id = 0
ThreadFunc: Thread Id = 1
從上面代碼和執(zhí)行結(jié)果可以看出，雖然GetValue()在ThreadFunc()函數(shù)中執(zhí)行，但是兩個(gè)線程執(zhí)行GetValue()得到的值是不同的，一個(gè)線程得到的是0，另外一個(gè)線程得到的是1。這主要是因?yàn)閮蓚€(gè)線程調(diào)用TlsSetValue()設(shè)置的值并不相同，一個(gè)為1，另一個(gè)為2。
需要注意的是，TlsGetValue()的返回值為0表示失敗，所以使用TlsSetValue()函數(shù)時(shí)，應(yīng)該從1開(kāi)始設(shè)置，然后在GetId()函數(shù)中，返回的是TlsGetValue()的返回值減1。
采用上面的方法，就可以設(shè)計(jì)出分布式隊(duì)列中的線程Id自動(dòng)編號(hào)和獲取功能了。下面是詳細(xì)的實(shí)現(xiàn)代碼：
- class CDistributedQueue {
- private:
- DWORD m_dwTlsIndex;
- LONG volatile m_lThreadIdIndex;
- public:
- CDistributedQueue();
- virtual ~CDistributedQueue();
- LONG ThreadIdGet();
- //可以添加其他成員函數(shù)在下面
- };
- CDistributedQueue::CDistributedQueue()
- {
- m_dwTlsIndex = TlsAlloc();
- m_lThreadIdIndex = 0;
- }
- CDistributedQueue::~CDistributedQueue()
- {
- TlsFree(m_dwTlsIndex);
- }
- LONG CDistributedQueue::ThreadIdGet()
- {
- LONG Id = (LONG )TlsGetValue(m_dwTlsIndex);
- if ( Id == 0 )
- {
- Id = AtomicIncrement(&m_lThreadIdIndex);
- TlsSetValue(Id);
- }
- return (Id - 1);
- }
上面的代碼中，設(shè)置或獲取線程編號(hào)都在ThreadIdGet()一個(gè)成員函數(shù)內(nèi)完成，先判斷獲取的Id是否為0，如果為0，表明線程還沒(méi)有被設(shè)置Id，因此將m_lThreadIdIndex原子加1，然后再設(shè)置給對(duì)應(yīng)的線程。每調(diào)用一次TlsSetValue()函數(shù)，其設(shè)置的Id值依次加1，這樣就可以得到一個(gè)1，2，3，…序列。每個(gè)線程調(diào)用了TlsSetValue()函數(shù)后，下一個(gè)調(diào)用TlsGetValue()函數(shù)時(shí)，獲得的值一定大于0，因此每個(gè)線程最多只能執(zhí)行TlsSetValue()函數(shù)一次。
采用上面的方法來(lái)獲取線程編號(hào)，必須保證創(chuàng)建的本地隊(duì)列數(shù)量大于等于訪問(wèn)隊(duì)列的線程數(shù)量，否則隊(duì)列數(shù)量不足，將會(huì)造成沒(méi)有足夠的本地隊(duì)列供線程使用，程序中可能會(huì)造成越界等不可預(yù)測(cè)的異常。常用的解決辦法是將本地隊(duì)列的數(shù)量擴(kuò)大一倍。
上面這種線程編號(hào)方法，非常方便，任何訪問(wèn)分布式隊(duì)列的線程都可以被自動(dòng)編號(hào)，調(diào)用分布式隊(duì)列的線程不需要為編號(hào)操心。
有了給線程自動(dòng)編號(hào)的方法后，就可以實(shí)現(xiàn)分布式隊(duì)列的各個(gè)具體操作如進(jìn)隊(duì)、出隊(duì)等。當(dāng)然在實(shí)現(xiàn)具體的操作代碼前，有必要了解一下分布式隊(duì)列中是如何進(jìn)行進(jìn)隊(duì)和出隊(duì)操作的。
#p#
5. 進(jìn)隊(duì)出隊(duì)操作
分布式隊(duì)列的進(jìn)隊(duì)出隊(duì)操作根據(jù)不同應(yīng)用類型具有不同的操作策略，但是不論何種類型的操作，其基本思想必須以本地隊(duì)列操作最大化作為前提條件。下面給出分布式隊(duì)列中常用的進(jìn)隊(duì)出隊(duì)操作類型。
1) 出隊(duì)操作
出隊(duì)操作比較簡(jiǎn)單，通常都是先從本地隊(duì)列中獲取數(shù)據(jù)，如果本地隊(duì)列為空，那么再?gòu)墓蚕黻?duì)列池中獲取數(shù)據(jù)。
由于先從本地隊(duì)列中獲取數(shù)據(jù)，因此有助于實(shí)現(xiàn)本地隊(duì)列操作的最大化。
出隊(duì)操作的流程圖如下：
圖2：分布式隊(duì)列的出隊(duì)操作流程圖
2:進(jìn)隊(duì)操作
進(jìn)隊(duì)操作相比于出隊(duì)操作要復(fù)雜一些，常用的操作策略有以下兩種：
策略1：先判斷本地隊(duì)列是否為空，如果為空則將數(shù)據(jù)放入本地隊(duì)列中；然后判斷共享隊(duì)列池中是否滿，如果滿則將數(shù)據(jù)放入本地隊(duì)列中，否則放入共享隊(duì)列中。
在這種策略的進(jìn)隊(duì)操作中，首先考慮的是本地隊(duì)列的操作問(wèn)題，本地隊(duì)列至少要有一個(gè)數(shù)據(jù)，然后考慮的問(wèn)題是負(fù)載平衡問(wèn)題，共享隊(duì)列池中的數(shù)據(jù)主要用于各線程間數(shù)據(jù)的負(fù)載均衡。共享隊(duì)列池的大小必須做出限制，否則數(shù)據(jù)全部都進(jìn)到共享隊(duì)列池中去了，本地隊(duì)列未得到有效使用。
共享隊(duì)列池到底設(shè)定多大，才能使得本地隊(duì)列操作最大化與負(fù)載平衡問(wèn)題之間取得一個(gè)好的均衡，是在實(shí)際情況中需要考慮的問(wèn)題，最好通過(guò)測(cè)試程序性能去獲取一個(gè)合適的值。
進(jìn)隊(duì)操作策略1的操作流程圖如下：
圖3：分布式隊(duì)列的進(jìn)隊(duì)操作策略1的流程圖
策略2：當(dāng)有多個(gè)數(shù)據(jù)需要進(jìn)隊(duì)時(shí)，先放入一些數(shù)據(jù)到本地隊(duì)列中，然后剩下的數(shù)據(jù)再放入共享隊(duì)列池中，如果隊(duì)列池滿的話，則仍然放入本地隊(duì)列中。
本策略中，通常是進(jìn)隊(duì)的線程馬上自己要從隊(duì)列中獲取數(shù)據(jù)，因此要先放入一些數(shù)據(jù)到自己的本地隊(duì)列中，保證下次從隊(duì)列中取數(shù)據(jù)一定是從本地隊(duì)列中獲取，可以大大提高本地化隊(duì)列操作的頻率，有效降低共享隊(duì)列池的操作，大大減少了同步操作。
進(jìn)隊(duì)操作策略2的操作流程圖如下：
圖4：分布式隊(duì)列的進(jìn)隊(duì)操作策略2的流程圖
有了進(jìn)隊(duì)操作和出隊(duì)操作的詳細(xì)流程后，實(shí)現(xiàn)分布式隊(duì)列的具體代碼就容易多了。
CDistributedQueue類的各個(gè)操作的詳細(xì)源代碼參見(jiàn)CAPI開(kāi)源項(xiàng)目中的CDistributedQueue.h。
原文鏈接：http://blog.csdn.net/drzhouweiming/article/details/4006930
