4.4 使用同步操作簡化代碼

同步工具的使用在本章稱為構(gòu)建塊，你可以之關(guān)注那些需要同步的操作，而非具體使用的機(jī)制。當(dāng)需要為程序的并發(fā)時，這是一種可以幫助你簡化你的代碼的方式，提供更多的函數(shù)化的方法。比起在多個線程間直接共享數(shù)據(jù)，每個任務(wù)擁有自己的數(shù)據(jù)會應(yīng)該會更好，并且結(jié)果可以對其他線程進(jìn)行廣播，這就需要使用“期望”來完成了。

4.4.1 使用“期望”的函數(shù)化編程

術(shù)語函數(shù)化編程(functional programming)引用于一種編程方式，這種方式中的函數(shù)結(jié)果只依賴于傳入函數(shù)的參數(shù)，并不依賴外部狀態(tài)。當(dāng)一個函數(shù)與數(shù)學(xué)概念相關(guān)時，當(dāng)你使用相同的函數(shù)調(diào)用這個函數(shù)兩次，這兩次的結(jié)果會完全相同。C++標(biāo)準(zhǔn)庫中很多與數(shù)學(xué)相關(guān)的函數(shù)都有這個特性，例如，sin(正弦),cos(余弦)和sqrt(平方根)；當(dāng)然，還有基本類型間的簡單運(yùn)算，例如，3+3，6*9，或1.3/4.7。一個純粹的函數(shù)不會改變?nèi)魏瓮獠繝顟B(tài)，并且這種特性完全限制了函數(shù)的返回值。

很容易想象這是一種什么樣的情況，特別是當(dāng)并行發(fā)生時，因為在第三章時我們討論過，很多問題發(fā)生在共享數(shù)據(jù)上。當(dāng)共享數(shù)據(jù)沒有被修改，那么就不存在條件競爭，并且沒有必要使用互斥量去保護(hù)共享數(shù)據(jù)。這可對編程進(jìn)行極大的簡化，例如Haskell語言[2]，在Haskell中函數(shù)默認(rèn)就是這么的“純粹”；這種純粹對的方式，在并發(fā)編程系統(tǒng)中越來越受歡迎。因為大多數(shù)函數(shù)都是純粹的，那么非純粹的函數(shù)對共享數(shù)據(jù)的修改就顯得更為突出，所以其很容易適應(yīng)應(yīng)用的整體結(jié)構(gòu)。

函數(shù)化編程的好處，并不限于那些將“純粹”作為默認(rèn)方式(范型)的語言。C++是一個多范型的語言，其也可以寫出FP類型的程序。在C++11中這種方式要比C++98簡單許多，因為C++11支持lambda表達(dá)式(詳見附錄A，A.6節(jié))，還加入了Boost和TR1中的std::bind，以及自動可以自行推斷類型的自動變量(詳見附錄A，A.7節(jié))?！捌谕弊鳛槠磮D的最后一塊，它使得函數(shù)化編程模式并發(fā)化(FP-style concurrency)在C++中成為可能；一個“期望”對象可以在線程間互相傳遞，并允許其中一個計算結(jié)果依賴于另外一個的結(jié)果，而非對共享數(shù)據(jù)的顯式訪問。

快速排序 FP模式版

為了描述在函數(shù)化(PF)并發(fā)中使用“期望”，讓我們來看看一個簡單的實現(xiàn)——快速排序算法。該算法的基本思想很簡單：給定一個數(shù)據(jù)列表，然后選取其中一個數(shù)為“中間”值，之后將列表中的其他數(shù)值分成兩組——一組比中間值大，另一組比中間值小。之后對小于“中間”值的組進(jìn)行排序，并返回排序好的列表；再返回“中間”值；再對比“中間”值大的組進(jìn)行排序，并返回排序的列表。圖4.2中展示了10個整數(shù)在這種方式下進(jìn)行排序的過程。

http://wiki.jikexueyuan.com/project/cplusplus-concurrency-action/images/chapter4/4-2.png" alt="" />

圖4.2 FP-模式的遞歸排序

