workflow的任務(wù)模型

今天，想聊聊workflow這個開源項目。

關(guān)于workflow，我之前特意寫過一篇文章【推薦學(xué)習(xí)這個C++開源項目】。

今天還是想再啰嗦啰嗦，因為自己這一年也在帶團(tuán)隊從0到1做項目，需要負(fù)責(zé)整個項目的架構(gòu)設(shè)計、接口設(shè)計、模塊劃分等工作。

做了一段時間后再回過頭復(fù)盤一下，深知架構(gòu)設(shè)計、接口設(shè)計的重要性，也感受到了架構(gòu)設(shè)計的困難程度，編碼和設(shè)計相比，真的容易的多了。

然后自己又回頭來研究了一下workflow，想著學(xué)習(xí)下其他項目的設(shè)計理念，隨著自己研究的越來越深入，越來越感覺它的高端，對外暴露特別簡單的接口卻能完成非常復(fù)雜的功能。

上篇文章是基礎(chǔ)篇，主要向大家普及一下workflow的特點和作用，感興趣的朋友可以移步到那里哈。

本篇文章是進(jìn)階篇，主要就是想介紹下workflow的任務(wù)模型，其他的框架一般只能處理普通的網(wǎng)絡(luò)通信，而workflow卻特別適用于通信與計算關(guān)系很復(fù)雜的應(yīng)用。其實我最感興趣的是它的內(nèi)存管理機(jī)制，下面也會詳細(xì)介紹。

優(yōu)秀的系統(tǒng)設(shè)計

在作者的設(shè)計理念中：程序 = 協(xié)議 + 算法 + 任務(wù)流。

**協(xié)議：**就是指通用的網(wǎng)絡(luò)協(xié)議，比如http、redis等，當(dāng)然還可以自定義網(wǎng)絡(luò)協(xié)議，這里只需要提供序列化和反序列化函數(shù)就可以達(dá)到想要的效果。

算法： workflow提供了一些通用的算法，比如sort、merge、reduce等，當(dāng)然還可以自定義算法，用過C++ STL的朋友應(yīng)該都知道如何自定義算法吧，在workflow中，任何復(fù)雜的計算都應(yīng)該包裝成算法。

**任務(wù)流：**我認(rèn)為這應(yīng)該就是整個系統(tǒng)設(shè)計的核心，通過任務(wù)流來抽象封裝實際的業(yè)務(wù)邏輯，就是把開發(fā)好的協(xié)議和算法組成一個任務(wù)流程圖，然后調(diào)度執(zhí)行這個圖。

任務(wù)流

這里多聊聊任務(wù)流，在workflow中，一切業(yè)務(wù)邏輯皆是任務(wù)，多個任務(wù)會組成任務(wù)流，任務(wù)流可組成圖，這個圖可能是串聯(lián)圖，可能是并聯(lián)圖，也有可能是串并聯(lián)圖，類似于這種：

也可能是這種復(fù)雜的DAG圖：

圖的層次結(jié)構(gòu)可以由用戶自定義，框架也是支持動態(tài)地創(chuàng)建任務(wù)流。

引用作者的一句話：

如果把業(yè)務(wù)邏輯想象成用設(shè)計好的電子元件搭建電路，那么每個電子元件內(nèi)部可能又是一個復(fù)雜電路。

workflow對任務(wù)做了很好的抽象和封裝。整個系統(tǒng)包含6種基礎(chǔ)任務(wù)：通訊、文件IO、CPU、定時器、計數(shù)器。

workflow提供了任務(wù)工廠，所有的任務(wù)都由任務(wù)工廠產(chǎn)生，并且會自動回收。

大多數(shù)情況下，通過任務(wù)工廠創(chuàng)建的任務(wù)都是一個復(fù)合任務(wù)，但用戶并不感知。例如一個http請求，可能包含很多次異步過程（DNS，重定向），內(nèi)部有很多復(fù)雜的任務(wù)，但對用戶來講，這就是一次簡單的通信任務(wù)。

