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 中执行。

参考资料

lib/internal/process/next_tick.js

src/node.cc

lib/internal/process/promises.js

/src/handle_wrap.h

chromium/src/v8/src/api.cc