Node中非同步和同步的實現
使用過node的朋友都知道，它最重要的也是最值得稱道的就是使用了非同步事件驅動的框架libuv，這個框架使得被稱為玩具語言的JavaScript也在後端語言中佔了一席之地（當然V8的高效能也是功不可沒，而且libuv的程式碼非常優雅，很值得大家的學習。不過libuv整個框架很大，我們不可能只通過一篇文章就能瞭解到它所有的東西，所以我挑選了node中最簡單fs模組同步讀和非同步讀檔案的過程講解來對libuv的一個大概過程有所瞭解。
fs.readSync
fs.readSync這個方法我相信沒有人會陌生，在node中同步讀取檔案，不過再很多文章中都不推薦使用這個方法，因為會造成node單執行緒的阻塞，對於一些比較繁忙的node例項來說是非常不友好的，不過今天我們不討論這些，只討論其中的實現。實現我們在node工程的lib目錄中找到fs.js可以看到它的程式碼:
function(fd, buffer, offset, length, position) {
if (length === 0) {
return 0;
}
return binding.read(fd, buffer, offset, length, position);
};
複製程式碼其中直接呼叫了binding.read其中的binding的申明是這樣的binding = process.binding('fs')，這個自然就是node的builtin_module，所以我們直接找到src/node_flie.cc檔案中。
從node::InitFs方法中我們可以看到所有的方法申明，其中返回物件中的read方法對應的是static void Read(const FunctionCallbackInfo<Value>& args)
static void Read(const FunctionCallbackInfo<Value>& args) {
//獲取傳入引數，並對引數進行處理
....
//將傳入的buffer的記憶體的地址取出並用來儲存read出的內容
char * buf = nullptr;
Local<Object> buffer_obj = args[1]->ToObject(env->isolate());
char *buffer_data = Buffer::Data(buffer_obj);
size_t buffer_length = Buffer::Length(buffer_obj);
...
buf = buffer_data + off;
uv_buf_t uvbuf = uv_buf_init(const_cast<char*>(buf), len);
//執行read操作
req = args[5];
if (req->IsObject()) {
ASYNC_CALL(read, req, UTF8, fd, &uvbuf, 1, pos);
} else {
SYNC_CALL(read, 0, fd, &uvbuf, 1, pos)
args.GetReturnValue().Set(SYNC_RESULT);
}
}
複製程式碼 從上面的程式碼，我們可以看出第六個引數是個很關鍵的引數，如果傳入了一個物件則使用非同步操作，而我們的fs.readSync方法沒有傳入第六個引數隨意使用的是同步操作，並且在操作完成後立即返回結果。
而SYNC_CALL這個巨集中具體做了什麼呢，他主要是呼叫另外一個巨集:
#define SYNC_CALL(func, path, ...) \
SYNC_DEST_CALL(func, path, nullptr, __VA_ARGS__) \
複製程式碼其中__VA_ARGS__表示的是除了func和path外其他傳入巨集的引數，接下來我們來看一下SYNC_DEST_CALL巨集:
#define SYNC_DEST_CALL(func, path, dest, ...) \
fs_req_wrap req_wrap; \
env->PrintSyncTrace(); \
int err = uv_fs_ ## func(env->event_loop(), \
&req_wrap.req, \
__VA_ARGS__, \
nullptr); \
if (err < 0) { \
return env->ThrowUVException(err, #func, nullptr, path, dest); \
} \
複製程式碼其中在巨集命令中的 ##標記是連線符的意思，所以這裡其實就是呼叫uv_fs_read方法，而env->PrintSyncTrace()是為了在node開啟--trace-sync-io時用來追蹤程式碼中何處使用了同步io時使用，可以通過這個方法打出程式碼中呼叫同步io的位置，所以當你的程式碼經常發生阻塞的時候你可以通過這個來調優你的程式碼（當然阻塞的原因未必是同步io造成的）。uv_fs_read方法是libuv是來讀取檔案的呼叫，我們找到這個方法的位置，就在deps/uv/src/unix/fs.c中:
int uv_fs_read(uv_loop_t* loop, uv_fs_t* req,
uv_file file,
const uv_buf_t bufs[],
unsigned int nbufs,
int64_t off,
uv_fs_cb cb) {
INIT(READ);
if (bufs == NULL || nbufs == 0)
return -EINVAL;
req->file = file;
req->nbufs = nbufs;
req->bufs = req->bufsml;
if (nbufs > ARRAY_SIZE(req->bufsml))
req->bufs = uv__malloc(nbufs * sizeof(*bufs));
if (req->bufs == NULL) {
if (cb != NULL)
uv__req_unregister(loop, req);
return -ENOMEM;
}
memcpy(req->bufs, bufs, nbufs * sizeof(*bufs));
req->off = off;
POST;
}
複製程式碼首先我們來看巨集呼叫INIT(READ):
#define INIT(subtype) \
do { \
if (req == NULL) \
return -EINVAL; \
req->type = UV_FS; \
if (cb != NULL) \
uv__req_init(loop, req, UV_FS); \
req->fs_type = UV_FS_ ## subtype; \
req->result = 0; \
req->ptr = NULL; \
req->loop = loop; \
req->path = NULL; \
req->new_path = NULL; \
req->cb = cb; \
} \
while (0)
複製程式碼這是一個很明顯的初始化操作，這裡主要說兩個最重要地方，首先是將req的loop指向node的event_loop，其次是指定了fs_type喂UV_FS_READ這個是一個重要的標誌，為後面的工作做識別。做了這些操作以後回到uv_fs_read方法來，我們可以看到在POST巨集呼叫之前都是一些對引數的處理工作，這個沒什麼可講的，我們主要來看看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); \
return req->result; \
} \
} \
while (0)
複製程式碼從上面的程式碼我們可以看到在有cb的時候呼叫的是uv__work_submit，這就是非同步的情況下呼叫，等會兒我們再講。現在我們先說uv__fs_work 的方法：
static void uv__fs_work(struct uv__work* w) {
int retry_on_eintr;
uv_fs_t* req;
ssize_t r;
req = container_of(w, uv_fs_t, work_req);
retry_on_eintr = !(req->fs_type == UV_FS_CLOSE);
do {
errno = 0;
#define X(type, action) \
case UV_FS_ ## type: \
r = action; \
break;
switch (req->fs_type) {
...
X(WRITE, uv__fs_buf_iter(req, uv__fs_write));
X(OPEN, uv__fs_open(req));
X(READ, uv__fs_buf_iter(req, uv__fs_read));
...
}
#undef X
} while (r == -1 && errno == EINTR && retry_on_eintr);
if (r == -1)
req->result = -errno;
else
req->result = r;
if (r == 0 && (req->fs_type == UV_FS_STAT ||
req->fs_type == UV_FS_FSTAT ||
req->fs_type == UV_FS_LSTAT)) {
req->ptr = &req->statbuf;
}
}
複製程式碼這個方法因為是fs檔案共用的方法，所以在其中會根據不同的型別的來執行不同的方法，剛剛我們看到了在初始化req的時候給了它UV_FS_READ的type，所以會執行方法uv__fs_buf_iter(req, uv__fs_read),uv__fs_buf_iter方法中主要是呼叫了傳入的第二個引數uv__fs_read函式，這裡的程式碼就不貼了很簡單，就是普通的read(還有readv和pread)操作，不過其中有個點就是這段程式碼:
#if defined(_AIX)
struct stat buf;
if(fstat(req->file, &buf))
return -1;
if(S_ISDIR(buf.st_mode)) {
errno = EISDIR;
return -1;
}
#endif
複製程式碼這段程式碼很好地解釋了node文件中關於fs.readFileSync的這一段
Note: Similar to fs.readFile(), when the path is a directory, the behavior of fs.readFileSync() is platform-specific.
// macOS, Linux, and Windows
fs.readFileSync('<directory>');
// => [Error: EISDIR: illegal operation on a directory, read <directory>]
// FreeBSD
fs.readFileSync('<directory>'); // => null, <data>
複製程式碼在uv__fs_read成功讀取檔案後,req->bufs中就已經有了所需的內容了，從node_file.cc的static void Read(const FunctionCallbackInfo<Value>& args)方法中我們可以知道req->bufs的記憶體所指向的則是binding.read(fd, buffer, offset, length, position)傳入的buffer記憶體段。這個時候就已經得到了想要讀取的內容了。而我們平時經常使用的fs.readFileSync則是先開啟檔案得到其fd，並生成一段buffer然後呼叫fs.readSync，是生成的buffer中取得檔案內容再返回，簡化了很多操作，所以更受到大家的青睞。到這裡我們的同步讀取就已經結束了，算是很簡單，因為read這些操作都是阻塞性的操作，所以對於單執行緒的node程序來說確實容易遇到效能瓶頸，下面我們來說一下node的非同步讀取fs.read函式。
fs.read
非同步的操作遠比同步要複雜很多，我們來一步步的瞭解。首先我們先來看 ocess.nextTick(function() { callback && callback(null, 0, buffer); }); }
function wrapper(err, bytesRead) {
// Retain a reference to buffer so that it can't be GC'ed too soon.
callback && callback(err, bytesRead || 0, buffer);
}
var req = new FSReqWrap();
req.oncomplete = wrapper;
binding.read(fd, buffer, offset, length, position, req);
};
複製程式碼從剛剛同步的分析中，我們知道當bingd.read傳入第六個引數的時候則會非同步執行read操作，這裡就傳入了第六個引數req, req = new FSReqWrap();req是FSReqWrap = binding.FSReqWrap的例項，所以我們從node::InitFs中可以看到如下程式碼:
Local<FunctionTemplate> fst = FunctionTemplate::New(env->isolate(), NewFSReqWrap);
fst->InstanceTemplate()->SetInternalFieldCount(1);
AsyncWrap::AddWrapMethods(env, fst);
Local<String> wrapString =
FIXED_ONE_BYTE_STRING(env->isolate(), "FSReqWrap");
fst->SetClassName(wrapString);
target->Set(wrapString, fst->GetFunction());
複製程式碼上面的程式碼使用v8提供的API生成FSReqWrap的建構函式而void NewFSReqWrap(const FunctionCallbackInfo<Value>& args)就會說起建構函式的內容。這個函式主要的主要工作只有一個object->SetAlignedPointerInInternalField(0, nullptr);，不過這個只跟C++物件的嵌入有關。從之前我們討論過的static void Read(const FunctionCallbackInfo<Value>& args)方法中聊到過，當傳入req物件的時候回撥用巨集命令ASYNC_CALL，這個巨集命令跟之前的SYNC_CALL一樣的呼叫，通過ASYNC_DEST_CALL(func, req, nullptr, encoding, __VA_ARGS__)去呼叫真正的邏輯，所以我們直接來看ASYNC_DEST_CALL的程式碼：
#define ASYNC_DEST_CALL(func, request, dest, encoding, ...) \
Environment* env = Environment::GetCurrent(args); \
CHECK(request->IsObject()); \
FSReqWrap* req_wrap = FSReqWrap::New(env, request.As<Object>(), \
#func, dest, encoding); \
int err = uv_fs_ ## func(env->event_loop(), \
req_wrap->req(), \
__VA_ARGS__, \
After); \
req_wrap->Dispatched(); \
if (err < 0) { \
uv_fs_t* uv_req = req_wrap->req(); \
uv_req->result = err; \
uv_req->path = nullptr; \
After(uv_req); \
req_wrap = nullptr; \
} else { \
args.GetReturnValue().Set(req_wrap->persistent()); \
}
複製程式碼上面的程式碼我們可以看到通過FSReqWrap::New來生成了req_wrap，這個方法的執行是node生成物件的一個基本邏輯，所以我們著重說一下，首先我們來看一下FSReqWrap::New的程式碼:
const bool copy = (data != nullptr && ownership == COPY);
const size_t size = copy ? 1 + strlen(data) : 0;
FSReqWrap* that;
char* const storage = new char[sizeof(*that) + size];
that = new(storage) FSReqWrap(env, req, syscall, data, encoding);
if (copy)
that->data_ = static_cast<char*>(memcpy(that->inline_data(), data, size));
return that;
複製程式碼這段程式碼我們主要了解一下new(storage) FSReqWrap(env, req, syscall, data, encoding);，首先我們通過一張圖來了解一下FSReqWrap的繼承關係：
上圖中我們給出了一些關鍵物件的關鍵屬性和方法，所以我們可以看出FSReqWrap各個繼承物件的主要作用：
1.繼承ReqWrap物件的關鍵屬性uv_fs_t，和關鍵方法ReqWrap<T>::Dispatched，使用該方法中的req_.data = this;在libuv的方法中傳遞自身。
2.繼承AsyncWrap中的MakeCallback，這個函式會執行我們傳入的非同步讀取完成後的回撥，在這個例子中就是使用js中通過req.oncomplete = wrapper;傳入的wrapper函式。
3.繼承BaseObject物件中的關鍵屬性Persistent<Object> persistent_handle_和Environment* env_，前者是v8中的持久化js物件，和Local的關係可以參見v8官方的解釋：
Local handles are held on a stack and are deleted when the appropriate destructor is called. These handles' lifetime is determined by a handle scope, which is often created at the beginning of a function call. When the handle scope is deleted, the garbage collector is free to deallocate those objects previously referenced by handles in the handle scope, provided they are no longer accessible from JavaScript or other handles.
Persistent handles provide a reference to a heap-allocated JavaScript Object, just like a local handle. There are two flavors, which differ in the lifetime management of the reference they handle. Use a persistent handle when you need to keep a reference to an object for more than one function call, or when handle lifetimes do not correspond to C++ scopes.
複製程式碼大概的意思就是，Local會隨著在棧上分配的scope析構而被GC清理掉，但是Persistent不會。有點類似棧上分配的記憶體和堆上分配記憶體的關係，想要在超過一個function中使用就要使用Persistent的v8物件，而後者是node的執行環境，幾乎囊括了node執行中所需要的一切方法和屬性（這一塊非常大，涉及的也很多，實在很難一兩句講清楚，跟本文討論內容無直接聯絡只能略過）。
最後，在FSReqWrap的建構函式中通過Wrap(object(), this)將我們上面提到的Persistent<Object> persistent_handle_持久化js物件和FSReqWrap的C++物件關聯起來，這是node中最常用的方式（也是ebmed開發中最常用的技巧）。我們回到巨集ASYNC_DEST_CALL中來，現在知道通過方法FSReqWrap::New方法使得FSReqWrap物件例項和剛剛在js中new的req物件連線了起來，也使libuv的uv_fs_t和其例項聯絡了起來。這個時候就跟前面一樣開始呼叫uv_fs_read，這次在最後一個引數cb中傳入了函式void After(uv_fs_t *req)作為回撥函式，從之前同步討論中我們就說過傳入回撥函式後情況的不同，首先是INIT巨集中在會多一步操作，通過uv__req_init中的QUEUE_INSERT_TAIL(&(loop)->active_reqs, &(req)->active_queue);巨集方法將req放入loop的acitve_reqs的迴圈連結串列中（libuv的迴圈連結串列實現非常的有意思，有興趣的朋友可以參考文章:libuv queue的實現)。而在POST中有回撥的函式的情況是直接通過uv__work_submit(loop, &req->work_req, uv__fs_work, uv__fs_done)呼叫來完成任務，我們來看一下uv__work_submit函式的程式碼，這個方法在deps/uv/src/threadpool中：
uv_once(&once, init_once);
w->loop = loop;
w->work = work;
w->done = done;
post(&w->wq);
複製程式碼該方法首先通過uv_once在第一次呼叫該方法是啟動幾個工作執行緒，這些執行緒主要執行static void worker(void* arg)方法：
for (;;) {
uv_mutex_lock(&mutex);
while (QUEUE_EMPTY(&wq)) {
idle_threads += 1;
uv_cond_wait(&cond, &mutex);
idle_threads -= 1;
}
q = QUEUE_HEAD(&wq);
if (q == &exit_message)
uv_cond_signal(&cond);
else {
QUEUE_REMOVE(q);
QUEUE_INIT(q);
}
uv_mutex_unlock(&mutex);
if (q == &exit_message)
break;
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_cond_wait使該執行緒阻塞，而在有任務的時候會在佇列中取出該任務再通過w->work(w)執行其任務，在執行完成後會將任務註冊在loop->wq的佇列中再通過uv_async_send(&w->loop->wq_async)通知主執行緒從loop->wq的佇列取出該任務並執行其回撥。
再回到uv__work_submit通過work方法我們就知道它接下來的工作是做什麼了，註冊work函式也就是傳入uv__fs_work函式，這個函式我們之前就介紹過了，這裡就不多做解釋了，只是在非同步中是通過worker執行緒來完成的，不會阻塞主執行緒。而第二個函式則是註冊完成後主線執行的回撥，也就是uv__fs_done:
req = container_of(w, uv_fs_t, work_req);
uv__req_unregister(req->loop, req);
if (status == -ECANCELED) {
assert(req->result == 0);
req->result = -ECANCELED;
}
req->cb(req);
複製程式碼從中我們可以看到，這個函式會將該任務的req從loop的acitve_reqs去去掉，然後執行傳入uv_fs_read中的回撥函式。而最後的post中主要是將當前任務註冊到wq的列表中，並使用條件變數的uv_cond_signal函式觸發uv_cond_wait中阻塞的函式運作起來，接著worker程序就能執行我們剛剛說的過程了。
上面我們講解了大概的過程，從這個過程中就能明白非同步的讀操作是如何執行的，通過使用wokrer執行緒來做實際的讀操作，而主執行緒則是在worker執行緒完成操作後，執行回撥。不過現在回過頭來我們看看，在worker執行緒以後是如何通知主執行緒呢？剛剛我們說到了是通過uv_async_send(&w->loop->wq_async)的呼叫通知的，這裡我們來看看他具體是如何做的。首先我們要回到loop的初始化處，函式uv_loop_init中，在這個函式中有這樣一個呼叫： uv_async_init(loop, &loop->wq_async, uv__work_done);。這個呼叫會生成一個管道，並通過以下語句：
uv__io_init(&loop->async_io_watcher, uv__async_io, pipefd[0]);
uv__io_start(loop, &loop->async_io_watcher, POLLIN);
loop->async_wfd = pipefd[1];
複製程式碼實現當有資料往pipefd[1]中寫時，主線會在讀取資料後執行uv__async_io的呼叫，在uv__async_io中最重要的工作就是執行其async_cb，而在loop初始化的時候註冊的async_cb是函式uv__work_done:
//取資料的操作
...
while (!QUEUE_EMPTY(&wq)) {
q = QUEUE_HEAD(&wq);
QUEUE_REMOVE(q);
w = container_of(q, struct uv__work, wq);
err = (w->work == uv__cancelled) ? UV_ECANCELED : 0;
w->done(w, err);
}
複製程式碼這裡我們可以看到，會從loop->wq佇列中取出放入其中所有任務，並通過w->done(w, err)執行其回撥，而剛剛在worker執行緒中的呼叫uv_async_send(&w->loop->wq_async)即是通過往loop->async_wfd，即上面提到的pipefd[1]寫一個位元組來觸發整個過程。到這裡最開始uv_fs_read中註冊的函式uv__fs_done就可以執行了，而這個函式的主要任務即是呼叫傳入uv_fs_read的cb引數，即void After(uv_fs_t *req)函式，這個函式處理的情況比較多，就不貼程式碼了唯一要講的就是他的第一句
FSReqWrap* req_wrap = static_cast<FSReqWrap*>(req->data);
複製程式碼這裡就回到了我們前面所說的通過req->data將FSReqWrap的物件例項串聯起來，到這裡就能順利的通過這個例項得到之前初始化的js物件，並執行它的oncomplete函數了。回到js的程式碼中我們可以看到這個函式執行的操作就是呼叫我們傳入的callback的函式:
callback && callback(err, bytesRead || 0, buffer);
複製程式碼至此，fs.read整個非同步操作就已經完成了，至於fs.readFile這個操作放在非同步中就複雜了許多，先非同步開啟檔案，再通過回撥中註冊非同步任務取得檔案的stat，最後通過回撥去讀取檔案，而且如果檔案太大不能一次讀完（一次最多讀8*1024的位元組），會不斷的回撥繼續讀取檔案，直到讀完才非同步關閉檔案，並且通過非同步關閉檔案的回撥執行傳入的回撥函式。可見為了我們平時開發中的方便，node的開發者還是付出了很多的努力的。
####總結
在瞭解了node對於檔案讀取的同步和非同步實現後，我們就能看出libuv的精妙之處了。特別是非同步時通過子執行緒處理任務，再用管道通知主線執行回撥的方式，真的是為node這樣的單執行緒語言量身定做，當然可能也有同學有疑問，主線是如何讀取管道值的呢？這又是一個很大的問題，我們只能以後的文章再來解釋了。這篇文章就先到此為止了，希望通過該文能幫助大家對node背後的邏輯會多一點了解。