下面清單中的代碼是FP-模式的順序?qū)崿F(xiàn)，它需要傳入列表，并且返回一個列表，而非與std::sort()做同樣的事情。 (譯者：std::sort()是無返回值的，因為參數(shù)接收的是迭代器，所以其可以對原始列表直進(jìn)行修改與排序。可參考sort())

清單4.12 快速排序——順序?qū)崿F(xiàn)版

template<typename T>
std::list<T> sequential_quick_sort(std::list<T> input)
{
  if(input.empty())
  {
    return input;
  }
  std::list<T> result;
  result.splice(result.begin(),input,input.begin());  // 1
  T const& pivot=*result.begin();  // 2

  auto divide_point=std::partition(input.begin(),input.end(),
             [&](T const& t){return t<pivot;});  // 3

  std::list<T> lower_part;
  lower_part.splice(lower_part.end(),input,input.begin(),
             divide_point);  // 4
  auto new_lower(
             sequential_quick_sort(std::move(lower_part)));  // 5
  auto new_higher(
             sequential_quick_sort(std::move(input)));  // 6

  result.splice(result.end(),new_higher);  // 7
  result.splice(result.begin(),new_lower);  // 8
  return result;
}

雖然接口的形式是FP模式的，但當(dāng)你使用FP模式時，你需要做大量的拷貝操作，所以在內(nèi)部你會使用“普通”的命令模式。你選擇第一個數(shù)為“中間”值，使用splice()①將輸入的首個元素(中間值)放入結(jié)果列表中。雖然這種方式產(chǎn)生的結(jié)果可能不是最優(yōu)的(會有大量的比較和交換操作)，但是對std::list做任何事都需要花費(fèi)較長的時間，因為鏈表是遍歷訪問的。你知道你想要什么樣的結(jié)果，所以你可以直接將要使用的“中間”值提前進(jìn)行拼接?，F(xiàn)在你還需要使用“中間”值進(jìn)行比較，所以這里使用了一個引用②，為了避免過多的拷貝。之后，你可以使用std::partition將序列中的值分成小于“中間”值的組和大于“中間”值的組③。最簡單的方法就是使用lambda函數(shù)指定區(qū)分的標(biāo)準(zhǔn)；使用已獲取的引用避免對“中間”值的拷貝(詳見附錄A，A.5節(jié)，更多有關(guān)lambda函數(shù)的信息)。

std::partition()對列表進(jìn)行重置，并返回一個指向首元素(不小于“中間”值)的迭代器。迭代器的類型全稱可能會很長，所以你可以使用auto類型說明符，讓編譯器幫助你定義迭代器類型的變量(詳見附錄A，A.7節(jié))。

現(xiàn)在，你已經(jīng)選擇了FP模式的接口；所以，當(dāng)你要使用遞歸對兩部分排序是，你將需要創(chuàng)建兩個列表。你可以用splice()函數(shù)來做這件事，將input列表小于divided_point的值移動到新列表lower_part④中。其他數(shù)繼續(xù)留在input列表中。而后，你可以使用遞歸調(diào)用⑤⑥的方式，對兩個列表進(jìn)行排序。這里顯式使用std::move()將列表傳遞到類函數(shù)中，這種方式還是為了避免大量的拷貝操作。最終，你可以再次使用splice()，將result中的結(jié)果以正確的順序進(jìn)行拼接。new_higher指向的值放在“中間”值的后面⑦，new_lower指向的值放在“中間”值的前面⑧。

快速排序 FP模式線程強(qiáng)化版

因為還是使用函數(shù)化模式，所以使用“期望”很容易將其轉(zhuǎn)化為并行的版本，如下面的程序清單所示。其中的操作與前面相同，不同的是它們現(xiàn)在并行運(yùn)行。

清單4.13 快速排序——“期望”并行版

template<typename T>
std::list<T> parallel_quick_sort(std::list<T> input)
{
  if(input.empty())
  {
    return input;
  }
  std::list<T> result;
  result.splice(result.begin(),input,input.begin());
  T const& pivot=*result.begin();

  auto divide_point=std::partition(input.begin(),input.end(),
                [&](T const& t){return t<pivot;});

  std::list<T> lower_part;
  lower_part.splice(lower_part.end(),input,input.begin(),
                divide_point);

  std::future<std::list<T> > new_lower(  // 1
                std::async(&parallel_quick_sort<T>,std::move(lower_part)));

  auto new_higher(
                parallel_quick_sort(std::move(input)));  // 2

  result.splice(result.end(),new_higher);  // 3
  result.splice(result.begin(),new_lower.get());  // 4
  return result;
}

