Node.js C++ 層的任務(wù)管理

作者：theanarkh 2022-12-02 23:20:06

我們都知道 Node.js 是基于事件循環(huán)來運行的，本質(zhì)上是一個生產(chǎn)者 / 消費者模型，所以就少不了任務(wù)的管理機制，不過本文不是介紹事件循環(huán)中的任務(wù)管理，而是 C++ 層的任務(wù)管理。本文主要介紹 SetImmediate、SetImmediateThreadsafe、RequestInterrupt、AddCleanupHook 這四個 API 產(chǎn)生的任務(wù)。時間關(guān)系，隨便寫寫，權(quán)當筆記。

好久沒更新了，今天寫個筆記。

任務(wù)管理機制的初始化

首先來看一下 Node.js 啟動的過程中，和任務(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_ 則用于線程間通信，即當子線程往主線程提交任務(wù)時，通過 task_queues_async_ 通知主線程，然后主線程執(zhí)行 uv_async_init 注冊的回調(diào)。上面的代碼就是消費者的邏輯。后面再詳細分析里面的處理流程。

提交任務(wù)

接下來逐個看一下生產(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 有一個很重要的區(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)時依然會被執(zhí)行。接著看 AddCleanupHook。

void Environment::AddCleanupHook(CleanupQueue::Callback fn, void* arg) {
  cleanup_queue_.Add(fn, arg);
}

AddCleanupHook 用于注冊線程退出前的回調(diào)。生產(chǎn)者的邏輯都比較簡單，就是往任務(wù)隊列里插入一個任務(wù)，如果是涉及到線程間的任務(wù)，則通知主線程。

消費者

接下來看一下消費者的邏輯，根據(jù)前面的分析可以知道，消費者有幾個：CheckImmediate，task_queues_async_ 的處理函數(shù)、RequestInterrupt 注冊的函數(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 階段時，只有子線程提交任務(wù)時會喚醒主線程，具體是通過 task_queues_async_ 結(jié)構(gòu)體，看一下處理函數(shù)。

env->RunAndClearNativeImmediates();

可以看到這時候也是處理了SetImmediate、SetImmediateThreadsafe 和 RequestInterrupt 產(chǎn)生的任務(wù)。最后來看一下處理退出前回調(diào)的函數(shù)，具體時機是 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 中同時處理了 SetImmediate、SetImmediateThreadsafe、 RequestInterrupt 產(chǎn)生的任務(wù)和注冊的退出前回調(diào)。

責任編輯：武曉燕來源：編程雜技

Node.js C++任務(wù)管理