nodejs学习笔记——node与libuv

Post on Aug 23, 2017 by Lightfish Zhang

前言

node的最重要的事件循环event loop，是通过libuv库实现的，这是google开发的C语言库，兼容不同的操作系统

事件类型

  • Network I/O事件， 根据OS平台不同，分别使用Linux上的epoll，OSX和BSD类OS上的kqueue，SunOS上的event ports以及Windows上的IOCP机制。
  • File I/O事件，则使用thread pool，在其中进行阻塞式I/O。利用thread pool的方式实现异步请求处理，在各类OS上都能获得很好的支持。
    • 社区有人提出，在Linux下使用原生的NIO(非阻塞io)替换thread pool的建议 ,测试发现有3%的提升. 考虑到 NIO 对内核版本的依赖，利用thread pool的方式实现异步请求处理，在各类OS上都能获得很好的支持，应该是libuv开发者权衡再三的结果。
  • 时间事件， 在event loop处理完后，从时间事件的表中获取下一次最近事件的时间间隔timeout，将timeout设置为等待事件的超时时间uv_run(loop, timeout);，周而复始

源码分析

学习资源

官方文档 社区文档 libuv常见示例

从Linux角度去看

  • 笔者是Linux爱好者（其实不懂其他操作系统），基于Linux环境下去研究libuv源码
  • libuv中对Linux封装的代码主要声明在libuv/src/unix/linux-core.c
  • 笔者一开始就去找epoll系统调用，发现libuv使用了syscall系统调用
    • 提高UNIX环境下的用户程序移植性
    • 进程可以跳转到的内核位置叫做sysem_call。这个过程检查系统调用号，这个号码告诉内核进程请求哪种服务。然后，它查看系统调用表sys_call_table找到所调用的内核函数入口地址。接着，就调用函数，等返回后，做一些系统检查，最后返回到进程
int uv__epoll_create(int size) {
#if defined(__NR_epoll_create)
  return syscall(__NR_epoll_create, size);
#else
  return errno = ENOSYS, -1;
#endif
}

int uv__epoll_ctl(int epfd, int op, int fd, struct uv__epoll_event* events) {
#if defined(__NR_epoll_ctl)
  return syscall(__NR_epoll_ctl, epfd, op, fd, events);
#else
  return errno = ENOSYS, -1;
#endif
}

int uv__epoll_wait(int epfd,
                   struct uv__epoll_event* events,
                   int nevents,
                   int timeout) {
#if defined(__NR_epoll_wait)
  return syscall(__NR_epoll_wait, epfd, events, nevents, timeout);
#else
  return errno = ENOSYS, -1;
#endif
}
  • 另外，多进程之间分享文件描述符的sendmsg/resvmsg
int uv__sendmmsg(int fd,
                 struct uv__mmsghdr* mmsg,
                 unsigned int vlen,
                 unsigned int flags);
int uv__utimesat(int dirfd,
                 const char* path,
                 const struct timespec times[2],
                 int flags);
  • 获取时间int clock_gettime(clockid_t clk_id, struct timespec *tp)
  • 修改进程的名字int prctl(PR_SET_NAME, title)
  • 获取cpu信息，直接通过文件路径获取 
    static unsigned long read_cpufreq(unsigned int cpunum) {
unsigned long val;
char buf[1024];
FILE* fp;

snprintf(buf,
         sizeof(buf),
         "/sys/devices/system/cpu/cpu%u/cpufreq/scaling_cur_freq",
         cpunum);

fp = fopen(buf, "r");
if (fp == NULL)
  return 0;

if (fscanf(fp, "%lu", &val) != 1)
  val = 0;

fclose(fp);

return val;
}
  • 获取进程使用物理内存的大小，通过读取文件/proc/self/stat获取procinfo结构体，里面包含rss成员，根据结构体字节对齐的规则，找到rss偏移量，读取rss。笔者奇怪的是，为什么不直接声明一个结构体指针，通过procinfo.rss读取呢？
int uv_resident_set_memory(size_t* rss) {
    //...
}
  • 获取执行文件的路径readlink("/proc/self/exe", buffer, n)
  • 多线程程序使用epoll时，要注意的信号掩码处理，最好使用epoll_pwait替代epoll_wait，看下libuv对epoll_wait的封装，errno == ENOSYS表示系统没有这个接口调用
