From 3e5803f91b9856a20da2c6bee1a88ba092dece55 Mon Sep 17 00:00:00 2001 From: Anna Henningsen Date: Sat, 1 Feb 2020 17:16:23 +0100 Subject: [PATCH] worker: improve MessagePort performance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Use a JS function as the single entry point for emitting `.onmessage()` calls, avoiding the overhead of manually constructing each message event object in C++. confidence improvement accuracy (*) (**) (***) worker/echo.js n=100000 sendsPerBroadcast=1 payload='object' workers=1 *** 16.34 % ±1.16% ±1.54% ±1.99% worker/echo.js n=100000 sendsPerBroadcast=1 payload='string' workers=1 *** 24.41 % ±1.50% ±1.99% ±2.58% worker/echo.js n=100000 sendsPerBroadcast=10 payload='object' workers=1 *** 26.66 % ±1.54% ±2.05% ±2.65% worker/echo.js n=100000 sendsPerBroadcast=10 payload='string' workers=1 *** 32.72 % ±1.60% ±2.11% ±2.73% worker/messageport.js n=1000000 payload='object' *** 40.28 % ±1.48% ±1.95% ±2.52% worker/messageport.js n=1000000 payload='string' *** 76.95 % ±2.19% ±2.90% ±3.75% Also fix handling exceptions returned from `MessagePort::New`. PR-URL: https://github.com/nodejs/node/pull/31605 Reviewed-By: James M Snell Reviewed-By: Denys Otrishko Reviewed-By: Rich Trott --- lib/internal/per_context/messageport.js | 12 +++++ node.gyp | 1 + src/api/environment.cc | 1 + src/env.h | 1 - src/node_messaging.cc | 63 ++++++++++++++++--------- src/node_messaging.h | 7 ++- 6 files changed, 61 insertions(+), 24 deletions(-) create mode 100644 lib/internal/per_context/messageport.js diff --git a/lib/internal/per_context/messageport.js b/lib/internal/per_context/messageport.js new file mode 100644 index 00000000000000..ee08a596122a51 --- /dev/null +++ b/lib/internal/per_context/messageport.js @@ -0,0 +1,12 @@ +'use strict'; +class MessageEvent { + constructor(data, target) { + this.data = data; + this.target = target; + } +} + +exports.emitMessage = function(data) { + if (typeof this.onmessage === 'function') + this.onmessage(new MessageEvent(data, this)); +}; diff --git a/node.gyp b/node.gyp index 3c6a85737078c0..8c7911732b944c 100644 --- a/node.gyp +++ b/node.gyp @@ -37,6 +37,7 @@ 'lib/internal/bootstrap/switches/is_not_main_thread.js', 'lib/internal/per_context/primordials.js', 'lib/internal/per_context/domexception.js', + 'lib/internal/per_context/messageport.js', 'lib/async_hooks.js', 'lib/assert.js', 'lib/buffer.js', diff --git a/src/api/environment.cc b/src/api/environment.cc index 39e9e48ced219a..f25a4cf183292f 100644 --- a/src/api/environment.cc +++ b/src/api/environment.cc @@ -488,6 +488,7 @@ bool InitializeContextForSnapshot(Local context) { static const char* context_files[] = {"internal/per_context/primordials", "internal/per_context/domexception", + "internal/per_context/messageport", nullptr}; for (const char** module = context_files; *module != nullptr; module++) { diff --git a/src/env.h b/src/env.h index 64e677bcb2af5e..cfac8f72dc06a1 100644 --- a/src/env.h +++ b/src/env.h @@ -403,7 +403,6 @@ constexpr size_t kFsStatsBufferLength = V(http2stream_constructor_template, v8::ObjectTemplate) \ V(http2ping_constructor_template, v8::ObjectTemplate) \ V(libuv_stream_wrap_ctor_template, v8::FunctionTemplate) \ - V(message_event_object_template, v8::ObjectTemplate) \ V(message_port_constructor_template, v8::FunctionTemplate) \ V(pipe_constructor_template, v8::FunctionTemplate) \ V(promise_wrap_template, v8::ObjectTemplate) \ diff --git a/src/node_messaging.cc b/src/node_messaging.cc index 1fb3c2e4f45a05..f510706fa6ed93 100644 --- a/src/node_messaging.cc +++ b/src/node_messaging.cc @@ -28,7 +28,6 @@ using v8::Maybe; using v8::MaybeLocal; using v8::Nothing; using v8::Object; -using v8::ObjectTemplate; using v8::SharedArrayBuffer; using v8::String; using v8::Symbol; @@ -169,6 +168,20 @@ uint32_t Message::AddWASMModule(WasmModuleObject::TransferrableModule&& mod) { namespace { +MaybeLocal GetEmitMessageFunction(Local context) { + Isolate* isolate = context->GetIsolate(); + Local per_context_bindings; + Local emit_message_val; + if (!GetPerContextExports(context).ToLocal(&per_context_bindings) || + !per_context_bindings->Get(context, + FIXED_ONE_BYTE_STRING(isolate, "emitMessage")) + .ToLocal(&emit_message_val)) { + return MaybeLocal(); + } + CHECK(emit_message_val->IsFunction()); + return emit_message_val.As(); +} + MaybeLocal GetDOMException(Local context) { Isolate* isolate = context->GetIsolate(); Local per_context_bindings; @@ -470,10 +483,14 @@ MessagePort::MessagePort(Environment* env, MessagePort* channel = ContainerOf(&MessagePort::async_, handle); channel->OnMessage(); }; + CHECK_EQ(uv_async_init(env->event_loop(), &async_, onmessage), 0); - async_.data = static_cast(this); + async_.data = nullptr; // Reset later to indicate success of the constructor. + auto cleanup = OnScopeLeave([&]() { + if (async_.data == nullptr) Close(); + }); Local fn; if (!wrap->Get(context, env->oninit_symbol()).ToLocal(&fn)) @@ -481,9 +498,16 @@ MessagePort::MessagePort(Environment* env, if (fn->IsFunction()) { Local init = fn.As(); - USE(init->Call(context, wrap, 0, nullptr)); + if (init->Call(context, wrap, 0, nullptr).IsEmpty()) + return; } + Local emit_message_fn; + if (!GetEmitMessageFunction(context).ToLocal(&emit_message_fn)) + return; + emit_message_fn_.Reset(env->isolate(), emit_message_fn); + + async_.data = static_cast(this); Debug(this, "Created message port"); } @@ -531,6 +555,11 @@ MessagePort* MessagePort::New( return nullptr; MessagePort* port = new MessagePort(env, context, instance); CHECK_NOT_NULL(port); + if (port->IsHandleClosing()) { + // Construction failed with an exception. + return nullptr; + } + if (data) { port->Detach(); port->data_ = std::move(data); @@ -623,16 +652,8 @@ void MessagePort::OnMessage() { continue; } - Local event; - Local cb_args[1]; - if (!env()->message_event_object_template()->NewInstance(context) - .ToLocal(&event) || - event->Set(context, env()->data_string(), payload).IsNothing() || - event->Set(context, env()->target_string(), object()).IsNothing() || - (cb_args[0] = event, false) || - MakeCallback(env()->onmessage_string(), - arraysize(cb_args), - cb_args).IsEmpty()) { + Local emit_message = PersistentToLocal::Strong(emit_message_fn_); + if (MakeCallback(emit_message, 1, &payload).IsEmpty()) { // Re-schedule OnMessage() execution in case of failure. if (data_) TriggerAsync(); @@ -901,6 +922,7 @@ void MessagePort::Entangle(MessagePort* a, MessagePortData* b) { void MessagePort::MemoryInfo(MemoryTracker* tracker) const { tracker->TrackField("data", data_); + tracker->TrackField("emit_message_fn", emit_message_fn_); } Local GetMessagePortConstructorTemplate(Environment* env) { @@ -910,8 +932,6 @@ Local GetMessagePortConstructorTemplate(Environment* env) { if (!templ.IsEmpty()) return templ; - Isolate* isolate = env->isolate(); - { Local m = env->NewFunctionTemplate(MessagePort::New); m->SetClassName(env->message_port_constructor_string()); @@ -922,13 +942,6 @@ Local GetMessagePortConstructorTemplate(Environment* env) { env->SetProtoMethod(m, "start", MessagePort::Start); env->set_message_port_constructor_template(m); - - Local event_ctor = FunctionTemplate::New(isolate); - event_ctor->SetClassName(FIXED_ONE_BYTE_STRING(isolate, "MessageEvent")); - Local e = event_ctor->InstanceTemplate(); - e->Set(env->data_string(), Null(isolate)); - e->Set(env->target_string(), Null(isolate)); - env->set_message_event_object_template(e); } return GetMessagePortConstructorTemplate(env); @@ -947,7 +960,13 @@ static void MessageChannel(const FunctionCallbackInfo& args) { Context::Scope context_scope(context); MessagePort* port1 = MessagePort::New(env, context); + if (port1 == nullptr) return; MessagePort* port2 = MessagePort::New(env, context); + if (port2 == nullptr) { + port1->Close(); + return; + } + MessagePort::Entangle(port1, port2); args.This()->Set(context, env->port1_string(), port1->object()) diff --git a/src/node_messaging.h b/src/node_messaging.h index c10264bb3495ff..1f9901dd2afc0f 100644 --- a/src/node_messaging.h +++ b/src/node_messaging.h @@ -131,12 +131,16 @@ class MessagePortData : public MemoryRetainer { // the uv_async_t handle that is used to notify the current event loop of // new incoming messages. class MessagePort : public HandleWrap { - public: + private: // Create a new MessagePort. The `context` argument specifies the Context // instance that is used for creating the values emitted from this port. + // This is called by MessagePort::New(), which is the public API used for + // creating MessagePort instances. MessagePort(Environment* env, v8::Local context, v8::Local wrap); + + public: ~MessagePort() override; // Create a new message port instance, optionally over an existing @@ -205,6 +209,7 @@ class MessagePort : public HandleWrap { std::unique_ptr data_ = nullptr; bool receiving_messages_ = false; uv_async_t async_; + v8::Global emit_message_fn_; friend class MessagePortData; };