本章主要內(nèi)容

• 等待一個事件
• 用期望等待一次性事件
• 帶時間限制的等待
• 使用操作的同步來簡化代碼

上一章中，我們看到各種在線程間保護共享數(shù)據(jù)的方法。但有時，你不僅需要保護數(shù)據(jù)，還需要同步不同線程上的操作。例如，一個線程可能需要等待另一個線程完成一個任務，然后第一個線程才能完成自己的任務。一般來說，通常希望線程等待特定事件發(fā)生或一個條件變?yōu)檎妗１M管可以通過定期檢查共享數(shù)據(jù)中存儲的“任務完成”標記或類似的東西來實現(xiàn)這一點，但這遠不夠理想。像這樣需要在線程之間同步操作的場景是如此的常見，以至于C++標準庫提供了條件變量(condition variables)和期望（futures）形式的設施來處理它。這些設施在并發(fā)技術規(guī)范(TS，Conncurrency Technical Specification)中得到了擴展，技術規(guī)范為期望（futures）提供了更多的操作，一起的還有新的同步設施鎖存器(latches)和屏障(barriers)。

本章將討論如何使用條件變量，期望，鎖存器以及屏障來等待事件，以及如何使用它們來簡化操作的同步。

4.1 等待一個事件或其他條件

假設你乘坐通宵火車旅行。一種確保你在正確的車站下車的方法是整晚保持清醒，并注意火車停在哪里。這樣你就不會誤站，但是等你到站的時候估計也累夠嗆。或者，你可以看一下時刻表，看看火車應該什么時候到達，然后把鬧鐘定得比到站時間稍微早一點， 接著就可以去睡覺了。這樣就可以了；你也不會誤站，但如果火車晚點，你就醒得太早了。當然，鬧鐘的電池也可能會沒電了，于是你就睡過了頭，以至于誤了站。理想的方式是，你可以去睡覺，不管什么時候，只要火車到站，就有人或其他東西能把你喚醒就好了。

這和線程有什么關系呢？嗯，如果一個線程正在等待另一個線程完成一個任務，它有幾個選項。首先，它可以不斷檢查共享數(shù)據(jù)中的標記(由互斥鎖保護)，并讓第二個線程在完成任務時設置該標記。這在兩個方面是浪費的:線程不斷檢查標記會消耗寶貴的處理時間，并且當互斥鎖被等待的線程鎖住時，其他線程不能鎖住它。這兩者對等待線程都不利：如果等待線程在運行，這就限制了可用的執(zhí)行資源去運行被等待的線程，同時為了檢查標記，等待線程鎖住了互斥鎖來保護它，被等待線程就不能在它完成任務后鎖住互斥鎖來設置標記。這種情況類似于你整晚和列車駕駛員交談：駕駛員不得不減慢火車的速度，因為你分散了他的注意力，所以火車需要更長的時間才能到站。類似地，正在等待的線程正在消耗系統(tǒng)中其他線程可以使用的資源，最終等待時間可能比必要的時間更長。

第二個選擇是讓等待線程在檢查的間隙用std::this_thread::sleep_for()函數(shù)休眠很短的時間(參見4.3節(jié))：

