workflow異步任務調度編程范式

workflow是搜狗開源的一個開發(fā)框架?？梢詽M足絕大多數(shù)日常服務器開發(fā)，性能優(yōu)異，給上層業(yè)務提供了易于開發(fā)的接口，卻只用了少量的代碼，舉重若輕，而且代碼整潔干凈易讀。

搜狗官方宣傳強調，workflow是一個異步任務調度編程范式，封裝了6種異步資源：CPU計算、GPU計算、網絡、磁盤I/O、定時器、計數(shù)器，以回調函數(shù)模式提供給用戶使用，概括起來實際上主要是兩個功能：1、屏蔽阻塞調用的影響，使阻塞調用的開發(fā)接口變?yōu)楫惒降?，充分利用計算資源；2、框架管理線程池，使開發(fā)者迅速構建并行計算程序。

往往單臺機器要服務于千千萬萬終端，我們最希望服務器資源都能充分利用，然而計算資源和I/O資源天然的效率不對等，使我們不得不采用一些其他技術手段實現(xiàn)基礎資源充分利用。所謂I/O資源包括文件I/O和網絡I/O，此外很多時候我們需要定時執(zhí)行某段邏輯，同樣不希望等待時間阻塞計算資源的使用。

所以框架最基礎的功能，是要為上層開發(fā)人員屏蔽底層資源的不對稱，使我們可以方便的開發(fā)業(yè)務邏輯而不需要把很多精力放在底層。

如何擬合計算資源和io資源

我們希望io等待或其他阻塞的時間，cpu還能充分利用，執(zhí)行一些任務。這要求發(fā)起io的線程不能調用阻塞接口原地等待，而是要切出去，往往采用I/O多路復用或者異步I/O的方式，分別對應reactor模型和proactor模型

對于網絡I/O，linux系統(tǒng)下缺乏對異步I/O的支持，即使近兩年有了iouring，支持了異步io，但性能上相對epoll未必會有多少提升，而且一切都交給系統(tǒng)調度，可控性上大大降低；另外開發(fā)難度也更大。反觀epoll，無論系統(tǒng)的支持還是相關設計模型都非常成熟了，所以近一二十年底層大都采用epoll，以reactor模式實現(xiàn)，reactor統(tǒng)一處理請求，將就緒的任務轉給下游的處理器。根據(jù)業(yè)務不同，又有幾種不同實現(xiàn)方式，有的就單線程之內調度，單線程循環(huán)處理（如redis），適合業(yè)務邏輯不復雜的場景；有的會單reactor處理請求，并通過消息隊列把請求轉發(fā)給下游多線程業(yè)務邏輯處理器處理；有的多線程多reactor處理請求，并通過消息隊列將任務分發(fā)給下游handler，單reactor模式可以認為是這種模式的特例，workflow便以這種方式實現(xiàn)。

對于文件I/O，linux下有兩種異步I/O的支持，posix aio（glibcaio）和linux 原生 aio，其中前者是一個通過多線程的異步，模擬的異步io，性能極差；linux 原生 aio是真正的aio，但是要求fd只能以O_DIRECT方式打開，所以只適用于文件I/O，workflow中支持了這種方式處理文件I/O。

對于定時器，常見的方式，有的通過epoll每次阻塞設置阻塞時間，用戶態(tài)管理定時器（如redis）；而epoll也支持時間事件，有的直接使用時間事件，workflow便采用這種方式。

提供給用戶的接口

計算資源得以充分利用，還需要考慮給用戶提供什么樣的接口，讓上層開發(fā)者能減少心智負擔，比如，以協(xié)程的方式，讓用戶像開發(fā)串行程序一樣開發(fā)異步程序，順序的寫邏輯；亦或者是提供讓用戶注冊回調的方式開發(fā)異步程序。workflow中提出了子任務的概念，以任務的方式提供給用戶。

子任務定義了一種管理回調的方式，用串行并行來組織子任務調度。用戶可以把邏輯寫在任務里，交給框架去調度。

把阻塞的任務交給epoll去異步調用，計算任務交給線程池去異步執(zhí)行，以至于所有的任務都是異步調起的，這種設計思想，就是workflow被稱為“異步任務調度框架”的原因。

代碼分析

根據(jù)上面的分析，對一般服務器框架結構已經有了一個整體認識。下面按這個順序，底層基礎數(shù)據(jù)結構——》純計算任務和Reactor層——》任務組織調度層——》用戶接口層，分四個層次逐步分析一下workflow的實現(xiàn)。

基礎數(shù)據(jù)結構

workflow使用到的基礎數(shù)據(jù)結構：鏈表、紅黑樹、消息隊列、線程池，workflow中這四個結構的設計都非常的精致。

鏈表（見文件 list.h)

