diff --git a/src/node_platform.cc b/src/node_platform.cc index 57a43eeb6459..db8f85859bac 100644 --- a/src/node_platform.cc +++ b/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& 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 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(this); @@ -355,12 +366,17 @@ void PerIsolatePlatformData::PostTaskImpl(std::unique_ptr 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( - 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( + 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(isolate, loop, debug_log_level_); + auto delegate = std::make_shared( + 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 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::TaskQueue() : lock_(), tasks_available_(), outstanding_tasks_drained_(), + drain_wakeup_generation_(0), outstanding_tasks_(0), stopped_(false), task_queue_() {} @@ -830,9 +833,40 @@ void TaskQueue::Locked::NotifyOfOutstandingCompletion() { } template -void TaskQueue::Locked::BlockingDrain() { - while (queue_->outstanding_tasks_ > 0) { - queue_->outstanding_tasks_drained_.Wait(lock_); +void TaskQueue::Locked::WakeDrain() { + queue_->drain_wakeup_generation_++; + queue_->outstanding_tasks_drained_.Broadcast(lock_); +} + +template +void TaskQueue::Locked::BlockingDrain( + const std::function& 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; } } diff --git a/src/node_platform.h b/src/node_platform.h index f47e2a46b66b..4710db7b4b9f 100644 --- a/src/node_platform.h +++ b/src/node_platform.h @@ -3,6 +3,7 @@ #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS +#include #include #include #include @@ -19,6 +20,7 @@ namespace node { class NodePlatform; class IsolateData; class PerIsolatePlatformData; +class WorkerThreadsTaskRunner; template struct has_priority : std::false_type {}; @@ -52,7 +54,9 @@ class TaskQueue { std::unique_ptr Pop(); std::unique_ptr BlockingPop(); void NotifyOfOutstandingCompletion(); - void BlockingDrain(); + void WakeDrain(); + void BlockingDrain( + const std::function& 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 worker_thread_task_runner = {}); ~PerIsolatePlatformData() override; std::shared_ptr GetForegroundTaskRunner() override; @@ -178,6 +184,7 @@ class PerIsolatePlatformData typedef std::unique_ptr DelayedTaskPointer; std::vector scheduled_delayed_tasks_; + std::weak_ptr 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& flush_foreground_tasks); + void NotifyForegroundTaskPosted(); void Shutdown(); int NumberOfWorkerThreads() const; diff --git a/test/cctest/test_platform.cc b/test/cctest/test_platform.cc index f1c1d52d92c7..69e59b84887a 100644 --- a/test/cctest/test_platform.cc +++ b/test/cctest/test_platform.cc @@ -1,6 +1,8 @@ #include "node_internals.h" #include "libplatform/libplatform.h" +#include +#include #include #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 foreground_task_runner, + uv_sem_t* foreground_task_finished, + std::atomic* 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(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 foreground_task_runner_; + uv_sem_t* foreground_task_finished_; + std::atomic* 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 unblocked_by_foreground{false}; + auto foreground_task_runner = platform->GetForegroundTaskRunner( + isolate_, v8::TaskPriority::kUserBlocking); + platform->PostTaskOnWorkerThread( + v8::TaskPriority::kUserBlocking, + std::make_unique( + 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) {