bool flag; std::mutex m;void wait_for_flag() {std::unique_lock<std::mutex> lk(m);while(!flag){lk.unlock(); // 1 解鎖互斥鎖std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 2 休眠100mslk.lock(); // 3 再次鎖住互斥鎖} }

循環(huán)體中，在休眠前②，函數(shù)對互斥鎖進行解鎖①，并且在休眠結(jié)束后再對互斥鎖進行上鎖③，因此另外的線程就有機會獲取鎖并設置標記。

這是一個進步，因為當線程休眠時，線程沒有浪費執(zhí)行時間，但是很難確定正確的休眠時間。太短的休眠仍然會浪費處理時間去做檢查；太長的休眠時間，會導致當被等待線程完成時，線程還處于休眠狀態(tài)，從而導致耽擱。這種睡過頭的情況很少會對程序的運行產(chǎn)生直接影響，但它可能意味著在快節(jié)奏的游戲中掉幀或在實時應用中超出時間片。

第三個也是首選的方法是使用C++標準庫提供的設施去等待事件本身。等待另一個線程觸發(fā)事件的最基本機制(例如前面提到的在流水線中存在的額外工作)是條件變量（condition variable）。從概念上講，條件變量與事件或其他條件（condition）相關聯(lián)，一個或多個線程可以等待該條件滿足。當一個線程確定滿足條件時，它可以通知一個或多個等待條件變量的線程，以喚醒它們并允許它們繼續(xù)處理。

4.1.1 使用條件變量等待條件

C++標準庫對條件變量有兩套實現(xiàn)：std::condition_variable和std::condition_variable_any。這兩個實現(xiàn)都包含在<condition_variable>庫頭文件中。兩者都需要與一個互斥鎖一起才能工作，因為需要互斥鎖提供適當?shù)耐?#xff1b;前者僅限于使用std::mutex，而后者可以使用任何滿足類似于互斥鎖的最低標準的對象，因而帶有_any后綴。由于std::condition_variable_any更通用，因此在大小、性能或操作系統(tǒng)資源方面有額外的潛在成本，所以除非需要額外的靈活性，否則應該首選std::condition_variable。

那么，如何使用std::condition_variable來處理簡介中的示例呢?如何讓正在等待工作的線程休眠，直到有數(shù)據(jù)要處理?下面的清單展示了使用條件變量實現(xiàn)的一種方法。

首先，有一個用來在兩個線程之間傳遞數(shù)據(jù)的隊列①。當數(shù)據(jù)準備好時，準備數(shù)據(jù)的線程使用std::lock_guard來保護隊列，并把數(shù)據(jù)推入隊列中②。然后它調(diào)用std::condition_variable實例的notify_one()成員函數(shù)通知等待線程 (如果有的話)③。注意,你把將數(shù)據(jù)推入隊列的代碼放在一個較小的作用域,所以你在解鎖之后通知條件變量——這是為了,如果等待線程立即醒來,它沒必要再被阻塞在等待你解鎖互斥鎖。

在柵欄的另一側(cè)，有一個正在處理數(shù)據(jù)的線程，這個線程首先鎖住互斥鎖，但這次使用std::unique_lock而不是std::lock_guard④——你馬上就會知道為什么。然后線程在std::condition_variable上調(diào)用wait()成員函數(shù)，并傳入鎖對象和表示等待條件的lambda函數(shù)⑤。Lambda函數(shù)是C++11添加的新特性，它可以讓一個匿名函數(shù)作為另一個表達式的一部分，并且它們非常適合被指定為wait()這種標準庫函數(shù)的謂詞。在這個例子中，簡單的Lambda函數(shù)[]{return !data_queue.empty();}會去檢查data_queue是否非空——也就是說，隊列中有數(shù)據(jù)準備要處理。附錄A的A.5節(jié)有Lambda函數(shù)更多的細節(jié)。

wait()會去檢查這些條件(通過調(diào)用所提供的lambda函數(shù))，當條件滿足(lambda函數(shù)返回true)時返回。如果條件不滿足(lambda函數(shù)返回false)，wait()函數(shù)將解鎖互斥鎖，并且將這個線程置于阻塞或等待狀態(tài)。當準備數(shù)據(jù)的線程調(diào)用notify_one()通知條件變量時，處理數(shù)據(jù)的線程從睡眠狀態(tài)中醒來，獲取互斥鎖上的鎖，并且再次檢查條件是否滿足。在條件滿足的情況下，從wait()返回并仍然持有鎖；當條件不滿足時，線程將對互斥鎖解鎖，并且重新開始等待。這就是為什么用std::unique_lock而不使用std::lock_guard——等待中的線程必須在等待期間解鎖互斥鎖，并在這之后對互斥鎖再次上鎖，而std::lock_guard沒有這么靈活。如果互斥鎖在線程休眠期間保持鎖住狀態(tài)，準備數(shù)據(jù)的線程將無法鎖住互斥鎖，也就無法添加數(shù)據(jù)項到隊列中，這樣等待線程也永遠看不到它的條件被滿足。

清單4.1為等待使用了一個簡單的lambda函數(shù)⑤，它檢查隊列是否非空，不過任何函數(shù)和可調(diào)用對象都可以擔此責任。如果已經(jīng)有了檢查條件的函數(shù)(可能因為它比像這樣簡單的測試要復雜一些)，那么可以直接傳入此函數(shù)，不一定非要包在一個lambda中。在調(diào)用wait()期間，條件變量可以對提供的條件檢查任意次數(shù)；但是它總是在鎖住互斥鎖的情況下才這么做，并且當(且僅當)用于測試條件的函數(shù)返回true時，它將立即返回。當?shù)却木€程重新獲得互斥鎖并檢查條件時，如果它不是直接響應來自另一個線程的通知，則稱為偽喚醒(spurious wakeup)。因為根據(jù)定義，任何這種偽喚醒的數(shù)量和頻率都是不確定的，所以不建議使用具有副作用的函數(shù)進行條件檢查。如果你這樣做，你必須為副作用發(fā)生多次做好準備。