哪有什么歲月靜好，只不過是有人替你負(fù)重前行。workflow的一大特點就是接口暴露的特別簡潔，非常方便用戶的接入，外部接入如此簡單，肯定是將很多組合的邏輯都放在了內(nèi)部，但其實workflow項目內(nèi)部代碼結(jié)構(gòu)層次非常簡潔清晰，感興趣的朋友可以自己研究研究哈。

內(nèi)存管理機(jī)制

還有就是項目的內(nèi)存管理機(jī)制，workflow整個項目都遵循著誰申請誰釋放的原則，內(nèi)部申請的內(nèi)存不需要外部顯式釋放，內(nèi)部會自動回收內(nèi)存。

而且整個項目都沒有使用shared_ptr，那多個對象共同使用同一塊裸內(nèi)存，workflow是怎么處理的呢？

內(nèi)部為這種需要共享的對象各自維護(hù)一個引用計數(shù)，手動incref和decref，至于為什么要這樣做，可以看看workflow美女架構(gòu)師的回答【https://www.zhihu.com/question/33084543/answer/2209929271】。

我總結(jié)了一下：

• shared_ptr是非侵入式指針，一層包一層，而且為了保持shared_ptr覆蓋對象整個生命周期，每次傳遞時都必須帶著智能指針模板，使用具有傳染性，而且也比較麻煩。
• shared_ptr引用計數(shù)的內(nèi)存區(qū)域和數(shù)據(jù)區(qū)域不一致，不連續(xù)，緩存失效可能導(dǎo)致性能問題，盡管有make_shared，但還是容易用錯。
• 手動為對象做incref和decref，使用起來更靈活，可以明確在引用計數(shù)為固定數(shù)字時做一些自定義操作，而且方便調(diào)試。因為手動管理對象的引用計數(shù)，就會要求開發(fā)者明晰對象的生命周期，明確什么時候該使用對象，什么時候該釋放對象。
• 如果使用shared_ptr可能會激起開發(fā)者的惰性，反正也不需要管理內(nèi)存啦，就無腦使用shared_ptr唄，最后出現(xiàn)問題時調(diào)試起來也比較困難。

那再深入源碼中研究研究，看看workflow是如何做到把對象指針給到外部后，內(nèi)部還自動回收的。

拿WFClientTask舉例說明一下，workflow中所有的Task都是通過Factory創(chuàng)建：

static WFHttpTask *create_http_task(const std::string& url,
                    int redirect_max,
                    int retry_max,
                    http_callback_t callback);


using WFHttpTask = WFNetworkTask;

template <class REQ, class RESP>
class WFClientTask : public WFNetworkTask {};

注意，create參數(shù)里有一個callback，workflow一定會執(zhí)行這個callback，然后內(nèi)部回收掉該WFClientTask占用的內(nèi)存，任何任務(wù)的生命周期都是從創(chuàng)建到callback函數(shù)結(jié)束。

它是怎么做到的？繼續(xù)看下WFClientTask的繼承層次結(jié)構(gòu)：

template <class REQ, class RESP>
class WFClientTask : public WFNetworkTask {};


template<class REQ, class RESP>
class WFNetworkTask : public CommRequest {};


class CommRequest : public SubTask, public CommSession {};


class SubTask {
public:
  virtual void dispatch() = 0;
private:
  virtual SubTask *done() = 0;
protected:
  void subtask_done();
};

WFClientTask繼承于WFNetworkTask，WFNetworkTask又繼承于SubTask。

SubTask內(nèi)部有subtask_done()方法，看下它的實現(xiàn)：

void SubTask::subtask_done() {
  SubTask *cur = this;
  ParallelTask *parent;
  SubTask **entry;


  while (1) {
    parent = cur->parent;
    entry = cur->entry;
    cur = cur->done();
    if (cur) {
      cur->parent = parent;
      cur->entry = entry;
      if (parent)
        *entry = cur;
      cur->dispatch();
    }
    else if (parent) {
      if (__sync_sub_and_fetch(&parent->nleft, 1) == 0) {
        cur = parent;
        continue;
      }
    }
    break;
  }
}

subtask_done()方法中會調(diào)用它的done()方法，然而這幾個方法都是virtual方法，看看WFClientTask是怎么重寫它們的：

