diff --git a/src/node_platform.cc b/src/node_platform.cc index 0c50e8468d0f44..65d1ef6c5dfe28 100644 --- a/src/node_platform.cc +++ b/src/node_platform.cc @@ -24,6 +24,43 @@ static void BackgroundRunner(void* data) { } } +BackgroundTaskRunner::BackgroundTaskRunner(int thread_pool_size) { + for (int i = 0; i < thread_pool_size; i++) { + std::unique_ptr t { new uv_thread_t() }; + if (uv_thread_create(t.get(), BackgroundRunner, &background_tasks_) != 0) + break; + threads_.push_back(std::move(t)); + } +} + +void BackgroundTaskRunner::PostTask(std::unique_ptr task) { + background_tasks_.Push(std::move(task)); +} + +void BackgroundTaskRunner::PostIdleTask(std::unique_ptr task) { + UNREACHABLE(); +} + +void BackgroundTaskRunner::PostDelayedTask(std::unique_ptr task, + double delay_in_seconds) { + UNREACHABLE(); +} + +void BackgroundTaskRunner::BlockingDrain() { + background_tasks_.BlockingDrain(); +} + +void BackgroundTaskRunner::Shutdown() { + background_tasks_.Stop(); + for (size_t i = 0; i < threads_.size(); i++) { + CHECK_EQ(0, uv_thread_join(threads_[i].get())); + } +} + +size_t BackgroundTaskRunner::NumberOfAvailableBackgroundThreads() const { + return threads_.size(); +} + PerIsolatePlatformData::PerIsolatePlatformData( v8::Isolate* isolate, uv_loop_t* loop) : isolate_(isolate), loop_(loop) { @@ -38,17 +75,20 @@ void PerIsolatePlatformData::FlushTasks(uv_async_t* handle) { platform_data->FlushForegroundTasksInternal(); } -void PerIsolatePlatformData::CallOnForegroundThread( - std::unique_ptr task) { +void PerIsolatePlatformData::PostIdleTask(std::unique_ptr task) { + UNREACHABLE(); +} + +void PerIsolatePlatformData::PostTask(std::unique_ptr task) { foreground_tasks_.Push(std::move(task)); uv_async_send(flush_tasks_); } -void PerIsolatePlatformData::CallDelayedOnForegroundThread( - std::unique_ptr task, double delay_in_seconds) { +void PerIsolatePlatformData::PostDelayedTask( + std::unique_ptr task, double delay_in_seconds) { std::unique_ptr delayed(new DelayedTask()); delayed->task = std::move(task); - delayed->platform_data = this; + delayed->platform_data = shared_from_this(); delayed->timeout = delay_in_seconds; foreground_delayed_tasks_.Push(std::move(delayed)); uv_async_send(flush_tasks_); @@ -80,49 +120,43 @@ NodePlatform::NodePlatform(int thread_pool_size, TracingController* controller = new TracingController(); tracing_controller_.reset(controller); } - for (int i = 0; i < thread_pool_size; i++) { - uv_thread_t* t = new uv_thread_t(); - if (uv_thread_create(t, BackgroundRunner, &background_tasks_) != 0) { - delete t; - break; - } - threads_.push_back(std::unique_ptr(t)); - } + background_task_runner_ = + std::make_shared(thread_pool_size); } 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) + std::shared_ptr existing = per_isolate_[isolate]; + if (existing) { existing->ref(); - else - per_isolate_[isolate] = new PerIsolatePlatformData(isolate, loop); + } else { + per_isolate_[isolate] = + std::make_shared(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); + std::shared_ptr existing = per_isolate_[isolate]; + CHECK(existing); 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())); + background_task_runner_->Shutdown(); + + { + Mutex::ScopedLock lock(per_isolate_mutex_); + per_isolate_.clear(); } - Mutex::ScopedLock lock(per_isolate_mutex_); - for (const auto& pair : per_isolate_) - delete pair.second; } size_t NodePlatform::NumberOfAvailableBackgroundThreads() { - return threads_.size(); + return background_task_runner_->NumberOfAvailableBackgroundThreads(); } void PerIsolatePlatformData::RunForegroundTask(std::unique_ptr task) { @@ -155,14 +189,14 @@ void PerIsolatePlatformData::CancelPendingDelayedTasks() { } void NodePlatform::DrainBackgroundTasks(Isolate* isolate) { - PerIsolatePlatformData* per_isolate = ForIsolate(isolate); + std::shared_ptr 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(); + background_task_runner_->BlockingDrain(); } while (per_isolate->FlushForegroundTasksInternal()); } @@ -198,24 +232,25 @@ bool PerIsolatePlatformData::FlushForegroundTasksInternal() { void NodePlatform::CallOnBackgroundThread(Task* task, ExpectedRuntime expected_runtime) { - background_tasks_.Push(std::unique_ptr(task)); + background_task_runner_->PostTask(std::unique_ptr(task)); } -PerIsolatePlatformData* NodePlatform::ForIsolate(Isolate* isolate) { +std::shared_ptr +NodePlatform::ForIsolate(Isolate* isolate) { Mutex::ScopedLock lock(per_isolate_mutex_); - PerIsolatePlatformData* data = per_isolate_[isolate]; - CHECK_NE(data, nullptr); + std::shared_ptr data = per_isolate_[isolate]; + CHECK(data); return data; } void NodePlatform::CallOnForegroundThread(Isolate* isolate, Task* task) { - ForIsolate(isolate)->CallOnForegroundThread(std::unique_ptr(task)); + ForIsolate(isolate)->PostTask(std::unique_ptr(task)); } void NodePlatform::CallDelayedOnForegroundThread(Isolate* isolate, Task* task, double delay_in_seconds) { - ForIsolate(isolate)->CallDelayedOnForegroundThread( + ForIsolate(isolate)->PostDelayedTask( std::unique_ptr(task), delay_in_seconds); } @@ -229,6 +264,16 @@ void NodePlatform::CancelPendingDelayedTasks(v8::Isolate* isolate) { bool NodePlatform::IdleTasksEnabled(Isolate* isolate) { return false; } +std::shared_ptr +NodePlatform::GetBackgroundTaskRunner(Isolate* isolate) { + return background_task_runner_; +} + +std::shared_ptr +NodePlatform::GetForegroundTaskRunner(Isolate* isolate) { + return ForIsolate(isolate); +} + double NodePlatform::MonotonicallyIncreasingTime() { // Convert nanos to seconds. return uv_hrtime() / 1e9; diff --git a/src/node_platform.h b/src/node_platform.h index 48301a05a11d8c..e5253cac10c3b0 100644 --- a/src/node_platform.h +++ b/src/node_platform.h @@ -43,17 +43,22 @@ struct DelayedTask { std::unique_ptr task; uv_timer_t timer; double timeout; - PerIsolatePlatformData* platform_data; + std::shared_ptr platform_data; }; -class PerIsolatePlatformData { +// This acts as the foreground task runner for a given Isolate. +class PerIsolatePlatformData : + public v8::TaskRunner, + public std::enable_shared_from_this { public: PerIsolatePlatformData(v8::Isolate* isolate, uv_loop_t* loop); ~PerIsolatePlatformData(); - void CallOnForegroundThread(std::unique_ptr task); - void CallDelayedOnForegroundThread(std::unique_ptr task, - double delay_in_seconds); + void PostTask(std::unique_ptr task) override; + void PostIdleTask(std::unique_ptr task) override; + void PostDelayedTask(std::unique_ptr task, + double delay_in_seconds) override; + bool IdleTasksEnabled() override { return false; }; void Shutdown(); @@ -84,6 +89,26 @@ class PerIsolatePlatformData { std::vector scheduled_delayed_tasks_; }; +// This acts as the single background task runner for all Isolates. +class BackgroundTaskRunner : public v8::TaskRunner { + public: + explicit BackgroundTaskRunner(int thread_pool_size); + + void PostTask(std::unique_ptr task) override; + void PostIdleTask(std::unique_ptr task) override; + void PostDelayedTask(std::unique_ptr task, + double delay_in_seconds) override; + bool IdleTasksEnabled() override { return false; }; + + void BlockingDrain(); + void Shutdown(); + + size_t NumberOfAvailableBackgroundThreads() const; + private: + TaskQueue background_tasks_; + std::vector> threads_; +}; + class NodePlatform : public MultiIsolatePlatform { public: NodePlatform(int thread_pool_size, v8::TracingController* tracing_controller); @@ -109,15 +134,20 @@ class NodePlatform : public MultiIsolatePlatform { void RegisterIsolate(IsolateData* isolate_data, uv_loop_t* loop) override; void UnregisterIsolate(IsolateData* isolate_data) override; + std::shared_ptr GetBackgroundTaskRunner( + v8::Isolate* isolate) override; + std::shared_ptr GetForegroundTaskRunner( + v8::Isolate* isolate) override; + private: - PerIsolatePlatformData* ForIsolate(v8::Isolate* isolate); + std::shared_ptr ForIsolate(v8::Isolate* isolate); Mutex per_isolate_mutex_; - std::unordered_map per_isolate_; - TaskQueue background_tasks_; - std::vector> threads_; + std::unordered_map> per_isolate_; std::unique_ptr tracing_controller_; + std::shared_ptr background_task_runner_; }; } // namespace node