基本上，std::condition_variable::wait是對忙-等待的優(yōu)化。事實上，一個合格(雖然不太理想)的實現(xiàn)技術可以只是一個簡單的循環(huán)：

template<typename Predicate> void minimal_wait(std::unique_lock<std::mutex>& lk, Predicate pred){while(!pred()){lk.unlock();lk.lock();} }

你的代碼必須準備不但能使用這種最小的wait()實現(xiàn)，而且還能使用只有在調(diào)用notify_one()或notify_all()時才會喚醒的實現(xiàn)。

解鎖std::unique_lock的靈活性，不僅適用于對wait()的調(diào)用；它還可以用在數(shù)據(jù)待處理但還未處理的時候⑥。處理數(shù)據(jù)可能是一個耗時的操作，正如你在第3章中看到的，在互斥鎖上持有的時間超過必要的時間不是一個好主意。

像清單4.1這樣，使用隊列在多個線程間轉(zhuǎn)移數(shù)據(jù)是很常見的。如果做得好，同步可以限制在隊列本身，這將極大地減少同步問題和競爭條件的可能數(shù)量。鑒于此，現(xiàn)在讓我們從清單4.1中提取一個通用的線程安全隊列

4.1.2 使用條件變量構(gòu)建線程安全隊列

如果你準備設計一個通用隊列，花點時間想想隊列需要哪些操作是值得的，就像在3.2.3節(jié)線程安全棧中做的一樣。我們可以從C++標準庫中找靈感，形式為std::queue<>的容器適配器如下所示的：

如果忽略構(gòu)造、賦值以及交換操作時，就只剩下了三組操作：查詢整個隊列的狀態(tài)的操作(empty()和size())；查詢隊列中元素的操作(front()和back())；修改隊列的操作(push(), pop()和emplace())。這和3.2.3中的棧一樣，因此也會遇到在接口上固有的競爭條件。所以，需要將front()和pop()合成一個函數(shù)調(diào)用，就像之前在棧實現(xiàn)時合并top()和pop()一樣。清單4.1中的代碼加入一些細微的變化：當使用隊列在線程之間傳遞數(shù)據(jù)時，接收線程通常需要等待數(shù)據(jù)。這里提供pop()函數(shù)的兩個變種：try_pop()和wait_and_pop()。try_pop()，嘗試從隊列中彈出數(shù)據(jù)，它總會直接返回(帶有失敗指示)，即使沒有值可檢索；wait_and_pop()，將會等到有值可檢索的時候才返回。如果你以棧示例為指引，接口可能會是下面這樣：

和棧一樣，為了簡化代碼，減少了構(gòu)造函數(shù)并刪除了賦值操作符。和之前一樣，也提供了兩個版本的try_pop()和wait_for_pop()。第一個重載的try_pop()①把檢索的值存儲在引用變量中，所以它可以用返回值做狀態(tài)；當檢索到一個值時，它將返回true，否則返回false(參見A.2節(jié))。第二個重載②就不能這樣了，因為它直接返回檢索到的值。不過，當沒有值可檢索時，這個函數(shù)可以返回NULL指針。

那么，所有這些與清單4.1有什么關系呢?嗯，你可以從中抽取代碼用于push()和wait_and_pop()，如下面的清單所示。

互斥鎖和條件變量現(xiàn)在包含在threadsafe_queue實例中，因此不再需要單獨的變量①，并且調(diào)用push()也不需要外部同步②。另外，wait_and_pop()負責條件變量的等待③。

另一個重載的wait_and_pop()現(xiàn)在編寫起來很簡單，剩下的函數(shù)幾乎可以逐字從清單3.5中的棧示例中拷貝。最終的隊列實現(xiàn)展示如下。

盡管empty()是一個const成員函數(shù)，并且拷貝構(gòu)造函數(shù)的other參數(shù)是一個const引用，但是其他線程可能有對該對象的非const引用，并且可能正在調(diào)用可變的成員函數(shù)，因此你仍然需要鎖住互斥鎖。因為鎖住互斥鎖是一種可變操作，所以互斥鎖對象必須標記為可變的(mutable)①，這樣就可以在empty()和拷貝構(gòu)造函數(shù)中鎖住它。