workflow中的鏈表貌似引自linux內核，實現(xiàn)了一種非常非常靈活的鏈表，甚至鏈表串起的不同節(jié)點之間可以是不同的數(shù)據(jù)結構

一般來說一個普通的鏈表節(jié)點如下：

定義節(jié)點時定義好數(shù)據(jù)段p_value_，這樣的話數(shù)據(jù)結構的實現(xiàn)就會與業(yè)務邏輯結合在一起。

這里不使用模板也實現(xiàn)了預定義獨立于業(yè)務邏輯的鏈表數(shù)據(jù)結構。

鏈表的節(jié)點：// 這是一個雙鏈表

可以把鏈表嵌入到任何一個數(shù)據(jù)結構中，

那如何通過鏈表節(jié)點拿到當前所在結構呢？

通過一個宏來實現(xiàn)：

簡單解釋下這個宏：ptr表示鏈表節(jié)點指針，type是當前節(jié)點數(shù)據(jù)結構類型名，member是鏈表節(jié)點在數(shù)據(jù)結構中的成員名

&((type *)0)->member)把指向地址空間起點的指針（空指針）轉化成指向節(jié)點數(shù)據(jù)結構的指針，然后取鏈表節(jié)點成員名，再取地址，就可以取到鏈表節(jié)點在這個數(shù)據(jù)結構中的偏移量。

ptr是鏈表節(jié)點指針，按(char *)減去偏移量，就可以回退到結構起始位置。再把這個位置轉化成(type *).就取到了指向當前數(shù)據(jù)結構的指針。

看接口甚至可以發(fā)現(xiàn)，當我想把當前數(shù)據(jù)結構從鏈表里刪除的時候，甚至不需要拿到鏈表，而是直接通過list_del(list_head * current_node)函數(shù)傳入當前節(jié)點就可以刪除，靈活的一塌糊涂。

并且提供了遍歷鏈表的接口宏：

每一行代碼都極其簡潔干凈，妙到毫巔！

其他鏈表基礎知識不多贅述。

紅黑樹（見rbtree.h/.c）

與鏈表類似，紅黑樹也使用了內核紅黑樹。

相同的風格，每個節(jié)點只有鏈接指針和節(jié)點顏色字段，而沒有數(shù)據(jù)。

當把紅黑樹node嵌入數(shù)據(jù)結構中之后，使用同樣原理的宏，來獲取節(jié)點所在結構的指針：

比較特別的是，由于節(jié)點不包含數(shù)據(jù)，數(shù)據(jù)結構不知道節(jié)點之間如何比較大小，所以需要用戶自己定義查找、插入函數(shù)，但給出了例子。

消息隊列（見msgqueue.h/.c）

這里實現(xiàn)了一個消息隊列，也是正常的提供一個put接口，供生產者reactor生產數(shù)據(jù)插入消息，一個get接口，傳遞給下游handler消費，消息隊列有消息上限，并提供阻塞和非阻塞兩種模式，阻塞模式下，當消息超過上限生產線成阻塞，等待消息小于上限了再插入。通過條件變量使沒有待處理的消息時，阻塞消費線程，于內核態(tài)等待消息出現(xiàn)。這里的生產者和消費者都是多線程的，所以需要考慮線程安全，消息隊列的常見實現(xiàn)是一個數(shù)據(jù)存儲段，一個鎖，一個條件變量，而workflow中的消息隊列的高妙之處就在于，他有兩個鎖，兩個條件變量，兩個數(shù)據(jù)空間，雙倍快樂。

這里使用了一個小技巧，大幅提升消息隊列性能，兩個數(shù)據(jù)段一個專門用來get，一個專門用來put，兩把鎖兩個條件變量，分別put時候和get時候使用。這樣的好處就是get和put操作之間幾乎互不干擾。put操作不會鎖消費線程。get操作絕大多數(shù)情況下不會鎖生產線程。

只有當get鏈表為空時，才會把put和get全鎖住，對兩個鏈表頭進行交換，極大的減少了生產線程和消費線程之間爭奪鎖產生的相互影響。

這里還有一個點就是消息隊列要求節(jié)點是自帶鏈表字段的，并指定鏈接節(jié)點相對于結構頭的偏移量（linkoff）。所以插進來的節(jié)點msg的結構是poller_result但是實際結構是poller_node強轉過來的，再對比這兩個結構體，發(fā)現(xiàn)前三個成員是一致的，而第四個成員就是鏈接節(jié)點。

線程池（見thrdpool.h/.c)

線程池實現(xiàn)的功能往往是創(chuàng)建一系列工作線程，工作線程執(zhí)行線程回調函數(shù)，從消息隊列中取任務并執(zhí)行，當消息隊列中沒有任務時，等待任務出現(xiàn)。

