libuv源码解析之信号处理

theanarkh

发布于 2020-01-15 03:48:31

89100

代码可运行

文章被收录于专栏：原创分享原创分享

运行总次数：0

代码可运行

libuv初始化的时候会初始化信号处理相关的逻辑。

// 保证只执行uv__signal_global_init一次
void uv__signal_global_once_init(void) {
  uv_once(&uv__signal_global_init_guard, uv__signal_global_init);
}

static void uv__signal_global_init(void) {
  if (uv__signal_lock_pipefd[0] == -1)
    // 注册fork之后，在父进程里执行的函数，保证父子进程的数据独立
    if (pthread_atfork(NULL, NULL, &uv__signal_global_reinit))
      abort();

  uv__signal_global_reinit();
}

static void uv__signal_global_reinit(void) {
  // 清除原来的（如果有的话）
  uv__signal_global_fini();
  // 新建一个管道用于互斥控制
  if (uv__make_pipe(uv__signal_lock_pipefd, 0))
    abort();
  // 先往管道写入数据，即解锁。后续才能顺利lock，unlock配对使用
  if (uv__signal_unlock())
    abort();
}

UV_DESTRUCTOR(static void uv__signal_global_fini(void)) {
  if (uv__signal_lock_pipefd[0] != -1) {
    uv__close(uv__signal_lock_pipefd[0]);
    uv__signal_lock_pipefd[0] = -1;
  }

  if (uv__signal_lock_pipefd[1] != -1) {
    uv__close(uv__signal_lock_pipefd[1]);
    uv__signal_lock_pipefd[1] = -1;
  }
}

经过一系列的操作后，主要是申请了一个用于互斥控制的管道，然后往管道里写数据。后面就可以使用lock和unlock进行加锁解锁。接着在第一个注册信号的时候，还会做一些初始化的工作。

int uv_signal_init(uv_loop_t* loop, uv_signal_t* handle) {
  int err;
  // 申请和libuv的通信管道并且注册io观察者
  err = uv__signal_loop_once_init(loop);
  if (err)
    return err;

  uv__handle_init(loop, (uv_handle_t*) handle, UV_SIGNAL);
  handle->signum = 0;
  handle->caught_signals = 0;
  handle->dispatched_signals = 0;

  return 0;
}

static int uv__signal_loop_once_init(uv_loop_t* loop) {
  int err;
  // 初始化过了
  if (loop->signal_pipefd[0] != -1)
    return 0;
  // 申请两个管道，用于其他进程和libuv主进程通信，并设置非阻塞标记
  err = uv__make_pipe(loop->signal_pipefd, UV__F_NONBLOCK);
  if (err)
    return err;
  // 设置信号io观察者的处理函数和文件描述符，libuv在poll io时，发现管道读端loop->signal_pipefd[0]可读，则执行uv__signal_event
  uv__io_init(&loop->signal_io_watcher,
              uv__signal_event,
              loop->signal_pipefd[0]);
  // 插入libuv的io观察者队列，并注册感兴趣的事件，即可读的时候，执行uv__signal_event
  uv__io_start(loop, &loop->signal_io_watcher, POLLIN);

  return 0;
}

上面的代码主要的工作有两个

1 申请一个管道，用于其他进程（libuv进程或fork出来的进程）和libuv进程通信。然后往libuv的io观察者队列注册一个观察者，libuv在poll io阶段会把观察者加到epoll中。io观察者里保存了管道读端的文件描述符loop->signal_pipefd[0]和回调函数uv__signal_event。

2 初始化信号相关的handle

接着会调uv__signal_start函数注册信号和处理函数。

static int uv__signal_start(uv_signal_t* handle,
                            uv_signal_cb signal_cb,
                            int signum,
                            int oneshot) {
  sigset_t saved_sigmask;
  int err;
  uv_signal_t* first_handle;

  assert(!uv__is_closing(handle));
  if (signum == 0)
    return UV_EINVAL;
  // 注册过了，重新设置处理函数就行
  if (signum == handle->signum) {
    handle->signal_cb = signal_cb;
    return 0;
  }
  // 这个handle之前已经设置了信号和处理函数，则先解除
  if (handle->signum != 0) {
    uv__signal_stop(handle);
  }
  // 屏蔽所有信号
  uv__signal_block_and_lock(&saved_sigmask);
  // 注册了该信号的第一个handle
  first_handle = uv__signal_first_handle(signum);
  // 之前没有注册过该信号的处理函数或者oneshot规则比之前的严格则重新修改该信号的处理规则
  if (first_handle == NULL ||
      (!oneshot && (first_handle->flags & UV_SIGNAL_ONE_SHOT))) {
    // 注册
    err = uv__signal_register_handler(signum, oneshot);
    if (err) {
      uv__signal_unlock_and_unblock(&saved_sigmask);
      return err;
    }
  }

  handle->signum = signum;
  if (oneshot)
    handle->flags |= UV_SIGNAL_ONE_SHOT;
  // 插入红黑树
  RB_INSERT(uv__signal_tree_s, &uv__signal_tree, handle);

  uv__signal_unlock_and_unblock(&saved_sigmask);

  handle->signal_cb = signal_cb;
  uv__handle_start(handle);

  return 0;
}

