nodejs 14.0.0源码分析之setTimeout

theanarkh

发布于 2020-02-25 15:20:43

7670

发布于 2020-02-25 15:20:43

这一篇我们来看看nodejs是如何实现定时器的。14.0.0的nodejs对定时器模块进行了重构，之前版本的实现是用一个map，以超时时间为键，每个键对应一个队列。即有同样超时时间的节点在同一个队列。每个队列对应一个底层的一个节点（二叉堆里的节点），nodejs在时间循环的timer阶段会从二叉堆里找出超时的节点，然后执行回答，回调里会遍历队列，哪个节点超时了。14.0.0重构后，只使用了一个二叉堆的节点。我们看一下他的实现。我们先看下定时器模块的组织结构。

在这里插入图片描述下面我们继续看一下定时器模块的几个重要的数据结果。

1 TimersList

超时时间一样的会被放到同一个队列，这个队列就是由TimersList来管理。对应图中的list那个方框。

// expiry是超时时间的绝对值。用来记录队列中最快到期的节点的时间，msecs是超时时间的相对值（相对插入时的当前时间） 
function TimersList(expiry, msecs) {
  // 用于链表
  this._idleNext = this; 
  this._idlePrev = this; 
  this.expiry = expiry;
  this.id = timerListId++;
  this.msecs = msecs;
  // 在优先队列里的位置
  this.priorityQueuePosition = null;
}

2 优先队列

const timerListQueue = new PriorityQueue(compareTimersLists, setPosition)

nodejs用优先队列对所有1中的链表进行管理，优先队列本质是一个二叉堆（小根堆），每个链表在二叉堆里对应一个节点。根据1中，我们知道每个链表都保存链表中最快到期的节点的过期时间。二叉堆以该事件为依据，即最快到期的list对应二叉堆中的根节点。我们判断根节点是否超时，如果没有超时，说明整个二叉堆的节点都没有超时。如果超时了，就需要不断遍历堆中的节点。

3 超时时间和链表的映射

1中已经提到，超时时间一样的节点，会排在同一个链表中个，nodejs中用一个map保存了超时时间到链表的映射关系。

了解完定时器整体的组织和基础数据结构，我们可以开始进入真正的源码分析了。

我们直接从setTimeout函数开始。

function setTimeout(callback, after, arg1, arg2, arg3) {
  if (typeof callback !== 'function') {
    throw new ERR_INVALID_CALLBACK(callback);
  }

  let i, args;
  switch (arguments.length) {
    // fast cases
    case 1:
    case 2:
      break;
    case 3:
      args = [arg1];
      break;
    case 4:
      args = [arg1, arg2];
      break;
    default:
      args = [arg1, arg2, arg3];
      for (i = 5; i < arguments.length; i++) {
        // Extend array dynamically, makes .apply run much faster in v6.0.0
        args[i - 2] = arguments[i];
      }
      break;
  }

  const timeout = new Timeout(callback, after, args, false, true);
  insert(timeout, timeout._idleTimeout);
  return timeout;
}

两个主要操作，new Timeout和insert。我们一个个来。

1 Timeout

function Timeout(callback, after, args, isRepeat, isRefed) {
  after *= 1; // Coalesce to number or NaN
  if (!(after >= 1 && after <= TIMEOUT_MAX)) {
    if (after > TIMEOUT_MAX) {
      process.emitWarning(`${after} does not fit into` +
                          ' a 32-bit signed integer.' +
                          '\nTimeout duration was set to 1.',
                          'TimeoutOverflowWarning');
    }
    after = 1; // Schedule on next tick, follows browser behavior
  }
  // 超时时间相对值
  this._idleTimeout = after;
  // 前后指针，用于链表
  this._idlePrev = this;
  this._idleNext = this;
  // 定时器的开始时间
  this._idleStart = null;
  // This must be set to null first to avoid function tracking
  // on the hidden class, revisit in V8 versions after 6.2
  // 超时回调
  this._onTimeout = null;
  this._onTimeout = callback;
  // 执行回调时传入的参数
  this._timerArgs = args;
  // 是否定期执行回调，用于setInterval
  this._repeat = isRepeat ? after : null;
  this._destroyed = false;
  // 激活底层的定时器节点（二叉堆的节点），说明有定时节点需要处理
  if (isRefed)
    incRefCount();
  this[kRefed] = isRefed;

  initAsyncResource(this, 'Timeout');
}

Timeout主要是新建一个对象记录一些定时器的上下文信息。

2 insert（对照上面的图理解）

function insert(item, msecs, start = getLibuvNow()) {
  msecs = MathTrunc(msecs);
  // 记录定时器的开始时间，见Timeout函数的定义
  item._idleStart = start;
  // 该超时时间是否已经存在对应的链表
  let list = timerListMap[msecs];
  // 还没有
  if (list === undefined) {
      // 算出绝对超时时间
    const expiry = start + msecs;
    // 新建一个链表
    timerListMap[msecs] = list = new TimersList(expiry, msecs);
    // 插入优先队列
    timerListQueue.insert(list);
    // 算出下一次超时的时间，即最快到期的时间
    if (nextExpiry > expiry) {
      // 设置底层的最后超时时间，这样保证可以尽量按时执行
      scheduleTimer(msecs);
      nextExpiry = expiry;
    }
  }
  // 把当前节点加到队列里
  L.append(list, item);
}

