Libuv是单线程事件驱动的异步IO库，对于阻塞式或耗时的操作，如果在Libuv的主循环里执行的话，就会阻塞后面的任务执行，所以Libuv里维护了一个线程池，它负责处理Libuv中耗时或者导致阻塞的操作，比如文件IO、DNS、自定义的耗时任务。线程池在Libuv架构中的位置如图4-1所示。
Libuv主线程通过线程池提供的接口把任务提交给线程池，然后立刻返回到事件循环中继续执行，线程池维护了一个任务队列，多个子线程会互斥地从中摘下任务节点执行，当子线程执行任务完毕后会通知主线程，主线程在事件循环的Poll IO阶段就会执行对应的回调。下面我们看一下线程池在Libuv中的实现。
Libuv子线程和主线程的通信是使用uv_async_t结构体实现的。Libuv使用loop->async_handles队列记录所有的uv_async_t结构体，使用loop->async_io_watcher作为所有uv_async_t结构体的IO观察者，即loop-> async_handles队列上所有的handle都是共享async_io_watcher这个IO观察者的。第一次插入一个uv_async_t结构体到async_handle队列时，会初始化IO观察者，如果再次注册一个async_handle，只会在loop->async_handle队列和handle队列插入一个节点，而不会新增一个IO观察者。当uv_async_t结构体对应的任务完成时，子线程会设置IO观察者为可读。Libuv在事件循环的Poll IO阶段就会处理IO观察者。下面我们看一下uv_async_t在Libuv中的使用。
使用uv_async_t之前首先需要执行uv_async_init进行初始化。
int uv_async_init(uv_loop_t* loop, uv_async_t* handle, uv_async_cb async_cb) { int err; // 给Libuv注册一个观察者io err = uv__async_start(loop); if (err) return err; // 设置相关字段，给Libuv插入一个handle uv__handle_init(loop, (uv_handle_t*)handle, UV_ASYNC); // 设置回调 handle->async_cb = async_cb; // 初始化标记字段，0表示没有任务完成 handle->pending = 0; // 把uv_async_t插入async_handle队列 QUEUE_INSERT_TAIL(&loop->async_handles, &handle->queue); uv__handle_start(handle); return 0; }
uv_async_init函数主要初始化结构体uv_async_t的一些字段，然后执行QUEUE_INSERT_TAIL给Libuv的async_handles队列追加一个节点。我们看到还有一个uv__async_start函数。我们看一下uv__async_start的实现。
static int uv__async_start(uv_loop_t* loop) { int pipefd[2]; int err; // uv__async_start只执行一次，有fd则不需要执行了 if (loop->async_io_watcher.fd != -1) return 0; // 获取一个用于进程间通信的fd（Linux的eventfd机制） err = uv__async_eventfd(); /* 成功则保存fd，失败说明不支持eventfd， 则使用管道通信作为进程间通信 */ if (err >= 0) { pipefd[0] = err; pipefd[1] = -1; } else if (err == UV_ENOSYS) { // 不支持eventfd则使用匿名管道 err = uv__make_pipe(pipefd, UV__F_NONBLOCK); #if defined(__Linux__) if (err == 0) { char buf[32]; int fd; snprintf(buf, sizeof(buf), "/proc/self/fd/%d", pipefd[0]); // 通过一个fd就可以实现对管道的读写，高级用法 fd = uv__open_cloexec(buf, O_RDWR); if (fd >= 0) { // 关掉旧的 uv__close(pipefd[0]); uv__close(pipefd[1]); // 赋值新的 pipefd[0] = fd; pipefd[1] = fd; } } #endif } // err大于等于0说明拿到了通信的读写两端 if (err < 0) return err; /* 初始化IO观察者async_io_watcher， 把读端文件描述符保存到IO观察者 */ uv__io_init(&loop->async_io_watcher, uv__async_io, pipefd[0]); // 注册IO观察者到loop里，并注册感兴趣的事件POLLIN，等待可读 uv__io_start(loop, &loop->async_io_watcher, POLLIN); // 保存写端文件描述符 loop->async_wfd = pipefd[1]; return 0; }
uv__async_start只会执行一次，时机在第一次执行uv_async_init的时候。uvasync_start主要的逻辑如下 1 获取通信描述符（通过eventfd生成一个通信的fd（充当读写两端）或者管道生成线程间通信的两个fd表示读端和写端）。 2 封装感兴趣的事件和回调到IO观察者然后追加到watcher_queue队列，在Poll IO阶段，Libuv会注册到epoll里面，如果有任务完成，也会在Poll IO阶段执行回调。 3 保存写端描述符。任务完成时通过写端fd通知主线程。 我们看到uvasync_start函数里有很多获取通信文件描述符的逻辑，总的来说，是为了完成两端通信的功能。初始化async结构体后，Libuv结构如图4-2所示。
初始化async结构体后，如果async结构体对应的任务完成后，就会通知主线程，子线程通过设置这个handle的pending为1标记任务完成，然后再往管道写端写入标记，通知主线程有任务完成了。
int uv_async_send(uv_async_t* handle) { /* Do a cheap read first. */ if (ACCESS_ONCE(int, handle->pending) != 0) return 0; /* 如pending是0，则设置为1，返回0，如果是1则返回1， 所以如果多次调用该函数是会被合并的 */ if (cmpxchgi(&handle->pending, 0, 1) == 0) uv__async_send(handle->loop); return 0; } static void uv__async_send(uv_loop_t* loop) { const void* buf; ssize_t len; int fd; int r; buf = ""; len = 1; fd = loop->async_wfd; #if defined(__Linux__) // 说明用的是eventfd而不是管道,eventfd时读写两端对应同一个fd if (fd == -1) { static const uint64_t val = 1; buf = &val; len = sizeof(val); // 见uv__async_start fd = loop->async_io_watcher.fd; /* eventfd */ } #endif // 通知读端 do r = write(fd, buf, len); while (r == -1 && errno == EINTR); if (r == len) return; if (r == -1) if (errno == EAGAIN || errno == EWOULDBLOCK) return; abort(); }
uv_async_send首先拿到写端对应的fd，然后调用write函数，此时，往管道的写端写入数据，标记有任务完成。有写则必然有读。读的逻辑是在uvio_poll中实现的。uv__io_poll函数即Libuv中Poll IO阶段执行的函数。在uvio_poll中会发现管道可读，然后执行对应的回调uv__async_io。
static void uv__async_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) { char buf[1024]; ssize_t r; QUEUE queue; QUEUE* q; uv_async_t* h; for (;;) { // 消费所有的数据 r = read(w->fd, buf, sizeof(buf)); // 数据大小大于buf长度（1024），则继续消费 if (r == sizeof(buf)) continue; // 成功消费完毕，跳出消费的逻辑 if (r != -1) break; // 读繁忙 if (errno == EAGAIN || errno == EWOULDBLOCK) break; // 读被中断，继续读 if (errno == EINTR) continue; abort(); } // 把async_handles队列里的所有节点都移到queue变量中 QUEUE_MOVE(&loop->async_handles, &queue); while (!QUEUE_EMPTY(&queue)) { // 逐个取出节点 q = QUEUE_HEAD(&queue); // 根据结构体字段获取结构体首地址 h = QUEUE_DATA(q, uv_async_t, queue); // 从队列中移除该节点 QUEUE_REMOVE(q); // 重新插入async_handles队列，等待下次事件 QUEUE_INSERT_TAIL(&loop->async_handles, q); /* 将第一个参数和第二个参数进行比较，如果相等， 则将第三参数写入第一个参数，返回第二个参数的值， 如果不相等，则返回第一个参数的值。 */ /* 判断触发了哪些async。pending在uv_async_send里设置成1， 如果pending等于1，则清0，返回1.如果pending等于0，则返回0 */ if (cmpxchgi(&h->pending, 1, 0) == 0) continue; if (h->async_cb == NULL) continue; // 执行上层回调 h->async_cb(h); } }
uv__async_io会遍历async_handles队列，pending等于1的话说明任务完成，然后执行对应的回调并清除标记位。
了解了Libuv中子线程和主线程的通信机制后，我们来看一下线程池的实现。
线程池是懒初始化的，Node.js启动的时候，并没有创建子线程，而是在提交第一个任务给线程池时，线程池才开始初始化。我们先看线程池的初始化逻辑，然后再看它的使用。
static void init_threads(void) { unsigned int i; const char* val; // 默认线程数4个，static uv_thread_t default_threads[4]; nthreads = ARRAY_SIZE(default_threads); // 判断用户是否在环境变量中设置了线程数，是的话取用户定义的 val = getenv("UV_THREADPOOL_SIZE"); if (val != NULL) nthreads = atoi(val); if (nthreads == 0) nthreads = 1; // #define MAX_THREADPOOL_SIZE 128最多128个线程 if (nthreads > MAX_THREADPOOL_SIZE) nthreads = MAX_THREADPOOL_SIZE; threads = default_threads; // 超过默认大小，重新分配内存 if (nthreads > ARRAY_SIZE(default_threads)) { threads = uv__malloc(nthreads * sizeof(threads[0])); } // 初始化条件变量，用于有任务时唤醒子线程，没有任务时挂起子线程 if (uv_cond_init(&cond)) abort(); // 初始化互斥变量，用于多个子线程互斥访问任务队列 if (uv_mutex_init(&mutex)) abort(); // 初始化三个队列 QUEUE_INIT(&wq); QUEUE_INIT(&slow_io_pending_wq); QUEUE_INIT(&run_slow_work_message); // 创建多个线程，工作函数为worker，sem为worker入参 for (i = 0; i < nthreads; i++) if (uv_thread_create(threads + i, worker, &sem)) abort(); }
线程池初始化时，会根据配置的子线程数创建对应数量的线程。默认是4个，最大128个子线程（不同版本的Libuv可能会不一样），我们也可以通过环境变量设置自定义的大小。线程池的初始化主要是初始化一些数据结构，然后创建多个线程，接着在每个线程里执行worker函数处理任务。后面我们会分析worker的逻辑。
了解线程池的初始化之后，我们看一下如何给线程池提交任务
// 给线程池提交一个任务 void uv__work_submit(uv_loop_t* loop, struct uv__work* w, enum uv__work_kind kind, void (*work)(struct uv__work* w), void (*done)(struct uv__work* w, int status)){ /* 保证已经初始化线程，并只执行一次，所以线程池是在提交第一个 任务的时候才被初始化，init_once -> init_threads */ uv_once(&once, init_once); w->loop = loop; w->work = work; w->done = done; post(&w->wq, kind); }
这里把业务相关的函数和任务完成后的回调函数封装到uvwork结构体中。uvwork结构定义如下。
struct uv__work { void (*work)(struct uv__work *w); void (*done)(struct uv__work *w, int status); struct uv_loop_s* loop; void* wq[2]; };
然后调调用post函数往线程池的队列中加入一个新的任务。Libuv把任务分为三种类型，慢IO（DNS解析）、快IO（文件操作）、CPU密集型等，kind就是说明任务的类型的。我们接着看post函数。
static void post(QUEUE* q, enum uv__work_kind kind) { // 加锁访问任务队列，因为这个队列是线程池共享的 uv_mutex_lock(&mutex); // 类型是慢IO if (kind == UV__WORK_SLOW_IO) { /* 插入慢IO对应的队列，Libuv这个版本把任务分为几种类型， 对于慢IO类型的任务，Libuv是往任务队列里面插入一个特殊的节点 run_slow_work_message，然后用slow_io_pending_wq维护了一个慢IO 任务的队列，当处理到run_slow_work_message这个节点的时候， Libuv会从slow_io_pending_wq队列里逐个取出任务节点来执行。 */ QUEUE_INSERT_TAIL(&slow_io_pending_wq, q); /* 有慢IO任务的时候，需要给主队列wq插入一个消息节点 run_slow_work_message,说明有慢IO任务，所以如果 run_slow_work_message是空，说明还没有插入主队列。需要进行 q = &run_slow_work_message;赋值，然后把 run_slow_work_message插入主队列。如果run_slow_work_message 非空，说明已经插入线程池的任务队列了。解锁然后直接返回。 */ if (!QUEUE_EMPTY(&run_slow_work_message)) { uv_mutex_unlock(&mutex); return; } // 说明run_slow_work_message还没有插入队列，准备插入队列 q = &run_slow_work_message; } // 把节点插入主队列，可能是慢IO消息节点或者一般任务 QUEUE_INSERT_TAIL(&wq, q); /* 有空闲线程则唤醒它，如果大家都在忙， 则等到它忙完后就会重新判断是否还有新任务 */ if (idle_threads > 0) uv_cond_signal(&cond); // 操作完队列，解锁 uv_mutex_unlock(&mutex); }
这就是Libuv中线程池的生产者逻辑。任务队列的架构如图4-3所示。
除了上面提到的，Libuv还提供了另外一种生产任务的方式，即uv_queue_work函数，它只提交CPU密集型的任务（在Node.js的crypto模块中使用）。下面我们看uv_queue_work的实现。
int uv_queue_work(uv_loop_t* loop, uv_work_t* req, uv_work_cb work_cb, uv_after_work_cb after_work_cb) { if (work_cb == NULL) return UV_EINVAL; uv__req_init(loop, req, UV_WORK); req->loop = loop; req->work_cb = work_cb; req->after_work_cb = after_work_cb; uv__work_submit(loop, &req->work_req, UV__WORK_CPU, uv__queue_work, uv__queue_done); return 0; }
uv_queue_work函数其实也没有太多的逻辑，它保存用户的工作函数和回调到request中。然后把uvqueue_work和uv__queue_done封装到uvwork中，接着提交任务到线程池中。所以当这个任务被执行的时候。它会执行工作函数uv__queue_work。
static void uv__queue_work(struct uv__work* w) { // 通过结构体某字段拿到结构体地址 uv_work_t* req = container_of(w, uv_work_t, work_req); req->work_cb(req); }
我们看到uv__queue_work其实就是对用户定义的任务函数进行了封装。这时候我们可以猜到，uv__queue_done也只是对用户回调的简单封装，即它会执行用户的回调。
我们提交了任务后，线程自然要处理，初始化线程池的时候我们分析过，worker函数是负责处理任务。我们看一下worker函数的逻辑。
static void worker(void* arg) { struct uv__work* w; QUEUE* q; int is_slow_work; // 线程启动成功 uv_sem_post((uv_sem_t*) arg); arg = NULL; // 加锁互斥访问任务队列 uv_mutex_lock(&mutex); for (;;) { /* 1 队列为空 2 队列不为空，但是队列中只有慢IO任务且正在执行的慢IO任务 个数达到阈值则空闲线程加一，防止慢IO占用过多线程，导致 其它快的任务无法得到执行 */ while (QUEUE_EMPTY(&wq) || (QUEUE_HEAD(&wq) == &run_slow_work_message && QUEUE_NEXT(&run_slow_work_message) == &wq && slow_io_work_running >= slow_work_thread_threshold())) { idle_threads += 1; // 阻塞，等待唤醒 uv_cond_wait(&cond, &mutex); // 被唤醒，开始干活，空闲线程数减一 idle_threads -= 1; } // 取出头结点，头指点可能是退出消息、慢IO，一般请求 q = QUEUE_HEAD(&wq); // 如果头结点是退出消息，则结束线程 if (q == &exit_message) { /* 唤醒其它因为没有任务正阻塞等待任务的线程， 告诉它们准备退出 */ uv_cond_signal(&cond); uv_mutex_unlock(&mutex); break; } // 移除节点 QUEUE_REMOVE(q); // 重置前后指针 QUEUE_INIT(q); is_slow_work = 0; /* 如果当前节点等于慢IO节点，上面的while只判断了是不是只有慢 IO任务且达到阈值，这里是任务队列里肯定有非慢IO任务，可能有 慢IO，如果有慢IO并且正在执行的个数达到阈值，则先不处理该慢 IO任务，继续判断是否还有非慢IO任务可执行。 */ if (q == &run_slow_work_message) { // 达到阈值，该节点重新入队，因为刚才被删除了 if (slow_io_work_running >= slow_work_thread_threshold()) { QUEUE_INSERT_TAIL(&wq, q); continue; } /* 没有慢IO任务则继续，这时候run_slow_work_message 已经从队列中被删除，下次有慢IO的时候重新入队 */ if (QUEUE_EMPTY(&slow_io_pending_wq)) continue; // 有慢IO，开始处理慢IO任务 is_slow_work = 1; /* 正在处理慢IO任务的个数累加，用于其它线程判断慢IO任务个 数是否达到阈值, slow_io_work_running是多个线程共享的变量 */ slow_io_work_running++; // 摘下一个慢IO任务 q = QUEUE_HEAD(&slow_io_pending_wq); // 从慢IO队列移除 QUEUE_REMOVE(q); QUEUE_INIT(q); /* 取出一个任务后，如果还有慢IO任务则把慢IO标记节点重新入 队，表示还有慢IO任务，因为上面把该标记节点出队了 */ if (!QUEUE_EMPTY(&slow_io_pending_wq)) { QUEUE_INSERT_TAIL(&wq, &run_slow_work_message); // 有空闲线程则唤醒它，因为还有任务处理 if (idle_threads > 0) uv_cond_signal(&cond); } } // 不需要操作队列了，尽快释放锁 uv_mutex_unlock(&mutex); // q是慢IO或者一般任务 w = QUEUE_DATA(q, struct uv__work, wq); // 执行业务的任务函数，该函数一般会阻塞 w->work(w); // 准备操作loop的任务完成队列，加锁 uv_mutex_lock(&w->loop->wq_mutex); // 置空说明执行完了，见cancel逻辑 w->work = NULL; /* 执行完任务,插入到loop的wq队列,在uv__work_done的时候会 执行该队列的节点 */ QUEUE_INSERT_TAIL(&w->loop->wq, &w->wq); // 通知loop的wq_async节点 uv_async_send(&w->loop->wq_async); uv_mutex_unlock(&w->loop->wq_mutex); // 为下一轮操作任务队列加锁 uv_mutex_lock(&mutex); /* 执行完慢IO任务，记录正在执行的慢IO个数变量减1， 上面加锁保证了互斥访问这个变量 */ if (is_slow_work) { slow_io_work_running--; } } }
我们看到消费者的逻辑似乎比较复杂，对于慢IO类型的任务，Libuv限制了处理慢IO任务的线程数，避免耗时比较少的任务得不到处理。其余的逻辑和一般的线程池类似，就是互斥访问任务队列，然后取出节点执行，执行完后通知主线程。结构如图4-4所示。
线程执行完任务后，并不是直接执行用户回调，而是通知主线程，由主线程统一处理，这是Node.js单线程事件循环的要求，也避免了多线程带来的复杂问题，我们看一下这块的逻辑。一切要从Libuv的初始化开始
uv_default_loop();-> uv_loop_init();-> uv_async_init(loop, &loop->wq_async, uv__work_done);
刚才我们已经分析过主线程和子线程的通信机制，wq_async是用于线程池中子线程和主线程通信的async handle，它对应的回调是uv__work_done。所以当一个线程池的线程任务完成时，通过uv_async_send(&w->loop->wq_async)设置loop->wq_async.pending = 1，然后通知IO观察者，Libuv在Poll IO阶段就会执行该handle对应的回调uv__work_done函数。那么我们就看看这个函数的逻辑。
void uv__work_done(uv_async_t* handle) { struct uv__work* w; uv_loop_t* loop; QUEUE* q; QUEUE wq; int err; // 通过结构体字段获得结构体首地址 loop = container_of(handle, uv_loop_t, wq_async); // 准备处理队列，加锁 uv_mutex_lock(&loop->wq_mutex); /* loop->wq是已完成的任务队列。把loop->wq队列的节点全部移到 wp变量中，这样一来可以尽快释放锁 */ QUEUE_MOVE(&loop->wq, &wq); // 不需要使用了，解锁 uv_mutex_unlock(&loop->wq_mutex); // wq队列的节点来自子线程插入 while (!QUEUE_EMPTY(&wq)) { q = QUEUE_HEAD(&wq); QUEUE_REMOVE(q); w = container_of(q, struct uv__work, wq); // 等于uv__canceled说明这个任务被取消了 err = (w->work == uv__cancelled) ? UV_ECANCELED : 0; // 执行回调 w->done(w, err); } }
该函数的逻辑比较简单，逐个处理已完成的任务节点，执行回调，在Node.js中，这里的回调是C++层，然后再到JS层。结构图如图4-5所示。
线程池的设计中，取消任务是一个比较重要的能力，因为在线程里执行的都是一些耗时或者引起阻塞的操作，如果能及时取消一个任务，将会减轻很多没必要的处理。不过Libuv实现中，只有当任务还在等待队列中才能被取消，如果一个任务正在被线程处理，则无法取消了。我们先看一下Libuv中是如何实现取消任务的。Libuv提供了uv__work_cancel函数支持用户取消提交的任务。我们看一下它的逻辑。
static int uv__work_cancel(uv_loop_t* loop, uv_req_t* req, struct uv__work* w) { int cancelled; // 加锁，为了把节点移出队列 uv_mutex_lock(&mutex); // 加锁，为了判断w->wq是否为空 uv_mutex_lock(&w->loop->wq_mutex); /* cancelled为true说明任务还在线程池队列等待处理 1 处理完，w->work == NULL 2 处理中，QUEUE_EMPTY(&w->wq)为true，因 为worker在摘下一个任务的时候，重置prev和next指针 3 未处理，!QUEUE_EMPTY(&w->wq)是true 且w->work != NULL */ cancelled = !QUEUE_EMPTY(&w->wq) && w->work != NULL; // 从线程池任务队列中删除该节点 if (cancelled) QUEUE_REMOVE(&w->wq); uv_mutex_unlock(&w->loop->wq_mutex); uv_mutex_unlock(&mutex); // 正在执行或者已经执行完了，则不能取消 if (!cancelled) return UV_EBUSY; // 打取消标记，Libuv执行回调的时候用到 w->work = uv__cancelled; uv_mutex_lock(&loop->wq_mutex); /* 插入loop的wq队列，对于取消的动作，Libuv认为是任务执行完了。 所以插入已完成的队列，执行回调的时候会通知用户该任务的执行结果 是取消，错误码是UV_ECANCELED */ QUEUE_INSERT_TAIL(&loop->wq, &w->wq); // 通知主线程有任务完成 uv_async_send(&loop->wq_async); uv_mutex_unlock(&loop->wq_mutex); return 0; }
在Libuv中，取消任务的方式就是把节点从线程池待处理队列中删除，然后打上取消的标记（w->work = uvcancelled），接着把该节点插入已完成队列，Libuv在处理已完成队列的节点时，判断如果w->work == uvcancelled则在执行用户回调时，传入错误码UV_ECANCELED，我们看到uv__work_cancel这个函数定义前面加了一个static，说明这个函数是只在本文件内使用的，Libuv对外提供的取消任务的接口是uv_cancel。
Copyright© 2013-2020
All Rights Reserved 京ICP备2023019179号-8