worker: refactor thread id management

- Assign thread IDs to `Environment` instances, rather than Workers.
  This is more embedder-friendly than the current system, in which
  all “main threads” (if there are multiple ones) would get the
  id `0`.
- Because that means that `isMainThread === (threadId === 0)` no longer
  holds, refactor `isMainThread` into a separate entity. Implement it
  in a way that allows for future extensibility, because we use
  `isMainThread` in multiple different ways (determining whether there
  is a parent thread; determining whether the current thread has control
  of the current process; etc.).

PR-URL: #25796
Reviewed-By: Joyee Cheung <[email protected]>
Reviewed-By: James M Snell <[email protected]>
Reviewed-By: Benjamin Gruenbaum <[email protected]>
Reviewed-By: Denys Otrishko <[email protected]>

lib/internal/worker.js

@@ -31,11 +31,10 @@ const { pathToFileURL } = require('url');
 
 const {
   Worker: WorkerImpl,
-  threadId
+  threadId,
+  isMainThread
 } = internalBinding('worker');
 
-const isMainThread = threadId === 0;
-
 const kHandle = Symbol('kHandle');
 const kPublicPort = Symbol('kPublicPort');
 const kDispose = Symbol('kDispose');

src/api/environment.cc

@@ -133,7 +133,9 @@ Environment* CreateEnvironment(IsolateData* isolate_data,
   // options than the global parse call.
   std::vector<std::string> args(argv, argv + argc);
   std::vector<std::string> exec_args(exec_argv, exec_argv + exec_argc);
-  Environment* env = new Environment(isolate_data, context);
+  // TODO(addaleax): Provide more sensible flags, in an embedder-accessible way.
+  Environment* env =
+      new Environment(isolate_data, context, Environment::kIsMainThread);
   env->Start(per_process::v8_is_profiling);
   env->ProcessCliArgs(args, exec_args);
   return env;

src/env-inl.h

@@ -647,17 +647,13 @@ inline void Environment::set_has_run_bootstrapping_code(bool value) {
 }
 
 inline bool Environment::is_main_thread() const {
-  return thread_id_ == 0;
+  return flags_ & kIsMainThread;
 }
 
 inline uint64_t Environment::thread_id() const {
   return thread_id_;
 }
 
-inline void Environment::set_thread_id(uint64_t id) {
-  thread_id_ = id;
-}
-
 inline worker::Worker* Environment::worker_context() const {
   return worker_context_;
 }

src/env.cc

@@ -16,6 +16,7 @@
 
 #include <stdio.h>
 #include <algorithm>
+#include <atomic>
 
 namespace node {
 
@@ -166,8 +167,11 @@ void Environment::TrackingTraceStateObserver::UpdateTraceCategoryState() {
            0, nullptr).ToLocalChecked();
 }
 
+static std::atomic<uint64_t> next_thread_id{0};
+
 Environment::Environment(IsolateData* isolate_data,
-                         Local<Context> context)
+                         Local<Context> context,
+                         Flags flags)
     : isolate_(context->GetIsolate()),
       isolate_data_(isolate_data),
       immediate_info_(context->GetIsolate()),
