src: wake worker drain for foreground tasks

src/node_platform.cc

@@ -300,8 +300,13 @@ void WorkerThreadsTaskRunner::PostDelayedTask(
       priority, std::move(task), delay_in_seconds);
 }
 
-void WorkerThreadsTaskRunner::BlockingDrain() {
-  pending_worker_tasks_.Lock().BlockingDrain();
+void WorkerThreadsTaskRunner::BlockingDrain(
+    const std::function<bool()>& flush_foreground_tasks) {
+  pending_worker_tasks_.Lock().BlockingDrain(flush_foreground_tasks);
+}
+
+void WorkerThreadsTaskRunner::NotifyForegroundTaskPosted() {
+  pending_worker_tasks_.Lock().WakeDrain();
 }
 
 void WorkerThreadsTaskRunner::Shutdown() {
@@ -317,8 +322,14 @@ int WorkerThreadsTaskRunner::NumberOfWorkerThreads() const {
 }
 
 PerIsolatePlatformData::PerIsolatePlatformData(
-    Isolate* isolate, uv_loop_t* loop, PlatformDebugLogLevel debug_log_level)
-    : isolate_(isolate), loop_(loop), debug_log_level_(debug_log_level) {
+    Isolate* isolate,
+    uv_loop_t* loop,
+    PlatformDebugLogLevel debug_log_level,
+    std::weak_ptr<WorkerThreadsTaskRunner> worker_thread_task_runner)
+    : isolate_(isolate),
+      loop_(loop),
+      worker_thread_task_runner_(std::move(worker_thread_task_runner)),
+      debug_log_level_(debug_log_level) {
   flush_tasks_ = new uv_async_t();
   CHECK_EQ(0, uv_async_init(loop, flush_tasks_, FlushTasks));
   flush_tasks_->data = static_cast<void*>(this);
@@ -355,12 +366,17 @@ void PerIsolatePlatformData::PostTaskImpl(std::unique_ptr<Task> task,
     fflush(stderr);
   }
 
-  auto locked = foreground_tasks_.Lock();
-  if (flush_tasks_ == nullptr) return;
-  // All foreground tasks are treated as user blocking tasks.
-  locked.Push(std::make_unique<TaskQueueEntry>(
-      std::move(task), v8::TaskPriority::kUserBlocking));
-  uv_async_send(flush_tasks_);
+  {
+    auto locked = foreground_tasks_.Lock();
+    if (flush_tasks_ == nullptr) return;
+    // All foreground tasks are treated as user blocking tasks.
+    locked.Push(std::make_unique<TaskQueueEntry>(
+        std::move(task), v8::TaskPriority::kUserBlocking));
+    uv_async_send(flush_tasks_);
+  }
+  if (auto runner = worker_thread_task_runner_.lock()) {
+    runner->NotifyForegroundTaskPosted();
+  }
 }
 
 void PerIsolatePlatformData::PostDelayedTaskImpl(
@@ -486,8 +502,8 @@ NodePlatform::~NodePlatform() {
 
 void NodePlatform::RegisterIsolate(Isolate* isolate, uv_loop_t* loop) {
   Mutex::ScopedLock lock(per_isolate_mutex_);
-  auto delegate =
-      std::make_shared<PerIsolatePlatformData>(isolate, loop, debug_log_level_);
+  auto delegate = std::make_shared<PerIsolatePlatformData>(
+      isolate, loop, debug_log_level_, worker_thread_task_runner_);
   IsolatePlatformDelegate* ptr = delegate.get();
   auto insertion = per_isolate_.emplace(
     isolate,
@@ -584,28 +600,14 @@ void NodePlatform::DrainTasks(Isolate* isolate) {
   std::shared_ptr<PerIsolatePlatformData> per_isolate = ForNodeIsolate(isolate);
   if (!per_isolate) return;
 
-  do {
-    // FIXME(54918): we should not be blocking on the worker tasks on the
-    // main thread in one go. Doing so leads to two problems:
-    // 1. If any of the worker tasks post another foreground task and wait
-    //    for it to complete, and that foreground task is posted right after
-    //    we flush the foreground task queue and before the foreground thread
-    //    goes into sleep, we'll never be able to wake up to execute that
-    //    foreground task and in turn the worker task will never complete, and
-    //    we have a deadlock.
-    // 2. Worker tasks can be posted from any thread, not necessarily associated
-    //    with the current isolate, and we can be blocking on a worker task that
-    //    is associated with a completely unrelated isolate in the event loop.
-    //    This is suboptimal.
-    //
-    // However, not blocking on the worker tasks at all can lead to loss of some
-    // critical user-blocking worker tasks e.g. wasm async compilation tasks,
-    // which should block the main thread until they are completed, as the
-    // documentation suggests. As a compromise, we currently only block on
-    // user-blocking tasks to reduce the chance of deadlocks while making sure
-    // that critical user-blocking tasks are not lost.
-    worker_thread_task_runner_->BlockingDrain();
-  } while (per_isolate->FlushForegroundTasksInternal());
+  // Worker tasks are shared by all isolates, so this can still wait for
+  // unrelated work. BlockingDrain() only waits for user-blocking tasks so
+  // critical work such as asynchronous Wasm compilation is not discarded.
+  // Flush this isolate's foreground tasks while waiting so user-blocking
+  // workers that depend on them can make progress.
+  worker_thread_task_runner_->BlockingDrain([per_isolate]() {
+    return per_isolate->FlushForegroundTasksInternal();
+  });
 }
 
 bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
@@ -780,6 +782,7 @@ TaskQueue<T>::TaskQueue()
     : lock_(),
       tasks_available_(),
       outstanding_tasks_drained_(),
+      drain_wakeup_generation_(0),
       outstanding_tasks_(0),
       stopped_(false),
       task_queue_() {}
@@ -830,9 +833,40 @@ void TaskQueue<T>::Locked::NotifyOfOutstandingCompletion() {
 }
 
 template <class T>
-void TaskQueue<T>::Locked::BlockingDrain() {
-  while (queue_->outstanding_tasks_ > 0) {
-    queue_->outstanding_tasks_drained_.Wait(lock_);
+void TaskQueue<T>::Locked::WakeDrain() {
+  queue_->drain_wakeup_generation_++;
+  queue_->outstanding_tasks_drained_.Broadcast(lock_);
+}
+
+template <class T>
+void TaskQueue<T>::Locked::BlockingDrain(
+    const std::function<bool()>& flush_foreground_tasks) {
+  // Foreground tasks are flushed without the queue lock. Remember wakeups that
+  // arrive in that interval so they cannot be lost before Wait().
+  uint64_t observed_generation = queue_->drain_wakeup_generation_;
+  while (true) {
+    bool did_work;
+    {
+      Mutex::ScopedUnlock unlock(lock_);
+      did_work = flush_foreground_tasks();
+    }
+
+    if (queue_->drain_wakeup_generation_ != observed_generation) {
+      observed_generation = queue_->drain_wakeup_generation_;
+      continue;
+    }
+
+    while (queue_->outstanding_tasks_ > 0 &&
+           queue_->drain_wakeup_generation_ == observed_generation) {
+      queue_->outstanding_tasks_drained_.Wait(lock_);
+    }
+
+    if (queue_->drain_wakeup_generation_ != observed_generation) {
+      observed_generation = queue_->drain_wakeup_generation_;
+      continue;
+    }
+
+    if (!did_work) return;
   }
 }
 

src/node_platform.h

@@ -3,6 +3,7 @@
 
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
+#include <cstdint>
 #include <functional>
 #include <queue>
 #include <type_traits>
@@ -19,6 +20,7 @@ namespace node {
 class NodePlatform;
 class IsolateData;
 class PerIsolatePlatformData;
+class WorkerThreadsTaskRunner;
 
 template <typename, typename = void>
 struct has_priority : std::false_type {};
@@ -52,7 +54,9 @@ class TaskQueue {
     std::unique_ptr<T> Pop();
     std::unique_ptr<T> BlockingPop();
     void NotifyOfOutstandingCompletion();
-    void BlockingDrain();
+    void WakeDrain();
+    void BlockingDrain(
+        const std::function<bool()>& flush_foreground_tasks);
     void Stop();
     PriorityQueue PopAll();
 
@@ -73,6 +77,7 @@ class TaskQueue {
   Mutex lock_;
   ConditionVariable tasks_available_;
   ConditionVariable outstanding_tasks_drained_;
+  uint64_t drain_wakeup_generation_;
   int outstanding_tasks_;
   bool stopped_;
   PriorityQueue task_queue_;
@@ -111,7 +116,8 @@ class PerIsolatePlatformData
   PerIsolatePlatformData(
       v8::Isolate* isolate,
       uv_loop_t* loop,
-      PlatformDebugLogLevel debug_log_level = PlatformDebugLogLevel::kNone);
+      PlatformDebugLogLevel debug_log_level = PlatformDebugLogLevel::kNone,
+      std::weak_ptr<WorkerThreadsTaskRunner> worker_thread_task_runner = {});
   ~PerIsolatePlatformData() override;
 
   std::shared_ptr<v8::TaskRunner> GetForegroundTaskRunner() override;
@@ -178,6 +184,7 @@ class PerIsolatePlatformData
   typedef std::unique_ptr<DelayedTask, void (*)(DelayedTask*)>
       DelayedTaskPointer;
   std::vector<DelayedTaskPointer> scheduled_delayed_tasks_;
+  std::weak_ptr<WorkerThreadsTaskRunner> worker_thread_task_runner_;
   PlatformDebugLogLevel debug_log_level_ = PlatformDebugLogLevel::kNone;
 };
 
@@ -195,7 +202,9 @@ class WorkerThreadsTaskRunner {
                        const v8::SourceLocation& location,
                        double delay_in_seconds);
 
-  void BlockingDrain();
+  void BlockingDrain(
+      const std::function<bool()>& flush_foreground_tasks);
+  void NotifyForegroundTaskPosted();
   void Shutdown();
 
   int NumberOfWorkerThreads() const;

test/cctest/test_platform.cc

@@ -1,6 +1,8 @@
 #include "node_internals.h"
 #include "libplatform/libplatform.h"
 
+#include <atomic>
+#include <cstdint>
 #include <string>
 #include "gtest/gtest.h"
 #include "node_test_fixture.h"
@@ -38,6 +40,51 @@ class RepostingTask : public v8::Task {
   node::NodePlatform* platform_;
 };
 
+class ForegroundSignalTask : public v8::Task {
+ public:
+  explicit ForegroundSignalTask(uv_sem_t* semaphore) : semaphore_(semaphore) {}
+
+  void Run() final { uv_sem_post(semaphore_); }
+
+ private:
+  uv_sem_t* semaphore_;
+};
+
+class WorkerTaskWaitingForForeground : public v8::Task {
+ public:
+  WorkerTaskWaitingForForeground(
+      std::shared_ptr<v8::TaskRunner> foreground_task_runner,
+      uv_sem_t* foreground_task_finished,
+      std::atomic<bool>* unblocked_by_foreground)
+      : foreground_task_runner_(std::move(foreground_task_runner)),
+        foreground_task_finished_(foreground_task_finished),
+        unblocked_by_foreground_(unblocked_by_foreground) {}
+
+  void Run() final {
+    foreground_task_runner_->PostTask(
+        std::make_unique<ForegroundSignalTask>(foreground_task_finished_));
+
+    // Bound the wait so an unfixed binary fails instead of hanging cctest.
+    constexpr uint64_t kTimeoutNanoseconds = 5'000'000'000ULL;
+    const uint64_t deadline = uv_hrtime() + kTimeoutNanoseconds;
+    while (true) {
+      int err = uv_sem_trywait(foreground_task_finished_);
+      if (err == 0) {
+        unblocked_by_foreground_->store(true);
+        return;
+      }
+      CHECK_EQ(UV_EAGAIN, err);
+      if (uv_hrtime() >= deadline) return;
+      uv_sleep(1);
+    }
+  }
+
+ private:
+  std::shared_ptr<v8::TaskRunner> foreground_task_runner_;
+  uv_sem_t* foreground_task_finished_;
+  std::atomic<bool>* unblocked_by_foreground_;
+};
+
 class PlatformTest : public EnvironmentTestFixture {};
 
 TEST_F(PlatformTest, SkipNewTasksInFlushForegroundTasks) {
@@ -60,6 +107,30 @@ TEST_F(PlatformTest, SkipNewTasksInFlushForegroundTasks) {
   EXPECT_FALSE(platform->FlushForegroundTasks(isolate_));
 }
 
+TEST_F(PlatformTest, DrainTasksRunsForegroundTasksNeededByWorker) {
+  v8::Isolate::Scope isolate_scope(isolate_);
+  const v8::HandleScope handle_scope(isolate_);
+  const Argv argv;
+  Env env {handle_scope, argv};
+
+  uv_sem_t foreground_task_finished;
+  CHECK_EQ(0, uv_sem_init(&foreground_task_finished, 0));
+  std::atomic<bool> unblocked_by_foreground{false};
+  auto foreground_task_runner = platform->GetForegroundTaskRunner(
+      isolate_, v8::TaskPriority::kUserBlocking);
+  platform->PostTaskOnWorkerThread(
+      v8::TaskPriority::kUserBlocking,
+      std::make_unique<WorkerTaskWaitingForForeground>(
+          std::move(foreground_task_runner),
+          &foreground_task_finished,
+          &unblocked_by_foreground));
+
+  platform->DrainTasks(isolate_);
+
+  EXPECT_TRUE(unblocked_by_foreground.load());
+  uv_sem_destroy(&foreground_task_finished);
+}
+
 // Tests the registration of an abstract `IsolatePlatformDelegate` instance as
 // opposed to the more common `uv_loop_s*` version of `RegisterIsolate`.
 TEST_F(NodeZeroIsolateTestFixture, IsolatePlatformDelegateTest) {