src: refactor thread stopping mechanism

- Follow style guide for naming, e.g. use lower_snake_case for simple setters/getters. - For performance, use atomics instead of a mutex, and inline the corresponding getter/setter pair.
nodejs · addaleax · Mar 18, 2019 · Mar 19, 2019 · Mar 18, 2019 · 47a0854aaa768ddfcfd653ab8e67baef35110d3c
commit 47a0854aaa768ddfcfd653ab8e67baef35110d3c
diff --git a/src/env-inl.h b/src/env-inl.h
@@ -709,7 +709,7 @@ inline void Environment::remove_sub_worker_context(worker::Worker* context) {
 }
 
 inline bool Environment::is_stopping() const {
-  return thread_stopper_.IsStopped();
+  return thread_stopper_.is_stopped();
 }
 
 inline performance::performance_state* Environment::performance_state() {
@@ -979,6 +979,14 @@ void Environment::ForEachBaseObject(T&& iterator) {
   }
 }
 
+bool AsyncRequest::is_stopped() const {
+  return stopped_.load(std::memory_order_relaxed);
+}
+
+void AsyncRequest::set_stopped(bool flag) {
+  stopped_.store(flag, std::memory_order_relaxed);
+}
+
 #define VP(PropertyName, StringValue) V(v8::Private, PropertyName)
 #define VY(PropertyName, StringValue) V(v8::Symbol, PropertyName)
 #define VS(PropertyName, StringValue) V(v8::String, PropertyName)

