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: https://github.com/nodejs/node/pull/25796 Reviewed-By: Joyee Cheung <joyeec9h3@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: Denys Otrishko <shishugi@gmail.com>
This commit is contained in:
parent
bcf2886a84
commit
393c196605
@ -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');
|
||||
|
@ -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;
|
||||
|
@ -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_;
|
||||
}
|
||||
|
@ -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) {
|
||||
|
12
src/env.h
12
src/env.h
@ -595,6 +595,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(
|
||||
@ -608,7 +613,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);
|
||||
@ -761,7 +767,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);
|
||||
@ -1005,7 +1010,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;
|
||||
|
@ -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);
|
||||
|
||||
|
@ -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
|
||||
|
Loading…
x
Reference in New Issue
Block a user