workflow中的線程池就是這樣一個很標準的線程池，同時很靈活的讓邏輯脫離于線程池，線程回調函數(shù)并非實際要執(zhí)行的邏輯，而是從消息隊列里get出的task，是一個包含了要執(zhí)行的回調和上下文的task，線程回調函數(shù)執(zhí)行了這個task。

這樣實現(xiàn)一個效果，就是可以運行時才動態(tài)決定要執(zhí)行什么邏輯，即每個task可以是不同的任務，靈活度大大提升。

基礎數(shù)據(jù)結構主要就這四種，這里只分析了其設計中比較可圈可點的部分，而沒有仔細講一些簡單的基礎細節(jié)。

純計算任務和Reactor調度層

把阻塞的任務交給epoll去異步調用，計算任務交給線程池去異步執(zhí)行，實現(xiàn)所有任務的異步調度，下面分別看看計算任務和reactor。

純計算任務

WorkFlow由框架統(tǒng)一管理原始任務線程池，單例__ExecManager內有一個單的封裝，優(yōu)雅的實現(xiàn)對線程池的管理。

這一層有三個新概念：

ExecQueue是一個有鎖鏈表隊列;

ExecSession的execute()接口由派生出來的任務自己去定義需要執(zhí)行的邏輯。

Executor類，創(chuàng)建并管理線程池，提供request()方法，request方法把對應任務放入到線程池去執(zhí)行。request的參數(shù)有兩個，分別是當前session和所在的ExecQueue，如果queue里面只有這一個session，則把這個session放入Executor管理的線程池里里執(zhí)行，如果不是首個任務，則只要放入隊列里就行了，線程routine會調度當前隊列中所有的任務進入線程池執(zhí)行，并用ExecQueue中的鎖保持隊列中任務調度的同步性。

Executor::executor_thread_routine是線程執(zhí)行routine，一共做了兩件事：

第一步會遞歸的調度所有當前Queue中的任務進線程池，并用ExecQueue中的鎖保持隊列中任務調度的同步性；

第二步是執(zhí)行當前session，并由session自己保持數(shù)據(jù)同步。

Reactor：

這里主要涉及四個文件poller.h/.c mpoller.h/.c Communicator.h/.cc CommScheduler.h/.cc

其中poller是對epoll的封裝，mpoller又集成多個poller線程；Communicator顧名思義，就是通信器，封裝了mpoller和線程池；CommScheduler是對Communicator的封裝，全局唯一，最后創(chuàng)建在__CommManager中，通過WFGlobal暴露出來。

這一層主要完成了右圖所示的工作，poller線程把epoll事件做初加工處理，生成一個poller_result，設置需要handle的類型，然后把處理結果put()進消息隊列，給工作線程去處理。handler線程等待任務，當隊列里有任務時，根據(jù)任務的operation類型做相應處理。

poller

poller.h/.c提供了poller的創(chuàng)建、啟動、stop、poller_add、poller_del、poller_mod和add_timer的接口。

poller_create創(chuàng)建了poller數(shù)據(jù)結構，分配了poller_node的指針數(shù)組nodes，這里的nodes是一個以fd為下標的數(shù)組，這時候只有一個指針數(shù)組，node還沒有創(chuàng)建，node是在poller_add的時候創(chuàng)建的，創(chuàng)建node的時候會檢查監(jiān)聽的操作是否需要result，需要的話同時分配result空間。但這時候poller線程還沒有跑起來，執(zhí)行poller_start時將poller線程跑起來；poller_add、poller_del、poller_mod分別是epoll的增加節(jié)點、刪除節(jié)點、改變監(jiān)聽事件 三種操作的簡單封裝；add_timer增加時間事件，

前面說過消息隊列里面裝的是poller_result(poller_node)，poller_result里面都會有一個poller_data。

poller_data封裝了需要處理的fd、對應的操作（operation）、上下文（可能是CommService或CommConnEntry）。

poller的核心是poller_thread，poller_start的時候啟動了是一個poller_thread，poller_thread處理的是epoll_event，主流程是一個經典的雙循環(huán)，外層循環(huán)epoll_wait，每次最多處理256個fd，epoll返回后，再根據(jù)每個epoll_event事件的類型，循環(huán)處理每個類型的事件，從枚舉可以看到對當前node的操作有讀、寫、listen、connect、timer等等，不管是什么類型的epoll事件，poller_thread處理的結果會生成一個.poller_result，并把這個結果插入到消息隊列中。

具體的操作非常的多了，不適合靜態(tài)分析，后面再動態(tài)分析請求的全流程。

poller的操作都是線程安全的，mpoller啟動多個線程的時候也可以直接使用。

mpoller

