• Libuv源码解析 - uv_run

    Libuv源码解析 - uv_run

    uv_run

    int uv_run(uv_loop_t *loop, uv_run_mode mode) {
  DWORD timeout;
  int r;
  int ran_pending;

  //判断事件循是否继续执行下去，即检查是否有任务
  r = uv__loop_alive(loop);
  //如果没有任务执行，那么即将退出，更新一下当前循环的时间
  if (!r)
    uv_update_time(loop);

  while (r != 0 && loop->stop_flag == 0) {
    //更新一下循环时间，这一轮循环的剩下操作可能使用这个变量获取当前时间，避免过多的系统调用影响性能
    uv_update_time(loop);
    //执行计时器队列超时回调
    uv__run_timers(loop);

    //执行pending队列回调，ran_pending表示pending队列是否为空，即是否没有节点可以执行
    ran_pending = uv_process_reqs(loop);    //一般来说，所有的io回调（网络，文件，dns）
                                            //都会在io轮询阶段执行。但是有的情况下，
                                            //io轮询阶段的回调会延迟到下一次循环执行，
                                            //那么这种回调就是在pending阶段执行的。
                                            //注意，当前版本所有异步io事件统一在这执行。

    //执行idle队列，遍历双链表执行回调事件
    uv_idle_invoke(loop);
    //执行prepare队列，遍历双链表执行回调事件
    uv_prepare_invoke(loop);

    timeout = 0;
    //执行的模式为UV_RUN_ONCE的时候，且当前这次loop循环没有pending队列任务
    //如果当前这次loop循环的pending队列有任务处理完，那么即使是UV_RUN_ONCE模式，
    //也不进行阻塞io，因为既然处理了io回调事件，那么响应方可能会再次响应我方
    //才会触发阻塞式poll io，同时默认模式也是这样
    if ((mode == UV_RUN_ONCE && !ran_pending) || mode == UV_RUN_DEFAULT)
      timeout = uv_backend_timeout(loop);   //计算阻塞超时时间(最长等待时间)
    
    //是否一次性处理多个io完成数据包
    if (pGetQueuedCompletionStatusEx)
      uv__poll(loop, timeout);              //一次性处理多个io完成数据包，性能效率高
    else
      uv__poll_wine(loop, timeout);         //一次只处理一个io完成数据包

    /* Run one final update on the provider_idle_time in case uv__poll*
     * returned because the timeout expired, but no events were received. This
     * call will be ignored if the provider_entry_time was either never set (if
     * the timeout == 0) or was already updated b/c an event was received.
     */                                     
    //如果对io轮询的循环进行额外配置，计算进程空闲时间
    //防止在多线程的情况下，UV_METRICS_IDLE_TIME状态可能不断更改，未记录到最终的空闲时间
    uv__metrics_update_idle_time(loop);

    uv_check_invoke(loop);                  //执行check队列，遍历双链表执行回调事件
    uv_process_endgames(loop);              //执行close回调队列

    if (mode == UV_RUN_ONCE) {
      /* UV_RUN_ONCE implies forward progress: at least one callback must have
       * been invoked when it returns. uv__io_poll() can return without doing
       * I/O (meaning: no callbacks) when its timeout expires - which means we
       * have pending timers that satisfy the forward progress constraint.
       *
       * UV_RUN_NOWAIT makes no guarantees about progress so it's omitted from
       * the check.
       */
      uv__run_timers(loop);                //由于可能是io轮询超时导致io轮询结束，
                                           //这是如果loop循环的模式为UV_RUN_ONCE，仍然是有一次执行定时器处理的机会
    }

    r = uv__loop_alive(loop);             //判断事件循是否继续执行下去，即检查是否有任务
    //只执行一次，退出循环，UV_RUN_NOWAIT表示在io轮询阶段不会阻塞并且循环只执行一次
    if (mode == UV_RUN_ONCE || mode == UV_RUN_NOWAIT)
      break;
  }

  /* The if statement lets the compiler compile it to a conditional store.
   * Avoids dirtying a cache line.
   */
  if (loop->stop_flag != 0) //重置标记位
    loop->stop_flag = 0;

  //返回是否还有活跃的任务（handles或者reqs），业务层可以根据需要选择是否再次执行该loop
  return r;
}

    • 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

    uv__hrtime

    uint64_t uv__hrtime(unsigned int scale) {
  LARGE_INTEGER counter;
  double scaled_freq;
  double result;

  assert(hrtime_frequency_ != 0);
  assert(scale != 0);
  //检索性能计数器的当前值，这是一个高分辨率 (<1us) 时间戳，可用于时间间隔度量。
  if (!QueryPerformanceCounter(&counter)) {
    uv_fatal_error(GetLastError(), "QueryPerformanceCounter");
  }
  assert(counter.QuadPart != 0);

  /* Because we have no guarantee about the order of magnitude of the
   * performance counter interval, integer math could cause this computation
   * to overflow. Therefore we resort to floating point math.
   */
  //因为我们无法保证性能计数器间隔的数量级，整数数学可能会导致此计算溢出。因此，我们采用
  //浮点数学。
  scaled_freq = (double) hrtime_frequency_ / scale;
  result = (double) counter.QuadPart / scaled_freq;
  return (uint64_t) result;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    uv__run_timers

    void uv__run_timers(uv_loop_t* loop) {
  struct heap_node* heap_node;                 //小顶堆结构
  uv_timer_t* handle;

  for (;;) {
    heap_node = heap_min(timer_heap(loop));     //获取小顶堆最小的结构
    if (heap_node == NULL)
      break;

    //字节偏移获取uv_timer_t结构首地址
    handle = container_of(heap_node, uv_timer_t, heap_node);  
    //由于整个过程中loop->time都未update，即每个定时器是否超时只能依据loop->time时间
    //即不会出现在某个定时器事件中添加断点等操作，导致定时器执行时间过长，从而导致某个
    //定时器事件无限循环
    if (handle->timeout > loop->time)           //最小元素的时间都大于当前loop循环的时间
      break;

    uv_timer_stop(handle);                      //停止当前定时器计时，防止出现时间延时
    uv_timer_again(handle);                     //停止定时器并把原先的timeout和repeat值都设置为之前的repeat值
    handle->timer_cb(handle);                   //执行回调方法
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    uv_process_reqs

    //没有pending节点返回0，有则返回1
INLINE static int uv_process_reqs(uv_loop_t* loop) {
  //pending队列结构为单向环形链表
  uv_req_t* req;
  uv_req_t* first;
  uv_req_t* next;

  if (loop->pending_reqs_tail == NULL)
    return 0;

  first = loop->pending_reqs_tail->next_req;
  next  = first;
  loop->pending_reqs_tail = NULL;               //置空

  while (next != NULL) {
    req  = next;
    next = req->next_req != first ? req->next_req : NULL;

    //根据处理类型进行相应处理
    switch (req->type) {
      case UV_READ:         //可读
        DELEGATE_STREAM_REQ(loop, req, read, data);
        break;

      case UV_WRITE:        //可写
        DELEGATE_STREAM_REQ(loop, (uv_write_t*) req, write, handle);
        break;

      case UV_ACCEPT:       //接收连接请求
        DELEGATE_STREAM_REQ(loop, req, accept, data);
        break;

      case UV_CONNECT:      //发送连接请求
        DELEGATE_STREAM_REQ(loop, (uv_connect_t*) req, connect, handle);
        break;

      case UV_SHUTDOWN:     //关闭请求
        /* Tcp shutdown requests don't come here. */
        assert(((uv_shutdown_t*) req)->handle->type == UV_NAMED_PIPE);
        uv_process_pipe_shutdown_req(
            loop,
            (uv_pipe_t*) ((uv_shutdown_t*) req)->handle,
            (uv_shutdown_t*) req);
        break;

      case UV_UDP_RECV:     //udp接收请求
        uv_process_udp_recv_req(loop, (uv_udp_t*) req->data, req);
        break;

      case UV_UDP_SEND:     //udp发送请求
        uv_process_udp_send_req(loop,
                                ((uv_udp_send_t*) req)->handle,
                                (uv_udp_send_t*) req);
        break;

      case UV_WAKEUP:       //线程之间通讯，类比唤醒一个线程（线程安全）
        uv_process_async_wakeup_req(loop, (uv_async_t*) req->data, req);
        break;

      case UV_SIGNAL_REQ:   //处理信号
        uv_process_signal_req(loop, (uv_signal_t*) req->data, req);
        break;

      case UV_POLL_REQ:     //处理轮询的文件描述符
        uv_process_poll_req(loop, (uv_poll_t*) req->data, req);
        break;

      case UV_PROCESS_EXIT:  //处理进程退出事件
        uv_process_proc_exit(loop, (uv_process_t*) req->data);
        break;

      case UV_FS_EVENT_REQ:  //处理文件的变化
        uv_process_fs_event_req(loop, req, (uv_fs_event_t*) req->data);
        break;

      default:
        assert(0);
    }
  }

  return 1;
}

    • 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

    uv_process_async_wakeup_req

    //调用回调事件并复位async_sent
void uv_process_async_wakeup_req(uv_loop_t* loop, uv_async_t* handle,
    uv_req_t* req) {
  assert(handle->type == UV_ASYNC);
  assert(req->type == UV_WAKEUP);

  handle->async_sent = 0;

  if (handle->flags & UV_HANDLE_CLOSING) {
    //销毁处理，加入到endgame队列中，可以调用uv_async_close或uv_close进行触发
    uv_want_endgame(loop, (uv_handle_t*)handle); 
  } else if (handle->async_cb != NULL) {
    handle->async_cb(handle);
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    uv_process_signal_req

    void uv_process_signal_req(uv_loop_t* loop, uv_signal_t* handle,
    uv_req_t* req) {
  long dispatched_signum;

  assert(handle->type == UV_SIGNAL);
  assert(req->type == UV_SIGNAL_REQ);

  dispatched_signum = InterlockedExchange(
          (volatile LONG*) &handle->pending_signum, 0);
  assert(dispatched_signum != 0);

  /* Check if the pending signal equals the signum that we are watching for.
   * These can get out of sync when the handler is stopped and restarted while
   * the signal_req is pending. */
  //如果接收到的信号与预期信号是一致的，则执行回调方法
  if (dispatched_signum == handle->signum)
    handle->signal_cb(handle, dispatched_signum);

  if (handle->flags & UV_SIGNAL_ONE_SHOT)
    uv_signal_stop(handle);

  if (handle->flags & UV_HANDLE_CLOSING) {
    /* When it is closing, it must be stopped at this point. */
    assert(handle->signum == 0);
    uv_want_endgame(loop, (uv_handle_t*) handle);
  }

    • 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

    uv_process_poll_req

    void uv_process_poll_req(uv_loop_t* loop, uv_poll_t* handle, uv_req_t* req) {
  if (!(handle->flags & UV_HANDLE_POLL_SLOW)) {
    uv__fast_poll_process_poll_req(loop, handle, req); //快速轮询
  } else {
    uv__slow_poll_process_poll_req(loop, handle, req); //慢速轮询
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    uv_process_proc_exit

    /* Called on main thread after a child process has exited. */
//一个子进程退出后，将在主线程中响应
void uv_process_proc_exit(uv_loop_t* loop, uv_process_t* handle) {
  int64_t exit_code;
  DWORD status;

  assert(handle->exit_cb_pending);
  handle->exit_cb_pending = 0;

  /* If we're closing, don't call the exit callback. Just schedule a close
   * callback now. */
  if (handle->flags & UV_HANDLE_CLOSING) {
    uv_want_endgame(loop, (uv_handle_t*) handle); //处理销毁方法
    return;
  }

  /* Unregister from process notification. */
  //从进程通知中注销该子进程
  if (handle->wait_handle != INVALID_HANDLE_VALUE) {
    UnregisterWait(handle->wait_handle);
    handle->wait_handle = INVALID_HANDLE_VALUE;
  }

  /* Set the handle to inactive: no callbacks will be made after the exit
   * callback. */
  //将进程设置为非活跃状态：退出回调后不会再进行回调
  uv__handle_stop(handle);

  //获取退出代码，检索指定进程的终止状态
  if (GetExitCodeProcess(handle->process_handle, &status)) {
    exit_code = status;
  } else {
    /* Unable to obtain the exit code. This should never happen. */
    exit_code = uv_translate_sys_error(GetLastError());
  }

  /* Fire the exit callback. */
  //执行退出回调
  if (handle->exit_cb) {
    handle->exit_cb(handle, exit_code, handle->exit_signal);
  }
}

    • 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

    uv_process_fs_event_req

    void uv_process_fs_event_req(uv_loop_t* loop, uv_req_t* req,
    uv_fs_event_t* handle) {
  FILE_NOTIFY_INFORMATION* file_info;
  int err, sizew, size;
  char* filename = NULL;
  WCHAR* filenamew = NULL;
  WCHAR* long_filenamew = NULL;
  DWORD offset = 0;

  assert(req->type == UV_FS_EVENT_REQ);
  assert(handle->req_pending);
  handle->req_pending = 0;

  /* Don't report any callbacks if: 在以下情况下不报告任何回调
   * - We're closing, just push the handle onto the endgame queue 处于销毁状态
   * - We are not active, just ignore the callback 已经处于非活跃状态
   */
  if (!uv__is_active(handle)) {
    if (handle->flags & UV_HANDLE_CLOSING) {
      uv_want_endgame(loop, (uv_handle_t*) handle);
    }
    return;
  }

  file_info = (FILE_NOTIFY_INFORMATION*)(handle->buffer + offset);

  if (REQ_SUCCESS(req)) {
    if (req->u.io.overlapped.InternalHigh > 0) {
      do {
        file_info = (FILE_NOTIFY_INFORMATION*)((char*)file_info + offset);
        assert(!filename);
        assert(!filenamew);
        assert(!long_filenamew);

        /*
         * Fire the event only if we were asked to watch a directory,
         * or if the filename filter matches.
         */
        //只有当我们被要求监听的目录或者文件名过滤器匹配时，才调用回调事件
        if (handle->dirw ||
            file_info_cmp(handle->filew,
                          file_info->FileName,
                          file_info->FileNameLength) == 0 ||
            file_info_cmp(handle->short_filew,
                          file_info->FileName,
                          file_info->FileNameLength) == 0) {

          if (handle->dirw) {
            /*
             * We attempt to resolve the long form of the file name explicitly.
             * We only do this for file names that might still exist on disk.
             * If this fails, we use the name given by ReadDirectoryChangesW.
             * This may be the long form or the 8.3 short name in some cases.
             */
            if (file_info->Action != FILE_ACTION_REMOVED &&
              file_info->Action != FILE_ACTION_RENAMED_OLD_NAME) {
              /* Construct a full path to the file. */
              size = wcslen(handle->dirw) +
                file_info->FileNameLength / sizeof(WCHAR) + 2;

              filenamew = (WCHAR*)uv__malloc(size * sizeof(WCHAR));
              if (!filenamew) {
                uv_fatal_error(ERROR_OUTOFMEMORY, "uv__malloc");
              }

              _snwprintf(filenamew, size, L"%s\\%.*s", handle->dirw,
                file_info->FileNameLength / (DWORD)sizeof(WCHAR),
                file_info->FileName);

              filenamew[size - 1] = L'\0';

              /* Convert to long name. */
              size = GetLongPathNameW(filenamew, NULL, 0);

              if (size) {
                long_filenamew = (WCHAR*)uv__malloc(size * sizeof(WCHAR));
                if (!long_filenamew) {
                  uv_fatal_error(ERROR_OUTOFMEMORY, "uv__malloc");
                }

                size = GetLongPathNameW(filenamew, long_filenamew, size);
                if (size) {
                  long_filenamew[size] = '\0';
                } else {
                  uv__free(long_filenamew);
                  long_filenamew = NULL;
                }
              }

              uv__free(filenamew);

              if (long_filenamew) {
                /* Get the file name out of the long path. */
                uv_relative_path(long_filenamew,
                                 handle->dirw,
                                 &filenamew);
                uv__free(long_filenamew);
                long_filenamew = filenamew;
                sizew = -1;
              } else {
                /* We couldn't get the long filename, use the one reported. */
                filenamew = file_info->FileName;
                sizew = file_info->FileNameLength / sizeof(WCHAR);
              }
            } else {
              /*
               * Removed or renamed events cannot be resolved to the long form.
               * We therefore use the name given by ReadDirectoryChangesW.
               * This may be the long form or the 8.3 short name in some cases.
               */
              filenamew = file_info->FileName;
              sizew = file_info->FileNameLength / sizeof(WCHAR);
            }
          } else {
            /* We already have the long name of the file, so just use it. */
            filenamew = handle->filew;
            sizew = -1;
          }

          /* Convert the filename to utf8. */
          uv__convert_utf16_to_utf8(filenamew, sizew, &filename);

          switch (file_info->Action) {
            case FILE_ACTION_ADDED:
            case FILE_ACTION_REMOVED:
            case FILE_ACTION_RENAMED_OLD_NAME:
            case FILE_ACTION_RENAMED_NEW_NAME:
              handle->cb(handle, filename, UV_RENAME, 0);
              break;

            case FILE_ACTION_MODIFIED:
              handle->cb(handle, filename, UV_CHANGE, 0);
              break;
          }

          uv__free(filename);
          filename = NULL;
          uv__free(long_filenamew);
          long_filenamew = NULL;
          filenamew = NULL;
        }

        offset = file_info->NextEntryOffset;
      } while (offset && !(handle->flags & UV_HANDLE_CLOSING));
    } else {
      handle->cb(handle, NULL, UV_CHANGE, 0);
    }
  } else {
    err = GET_REQ_ERROR(req);
    handle->cb(handle, NULL, 0, uv_translate_sys_error(err));
  }

  if (!(handle->flags & UV_HANDLE_CLOSING)) {
    uv_fs_event_queue_readdirchanges(loop, handle);
  } else {
    uv_want_endgame(loop, (uv_handle_t*)handle);
  }
}

    • 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
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138
    • 139
    • 140
    • 141
    • 142
    • 143
    • 144
    • 145
    • 146
    • 147
    • 148
    • 149
    • 150
    • 151
    • 152
    • 153
    • 154
    • 155
    • 156
    • 157
    • 158

    uv_backend_timeout

    int uv_backend_timeout(const uv_loop_t* loop) {
  //如果loop stop标记为已经标记为停止
  if (loop->stop_flag != 0)
    return 0;

  //检查是否有活跃的handles和活跃的reqs，没有活跃的则退出，超时时间为0
  if (!uv__has_active_handles(loop) && !uv__has_active_reqs(loop))
    return 0;

  //pending回调是否为空，有io回调事件未执行则超时时间为0
  if (loop->pending_reqs_tail)
    return 0;

  //close回调是否为空，有close回调事件则超时时间为0
  if (loop->endgame_handles)
    return 0;

  //idle回调是否为空，有idle事件则超时时间为0
  if (loop->idle_handles)
    return 0;

  return uv__next_timeout(loop);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    uv__next_timeout

    int uv__next_timeout(const uv_loop_t* loop) {
  const struct heap_node* heap_node;
  const uv_timer_t* handle;
  uint64_t diff;

  heap_node = heap_min(timer_heap(loop));   //获取当前定时器队列中最小的节点
  if (heap_node == NULL)
    return -1; /* block indefinitely */     //无限期阻塞，直至有网络io事件到来

  //指针偏移获取uv_timer_t节点首地址
  handle = container_of(heap_node, uv_timer_t, heap_node);  
  if (handle->timeout <= loop->time)
    return 0;                           //超时时间为0，即从阻塞io变成非阻塞io

  diff = handle->timeout - loop->time;  //阻塞io只能阻塞定时器队列中最快执行的定时器之间的差值
  if (diff > INT_MAX)
    diff = INT_MAX;                     //最长阻塞时间

  return (int) diff;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    uv__poll

    static void uv__poll(uv_loop_t* loop, DWORD timeout) {
  BOOL success;
  uv_req_t* req;
  OVERLAPPED_ENTRY overlappeds[128];
  ULONG count;
  ULONG i;
  int repeat;
  uint64_t timeout_time;
  uint64_t user_timeout;
  int reset_timeout;
  
  timeout_time = loop->time + timeout;          //当前时间 + 超时时间，即具体退出时间

  //检查是否对io轮询的循环进行额外配置，通过在uv_run之前调用uv_loop_configure方法产生
  if (uv__get_internal_fields(loop)->flags & UV_METRICS_IDLE_TIME) {
    reset_timeout = 1;
    user_timeout = timeout;
    timeout = 0;
  } else {
    reset_timeout = 0;      //通常reset_timeout = 0
  }

  for (repeat = 0; ; repeat++) {
    /* Only need to set the provider_entry_time if timeout != 0. The function
     * will return early if the loop isn't configured with UV_METRICS_IDLE_TIME.
     */
    //只有当超时时间不为0时需要设置provider_entry_time，如果io轮询没有进行额外配置，如
    //UV_METRICS_IDLE_TIME，那么uv__metrics_set_provider_entry_time方法将直接return
    if (timeout != 0)
      uv__metrics_set_provider_entry_time(loop);

    //获取排队队列的完成状态，同时索引多个完整端口状态。
    success = pGetQueuedCompletionStatusEx(loop->iocp,      //I/O完整端口
                                           overlappeds,     //该数组包含GetQueuedCompletionStatusEx函数返回的信息
                                           ARRAY_SIZE(overlappeds),//OVERLAPPED_ENTRY数组元素的数目
                                           &count,          //返回需要处理io事件的个数
                                           timeout,         //超时事件
                                           FALSE);          //该参数为FALSE，那么意味着是阻塞IO，阻塞时间由超时时间和IO事件决定

    //如果io轮询的循环设置额外配置，那么io轮询是非阻塞io，同时超时事件重置为上一次需要阻塞的超时时间
    if (reset_timeout != 0) {
      timeout = user_timeout;
      reset_timeout = 0;
    }

    /* Placed here because on success the loop will break whether there is an
     * empty package or not, or if GetQueuedCompletionStatus returned early then
     * the timeout will be updated and the loop will run again. In either case
     * the idle time will need to be updated.
     */
    //更新空闲时间，前提是io轮询的循环设置额外配置
    uv__metrics_update_idle_time(loop);

    //如果获取排队队列的完成状态成功
    if (success) {
      for (i = 0; i < count; i++) {
        /* Package was dequeued, but see if it is not a empty package
         * meant only to wake us up.
         * 数据包已经出队，但需要提前查看它是否只是一个单纯想唤醒我们的空包
         * 如果仅仅只是一个唤醒我们的空包，I/O 完成端口不做任何处理
         */
        if (overlappeds[i].lpOverlapped) {
          //将一个重叠结构转换为uv_req_t对象，利用结构体成员计算出结构体对象首地址
          //高效数据结构，可以节省一个数据域指针的开销
          req = uv_overlapped_to_req(overlappeds[i].lpOverlapped);
          //插入pending队列单向环形链表尾部
          uv_insert_pending_req(loop, req);
        }
      }

      /* Some time might have passed waiting for I/O,
       * so update the loop time here.
       */
      //等待I/O轮询可能已经过了一段时间，所以请在此处更新循环时间。
      uv_update_time(loop);
    } else if (GetLastError() != WAIT_TIMEOUT) {
      /* Serious error */   //重大错误
      uv_fatal_error(GetLastError(), "GetQueuedCompletionStatusEx");
    } else if (timeout > 0) {
      /* GetQueuedCompletionStatus can occasionally return a little early.
       * Make sure that the desired timeout target time is reached.
       */
      //GetQueuedCompletionStatus当有IO事件发生时会提前返回
      //需要保证超时时间达到我们计算所得的超时时间才真正退出
      uv_update_time(loop);
      //如果未到达预定的超时时间则继续获取排队队列的完成状态
      if (timeout_time > loop->time) {
        //减去已经消耗掉的时间
        timeout = (DWORD)(timeout_time - loop->time);
        /* The first call to GetQueuedCompletionStatus should return very
         * close to the target time and the second should reach it, but
         * this is not stated in the documentation. To make sure a busy
         * loop cannot happen, the timeout is increased exponentially
         * starting on the third round.
         */
        //为了防止GetQueuedCompletionStatus第一次之后返回之后已经非常接近目标时间timeout_time，
        //然后第二次如果差点达到目标时间，那么按照流程应当调用第三次GetQueuedCompletionStatus
        //但这实际上只需要将第二次的timeout时间延长一点点，那么就可以不用调用第三次
        //GetQueuedCompletionStatus IO轮询，所以我们每次timeout + repeat，然后为了
        //保证高效，所以让repeat成指数增长。
        //有好处也有坏处，那就是定时器不会实时执行，存在一定的误差，应该repeat是不可控的
        timeout += repeat ? (1 << (repeat - 1)) : 0;//0,2^0,2^1,2^2...
        continue;
      }
    }
    break;
  }
}

    • 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
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108

    uv__metrics_set_provider_entry_time

    void uv__metrics_set_provider_entry_time(uv_loop_t* loop) {
  uv__loop_metrics_t* loop_metrics;
  uint64_t now;

  if (!(uv__get_internal_fields(loop)->flags & UV_METRICS_IDLE_TIME))
    return;

  now = uv_hrtime();                        //获取当前时间戳
  loop_metrics = uv__get_loop_metrics(loop);
  uv_mutex_lock(&loop_metrics->lock);       //加锁
  loop_metrics->provider_entry_time = now;  //添加进入时间（起始时间）
  uv_mutex_unlock(&loop_metrics->lock);     //解锁
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    uv__metrics_update_idle_time

    void uv__metrics_update_idle_time(uv_loop_t* loop) {
  uv__loop_metrics_t* loop_metrics;
  uint64_t entry_time;
  uint64_t exit_time;

  if (!(uv__get_internal_fields(loop)->flags & UV_METRICS_IDLE_TIME))
    return;

  loop_metrics = uv__get_loop_metrics(loop);

  /* The thread running uv__metrics_update_idle_time() is always the same
   * thread that sets provider_entry_time. So it's unnecessary to lock before
   * retrieving this value.
   */
  if (loop_metrics->provider_entry_time == 0)
    return;

  exit_time = uv_hrtime();

  uv_mutex_lock(&loop_metrics->lock);               //加锁
  entry_time = loop_metrics->provider_entry_time;   
  loop_metrics->provider_entry_time = 0;            //将起始时间置为0
  loop_metrics->provider_idle_time += exit_time - entry_time; //记录从起始时间到目前为止进程的空闲时间
  uv_mutex_unlock(&loop_metrics->lock);             //解锁
}

    • 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

    uv_process_endgames

    INLINE static void uv_process_endgames(uv_loop_t* loop) {
  uv_handle_t* handle;
  //endgame_handles为单链表结构
  while (loop->endgame_handles) {
    handle = loop->endgame_handles;
    loop->endgame_handles = handle->endgame_next;

    handle->flags &= ~UV_HANDLE_ENDGAME_QUEUED;     //设置句柄状态

    switch (handle->type) {
      case UV_TCP:          //Tcp关闭回调事件
        uv_tcp_endgame(loop, (uv_tcp_t*) handle);
        break;

      case UV_NAMED_PIPE:   //PIPE关闭回调事件
        uv_pipe_endgame(loop, (uv_pipe_t*) handle);
        break;

      case UV_TTY:          //TTY关闭回调事件
        uv_tty_endgame(loop, (uv_tty_t*) handle);
        break;

      case UV_UDP:          //UDP关闭回调事件
        uv_udp_endgame(loop, (uv_udp_t*) handle);
        break;

      case UV_POLL:         //POLL轮询关闭回调事件
        uv_poll_endgame(loop, (uv_poll_t*) handle);
        break;

      case UV_TIMER:        //定时器关闭回调事件
        uv__timer_close((uv_timer_t*) handle);
        uv__handle_close(handle);
        break;

      case UV_PREPARE:      //prepare、check、idle句柄关闭回调事件
      case UV_CHECK:
      case UV_IDLE:
        uv_loop_watcher_endgame(loop, handle);
        break;

      case UV_ASYNC:        //异步（线程之间通信）关闭回调事件
        uv_async_endgame(loop, (uv_async_t*) handle);
        break;

      case UV_SIGNAL:       //信号关闭回调事件
        uv_signal_endgame(loop, (uv_signal_t*) handle);
        break;

      case UV_PROCESS:      //进程关闭回调事件
        uv_process_endgame(loop, (uv_process_t*) handle);
        break;

      case UV_FS_EVENT:     //文件系统关闭回调事件
        uv_fs_event_endgame(loop, (uv_fs_event_t*) handle);
        break;

      case UV_FS_POLL:      //文件轮询关闭回调事件
        uv__fs_poll_endgame(loop, (uv_fs_poll_t*) handle);
        break;

      default:
        assert(0);
        break;
    }
  }
}

    • 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

    uv_tcp_endgame

    void uv_tcp_endgame(uv_loop_t* loop, uv_tcp_t* handle) {
  int err;
  unsigned int i;
  uv_tcp_accept_t* req;

  //uv_shutdown注册的shutdown回调事件属于当前状态
  //如果句柄处于连接状态，且shutdown_req不为空（shutdown_req的句柄应为UV_SHUTDOWN类型）
  //同时当前的写缓冲队列数为0，才执行shutdown回调方法
  if (handle->flags & UV_HANDLE_CONNECTION &&
      handle->stream.conn.shutdown_req != NULL &&
      handle->stream.conn.write_reqs_pending == 0) {
    
    //从loop的handle队列移除
    UNREGISTER_HANDLE_REQ(loop, handle, handle->stream.conn.shutdown_req);

    err = 0;
    if (handle->flags & UV_HANDLE_CLOSING) {
      err = ERROR_OPERATION_ABORTED;
    } else if (shutdown(handle->socket, SD_SEND) == SOCKET_ERROR) { //只关闭写状态
      err = WSAGetLastError();
    }

    //响应回调事件
    if (handle->stream.conn.shutdown_req->cb) {
      handle->stream.conn.shutdown_req->cb(handle->stream.conn.shutdown_req,
                               uv_translate_sys_error(err));
    }

    handle->stream.conn.shutdown_req = NULL;
    DECREASE_PENDING_REQ_COUNT(handle);
    return;
  }

  //uv_close注册close回调事件属于当前状态
  //句柄要处于closing状态，同时当前的读缓冲队列数为0
  if (handle->flags & UV_HANDLE_CLOSING &&
      handle->reqs_pending == 0) {
    assert(!(handle->flags & UV_HANDLE_CLOSED));
    assert(handle->socket == INVALID_SOCKET);

    //如果为服务端句柄且为且该句柄支持重叠结构，清除所有连接的句柄信息
    if (!(handle->flags & UV_HANDLE_CONNECTION) && handle->tcp.serv.accept_reqs) {
      if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
        for (i = 0; i < uv_simultaneous_server_accepts; i++) {
          req = &handle->tcp.serv.accept_reqs[i];
          if (req->wait_handle != INVALID_HANDLE_VALUE) {
            UnregisterWait(req->wait_handle);
            req->wait_handle = INVALID_HANDLE_VALUE;
          }
          if (req->event_handle != NULL) {
            CloseHandle(req->event_handle);
            req->event_handle = NULL;
          }
        }
      }

      uv__free(handle->tcp.serv.accept_reqs);
      handle->tcp.serv.accept_reqs = NULL;
    }

    //如果句柄为连接状态且支持重叠结构，消除句柄信息
    if (handle->flags & UV_HANDLE_CONNECTION &&
        handle->flags & UV_HANDLE_EMULATE_IOCP) {
      if (handle->read_req.wait_handle != INVALID_HANDLE_VALUE) {
        UnregisterWait(handle->read_req.wait_handle);
        handle->read_req.wait_handle = INVALID_HANDLE_VALUE;
      }
      if (handle->read_req.event_handle != NULL) {
        CloseHandle(handle->read_req.event_handle);
        handle->read_req.event_handle = NULL;
      }
    }

    //将其从handle队列中移除，同时响应回调事件
    uv__handle_close(handle);
    loop->active_tcp_streams--;
  }
}

    • 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
  • 相关阅读:
    CSS-0_2 CSS和继承（inherit & initial）
    数据可视化模块 Matplotlib详解
    shiro注解和过滤器
    jvm调优
    Node.js和cnpm环境搭建
    LyScript 内存交换与差异对比
    2020全栈学习Demo大合集 AllDemo-996station GitHub鉴赏官
    Python 3.10里面的Match-Case语法详解
    表单判断0也生效相关方法
    rust编译器教我做人，为啥还要学习rust语言，因为想使用rust做一些底层服务，更深入的研究技术。
  • 原文地址：https://blog.csdn.net/qq135595696/article/details/127954980