Node.js 代碼閱讀筆記系列 — process.nextTick() 的實現

時間 2019-11-15

標籤 node.js node 代碼閱讀筆記系列 process.nexttick process nexttick 實現欄目 Node.js 简体版

原文原文鏈接

process.nextTick()

process 是一個全局對象，它提供了當前 Node.js 線程的相關信息和一些控制方法。由於 process 掛載了太多屬性和方法，這篇文章先從 process.nextTick() 開始吧。javascript

setupNextTick

function setupNextTick() {
    // 設置 Promise 模塊的調度方法
  const promises = require('internal/process/promises');
  const emitPendingUnhandledRejections = promises.setup(scheduleMicrotasks);

  var nextTickQueue = [];
  // microtask 標記
  var microtasksScheduled = false;

  // 接收 V8 micro task 隊列的運行的對象.
  var _runMicrotasks = {};

  // 這裏 kIndex kLength 是一個約定的 Environment::TickInfo 的 index 和 length 的索引 
  var kIndex = 0;
  var kLength = 1;

  process.nextTick = nextTick;
  // Needs to be accessible from beyond this scope.
  process._tickCallback = _tickCallback;
  process._tickDomainCallback = _tickDomainCallback;

  // 經過 process._setupNextTick 註冊 _tickCallback， 獲取 _runMicrotasks
  // `tickInfo` 也接收了 `process._setupNextTick()` 的返回參數，經過 `tickInfo` 能使 C++ 模塊能訪問到 nextTick 隊列的狀態。

  const tickInfo = process._setupNextTick(_tickCallback, _runMicrotasks);

    // 接收驅動 V8's micro task 隊列的方法
  _runMicrotasks = _runMicrotasks.runMicrotasks;

    function tickDone() {
    ...
    }

    function scheduleMicrotasks() {
    ...
    }

    function runMicrotasksCallback() {
    ...
    }

    function _combinedTickCallback() {
    ...
    }

    function _tickCallback() {
    ...
    }

    function _tickDomainCallback() {
    ...
    }

    function nextTick() {
    ...
    }
}複製代碼

_tickCallback & _tickDomainCallback

這裏兩個大致都是執行必定數量（最大 1e4 ）的數量 callbacks，前者不須要執行 domain 進入上下文。java

function _tickCallback() {
    var callback, args, tock;

    do {
      while (tickInfo[kIndex] < tickInfo[kLength]) {
        tock = nextTickQueue[tickInfo[kIndex]++];
        callback = tock.callback;
        args = tock.args;
        _combinedTickCallback(args, callback);
        if (kMaxCallbacksPerLoop < tickInfo[kIndex])
          tickDone();
      }
      tickDone();
      // V8 promise microtasks
      _runMicrotasks();
      emitPendingUnhandledRejections();
    } while (tickInfo[kLength] !== 0);
  }複製代碼

_combinedTickCallback

這裏的參數處理仍是體現了 Nodejs 中貫穿的性能追求以及 80/20 的理念。node

function _combinedTickCallback(args, callback) {
    if (args === undefined) {
      callback();
    } else {
      switch (args.length) {
        case 1:
          callback(args[0]);
          break;
        case 2:
          callback(args[0], args[1]);
          break;
        case 3:
          callback(args[0], args[1], args[2]);
          break;
        default:
          callback.apply(null, args);
      }
    }
  }複製代碼

tickDone

執行正常的清理操做，刪除剛執行完的 callback 或者清空隊列。c++

function tickDone() {
    if (tickInfo[kLength] !== 0) {
      if (tickInfo[kLength] <= tickInfo[kIndex]) {
        nextTickQueue = [];
        tickInfo[kLength] = 0;
      } else {
           // 推出隊列的首個元素
        nextTickQueue.splice(0, tickInfo[kIndex]);
        tickInfo[kLength] = nextTickQueue.length;
      }
    }
    tickInfo[kIndex] = 0;
  }複製代碼

scheduleMicrotasks