可以看到實際上使用的并不是poller而是mpoller，mpoller是對多線程poller的封裝，一個mpoller包括至少一個poller，實際配幾個線程就創(chuàng)建幾個poller，并統(tǒng)一分配poller_node，所有poller共享poller_node數(shù)組。實際使用的時候可以根據(jù)運算核心數(shù)和業(yè)務邏輯的復雜程度調整poller_thread和handler_thread的配比。mpoller的add、del、mod接口會對傳入的fd對線程數(shù)求模，將fd均勻的分配到各個poller。

關于數(shù)據(jù)同步

可以看到對fd的[]操作并沒有加鎖，以mpoller_add為例

第4行計算index，fd和nthreads都是不會發(fā)生變化，不會修改的，線程之間無沖突，所以不需要加鎖。

第5行由poller_add來保證線程安全，每個poller中都有一個鎖，poller_add、poller_del、poller_mod的操作都是加鎖的，因為這三種操作都可能發(fā)生在不同的線程。

Communicator

Communicator是通訊器，是底層和業(yè)務層的樞紐，創(chuàng)建了mpoller和handler線程池，初始化時候啟動兩個線程池，bind的時候會把服務綁到communicator上，把服務創(chuàng)建的listen_fd放入到poller中開始監(jiān)聽。handler_thread就是在Communicator中啟動的，handler_thread從消息隊列里拿到的是poller_result，handler_thread做的是拿到任務以后根據(jù)poller_result::poller_data::operation類型做相應處理。

相關的結構有：

鏈接：

class WFConnection : public CommConnection 創(chuàng)建的鏈接

對端：

CommTarget通訊目標，封裝了對端的地址、port、超時時間

消息：

很明顯CommMessageIn是一次通信中的輸入消息，CommMessageOut是返回的消息的基類，輸入消息的基類是__poller_message，這里又使用了一個c程序員常用的小技巧，成員char data[0]是一個柔性數(shù)組，把__poller_message變成了一個變長結構體。

結構體中末尾成員是一個長度為0的char數(shù)組，這樣聲明看起來和char *data是一樣的，但是這樣寫相對于char指針有一些優(yōu)勢。

對比如下結構，考慮__poller_message_test和__poller_message有什么區(qū)別

首先，數(shù)組長度是0，說明沒分配空間。所以64位系統(tǒng)中，sizeof(struct __poller_message_test) == 16 而 sizeof(struct __poller_message) == 8。其次，如果使用一個char指針,需要為指針分配內存。而使用data[0]則不需要二次給指針分配內存，直接為結構分配適量大小內存即可，成員data會自動指向結構尾部的下一個字節(jié)。

輸入消息有一個append的虛方法，子類自己去定義如何反序列化，輸出消息有一個encode的虛方法，子類消息自己去定義序列化發(fā)送消息?；恄_poller_message中的函數(shù)指針會被賦值為Communicator::append(const void *buf, size_t *size, poller_message_t *msg)，實際運行時由函數(shù)指針append去調用各子類消息的virtual int append(const void *buf, size_t *size)對消息進行反序列化。

框架內已經定義好一些常用協(xié)議了：

會話：CommSession

CommSession封裝了一次會話所有組成單位，包括輸入/輸出消息、CommConnection、CommTarget

定義了消息的生產方式

服務器：CommService

類圖：

class WFServerBase : protected CommService 服務器的抽象。封裝了服務器地址、監(jiān)聽套接字、活躍鏈接和連接數(shù)、服務器參數(shù)。

基類定義了newsession、newconnect接口。WFServerBase類中實現(xiàn)了服務啟動start()、停止stop()、創(chuàng)建/刪除鏈接newconnect（）。

WFServer是一個模板類，模板參數(shù)是輸入輸出消息類型，可以實例化為各種類型的服務器，不同類型的服務器就是消息類型不同的服務器實例化，因為不同類型服務器實例消息類型不同，處理消息方式也不同，WFServer中保存了處理消息方式的回調——processer，并在服務創(chuàng)建的時候初始化。在WFServer中定義session創(chuàng)建方式new_session()的時候，會用processer來創(chuàng)建task，process實際上是task的處理方式。

服務Start()的時候會被bind()到全局的Communicator上，包括創(chuàng)建fd、bind、listen、放入epoll監(jiān)聽，成為epoll監(jiān)聽的第一個fd。服務實際上是交給Communicator創(chuàng)建的handler_thread線程池來驅動起來的。

Entry:CommConnEntry

打包了所有一次會話需要的上下文，包括poller、servide、session、target、socket等，處理accept事件（handle_listen_result）的時候由Communicator::accept_conn創(chuàng)建，創(chuàng)建后放在poller_data中，mpoller_add監(jiān)聽

Communicator:

有了上面這些基礎結構，Communicator就是一個完全體了，Communicator初始化的時候，啟動了poller_thread、handler_thread驅動服務進行消息處理。

