src: prepare v8 platform for multi-isolate support

This splits the task queue used for asynchronous tasks scheduled
by V8 in per-isolate queues, so that multiple threads can be supported.

Original-PR-URL: https://github.com/ayojs/ayo/pull/89
Original-Reviewed-By: Timothy Gu <timothygu99@gmail.com>
PR-URL: https://github.com/nodejs/node/pull/16700
Reviewed-By: James M Snell <jasnell@gmail.com>

1 files changed, 94 insertions, 26 deletions

diff --git a/src/node_platform.cc b/src/node_platform.cc
index 56b13b8437..ec2fca6c41 100644
--- a/src/node_platform.cc
+++ b/src/node_platform.cc
@@ -1,6 +1,8 @@
 #include "node_platform.h"
 #include "node_internals.h"
 
+#include "env.h"
+#include "env-inl.h"
 #include "util.h"
 
 namespace node {
@@ -13,11 +15,6 @@ using v8::Platform;
 using v8::Task;
 using v8::TracingController;
 
-static void FlushTasks(uv_async_t* handle) {
-  NodePlatform* platform = static_cast<NodePlatform*>(handle->data);
-  platform->FlushForegroundTasksInternal();
-}
-
 static void BackgroundRunner(void* data) {
   TaskQueue<Task>* background_tasks = static_cast<TaskQueue<Task>*>(data);
   while (Task* task = background_tasks->BlockingPop()) {
@@ -27,12 +24,51 @@ static void BackgroundRunner(void* data) {
   }
 }
 
-NodePlatform::NodePlatform(int thread_pool_size, uv_loop_t* loop,
-                           TracingController* tracing_controller)
-    : loop_(loop) {
-  CHECK_EQ(0, uv_async_init(loop, &flush_tasks_, FlushTasks));
-  flush_tasks_.data = static_cast<void*>(this);
-  uv_unref(reinterpret_cast<uv_handle_t*>(&flush_tasks_));
+PerIsolatePlatformData::PerIsolatePlatformData(
+    v8::Isolate* isolate, uv_loop_t* loop)
+  : isolate_(isolate), loop_(loop) {
+  flush_tasks_ = new uv_async_t();
+  CHECK_EQ(0, uv_async_init(loop, flush_tasks_, FlushTasks));
+  flush_tasks_->data = static_cast<void*>(this);
+  uv_unref(reinterpret_cast<uv_handle_t*>(flush_tasks_));
+}
+
+void PerIsolatePlatformData::FlushTasks(uv_async_t* handle) {
+  auto platform_data = static_cast<PerIsolatePlatformData*>(handle->data);
+  platform_data->FlushForegroundTasksInternal();
+}
+
+void PerIsolatePlatformData::CallOnForegroundThread(Task* task) {
+  foreground_tasks_.Push(task);
+  uv_async_send(flush_tasks_);
+}
+
+void PerIsolatePlatformData::CallDelayedOnForegroundThread(
+    Task* task, double delay_in_seconds) {
+  auto pair = new std::pair<Task*, double>(task, delay_in_seconds);
+  foreground_delayed_tasks_.Push(pair);
+  uv_async_send(flush_tasks_);
+}
+
+PerIsolatePlatformData::~PerIsolatePlatformData() {
+  FlushForegroundTasksInternal();
+
+  uv_close(reinterpret_cast<uv_handle_t*>(flush_tasks_),
+           [](uv_handle_t* handle) {
+    delete reinterpret_cast<uv_async_t*>(handle);
+  });
+}
+
+void PerIsolatePlatformData::ref() {
+  ref_count_++;
+}
+
+int PerIsolatePlatformData::unref() {
+  return --ref_count_;
+}
+
+NodePlatform::NodePlatform(int thread_pool_size,
+                           TracingController* tracing_controller) {
   if (tracing_controller) {
     tracing_controller_.reset(tracing_controller);
   } else {
@@ -49,18 +85,35 @@ NodePlatform::NodePlatform(int thread_pool_size, uv_loop_t* loop,
   }
 }
 
+void NodePlatform::RegisterIsolate(IsolateData* isolate_data, uv_loop_t* loop) {
+  Isolate* isolate = isolate_data->isolate();
+  Mutex::ScopedLock lock(per_isolate_mutex_);
+  PerIsolatePlatformData* existing = per_isolate_[isolate];
+  if (existing != nullptr)
+    existing->ref();
+  else
+    per_isolate_[isolate] = new PerIsolatePlatformData(isolate, loop);
+}
+
+void NodePlatform::UnregisterIsolate(IsolateData* isolate_data) {
+  Isolate* isolate = isolate_data->isolate();
+  Mutex::ScopedLock lock(per_isolate_mutex_);
+  PerIsolatePlatformData* existing = per_isolate_[isolate];
+  CHECK_NE(existing, nullptr);
+  if (existing->unref() == 0) {
+    delete existing;
+    per_isolate_.erase(isolate);
+  }
+}
+
 void NodePlatform::Shutdown() {
   background_tasks_.Stop();
   for (size_t i = 0; i < threads_.size(); i++) {
     CHECK_EQ(0, uv_thread_join(threads_[i].get()));
   }
-  // uv_run cannot be called from the time before the beforeExit callback
-  // runs until the program exits unless the event loop has any referenced
-  // handles after beforeExit terminates. This prevents unrefed timers
-  // that happen to terminate during shutdown from being run unsafely.
-  // Since uv_run cannot be called, this handle will never be fully cleaned
-  // up.
-  uv_close(reinterpret_cast<uv_handle_t*>(&flush_tasks_), nullptr);
+  Mutex::ScopedLock lock(per_isolate_mutex_);
+  for (const auto& pair : per_isolate_)
+    delete pair.second;
 }
 
 size_t NodePlatform::NumberOfAvailableBackgroundThreads() {
@@ -85,13 +138,19 @@ static void RunForegroundTask(uv_timer_t* handle) {
   });
 }
 
-void NodePlatform::DrainBackgroundTasks() {
+void NodePlatform::DrainBackgroundTasks(Isolate* isolate) {
+  PerIsolatePlatformData* per_isolate = ForIsolate(isolate);
+
   do {
+    // Right now, there is no way to drain only background tasks associated with
+    // a specific isolate, so this sometimes does more work than necessary.
+    // In the long run, that functionality is probably going to be available
+    // anyway, though.
     background_tasks_.BlockingDrain();
-  } while (FlushForegroundTasksInternal());
+  } while (per_isolate->FlushForegroundTasksInternal());
 }
 
-bool NodePlatform::FlushForegroundTasksInternal() {
+bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
   bool did_work = false;
   while (auto delayed = foreground_delayed_tasks_.Pop()) {
     did_work = true;
@@ -118,17 +177,26 @@ void NodePlatform::CallOnBackgroundThread(Task* task,
   background_tasks_.Push(task);
 }
 
+PerIsolatePlatformData* NodePlatform::ForIsolate(Isolate* isolate) {
+  Mutex::ScopedLock lock(per_isolate_mutex_);
+  PerIsolatePlatformData* data = per_isolate_[isolate];
+  CHECK_NE(data, nullptr);
+  return data;
+}
+
 void NodePlatform::CallOnForegroundThread(Isolate* isolate, Task* task) {
-  foreground_tasks_.Push(task);
-  uv_async_send(&flush_tasks_);
+  ForIsolate(isolate)->CallOnForegroundThread(task);
 }
 
 void NodePlatform::CallDelayedOnForegroundThread(Isolate* isolate,
                                                  Task* task,
                                                  double delay_in_seconds) {
-  auto pair = new std::pair<Task*, double>(task, delay_in_seconds);
-  foreground_delayed_tasks_.Push(pair);
-  uv_async_send(&flush_tasks_);
+  ForIsolate(isolate)->CallDelayedOnForegroundThread(task,
+                                                     delay_in_seconds);
+}
+
+void NodePlatform::FlushForegroundTasks(v8::Isolate* isolate) {
+  ForIsolate(isolate)->FlushForegroundTasksInternal();
 }
 
 bool NodePlatform::IdleTasksEnabled(Isolate* isolate) { return false; }