這里最大的變化是，當(dāng)前線程不對小于“中間”值部分的列表進(jìn)行排序，使用std::async()①在另一線程對其進(jìn)行排序。大于部分列表，如同之前一樣，使用遞歸的方式進(jìn)行排序②。通過遞歸調(diào)用parallel_quick_sort()，你就可以利用可用的硬件并發(fā)了。std::async()會啟動一個新線程，這樣當(dāng)你遞歸三次時，就會有八個線程在運(yùn)行了；當(dāng)你遞歸十次(對于大約有1000個元素的列表)，如果硬件能處理這十次遞歸調(diào)用，你將會創(chuàng)建1024個執(zhí)行線程。當(dāng)運(yùn)行庫認(rèn)為這樣做產(chǎn)生了太多的任務(wù)時(也許是因為數(shù)量超過了硬件并發(fā)的最大值)，運(yùn)行庫可能會同步的切換新產(chǎn)生的任務(wù)。當(dāng)任務(wù)過多時(已影響性能)，這些任務(wù)應(yīng)該在使用get()函數(shù)獲取的線程上運(yùn)行，而不是在新線程上運(yùn)行，這樣就能避免任務(wù)向線程傳遞的開銷。值的注意的是，這完全符合std::async的實現(xiàn)，為每一個任務(wù)啟動一個線程(甚至在任務(wù)超額時；在std::launch::deferred沒有明確規(guī)定的情況下)；或為了同步執(zhí)行所有任務(wù)(在std::launch::async有明確規(guī)定的情況下)。當(dāng)你依賴運(yùn)行庫的自動縮放，建議你去查看一下你的實現(xiàn)文檔，了解一下將會有怎么樣的行為表現(xiàn)。

比起使用std::async()，你可以寫一個spawn_task()函數(shù)對std::packaged_task和std::thread做簡單的包裝，如清單4.14中的代碼所示；你需要為函數(shù)結(jié)果創(chuàng)建一個std::packaged_task對象， 可以從這個對象中獲取“期望”，或在線程中執(zhí)行它，返回“期望”。其本身并不提供太多的好處(并且事實上會造成大規(guī)模的超額任務(wù))，但是它會為轉(zhuǎn)型成一個更復(fù)雜的實現(xiàn)鋪平道路，將會實現(xiàn)向一個隊列添加任務(wù)，而后使用線程池的方式來運(yùn)行它們。我們將在第9章再討論線程池。使用std::async更適合于當(dāng)你知道你在干什么，并且要完全控制在線程池中構(gòu)建或執(zhí)行過任務(wù)的線程。

清單4.14 spawn_task的簡單實現(xiàn)

template<typename F,typename A>
std::future<std::result_of<F(A&&)>::type>
   spawn_task(F&& f,A&& a)
{
  typedef std::result_of<F(A&&)>::type result_type;
  std::packaged_task<result_type(A&&)>
       task(std::move(f)));
  std::future<result_type> res(task.get_future());
  std::thread t(std::move(task),std::move(a));
  t.detach();
  return res;
}

其他先不管，回到parallel_quick_sort函數(shù)。因為你只是直接遞歸去獲取new_higher列表，你可以如之前一樣對new_higher進(jìn)行拼接③。但是，new_lower列表是std::future<std::list<T>>的實例，而非是一個簡單的列表，所以你需要調(diào)用get()成員函數(shù)在調(diào)用splice()④之前去檢索數(shù)值。在這之后，等待后臺任務(wù)完成，并且將結(jié)果移入splice()調(diào)用中；get()返回一個包含結(jié)果的右值引用，所以這個結(jié)果是可以移出的(詳見附錄A，A.1.1節(jié)，有更多有關(guān)右值引用和移動語義的信息)。

