[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8},"actionData","errors","root",{"_11":-5},"routes/user.$name.topics",{"_9":10,"_11":12,"_13":14,"_15":16,"_17":18,"_19":16},"topics",[34],"user",{"_20":21,"_13":14,"_22":23,"_24":14,"_25":26,"_27":28,"_29":16,"_30":31,"_32":33},"loginname","lolBig","page",1,"limit",20,"total","id","29896","avatar_url","https://avatars3.githubusercontent.com/u/3438038?v=4","githubUsername","create_at","2017-08-30 01:45:32.404","score",5,"topic_count","reply_count",0,"is_star",false,{"_20":35,"_36":21,"_37":38,"_39":40,"_41":42,"_43":44,"_45":33,"_46":33,"_30":16,"_47":48,"_25":49,"_50":51},"40630","author_id","tab","share","content","通常我们在讨论 Node.js 的时候都会涉及到异步这个特性。实际上 Node.js 在执行异步调用的时候，不同的场景下有着不同的处理方式。本文将通过 libuv 源码来分析 Node.js 是如何通过 libuv 的线程池完成异步调用。本文描述的 Node.js 版本为 [v11.15.0][]，libuv 版本为 [1.24.0][] 。\r\n\r\n以下面的代码为例，它通过调用 [fs.access][] 来异步地判断文件是否存在并在回调中打印日志，在 Node.js 中这是一个典型的异步调用。\r\n\r\n```javascript\r\nconst fs = require('fs')\r\nconst cb = function (err) {\r\n console.log(`Is myfile exists: ${!err}`)\r\n}\r\nfs.access('myfile', cb)\r\n```\r\n\r\n在分析上面这段代码的调用过程之前，我们先来了解一些 libuv 概念。\r\n\r\n### 什么类型的请求 libuv 会把它放到线程池去执行\r\n\r\n主动通过 libuv 发起的操作被 libuv 称为[请求][]( `uv_req_t` )，libuv 的线程池作用于以下 4 种枚举的异步请求：\r\n\r\n- `UV_FS`： [fs][] 模块的异步函数（除了 [uv_fs_req_cleanup][] ），[fs.access][]、[fs.stat][] 等。\r\n- `UV_GETADDRINFO`：[dns][] 模块的异步函数，[dns.lookup][] 等。\r\n- `UV_GETNAMEINFO`：[dns][] 模块的异步函数，[dns.lookupService][] 等。\r\n- `UV_WORK`：[zlib][] 模块的 [zlib.unzip][]、[zlib.gzip][] 等；在 Node.js 的 Addon(C/C++) 中通过 [uv_queue_work][] 创建的多线程请求。\r\n\r\n其它的 `UV_CONNECT`、`UV_WRITE`、`UDP_SEND` 等则并不会通过线程池去执行。\r\n\r\n### 线程池请求分类\r\n\r\n这 4 种枚举请求 libuv 内部把它们分为 3 种[任务类型][]( `uv__work_kind` )：\r\n\r\n- `UV__WORK_CPU`：CPU 密集型，`UV_WORK` 类型的请求被定义为这种类型。因此根据这个分类，不推荐在 `uv_queue_work` 中做 I/O 密集的操作。\r\n- `UV__WORK_FAST_IO`：快 IO 型，`UV_FS` 类型的请求被定义为这种类型。\r\n- `UV__WORK_SLOW_IO`：慢 IO 型，`UV_GETADDRINFO` 和 `UV_GETNAMEINFO` 类型的请求被定义为这种类型。\r\n\r\n`UV__WORK_SLOW_IO` 执行不同于 `UV__WORK_CPU` 与 `UV__WORK_FAST_IO` ，libuv 执行它的时候流程会有些差异，这个后面会提到。\r\n\r\n### 线程池是如何初始化的\r\n\r\nlibuv 通过[init_threads][] 函数初始化线程池，初始化时会根据一个名为 [UV_THREADPOOL_SIZE][] 的环境变量来初始化内部线程池的大小，线程最大数量为 [128][] ，默认为 [4][] 。如果以单进程的架构去部署服务，可以根据服务器 CPU 的核心数量及业务情况来设置线程池大小，达到资源利用的最大化。uv loop 线程在创建 worker 线程时，会初始化以下变量：\r\n\r\n- 信号量 sem：在创建线程时与线程进行同步，每个线程创建好后将会通过这个信号量告知 uv loop 线程自己已经初始化完毕，可以开始处理请求了。当所有线程都初始化完成后这个信号量将被销毁，即完成线程池的初始化。\r\n- 条件变量 cond：线程创建完成后通过这个条件变量进入阻塞状态( [uv_cond_wait][] )，直到其它线程通过 [uv_cond_signal][] 将其唤醒。\r\n- 互斥量 mutex：对下面 3 个临界资源进行互斥访问。\r\n- 请求队列 wq：线程池收到 `UV__WORK_CPU` 和 `UV__WORK_FAST_IO` 类型的请求后将其插到此队列的尾部，并通过 [uv_cond_signal][] 唤醒 worker 线程去处理，这是线程池请求的主队列。\r\n- 慢 I/O 队列 slow_io_pending_wq：线程池收到 `UV__WORK_SLOW_IO` 类型的请求后将其插到此队列的尾部。\r\n- 慢 I/O 标志位节点 run_slow_work_message：当存在慢 I/O 请求时，用来作为一个标志位放在请求队列 wq 中，表示当前有慢 I/O 请求，worker 线程处理请求时需要关注慢 I/O 队列的请求；当慢 I/O 队列的请求都处理完毕后这个标志位将从请求队列 wq 中移除。\r\n\r\nworker 线程的入口函数均为 [worker][] 函数，这个我们后面再说。 [init_threads][] 实现如下：\r\n\r\n```c\r\nstatic void init_threads(void) {\r\n unsigned int i;\r\n const char* val;\r\n uv_sem_t sem;\r\n\r\n // 6-23 行初始化线程池大小\r\n nthreads = ARRAY_SIZE(default_threads);\r\n val = getenv(\"UV_THREADPOOL_SIZE\"); // 根据环境变量设置线程池大小\r\n if (val != NULL)\r\n nthreads = atoi(val);\r\n if (nthreads == 0)\r\n nthreads = 1;\r\n if (nthreads > MAX_THREADPOOL_SIZE)\r\n nthreads = MAX_THREADPOOL_SIZE;\r\n\r\n threads = default_threads;\r\n if (nthreads > ARRAY_SIZE(default_threads)) {\r\n threads = uv__malloc(nthreads * sizeof(threads[0]));\r\n if (threads == NULL) {\r\n nthreads = ARRAY_SIZE(default_threads);\r\n threads = default_threads;\r\n }\r\n }\r\n // 初始化条件变量\r\n if (uv_cond_init(&cond))\r\n abort();\r\n\r\n // 初始化互斥量\r\n if (uv_mutex_init(&mutex))\r\n abort();\r\n\r\n // 初始化队列和节点\r\n QUEUE_INIT(&wq); // 工作队列\r\n QUEUE_INIT(&slow_io_pending_wq); // 慢 I/O 队列\r\n QUEUE_INIT(&run_slow_work_message); // 如果有慢 I/O 请求，将此节点作为标志位插入到 wq 中\r\n\r\n // 初始化信号量\r\n if (uv_sem_init(&sem, 0))\r\n abort(); // 后续线程同步需要依赖这个信号量，因此这个信号量创建失败了则终止进程\r\n\r\n // 创建 worker 线程\r\n for (i = 0; i < nthreads; i++)\r\n if (uv_thread_create(threads + i, worker, &sem)) // 初始化 worker 线程\r\n abort(); // woker 线程创建错误原因为 EAGAIN、EINVAL、EPERM 其中之一，具体请参考 man3\r\n \r\n // 等待 worker 创建完成\r\n for (i = 0; i < nthreads; i++)\r\n uv_sem_wait(&sem); // 等待 worker 线程创建完毕\r\n\r\n // 回收信号量资源\r\n uv_sem_destroy(&sem);\r\n}\r\n```\r\n\r\n### 请求是如何放到线程池去执行的\r\n\r\nlibuv 有两个函数可以创建多线程请求：\r\n\r\n- [uv_queue_work][]：开发者常用的创建多线程请求的函数。\r\n- [uv__work_submit][]：libuv 内部创建多线程请求的函数，实际上 [uv_queue_work][] 最终也是调用的这个函数。\r\n\r\n[uv__work_submit][] 函数主要做 2 件事：\r\n\r\n1. 调用 [init_threads][] 初始化线程池，因为线程池的创建是惰性的，只有用到的时候才会创建。\r\n2. 调用内部的 [post][] 函数将请求插入到请求队列中。\r\n\r\n实现如下：\r\n\r\n```c\r\nvoid uv__work_submit(uv_loop_t* loop,\r\n struct uv__work* w,\r\n enum uv__work_kind kind,\r\n void (*work)(struct uv__work* w),\r\n void (*done)(struct uv__work* w, int status)) {\r\n // 在收到请求后才开始初始化线程池，但是只会初始化一次\r\n uv_once(&once, init_once);\r\n w->loop = loop;\r\n w->work = work;\r\n w->done = done;\r\n post(&w->wq, kind);\r\n}\r\n\r\nstatic void init_once(void) {\r\n // fork 后子进程的 mutex 、condition variables 等 pthread 变量的状态是父进程 fork 时的复制，所以子进程创建时需要重置状态\r\n // 具体请参考 http://man7.org/linux/man-pages/man2/fork.2.html\r\n if (pthread_atfork(NULL, NULL, &reset_once))\r\n abort();\r\n // 初始化线程池\r\n init_threads();\r\n}\r\n\r\nstatic void reset_once(void) {\r\n // 重置 once 变量\r\n uv_once_t child_once = UV_ONCE_INIT;\r\n memcpy(&once, &child_once, sizeof(child_once));\r\n}\r\n```\r\n\r\n[post][] 函数主要做 2 件事：\r\n\r\n1. 判断请求的请求类型是否是 `UV__WORK_SLOW_IO`：\r\n - 如果是，将这个请求插到慢 I/O 请求队列 `slow_io_pending_wq` 的尾部，同时在请求队列 `wq` 的尾部插入一个 `run_slow_work_message` 节点作为标志位，告知请求队列 `wq` 当前存在慢 I/O 请求。\r\n - 如果不是，将请求插到请求队列 `wq` 尾部。\r\n2. 如果有空闲的线程，唤醒某一个去执行请求。\r\n\r\n并发的慢 I/O 的请求数量不会超过线程池大小的一半，这样做的好处是避免多个慢 I/O 的请求在某段时间内把所有线程都占满，导致其它能够快速执行的请求需要排队。\r\n\r\n[post][] 函数实现如下：\r\n\r\n```c\r\nstatic void post(QUEUE* q, enum uv__work_kind kind) {\r\n // 加锁\r\n uv_mutex_lock(&mutex);\r\n if (kind == UV__WORK_SLOW_IO) {\r\n /* 插入到慢 I/O 队列中 */\r\n QUEUE_INSERT_TAIL(&slow_io_pending_wq, q);\r\n /* 如果 run_slow_work_message 节点不为空代表其已在 wq 队列中，无需再次插入 */\r\n if (!QUEUE_EMPTY(&run_slow_work_message)) {\r\n uv_mutex_unlock(&mutex);\r\n return;\r\n }\r\n // 不在 wq 队列中则将 run_slow_work_message 作为标志位插到 wq 尾部\r\n q = &run_slow_work_message;\r\n }\r\n // 将请求插到请求队列尾部\r\n QUEUE_INSERT_TAIL(&wq, q);\r\n // 如果有空闲的线程，唤醒某一个去执行请求\r\n if (idle_threads > 0)\r\n uv_cond_signal(&cond); // 唤醒一个 worker 线程\r\n uv_mutex_unlock(&mutex);\r\n}\r\n```\r\n\r\nworker 线程的入口函数 [worker][] 在线程创建好并初始化完成后将按照下面的步骤不断的循环：\r\n\r\n1. 等待唤醒。\r\n2. 取出请求队列 wq 或者慢 I/O 请求队列的头部请求去执行。\r\n3. 通知 uv loop 线程完成了一个请求的处理。\r\n4. 回到 1 。\r\n\r\n```c\r\nstatic void worker(void* arg) {\r\n struct uv__work* w;\r\n QUEUE* q;\r\n int is_slow_work;\r\n\r\n // 通知 uv loop 线程此 worker 线程已创建完毕\r\n uv_sem_post((uv_sem_t*) arg);\r\n arg = NULL;\r\n\r\n uv_mutex_lock(&mutex);\r\n // 通过这个死循环来不断的执行请求\r\n for (;;) {\r\n /*\r\n \t这个 while 有2个判断\r\n \t1. 在多核处理器下，pthread_cond_signal 可能会激活多于一个线程，通过一个 while 来避免这种情况导致的问题，具体请参考 https://linux.die.net/man/3/pthread_cond_signal\r\n \t2. 限制慢 I/O 请求的数量小于线程数量的一半\r\n */\r\n while (QUEUE_EMPTY(&wq) ||\r\n (QUEUE_HEAD(&wq) == &run_slow_work_message &&\r\n QUEUE_NEXT(&run_slow_work_message) == &wq &&\r\n slow_io_work_running >= slow_work_thread_threshold())) {\r\n idle_threads += 1;\r\n // worker 线程初始化完成或没有请求执行时进入阻塞状态，直到被新的请求唤醒\r\n uv_cond_wait(&cond, &mutex);\r\n idle_threads -= 1;\r\n }\r\n // 唤醒并且达到执行请求的条件后取出队列头部的请求\r\n q = QUEUE_HEAD(&wq);\r\n // 如果头部请求是退出，则跳出循环，结束 worker 线程\r\n if (q == &exit_message) {\r\n // 继续唤醒其它 worker 去结束线程\r\n uv_cond_signal(&cond);\r\n uv_mutex_unlock(&mutex);\r\n break;\r\n }\r\n\r\n // 将这个请求节点从请求队列 wq 中移除\r\n QUEUE_REMOVE(q);\r\n QUEUE_INIT(q);\r\n\r\n is_slow_work = 0;\r\n // 如果这个请求是慢 I/O 的标志位\r\n if (q == &run_slow_work_message) {\r\n /* 控制慢 I/O 请求数量，超过则插到队列尾部，等待前面的请求执行完 */\r\n if (slow_io_work_running >= slow_work_thread_threshold()) {\r\n QUEUE_INSERT_TAIL(&wq, q);\r\n continue;\r\n }\r\n\r\n /* 判断慢 I/O 请求队列中是否有请求，请求有可能被取消 */\r\n if (QUEUE_EMPTY(&slow_io_pending_wq))\r\n continue;\r\n\r\n is_slow_work = 1;\r\n slow_io_work_running++;\r\n\r\n // 取出慢 I/O 请求队列中头部的请求\r\n q = QUEUE_HEAD(&slow_io_pending_wq);\r\n QUEUE_REMOVE(q);\r\n QUEUE_INIT(q);\r\n\r\n // 如果慢 I/O 请求队列中还有请求，则将 run_slow_work_message 这个标志位重新插到请求队列 wq 的尾部\r\n if (!QUEUE_EMPTY(&slow_io_pending_wq)) {\r\n QUEUE_INSERT_TAIL(&wq, &run_slow_work_message);\r\n if (idle_threads > 0)\r\n uv_cond_signal(&cond); // 唤醒一个线程继续执行\r\n }\r\n }\r\n\r\n uv_mutex_unlock(&mutex);\r\n\r\n w = QUEUE_DATA(q, struct uv__work, wq);\r\n // 上面处理了这多，终于在这里开始执行请求的函数了\r\n w->work(w);\r\n\r\n uv_mutex_lock(&w->loop->wq_mutex);\r\n w->work = NULL;\r\n \r\n // 为保证线程安全，请求执行完后不会立即回调请求，而是将完成的请求插到已完成的请求队列中，在uv loop 线程完成回调\r\n QUEUE_INSERT_TAIL(&w->loop->wq, &w->wq);\r\n // 通过 uv_async_send 同步 uv loop 线程：线程池完成了一个请求\r\n uv_async_send(&w->loop->wq_async);\r\n uv_mutex_unlock(&w->loop->wq_mutex);\r\n\r\n uv_mutex_lock(&mutex);\r\n if (is_slow_work) {\r\n slow_io_work_running--;\r\n }\r\n }\r\n}\r\n```\r\n\r\n### 请求在 worker 执行完后是如何同步 uv loop 所在的线程\r\n\r\n在 [uv_loop_init][] 时，线程池的 [wq_async][]([uv_async_t][]) 句柄通过 [uv_async_init][] 初始化并插入到 uv loop 的 [async_handles][] 队列中，然后在 uv loop 线程中遍历 [async_handles][] 队列并完成回调。\r\n\r\nworker 线程和 uv loop 线程通过 [uv_async_send][] 进行同步，而[uv_async_send][] 只做了一件事：向 [async_wfd][] 句柄写了一个长度为 1 个字节的字符串（只有 `\\0` 这个字符）。\r\n\r\n[uv_async_send][] 实现如下：\r\n\r\n```c\r\nint uv_async_send(uv_async_t* handle) {\r\n if (ACCESS_ONCE(int, handle->pending) != 0)\r\n return 0;\r\n // cmpxchgi 函数设置标志位，如果已经设置过则不会重复调用 uv__async_send\r\n if (cmpxchgi(&handle->pending, 0, 1) == 0)\r\n uv__async_send(handle->loop);\r\n\r\n return 0;\r\n}\r\n\r\nstatic void uv__async_send(uv_loop_t* loop) {\r\n const void* buf;\r\n ssize_t len;\r\n int fd;\r\n int r;\r\n\r\n buf = \"\";\r\n len = 1;\r\n fd = loop->async_wfd;\r\n\r\n#if defined(__linux__)\r\n if (fd == -1) {\r\n static const uint64_t val = 1;\r\n buf = &val;\r\n len = sizeof(val);\r\n fd = loop->async_io_watcher.fd; /* eventfd */\r\n }\r\n#endif\r\n\r\n do\r\n r = write(fd, buf, len); // 向 fd 写入内容\r\n while (r == -1 && errno == EINTR);\r\n\r\n if (r == len)\r\n return;\r\n\r\n if (r == -1)\r\n if (errno == EAGAIN || errno == EWOULDBLOCK)\r\n return;\r\n\r\n abort();\r\n}\r\n```\r\n\r\n对 [async_wfd][] 写内容为什么能做到同步呢？实际上在 worker 线程对 [async_wfd][] 写入时，uv loop 线程同时也在不断的[循环][]去接收处理各种各样的事件或请求，其中就包括对 [async_wfd][] 可读事件的监听。\r\n\r\nuv loop 是在 [uv_run][] 函数中执行的，它在 Node.js [启动时][] 被调用， [uv_run][] 实现如下：\r\n\r\n```c\r\nint uv_run(uv_loop_t* loop, uv_run_mode mode) {\r\n int timeout;\r\n int r;\r\n int ran_pending;\r\n\r\n r = uv__loop_alive(loop);\r\n if (!r)\r\n uv__update_time(loop);\r\n\r\n while (r != 0 && loop->stop_flag == 0) {\r\n // 更新计时器时间\r\n uv__update_time(loop);\r\n // 回调超时的计时器，setTimeout、setInterval 都是由这个函数回调\r\n uv__run_timers(loop);\r\n // 处理某些没有在 uv__io_poll 完成的回调\r\n ran_pending = uv__run_pending(loop);\r\n // 官方解释：Idle handle is needed only to stop the event loop from blocking in poll.\r\n // 实际上 napi 中某些函数比如 napi_call_threadsafe_function 会往 idle 队列中插入回调，然后在这个阶段执行\r\n uv__run_idle(loop);\r\n // process._startProfilerIdleNotifier 的回调\r\n uv__run_prepare(loop);\r\n\r\n timeout = 0;\r\n if ((mode == UV_RUN_ONCE && !ran_pending) || mode == UV_RUN_DEFAULT)\r\n timeout = uv_backend_timeout(loop); // 计算 uv__io_poll 超时时间，算法请参考 https://github.com/libuv/libuv/blob/v1.24.0/src/unix/core.c#L318\r\n\r\n // 对 async_wfd 可读的监听在 uv__io_poll 这个函数中\r\n // 第二个参数 timeout 为上面计算出来，用来设置 epoll_wait 等函数等待 I/O 事件的时间\r\n uv__io_poll(loop, timeout);\r\n // setImmediate 的回调\r\n // ps: 个人觉得从实现上讲 setImmediate 和 nextTick 应该互换名字 :-)\r\n uv__run_check(loop);\r\n // 关闭句柄是个异步操作\r\n // 一般结束 uv loop 时会先调用 uv_walk 遍历所有句柄并关闭它们，然后再执行一次 uv loop 通过这个函数来完成关闭，最后再调用 uv_loop_close，否则的话会出现内存泄露\r\n uv__run_closing_handles(loop);\r\n\r\n if (mode == UV_RUN_ONCE) {\r\n /* UV_RUN_ONCE implies forward progress: at least one callback must have\r\n * been invoked when it returns. uv__io_poll() can return without doing\r\n * I/O (meaning: no callbacks) when its timeout expires - which means we\r\n * have pending timers that satisfy the forward progress constraint.\r\n *\r\n * UV_RUN_NOWAIT makes no guarantees about progress so it's omitted from\r\n * the check.\r\n */\r\n uv__update_time(loop);\r\n uv__run_timers(loop);\r\n }\r\n\r\n r = uv__loop_alive(loop);\r\n if (mode == UV_RUN_ONCE || mode == UV_RUN_NOWAIT)\r\n break;\r\n }\r\n\r\n /* The if statement lets gcc compile it to a conditional store. Avoids\r\n * dirtying a cache line.\r\n */\r\n if (loop->stop_flag != 0)\r\n loop->stop_flag = 0;\r\n\r\n return r;\r\n}\r\n```\r\n\r\n可以看到 uv loop 里面其实就是在不断的循环去更新计时器、处理各种类型的回调、轮询 I/O 事件，Node.js 的异步便是通过 uv loop 完成的。\r\n\r\nlibuv 的异步采用的是 [Reactor][] 模型进行多路复用，在 [uv__io_poll][] 中去处理 I/O 相关的事件， [uv__io_poll][] 在不同的平台下通过 [epoll][]、[kqueue][] 等不同的方式实现。所以当往 [async_wfd][] 写入内容时，在 [uv__io_poll][] 中将会轮询到 [async_wfd][] 可读的事件，这个事件仅仅是用来通知 uv loop 线程：非 uv loop 线程有回调需要在 uv loop 线程执行。\r\n\r\n当轮询到 [async_wfd][] 可读后，[uv__io_poll][] 会回调对应的函数 [uv__async_io][]，它主要做了下面 2 件事：\r\n\r\n1. 读取数据，确认是否有 [uv_async_send][] 调用，数据内容并不关心。\r\n2. 遍历 [async_handles][] 句柄队列，判断是否有事件，如果有的话执行它的回调。\r\n\r\n实现如下：\r\n\r\n```c\r\nstatic void uv__async_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {\r\n char buf[1024];\r\n ssize_t r;\r\n QUEUE queue;\r\n QUEUE* q;\r\n uv_async_t* h;\r\n\r\n assert(w == &loop->async_io_watcher);\r\n\r\n // 这个 for 循环用来确认是否有 uv_async_send 调用\r\n for (;;) {\r\n r = read(w->fd, buf, sizeof(buf));\r\n\r\n if (r == sizeof(buf))\r\n continue;\r\n\r\n if (r != -1)\r\n break;\r\n\r\n if (errno == EAGAIN || errno == EWOULDBLOCK)\r\n break;\r\n\r\n if (errno == EINTR)\r\n continue;\r\n\r\n abort();\r\n }\r\n \r\n // 交换 loop->async_handle 和 queue内容，避免在遍历 loop->async_handles 时有新的 async_handle 插入到队列\r\n // loop->async_handles 队列中除了线程池的句柄还有其它的\r\n QUEUE_MOVE(&loop->async_handles, &queue);\r\n while (!QUEUE_EMPTY(&queue)) {\r\n q = QUEUE_HEAD(&queue);\r\n h = QUEUE_DATA(q, uv_async_t, queue);\r\n\r\n QUEUE_REMOVE(q);\r\n // 将 uv_async_t 重新插入到 loop->async_handles 中，uv_async_t 需要手动调用 uv__async_stop 才会从队列中移除\r\n QUEUE_INSERT_TAIL(&loop->async_handles, q);\r\n\r\n // 确认这个 async_handle 是否需要回调\r\n if (cmpxchgi(&h->pending, 1, 0) == 0)\r\n continue;\r\n\r\n if (h->async_cb == NULL)\r\n continue;\r\n\r\n // 调用通过 uv_async_init 初始化 uv_async_t 时绑定的回调函数\r\n // 线程池的 uv_async_t 是在 uv_loop_init 时初始化的，它绑定的回调是 uv__work_done\r\n // 因此如果 h == loop->wq_async，这里 h->async_cb 实际是调用了 uv__work_done(h);\r\n // 详情请参考 https://github.com/libuv/libuv/blob/v1.24.0/src/unix/loop.c#L88\r\n h->async_cb(h);\r\n }\r\n}\r\n```\r\n\r\n调用线程池的 `h->async_cb` 后会回到线程池的 [uv__work_done][] 函数：\r\n\r\n```c\r\nvoid uv__work_done(uv_async_t* handle) {\r\n struct uv__work* w;\r\n uv_loop_t* loop;\r\n QUEUE* q;\r\n QUEUE wq;\r\n int err;\r\n\r\n loop = container_of(handle, uv_loop_t, wq_async);\r\n uv_mutex_lock(&loop->wq_mutex);\r\n // 清空已完成的 loop->wq 队列\r\n QUEUE_MOVE(&loop->wq, &wq);\r\n uv_mutex_unlock(&loop->wq_mutex);\r\n\r\n while (!QUEUE_EMPTY(&wq)) {\r\n q = QUEUE_HEAD(&wq);\r\n QUEUE_REMOVE(q);\r\n\r\n w = container_of(q, struct uv__work, wq);\r\n // 如果在回调前调用了 uv_cancel 取消请求，则即使请求已经执行完，依旧算出错\r\n err = (w->work == uv__cancelled) ? UV_ECANCELED : 0;\r\n w->done(w, err);\r\n }\r\n}\r\n\r\n```\r\n\r\n最后通过 `w->done(w, err)` 回调 [uv__fs_done][]，并由 [uv__fs_done][] 回调 JS 函数：\r\n\r\n```c\r\nstatic void uv__fs_done(struct uv__work* w, int status) {\r\n uv_fs_t* req;\r\n\r\n req = container_of(w, uv_fs_t, work_req);\r\n uv__req_unregister(req->loop, req);\r\n\r\n // 如果取消了则抛出异常\r\n if (status == UV_ECANCELED) {\r\n assert(req->result == 0);\r\n req->result = UV_ECANCELED;\r\n }\r\n\r\n // 回调 JS\r\n req->cb(req);\r\n}\r\n```\r\n\r\n以上就是 libuv 是线程池从创建到执行多线程请求的过程。\r\n\r\n### fs.access 调用过程分析\r\n\r\n再回到文章开头提到的代码，我们来分析它的调用过程。\r\n\r\n```javascript\r\nconst fs = require('fs')\r\nconst cb = function (err) {\r\n console.log(`Is myfile exists: ${!err}`)\r\n}\r\nfs.access('myfile', cb)\r\n```\r\n\r\n假设线程池大小为 2 ，下面描述了执行 [fs.access][] 时 3 个线程的状态（略过了 Node.js 启动和 JavaScript 和 Native 函数调用过程），时间轴从上到下：\r\n\r\n**`空白`代表处于阻塞状态，`-`代表线程尚未启动**\r\n\r\n| uv loop thread | worker thread 1 | worker thread 2 |\r\n| :------------------------------------------: | :---------------: | :--------------: |\r\n| [fs.access][]('myfile', cb) | - | - |\r\n| JavaScript 通过 v8 调用 Native 函数 | - | - |\r\n| [uv_fs_access][] | - | - |\r\n| [uv__work_submit][] | - | - |\r\n| [init_threads][] | [worker][] | [worker][] |\r\n| [uv_sem_wait][] | [uv_sem_post][] | [uv_sem_post][] |\r\n| | [uv_cond_wait][] | [uv_cond_wait][] |\r\n| [uv_cond_signal][] | | |\r\n| [uv__io_poll][] | [access][] | |\r\n| [uv__io_poll][] | | |\r\n| [uv__io_poll][] | [uv_async_send][] | |\r\n| [uv__io_poll][] | [uv_cond_wait][] | |\r\n| [uv__io_poll][] | | |\r\n| [uv__async_io][] | | |\r\n| [uv__work_done][] | | |\r\n| [uv__fs_done][] | | |\r\n| Native 通过 v8 回调 JavaScript 函数 | | |\r\n| cb | | |\r\n| console.log(\\`Is myfile exists: ${exists}\\`) | | |\r\n\r\n可以看到调用过程如下：\r\n\r\n1. 通过 Node.js 启动时对 JavaScript 函数与 Native 函数的绑定，[fs.access][] 最终会进入到 Native 函数中，而 Native 函数会调用 libuv 的 [uv_fs_access][] 函数来判断文件是否可以访问。（这里略过 JavaScript 如何通过 v8 调用 Native 函数）\r\n2. [uv_fs_access][] 在 uv loop 线程向线程池提交了一个多线程请求。\r\n3. 由于线程池是惰性的，在执行请求前，先进行了初始化线程池的操作。\r\n4. 线程池初始化完成后唤醒了 `worker thread 1` 去执行请求，同时 uv loop 线程不断的轮询是否完成了请求。\r\n5. `worker thread 1` 同步的调用 [access][] 函数判断目标文件是否可读。\r\n6. [access][] 函数完成后， `worker thread 1` 通过 [uv_async_send][] 同步 uv loop 线程请求已完成，同时自身进入阻塞状态，等待新的请求将其唤醒。\r\n7. uv loop 线程发现请求执行完成后通过一系列回调回到 [uv__fs_done][]。\r\n8. [uv__fs_done][] 回调 JavaScript 函数打印日志。（这里略过 [uv__fs_done][] 是如何通过 v8 回调到 JavaScript）\r\n\r\n整个过程由于没有新的请求进来， `worker thread 2` 始终处于阻塞状态。\r\n\r\n### 结束语\r\n通过对 [fs.access][] 的调用过程分析，我们了解了 libuv 是如何通过线程池进行异步调用的。另外也可以看到针对不同的平台，libuv 对 [uv__io_poll][] 的实现是不同的，后面我们将介绍 [uv__io_poll][] 实现异步 I/O 的方式。\r\n\r\n[v11.15.0]: \r\n[1.24.0]: \r\n[dns]: \r\n[dns.lookup]: \r\n[dns.lookupService]: https://nodejs.org/dist/latest-v11.x/docs/api/dns.html#dns_dns_lookupservice_address_port_callback>\r\n[zlib]: \r\n[zlib.unzip]: \r\n[zlib.gzip]: \r\n[fs]: \r\n[fs.access]: \r\n[fs.stat]: https://nodejs.org/dist/latest-v11.x/docs/api/fs.html#fs_fs_stat_path_options_callback\r\n[Node.js v11.5.0]: https://github.com/nodejs/node/blob/v11.5.0/\r\n[libuv v1.27.0]: https://github.com/libuv/libuv/blob/v1.24.0/\r\n[init_threads]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L188\r\n[UV_THREADPOOL_SIZE]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L194\t\"设置线程数量的环境变量\"\r\n[启动时]: https://github.com/nodejs/node/blob/v11.5.0/src/node.cc#L2061\r\n[请求]: https://github.com/libuv/libuv/blob/v1.24.0/include/uv.h#L165\r\n[任务类型]: https://github.com/libuv/libuv/blob/v1.24.0/src/uv-common.h#L177\r\n[uv_fs_req_cleanup]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/fs.c#L1552\r\n[128]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L30\t\"线程最大数量\"\r\n[4]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L39\t\"线\"\r\n[uv_run]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/core.c#L350\t\"程池默认数量\"\r\n[uv_queue_work]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L338\r\n[uv__work_submit]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L256\r\n[4]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L39\t\"线程池默认数量\"\r\n[worker]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L57\t\"worker线程入口函数\"\r\n[post]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L142\r\n[uv_async_send]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/async.c#L59\r\n[uv_loop_init]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/loop.c#L30\r\n[wq_async]: \r\n[async_handles]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/loop.c#L42\r\n[uv_async_init]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/loop.c#L88\r\n[循环]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/core.c#L353\t\"uv主循环\"\r\n[Reactor]: \r\n[uv__io_poll]: \t\"uv__io_poll定义\"\r\n[epoll]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/linux-core.c#L189\r\n[kqueue]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/kqueue.c#L112\r\n[uv__async_io]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/async.c#L77\r\n[uv_async_send]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/async.c#L59\r\n[uv_async_t]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/async.c#L104\r\n[uv__work_done]: https://github.com/libuv/libuv/blob/v1.24.0/src/threadpool.c#L295\r\n[uv__fs_done]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/fs.c#L1352\r\n[uv_fs_access]: https://github.com/libuv/libuv/blob/v1.24.0/src/unix/fs.c#L1200\r\n[access]: \r\n[uv_sem_wait]: \r\n[uv_sem_post]: https://github.com/libuv/libuv/blob/v1.24.0/include/uv.h#L1536\r\n[uv_cond_wait]: \r\n[uv_cond_signal]: \r\n[async_wfd]: ","title","Node.js 是如何异步判断文件是否存在？","last_reply_at","2019-05-10 15:58:24.066","good","top","visit_count",5133,"2019-05-10 08:10:31.578","author",{"_13":14,"_22":23}]