1. 程式人生 > >淺析libuv原始碼-node事件輪詢解析(4)

淺析libuv原始碼-node事件輪詢解析(4)

  這篇應該能結,簡圖如下。

  上一篇講到了uv__work_submit方法,接著寫了。

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)) {
  // 上篇主要講的這裡 初始化執行緒池等
  uv_once(&once, init_once);
  w->loop = loop;
  w->work = work;
  w->done = done;
  post(&w->wq, kind);
}

  從post開始。

static void post(QUEUE* q, enum uv__work_kind kind) {
  // 因為存在佇列插入操作 需要加鎖
  uv_mutex_lock(&mutex);
  if (kind == UV__WORK_SLOW_IO) {
    //跳...
  }

  QUEUE_INSERT_TAIL(&wq, q);
  // 如果有空閒執行緒 喚醒
  if (idle_threads > 0)
    uv_cond_signal(&cond);
  uv_mutex_unlock(&mutex);
}

  wq就是上一篇講的執行緒都會用到的那個佇列,這裡負責插入任務,worker中取出任務。

  沒想到post到這裡沒了,這點東西併到上一篇就好了。以後寫這種系列部落格還是先規劃一下,不能邊看原始碼邊寫……

  函式到這裡就斷了,看似沒有線索,實際上在上一節的worker方法中,還漏了一個地方。

static void worker(void* arg) {
  // ...
  for (;;) {
    // 這裡呼叫內部fs方法處理任務
    w = QUEUE_DATA(q, struct uv__work, wq);
    w->work(w);

    uv_mutex_lock(&w->loop->wq_mutex);
    w->work = NULL; 
    QUEUE_INSERT_TAIL(&w->loop->wq, &w->wq);
    // 這個是漏了的關鍵
    uv_async_send(&w->loop->wq_async);
    uv_mutex_unlock(&w->loop->wq_mutex);

    // ...
  }
}

  每一條執行緒在每次處理完一條事務並將其插入工作佇列wq後,都會呼叫一下這個uv_async_send方法,上一篇沒講這個。

  這裡的wq_async是一個在loop上面的變數,在輪詢初始化的時候出現過,這裡先不看。

  uv_async_send這個方法又涉及到另外一個大模組,如下。

int uv_async_send(uv_async_t* handle) {
  // 錯誤處理...
  if (!uv__atomic_exchange_set(&handle->async_sent)) {
    POST_COMPLETION_FOR_REQ(loop, &handle->async_req);
  }

  return 0;
}

// 將操作結果推到iocp上面
#define POST_COMPLETION_FOR_REQ(loop, req)                              \
  if (!PostQueuedCompletionStatus((loop)->iocp,                         \
                                  0,                                    \
                                  0,                                    \
                                  &((req)->u.io.overlapped))) {         \
    uv_fatal_error(GetLastError(), "PostQueuedCompletionStatus");       \
  }

  這個地方說實話我並不是明白windows底層API的操作原理,IOCP這部分我沒有去研究,只能從字面上去理解。

  關於PostXXX方法官網解釋如下:

Posts an I/O completion packet to an I/O completion port.

  將一個I/O完成的資料打包到I/O完成的埠,翻譯過來就是這樣,個人理解上的話大概是把一個async_req丟到IOCP那裡儲存起來。

 

  接下來終於可以回到事件輪詢部分,點題了。

int uv_run(uv_loop_t *loop, uv_run_mode mode) {
  // ...

  while (r != 0 && loop->stop_flag == 0) {
    // ...
    // call pending callbacks
    ran_pending = uv_process_reqs(loop);
    // ...
    // poll for I/O
    if (pGetQueuedCompletionStatusEx)
      uv__poll(loop, timeout);
    else
      uv__poll_wine(loop, timeout);
    // ...
  }
  // ...
}

  截取了剩下的poll for I/O、call pending callback,也就是剩下的兩部分了。if判斷不用管,只是一個方法相容,最終的目的是一樣的。

  所以只看uv__poll部分。

static void uv__poll(uv_loop_t* loop, DWORD timeout) {
  // ...
  // 設定阻塞時間
  uint64_t timeout_time;
  timeout_time = loop->time + timeout;

  for (repeat = 0; ; repeat++) {
    success = GetQueuedCompletionStatusEx(loop->iocp,
                                          overlappeds,
                                          ARRAY_SIZE(overlappeds),
                                          &count,
                                          timeout,
                                          FALSE);

    if (success) {
      for (i = 0; i < count; i++) {
        if (overlappeds[i].lpOverlapped) {
          req = uv_overlapped_to_req(overlappeds[i].lpOverlapped);
          uv_insert_pending_req(loop, req);
        }
      }
      uv_update_time(loop);
    } else if (GetLastError() != WAIT_TIMEOUT) {
      // ...
    } else if (timeout > 0) {
      // 超時處理...
    }
    break;
  }
}

  這裡的GetQueueXXX方法與之前的PostQueueXXX正好是一對方法,都是基於IOCP,一個是儲存,一個是取出。

  遍歷操作就很容易懂了,取出資料後,一個個的塞到pending callback的佇列中。

  把uv_insert_pending_req、uv_process_reqs兩個方法結合起來看。

