我們都知道 Node.js 是基于事件循環來運行的，本質上是一個生產者 / 消費者模型，所以就少不了任務的管理機制，不過本文不是介紹事件循環中的任務管理，而是 C++ 層的任務管理。本文主要介紹 SetImmediate、SetImmediateThreadsafe、RequestInterrupt、AddCleanupHook 這四個 API 產生的任務。時間關系，隨便寫寫，權當筆記。
好久沒更新了，今天寫個筆記。
我們都知道 Node.js 是基于事件循環來運行的，本質上是一個生產者 / 消費者模型，所以就少不了任務的管理機制，不過本文不是介紹事件循環中的任務管理，而是 C++ 層的任務管理。本文主要介紹 SetImmediate、SetImmediateThreadsafe、RequestInterrupt、AddCleanupHook 這四個 API 產生的任務。時間關系，隨便寫寫，權當筆記。
任務管理機制的初始化
首先來看一下 Node.js 啟動的過程中，和任務管理相關的邏輯。
uv_check_start(immediate_check_handle(), CheckImmediate)
uv_async_init(
event_loop(),
&task_queues_async_,
[](uv_async_t* async) {
Environment* env = ContainerOf(&Environment::task_queues_async_, async);
env->RunAndClearNativeImmediates();
})
CheckImmediate 是在 check 階段執行的函數，task_queues_async_ 則用于線程間通信，即當子線程往主線程提交任務時，通過 task_queues_async_ 通知主線程，然后主線程執行 uv_async_init 注冊的回調。上面的代碼就是消費者的邏輯。后面再詳細分析里面的處理流程。
提交任務
接下來逐個看一下生產者的邏輯。
template <typename Fn>
void Environment::SetImmediate(Fn&& cb, CallbackFlags::Flags flags) {
auto callback = native_immediates_.CreateCallback(std::move(cb), flags);
native_immediates_.Push(std::move(callback));
// ...
}
SetImmediate 用于同線程的代碼提交任務。
template <typename Fn>
void Environment::SetImmediateThreadsafe(Fn&& cb, CallbackFlags::Flags flags) {
auto callback = native_immediates_threadsafe_.CreateCallback(
std::move(cb), flags);
{
Mutex::ScopedLock lock(native_immediates_threadsafe_mutex_);
native_immediates_threadsafe_.Push(std::move(callback));
if (task_queues_async_initialized_)
uv_async_send(&task_queues_async_);
}
}
SetImmediateThreadsafe 用于子線程給主線程提交任務，所以需要加鎖。
template <typename Fn>
void Environment::RequestInterrupt(Fn&& cb) {
auto callback = native_immediates_interrupts_.CreateCallback(
std::move(cb), CallbackFlags::kRefed);
{
Mutex::ScopedLock lock(native_immediates_threadsafe_mutex_);
native_immediates_interrupts_.Push(std::move(callback));
if (task_queues_async_initialized_)
uv_async_send(&task_queues_async_);
}
RequestInterruptFromV8();
}
RequestInterrupt 用于子線程給主線程提交代碼，他和 SetImmediateThreadsafe 有一個很重要的區別是調用了 RequestInterruptFromV8。
void Environment::RequestInterruptFromV8() {
isolate()->RequestInterrupt([](Isolate* isolate, void* data) {
std::unique_ptr<Environment*> env_ptr { static_cast<Environment**>(data) };
Environment* env = *env_ptr;
env->RunAndClearInterrupts();
}, interrupt_data);
}RequestInterrupt 可以使得提交的代碼在 JS 代碼死循環時依然會被執行。接著看 AddCleanupHook。
void Environment::AddCleanupHook(CleanupQueue::Callback fn, void* arg) {
cleanup_queue_.Add(fn, arg);
}AddCleanupHook 用于注冊線程退出前的回調。生產者的邏輯都比較簡單，就是往任務隊列里插入一個任務，如果是涉及到線程間的任務，則通知主線程。
消費者
接下來看一下消費者的邏輯，根據前面的分析可以知道，消費者有幾個：CheckImmediate，task_queues_async_ 的處理函數、RequestInterrupt 注冊的函數、退出前回調處理函數。先看 CheckImmediate。
void Environment::CheckImmediate(uv_check_t* handle) {
Environment* env = Environment::from_immediate_check_handle(handle);
env->RunAndClearNativeImmediates();
}
void Environment::RunAndClearNativeImmediates(bool only_refed) {
RunAndClearInterrupts();
auto drain_list = [&](NativeImmediateQueue* queue) {
while (auto head = queue->Shift()) {
head->Call(this);
}
return false;
};
while (drain_list(&native_immediates_)) {}
NativeImmediateQueue threadsafe_immediates;
if (native_immediates_threadsafe_.size() > 0) {
Mutex::ScopedLock lock(native_immediates_threadsafe_mutex_);
threadsafe_immediates.ConcatMove(std::move(native_immediates_threadsafe_));
}
while (drain_list(&threadsafe_immediates)) {}
}
void Environment::RunAndClearInterrupts() {
while (native_immediates_interrupts_.size() > 0) {
NativeImmediateQueue queue;
{
Mutex::ScopedLock lock(native_immediates_threadsafe_mutex_);
queue.ConcatMove(std::move(native_immediates_interrupts_));
}
while (auto head = queue.Shift())
head->Call(this);
}
}CheckImmediate 函數中處理了SetImmediate、SetImmediateThreadsafe 和 RequestInterrupt 產生的任務。但是如果主線程阻塞在 Poll IO 階段時，只有子線程提交任務時會喚醒主線程，具體是通過 task_queues_async_ 結構體，看一下處理函數。
env->RunAndClearNativeImmediates();
可以看到這時候也是處理了SetImmediate、SetImmediateThreadsafe 和 RequestInterrupt 產生的任務。最后來看一下處理退出前回調的函數，具體時機是 FreeEnvironment 函數中的 env->RunCleanup()。
void Environment::RunCleanup() {
RunAndClearNativeImmediates(true);
while (!cleanup_queue_.empty() || principal_realm_->HasCleanupHooks() ||
native_immediates_.size() > 0 ||
native_immediates_threadsafe_.size() > 0 ||
native_immediates_interrupts_.size() > 0) {
// 見 CleanupQueue::Drain
cleanup_queue_.Drain();
RunAndClearNativeImmediates(true);
}
}
// cleanup_queue_.Drain();
void CleanupQueue::Drain() {
std::vector<CleanupHookCallback> callbacks(cleanup_hooks_.begin(),
cleanup_hooks_.end());
std::sort(callbacks.begin(),
callbacks.end(),
[](const CleanupHookCallback& a, const CleanupHookCallback& b) {
return a.insertion_order_counter_ > b.insertion_order_counter_;
});
for (const CleanupHookCallback& cb : callbacks) {
cb.fn_(cb.arg_);
cleanup_hooks_.erase(cb);
}
}RunCleanup 中同時處理了 SetImmediate、SetImmediateThreadsafe、 RequestInterrupt 產生的任務和注冊的退出前回調。