以示例代碼的hello_world程序為例，觀察一次網絡請求過程，看看poller_thread和handler_thread分別都做了什么。

從hello_world啟服到線程工作：

這里特別看一下poller_add的時候創(chuàng)建了poller_node實體,poller_node中有一個成員struct __poller_node *res，__poller_data_get_event()的時候會返回一個bool值，表示是否需要創(chuàng)建res?？梢钥吹讲僮黝愋蜑閘isten的情況。是需要res的。

經過這個過程，服務器就啟動開始接受請求了，service創(chuàng)建listen_fd交由poller管理，當監(jiān)聽到有客戶端鏈接時，accept+read。下面分析接收到一個請求時，poller_thread和handler_thread分別做了什么。

poller_thread知道listenfd可讀，則accept一個readfd，創(chuàng)建了對端target，把這個poller_result(poller_node)放進消息隊列。

handler_thread拿到這個poller_result之后，主要是創(chuàng)建了完整的CommConnEntry，并把負責read的poller_node放入epoll監(jiān)聽，等待內核緩沖區(qū)有數(shù)據(jù)可讀。

這里有個細節(jié)，readfd是無阻塞模式，因為使用了epoll的邊緣觸發(fā)模式，即每個fd的狀態(tài)變化只通知一次，這樣的話需要把readfd上的數(shù)據(jù)全讀完，所以readfd必須設置成無阻塞模式，否則循環(huán)讀到最后肯定會被阻塞。

如果遇到errorno==EAGAIN則直接返回，因為對于fd阻塞調用eagain表示提示重試，對于非阻塞fd，errorno==EAGAIN則表示緩沖區(qū)已經寫滿，直接return本次處理結束。

readfd放入epoll之后，readfd上有數(shù)據(jù)到來后會被操作系統(tǒng)拷進內核緩沖區(qū)，然后epoll提示readfd可讀。poller_thread會進入處理可讀事件(handle_read)。

poller_thread對可讀事件的處理主要是把字節(jié)流讀出來，并反序列化，放入隊列提供給handler_thread，handler_thread調service處理業(yè)務邏輯。

handler對收到的消息的處理分兩種情況，如果是服務端，當做請求處理，如果是客戶端，當回復處理，所以hello_world程序進入請求處理流程。

服務器對請求的處理是創(chuàng)建服務對應類型的CommRequest，helloworld中實際是執(zhí)行了一個WFHttpServerTask。

繼承關系：WFHttpServerTask——>WFServerTask——>WFNetworkTask——>CommRequest——>SubTask，CommSession。

SubTask和CommSession后面再仔細分析，這里先從字面理解，SubTask就是任務，就是處理自定義邏輯的過程，CommSession是會話。那handle的時候會先調用當前Task的processor.dispatch()執(zhí)行任務，任務執(zhí)行完自動subtask_done()的時候會調用scheduler->reply()，將結果返回 Send_message()?？梢钥吹絊end_message是先嘗試同步寫，如果同步寫失敗了，再嘗試異步寫，異步寫的過程就是先把文件描述符加入epoll監(jiān)聽，等待可寫信號出現(xiàn)后，再寫入。寫的時候使用iovec，聚集寫盡量減少拷貝次數(shù)。

至此poller事件各種operation的處理，已經分析過PD_OP_READ、PD_OP_WRITE、PD_OP_LISTEN，再通過wget看一下PD_OP_CONNECT。

connect主要是處理客戶端鏈接服務端時，服務端無法立刻建立鏈接時的等待，異步等待屏蔽等待時間。

request的時候會優(yōu)先檢查目標上有沒有idle鏈接，如果有的話直接復用，如果沒有會創(chuàng)建connect，conn_fd是非阻塞的，operation設置為PD_OP_CONNECT，放在epoll中管理，等待fd可用。

可以看到，是一個簡單的發(fā)送請求，等待結果的過程。

poller事件共有10種operation，這里分析過讀、寫、connect、listen四種流程，PD_OP_SSL_ACCEPT、PD_OP_SSL_CONNECT、PD_OP_SSL_SHUTDOWN三個只是使用openssl庫時的創(chuàng)建和關閉鏈接。還有另外兩種事件：PD_OP_EVENT、PD_OP_NOTIFY，這兩種分別是linux和mac環(huán)境下處理異步文件I/O用的。

異步文件I/O：

TODO

任務組織調度層

下面分析任務線程是如何執(zhí)行任務的邏輯。這個層次有兩個核心基礎概念，一個是任務的抽象，一個是會話(session)的抽象，二者是所有執(zhí)行邏輯的祖爺爺和祖奶奶。

任務：