再回頭看一下 setupNextTick 中的 Promise setup 那段 promises.setup(scheduleMicrotasks)。下面咱們來看看 scheduleMicrotasks.git

function setupNextTick() {
  const promises = require('internal/process/promises');
  const emitPendingUnhandledRejections = promises.setup(scheduleMicrotasks);
  var microtasksScheduled = false;
  ...
}複製代碼

先判斷 microtasksScheduled ，若是爲 false 就會執行到給 nextTickQueue 添加一個新的節點，callback 爲 runMicrotasksCallback。接着 tickInfo[kLength] 增長 1 並將 microtasksScheduled 設置爲 true , 確保在未執行 microtask 以前不會重複執行。github

function scheduleMicrotasks() {
    if (microtasksScheduled)
      return;

    nextTickQueue.push({
      callback: runMicrotasksCallback,
      domain: null
    });

    tickInfo[kLength]++;
    microtasksScheduled = true;
  }複製代碼

runMicrotasksCallback

能夠看到這裏與以前對應的，這裏首先執行 microtasksScheduled = false, 接着調用 _runMicrotasks。在 nextTickQueue 以及 Promise 還有 Listeners 時繼續調用 scheduleMicrotasks 來向 nextTickQueue 添加 callback。bootstrap

function runMicrotasksCallback() {
    microtasksScheduled = false;
    _runMicrotasks();

    if (tickInfo[kIndex] < tickInfo[kLength] ||
        emitPendingUnhandledRejections())
      scheduleMicrotasks();
  }複製代碼

SetupNextTick

經過上面的 JS 部分咱們瞭解到，process.nextTick, Microtasks 以及 Promise 的 callback 都是經過一個隊列 nextTickQueue 調度, 而這一切都是從
_tickCallback （ _tickDomainCallback ）開始的。api

// src/node.cc
void SetupNextTick(const FunctionCallbackInfo<Value>& args) {
  Environment* env = Environment::GetCurrent(args);

  CHECK(args[0]->IsFunction());
  CHECK(args[1]->IsObject());

  // 將以前的 `_tickCallback` 設置到環境變量中 tick_callback_function
  env->set_tick_callback_function(args[0].As<Function>());

    // 將傳過來的 _runMicrotasks ({}) 對象添加 runMicrotasks 方法
  env->SetMethod(args[1].As<Object>(), "runMicrotasks", RunMicrotasks);

  // Do a little housekeeping.
  // 刪除當前執行環境的線程上的 _setupNextTick
  env->process_object()->Delete(
      env->context(),
      FIXED_ONE_BYTE_STRING(args.GetIsolate(), "_setupNextTick")).FromJust();


  // 返回 tick_info 用於和 processNextTick js部分能同步狀態

  uint32_t* const fields = env->tick_info()->fields();
  uint32_t const fields_count = env->tick_info()->fields_count();

  Local<ArrayBuffer> array_buffer =
      ArrayBuffer::New(env->isolate(), fields, sizeof(*fields) * fields_count);

    //返回一個數組 [0, 0] 
  // 和 lib/internal/process/next_tick.js 中
  // kIndex， kLength 對應

  args.GetReturnValue().Set(Uint32Array::New(array_buffer, 0, fields_count));

}複製代碼

RunMicrotasks() 是 v8 暴露的一個 API 方法
cs.chromium.org/chromium/sr…數組

上面設置 tick_callback_function，那麼這個 process.nextTick() 是何時被調用？promise

AsyncWrap

// src/async-wrap.cc

Local<Value> AsyncWrap::MakeCallback(const Local<Function> cb,
                                     int argc,
                                     Local<Value>* argv) {

  ...

  Local<Value> ret = cb->Call(context, argc, argv);

  ...

  Environment::TickInfo* tick_info = env()->tick_info();

    // 若是 nextTick 隊列爲空時執行 RunMicrotasks
  if (tick_info->length() == 0) {
    env()->isolate()->RunMicrotasks();
  }

  Local<Object> process = env()->process_object();

  if (tick_info->length() == 0) {
    tick_info->set_index(0);
    return ret;
  }

  // 直接執行 _tickCallback
  if (env()->tick_callback_function()->Call(process, 0, nullptr).IsEmpty()) {
    return Local<Value>();
  }

  return ret;
}複製代碼

