主要有三部分組成，threadpool，scheduler，task。

三者關(guān)系如上圖示，pplx只著重實現(xiàn)了task部分功能，scheduler跟threadpool只是簡略實現(xiàn)。

threadpool主要依賴boost.asio達(dá)到跨平臺的目標(biāo)，cpprestsdk的 io操作同時也依賴這個threadpool。

pplx提供了兩個版本的scheduler，分別是

linux_scheduler依賴boost.asio.threadpool。

window_schedule依賴win32 ThreadPool。

默認(rèn)的scheduler只是簡單地將work投遞到threadpool進(jìn)行分派。

用戶可以根據(jù)自己需要，實現(xiàn)scheduler_interface，提供復(fù)雜的調(diào)度。

每個task關(guān)聯(lián)著一個_Task_impl實現(xiàn)體，一個_TaskCollection_t(喚醒事件，后繼任務(wù)隊列，這個隊列的任務(wù)之間的關(guān)系是并列的)，還有一個_PPLTaskHandle代碼執(zhí)行單元。

task，并行執(zhí)行的單位任務(wù)。通過scheduler將代碼執(zhí)行單元調(diào)度到線程去執(zhí)行。

task提供類似activeobject模式的功能，可以看作是一個future，通過get()同步阻塞等待執(zhí)行結(jié)果。

task提供拓?fù)淠Ｐ?#xff0c;通過then()創(chuàng)建后續(xù)task，并作為后繼執(zhí)行任務(wù)。注意的是每個task可以接受不限數(shù)量的then()，這些后繼任務(wù)之間并不串行。例 task().then().then()串行，(task1.then(), task1.then())并行。一個任務(wù)在執(zhí)行完成時，會將結(jié)果傳遞給它的所有直接后繼執(zhí)行任務(wù)。

此外，task拓?fù)涑藅hen()函數(shù)外，還可以在執(zhí)行l(wèi)ambda中添加并行分支，然后可以在后繼任務(wù)中同步這些分支。

也就是說后繼任務(wù)同步原本task拓?fù)渫獾膖ask拓?fù)洳拍芾^續(xù)執(zhí)行。

1 auto fork0 =

2 task([]()->task{

3 auto fork1 =

4 task([]()->task{

5 auto fork2 =

6 task([](){

7 // do your fork2 work

8

9 });

10 // do your fork1 work

11

12 returnfork2;

13 }).then([](task& frk2){ frk2.wait(); }); // will sync fork2

14 // do your fork0 work

15

16 returnfork1;

17 }).then([](task& frk1){ frk1.wait(); }); // will sync fork1

18 fork0.wait(); // sync fork1, fork2

上面的方式有一個問題，如果里層的fork先完成，將不要阻塞線程，但是外層fork先完成就不得不阻塞線程等待內(nèi)層fork完成。

所以可以用when_all

task >([]()->task{

std::vector >forks;

forks.push_back( task([]() { /* do fork0 work */}) );

forks.push_back( task([]() { /* do fork1 work */}) );

forks.push_back( task([]() { /* do fork2 work */}) );

forks.push_back( task([]() { /* do fork3 work */}) );

returnwhen_all(std::begin(forks), std::end(forks));

}).then([](taskforks){

forks.wait();

}).wait();

通過上面的方式，也可以在lambda中，將其它task拓?fù)洳迦氲侥阍瓉淼膖ask拓?fù)洹?/p>

task結(jié)束，分兩種情況，完成以及取消。取消執(zhí)行，只能在執(zhí)行代碼時通過拋出異常，task并沒有提供取消的接口。任務(wù)在執(zhí)行過程中拋出的異常，就會被task捕捉，并暫存異常，然后取消執(zhí)行。異常在wait()時重新拋出。下面的時序分析可以看到全過程 。

值得注意的是，PPL中task原本的設(shè)計是的有Async與Inline之分的。在_Task_impl_base::_Wait()有一小段注釋說明

// If this task was created from a Windows Runtime async operation, do not attempt to inline it. The

// async operation will take place on a thread in the appropriate apartment Simply wait for the completed

// event to be set.

也就是task除了由scheduler調(diào)度到線程池分派執(zhí)行，還可以強(qiáng)制在wait()函數(shù)內(nèi)分派執(zhí)行，后繼task也不必再次調(diào)度而可以在當(dāng)前線程繼續(xù)分派執(zhí)行。但是pplx沒有實現(xiàn)

class_TaskCollectionImpl

{

...

void_Cancel()

{

// No cancellation support

}

void_RunAndWait()

{

// No inlining support yet

_Wait();

}

下面是對task的時序分析。

開始的task創(chuàng)建_InitialTaskHandle， 一種只能用于始首的Handle執(zhí)行單元。

通過then()添加的task，創(chuàng)建_ContinuationTaskHandle，(一種可以入鏈的后繼執(zhí)行單元)，并暫存起來。

當(dāng)一個任務(wù)在線程池中分派結(jié)束時，就會將所有通過then()添加到它結(jié)尾的后繼任務(wù)一次過向scheduler調(diào)度出去。

任務(wù)只能通過拋出異常從而自己中止執(zhí)行，task并暫存異常(及錯誤信息)。

后繼任務(wù)被調(diào)度到線程池繼續(xù)分派執(zhí)行。

這里順便討論一個開銷，在window版本中，每個task都有一個喚醒事件，使用事件內(nèi)核對象，都要創(chuàng)建釋放一個內(nèi)核對象，在高并行任務(wù)時，可能會消耗過多內(nèi)核對象，消耗句柄數(shù)。

并且continuation后繼任務(wù)，在默認(rèn)scheduler調(diào)度下，不會在同一線程中分派，所有后繼任務(wù)都會簡單投遞到線程池。由線程池去決定分派的線程。所以由then()串行起來的任務(wù)可能會由不同的線程順序分派，從而產(chǎn)生開銷。因為pplx并沒有實現(xiàn) Inline功能，所有task都會視作Async重新調(diào)度到線程池。

總結(jié)

以上是生活随笔為你收集整理的mysql ppl_浅析pplx库的设计与实现。的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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