void uv__io_poll(uv_loop_t* loop, int timeout) {
  //....
  for (;;) {
    if (sigmask != 0 && no_epoll_pwait != 0)
      if (pthread_sigmask(SIG_BLOCK, &sigset, NULL))
        abort();

    if (sigmask != 0 && no_epoll_pwait == 0) {
      nfds = uv__epoll_pwait(loop->backend_fd,
                             events,
                             ARRAY_SIZE(events),
                             timeout,
                             sigmask);
      if (nfds == -1 && errno == ENOSYS)
        no_epoll_pwait = 1;
    } else {
      nfds = uv__epoll_wait(loop->backend_fd,
                            events,
                            ARRAY_SIZE(events),
                            timeout);
      if (nfds == -1 && errno == ENOSYS)
        no_epoll_wait = 1;
    }

看看libuv是怎么工作的

  • 一个比较简单的例子，查看本机文件内容uvcat
#include <assert.h>
#include <stdio.h>
#include <fcntl.h>
#include <unistd.h>
#include <uv.h>

void on_read(uv_fs_t *req);

uv_fs_t open_req;
uv_fs_t read_req;
uv_fs_t write_req;

static char buffer[1024];

static uv_buf_t iov;

void on_write(uv_fs_t *req) {
    if (req->result < 0) {
        fprintf(stderr, "Write error: %s\n", uv_strerror((int)req->result));
    }
    else {
        uv_fs_read(uv_default_loop(), &read_req, open_req.result, &iov, 1, -1, on_read);
    }
}

void on_read(uv_fs_t *req) {
    if (req->result < 0) {
        fprintf(stderr, "Read error: %s\n", uv_strerror(req->result));
    }
    else if (req->result == 0) {
        uv_fs_t close_req;
        // synchronous
        uv_fs_close(uv_default_loop(), &close_req, open_req.result, NULL);
    }
    else if (req->result > 0) {
        iov.len = req->result;
        // 第三个参数是文件描述符，将iov缓存写到标准输出
        uv_fs_write(uv_default_loop(), &write_req, 1, &iov, 1, -1, on_write);
    }
}

void on_open(uv_fs_t *req) {
    // The request passed to the callback is the same as the one the call setup
    // function was passed.
    assert(req == &open_req);
    if (req->result >= 0) { // req->result为open返回的文件描述符
        iov = uv_buf_init(buffer, sizeof(buffer));
        /*
        # 读取文件
        - 使用uv的默认loop
        - 使用uv封装的`uv_fs_s`
        - 目标文件的文件描述符
        - flag
        - mode
        - 回调函数 on_read
        */
        uv_fs_read(uv_default_loop(), &read_req, req->result,
                   &iov, 1, -1, on_read);
    }
    else {
        fprintf(stderr, "error opening file: %s\n", uv_strerror((int)req->result));
    }
}

int main(int argc, char **argv) {
    /*
    # 打开文件
    - 使用uv的默认loop
    - 使用uv封装的`uv_fs_s`，打开文件成功后，成员变量result为文件描述符
    - 路径名
    - flag 只读
    - mode
    - 回调函数 on_open
    */
    uv_fs_open(uv_default_loop(), &open_req, argv[1], O_RDONLY, 0, on_open);

    // uv的执行，当loop中的事件都处理完毕，返回
    uv_run(uv_default_loop(), UV_RUN_DEFAULT);

    uv_fs_req_cleanup(&open_req);
    uv_fs_req_cleanup(&read_req);
    uv_fs_req_cleanup(&write_req);
    return 0;
}
  • 这个例子中，uv_run()执行，其内部是while循环，UV_RUN_DEFAULT表示事件都处理完后，跳出循环，uv_run()返回
  • 这个例子的代码不难看，沿着回调函数一直看就行，如果程序回调函数一多，就容易陷入回调地狱，像node.js还没有async/await或者yeild的时候，也容易陷入这个境地

如何加入文件事件

  • uv_default_loop()第一次执行，初始化工作，主要工作有:
    • uv__platform_loop_init()对操作系统的io多路复用的事件监听句柄的初始化
    • uv_rwlock_init()与uv_mutex_init()对libuv的锁与互斥量初始化
  • uv_fs_open()的执行过程，源码中, 主要执行两个代码段的宏定义INIT,POST, INIT是初始化一些变量，有意思的是POST,派发事件到loop的过程：
  • 文件事件，libuv的宏定义POST的执行过程：
#define POST                                                                  \
  do {                                                                        \
    if ((cb) != NULL) {                                                       \
      uv__work_submit((loop), &(req)->work_req, uv__fs_work, uv__fs_done);    \
      return 0;                                                               \
    }                                                                         \
    else {                                                                    \
      uv__fs_work(&(req)->work_req);                                          \
      uv__fs_done(&(req)->work_req, 0);                                       \
      return (req)->result;                                                   \
    }                                                                         \
  }                                                                           \
  while (0)
  • uv__work_submit()会调用uv_once(&once, init_once);，其实uv_once()是pthread_once()的封装，使init_once()这个函数在一个进程（多个线程）中只执行一次，init_once()是libuv的线程池的初始化
static void init_once(void) {
    //...
  for (i = 0; i < nthreads; i++)
    if (uv_thread_create(threads + i, worker, NULL))
      abort();
  • uv__work_submit()接下来会调用post()，抢占锁后，把文件事件的加入队列

多线程处理文件事件

  • uv_run()->uv__io_poll()
    • Linux环境下，调用epoll接口，但是libuv的文件事件不使用epoll，而是使用线程池去执行本机文件io任务
    • 除了epoll的调用外，uv__io_poll()还处理的队列中数据，已完成任务或者任务之后还有回调需要继续执行等，都处理好
  • libuv/src/threadpool.c的静态函数worker()，就是libuv的线程池的各个子线程执行体
    • 当wq队列为空时，线程阻塞等待条件变量uv_cond_wait()，Linux是pthread_cond_wait()，细节是：释放mutex锁，进入条件变量等待的队列，获取到条件变量的同时占用锁mutex.
    • 获取到条件变量，也就是wq队列不为空，子线程从队列QUEUE_HEAD(&wq)获取任务，然后释放锁mutex
    • 获得任务就执行呗，w->work(w)，处理完后，将执行完后的数据发送uv_async_send(&w->loop->wq_async)，然后子线程继续循环之前的步骤