diff --git a/src/env.cc b/src/env.cc
@@ -340,13 +340,13 @@ void Environment::InitializeLibuv(bool start_profiler_idle_notifier) {
   uv_unref(reinterpret_cast<uv_handle_t*>(&idle_prepare_handle_));
   uv_unref(reinterpret_cast<uv_handle_t*>(&idle_check_handle_));
 
-  GetAsyncRequest()->Install(
+  thread_stopper()->Install(
     this, static_cast<void*>(this), [](uv_async_t* handle) {
       Environment* env = static_cast<Environment*>(handle->data);
       uv_stop(env->event_loop());
     });
-  GetAsyncRequest()->SetStopped(false);
-  uv_unref(reinterpret_cast<uv_handle_t*>(GetAsyncRequest()->GetHandle()));
+  thread_stopper()->set_stopped(false);
+  uv_unref(reinterpret_cast<uv_handle_t*>(thread_stopper()->GetHandle()));
 
   // Register clean-up cb to be called to clean up the handles
   // when the environment is freed, note that they are not cleaned in
@@ -365,7 +365,7 @@ void Environment::InitializeLibuv(bool start_profiler_idle_notifier) {
 
 void Environment::ExitEnv() {
   set_can_call_into_js(false);
-  GetAsyncRequest()->Stop();
+  thread_stopper()->Stop();
   isolate_->TerminateExecution();
 }
 
@@ -533,7 +533,7 @@ void Environment::RunCleanup() {
   started_cleanup_ = true;
   TraceEventScope trace_scope(TRACING_CATEGORY_NODE1(environment),
                               "RunCleanup", this);
-  GetAsyncRequest()->Uninstall();
+  thread_stopper()->Uninstall();
   CleanupHandles();
 
   while (!cleanup_hooks_.empty()) {
@@ -948,49 +948,34 @@ char* Environment::Reallocate(char* data, size_t old_size, size_t size) {
 }
 
 void AsyncRequest::Install(Environment* env, void* data, uv_async_cb target) {
-  Mutex::ScopedLock lock(mutex_);
+  CHECK_NULL(async_);
   env_ = env;
   async_ = new uv_async_t;
   async_->data = data;
   CHECK_EQ(uv_async_init(env_->event_loop(), async_, target), 0);
 }
 
 void AsyncRequest::Uninstall() {
-  Mutex::ScopedLock lock(mutex_);
   if (async_ != nullptr) {
     env_->CloseHandle(async_, [](uv_async_t* async) { delete async; });
     async_ = nullptr;
   }
 }
 
 void AsyncRequest::Stop() {
-  Mutex::ScopedLock lock(mutex_);
-  stop_ = true;
+  set_stopped(true);
   if (async_ != nullptr) uv_async_send(async_);
 }
 
-void AsyncRequest::SetStopped(bool flag) {
-  Mutex::ScopedLock lock(mutex_);
-  stop_ = flag;
-}
-
-bool AsyncRequest::IsStopped() const {
-  Mutex::ScopedLock lock(mutex_);
-  return stop_;
-}
-
 uv_async_t* AsyncRequest::GetHandle() {
-  Mutex::ScopedLock lock(mutex_);
   return async_;
 }
 
 void AsyncRequest::MemoryInfo(MemoryTracker* tracker) const {
-  Mutex::ScopedLock lock(mutex_);
   if (async_ != nullptr) tracker->TrackField("async_request", *async_);
 }
 
 AsyncRequest::~AsyncRequest() {
-  Mutex::ScopedLock lock(mutex_);
   CHECK_NULL(async_);
 }
 

diff --git a/src/env.h b/src/env.h
@@ -38,6 +38,7 @@
 #include "uv.h"
 #include "v8.h"
 
+#include <atomic>
 #include <cstdint>
 #include <functional>
 #include <list>
@@ -518,18 +519,19 @@ class AsyncRequest : public MemoryRetainer {
   void Install(Environment* env, void* data, uv_async_cb target);
   void Uninstall();
   void Stop();
-  void SetStopped(bool flag);
-  bool IsStopped() const;
+  inline void set_stopped(bool flag);
+  inline bool is_stopped() const;
   uv_async_t* GetHandle();
   void MemoryInfo(MemoryTracker* tracker) const override;
+
+
   SET_MEMORY_INFO_NAME(AsyncRequest)
   SET_SELF_SIZE(AsyncRequest)
 
  private:
   Environment* env_;
   uv_async_t* async_ = nullptr;
-  mutable Mutex mutex_;
-  bool stop_ = true;
+  std::atomic_bool stopped_ {true};
 };
 
 class Environment {
@@ -1048,7 +1050,7 @@ class Environment {
   inline ExecutionMode execution_mode() { return execution_mode_; }
 
   inline void set_execution_mode(ExecutionMode mode) { execution_mode_ = mode; }
-  inline AsyncRequest* GetAsyncRequest() { return &thread_stopper_; }
+  inline AsyncRequest* thread_stopper() { return &thread_stopper_; }
 
  private:
   inline void CreateImmediate(native_immediate_callback cb,

diff --git a/src/node.cc b/src/node.cc
@@ -832,14 +832,14 @@ inline int StartNodeWithIsolate(Isolate* isolate,
       per_process::v8_platform.DrainVMTasks(isolate);
 
       more = uv_loop_alive(env.event_loop());
-      if (more && !env.GetAsyncRequest()->IsStopped()) continue;
+      if (more && !env.is_stopping()) continue;
 
       RunBeforeExit(&env);
 
       // Emit `beforeExit` if the loop became alive either after emitting
       // event, or after running some callbacks.
       more = uv_loop_alive(env.event_loop());
-    } while (more == true && !env.GetAsyncRequest()->IsStopped());
+    } while (more == true && !env.is_stopping());
     env.performance_state()->Mark(
         node::performance::NODE_PERFORMANCE_MILESTONE_LOOP_EXIT);
   }

diff --git a/src/node_worker.cc b/src/node_worker.cc
@@ -104,7 +104,7 @@ Worker::Worker(Environment* env,
 bool Worker::is_stopped() const {
   Mutex::ScopedLock lock(mutex_);
   if (env_ != nullptr)
-    return env_->GetAsyncRequest()->IsStopped();
+    return env_->is_stopping();
   return stopped_;
 }
 
@@ -222,7 +222,7 @@ void Worker::Run() {
           stopped_ = true;
           this->env_ = nullptr;
         }
-        env_->GetAsyncRequest()->SetStopped(true);
+        env_->thread_stopper()->set_stopped(true);
         env_->stop_sub_worker_contexts();
         env_->RunCleanup();
         RunAtExit(env_.get());
@@ -381,7 +381,8 @@ void Worker::OnThreadStopped() {
 Worker::~Worker() {
   Mutex::ScopedLock lock(mutex_);
 
-  CHECK(stopped_ || env_ == nullptr || env_->GetAsyncRequest()->IsStopped());
+  CHECK(stopped_);
+  CHECK_NULL(env_);
   CHECK(thread_joined_);
 
   Debug(this, "Worker %llu destroyed", thread_id_);