即使假設(shè)，使用std::async()是對可用硬件并發(fā)最好的選擇，但是這樣的并行實現(xiàn)對于快速排序來說，依然不是最理想的。其中，std::partition做了很多工作，即使做了依舊是順序調(diào)用，但就現(xiàn)在的情況來說，已經(jīng)足夠好了。如果你對實現(xiàn)最快并行的可能性感興趣的話，你可以去查閱一些學(xué)術(shù)文獻(xiàn)。

因為避開了共享易變數(shù)據(jù)，函數(shù)化編程可算作是并發(fā)編程的范型；并且也是通訊順序進(jìn)程(CSP,Communicating Sequential Processer[3],)的范型，這里線程理論上是完全分開的，也就是沒有共享數(shù)據(jù)，但是有通訊通道允許信息在不同線程間進(jìn)行傳遞。這種范型被Erlang語言所采納，并且在MPI(Message Passing Interface，消息傳遞接口)上常用來做C和C++的高性能運(yùn)算。現(xiàn)在你應(yīng)該不會在對學(xué)習(xí)它們而感到驚奇了吧，因為只需遵守一些約定，C++就能支持它們；在接下來的一節(jié)中，我們會討論實現(xiàn)這種方式。

4.4.2 使用消息傳遞的同步操作

CSP的概念十分簡單：當(dāng)沒有共享數(shù)據(jù)，每個線程就可以進(jìn)行獨(dú)立思考，其行為純粹基于其所接收到的信息。每個線程就都有一個狀態(tài)機(jī)：當(dāng)線程收到一條信息，它將會以某種方式更新其狀態(tài)，并且可能向其他線程發(fā)出一條或多條信息，對于消息的處理依賴于線程的初始化狀態(tài)。這是一種正式寫入這些線程的方式，并且以有限狀態(tài)機(jī)的模式實現(xiàn)，但是這不是唯一的方案；狀態(tài)機(jī)可以在應(yīng)用程序中隱式實現(xiàn)。這種方法咋任何給定的情況下，都更加依賴于特定情形下明確的行為要求和編程團(tuán)隊的專業(yè)知識。無論你選擇用什么方式去實現(xiàn)每個線程，任務(wù)都會分成獨(dú)立的處理部分，這樣會消除潛在的混亂(數(shù)據(jù)共享并發(fā))，這樣就讓編程變的更加簡單，且擁有低錯誤率。

真正通訊順序處理是沒有共享數(shù)據(jù)的，所有消息都是通過消息隊列傳遞，但是因為C++線程共享一塊地址空間，所以達(dá)不到真正通訊順序處理的要求。這里就需要有一些約定了：作為一款應(yīng)用或者是一個庫的作者，我們有責(zé)任確保在我們的實現(xiàn)中，線程不存在共享數(shù)據(jù)。當(dāng)然，為了線程間的通信，消息隊列是必須要共享的，具體的細(xì)節(jié)可以包含在庫中。

試想，有一天你要為實現(xiàn)ATM(自動取款機(jī))寫一段代碼。這段代碼需要處理，人們嘗試取錢時和銀行之間的交互情況，以及控制物理器械接受用戶的卡片，顯示適當(dāng)?shù)男畔?，處理按鈕事件，吐出現(xiàn)金，還有退還用戶的卡。

一種處理所有事情的方法是讓代碼將所有事情分配到三個獨(dú)立線程上去：一個線程去處理物理機(jī)械，一個去處理ATM機(jī)的邏輯，還有一個用來與銀行通訊。這些線程可以通過信息進(jìn)行純粹的通訊，而非共享任何數(shù)據(jù)。比如，當(dāng)有人在ATM機(jī)上插入了卡片或者按下按鈕，處理物理機(jī)械的線程將會發(fā)送一條信息到邏輯線程上，并且邏輯線程將會發(fā)送一條消息到機(jī)械線程，告訴機(jī)械線程可以分配多少錢，等等。

