libuv & Node.js EventLoop （一）

JavaScript

在网络上查询libuv和EventLoop相关信息的时候，经常看到不同的文章所表达的意思差距较多，主要原因有二吧：

  • 它们的libuv和V8大版本不同，导致具体的实现略有差异
  • 另外它们的代码错综复杂，又是大多数JavaScript工作者不擅长的C/C++，只是从上而下的看，或许一些细节无法完全理解或是认知的分歧

与其受他人影响，不如自己来好好梳理下。

版本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// libuv
#define UV_VERSION_MAJOR 1
#define UV_VERSION_MINOR 33
#define UV_VERSION_PATCH 1

// V8
#define V8_MAJOR_VERSION 7
#define V8_MINOR_VERSION 8
#define V8_BUILD_NUMBER 279
#define V8_PATCH_LEVEL 17

// Node.js
#define NODE_MAJOR_VERSION 14
#define NODE_MINOR_VERSION 0
#define NODE_PATCH_VERSION 0

没有列出具体版本号的代码分析都是耍流氓，2010年的代码和2020年的代码可能差距甚远，“上古”分析固然在当时是对的，但是在今日也许是错误的。



libuv & EventLoop文档比较

_images/loop_iteration.png

上图出自libuv的The I/O loop介绍，具体的信息可以看其文档。

image-20200301145547994

上图出自Event Loop Explained，两者的各个阶段基本是对应的。哪怕是很多其它自行画的图解中，也基本差不多，但唯独pending callbacks阶段，我看到很多文章里面把它标为I/O callbacks，不知道是不是历史原因，但是就从二者的目前文档解释来看是不妥的。

Pending callbacks are called. All I/O callbacks are called right after polling for I/O, for the most part. There are cases, however, in which calling such a callback is deferred for the next loop iteration. If the previous iteration deferred any I/O callback it will be run at this point. —— libuv

libuv里的pending callbacks：大多数的I/O callbacks应该在polling阶段完成，有部分会被延迟到下一个pending callbacks阶段执行。


This phase executes callbacks for some system operations such as types of TCP errors. For example if a TCP socket receives ECONNREFUSED when attempting to connect, some *nix systems want to wait to report the error. This will be queued to execute in the pending callbacks phase. —— Node.js

Node.js里的pending callbacks：这个阶段会执行系统上因一些错误而引起的callbacks，如TCP错误。

我觉得pending callbacks比较合理，一方面如libuv所说，它是一部分延迟的I/O回调，在Node.js里面指的是一些系统上的错误（这些错误也是I/O引起的）。而大多数的I/O操作其实是在 poll阶段。



libuv的大致流程代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
int uv_run(uv_loop_t* loop, uv_run_mode mode) { // 默认 mode 是 UV_RUN_DEFAULT
  int timeout;
  int r;
  int ran_pending;

  // 是否还存在alive的事件
  r = uv__loop_alive(loop);
  if (!r)
    uv__update_time(loop); // 存在的话更新当前的时间，可以把这个时间理解为libuv里面统一的时间，方便触发定时任务

  while (r != 0 && loop->stop_flag == 0) {
    uv__update_time(loop); // 和libuv图里的 Update loop time对应
    uv__run_timers(loop); // 执行 timers 阶段
    ran_pending = uv__run_pending(loop); // 执行pending 阶段；返回0表示空，1表示有；
    uv__run_idle(loop); // idle 阶段； Node.js里面不太关心
    uv__run_prepare(loop); // prepare 阶段； Node.js里面不太关心

    timeout = 0; // 初始化阻塞 poll 阶段的超时时间
    if ((mode == UV_RUN_ONCE && !ran_pending) || mode == UV_RUN_DEFAULT)
      timeout = uv_backend_timeout(loop); // 算出阻塞 poll 阶段的超时时间
    // 根据timers里最近的超时时间算出一个差值 diff = loop.time - min.timeout
    // 如果 diff >= 0 , timeout = 0
    // 否则 timeout = min(diff, INT_MAX)
    
    uv__io_poll(loop, timeout); // 执行 poll 阶段
    uv__run_check(loop); // 执行 check 阶段
    uv__run_closing_handles(loop); // 执行 close 阶段

    // mode 默认 UV_RUN_DEFAULT 所以不执行下面
    if (mode == UV_RUN_ONCE) {
      uv__update_time(loop);
      uv__run_timers(loop);
    }

    // 查看是否还有alive事件
    r = uv__loop_alive(loop);
    if (mode == UV_RUN_ONCE || mode == UV_RUN_NOWAIT)
      break;
  }

  /* The if statement lets gcc compile it to a conditional store. Avoids
   * dirtying a cache line.
   */
  if (loop->stop_flag != 0)
    loop->stop_flag = 0;

  return r;
}