在多個線程等待同一事件時，條件變量也很有用。如果線程用于劃分工作負載，因此只有一個線程應該響應通知，那么可以使用與清單4.1中所示完全相同的結(jié)構(gòu)，只需運行多個數(shù)據(jù)處理線程實例。當新數(shù)據(jù)準備好時，調(diào)用notify_one()將會觸發(fā)一個正在執(zhí)行wait()的線程去檢查它的條件并且從wait()函數(shù)返回(因為你剛向data_queue中添加一個數(shù)據(jù)項)。 不能保證哪個線程會被通知，甚至不能保證是否有線程在等待被通知，因為有可能所有的處理線程仍然在處理數(shù)據(jù)。

另一種可能是幾個線程在等待同一事件，并且它們都需要響應該事件。這可能發(fā)生在共享數(shù)據(jù)初始化的情況下，所有的處理線程可以使用相同的數(shù)據(jù)，但是需要等待它被初始化(盡管可能有更好的機制，比如std::call once；關于這個選項的討論，請參閱第3章的3.3.1節(jié))，或者線程需要等待共享數(shù)據(jù)的更新，比如定期的重新初始化。在這些情況下，準備數(shù)據(jù)的線程可以對條件變量調(diào)用notify_all()成員函數(shù)，而不是notify_one()。顧名思義，這將導致當前執(zhí)行wait()的所有線程檢查它們正在等待的條件。

如果等待線程只等待一次，因此當條件為真時，它將不再等待該條件變量，那么條件變量可能不是同步機制的最佳選擇。如果等待的條件是某一特定數(shù)據(jù)的可用性，則尤其如此。在這種情況下，期望（future）可能更合適。

4.2 使用期望等待一次性事件

假設你要乘飛機去國外度假。一旦你到達機場，完成了各種登機手續(xù)，你還得等待你的航班準備登機的通知，這可能要等上好幾個小時。是的，你也許能找到一些消磨時間的方式，比如看書、上網(wǎng)，或者在機場價格高昂的咖啡館用餐，但基本上你只是在等待一件事:登機的信號。不僅如此，一個給定的航班只會有一次;下次你去度假時，你將等待不同的航班。

C++標準庫將這種一次性事件建模為所謂的期望（future）。如果一個線程需要等待一個特定的一次性事件，它會以某種方式獲得一個表示該事件的期望。然后，線程可以周期性地等待很短的一段時間，以查看事件是否已經(jīng)發(fā)生(查看出發(fā)時刻表)，同時在檢查的間隙執(zhí)行其他任務(在價格高昂的咖啡館用餐)。或者，它可以執(zhí)行另一個任務，直到它需要事件在它繼續(xù)之前發(fā)生，然后就等待期望變成就緒（ready）。期望可能有與之相關的數(shù)據(jù)(比如你的航班在哪個登機口登機)，也可能沒有。一旦事件發(fā)生(因此期望已經(jīng)變成就緒)，期望就不能被重置。

C++標準庫中，有兩種期望，實現(xiàn)為兩個類模板，聲明在<future>庫頭文件中：唯一的期望（unique futures）(std::future<>)和共享的期望（shared futures） (std::shared_future<>)。它們仿照了std::unique_ptr和std::shared_ptr。一個std::future的實例是唯一一個引用其關聯(lián)事件的實例，而多個std::shared_future實例可能引用同一事件。后一種情況中，所有實例會在同時變?yōu)榫途w狀態(tài)，然后他們可以訪問與事件相關的任何數(shù)據(jù)。這些關聯(lián)的數(shù)據(jù)是這些類成為模板的原因；就像std::unique_ptr和std::shared_ptr一樣，模板參數(shù)是關聯(lián)數(shù)據(jù)的類型。如果沒有相關聯(lián)的數(shù)據(jù)，可以使用std::future<void>與std::shared_future<void>的特化模板。盡管期望用于線程間通信，但是期望對象本身不提供同步訪問。如果多個線程需要訪問一個期望對象，它們必須通過互斥鎖或其他同步機制來保護訪問，如第3章所述。但是，正如你將在4.2.5節(jié)中看到的，多個線程可以訪問它們自己的std::shared_future<>副本，而無需進一步同步，即使它們都引用相同的異步結(jié)果。