一種為ATM機(jī)邏輯建模的方式是將其當(dāng)做一個狀態(tài)機(jī)。線程的每一個狀態(tài)都會等待一條可接受的信息，這條信息包含需要處理的內(nèi)容。這可能會讓線程過渡到一個新的狀態(tài)，并且循環(huán)繼續(xù)。在圖4.3中將展示，有狀態(tài)參與的一個簡單是實現(xiàn)。在這個簡化實現(xiàn)中，系統(tǒng)在等待一張卡插入。當(dāng)有卡插入時，系統(tǒng)將會等待用戶輸入它的PIN(類似身份碼的東西)，每次輸入一個數(shù)字。用戶可以將最后輸入的數(shù)字刪除。當(dāng)數(shù)字輸入完成，PIN就需要驗證。當(dāng)驗證有問題，你的程序就需要終止，就需要為用戶退出卡，并且繼續(xù)等待其他人將卡插入到機(jī)器中。當(dāng)PIN驗證通過，你的程序要等待用戶取消交易或選擇取款。當(dāng)用戶選擇取消交易，你的程序就可以結(jié)束，并返還卡片。當(dāng)用戶選擇取出一定量的現(xiàn)金，你的程序就要在吐出現(xiàn)金和返還卡片前等待銀行方面的確認(rèn)，或顯示“余額不足”的信息，并返還卡片。很明顯，一個真正的ATM機(jī)要考慮的東西更多、更復(fù)雜，但是我們來說這樣描述已經(jīng)足夠了。

http://wiki.jikexueyuan.com/project/cplusplus-concurrency-action/images/chapter4/4-3.png" alt="" />

圖4.3 一臺ATM機(jī)的狀態(tài)機(jī)模型(簡化)

我們已經(jīng)為你的ATM機(jī)邏輯設(shè)計了一個狀態(tài)機(jī)，你可以使用一個類實現(xiàn)它，這個類中有一個成員函數(shù)可以代表每一個狀態(tài)。每一個成員函數(shù)可以等待從指定集合中傳入的信息，以及當(dāng)他們到達(dá)時進(jìn)行處理，這就有可能觸發(fā)原始狀態(tài)向另一個狀態(tài)的轉(zhuǎn)化。每種不同的信息類型由一個獨(dú)立的struct表示。清單4.15展示了ATM邏輯部分的簡單實現(xiàn)(在以上描述的系統(tǒng)中，有主循環(huán)和對第一狀態(tài)的實現(xiàn))，并且一直在等待卡片插入。

如你所見，所有信息傳遞所需的的同步，完全包含在“信息傳遞”庫中(基本實現(xiàn)在附錄C中，是清單4.15代碼的完整版)

清單4.15 ATM邏輯類的簡單實現(xiàn)

struct card_inserted
{
  std::string account;
};

class atm
{
  messaging::receiver incoming;
  messaging::sender bank;
  messaging::sender interface_hardware;
  void (atm::*state)();

  std::string account;
  std::string pin;

  void waiting_for_card()  // 1
  {
    interface_hardware.send(display_enter_card());  // 2
    incoming.wait().  // 3
      handle<card_inserted>(
      [&](card_inserted const& msg)  // 4
      {
       account=msg.account;
       pin="";
       interface_hardware.send(display_enter_pin());
       state=&atm::getting_pin;
      }
    );
  }
  void getting_pin();
public:
  void run()  // 5
  {
    state=&atm::waiting_for_card;  // 6
    try
    {
      for(;;)
      {
        (this->*state)();  // 7
      }
    }
    catch(messaging::close_queue const&)
    {
    }
  }
};

如之前提到的，這個實現(xiàn)對于實際ATM機(jī)的邏輯來說是非常簡單的，但是他能讓你感受到信息傳遞編程的方式。這里無需考慮同步和并發(fā)問題，只需要考慮什么時候接收信息和發(fā)送信息即可。為ATM邏輯所設(shè)的狀態(tài)機(jī)運(yùn)行在獨(dú)立的線程上，與系統(tǒng)的其他部分一起，比如與銀行通訊的接口，以及運(yùn)行在獨(dú)立線程上的終端接口。這種程序設(shè)計的方式被稱為參與者模式(Actor model)——在系統(tǒng)中有很多獨(dú)立的(運(yùn)行在一個獨(dú)立的線程上)參與者，這些參與者會互相發(fā)送信息，去執(zhí)行手頭上的任務(wù)，并且它們不會共享狀態(tài)，除非是通過信息直接傳入的。