上述uv_run方法同样表达了之前图的循环流程，接下来我们看看各个阶段的具体执行方法。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
void uv__run_timers(uv_loop_t* loop) {
  struct heap_node* heap_node; // timers 里的都是按照最小堆存放的
  uv_timer_t* handle;

  for (;;) {
    heap_node = heap_min(timer_heap(loop)); // 从堆顶取出一个
    if (heap_node == NULL) // 不存在就退出
      break;

    handle = container_of(heap_node, uv_timer_t, heap_node); 
    if (handle->timeout > loop->time) // 触发时间没达到也退出
      break;

    uv_timer_stop(handle);
    uv_timer_again(handle);
    
    handle->timer_cb(handle); // 执行回调
  }
}

timers里存放的事件都是最小堆的数据结构排列的，不断的取出根节点比较当前的loop->time就能知道是执行还是退出，那么这里的handle->timer_cb是我们平时JavaScript里的setTimeout回调吗？后续解答


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
static int uv__run_pending(uv_loop_t* loop) {
  QUEUE* q;
  QUEUE pq;
  uv__io_t* w;

  if (QUEUE_EMPTY(&loop->pending_queue))
    return 0;

  QUEUE_MOVE(&loop->pending_queue, &pq);

  while (!QUEUE_EMPTY(&pq)) { // 双向链表，为空就跳出
    q = QUEUE_HEAD(&pq);
    QUEUE_REMOVE(q);
    QUEUE_INIT(q);
    w = QUEUE_DATA(q, uv__io_t, pending_queue);
    w->cb(loop, w, POLLOUT); // 执行回调
  }

  return 1;
}

