src: add a threadsafe variant of SetImmediate()

Add a variant of `SetImmediate()` that can be called from any thread.
This allows removing the `AsyncRequest` abstraction and replaces it
with a more generic mechanism.

PR-URL: https://github.com/nodejs/node/pull/31386
Refs: https://github.com/openjs-foundation/summit/pull/240
Reviewed-By: Gireesh Punathil <gpunathi@in.ibm.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

1 files changed, 15 insertions, 5 deletions

diff --git a/src/env.cc b/src/env.cc
index b77d4f294d0..a07e1dafcfe 100644
--- a/src/env.cc
+++ b/src/env.cc
@@ -463,15 +463,16 @@ void Environment::InitializeLibuv(bool start_profiler_idle_notifier) {
   uv_check_init(event_loop(), &idle_check_handle_);
   uv_async_init(
       event_loop(),
-      &cleanup_finalization_groups_async_,
+      &task_queues_async_,
       [](uv_async_t* async) {
         Environment* env = ContainerOf(
-            &Environment::cleanup_finalization_groups_async_, async);
+            &Environment::task_queues_async_, async);
         env->CleanupFinalizationGroups();
+        env->RunAndClearNativeImmediates();
       });
   uv_unref(reinterpret_cast<uv_handle_t*>(&idle_prepare_handle_));
   uv_unref(reinterpret_cast<uv_handle_t*>(&idle_check_handle_));
-  uv_unref(reinterpret_cast<uv_handle_t*>(&cleanup_finalization_groups_async_));
+  uv_unref(reinterpret_cast<uv_handle_t*>(&task_queues_async_));
 
   thread_stopper()->Install(
     this, static_cast<void*>(this), [](uv_async_t* handle) {
@@ -535,7 +536,7 @@ void Environment::RegisterHandleCleanups() {
       close_and_finish,
       nullptr);
   RegisterHandleCleanup(
-      reinterpret_cast<uv_handle_t*>(&cleanup_finalization_groups_async_),
+      reinterpret_cast<uv_handle_t*>(&task_queues_async_),
       close_and_finish,
       nullptr);
 }
@@ -666,6 +667,15 @@ void Environment::RunAndClearNativeImmediates(bool only_refed) {
                               "RunAndClearNativeImmediates", this);
   size_t ref_count = 0;
 
+  // It is safe to check .size() first, because there is a causal relationship
+  // between pushes to the threadsafe and this function being called.
+  // For the common case, it's worth checking the size first before establishing
+  // a mutex lock.
+  if (native_immediates_threadsafe_.size() > 0) {
+    Mutex::ScopedLock lock(native_immediates_threadsafe_mutex_);
+    native_immediates_.ConcatMove(std::move(native_immediates_threadsafe_));
+  }
+
   NativeImmediateQueue queue;
   queue.ConcatMove(std::move(native_immediates_));
 
@@ -1087,7 +1097,7 @@ void Environment::CleanupFinalizationGroups() {
       if (try_catch.HasCaught() && !try_catch.HasTerminated())
         errors::TriggerUncaughtException(isolate(), try_catch);
       // Re-schedule the execution of the remainder of the queue.
-      uv_async_send(&cleanup_finalization_groups_async_);
+      uv_async_send(&task_queues_async_);
       return;
     }
   }