@@ -176,6 +180,8 @@ Environment::Environment(IsolateData* isolate_data,
       should_abort_on_uncaught_toggle_(isolate_, 1),
       trace_category_state_(isolate_, kTraceCategoryCount),
       stream_base_state_(isolate_, StreamBase::kNumStreamBaseStateFields),
+      flags_(flags),
+      thread_id_(next_thread_id++),
       fs_stats_field_array_(isolate_, kFsStatsBufferLength),
       fs_stats_field_bigint_array_(isolate_, kFsStatsBufferLength),
       context_(context->GetIsolate(), context) {

src/env.h

@@ -593,6 +593,11 @@ class Environment {
     DISALLOW_COPY_AND_ASSIGN(TickInfo);
   };
 
+  enum Flags {
+    kNoFlags = 0,
+    kIsMainThread = 1
+  };
+
   static inline Environment* GetCurrent(v8::Isolate* isolate);
   static inline Environment* GetCurrent(v8::Local<v8::Context> context);
   static inline Environment* GetCurrent(
@@ -606,7 +611,8 @@ class Environment {
   static inline Environment* GetThreadLocalEnv();
 
   Environment(IsolateData* isolate_data,
-              v8::Local<v8::Context> context);
+              v8::Local<v8::Context> context,
+              Flags flags = Flags());
   ~Environment();
 
   void Start(bool start_profiler_idle_notifier);
@@ -759,7 +765,6 @@ class Environment {
 
   inline bool is_main_thread() const;
   inline uint64_t thread_id() const;
-  inline void set_thread_id(uint64_t id);
   inline worker::Worker* worker_context() const;
   inline void set_worker_context(worker::Worker* context);
   inline void add_sub_worker_context(worker::Worker* context);
@@ -1003,7 +1008,8 @@ class Environment {
 
   bool has_run_bootstrapping_code_ = false;
   bool can_call_into_js_ = true;
-  uint64_t thread_id_ = 0;
+  Flags flags_;
+  uint64_t thread_id_;
   std::unordered_set<worker::Worker*> sub_worker_contexts_;
 
   static void* const kNodeContextTagPtr;

src/node.cc

@@ -743,7 +743,7 @@ inline int Start(Isolate* isolate, IsolateData* isolate_data,
   HandleScope handle_scope(isolate);
   Local<Context> context = NewContext(isolate);
   Context::Scope context_scope(context);
-  Environment env(isolate_data, context);
+  Environment env(isolate_data, context, Environment::kIsMainThread);
   env.Start(per_process::v8_is_profiling);
   env.ProcessCliArgs(args, exec_args);
 

src/node_worker.cc

@@ -13,6 +13,7 @@
 
 using node::options_parser::kDisallowedInEnvironment;
 using v8::ArrayBuffer;
+using v8::Boolean;
 using v8::Context;
 using v8::Function;
 using v8::FunctionCallbackInfo;
@@ -33,9 +34,6 @@ namespace worker {
 
 namespace {
 
-uint64_t next_thread_id = 1;
-Mutex next_thread_id_mutex;
-
 #if NODE_USE_V8_PLATFORM && HAVE_INSPECTOR
 void StartWorkerInspector(Environment* child, const std::string& url) {
   child->inspector_agent()->Start(url,
@@ -74,17 +72,7 @@ Worker::Worker(Environment* env,
                const std::string& url,
                std::shared_ptr<PerIsolateOptions> per_isolate_opts)
     : AsyncWrap(env, wrap, AsyncWrap::PROVIDER_WORKER), url_(url) {
-  // Generate a new thread id.
-  {
-    Mutex::ScopedLock next_thread_id_lock(next_thread_id_mutex);
-    thread_id_ = next_thread_id++;
-  }
-
-  Debug(this, "Creating worker with id %llu", thread_id_);
-  wrap->Set(env->context(),
-            env->thread_id_string(),
-            Number::New(env->isolate(),
-                        static_cast<double>(thread_id_))).FromJust();
+  Debug(this, "Creating new worker instance at %p", static_cast<void*>(this));
 
   // Set up everything that needs to be set up in the parent environment.
   parent_port_ = MessagePort::New(env, env->context());
@@ -130,7 +118,7 @@ Worker::Worker(Environment* env,
     CHECK_NE(env_, nullptr);
     env_->set_abort_on_uncaught_exception(false);
     env_->set_worker_context(this);
-    env_->set_thread_id(thread_id_);
+    thread_id_ = env_->thread_id();
 
     env_->Start(env->profiler_idle_notifier_started());
     env_->ProcessCliArgs(std::vector<std::string>{},
@@ -142,7 +130,15 @@ Worker::Worker(Environment* env,
   // The new isolate won't be bothered on this thread again.
   isolate_->DiscardThreadSpecificMetadata();
 
-  Debug(this, "Set up worker with id %llu", thread_id_);
+  wrap->Set(env->context(),
+            env->thread_id_string(),
+            Number::New(env->isolate(), static_cast<double>(thread_id_)))
+      .FromJust();
+
+  Debug(this,
+        "Set up worker at %p with id %llu",
+        static_cast<void*>(this),
+        thread_id_);
 }
 
 bool Worker::is_stopped() const {
@@ -562,11 +558,17 @@ void InitWorker(Local<Object> target,
 
   env->SetMethod(target, "getEnvMessagePort", GetEnvMessagePort);
 
-  auto thread_id_string = FIXED_ONE_BYTE_STRING(env->isolate(), "threadId");
-  target->Set(env->context(),
-              thread_id_string,
-              Number::New(env->isolate(),
-                          static_cast<double>(env->thread_id()))).FromJust();
+  target
+      ->Set(env->context(),
+            env->thread_id_string(),
+            Number::New(env->isolate(), static_cast<double>(env->thread_id())))
+      .FromJust();
+
+  target
+      ->Set(env->context(),
+            FIXED_ONE_BYTE_STRING(env->isolate(), "isMainThread"),
+            Boolean::New(env->isolate(), env->is_main_thread()))
+      .FromJust();
 }
 
 }  // anonymous namespace