HandleWrap

// - MakeCallback may only be made directly off the event loop.
//   That is there can be no JavaScript stack frames underneath it.

// MakeCallback 會直接在 event loop 中執行。

class HandleWrap : public AsyncWrap {
 public:
  ...
  uv_handle_t* const handle_;
};複製代碼

好比下面的 UDPWrap 是 Nodejs 的 udp 協議 (User Datagram Protocol)的封裝層，它繼承自 HandleWrap

// src/udp_wrap.cc
class UDPWrap: public HandleWrap {
 public:
  ...
 private:
  ...
  uv_udp_t handle_;
};複製代碼

AsyncWrap 是 Nodejs 中大多數 IO 封裝層都是基於 HandleWrap。HandleWrap 繼承自 AsyncWrap , 因此 process.nextTick 和 microtask 基本是在 uv__io_poll 階段調用, 爲何說是主要，由於有兩個其餘狀況，繼續往下看。

node 初始化

node 初始化運行的時候會調用 process._tickCallback()

// lib/module.js

// bootstrap main module.
Module.runMain = function() {
  // Load the main module--the command line argument.
  Module._load(process.argv[1], null, true);
  // Handle any nextTicks added in the first tick of the program
  process._tickCallback();
};複製代碼

捕獲異常

若是應用中拋出異常，未被捕獲的話退出線程，有捕獲的話，應用不會崩潰退出，而是調用 setImmediate 執行 process._tickCallback(), 也就是說 process.nextTick 也可能在 Check 階段被調用。

function setupProcessFatal() {

    process._fatalException = function(er) {
      var caught;

      if (process.domain && process.domain._errorHandler)
        caught = process.domain._errorHandler(er) || caught;

      if (!caught)
        caught = process.emit('uncaughtException', er);
       // 若是沒有函數處理這個異常，C++ 結束
      if (!caught) {
        try {
          if (!process._exiting) {
            process._exiting = true;
            process.emit('exit', 1);
          }
        } catch (er) {
          // nothing to be done about it at this point.
        }
      } else {
        // 若是捕獲了這個異常，在 `setImmediate` 中調用 `_tickCallback()` 繼續處理 nextTick 隊列
        NativeModule.require('timers').setImmediate(process._tickCallback);
      }

      return caught;
    };
  }複製代碼

scheduleMicrotasks?

function scheduleMicrotasks() {
    if (microtasksScheduled)
      return;

    nextTickQueue.push({
      callback: runMicrotasksCallback,
      domain: null
    });

    tickInfo[kLength]++;
    microtasksScheduled = true;
}複製代碼

MakeCallback

// src/node.cc
Local<Value> MakeCallback() {
  ...
   // V8 RunMicrotasks 
  if (tick_info->length() == 0) {
    env->isolate()->RunMicrotasks();
  }
  ...

  return ret;
}複製代碼

Promise

V8 中 microtask 默認是自動運行的。由於 Promise 處理的異步場景和絕大多數 Nodejs 中異步IO 是緊密相關的，因此在 Nodejs 中默認關閉了自動運行而經過 Nodejs 自行觸發 RunMicrotasks（）。結合上面的代碼也能夠基本得出結論 Nodejs 中 Promise 和 process.nextTick() 回調的執行階段是比較類似的。

inline int Start(..) {
...
  isolate->SetAutorunMicrotasks(false);
...複製代碼

總結

process.nextTick() 通常是在 poll 階段被執行，也有可能在 check 階段執行。Promise 所處的 Microtasks 是經過調用 V8 暴露的 RunMicrotasks() 方法執行，RunMicrotasks() 會在 process.nextTick() 隊列執行，也會在 node::MakeCallback 中執行。