INLINE static void uv_insert_pending_req(uv_loop_t* loop, uv_req_t* req) {
  req->next_req = NULL;
  // 插入到pending_reqs_tail上
  if (loop->pending_reqs_tail) {
    // DEBUG...

    req->next_req = loop->pending_reqs_tail->next_req;
    loop->pending_reqs_tail->next_req = req;
    loop->pending_reqs_tail = req;
  } else {
    req->next_req = req;
    loop->pending_reqs_tail = req;
  }
}

INLINE static int uv_process_reqs(uv_loop_t* loop) {
  // ...

  // 處理pending_reqs_tail
  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) {
      // handle各類req...
    }
  }

  return 1;
}

  就這樣,完美的把poll for I/O與call pending callback兩塊內容連線到了一起,也同時理解了一個非同步I/O操作是如何在node內部被處理的。

 

  最後還是剩一個尾巴,就是丟到IOCP的那個async_req怎麼回事?這個變數在輪詢的初始化方法中出現,如下。

typedef struct uv_loop_s uv_loop_t;

struct uv_loop_s {
  // ...
  UV_LOOP_PRIVATE_FIELDS
};

#define UV_LOOP_PRIVATE_FIELDS                                                \
  // 其餘變數
  uv_async_t wq_async;

// uv__word_done是這個handle的回撥函式
int uv_loop_init(uv_loop_t* loop) {
  // ...
  err = uv_async_init(loop, &loop->wq_async, uv__work_done);
  // ...
}

// 第一篇中演示過handle的初始化和執行 很常規的init、start兩步
int uv_async_init(uv_loop_t* loop, uv_async_t* handle, uv_async_cb async_cb) {
  uv_req_t* req;

  uv__handle_init(loop, (uv_handle_t*) handle, UV_ASYNC);
  handle->async_sent = 0;
  handle->async_cb = async_cb;

  req = &handle->async_req;
  UV_REQ_INIT(req, UV_WAKEUP);
  req->data = handle;

  uv__handle_start(handle);

  return 0;
}

# define UV_REQ_INIT(req, typ)                                                \
  do {                                                                        \
    (req)->type = (typ);                                                      \
  }                                                                           \
  while (0)

  從程式碼裡面可以知道,loop上本身帶有一個uv_async_t的變數wq_async,初始化後有四個屬性。其中需要注意,這個型別的type被設定為UV_WAKEUP。

  再回到uv_process_reqs中,處理從IOCP取出的req那塊。

INLINE static int uv_process_reqs(uv_loop_t* loop) {
  // ...

  while (next != NULL) {
    // ...
    switch (req->type) {
      // ...
      case UV_WAKEUP:
        uv_process_async_wakeup_req(loop, (uv_async_t*) req->data, req);
        break;
      // ...
    }
  }

  return 1;
}

  我們找到了處理UV_WAKEUP的case,引數參考上面那個初始化的程式碼也很容易得知,req->data就是loop初始化的那個handle,req是那個async_req。

  方法程式碼如下。

void uv_process_async_wakeup_req(uv_loop_t* loop, uv_async_t* handle, uv_req_t* req) {
  // 丟進IOCP的時候被設定為1了 具體在uv_async_send的uv__atomic_exchange_set方法中
  handle->async_sent = 0;
  if (handle->flags & UV_HANDLE_CLOSING) {
    uv_want_endgame(loop, (uv_handle_t*)handle);
  } else if (handle->async_cb != NULL) {
    // 進的else分支
    handle->async_cb(handle);
  }
}

  這裡的async_cb也是初始化就定義了,實際函式名是uv__work_done。

void uv__work_done(uv_async_t* handle) {
  // ...

  loop = container_of(handle, uv_loop_t, wq_async);
  uv_mutex_lock(&loop->wq_mutex);
  // 還是那個熟悉的佇列
  QUEUE_MOVE(&loop->wq, &wq);
  uv_mutex_unlock(&loop->wq_mutex);

  while (!QUEUE_EMPTY(&wq)) {
    // ...
    w->done(w, err);
  }
}

  這個done,就是使用者從JS傳過去的callback……

  也就是說call pending callback實際上是呼叫使用者傳過來的callback,第二篇的圖其實是有問題的,系列完結撒