前面看到對于請求的處理，實際是執(zhí)行了CommRequest，CommRequest既是一個SubTask又是一個CommSession，最后是通過執(zhí)行的是SubTask的接口dispatch()執(zhí)行起來的，這里最重要的概念——子任務。workflow里面所有的邏輯，最后都是通過子任務執(zhí)行起來的；子任務又可以通過各種組合關系，串并聯(lián)的組織起來。

這里有四個重要的基本元素：

1，SubTask——子任務，是一切任務的祖先。

2、ParallelTask——并行任務，并行任務里面管理SubTask數(shù)組，啟動時會把自己管理的SubTask一個一個全部dispatch一遍。

3、SeriesWork——串聯(lián)工作組，里面管理了一個數(shù)組的子任務，逐個執(zhí)行。

4、ParallelWork——并聯(lián)工作組，里面管理了一個SeriesWork數(shù)組，其本身的祖先是一個SubTask，所以他可以被SeriesWork管理。

這樣就實現(xiàn)了任務的串并聯(lián)執(zhí)行甚至以DAG的形式復合。

下面逐一分析：

SubTask是一切執(zhí)行任務的祖先，不同的任務實現(xiàn)，實現(xiàn)不同的dispatch()和done()接口，提供兩個接口留給用戶自定義：

1、dispatch()接口 就是執(zhí)行任務，用戶任務自定義執(zhí)行邏輯，而在執(zhí)行結束后，必須調用subtask_done()。

2、done()接口 在任務邏輯執(zhí)行結束后，由subtask_done()調起done()，這個接口是用戶自定義的結束回調，在done()接口里面回收資源，銷毀任務。done()函數(shù)還會返回一個子任務的指針，當當前任務執(zhí)行完還要執(zhí)行下一個任務的時候，返回下一個任務，如果沒有下一個任務，則返回nullptr。為什么這么約定呢？這需要看一下subtask_done()函數(shù)的工作方式。

需要知道成員變量的意思才能明白調度方式：

pointer 一般指向當前所在SeriesWork，SubWork最后也是放在SeriesWork之中啟動起來的；

parent 當一個子任務被ParallelTask任務管理的時候，parent指向被管理的并行任務。

entry 指向待執(zhí)行任務數(shù)組的首位。

subtask_done()：仔細解讀一下subtask_done()的工作方式：

可以看到先保存了當前任務的parent和entry，然后直接調用了當前任務的done()接口。如果又返回了一個子任務，則調用新任務的dispatch()，使其運行起來，dispatch()到最后必然又會調用新任務的subtask_done()；從而遞歸執(zhí)行這條線上所有任務，直至done()不會再返回任務；當不再返回任務時，說明parent的孩子都執(zhí)行完，就可以繼續(xù)再往上執(zhí)行(parent也是一個SubTask)，直至根任務執(zhí)行完。

ParallelTask:

ParallelTask是SubTask的兒子，結構很簡單，管理了一個SubTask數(shù)組，ParallelTask::dispatch()的時候會把數(shù)組內管理的所有SubTask逐一dispatch()一遍，這樣的話就實現(xiàn)了同級任務的并列執(zhí)行，特別注意并列執(zhí)行不一定是并行，是否并行取決于調度。任務本身是順序dispatch()的，如果dispatch調度的時候把任務放入線程池執(zhí)行任務就是并行的。

SeriesWork：

SeriesWork是一個有鎖的線程安全隊列，隊列中存儲了需要按順序執(zhí)行的SubTask，預分配4個空間，如果入隊時隊列已滿，則像vector一樣拓展二倍空間。

SubTask都是放到SeriesWork中執(zhí)行的。SeriesWork是怎么調度執(zhí)行任務的？啟動函數(shù)Start()，會從第一個SubTask開始dispatch()，可以看到多數(shù)任務Task的done()的實現(xiàn)都是返回return series->pop();意思就是當前任務執(zhí)行完了，返回當前所在的SeriesWork中的下一個任務，繼續(xù)執(zhí)行，直至所有任務執(zhí)行完。

注意SeriesWork本身不是一個SubTask，所以無法被SeriesWork管理。

ParallelWork：

ParallelWork稍微復雜一點

繼承關系：ParallelWork——>ParallelTask——>SubTask

可見：1、ParallelWork是一個SubTask，所以可以被SeriesWork管理；2、ParallelWork同時也是一個ParallelTask，管理了一個數(shù)組的SubTask；3、ParallelWork管理了一個SeriesWork數(shù)組，這個數(shù)組的長度和SubTask數(shù)組的長度相同。并且讓SubTask指向同索引SeriesWork的首個SubTask。

ParallelWork是怎樣啟動和調度任務的：

ParallelWork本身是一個SubTask，所以啟動時把他放入一個SeriesWork，作為SeriesWork的firsttask被調起dispatch()；然后ParallelWork本身是一個ParallelTask，dispatch的時候會把其下管理的所有的SubTask逐個啟動dispatch()；如圖，SubTask指向的實際是管理的SeriesWork的first Task，所以實際上相當于啟動了管理的所有SeriesWork。