并發(fā)技術規(guī)范在std::experimental名空間中提供了這些類模板的擴展版本：std::experimental::future<>和std::experimental:: shared_future<>。這些類的行為與std名空間中的對應類相同，但是它們有額外的成員函數(shù)來提供額外的功能。需要重點注意的是，名字std::experimental并非暗示代碼的質(zhì)量(我希望實現(xiàn)的質(zhì)量和你的庫供應商提供的其他東西是一樣的)，但是需要強調(diào)的是，這些都是非標準的類和函數(shù)，因此，如果它們最終被采用到未來的C++標準中，它們的語法和語義可能會有變化。如果想要使用這些設施，需要包含<experimental/future>頭文件。

最基本的一次性事件是在后臺運行的計算的結(jié)果。在第2章中，你看到std::thread并沒有提供一種簡單方法從這樣的任務中返回一個值，并且我承諾過將在第4章中用期望來解決——現(xiàn)在是時候看看怎么解決了。

4.2.1 從后臺任務返回值

假設有一個長時間運行的計算，你希望最終產(chǎn)生一個有用的結(jié)果，但當前不需要該值。也許你已經(jīng)找到了一種方法來確定生命,宇宙和萬物的答案——從道格拉斯·亞當斯[1]那取一個例子（譯注：作者這里開玩笑，扯遠了，可以無視）。你可以啟動一個新的線程來執(zhí)行計算，但這意味著你必須負責把結(jié)果傳送回來,因為std::thread沒有提供直接的機制來做這個事情。這就是需要std::async函數(shù)模板(也聲明在<future>頭文件中)的地方。

如果你不需要立即得到結(jié)果，可以使用std::async來啟動一個異步任務（asynchronous task）。而不是給你一個std::thread對象去等待，std::async會返回一個std::future對象，它將最終持有函數(shù)的返回值。當你需要該值時，只需在期望上調(diào)用get()，線程就會阻塞，直到期望就緒（ready），然后返回該值。下面的清單顯示了一個簡單的示例。

std::async允許你通過向調(diào)用中添加更多的參數(shù)來傳遞額外的參數(shù)給函數(shù)，這與std::thread的方法相同。如果第一個參數(shù)是指向成員函數(shù)的指針，那么第二個參數(shù)提供了應用成員函數(shù)的對象(要么直接是對象，要么通過指針，亦或包裝在std::ref中)，其余的參數(shù)作為成員函數(shù)的參數(shù)傳遞。否則，第二個和隨后的參數(shù)將作為函數(shù)或可調(diào)用對象的第一個參數(shù)。就如std::thread，當參數(shù)為右值時，拷貝操作將使用移動（moving）的方式轉(zhuǎn)移原始數(shù)據(jù)。這就允許使用只支持移動的類型作為函數(shù)對象和參數(shù)。參見下面的清單：

默認情況下，當?shù)却谕麜r，std::async是否啟動一個新線程，還是同步執(zhí)行任務，取決于實現(xiàn)。在大多數(shù)情況下，這是你想要的，但是你可以在調(diào)用函數(shù)之前，通過std::async的附加參數(shù)指定要使用哪種模式。這個參數(shù)的類型是std::launch，它可以是std::launch::defered，表明函數(shù)調(diào)用被推遲到wait()或get()函數(shù)調(diào)用時才執(zhí)行，或者是std::launch::async，表明函數(shù)必須在它自己的線程上運行，還可以是std::launch::deferred | std::launch::async表明讓具體實現(xiàn)來選擇哪種方式。最后一個選項是默認的。如果函數(shù)調(diào)用是推遲的，它可能永遠也不會運行。例如：

auto f6=std::async(std::launch::async,Y(),1.2); // 在新線程上執(zhí)行 auto f7=std::async(std::launch::deferred,baz,std::ref(x)); // 在wait()或get()調(diào)用時執(zhí)行 auto f8=std::async(std::launch::deferred | std::launch::async,baz,std::ref(x)); // 實現(xiàn)選擇執(zhí)行方式 auto f9=std::async(baz,std::ref(x)); // 實現(xiàn)選擇執(zhí)行方式 f7.wait(); // 調(diào)用延遲函數(shù)

正如你將在本章后面以及第8章中看到的，使用std::async可以很容易地將算法劃分為可以并發(fā)運行的任務。然而，這并不是將std::future與任務聯(lián)系起來的唯一方法；你還可以通過將任務包裝到std::packaged_task<>類模板的實例中，或者通過編寫代碼使用std::promise<>類模板顯式地設置值來實現(xiàn)。std::packaged_task是一個比std::promise更高層次的抽象，所以我將從它開始。

總結(jié)

以上是生活随笔為你收集整理的c++高级编程(第4版).pdf_《C++并发编程实战第2版》第四章：同步并发操作（1/4）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。