試想，你被要求負(fù)責(zé)建造一座房子。為了完成任務(wù)，你需要挖地基、砌墻、添加水暖、接入電線，等等。理論上，如果你很擅長建造屋子，那么這些事情都可以由你來完成，但是這樣就要花費很長很長時間，并且需要不斷的切換任務(wù)。或者，你可以雇傭一些人來幫助你完成房子的建造。那么現(xiàn)在你需要決定雇多少人，以及雇傭人員具有什么樣的技能。比如，你可以雇幾個人，這幾個人什么都會?，F(xiàn)在你還得不斷的切換任務(wù)，不過因為雇傭了很多人，就要比之前的速度快很多。
或者，你可以雇傭一個包工隊(專家組)，由瓦工，木匠，電工和水管工組成。你的包工隊員只做其擅長的，所以當(dāng)沒有水暖任務(wù)時，水管工會坐在那里休息，喝茶或咖啡。因為人多的緣故，要比之前一個人的速度快很多，并且水管工在收拾廁所的時候，電工可以將電線連接到廚房，不過當(dāng)沒有屬于自己的任務(wù)時，有人就會休息。即使有人在休息，你可能還是能感覺到包工隊要比雇傭一群什么都會的人快。包工隊不需要更換工具，并且每個人的任務(wù)都要比會的人做的快。是快還是慢，取決于特定的情況——需要嘗試，進(jìn)行觀察。
即使雇傭包工隊，你依舊可以選擇人數(shù)不同的團(tuán)隊(可能在一個團(tuán)隊中，瓦工的數(shù)量超過電工)。同樣，這會是一種補(bǔ)足，并且在建造不止一座房子的時候，會改變整體效率。即使水管工沒有太多的任務(wù)，在建造過一次房子后，你依舊能讓他總是處于忙碌的狀態(tài)。當(dāng)包工隊無事可做的時候，你是不會給他們錢的；即使每次工作只有那么幾個人工作，你還需要負(fù)擔(dān)整個團(tuán)隊的開銷。
建造例子已經(jīng)足夠說明問題；這與線程所做的事情有什么關(guān)系呢？好吧，這些問題也會發(fā)生在線程上。你需要決定使用多少個線程，并且這些線程應(yīng)該去做什么。還需要決定是使用“全能”的線程去完成所有的任務(wù)，還是使用“專業(yè)”線程只去完成一件事情，或?qū)煞N方法混合。使用并發(fā)的時候，需要作出諸多選擇來驅(qū)動并發(fā)，這里的選擇會決定代碼的性能和清晰度。因此，這里的選擇至關(guān)重要，所以在你設(shè)計應(yīng)用程序的結(jié)構(gòu)時，再作出適當(dāng)?shù)臎Q定。在本節(jié)中，將看到很多劃分任務(wù)的技術(shù)，就先從線程間劃分?jǐn)?shù)據(jù)開始吧！
最簡單的并行算法，就是并行化的std::for_each，其會對一個數(shù)據(jù)集中每個元素執(zhí)行同一個操作。為了并行化該算法，可以為數(shù)據(jù)集中每個元素分配一個處理線程。如何劃分才能獲得最佳的性能，很大程度上取決于數(shù)據(jù)結(jié)構(gòu)實現(xiàn)的細(xì)節(jié)，在之后有關(guān)性能問題的章節(jié)會再提及此問題。
最簡單的分配方式：第一組N個元素分配一個線程，下一組N個元素再分配一個線程，以此類推，如圖8.1所示。不管數(shù)據(jù)怎么分，每個線程都會對分配給它的元素進(jìn)行操作，不過并不會和其他線程進(jìn)行溝通，直到處理完成。
http://wiki.jikexueyuan.com/project/cplusplus-concurrency-action/images/chapter8/8-1.png" alt="" />
圖8.1 向線程分發(fā)連續(xù)的數(shù)據(jù)塊
使用過MPI(Message Passing Interface)[1]和OpenMP[2]的人對這個結(jié)構(gòu)一定很熟悉：一項任務(wù)被分割成多個，放入一個并行任務(wù)集中，執(zhí)行線程獨立的執(zhí)行這些任務(wù)，結(jié)果在會有主線程中合并。這種方式在2.4節(jié)中的accumulate的例子中使用過了；在這個例子中，所有并行任務(wù)和主線程的任務(wù)都是累加和。對于for_each來說，主線程將無事可做，因為這個計算不需要最終處理。
最后一步對于并行程序來說十分重要；如清單2.8中那樣原始的實現(xiàn)，最后一步就是一個串行的。不過，這一步同樣也是能被并行化的；accumulate實際上是一個遞減操作，所以清單2.8中，當(dāng)線程數(shù)量大于一個線程上最小處理項時，可以對accumulate進(jìn)行遞歸調(diào)用?；蛘?，工作線程就像做一個完整的任務(wù)一樣，對步驟進(jìn)行遞減，而非每次都產(chǎn)生新的線程。
雖然這個技術(shù)十分強(qiáng)大，但是并不是哪都適用。有時不能像之前那樣，對任務(wù)進(jìn)行整齊的劃分，因為只有對數(shù)據(jù)進(jìn)行處理后，才能進(jìn)行明確的劃分。這里特別適用了遞歸算法，就像快速排序；下面就來看看這種特別的方式。
快速排序有兩個最基本的步驟：將數(shù)據(jù)劃分到中樞元素之前或之后，然后對中樞元素之前和之后的兩半數(shù)組再次進(jìn)行快速排序。這里不能通過對數(shù)據(jù)的簡單劃分達(dá)到并行，因為，只有在一次排序結(jié)束后，才能知道哪些項在中樞元素之前和之后。當(dāng)要對這種算法進(jìn)行并行化，很自然的會想到使用遞歸。每一級的遞歸都會多次調(diào)用quick_sort函數(shù)，因為需要知道哪些元素在中樞元素之前和之后。遞歸調(diào)用是完全獨立的，因為其訪問的是不同的數(shù)據(jù)集，并且每次迭代都能并發(fā)執(zhí)行。圖8.2展示了這樣的遞歸劃分。
http://wiki.jikexueyuan.com/project/cplusplus-concurrency-action/images/chapter8/8-2.png" alt="" />
圖 8.2 遞歸劃分?jǐn)?shù)據(jù)
在第4章中，已經(jīng)見過這種實現(xiàn)。比起對大于和小于的數(shù)據(jù)塊遞歸調(diào)用函數(shù)，使用std::async()可以為每一級生成小于數(shù)據(jù)塊的異步任務(wù)。使用std::async()時，C++線程庫就能決定何時讓一個新線程執(zhí)行任務(wù)，以及同步執(zhí)行任務(wù)。
重要的是：對一個很大的數(shù)據(jù)集進(jìn)行排序時，當(dāng)每層遞歸都產(chǎn)生一個新線程，最后就會產(chǎn)生大量的線程。你會看到其對性能的影響，如果有太多的線程存在，那么你的應(yīng)用將會運行的很慢。如果數(shù)據(jù)集過于龐大，會將線程耗盡。那么在遞歸的基礎(chǔ)上進(jìn)行任務(wù)的劃分，就是一個不錯的主意；你只需要將一定數(shù)量的數(shù)據(jù)打包后，交給線程即可。std::async()可以出里這種簡單的情況，不過這不是唯一的選擇。
另一種選擇是使用std::thread::hardware_concurrency()函數(shù)來確定線程的數(shù)量，就像在清單2.8中的并行版accumulate()一樣。然后，你可以將已排序的數(shù)據(jù)推到線程安全的棧上(如第6、7章中提及的棧)。當(dāng)線程無所事事，不是已經(jīng)完成對自己數(shù)據(jù)塊的梳理，就是在等待一組排序數(shù)據(jù)的產(chǎn)生；線程可以從棧上獲取這組數(shù)據(jù)，并且對其排序。
下面的代碼就是使用以上方式進(jìn)行的實現(xiàn)。
清單8.1 使用棧的并行快速排序算法——等待數(shù)據(jù)塊排序
template<typename T>
struct sorter // 1
{
struct chunk_to_sort
{
std::list<T> data;
std::promise<std::list<T> > promise;
};
thread_safe_stack<chunk_to_sort> chunks; // 2
std::vector<std::thread> threads; // 3
unsigned const max_thread_count;
std::atomic<bool> end_of_data;
sorter():
max_thread_count(std::thread::hardware_concurrency()-1),
end_of_data(false)
{}
~sorter() // 4
{
end_of_data=true; // 5
for(unsigned i=0;i<threads.size();++i)
{
threads[i].join(); // 6
}
}
void try_sort_chunk()
{
boost::shared_ptr<chunk_to_sort > chunk=chunks.pop(); // 7
if(chunk)
{
sort_chunk(chunk); // 8
}
}
std::list<T> do_sort(std::list<T>& chunk_data) // 9
{
if(chunk_data.empty())
{
return chunk_data;
}
std::list<T> result;
result.splice(result.begin(),chunk_data,chunk_data.begin());
T const& partition_val=*result.begin();
typename std::list<T>::iterator divide_point= // 10
std::partition(chunk_data.begin(),chunk_data.end(),
[&](T const& val){return val<partition_val;});
chunk_to_sort new_lower_chunk;
new_lower_chunk.data.splice(new_lower_chunk.data.end(),
chunk_data,chunk_data.begin(),
divide_point);
std::future<std::list<T> > new_lower=
new_lower_chunk.promise.get_future();
chunks.push(std::move(new_lower_chunk)); // 11
if(threads.size()<max_thread_count) // 12
{
threads.push_back(std::thread(&sorter<T>::sort_thread,this));
}
std::list<T> new_higher(do_sort(chunk_data));
result.splice(result.end(),new_higher);
while(new_lower.wait_for(std::chrono::seconds(0)) !=
std::future_status::ready) // 13
{
try_sort_chunk(); // 14
}
result.splice(result.begin(),new_lower.get());
return result;
}
void sort_chunk(boost::shared_ptr<chunk_to_sort> const& chunk)
{
chunk->promise.set_value(do_sort(chunk->data)); // 15
}
void sort_thread()
{
while(!end_of_data) // 16
{
try_sort_chunk(); // 17
std::this_thread::yield(); // 18
}
}
};
template<typename T>
std::list<T> parallel_quick_sort(std::list<T> input) // 19
{
if(input.empty())
{
return input;
}
sorter<T> s;
return s.do_sort(input); // 20
}這里，parallel_quick_sort函數(shù)?代表了sorter類①的功能，其支持在棧上簡單的存儲無序數(shù)據(jù)塊②，并且對線程進(jìn)行設(shè)置③。do_sort成員函數(shù)⑨主要做的就是對數(shù)據(jù)進(jìn)行劃分⑩。相較于對每一個數(shù)據(jù)塊產(chǎn)生一個新的線程，這次會將這些數(shù)據(jù)塊推到棧上?；并在有備用處理器?的時候，產(chǎn)生新線程。因為小于部分的數(shù)據(jù)塊可能由其他線程進(jìn)行處理，那么就得等待這個線程完成?。為了讓所有事情順利進(jìn)行(只有一個線程和其他所有線程都忙碌時)，當(dāng)線程處于等待狀態(tài)時?，就讓當(dāng)前線程嘗試處理棧上的數(shù)據(jù)。try_sort_chunk只是從棧上彈出一個數(shù)據(jù)塊⑦，并且對其進(jìn)行排序⑧，將結(jié)果存在promise中，讓線程對已經(jīng)存在于棧上的數(shù)據(jù)塊進(jìn)行提取?。
當(dāng)end_of_data沒有被設(shè)置時?，新生成的線程還在嘗試從棧上獲取需要排序的數(shù)據(jù)塊?。在循環(huán)檢查中，也要給其他線程機(jī)會?，可以從棧上取下數(shù)據(jù)塊進(jìn)行更多的操作。這里的實現(xiàn)依賴于sorter類④對線程的清理。當(dāng)所有數(shù)據(jù)都已經(jīng)排序完成，do_sort將會返回(即使還有工作線程在運行)，所以主線程將會從parallel_quick_sort?中返回，在這之后會銷毀sorter對象。析構(gòu)函數(shù)會設(shè)置end_of_data標(biāo)志⑤，以及等待所有線程完成工作⑥。標(biāo)志的設(shè)置將終止線程函數(shù)內(nèi)部的循環(huán)?。
在這個方案中，不用為spawn_task產(chǎn)生的無數(shù)線程所困擾，并且也不用再依賴C++線程庫，為你選擇執(zhí)行線程的數(shù)量(就像std::async()那樣)。該方案制約線程數(shù)量的值就是std::thread::hardware_concurrency()的值，這樣就能避免任務(wù)過于頻繁的切換。不過，這里還有兩個問題：線程管理和線程通訊。要解決這兩個問題就要增加代碼的復(fù)雜程度。雖然，線程對數(shù)據(jù)項是分開處理的，不過所有對棧的訪問，都可以向棧添加新的數(shù)據(jù)塊，并且移出數(shù)據(jù)塊以作處理。這里重度的競爭會降低性能(即使使用無鎖(無阻塞)棧)，原因?qū)诤竺嫣岬健?/p>
這個方案使用到了一個特殊的線程池——所有線程的任務(wù)都來源于一個等待鏈表，然后線程會去完成任務(wù)，完成任務(wù)后會再來鏈表提取任務(wù)。這個線程池很有問題(包括對工作鏈表的競爭)，這個問題的解決方案將在第9章提到。關(guān)于多處理器的問題，將會在本章后面的章節(jié)中做出更為詳細(xì)的介紹(詳見8.2.1)。
幾種劃分方法：1，處理前劃分；2，遞歸劃分(都需要事先知道數(shù)據(jù)的長度固定)，還有上面的那種劃分方式。事情并非總是這樣好解決；當(dāng)數(shù)據(jù)是動態(tài)生成，或是通過外部輸入，那么這里的辦法就不適用了。在這種情況下，基于任務(wù)類型的劃分方式，就要好于基于數(shù)據(jù)的劃分方式。
雖然為每個線程分配不同的數(shù)據(jù)塊，但工作的劃分(無論是之前就劃分好，還是使用遞歸的方式劃分)仍然在理論階段，因為這里每個線程對每個數(shù)據(jù)塊的操作是相同的。而另一種選擇是讓線程做專門的工作，也就是每個線程做不同的工作，就像水管工和電工在建造一所屋子的時候所做的不同工作那樣。線程可能會對同一段數(shù)據(jù)進(jìn)行操作，但它們對數(shù)據(jù)進(jìn)行不同的操作。
對分工的排序，也就是從并發(fā)分離關(guān)注結(jié)果；每個線程都有不同的任務(wù)，這就意味著真正意義上的線程獨立。其他線程偶爾會向特定線程交付數(shù)據(jù)，或是通過觸發(fā)事件的方式來進(jìn)行處理；不過總體而言，每個線程只需要關(guān)注自己所要做的事情即可。其本身就是基本良好的設(shè)計，每一段代碼只對自己的部分負(fù)責(zé)。
分離關(guān)注
當(dāng)有多個任務(wù)需要持續(xù)運行一段時間，或需要及時進(jìn)行處理的事件(比如，按鍵事件或傳入網(wǎng)絡(luò)數(shù)據(jù))，且還有其他任務(wù)正在運行時，單線程應(yīng)用采用的是單職責(zé)原則處理沖突。單線程的世界中，代碼會執(zhí)行任務(wù)A(部分)后，再去執(zhí)行任務(wù)B(部分)，再檢查按鈕事件，再檢查傳入的網(wǎng)絡(luò)包，然后在循環(huán)回去，執(zhí)行任務(wù)A。這將會使得任務(wù)A復(fù)雜化，因為需要存儲完成狀態(tài)，以及定期從主循環(huán)中返回。如果在循環(huán)中添加了很多任務(wù)，那么程序?qū)⑦\行的很慢；并且用戶會發(fā)現(xiàn)，在他/她按下按鍵后，很久之后才會有反應(yīng)。我確定你已經(jīng)在一些程序中見過這種情況：你給程序分配一項任務(wù)后，發(fā)現(xiàn)接口會封鎖，直到這項任務(wù)完成。
當(dāng)使用獨立線程執(zhí)行任務(wù)時，操作系統(tǒng)會幫你處理接口問題。在執(zhí)行任務(wù)A時，線程可以專注于執(zhí)行任務(wù)，而不用為保存狀態(tài)從主循環(huán)中返回。操作系統(tǒng)會自動保存狀態(tài)，當(dāng)需要的時候，將線程切換到任務(wù)B或任務(wù)C。如果目標(biāo)系統(tǒng)是帶有多核或多個處理器，任務(wù)A和任務(wù)B可很可能真正的并發(fā)執(zhí)行。這樣處理按鍵時間或網(wǎng)絡(luò)包的代碼，就能及時執(zhí)行了。所有事情都完成的很好，用戶得到了及時的響應(yīng)；當(dāng)然，作為開發(fā)者只需要寫具體操作的代碼即可，不用再將控制分支和使用用戶交互混在一起了。
聽起來不錯，玫瑰色的愿景呀。事實真像上面所說的那樣簡單？一切取決于細(xì)節(jié)。如果每件事都是獨立的，那么線程間就不需要交互，這樣的話一切都很簡單了。不幸的是，現(xiàn)實沒那么美好。后臺那些優(yōu)雅的任務(wù)，經(jīng)常會被用戶要求做一些事情，并且它們需要通過更新用戶接口的方式，來讓用戶知道它們完成了任務(wù)?；蛘撸脩艨赡芟胍∠蝿?wù)，這就需要用戶向接口發(fā)送一條消息，告知后臺任務(wù)停止運行。這兩種情況都需要認(rèn)真考慮，設(shè)計，以及適當(dāng)?shù)耐?，不過這里擔(dān)心的部分還是分離的。用戶接口線程只能處理用戶接口，當(dāng)其他線程告訴該線程要做什么時，用戶接口線程會進(jìn)行更新。同樣，后臺線程只運行它們所關(guān)注的任務(wù)；只是，有時會發(fā)生“允許任務(wù)被其他線程所停止”的情況。在這兩種情況下，后臺線程需要照顧來自其他線程的請求，線程本身只知道它們請求與自己的任務(wù)有所關(guān)聯(lián)。
多線程下有兩個危險需要分離關(guān)注。第一個是對錯誤擔(dān)憂的分離，主要表現(xiàn)為線程間共享著很多的數(shù)據(jù)，或者不同的線程要相互等待；這兩種情況都是因為線程間很密切的交互。當(dāng)這種情況發(fā)生，就需要看一下為什么需要這么多交互。當(dāng)所有交互都有關(guān)于同樣的問題，就應(yīng)該使用單線程來解決，并將引用同一原因的線程提取出來?；蛘撸?dāng)有兩個線程需要頻繁的交流，且沒有其他線程時，那么就可以將這兩個線程合為一個線程。
當(dāng)通過任務(wù)類型對線程間的任務(wù)進(jìn)行劃分時，不應(yīng)該讓線程處于完全隔離的狀態(tài)。當(dāng)多個輸入數(shù)據(jù)集需要使用同樣的操作序列，可以將序列中的操作分成多個階段，來讓每個線程執(zhí)行。
劃分任務(wù)序列
當(dāng)任務(wù)會應(yīng)用到相同操作序列，去處理獨立的數(shù)據(jù)項時，就可以使用流水線(pipeline)系統(tǒng)進(jìn)行并發(fā)。這好比一個物理管道：數(shù)據(jù)流從管道一端進(jìn)入，在進(jìn)行一系列操作后，從管道另一端出去。
使用這種方式劃分工作，可以為流水線中的每一階段操作創(chuàng)建一個獨立線程。當(dāng)一個操作完成，數(shù)據(jù)元素會放在隊列中，以供下一階段的線程提取使用。這就允許第一個線程在完成對于第一個數(shù)據(jù)塊的操作，并要對第二個數(shù)據(jù)塊進(jìn)行操作時，第二個線程可以對第一個數(shù)據(jù)塊執(zhí)行管線中的第二個操作。
這就是在線程間劃分?jǐn)?shù)據(jù)的一種替代方案(如8.1.1描述)；這種方式適合于操作開始前，且對輸入數(shù)據(jù)處長度不清楚的情況。例如，數(shù)據(jù)來源可能是從網(wǎng)絡(luò)，或者可能是通過掃描文件系統(tǒng)來確定要處理的文件。
流水線對于隊列中耗時的操作處理的也很合理；通過對線程間任務(wù)的劃分，就能對應(yīng)用的性能所有改善。假設(shè)有20個數(shù)據(jù)項，需要在四核的機(jī)器上處理，并且每一個數(shù)據(jù)項需要四個步驟來完成操作，每一步都需要3秒來完成。如果你將數(shù)據(jù)分給了四個線程，那么每個線程上就有5個數(shù)據(jù)項要處理。假設(shè)在處理的時候，沒有其他線程對處理過程進(jìn)行影響，在12秒后4個數(shù)據(jù)項處理完成，24秒后8個數(shù)據(jù)項處理完成，以此類推。當(dāng)20個數(shù)據(jù)項都完成操作，就需要1分鐘的時間。在管線中就會完全不同。四步可以交給四個內(nèi)核。那么現(xiàn)在，第一個數(shù)據(jù)項可以被每一個核進(jìn)行處理，所以其還是會消耗12秒。的確，在12秒后你就能得到一個處理過的數(shù)據(jù)項，這相較于數(shù)據(jù)劃分并沒有好多少。不過，當(dāng)流水線流動起來，事情就會不一樣了；在第一個核處理第一個數(shù)據(jù)項后，數(shù)據(jù)項就會交給下一個內(nèi)核，所以第一個核在處理完第一個數(shù)據(jù)項后，其還可以對第二個數(shù)據(jù)項進(jìn)行處理。那么在12秒后，每3秒將會得到一個已處理的數(shù)據(jù)項，這就要好于每隔12秒完成4個數(shù)據(jù)項。
為什么整批處理的時間要長于流水線呢？因為你需要在最終核開始處理第一個元素前等待9秒。更平滑的操作，能在某些情況下獲益更多。考慮如下情況：當(dāng)一個系統(tǒng)用來播放高清數(shù)字視頻。為了讓視頻能夠播放，你至少要保證25幀每秒的解碼速度。同樣的，這些圖像需要有均勻的間隔，才會給觀眾留有連續(xù)播放的感覺；一個應(yīng)用可以在1秒解碼100幀，不過在解完就需要暫停1s的時候，這個應(yīng)用就沒有意義了。另一方面，觀眾能接受在視頻開始播放的時候有一定的延遲。這種情況，并行使用流水線就能得到穩(wěn)定的解碼率。
看了這么多線程間劃分工作的技術(shù)，接下來讓我們來看一下在多線程系統(tǒng)中有哪些因素會影響性能，并且這些因素是如何影響你選擇劃分方案的。
掃碼關(guān)注鍍金池微信號
手機(jī)訂課，優(yōu)惠多多