libuv里面事件很多是由双向链表构建而成，等着被一个个执行，双向链表的好处就是插入很容易。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#define UV_LOOP_WATCHER_DEFINE(name, type)                                    \
	void uv__run_##name(uv_loop_t* loop) {                                      \
    uv_##name##_t* h;                                                         \
    QUEUE queue;                                                              \
    QUEUE* q;                                                                 \
    QUEUE_MOVE(&loop->name##_handles, &queue);                                \
    while (!QUEUE_EMPTY(&queue)) {                                            \
      q = QUEUE_HEAD(&queue);                                                 \
      h = QUEUE_DATA(q, uv_##name##_t, queue);                                \
      QUEUE_REMOVE(q);                                                        \
      QUEUE_INSERT_TAIL(&loop->name##_handles, q);                            \
      h->name##_cb(h);                                                        \
    }                                                                         \
  }     

UV_LOOP_WATCHER_DEFINE(prepare, PREPARE)
UV_LOOP_WATCHER_DEFINE(check, CHECK)
UV_LOOP_WATCHER_DEFINE(idle, IDLE)

UV_LOOP_WATCHER_DEFINE宏直接初始化了prepare、check和idle3个阶段，都是双向链表。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
// uv__io_poll 太长了，不重要的代码已经移除
void uv__io_poll(uv_loop_t* loop, int timeout) {
  // ...
  if (loop->nfds == 0) { // 没有需要执行的 直接退出
    assert(QUEUE_EMPTY(&loop->watcher_queue));
    return;
  }

  // 将loop->watcher_queue列队里的待观察的文件描述符绑定到epoll上
  while (!QUEUE_EMPTY(&loop->watcher_queue)) {
    q = QUEUE_HEAD(&loop->watcher_queue);
    QUEUE_REMOVE(q);
    QUEUE_INIT(q);

    w = QUEUE_DATA(q, uv__io_t, watcher_queue);
    // ...
    if (w->events == 0)
      op = EPOLL_CTL_ADD; // 添加
    else
      op = EPOLL_CTL_MOD; // 修改

    epoll_ctl(loop->backend_fd, op, w->fd, &e) // epoll_ctl 底层的系统函数，将文件描述符关联起来

    w->events = w->pevents;
  }

  sigmask = 0;
  if (loop->flags & UV_LOOP_BLOCK_SIGPROF) {
    sigemptyset(&sigset);
    sigaddset(&sigset, SIGPROF);
    sigmask |= 1 << (SIGPROF - 1);
  }

  base = loop->time;
  real_timeout = timeout;

  for (;;) {
    nfds = epoll_wait(loop->backend_fd,
                        events,
                        ARRAY_SIZE(events),
                        timeout);
    
    if (nfds == 0) { // 超时，没有新的事件准备好
      assert(timeout != -1); // -1 表示不会超时，而nfds为0表示超时，存在矛盾所以抛出异常
      if (timeout == 0) // 退出
        return;
      
      goto update_timeout; // 重新更新时间 准备下次循环 epoll_wait 
    }

    if (nfds == -1) { // 出错
      if (errno == ENOSYS) {
        /* epoll_wait() or epoll_pwait() failed, try the other system call. */
        assert(no_epoll_wait == 0 || no_epoll_pwait == 0);
        continue;
      }

      if (errno != EINTR)
        abort();

      if (timeout == -1)
        continue;

      if (timeout == 0)
        return;

      goto update_timeout;
    }

    have_signals = 0;
    nevents = 0;

    loop->watchers[loop->nwatchers] = (void*) events;
    loop->watchers[loop->nwatchers + 1] = (void*) (uintptr_t) nfds;
    for (i = 0; i < nfds; i++) {
      pe = events + i;
      fd = pe->data.fd;
      w = loop->watchers[fd];
      w->cb(loop, w, pe->events); // 这个for循环很长，这里我简化了，其实就是准备好的IO就执行回调
    }

    if (timeout == 0)
      return;

    if (timeout == -1)
      continue;

update_timeout:
    assert(timeout > 0);

    real_timeout -= (loop->time - base);
    if (real_timeout <= 0)
      return;

    timeout = real_timeout;
  }
}

uv__io_poll里的timeout参数是根据timers里的最近一次定时时间计算出来的。方法内部使用了epoll，是IO多路复用的一个概念。一共有三种：select、poll和epoll。

epoll是Linux内核为处理大批量文件描述符而作了改进的poll，是Linux下多路复用IO接口select/poll的增强版本，它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。另一点原因就是获取事件的时候，它无须遍历整个被侦听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就行了。epoll除了提供select/poll那种IO事件的水平触发（Level Triggered）外，还提供了边缘触发（Edge Triggered），这就使得用户空间程序有可能缓存IO状态，减少epoll_wait/epoll_pwait的调用，提高应用程序效率。 ——百度百科

我觉得libuv使用epoll主要是因为支持其自身的异步特点

  • epoll监控的文件描述符远远大于select/poll
  • 在大量并发的时候epoll性能远远高于select/poll，而少量异步select/poll相对好点。Node.js利用了libuv的高并发特点。

1
2
3
4
5
6
7
8
9
10
11
12
13
static void uv__run_closing_handles(uv_loop_t* loop) {
  uv_handle_t* p;
  uv_handle_t* q;

  p = loop->closing_handles;
  loop->closing_handles = NULL;

  while (p) {
    q = p->next_closing;
    uv__finish_close(p);
    p = q;
  }
}

一个个触发close回调


上述就是libuv各个阶段的大体流程，poll最为复杂，还有很多东西可以留着细品。



Node.js setTimeout

上文中提到那么这里的handle->timer_cb是我们平时JavaScript里的setTimeout回调吗？回答：不是的

看看Node.js里面如何具体实现这个setTimemout方法的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
function setTimeout(callback, after, arg1, arg2, arg3) {
  var args = [arg1, arg2, arg3]; // 简化了代码
  const timeout = new Timeout(callback, after, args, false);
  active(timeout);

  return timeout;
}
function Timeout(callback, after, args, isRepeat) {
  // 省略了部分代码
  after *= 1; 
  if (!(after >= 1 && after <= TIMEOUT_MAX)) {
    after = 1; // 定时验证不过 就为1 
  }
  this._idleTimeout = after;
  this._onTimeout = callback;
  this._timerArgs = args;
}
function active(item) {
  insert(item, true, getLibuvNow());  // getLibuvNow 获取 loop->time时间
}
function insert(item, refed, start) {
  let msecs = item._idleTimeout;
  if (msecs < 0 || msecs === undefined)
    return;

  // Truncate so that accuracy of sub-millisecond timers is not assumed.
  msecs = Math.trunc(msecs);

  item._idleStart = start;

  // Use an existing list if there is one, otherwise we need to make a new one.
  // timerListMap 是一个键值对，key 是定时时间，value 是一个TimersList
  var list = timerListMap[msecs];
  if (list === undefined) {
    const expiry = start + msecs; // 过期时间
    timerListMap[msecs] = list = new TimersList(expiry, msecs); // 双向链表
    timerListQueue.insert(list); // timerListQueue 用数组实现的最小堆

    if (nextExpiry > expiry) {
      scheduleTimer(msecs); // 如果此次定时任务的有效时间小的话，调用 V8 scheduleTimer
      nextExpiry = expiry;
    }
  }
  // ... 这边简略代码
  L.append(list, item); // 将setTimeout创建的Timeout对象添加到list尾部
}

通过上述代码，我们也可以看出JavaScript层面也有一个维护timers的最小堆，并没有吧具体的某个setTimeout注册到V8里面，只是将定时时间告诉了V8scheduleTimer(msecs);，下面是V8相关代码。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
// timers.cc
void ScheduleTimer(const FunctionCallbackInfo<Value>& args) {
  auto env = Environment::GetCurrent(args);
  env->ScheduleTimer(args[0]->IntegerValue(env->context()).FromJust());
}
// env.cc
void Environment::ScheduleTimer(int64_t duration_ms) {
  if (started_cleanup_) return;
  uv_timer_start(timer_handle(), RunTimers, duration_ms, 0);
}
void Environment::RunTimers(uv_timer_t* handle) {
  Environment* env = Environment::from_timer_handle(handle);
  
  // timers_callback_function 从何而来呢？
  Local<Function> cb = env->timers_callback_function();
  do {
    ret = cb->Call(env->context(), process, 1, &arg);
  } while (ret.IsEmpty() && env->can_call_into_js());
}
// uv/timer.c
int uv_timer_start(uv_timer_t* handle,
                   uv_timer_cb cb,
                   uint64_t timeout,
                   uint64_t repeat) {
  // 给uv_timer_t对象初始化相关属性
  // 并没有JavaScript层面的回调方法，具体的回调也只是C++的RunTimers
  handle->timer_cb = cb;
  handle->timeout = clamped_timeout;
  handle->repeat = repeat;
  handle->start_id = handle->loop->timer_counter++;

  // 将uv_timer_t添加到timer_heap上
  heap_insert(timer_heap(handle->loop),
              (struct heap_node*) &handle->heap_node,
              timer_less_than);
  uv__handle_start(handle);

  return 0;
}

V8部分注册在timers阶段的是一个C++的回调方法，其内部是执行timers_callback_function方法，它从何来来？

1
2
3
4
5
6
7
8
9
// timers.cc
void SetupTimers(const FunctionCallbackInfo<Value>& args) {
  CHECK(args[0]->IsFunction());
  CHECK(args[1]->IsFunction());
  auto env = Environment::GetCurrent(args);

  env->set_immediate_callback_function(args[0].As<Function>()); // 注册了immediate回调
  env->set_timers_callback_function(args[1].As<Function>()); // 注册了timers回调
}
1
2
3
4
5
6
7
8
9
10
11
12
13
// node.js
const {
  setupTaskQueue,
  queueMicrotask
} = require('internal/process/task_queues');
const { nextTick, runNextTicks } = setupTaskQueue();
process.nextTick = nextTick;
process._tickCallback = runNextTicks;

const { getTimerCallbacks } = require('internal/timers');
const { setupTimers } = internalBinding('timers'); // c++ SetupTimers 方法传到了JavaScript层
const { processImmediate, processTimers } = getTimerCallbacks(runNextTicks);
setupTimers(processImmediate, processTimers);

我们通过上述代码看到在node.js里面会将processImmediate和processTimers这两个JavaScript方法注册到对应的C++回调里，之后执行timers_callback_function其实就是执行processTimers方法。

另外，需要注意微任务相关的代码会被带到getTimerCallbacks。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
function getTimerCallbacks(runNextTicks) {
  function processImmediate() {
    // 之后文章在做介绍
  }


  function processTimers(now) {
    nextExpiry = Infinity;

    let list;
    let ranAtLeastOneList = false;
    while (list = timerListQueue.peek()) {
      if (list.expiry > now) { // 当前列表的过期时间大于now(libuv loop->time), 还没有到过期或到触发时间
        nextExpiry = list.expiry; // nextExpiry 设置为 当前列表的过期时间
        return refCount > 0 ? nextExpiry : -nextExpiry;
      }
      if (ranAtLeastOneList)
        runNextTicks(); // 微任务会被触发
      else
        ranAtLeastOneList = true;
      listOnTimeout(list, now);
    }
    return 0;
  }

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

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

    var diff, timer;
    let ranAtLeastOneTimer = false;
    while (timer = L.peek(list)) { // 如果不是一个空的list，持续执行
      // _idleStart 是插入libuv (loop->time)的时间
      // now 是当前libuv 执行此次callback的时间
      diff = now - timer._idleStart; 

      // Check if this loop iteration is too early for the next timer.
      // This happens if there are more timers scheduled for later in the list.
      if (diff < msecs) {
        list.expiry = Math.max(timer._idleStart + msecs, now + 1);
        list.id = timerListId++;
        timerListQueue.percolateDown(1);
        debug('%d list wait because diff is %d', msecs, diff);
        return;
      }

      if (ranAtLeastOneTimer)
        runNextTicks(); // 微任务会被触发
      else
        ranAtLeastOneTimer = true;

      // The actual logic for when a timeout happens.
      L.remove(timer);

      const asyncId = timer[async_id_symbol];
      if (!timer._onTimeout) {
        continue;
      }
      try {
        const args = timer._timerArgs;
        if (args === undefined)
          timer._onTimeout(); // _onTimeout setTimeout 回调
        else
          timer._onTimeout(...args);
      } finally {
        // ...
      }
    }
    if (list === timerListMap[msecs]) {
      delete timerListMap[msecs];
      timerListQueue.shift();
    }
  }

  return {
    processImmediate,
    processTimers
  };
}

上述代码很长，但是很容易理解，就是根据libuv的时间，将到期的定时任务一一执行了，格外需要注意的是微任务的执行runNextTicks，每次setTimeout的callback之后都会将微任务清空，而不是网上很多文章说的timer阶段之后将微任务清空，这个改动在Node.js 11版本

Timers
Interval timers will be rescheduled even if previous interval threw an error. #20002
nextTick queue will be run after each immediate and timer. #22842

具体的原因是希望和浏览器里的行为更为一直，下面的代码在11之前和之后的执行结果略有不同。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
setTimeout(() => {
  console.log('time1');
  Promise.resolve().then(() => {
    console.log('promise1');
  });
});

setTimeout(() => {
  console.log('time2');
  Promise.resolve().then(() => {
    console.log('promise2');
  });
});

/* 11之后
time1
promise1
time2
promise2
*/

/* 11之前
time1
time2
promise1
promise2
*/



总结

至此，libuv和Node.js EventLoop关于各个阶段和setTimeout的实现已经做了简单介绍。具体的还是需要看各自版本的代码而定，不能轻易去“相信”网上的介绍，比如最后一个例子就很容易在不同版本出现不同的执行结果。之后，有时间介绍 nextTick的回调实现，这个比较复杂。

#Node.js #JavaScript
Node.js 监控中心架构迭代 收集Node.js应用的内存堆栈快照和CPU火焰图

Comments

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×