scheduleTimer函数是对c++函数的封装。

void ScheduleTimer(const FunctionCallbackInfo<Value>& args) {
  auto env = Environment::GetCurrent(args);
  env->ScheduleTimer(args[0]->IntegerValue(env->context()).FromJust());
}

void Environment::ScheduleTimer(int64_t duration_ms) {
  if (started_cleanup_) return;
  uv_timer_start(timer_handle(), RunTimers, duration_ms, 0);
}

uv_timer_start就是开启底层计时，即往libuv的二叉堆插入一个节点。超时时间是duration_ms，就是最快到期的时间，在timer阶段会判断是否过期。是的话执行RunTimers函数。我们先看一下该函数的主要代码。

Local<Function> cb = env->timers_callback_function();
ret = cb->Call(env->context(), process, 1, &arg);

RunTimers会执行timers_callback_function。timers_callback_function是在nodejs初始化的时候设置的。我们先暂定一下，看一下定时器模块的初始化流程。再回来这里分析。

nodejs在初始化的时候通过一下代码对定时器进行了初始化工作。

setupTimers(processImmediate, processTimers);

setupTimers对应的c++函数是

void SetupTimers(const FunctionCallbackInfo<Value>& args) {
  auto env = Environment::GetCurrent(args);
  env->set_immediate_callback_function(args[0].As<Function>());
  env->set_timers_callback_function(args[1].As<Function>());
}

nodejs把processTimers设置为超时的回调函数。现在我们知道了nodejs是如何设置超时的处理函数，也知道了什么时候会执行该回调。那我们就来看一下回调时具体处理逻辑。

void Environment::RunTimers(uv_timer_t* handle) {
  Local<Function> cb = env->timers_callback_function();
  MaybeLocal<Value> ret;
  Local<Value> arg = env->GetNow();

  do {
    // 执行js回调，即下面的processTimers函数
    ret = cb->Call(env->context(), process, 1, &arg);
  } while (ret.IsEmpty() && env->can_call_into_js());

  // 是否执行了所有的节点
  if (ret.IsEmpty())
    return;

  int64_t expiry_ms = ret.ToLocalChecked()->IntegerValue(env->context()).FromJust();

  uv_handle_t* h = reinterpret_cast<uv_handle_t*>(handle);
  // 还有超时节点，开块超时时间是expiry_ms ，需要重新插入底层的二叉堆。
  if (expiry_ms != 0) {
    // 算出下次超时的相对值int64_t duration_ms =
        llabs(expiry_ms) - (uv_now(env->event_loop()) - env->timer_base());
    // 重新把handle插入libuv的二叉堆
    env->ScheduleTimer(duration_ms > 0 ? duration_ms : 1);

  }
}

该函数主要是执行回调，然后如果还有没超时的节点，重新设置libuv定时器的时间。看看js层面。

  function processTimers(now) {
    nextExpiry = Infinity;

    let list;
    let ranAtLeastOneList = false;
    // 取出优先队列的根节点，即最快到期的节点
    while (list = timerListQueue.peek()) {
      // 还没过期，
      if (list.expiry > now) {
        nextExpiry = list.expiry;
        // 返回下一次过期的时间
        return refCount > 0 ? nextExpiry : -nextExpiry;
      }

      listOnTimeout(list, now);
    }
    return 0;
  }

  function listOnTimeout(list, now) {
    const msecs = list.msecs;

    debug('timeout callback %d', msecs);

    let ranAtLeastOneTimer = false;
    let timer;
    // 遍历具有统一相对过期时间的队列
    while (timer = L.peek(list)) {
      // 算出已经过去的时间
      const diff = now - timer._idleStart;
      // 过期的时间比超时时间小，还没过期
      if (diff < msecs) {
        // 整个链表节点的最快过期时间等于当前还没过期节点的值，链表是有序的
        list.expiry = MathMax(timer._idleStart + msecs, now + 1);
        // 更新id，用于决定在优先队列里的位置
        list.id = timerListId++;
        // 调整过期时间后，当前链表对应的节点不一定是优先队列里的根节点了，可能有他更快到期，即当前链表需要往下沉
        timerListQueue.percolateDown(1);
        return;
      }

      // 准备执行用户设置的回调，删除这个节点
      L.remove(timer);

      let start;
      if (timer._repeat)
        start = getLibuvNow();

      try {
        const args = timer._timerArgs;
        // 执行用户设置的回调
        if (args === undefined)
          timer._onTimeout();
        else
          timer._onTimeout(...args);
      } finally {
        // 设置了重复执行回调，即来自setInterval。则需要重新加入链表。
        if (timer._repeat && timer._idleTimeout !== -1) {
          // 更新超时时间，一样的时间间隔
          timer._idleTimeout = timer._repeat;
          // 重新插入链表
          insert(timer, timer._idleTimeout, start);
        } else if (!timer._idleNext && !timer._idlePrev && !timer._destroyed) {
          timer._destroyed = true;
          if (timer[kRefed])
            refCount--;
    }
    // 为空则删除
    if (list === timerListMap[msecs]) {
      delete timerListMap[msecs];
      timerListQueue.shift();
    }
  }

上面的代码主要是遍历优先队列，