這四個結構就是整個任務調度的基石，所有的邏輯都是作為任務執(zhí)行起來的。并行任務管理串行任務，串行任務管理SubTask（并行任務也是SubTask），這套設定使任務可以自由復合DAG復合。

這時可以明白這個框架名字所謂WorkFlow，其核心就是組織任務的執(zhí)行流，所有的執(zhí)行邏輯都是任務。

會話(session)：

想要執(zhí)行的邏輯，通過成為SubTask可以啟動起來，并按一定的順序調度，那具體做的事，則被抽象為會話。

基礎session有四種：CommSession、ExecSession、IOSession、SleepSession，分別代表網絡操作、運算操作、I/O操作、睡眠操作，session都需要實現(xiàn)handle()接口，所有最后執(zhí)行的任務都是這四種操作派生出來的。

SubTask這個大渣男分別和四種session結合生成了CommRequest、ExecRequest、SleepRequest、IORequest，使得所有的request都可以被作為子任務調度，都有state和error。

四種request分別派生出了WFNetWorkTask、WFThreadTask、WFTimerTask、WFFileTask。其中WFNetWorkTask和WFThreadTask都是兩個參數(shù)的模板類。對通信任務來說，參數(shù)是請求消息和回復消息，對于計算任務來說參數(shù)是輸入和輸出，WFReduceTask、WFSortTask、WFMergeTask是不用參數(shù)的的實例化，WFHttpTask、WFRedisTask、WFMysqlTask、WFKafkaTask只不過是不同協(xié)議的WFNetWorkTask的實例化。

CommRequest派生了WFNetworkTask；ExecRequest派生了WFThreadTask，二者都加入了輸入輸出模板參數(shù)，和一些控制參數(shù)，提供了方便的啟動多線程任務和網絡任務的方式。更有WFMultiThreadTask任務，批量管理多線程任務。

這里還有一個WFTimerTask，實現(xiàn)了不占線程的定時功能.。

WFTimerTask：

WFTimerTask可以讓任務休眠一定時長后執(zhí)行，不占線程，達到時長之后返回執(zhí)行回調，就是定時任務。

如果一個WFTimerTask被直接start()，則創(chuàng)建一個SeriesWork，并dispatch()起來，如果是串在其他的SeriesWork，當執(zhí)行到這個task的時候直接dispatch()。

當SleepRequest被dispatch()時候，實際是調用當前scheduler(即communicator)的sleep()，實際是取出當前WFTimerTask的休眠時間，然后創(chuàng)建一個定時任務mpoller_add_timer交給epoll管理，等epoll提示時間到了，再切回來執(zhí)行。

層次結構：

借用一張官圖非常清楚的表達清楚任務之間的層次關系。

用戶接口

至此，底層支持都分析過了，下面看看通過這些底層結構可以組織出什么花樣。

其他Tasks

WFCounterTask：

CounterTask是一個計數(shù)器Task，任務里保存了一個原子的unsigned用來計數(shù)，初始化時候傳入需要記的個數(shù)，每次任務被dispatch()的時候，計數(shù)器減一，直到計數(shù)器為0時，執(zhí)行回調，配合一個阻塞信號量，可以實現(xiàn)一批并行任務的統(tǒng)一等待，如:WaitGroup。

可能是覺得手動創(chuàng)建CounterTask不夠優(yōu)雅，框架還創(chuàng)建了CounterTask管理器，用一個紅黑樹以名字為key統(tǒng)一管理CounterTask，可以通過名字全局操作CounterTask。

WaitGroup

既然說到了就順便說一下WaitGroup。

WaitGroup實現(xiàn)了阻塞等待多個任務完成的效果。

WaitGroup由一個原子的等待個數(shù)，一個WFCounterTask和一個std::future組成。構造時創(chuàng)建一個std::promise，并綁定到future上；創(chuàng)建一個計數(shù)1的CounterTask并注冊回調，回調中時給promise->setvalue()。

每次調用done會給剩余個數(shù)減一，當減完時，counter->done()，這時回調會告訴futrue，所有任務都完成了，阻塞結束。

WFGraphNode和WFGraphTask：

WFGraphTask實現(xiàn)了將任務迅速的組織成有向無環(huán)圖的方法，一個WFGraphTask管理了一張由多個WFGraphNode組成。

WFGraphNode是一個WFCounterTask，并加入了一個WFGraphNode*列表：follower，follower表達了鄰接關系，保存的就是依賴當前任務的下游節(jié)點。因為是counter任務，所以具有計數(shù)的功能，記的數(shù)就是當前Node的入度。在當前任務執(zhí)行完之后，會把所有下游節(jié)點都dispatch（計數(shù)）一次，當計數(shù)減少到0時，說明當前Node所有依賴已經完成了，就把當前graphNode上掛的SeriesWork執(zhí)行起來。

