我們都知道 Node.js 是基于事件循環(huán)來運(yùn)行的，本質(zhì)上是一個(gè)生產(chǎn)者 / 消費(fèi)者模型，所以就少不了任務(wù)的管理機(jī)制，不過本文不是介紹事件循環(huán)中的任務(wù)管理，而是 C++ 層的任務(wù)管理。本文主要介紹 SetImmediate、SetImmediateThreadsafe、RequestInterrupt、AddCleanupHook 這四個(gè) API 產(chǎn)生的任務(wù)。時(shí)間關(guān)系，隨便寫寫，權(quán)當(dāng)筆記。
好久沒更新了，今天寫個(gè)筆記。
我們都知道 Node.js 是基于事件循環(huán)來運(yùn)行的，本質(zhì)上是一個(gè)生產(chǎn)者 / 消費(fèi)者模型，所以就少不了任務(wù)的管理機(jī)制，不過本文不是介紹事件循環(huán)中的任務(wù)管理，而是 C++ 層的任務(wù)管理。本文主要介紹 SetImmediate、SetImmediateThreadsafe、RequestInterrupt、AddCleanupHook 這四個(gè) API 產(chǎn)生的任務(wù)。時(shí)間關(guān)系，隨便寫寫，權(quán)當(dāng)筆記。
任務(wù)管理機(jī)制的初始化
首先來看一下 Node.js 啟動(dòng)的過程中，和任務(wù)管理相關(guān)的邏輯。
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 階段執(zhí)行的函數(shù)，task_queues_async_ 則用于線程間通信，即當(dāng)子線程往主線程提交任務(wù)時(shí)，通過 task_queues_async_ 通知主線程，然后主線程執(zhí)行 uv_async_init 注冊(cè)的回調(diào)。上面的代碼就是消費(fèi)者的邏輯。后面再詳細(xì)分析里面的處理流程。
提交任務(wù)
接下來逐個(gè)看一下生產(chǎn)者的邏輯。
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 用于同線程的代碼提交任務(wù)。
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 用于子線程給主線程提交任務(wù)，所以需要加鎖。
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 有一個(gè)很重要的區(qū)別是調(diào)用了 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 代碼死循環(huán)時(shí)依然會(huì)被執(zhí)行。接著看 AddCleanupHook。
void Environment::AddCleanupHook(CleanupQueue::Callback fn, void* arg) {
cleanup_queue_.Add(fn, arg);
}AddCleanupHook 用于注冊(cè)線程退出前的回調(diào)。生產(chǎn)者的邏輯都比較簡單，就是往任務(wù)隊(duì)列里插入一個(gè)任務(wù)，如果是涉及到線程間的任務(wù)，則通知主線程。
消費(fèi)者
接下來看一下消費(fèi)者的邏輯，根據(jù)前面的分析可以知道，消費(fèi)者有幾個(gè)：CheckImmediate，task_queues_async_ 的處理函數(shù)、RequestInterrupt 注冊(cè)的函數(shù)、退出前回調(diào)處理函數(shù)。先看 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 函數(shù)中處理了SetImmediate、SetImmediateThreadsafe 和 RequestInterrupt 產(chǎn)生的任務(wù)。但是如果主線程阻塞在 Poll IO 階段時(shí)，只有子線程提交任務(wù)時(shí)會(huì)喚醒主線程，具體是通過 task_queues_async_ 結(jié)構(gòu)體，看一下處理函數(shù)。
env->RunAndClearNativeImmediates();
可以看到這時(shí)候也是處理了SetImmediate、SetImmediateThreadsafe 和 RequestInterrupt 產(chǎn)生的任務(wù)。最后來看一下處理退出前回調(diào)的函數(shù)，具體時(shí)機(jī)是 FreeEnvironment 函數(shù)中的 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 中同時(shí)處理了 SetImmediate、SetImmediateThreadsafe、 RequestInterrupt 產(chǎn)生的任務(wù)和注冊(cè)的退出前回調(diào)。