template <class REQ, class RESP>
class WFClientTask : public WFNetworkTask<REQ, RESP> {
protected:
  virtual SubTask *done() {
    SeriesWork *series = series_of(this);
    if (this->state == WFT_STATE_SYS_ERROR && this->error < 0) {
      this->state = WFT_STATE_SSL_ERROR;
      this->error = -this->error;
    }
    if (this->callback)
      this->callback(this);
    delete this;
    return series->pop();
  }
};

子類重寫了done()方法，可以看到在它的實現(xiàn)里，觸發(fā)了callback，然后調(diào)用了delete this，釋放掉了當(dāng)前占用的這塊內(nèi)存。

那誰調(diào)用的done()，可以看下上面的代碼，subtask_done()會觸發(fā)done()，那誰觸發(fā)的subtask_done()：

void CommRequest::dispatch() {
  if (this->scheduler->request(this, this->object, this->wait_timeout,
                 &this->target) < 0) {
    this->state = CS_STATE_ERROR;
    this->error = errno;
    if (errno != ETIMEDOUT)
      this->timeout_reason = TOR_NOT_TIMEOUT;
    else
      this->timeout_reason = TOR_WAIT_TIMEOUT;
    this->subtask_done();
  }
}

可以看到，dispatch()里觸發(fā)了subtask_done()，那誰觸發(fā)的dispatch()：

template<class REQ, class RESP>
class WFNetworkTask : public CommRequest {
public:
  /* start(), dismiss() are for client tasks only. */
  void start() {
    assert(!series_of(this));
    Workflow::start_series_work(this, nullptr);
  }
};


inline void
Workflow::start_series_work(SubTask *first, SubTask *last,
              series_callback_t callback) {
  SeriesWork *series = new SeriesWork(first, std::move(callback));
  series->set_last_task(last);
  first->dispatch();
}

這里可以看到，Task的start()方法觸發(fā)start_series_work()，進(jìn)而觸發(fā)dispatch()方法。

總結(jié)一下：

● 步驟一

通過工廠方法創(chuàng)建WFClientTask，同時設(shè)置callback；

● 步驟二

外部調(diào)用start()方法，start()中調(diào)用Workflow::start_series_work()方法；

● 步驟三

start_series_work()中調(diào)用SubTask的dispatch()方法，這個dispatch()方法由SubTask的子類CommRequest（WFClientTask的父類）實現(xiàn)；

● 步驟四

dispatch()方法在異步操作結(jié)束后會觸發(fā)subtask_done()方法；

● 步驟五

subtask_done()方法內(nèi)會觸發(fā)done()方法；

● 步驟六

done()方法內(nèi)會觸發(fā)callback，然后delete this；

● 步驟七

內(nèi)存釋放完成。

其實這塊可以猜到，想要銷毀自己的內(nèi)存，基本上也就delete this這個方法。

然而我認(rèn)為這中間調(diào)用的思想和過程才是我們需要重點關(guān)注和學(xué)習(xí)的，遠(yuǎn)不止我上面描述的這么簡單，感興趣的朋友可以自己去研究研究哈。

關(guān)于workflow還有很多優(yōu)點，這里就不一一列舉啦，它的源碼也值得我們CPP開發(fā)者學(xué)習(xí)和進(jìn)階，具體可以看我之前的文章。

發(fā)現(xiàn)workflow團(tuán)隊對這個項目相當(dāng)重視，還特意建了個QQ交流群（群號碼是618773193），對此項目有任何問題都可以在這個群里探討，也方便了我們學(xué)習(xí)，真的不錯。項目地址如下：https://github.com/sogou/workflow，也可以點擊閱讀原文直達(dá)。

在訪問GitHub遇到困難時，可使用他們的Gitee官方倉庫：https://gitee.com/sogou/workflow。

項目也特別實用，據(jù)說搜狗內(nèi)外很多團(tuán)隊都在使用workflow。感覺這個項目值得學(xué)習(xí)的話就給人家個star，不要白嫖哈，對項目團(tuán)隊來說也是一種認(rèn)可和鼓勵。