運(yùn)行從run()成員函數(shù)開始⑤，其將會初始化waiting_for_card⑥的狀態(tài)，然后反復(fù)執(zhí)行當(dāng)前狀態(tài)的成員函數(shù)(無論這個狀態(tài)時怎么樣的)⑦。狀態(tài)函數(shù)是簡易atm類的成員函數(shù)。wait_for_card函數(shù)①依舊很簡單：它發(fā)送一條信息到接口，讓終端顯示“等待卡片”的信息②，之后就等待傳入一條消息進(jìn)行處理③。這里處理的消息類型只能是card_inserted類的，這里使用一個lambda函數(shù)④對其進(jìn)行處理。當(dāng)然，你可以傳遞任何函數(shù)或函數(shù)對象，去處理函數(shù)，但對于一個簡單的例子來說，使用lambda表達(dá)式是最簡單的方式。注意handle()函數(shù)調(diào)用是連接到wait()函數(shù)上的；當(dāng)收到的信息類型與處理類型不匹配，收到的信息會被丟棄，并且線程繼續(xù)等待，直到接收到一條類型匹配的消息。

lambda函數(shù)自身，只是將用戶的賬號信息緩存到一個成員變量中去，并且清除PIN信息，再發(fā)送一條消息到硬件接口，讓顯示界面提示用戶輸入PIN，然后將線程狀態(tài)改為“獲取PIN”。當(dāng)消息處理程序結(jié)束，狀態(tài)函數(shù)就會返回，然后主循環(huán)會調(diào)用新的狀態(tài)函數(shù)⑦。

如圖4.3，getting_pin狀態(tài)函數(shù)會負(fù)載一些，因為其要處理三個不同的信息類型。具體代碼展示如下：

清單4.16 簡單ATM實現(xiàn)中的getting_pin狀態(tài)函數(shù)

void atm::getting_pin()
{
  incoming.wait()
    .handle<digit_pressed>(  // 1
      [&](digit_pressed const& msg)
      {
        unsigned const pin_length=4;
        pin+=msg.digit;
        if(pin.length()==pin_length)
        {
          bank.send(verify_pin(account,pin,incoming));
          state=&atm::verifying_pin;
        }
      }
      )
    .handle<clear_last_pressed>(  // 2
      [&](clear_last_pressed const& msg)
      {
        if(!pin.empty())
        {
          pin.resize(pin.length()-1);
        }
      }
      )
    .handle<cancel_pressed>(  // 3
      [&](cancel_pressed const& msg)
      {
        state=&atm::done_processing;
      }
      );
}

這次需要處理三種消息類型，所以wait()函數(shù)后面接了三個handle()函數(shù)調(diào)用①②③。每個handle()都有對應(yīng)的消息類型作為模板參數(shù)，并且將消息傳入一個lambda函數(shù)中(其獲取消息類型作為一個參數(shù))。因為這里的調(diào)用都被連接在了一起，wait()的實現(xiàn)知道它是等待一條digit_pressed消息，或是一條clear_last_pressed肖息，亦或是一條cancel_pressed消息，其他的消息類型將會被丟棄。

這次當(dāng)你獲取一條消息時，無需再去改變狀態(tài)。比如，當(dāng)你獲取一條digit_pressed消息時，你僅需要將其添加到pin中，除非那些數(shù)字是最終的輸入。(清單4.15中)主循環(huán)⑦將會再次調(diào)用getting_pin()去等待下一個數(shù)字(或清除數(shù)字，或取消交易)。

這里對應(yīng)的動作如圖4.3所示。每個狀態(tài)盒的實現(xiàn)都由一個不同的成員函數(shù)構(gòu)成，等待相關(guān)信息并適當(dāng)?shù)母聽顟B(tài)。

如你所見，在一個并發(fā)系統(tǒng)中這種編程方式可以極大的簡化任務(wù)的設(shè)計，因為每一個線程都完全被獨(dú)立對待。因此，在使用多線程去分離關(guān)注點(diǎn)時，需要你明確如何分配線程之間的任務(wù)。

[2] 詳見 http://www.haskell.org/.

[3] 《通信順序進(jìn)程》(Communicating Sequential Processes), C.A.R. Hoare, Prentice Hall, 1985. 免費(fèi)在線閱讀地址 http://www.usingcsp.com/cspbook.pdf.