Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

src: set up process.nextTick and promise handlers with internalBinding('task_queue') #25163

Closed
wants to merge 3 commits into from
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions lib/internal/bootstrap/node.js
Original file line number Diff line number Diff line change
@@ -14,13 +14,9 @@

// This file is compiled as if it's wrapped in a function with arguments
// passed by node::LoadEnvironment()
/* global process, bootstrappers, loaderExports, triggerFatalException */
/* global process, loaderExports, triggerFatalException */
/* global isMainThread */

const {
_setupNextTick,
_setupPromises
} = bootstrappers;
const { internalBinding, NativeModule } = loaderExports;

const exceptionHandlerState = { captureFn: null };
@@ -105,8 +101,18 @@ function startup() {
}

NativeModule.require('internal/process/warning').setup();
NativeModule.require('internal/process/next_tick').setup(_setupNextTick,
_setupPromises);
const {
nextTick,
runNextTicks
} = NativeModule.require('internal/process/next_tick').setup();

process.nextTick = nextTick;
// Used to emulate a tick manually in the JS land.
// A better name for this function would be `runNextTicks` but
// it has been exposed to the process object so we keep this legacy name
// TODO(joyeecheung): either remove it or make it public
process._tickCallback = runNextTicks;

const credentials = internalBinding('credentials');
if (credentials.implementsPosixCredentials) {
process.getuid = credentials.getuid;
48 changes: 29 additions & 19 deletions lib/internal/process/next_tick.js
Original file line number Diff line number Diff line change
@@ -1,8 +1,20 @@
'use strict';

exports.setup = setupNextTick;
const {
// For easy access to the nextTick state in the C++ land,
// and to avoid unnecessary calls into JS land.
tickInfo,
// Used to run V8's micro task queue.
runMicrotasks,
setTickCallback,
initializePromiseRejectCallback
} = internalBinding('task_queue');

const {
promiseRejectHandler,
emitPromiseRejectionWarnings
} = require('internal/process/promises');

function setupNextTick(_setupNextTick, _setupPromises) {
const {
getDefaultTriggerAsyncId,
newAsyncId,
@@ -14,31 +26,16 @@ function setupNextTick(_setupNextTick, _setupPromises) {
emitDestroy,
symbols: { async_id_symbol, trigger_async_id_symbol }
} = require('internal/async_hooks');
const emitPromiseRejectionWarnings =
require('internal/process/promises').setup(_setupPromises);
const { ERR_INVALID_CALLBACK } = require('internal/errors').codes;
const FixedQueue = require('internal/fixed_queue');

// tickInfo is used so that the C++ code in src/node.cc can
// have easy access to our nextTick state, and avoid unnecessary
// calls into JS land.
// runMicrotasks is used to run V8's micro task queue.
const [
tickInfo,
runMicrotasks
] = _setupNextTick(internalTickCallback);

// *Must* match Environment::TickInfo::Fields in src/env.h.
const kHasScheduled = 0;
const kHasPromiseRejections = 1;

const queue = new FixedQueue();

process.nextTick = nextTick;
// Needs to be accessible from beyond this scope.
process._tickCallback = _tickCallback;

function _tickCallback() {
function runNextTicks() {
if (tickInfo[kHasScheduled] === 0 && tickInfo[kHasPromiseRejections] === 0)
runMicrotasks();
if (tickInfo[kHasScheduled] === 0 && tickInfo[kHasPromiseRejections] === 0)
@@ -125,4 +122,17 @@ function setupNextTick(_setupNextTick, _setupPromises) {
tickInfo[kHasScheduled] = 1;
queue.push(new TickObject(callback, args, getDefaultTriggerAsyncId()));
}
}

// TODO(joyeecheung): make this a factory class so that node.js can
// control the side effects caused by the initializers.
exports.setup = function() {
// Initializes the per-isolate promise rejection callback which
// will call the handler being passed into this function.
initializePromiseRejectCallback(promiseRejectHandler);
// Sets the callback to be run in every tick.
setTickCallback(internalTickCallback);
return {
nextTick,
runNextTicks
};
};
18 changes: 9 additions & 9 deletions lib/internal/process/promises.js
Original file line number Diff line number Diff line change
@@ -1,21 +1,16 @@
'use strict';

const { safeToString } = internalBinding('util');
const {
promiseRejectEvents
} = internalBinding('task_queue');

const maybeUnhandledPromises = new WeakMap();
const pendingUnhandledRejections = [];
const asyncHandledRejections = [];
const promiseRejectEvents = {};
let lastPromiseId = 0;

exports.setup = setupPromises;

function setupPromises(_setupPromises) {
_setupPromises(handler, promiseRejectEvents);
return emitPromiseRejectionWarnings;
}

function handler(type, promise, reason) {
function promiseRejectHandler(type, promise, reason) {
switch (type) {
case promiseRejectEvents.kPromiseRejectWithNoHandler:
return unhandledRejection(promise, reason);
@@ -124,3 +119,8 @@ function emitPromiseRejectionWarnings() {
}
return maybeScheduledTicks || pendingUnhandledRejections.length !== 0;
}

module.exports = {
promiseRejectHandler,
emitPromiseRejectionWarnings
};
3 changes: 2 additions & 1 deletion node.gyp
Original file line number Diff line number Diff line change
@@ -325,7 +325,6 @@

'sources': [
'src/async_wrap.cc',
'src/bootstrapper.cc',
'src/callback_scope.cc',
'src/cares_wrap.cc',
'src/connect_wrap.cc',
@@ -371,6 +370,8 @@
'src/node_process.cc',
'src/node_serdes.cc',
'src/node_stat_watcher.cc',
'src/node_symbols.cc',
'src/node_task_queue.cc',
'src/node_trace_events.cc',
'src/node_types.cc',
'src/node_url.cc',
4 changes: 2 additions & 2 deletions src/env.h
Original file line number Diff line number Diff line change
@@ -358,7 +358,7 @@ constexpr size_t kFsStatsBufferLength = kFsStatsFieldsNumber * 2;
V(performance_entry_template, v8::Function) \
V(pipe_constructor_template, v8::FunctionTemplate) \
V(process_object, v8::Object) \
V(promise_handler_function, v8::Function) \
V(promise_reject_callback, v8::Function) \
V(promise_wrap_template, v8::ObjectTemplate) \
V(sab_lifetimepartner_constructor_template, v8::FunctionTemplate) \
V(script_context_constructor_template, v8::FunctionTemplate) \
@@ -373,7 +373,7 @@ constexpr size_t kFsStatsBufferLength = kFsStatsFieldsNumber * 2;
V(tty_constructor_template, v8::FunctionTemplate) \
V(udp_constructor_function, v8::Function) \
V(url_constructor_function, v8::Function) \
V(write_wrap_template, v8::ObjectTemplate) \
V(write_wrap_template, v8::ObjectTemplate)

class Environment;

6 changes: 0 additions & 6 deletions src/node.cc
Original file line number Diff line number Diff line change
@@ -1200,20 +1200,14 @@ void LoadEnvironment(Environment* env) {
return;
}

// Bootstrap Node.js
Local<Object> bootstrapper = Object::New(env->isolate());
SetupBootstrapObject(env, bootstrapper);

// process, bootstrappers, loaderExports, triggerFatalException
std::vector<Local<String>> node_params = {
env->process_string(),
FIXED_ONE_BYTE_STRING(isolate, "bootstrappers"),
FIXED_ONE_BYTE_STRING(isolate, "loaderExports"),
FIXED_ONE_BYTE_STRING(isolate, "triggerFatalException"),
FIXED_ONE_BYTE_STRING(isolate, "isMainThread")};
std::vector<Local<Value>> node_args = {
process,
bootstrapper,
loader_exports.ToLocalChecked(),
env->NewFunctionTemplate(FatalException)
->GetFunction(context)
1 change: 1 addition & 0 deletions src/node_binding.cc
Original file line number Diff line number Diff line change
@@ -52,6 +52,7 @@
V(stream_wrap) \
V(string_decoder) \
V(symbols) \
V(task_queue) \
V(tcp_wrap) \
V(timers) \
V(trace_events) \
2 changes: 0 additions & 2 deletions src/node_internals.h
Original file line number Diff line number Diff line change
@@ -185,8 +185,6 @@ v8::Maybe<bool> ProcessEmitDeprecationWarning(Environment* env,
const char* warning,
const char* deprecation_code);

void SetupBootstrapObject(Environment* env,
v8::Local<v8::Object> bootstrapper);
void SetupProcessObject(Environment* env,
const std::vector<std::string>& args,
const std::vector<std::string>& exec_args);
29 changes: 29 additions & 0 deletions src/node_symbols.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
#include "env-inl.h"
#include "node_binding.h"

namespace node {

using v8::Context;
using v8::Local;
using v8::Object;
using v8::Value;

namespace symbols {

static void Initialize(Local<Object> target,
Local<Value> unused,
Local<Context> context,
void* priv) {
Environment* env = Environment::GetCurrent(context);
#define V(PropertyName, StringValue) \
target \
->Set(env->context(), env->PropertyName()->Name(), env->PropertyName()) \
.FromJust();
PER_ISOLATE_SYMBOL_PROPERTIES(V)
#undef V
}

} // namespace symbols
} // namespace node

NODE_MODULE_CONTEXT_AWARE_INTERNAL(symbols, node::symbols::Initialize)
90 changes: 35 additions & 55 deletions src/bootstrapper.cc → src/node_task_queue.cc
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
#include "node.h"
#include "env-inl.h"
#include "node.h"
#include "node_internals.h"
#include "v8.h"

@@ -23,36 +23,21 @@ using v8::Object;
using v8::Promise;
using v8::PromiseRejectEvent;
using v8::PromiseRejectMessage;
using v8::String;
using v8::Value;

void RunMicrotasks(const FunctionCallbackInfo<Value>& args) {
namespace task_queue {

static void RunMicrotasks(const FunctionCallbackInfo<Value>& args) {
args.GetIsolate()->RunMicrotasks();
}

void SetupNextTick(const FunctionCallbackInfo<Value>& args) {
static void SetTickCallback(const FunctionCallbackInfo<Value>& args) {
Environment* env = Environment::GetCurrent(args);
Isolate* isolate = env->isolate();
Local<Context> context = env->context();

CHECK(args[0]->IsFunction());

env->set_tick_callback_function(args[0].As<Function>());

Local<Function> run_microtasks_fn =
env->NewFunctionTemplate(RunMicrotasks)->GetFunction(context)
.ToLocalChecked();
run_microtasks_fn->SetName(FIXED_ONE_BYTE_STRING(isolate, "runMicrotasks"));

Local<Value> ret[] = {
env->tick_info()->fields().GetJSArray(),
run_microtasks_fn
};

args.GetReturnValue().Set(Array::New(isolate, ret, arraysize(ret)));
}

void PromiseRejectCallback(PromiseRejectMessage message) {
static void PromiseRejectCallback(PromiseRejectMessage message) {
static std::atomic<uint64_t> unhandledRejections{0};
static std::atomic<uint64_t> rejectionsHandledAfter{0};

@@ -64,7 +49,7 @@ void PromiseRejectCallback(PromiseRejectMessage message) {

if (env == nullptr) return;

Local<Function> callback = env->promise_handler_function();
Local<Function> callback = env->promise_reject_callback();
Local<Value> value;
Local<Value> type = Number::New(env->isolate(), event);

@@ -104,53 +89,48 @@ void PromiseRejectCallback(PromiseRejectMessage message) {
env->tick_info()->promise_rejections_toggle_on();
}

void SetupPromises(const FunctionCallbackInfo<Value>& args) {
static void InitializePromiseRejectCallback(
const FunctionCallbackInfo<Value>& args) {
Environment* env = Environment::GetCurrent(args);
Isolate* isolate = env->isolate();

CHECK(args[0]->IsFunction());
CHECK(args[1]->IsObject());

Local<Object> constants = args[1].As<Object>();

NODE_DEFINE_CONSTANT(constants, kPromiseRejectWithNoHandler);
NODE_DEFINE_CONSTANT(constants, kPromiseHandlerAddedAfterReject);
NODE_DEFINE_CONSTANT(constants, kPromiseResolveAfterResolved);
NODE_DEFINE_CONSTANT(constants, kPromiseRejectAfterResolved);

// TODO(joyeecheung): this may be moved to somewhere earlier in the bootstrap
// to make sure it's only called once
isolate->SetPromiseRejectCallback(PromiseRejectCallback);
env->set_promise_handler_function(args[0].As<Function>());
}

#define BOOTSTRAP_METHOD(name, fn) env->SetMethod(bootstrapper, #name, fn)

// The Bootstrapper object is an ephemeral object that is used only during
// the bootstrap process of the Node.js environment. A reference to the
// bootstrap object must not be kept around after the bootstrap process
// completes so that it can be gc'd as soon as possible.
void SetupBootstrapObject(Environment* env,
Local<Object> bootstrapper) {
BOOTSTRAP_METHOD(_setupNextTick, SetupNextTick);
BOOTSTRAP_METHOD(_setupPromises, SetupPromises);
env->set_promise_reject_callback(args[0].As<Function>());
}
#undef BOOTSTRAP_METHOD

namespace symbols {

void Initialize(Local<Object> target,
static void Initialize(Local<Object> target,
Local<Value> unused,
Local<Context> context,
void* priv) {
Environment* env = Environment::GetCurrent(context);
#define V(PropertyName, StringValue) \
target->Set(env->context(), \
env->PropertyName()->Name(), \
env->PropertyName()).FromJust();
PER_ISOLATE_SYMBOL_PROPERTIES(V)
#undef V
Isolate* isolate = env->isolate();

env->SetMethod(target, "setTickCallback", SetTickCallback);
env->SetMethod(target, "runMicrotasks", RunMicrotasks);
target->Set(env->context(),
FIXED_ONE_BYTE_STRING(isolate, "tickInfo"),
env->tick_info()->fields().GetJSArray()).FromJust();

Local<Object> events = Object::New(isolate);
NODE_DEFINE_CONSTANT(events, kPromiseRejectWithNoHandler);
NODE_DEFINE_CONSTANT(events, kPromiseHandlerAddedAfterReject);
NODE_DEFINE_CONSTANT(events, kPromiseResolveAfterResolved);
NODE_DEFINE_CONSTANT(events, kPromiseRejectAfterResolved);

target->Set(env->context(),
FIXED_ONE_BYTE_STRING(isolate, "promiseRejectEvents"),
events).FromJust();
env->SetMethod(target,
"initializePromiseRejectCallback",
InitializePromiseRejectCallback);
}

} // namespace symbols
} // namespace task_queue
} // namespace node

NODE_MODULE_CONTEXT_AWARE_INTERNAL(symbols, node::symbols::Initialize)
NODE_MODULE_CONTEXT_AWARE_INTERNAL(task_queue, node::task_queue::Initialize)
2 changes: 1 addition & 1 deletion test/message/events_unhandled_error_nexttick.out
Original file line number Diff line number Diff line change
@@ -16,7 +16,7 @@ Error
Emitted 'error' event at:
at process.nextTick (*events_unhandled_error_nexttick.js:*:*)
at internalTickCallback (internal/process/next_tick.js:*:*)
at process._tickCallback (internal/process/next_tick.js:*:*)
at process.runNextTicks [as _tickCallback] (internal/process/next_tick.js:*:*)
at Function.Module.runMain (internal/modules/cjs/loader.js:*:*)
at executeUserCode (internal/bootstrap/node.js:*:*)
at startExecution (internal/bootstrap/node.js:*:*)
2 changes: 1 addition & 1 deletion test/message/nexttick_throw.out
Original file line number Diff line number Diff line change
@@ -5,7 +5,7 @@
ReferenceError: undefined_reference_error_maker is not defined
at *test*message*nexttick_throw.js:*:*
at internalTickCallback (internal/process/next_tick.js:*:*)
at process._tickCallback (internal/process/next_tick.js:*:*)
at process.runNextTicks [as _tickCallback] (internal/process/next_tick.js:*:*)
at Function.Module.runMain (internal/modules/cjs/loader.js:*:*)
at executeUserCode (internal/bootstrap/node.js:*:*)
at startExecution (internal/bootstrap/node.js:*:*)
2 changes: 1 addition & 1 deletion test/message/unhandled_promise_trace_warnings.out
Original file line number Diff line number Diff line change
@@ -42,7 +42,7 @@
at *
(node:*) PromiseRejectionHandledWarning: Promise rejection was handled asynchronously (rejection id: 1)
at handledRejection (internal/process/promises.js:*)
at handler (internal/process/promises.js:*)
at promiseRejectHandler (internal/process/promises.js:*)
at Promise.then *
at Promise.catch *
at Immediate.setImmediate (*test*message*unhandled_promise_trace_warnings.js:*)
2 changes: 1 addition & 1 deletion test/parallel/test-bootstrap-modules.js
Original file line number Diff line number Diff line change
@@ -9,7 +9,7 @@ const common = require('../common');
const assert = require('assert');

const isMainThread = common.isMainThread;
const kMaxModuleCount = isMainThread ? 61 : 83;
const kMaxModuleCount = isMainThread ? 62 : 84;

assert(list.length <= kMaxModuleCount,
`Total length: ${list.length}\n` + list.join('\n')