上面的代码比较多，大致的逻辑如下

1 给进程注册一个信号和信号处理函数。主要是调用操作系统的函数来处理的，代码如下

// 给当前进程注册信号处理函数，会覆盖之前设置的signum对应的处理函数
static int uv__signal_register_handler(int signum, int oneshot) {
  struct sigaction sa;

  memset(&sa, 0, sizeof(sa));
  // 全置一，说明收到signum信号的时候，暂时屏蔽其他信号
  if (sigfillset(&sa.sa_mask))
    abort();
  sa.sa_handler = uv__signal_handler;
  sa.sa_flags = SA_RESTART;
  // 设置了oneshot，说明信号处理函数只执行一次，然后被恢复为系统的默认处理函数
  if (oneshot)
    sa.sa_flags |= SA_RESETHAND;

  // 注册
  if (sigaction(signum, &sa, NULL))
    return UV__ERR(errno);

  return 0;
}

2 进程注册的信号和回调是在一棵红黑树管理的，每次注册的时候会往红黑树插入一个节点会修改他的节点。

至此，信号的注册就完成了。我们发现，在uv__signal_register_handler函数中有这样一句代码。

  sa.sa_handler = uv__signal_handler;

我们发现，不管注册什么信号，他的处理函数都是这个。我们自己的业务回调函数，是保存在handle里的。那么当有信号到来的时候。uv__signal_handler就会被调用。下面我们看看uv__signal_handler函数。

// 信号处理函数，signum为收到的信号，每个子进程收到信号的时候都由该函数处理，然后通过管道通知libuv
static void uv__signal_handler(int signum) {
  uv__signal_msg_t msg;
  uv_signal_t* handle;
  int saved_errno;
  // 保持上一个系统调用的错误码
  saved_errno = errno;
  memset(&msg, 0, sizeof msg);

  if (uv__signal_lock()) {
    errno = saved_errno;
    return;
  }

  for (handle = uv__signal_first_handle(signum);
       handle != NULL && handle->signum == signum;
       handle = RB_NEXT(uv__signal_tree_s, &uv__signal_tree, handle)) {
    int r;

    msg.signum = signum;
    msg.handle = handle;

    do {
      // 通知libuv，哪些handle需要处理该信号，在poll io阶段处理
      r = write(handle->loop->signal_pipefd[1], &msg, sizeof msg);
    } while (r == -1 && errno == EINTR);

    assert(r == sizeof msg ||
           (r == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)));
    // 该handle收到信号的次数
    if (r != -1)
      handle->caught_signals++;
  }

  uv__signal_unlock();
  errno = saved_errno;
}

该函数遍历红黑树，找到注册了该信号的handle，然后封装一个msg写入管道（即可libuv通信的管道）。信号的处理就完成了。接下来在libuv的poll io阶段才做真正的处理。回到文章开头我们知道在poll io阶段。epoll会检测到管道loop->signal_pipefd[0]可读，然后会执行uv__signal_event函数。我们看看这个函数的代码。

// 如果收到信号,libuv poll io阶段,会执行该函数
static void uv__signal_event(uv_loop_t* loop,
                             uv__io_t* w,
                             unsigned int events) {
  uv__signal_msg_t* msg;
  uv_signal_t* handle;
  char buf[sizeof(uv__signal_msg_t) * 32];
  size_t bytes, end, i;
  int r;

  bytes = 0;
  end = 0;

  do {
    // 独处所有的uv__signal_msg_t
    r = read(loop->signal_pipefd[0], buf + bytes, sizeof(buf) - bytes);

    if (r == -1 && errno == EINTR)
      continue;

    if (r == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
      if (bytes > 0)
        continue;
      return;
    }

    if (r == -1)
      abort();

    bytes += r;

    /* `end` is rounded down to a multiple of sizeof(uv__signal_msg_t). */
    end = (bytes / sizeof(uv__signal_msg_t)) * sizeof(uv__signal_msg_t);

    for (i = 0; i < end; i += sizeof(uv__signal_msg_t)) {
      msg = (uv__signal_msg_t*) (buf + i);
      handle = msg->handle;
      // 收到的信号和handle感兴趣的信号一致，执行回调
      if (msg->signum == handle->signum) {
        assert(!(handle->flags & UV_HANDLE_CLOSING));
        handle->signal_cb(handle, handle->signum);
      }
      // 处理信号个数
      handle->dispatched_signals++;
      // 只执行一次，恢复系统默认的处理函数
      if (handle->flags & UV_SIGNAL_ONE_SHOT)
        uv__signal_stop(handle);

      // 处理完了关闭
      if ((handle->flags & UV_HANDLE_CLOSING) &&
          (handle->caught_signals == handle->dispatched_signals)) {
        uv__make_close_pending((uv_handle_t*) handle);
      }
    }

    bytes -= end;

    if (bytes) {
      memmove(buf, buf + end, bytes);
      continue;
    }
  } while (end == sizeof buf);
}