依賴處理：當一個node1依賴Node2時候，Node2的下游節(jié)點列表里加入Node1，Node1的入度自增。

執(zhí)行處理：當Node2執(zhí)行完，Node1的入度減一。

框架的重載了GraphNode的自增運算符和大于號、小于號，自增運算符返回Node本身。大于號、小于號運算符調用依賴關系函數(shù)。從而很形象的可以通過如下語法表達節(jié)點之間的依賴關系：

是不是很秀？簡直妙不可言

再說一個細節(jié)：DAG建立起來了，但是Node上是怎么掛的任務呢？

答：創(chuàng)建WFGraphNode通過統(tǒng)一接口：WFGraphNode& WFGraphTask::create_graph_node(SubTask *task)，創(chuàng)建的時候傳入你想要執(zhí)行的任務，然后把要執(zhí)行的任務和當前Counter任務串在一個Series里面。當當前Node計數(shù)器第一次變0的時候，會調到Done()，看一下關鍵的done()實現(xiàn)：

首次done()的時候不析構，并將狀態(tài)置為下次進來析構（value賦1&&user_data非空)。

然后將本series里面要執(zhí)行的用戶任務執(zhí)行起來。當用戶任務執(zhí)行完，會再次執(zhí)行到GraphNode->Done();這時侯，Node析構，并將所有follower->dispatch()起來。這就是圖任務的整體執(zhí)行路徑。

WFRepeaterTask：

這是一個遞歸Task，繼承自GenericTask，也就是說啟動時，會創(chuàng)建一個Series，并把Series啟動起來。創(chuàng)建的時候傳入創(chuàng)建任務的回調Create，在dispatch()得時候，往當前Series里傳入兩個任務，一個是Create回調創(chuàng)建出來的新任務，一個是當前任務。這樣的話，順序任務的調度就變成：執(zhí)行任務—》創(chuàng)建任務—》執(zhí)行任務。。。

WFConditional：

WFConditional是條件任務包裝器，可以把其他任務包裝成條件任務，通過一個atomic變量實現(xiàn)。新增加一個signal接口，當dispatch和signal都執(zhí)行后，任務會被執(zhí)行。原理：當任務被dispatch或者signal時，都會去設置原子bool的值，并檢查狀態(tài)，如果設置過狀態(tài)，就調起任務，可見第一次不會調起，第二次才會調起任務。

為了避免發(fā)送signal者持有條件任務的裸指針，框架還提供了全局的命名的條件任務，發(fā)送者可以根據(jù)名字給conditional發(fā)signal，內部是一個觀察者模式，以cond的名字為key構建了一個紅黑樹管理，當signal某個key的時候，找到對應的條件任務發(fā)送signal()。

WFModuleTask：

WFModuleTask提供了一個模塊級的封裝，可以把一系列任務封裝到一個模塊里，可以注冊一個模塊的回調函數(shù)。WFModuleTask本質上還是一個SeriesWork，把一系列任務封裝在一起，降低功能任務之間的耦合程度。

服務

基于workflow框架我們可以迅速的構建http服務器，只需要幾行代碼：

可以看到構造一個WFHttpServer，只要傳入一個處理WFHttpTask的回調函數(shù)即可。

下面分別看 WFHttpServer 、WFServerTask

WFHttpServer

首先WFHttpServer是WFServer的http消息時的特化版本。WFServer在BaseServer的基礎上增加了輸入輸出模板參數(shù)，并增加了一個可以處理WFNetworkTask的回調函數(shù)，同時重寫了new_session方法；

poller在create_message的時候會調到new_session，創(chuàng)建WFServerTask；

Communicator并不知道Service是什么類型的service，在create_message的時候不管是什么類型的service，都調用service對應的new_session接口去生產session交給Poller去生成任務交由線程池執(zhí)行。

WFServerTask

WFServerTask繼承自WFHttpTask，WFServerTask內定義了兩個局部類，Processor和Series。

前者Processor保存著服務初始化時傳入的回調和當前WFServerTask的指針，dispatch時執(zhí)行回調處理當前任務。

后者Series本質上是一個SeriesWork，把Processor和當前任務串起來，并先執(zhí)行Processor，最后執(zhí)行當前WFServerTask，當前任務負責reply。同時負責引用計數(shù)，讓service知道有多少任務在引用。

服務小結

session是被動產生的，服務是靜態(tài)定義的，服務定義了自己的服務類型、和產生任務的方法、處理任務的回調等等，然后在服務啟動的時候綁定地址創(chuàng)建fd，把自己綁定到Communicator上，交給Reactor去調度。