#include "node_messaging.h" #include "async_wrap-inl.h" #include "debug_utils-inl.h" #include "memory_tracker-inl.h" #include "node_buffer.h" #include "node_contextify.h" #include "node_errors.h" #include "node_external_reference.h" #include "node_process.h" #include "util-inl.h" using node::contextify::ContextifyContext; using node::errors::TryCatchScope; using v8::Array; using v8::ArrayBuffer; using v8::BackingStore; using v8::CompiledWasmModule; using v8::Context; using v8::EscapableHandleScope; using v8::Function; using v8::FunctionCallbackInfo; using v8::FunctionTemplate; using v8::Global; using v8::HandleScope; using v8::Isolate; using v8::Just; using v8::Local; using v8::Maybe; using v8::MaybeLocal; using v8::Nothing; using v8::Object; using v8::SharedArrayBuffer; using v8::String; using v8::Symbol; using v8::Value; using v8::ValueDeserializer; using v8::ValueSerializer; using v8::WasmModuleObject; namespace node { using BaseObjectList = std::vector>; BaseObject::TransferMode BaseObject::GetTransferMode() const { return BaseObject::TransferMode::kUntransferable; } std::unique_ptr BaseObject::TransferForMessaging() { return CloneForMessaging(); } std::unique_ptr BaseObject::CloneForMessaging() const { return {}; } Maybe BaseObject::NestedTransferables() const { return Just(BaseObjectList {}); } Maybe BaseObject::FinalizeTransferRead( Local context, ValueDeserializer* deserializer) { return Just(true); } namespace worker { Maybe TransferData::FinalizeTransferWrite( Local context, ValueSerializer* serializer) { return Just(true); } Message::Message(MallocedBuffer&& buffer) : main_message_buf_(std::move(buffer)) {} bool Message::IsCloseMessage() const { return main_message_buf_.data == nullptr; } namespace { // This is used to tell V8 how to read transferred host objects, like other // `MessagePort`s and `SharedArrayBuffer`s, and make new JS objects out of them. class DeserializerDelegate : public ValueDeserializer::Delegate { public: DeserializerDelegate( Message* m, Environment* env, const std::vector>& host_objects, const std::vector>& shared_array_buffers, const std::vector& wasm_modules) : host_objects_(host_objects), shared_array_buffers_(shared_array_buffers), wasm_modules_(wasm_modules) {} MaybeLocal ReadHostObject(Isolate* isolate) override { // Identifying the index in the message's BaseObject array is sufficient. uint32_t id; if (!deserializer->ReadUint32(&id)) return MaybeLocal(); CHECK_LE(id, host_objects_.size()); return host_objects_[id]->object(isolate); } MaybeLocal GetSharedArrayBufferFromId( Isolate* isolate, uint32_t clone_id) override { CHECK_LE(clone_id, shared_array_buffers_.size()); return shared_array_buffers_[clone_id]; } MaybeLocal GetWasmModuleFromId( Isolate* isolate, uint32_t transfer_id) override { CHECK_LE(transfer_id, wasm_modules_.size()); return WasmModuleObject::FromCompiledModule( isolate, wasm_modules_[transfer_id]); } ValueDeserializer* deserializer = nullptr; private: const std::vector>& host_objects_; const std::vector>& shared_array_buffers_; const std::vector& wasm_modules_; }; } // anonymous namespace MaybeLocal Message::Deserialize(Environment* env, Local context) { CHECK(!IsCloseMessage()); EscapableHandleScope handle_scope(env->isolate()); Context::Scope context_scope(context); // Create all necessary objects for transferables, e.g. MessagePort handles. std::vector> host_objects(transferables_.size()); auto cleanup = OnScopeLeave([&]() { for (BaseObjectPtr object : host_objects) { if (!object) continue; // If the function did not finish successfully, host_objects will contain // a list of objects that will never be passed to JS. Therefore, we // destroy them here. object->Detach(); } }); for (uint32_t i = 0; i < transferables_.size(); ++i) { TransferData* data = transferables_[i].get(); host_objects[i] = data->Deserialize( env, context, std::move(transferables_[i])); if (!host_objects[i]) return {}; } transferables_.clear(); std::vector> shared_array_buffers; // Attach all transferred SharedArrayBuffers to their new Isolate. for (uint32_t i = 0; i < shared_array_buffers_.size(); ++i) { Local sab = SharedArrayBuffer::New(env->isolate(), shared_array_buffers_[i]); shared_array_buffers.push_back(sab); } DeserializerDelegate delegate( this, env, host_objects, shared_array_buffers, wasm_modules_); ValueDeserializer deserializer( env->isolate(), reinterpret_cast(main_message_buf_.data), main_message_buf_.size, &delegate); delegate.deserializer = &deserializer; // Attach all transferred ArrayBuffers to their new Isolate. for (uint32_t i = 0; i < array_buffers_.size(); ++i) { Local ab = ArrayBuffer::New(env->isolate(), std::move(array_buffers_[i])); deserializer.TransferArrayBuffer(i, ab); } if (deserializer.ReadHeader(context).IsNothing()) return {}; Local return_value; if (!deserializer.ReadValue(context).ToLocal(&return_value)) return {}; for (BaseObjectPtr base_object : host_objects) { if (base_object->FinalizeTransferRead(context, &deserializer).IsNothing()) return {}; } host_objects.clear(); return handle_scope.Escape(return_value); } void Message::AddSharedArrayBuffer( std::shared_ptr backing_store) { shared_array_buffers_.emplace_back(std::move(backing_store)); } void Message::AddTransferable(std::unique_ptr&& data) { transferables_.emplace_back(std::move(data)); } uint32_t Message::AddWASMModule(CompiledWasmModule&& mod) { wasm_modules_.emplace_back(std::move(mod)); return wasm_modules_.size() - 1; } 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; Local domexception_ctor_val; if (!GetPerContextExports(context).ToLocal(&per_context_bindings) || !per_context_bindings->Get(context, FIXED_ONE_BYTE_STRING(isolate, "DOMException")) .ToLocal(&domexception_ctor_val)) { return MaybeLocal(); } CHECK(domexception_ctor_val->IsFunction()); Local domexception_ctor = domexception_ctor_val.As(); return domexception_ctor; } void ThrowDataCloneException(Local context, Local message) { Isolate* isolate = context->GetIsolate(); Local argv[] = {message, FIXED_ONE_BYTE_STRING(isolate, "DataCloneError")}; Local exception; Local domexception_ctor; if (!GetDOMException(context).ToLocal(&domexception_ctor) || !domexception_ctor->NewInstance(context, arraysize(argv), argv) .ToLocal(&exception)) { return; } isolate->ThrowException(exception); } // This tells V8 how to serialize objects that it does not understand // (e.g. C++ objects) into the output buffer, in a way that our own // DeserializerDelegate understands how to unpack. class SerializerDelegate : public ValueSerializer::Delegate { public: SerializerDelegate(Environment* env, Local context, Message* m) : env_(env), context_(context), msg_(m) {} void ThrowDataCloneError(Local message) override { ThrowDataCloneException(context_, message); } Maybe WriteHostObject(Isolate* isolate, Local object) override { if (env_->base_object_ctor_template()->HasInstance(object)) { return WriteHostObject( BaseObjectPtr { Unwrap(object) }); } ThrowDataCloneError(env_->clone_unsupported_type_str()); return Nothing(); } Maybe GetSharedArrayBufferId( Isolate* isolate, Local shared_array_buffer) override { uint32_t i; for (i = 0; i < seen_shared_array_buffers_.size(); ++i) { if (PersistentToLocal::Strong(seen_shared_array_buffers_[i]) == shared_array_buffer) { return Just(i); } } seen_shared_array_buffers_.emplace_back( Global { isolate, shared_array_buffer }); msg_->AddSharedArrayBuffer(shared_array_buffer->GetBackingStore()); return Just(i); } Maybe GetWasmModuleTransferId( Isolate* isolate, Local module) override { return Just(msg_->AddWASMModule(module->GetCompiledModule())); } Maybe Finish(Local context) { for (uint32_t i = 0; i < host_objects_.size(); i++) { BaseObjectPtr host_object = std::move(host_objects_[i]); std::unique_ptr data; if (i < first_cloned_object_index_) data = host_object->TransferForMessaging(); if (!data) data = host_object->CloneForMessaging(); if (!data) return Nothing(); if (data->FinalizeTransferWrite(context, serializer).IsNothing()) return Nothing(); msg_->AddTransferable(std::move(data)); } return Just(true); } inline void AddHostObject(BaseObjectPtr host_object) { // Make sure we have not started serializing the value itself yet. CHECK_EQ(first_cloned_object_index_, SIZE_MAX); host_objects_.emplace_back(std::move(host_object)); } // Some objects in the transfer list may register sub-objects that can be // transferred. This could e.g. be a public JS wrapper object, such as a // FileHandle, that is registering its C++ handle for transfer. inline Maybe AddNestedHostObjects() { for (size_t i = 0; i < host_objects_.size(); i++) { std::vector> nested_transferables; if (!host_objects_[i]->NestedTransferables().To(&nested_transferables)) return Nothing(); for (auto nested_transferable : nested_transferables) { if (std::find(host_objects_.begin(), host_objects_.end(), nested_transferable) == host_objects_.end()) { AddHostObject(nested_transferable); } } } return Just(true); } ValueSerializer* serializer = nullptr; private: Maybe WriteHostObject(BaseObjectPtr host_object) { BaseObject::TransferMode mode = host_object->GetTransferMode(); if (mode == BaseObject::TransferMode::kUntransferable) { ThrowDataCloneError(env_->clone_unsupported_type_str()); return Nothing(); } for (uint32_t i = 0; i < host_objects_.size(); i++) { if (host_objects_[i] == host_object) { serializer->WriteUint32(i); return Just(true); } } if (mode == BaseObject::TransferMode::kTransferable) { THROW_ERR_MISSING_TRANSFERABLE_IN_TRANSFER_LIST(env_); return Nothing(); } CHECK_EQ(mode, BaseObject::TransferMode::kCloneable); uint32_t index = host_objects_.size(); if (first_cloned_object_index_ == SIZE_MAX) first_cloned_object_index_ = index; serializer->WriteUint32(index); host_objects_.push_back(host_object); return Just(true); } Environment* env_; Local context_; Message* msg_; std::vector> seen_shared_array_buffers_; std::vector> host_objects_; size_t first_cloned_object_index_ = SIZE_MAX; friend class worker::Message; }; } // anonymous namespace Maybe Message::Serialize(Environment* env, Local context, Local input, const TransferList& transfer_list_v, Local source_port) { HandleScope handle_scope(env->isolate()); Context::Scope context_scope(context); // Verify that we're not silently overwriting an existing message. CHECK(main_message_buf_.is_empty()); SerializerDelegate delegate(env, context, this); ValueSerializer serializer(env->isolate(), &delegate); delegate.serializer = &serializer; std::vector> array_buffers; for (uint32_t i = 0; i < transfer_list_v.length(); ++i) { Local entry = transfer_list_v[i]; if (entry->IsObject()) { // See https://github.com/nodejs/node/pull/30339#issuecomment-552225353 // for details. bool untransferable; if (!entry.As()->HasPrivate( context, env->untransferable_object_private_symbol()) .To(&untransferable)) { return Nothing(); } if (untransferable) continue; } // Currently, we support ArrayBuffers and BaseObjects for which // GetTransferMode() does not return kUntransferable. if (entry->IsArrayBuffer()) { Local ab = entry.As(); // If we cannot render the ArrayBuffer unusable in this Isolate, // copying the buffer will have to do. // Note that we can currently transfer ArrayBuffers even if they were // not allocated by Node’s ArrayBufferAllocator in the first place, // because we pass the underlying v8::BackingStore around rather than // raw data *and* an Isolate with a non-default ArrayBuffer allocator // is always going to outlive any Workers it creates, and so will its // allocator along with it. if (!ab->IsDetachable()) continue; if (std::find(array_buffers.begin(), array_buffers.end(), ab) != array_buffers.end()) { ThrowDataCloneException( context, FIXED_ONE_BYTE_STRING( env->isolate(), "Transfer list contains duplicate ArrayBuffer")); return Nothing(); } // We simply use the array index in the `array_buffers` list as the // ID that we write into the serialized buffer. uint32_t id = array_buffers.size(); array_buffers.push_back(ab); serializer.TransferArrayBuffer(id, ab); continue; } else if (env->base_object_ctor_template()->HasInstance(entry)) { // Check if the source MessagePort is being transferred. if (!source_port.IsEmpty() && entry == source_port) { ThrowDataCloneException( context, FIXED_ONE_BYTE_STRING(env->isolate(), "Transfer list contains source port")); return Nothing(); } BaseObjectPtr host_object { Unwrap(entry.As()) }; if (env->message_port_constructor_template()->HasInstance(entry) && (!host_object || static_cast(host_object.get())->IsDetached())) { ThrowDataCloneException( context, FIXED_ONE_BYTE_STRING( env->isolate(), "MessagePort in transfer list is already detached")); return Nothing(); } if (std::find(delegate.host_objects_.begin(), delegate.host_objects_.end(), host_object) != delegate.host_objects_.end()) { ThrowDataCloneException( context, String::Concat(env->isolate(), FIXED_ONE_BYTE_STRING( env->isolate(), "Transfer list contains duplicate "), entry.As()->GetConstructorName())); return Nothing(); } if (host_object && host_object->GetTransferMode() != BaseObject::TransferMode::kUntransferable) { delegate.AddHostObject(host_object); continue; } } THROW_ERR_INVALID_TRANSFER_OBJECT(env); return Nothing(); } if (delegate.AddNestedHostObjects().IsNothing()) return Nothing(); serializer.WriteHeader(); if (serializer.WriteValue(context, input).IsNothing()) { return Nothing(); } for (Local ab : array_buffers) { // If serialization succeeded, we render it inaccessible in this Isolate. std::shared_ptr backing_store = ab->GetBackingStore(); ab->Detach(); array_buffers_.emplace_back(std::move(backing_store)); } if (delegate.Finish(context).IsNothing()) return Nothing(); // The serializer gave us a buffer allocated using `malloc()`. std::pair data = serializer.Release(); CHECK_NOT_NULL(data.first); main_message_buf_ = MallocedBuffer(reinterpret_cast(data.first), data.second); return Just(true); } void Message::MemoryInfo(MemoryTracker* tracker) const { tracker->TrackField("array_buffers_", array_buffers_); tracker->TrackField("shared_array_buffers", shared_array_buffers_); tracker->TrackField("transferables", transferables_); } MessagePortData::MessagePortData(MessagePort* owner) : owner_(owner) { } MessagePortData::~MessagePortData() { CHECK_NULL(owner_); Disentangle(); } void MessagePortData::MemoryInfo(MemoryTracker* tracker) const { Mutex::ScopedLock lock(mutex_); tracker->TrackField("incoming_messages", incoming_messages_); } void MessagePortData::AddToIncomingQueue(std::shared_ptr message) { // This function will be called by other threads. Mutex::ScopedLock lock(mutex_); incoming_messages_.emplace_back(std::move(message)); if (owner_ != nullptr) { Debug(owner_, "Adding message to incoming queue"); owner_->TriggerAsync(); } } void MessagePortData::Entangle(MessagePortData* a, MessagePortData* b) { auto group = std::make_shared(); group->Entangle({a, b}); } void MessagePortData::Disentangle() { if (group_) { group_->Disentangle(this); } } MessagePort::~MessagePort() { if (data_) Detach(); } MessagePort::MessagePort(Environment* env, Local context, Local wrap) : HandleWrap(env, wrap, reinterpret_cast(&async_), AsyncWrap::PROVIDER_MESSAGEPORT), data_(new MessagePortData(this)) { auto onmessage = [](uv_async_t* handle) { // Called when data has been put into the queue. MessagePort* channel = ContainerOf(&MessagePort::async_, handle); channel->OnMessage(); }; CHECK_EQ(uv_async_init(env->event_loop(), &async_, onmessage), 0); // Reset later to indicate success of the constructor. bool succeeded = false; auto cleanup = OnScopeLeave([&]() { if (!succeeded) Close(); }); Local fn; if (!wrap->Get(context, env->oninit_symbol()).ToLocal(&fn)) return; if (fn->IsFunction()) { Local init = fn.As(); 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); succeeded = true; Debug(this, "Created message port"); } bool MessagePort::IsDetached() const { return data_ == nullptr || IsHandleClosing(); } void MessagePort::TriggerAsync() { if (IsHandleClosing()) return; CHECK_EQ(uv_async_send(&async_), 0); } void MessagePort::Close(v8::Local close_callback) { Debug(this, "Closing message port, data set = %d", static_cast(!!data_)); if (data_) { // Wrap this call with accessing the mutex, so that TriggerAsync() // can check IsHandleClosing() without race conditions. Mutex::ScopedLock sibling_lock(data_->mutex_); HandleWrap::Close(close_callback); } else { HandleWrap::Close(close_callback); } } void MessagePort::New(const FunctionCallbackInfo& args) { // This constructor just throws an error. Unfortunately, we can’t use V8’s // ConstructorBehavior::kThrow, as that also removes the prototype from the // class (i.e. makes it behave like an arrow function). Environment* env = Environment::GetCurrent(args); THROW_ERR_CONSTRUCT_CALL_INVALID(env); } MessagePort* MessagePort::New( Environment* env, Local context, std::unique_ptr data, std::shared_ptr sibling_group) { Context::Scope context_scope(context); Local ctor_templ = GetMessagePortConstructorTemplate(env); // Construct a new instance, then assign the listener instance and possibly // the MessagePortData to it. Local instance; if (!ctor_templ->InstanceTemplate()->NewInstance(context).ToLocal(&instance)) 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) { CHECK(!sibling_group); port->Detach(); port->data_ = std::move(data); // This lock is here to avoid race conditions with the `owner_` read // in AddToIncomingQueue(). (This would likely be unproblematic without it, // but it's better to be safe than sorry.) Mutex::ScopedLock lock(port->data_->mutex_); port->data_->owner_ = port; // If the existing MessagePortData object had pending messages, this is // the easiest way to run that queue. port->TriggerAsync(); } else if (sibling_group) { sibling_group->Entangle(port->data_.get()); } return port; } MaybeLocal MessagePort::ReceiveMessage(Local context, bool only_if_receiving) { std::shared_ptr received; { // Get the head of the message queue. Mutex::ScopedLock lock(data_->mutex_); Debug(this, "MessagePort has message"); bool wants_message = receiving_messages_ || !only_if_receiving; // We have nothing to do if: // - There are no pending messages // - We are not intending to receive messages, and the message we would // receive is not the final "close" message. if (data_->incoming_messages_.empty() || (!wants_message && !data_->incoming_messages_.front()->IsCloseMessage())) { return env()->no_message_symbol(); } received = data_->incoming_messages_.front(); data_->incoming_messages_.pop_front(); } if (received->IsCloseMessage()) { Close(); return env()->no_message_symbol(); } if (!env()->can_call_into_js()) return MaybeLocal(); return received->Deserialize(env(), context); } void MessagePort::OnMessage() { Debug(this, "Running MessagePort::OnMessage()"); HandleScope handle_scope(env()->isolate()); Local context = object(env()->isolate())->CreationContext(); size_t processing_limit; { Mutex::ScopedLock(data_->mutex_); processing_limit = std::max(data_->incoming_messages_.size(), static_cast(1000)); } // data_ can only ever be modified by the owner thread, so no need to lock. // However, the message port may be transferred while it is processing // messages, so we need to check that this handle still owns its `data_` field // on every iteration. while (data_) { if (processing_limit-- == 0) { // Prevent event loop starvation by only processing those messages without // interruption that were already present when the OnMessage() call was // first triggered, but at least 1000 messages because otherwise the // overhead of repeatedly triggering the uv_async_t instance becomes // noticable, at least on Windows. // (That might require more investigation by somebody more familiar with // Windows.) TriggerAsync(); return; } HandleScope handle_scope(env()->isolate()); Context::Scope context_scope(context); Local emit_message = PersistentToLocal::Strong(emit_message_fn_); Local payload; Local message_error; Local argv[2]; { // Catch any exceptions from parsing the message itself (not from // emitting it) as 'messageeror' events. TryCatchScope try_catch(env()); if (!ReceiveMessage(context, true).ToLocal(&payload)) { if (try_catch.HasCaught() && !try_catch.HasTerminated()) message_error = try_catch.Exception(); goto reschedule; } } if (payload == env()->no_message_symbol()) break; if (!env()->can_call_into_js()) { Debug(this, "MessagePort drains queue because !can_call_into_js()"); // In this case there is nothing to do but to drain the current queue. continue; } argv[0] = payload; argv[1] = env()->message_string(); if (MakeCallback(emit_message, arraysize(argv), argv).IsEmpty()) { reschedule: if (!message_error.IsEmpty()) { argv[0] = message_error; argv[1] = env()->messageerror_string(); USE(MakeCallback(emit_message, arraysize(argv), argv)); } // Re-schedule OnMessage() execution in case of failure. if (data_) TriggerAsync(); return; } } } void MessagePort::OnClose() { Debug(this, "MessagePort::OnClose()"); if (data_) { // Detach() returns move(data_). Detach()->Disentangle(); } } std::unique_ptr MessagePort::Detach() { CHECK(data_); Mutex::ScopedLock lock(data_->mutex_); data_->owner_ = nullptr; return std::move(data_); } BaseObject::TransferMode MessagePort::GetTransferMode() const { if (IsDetached()) return BaseObject::TransferMode::kUntransferable; return BaseObject::TransferMode::kTransferable; } std::unique_ptr MessagePort::TransferForMessaging() { Close(); return Detach(); } BaseObjectPtr MessagePortData::Deserialize( Environment* env, Local context, std::unique_ptr self) { return BaseObjectPtr { MessagePort::New( env, context, static_unique_pointer_cast(std::move(self))) }; } Maybe MessagePort::PostMessage(Environment* env, Local message_v, const TransferList& transfer_v) { Isolate* isolate = env->isolate(); Local obj = object(isolate); Local context = obj->CreationContext(); std::shared_ptr msg = std::make_shared(); // Per spec, we need to both check if transfer list has the source port, and // serialize the input message, even if the MessagePort is closed or detached. Maybe serialization_maybe = msg->Serialize(env, context, message_v, transfer_v, obj); if (data_ == nullptr) { return serialization_maybe; } if (serialization_maybe.IsNothing()) { return Nothing(); } std::string error; Maybe res = data_->Dispatch(msg, &error); if (res.IsNothing()) return res; if (!error.empty()) ProcessEmitWarning(env, error.c_str()); return res; } Maybe MessagePortData::Dispatch( std::shared_ptr message, std::string* error) { if (!group_) { if (error != nullptr) *error = "MessagePortData is not entangled."; return Nothing(); } return group_->Dispatch(this, message, error); } static Maybe ReadIterable(Environment* env, Local context, // NOLINTNEXTLINE(runtime/references) TransferList& transfer_list, Local object) { if (!object->IsObject()) return Just(false); if (object->IsArray()) { Local arr = object.As(); size_t length = arr->Length(); transfer_list.AllocateSufficientStorage(length); for (size_t i = 0; i < length; i++) { if (!arr->Get(context, i).ToLocal(&transfer_list[i])) return Nothing(); } return Just(true); } Isolate* isolate = env->isolate(); Local iterator_method; if (!object.As()->Get(context, Symbol::GetIterator(isolate)) .ToLocal(&iterator_method)) return Nothing(); if (!iterator_method->IsFunction()) return Just(false); Local iterator; if (!iterator_method.As()->Call(context, object, 0, nullptr) .ToLocal(&iterator)) return Nothing(); if (!iterator->IsObject()) return Just(false); Local next; if (!iterator.As()->Get(context, env->next_string()).ToLocal(&next)) return Nothing(); if (!next->IsFunction()) return Just(false); std::vector> entries; while (env->can_call_into_js()) { Local result; if (!next.As()->Call(context, iterator, 0, nullptr) .ToLocal(&result)) return Nothing(); if (!result->IsObject()) return Just(false); Local done; if (!result.As()->Get(context, env->done_string()).ToLocal(&done)) return Nothing(); if (done->BooleanValue(isolate)) break; Local val; if (!result.As()->Get(context, env->value_string()).ToLocal(&val)) return Nothing(); entries.push_back(val); } transfer_list.AllocateSufficientStorage(entries.size()); std::copy(entries.begin(), entries.end(), &transfer_list[0]); return Just(true); } void MessagePort::PostMessage(const FunctionCallbackInfo& args) { Environment* env = Environment::GetCurrent(args); Local obj = args.This(); Local context = obj->CreationContext(); if (args.Length() == 0) { return THROW_ERR_MISSING_ARGS(env, "Not enough arguments to " "MessagePort.postMessage"); } if (!args[1]->IsNullOrUndefined() && !args[1]->IsObject()) { // Browsers ignore null or undefined, and otherwise accept an array or an // options object. return THROW_ERR_INVALID_ARG_TYPE(env, "Optional transferList argument must be an iterable"); } TransferList transfer_list; if (args[1]->IsObject()) { bool was_iterable; if (!ReadIterable(env, context, transfer_list, args[1]).To(&was_iterable)) return; if (!was_iterable) { Local transfer_option; if (!args[1].As()->Get(context, env->transfer_string()) .ToLocal(&transfer_option)) return; if (!transfer_option->IsUndefined()) { if (!ReadIterable(env, context, transfer_list, transfer_option) .To(&was_iterable)) return; if (!was_iterable) { return THROW_ERR_INVALID_ARG_TYPE(env, "Optional options.transfer argument must be an iterable"); } } } } MessagePort* port = Unwrap(args.This()); // Even if the backing MessagePort object has already been deleted, we still // want to serialize the message to ensure spec-compliant behavior w.r.t. // transfers. if (port == nullptr) { Message msg; USE(msg.Serialize(env, context, args[0], transfer_list, obj)); return; } Maybe res = port->PostMessage(env, args[0], transfer_list); if (res.IsJust()) args.GetReturnValue().Set(res.FromJust()); } void MessagePort::Start() { Debug(this, "Start receiving messages"); receiving_messages_ = true; Mutex::ScopedLock lock(data_->mutex_); if (!data_->incoming_messages_.empty()) TriggerAsync(); } void MessagePort::Stop() { Debug(this, "Stop receiving messages"); receiving_messages_ = false; } void MessagePort::Start(const FunctionCallbackInfo& args) { MessagePort* port; ASSIGN_OR_RETURN_UNWRAP(&port, args.This()); if (!port->data_) { return; } port->Start(); } void MessagePort::Stop(const FunctionCallbackInfo& args) { MessagePort* port; CHECK(args[0]->IsObject()); ASSIGN_OR_RETURN_UNWRAP(&port, args[0].As()); if (!port->data_) { return; } port->Stop(); } void MessagePort::CheckType(const FunctionCallbackInfo& args) { Environment* env = Environment::GetCurrent(args); args.GetReturnValue().Set( GetMessagePortConstructorTemplate(env)->HasInstance(args[0])); } void MessagePort::Drain(const FunctionCallbackInfo& args) { MessagePort* port; ASSIGN_OR_RETURN_UNWRAP(&port, args[0].As()); port->OnMessage(); } void MessagePort::ReceiveMessage(const FunctionCallbackInfo& args) { Environment* env = Environment::GetCurrent(args); if (!args[0]->IsObject() || !env->message_port_constructor_template()->HasInstance(args[0])) { return THROW_ERR_INVALID_ARG_TYPE(env, "The \"port\" argument must be a MessagePort instance"); } MessagePort* port = Unwrap(args[0].As()); if (port == nullptr) { // Return 'no messages' for a closed port. args.GetReturnValue().Set( Environment::GetCurrent(args)->no_message_symbol()); return; } MaybeLocal payload = port->ReceiveMessage(port->object()->CreationContext(), false); if (!payload.IsEmpty()) args.GetReturnValue().Set(payload.ToLocalChecked()); } void MessagePort::MoveToContext(const FunctionCallbackInfo& args) { Environment* env = Environment::GetCurrent(args); if (!args[0]->IsObject() || !env->message_port_constructor_template()->HasInstance(args[0])) { return THROW_ERR_INVALID_ARG_TYPE(env, "The \"port\" argument must be a MessagePort instance"); } MessagePort* port = Unwrap(args[0].As()); CHECK_NOT_NULL(port); Local context_arg = args[1]; ContextifyContext* context_wrapper; if (!context_arg->IsObject() || (context_wrapper = ContextifyContext::ContextFromContextifiedSandbox( env, context_arg.As())) == nullptr) { return THROW_ERR_INVALID_ARG_TYPE(env, "Invalid context argument"); } std::unique_ptr data; if (!port->IsDetached()) data = port->Detach(); Context::Scope context_scope(context_wrapper->context()); MessagePort* target = MessagePort::New(env, context_wrapper->context(), std::move(data)); if (target != nullptr) args.GetReturnValue().Set(target->object()); } void MessagePort::Entangle(MessagePort* a, MessagePort* b) { MessagePortData::Entangle(a->data_.get(), b->data_.get()); } void MessagePort::Entangle(MessagePort* a, MessagePortData* b) { MessagePortData::Entangle(a->data_.get(), b); } void MessagePort::MemoryInfo(MemoryTracker* tracker) const { tracker->TrackField("data", data_); tracker->TrackField("emit_message_fn", emit_message_fn_); } Local GetMessagePortConstructorTemplate(Environment* env) { // Factor generating the MessagePort JS constructor into its own piece // of code, because it is needed early on in the child environment setup. Local templ = env->message_port_constructor_template(); if (!templ.IsEmpty()) return templ; { Local m = env->NewFunctionTemplate(MessagePort::New); m->SetClassName(env->message_port_constructor_string()); m->InstanceTemplate()->SetInternalFieldCount( MessagePort::kInternalFieldCount); m->Inherit(HandleWrap::GetConstructorTemplate(env)); env->SetProtoMethod(m, "postMessage", MessagePort::PostMessage); env->SetProtoMethod(m, "start", MessagePort::Start); env->set_message_port_constructor_template(m); } return GetMessagePortConstructorTemplate(env); } JSTransferable::JSTransferable(Environment* env, Local obj) : BaseObject(env, obj) { MakeWeak(); } void JSTransferable::New(const FunctionCallbackInfo& args) { CHECK(args.IsConstructCall()); new JSTransferable(Environment::GetCurrent(args), args.This()); } JSTransferable::TransferMode JSTransferable::GetTransferMode() const { // Implement `kClone in this ? kCloneable : kTransferable`. HandleScope handle_scope(env()->isolate()); errors::TryCatchScope ignore_exceptions(env()); bool has_clone; if (!object()->Has(env()->context(), env()->messaging_clone_symbol()).To(&has_clone)) { return TransferMode::kUntransferable; } return has_clone ? TransferMode::kCloneable : TransferMode::kTransferable; } std::unique_ptr JSTransferable::TransferForMessaging() { return TransferOrClone(TransferMode::kTransferable); } std::unique_ptr JSTransferable::CloneForMessaging() const { return TransferOrClone(TransferMode::kCloneable); } std::unique_ptr JSTransferable::TransferOrClone( TransferMode mode) const { // Call `this[symbol]()` where `symbol` is `kClone` or `kTransfer`, // which should return an object with `data` and `deserializeInfo` properties; // `data` is written to the serializer later, and `deserializeInfo` is stored // on the `TransferData` instance as a string. HandleScope handle_scope(env()->isolate()); Local context = env()->isolate()->GetCurrentContext(); Local method_name = mode == TransferMode::kCloneable ? env()->messaging_clone_symbol() : env()->messaging_transfer_symbol(); Local method; if (!object()->Get(context, method_name).ToLocal(&method)) { return {}; } if (method->IsFunction()) { Local result_v; if (!method.As()->Call( context, object(), 0, nullptr).ToLocal(&result_v)) { return {}; } if (result_v->IsObject()) { Local result = result_v.As(); Local data; Local deserialize_info; if (!result->Get(context, env()->data_string()).ToLocal(&data) || !result->Get(context, env()->deserialize_info_string()) .ToLocal(&deserialize_info)) { return {}; } Utf8Value deserialize_info_str(env()->isolate(), deserialize_info); if (*deserialize_info_str == nullptr) return {}; return std::make_unique( *deserialize_info_str, Global(env()->isolate(), data)); } } if (mode == TransferMode::kTransferable) return TransferOrClone(TransferMode::kCloneable); else return {}; } Maybe JSTransferable::NestedTransferables() const { // Call `this[kTransferList]()` and return the resulting list of BaseObjects. HandleScope handle_scope(env()->isolate()); Local context = env()->isolate()->GetCurrentContext(); Local method_name = env()->messaging_transfer_list_symbol(); Local method; if (!object()->Get(context, method_name).ToLocal(&method)) { return Nothing(); } if (!method->IsFunction()) return Just(BaseObjectList {}); Local list_v; if (!method.As()->Call( context, object(), 0, nullptr).ToLocal(&list_v)) { return Nothing(); } if (!list_v->IsArray()) return Just(BaseObjectList {}); Local list = list_v.As(); BaseObjectList ret; for (size_t i = 0; i < list->Length(); i++) { Local value; if (!list->Get(context, i).ToLocal(&value)) return Nothing(); if (env()->base_object_ctor_template()->HasInstance(value)) ret.emplace_back(Unwrap(value)); } return Just(ret); } Maybe JSTransferable::FinalizeTransferRead( Local context, ValueDeserializer* deserializer) { // Call `this[kDeserialize](data)` where `data` comes from the return value // of `this[kTransfer]()` or `this[kClone]()`. HandleScope handle_scope(env()->isolate()); Local data; if (!deserializer->ReadValue(context).ToLocal(&data)) return Nothing(); Local method_name = env()->messaging_deserialize_symbol(); Local method; if (!object()->Get(context, method_name).ToLocal(&method)) { return Nothing(); } if (!method->IsFunction()) return Just(true); if (method.As()->Call(context, object(), 1, &data).IsEmpty()) { return Nothing(); } return Just(true); } JSTransferable::Data::Data(std::string&& deserialize_info, v8::Global&& data) : deserialize_info_(std::move(deserialize_info)), data_(std::move(data)) {} BaseObjectPtr JSTransferable::Data::Deserialize( Environment* env, Local context, std::unique_ptr self) { // Create the JS wrapper object that will later be filled with data passed to // the `[kDeserialize]()` method on it. This split is necessary, because here // we need to create an object with the right prototype and internal fields, // but the actual JS data stored in the serialized data can only be read at // the end of the stream, after the main message has been read. if (context != env->context()) { THROW_ERR_MESSAGE_TARGET_CONTEXT_UNAVAILABLE(env); return {}; } HandleScope handle_scope(env->isolate()); Local info; if (!ToV8Value(context, deserialize_info_).ToLocal(&info)) return {}; Local ret; CHECK(!env->messaging_deserialize_create_object().IsEmpty()); if (!env->messaging_deserialize_create_object()->Call( context, Null(env->isolate()), 1, &info).ToLocal(&ret) || !env->base_object_ctor_template()->HasInstance(ret)) { return {}; } return BaseObjectPtr { Unwrap(ret) }; } Maybe JSTransferable::Data::FinalizeTransferWrite( Local context, ValueSerializer* serializer) { HandleScope handle_scope(context->GetIsolate()); auto ret = serializer->WriteValue(context, PersistentToLocal::Strong(data_)); data_.Reset(); return ret; } std::shared_ptr SiblingGroup::Get(const std::string& name) { Mutex::ScopedLock lock(SiblingGroup::groups_mutex_); std::shared_ptr group; auto i = groups_.find(name); if (i == groups_.end() || i->second.expired()) { group = std::make_shared(name); groups_[name] = group; } else { group = i->second.lock(); } return group; } void SiblingGroup::CheckSiblingGroup(const std::string& name) { Mutex::ScopedLock lock(SiblingGroup::groups_mutex_); auto i = groups_.find(name); if (i != groups_.end() && i->second.expired()) groups_.erase(name); } SiblingGroup::SiblingGroup(const std::string& name) : name_(name) { } SiblingGroup::~SiblingGroup() { // If this is a named group, check to see if we can remove the group if (!name_.empty()) CheckSiblingGroup(name_); } Maybe SiblingGroup::Dispatch( MessagePortData* source, std::shared_ptr message, std::string* error) { Mutex::ScopedLock lock(group_mutex_); // The source MessagePortData is not part of this group. if (ports_.find(source) == ports_.end()) { if (error != nullptr) *error = "Source MessagePort is not entangled with this group."; return Nothing(); } // There are no destination ports. if (size() <= 1) return Just(false); // Transferables cannot be used when there is more // than a single destination. if (size() > 2 && message->has_transferables()) { if (error != nullptr) *error = "Transferables cannot be used with multiple destinations."; return Nothing(); } for (MessagePortData* port : ports_) { if (port == source) continue; // This loop should only be entered if there's only a single destination for (const auto& transferable : message->transferables()) { if (port == transferable.get()) { if (error != nullptr) { *error = "The target port was posted to itself, and the " "communication channel was lost"; } return Just(true); } } port->AddToIncomingQueue(message); } return Just(true); } void SiblingGroup::Entangle(MessagePortData* port) { Entangle({ port }); } void SiblingGroup::Entangle(std::initializer_list ports) { Mutex::ScopedLock lock(group_mutex_); for (MessagePortData* data : ports) { ports_.insert(data); CHECK(!data->group_); data->group_ = shared_from_this(); } } void SiblingGroup::Disentangle(MessagePortData* data) { auto self = shared_from_this(); // Keep alive until end of function. Mutex::ScopedLock lock(group_mutex_); ports_.erase(data); data->group_.reset(); data->AddToIncomingQueue(std::make_shared()); // If this is an anonymous group and there's another port, close it. if (size() == 1 && name_.empty()) (*(ports_.begin()))->AddToIncomingQueue(std::make_shared()); } SiblingGroup::Map SiblingGroup::groups_; Mutex SiblingGroup::groups_mutex_; namespace { static void SetDeserializerCreateObjectFunction( const FunctionCallbackInfo& args) { Environment* env = Environment::GetCurrent(args); CHECK(args[0]->IsFunction()); env->set_messaging_deserialize_create_object(args[0].As()); } static void MessageChannel(const FunctionCallbackInfo& args) { Environment* env = Environment::GetCurrent(args); if (!args.IsConstructCall()) { THROW_ERR_CONSTRUCT_CALL_REQUIRED(env); return; } Local context = args.This()->CreationContext(); 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()) .Check(); args.This()->Set(context, env->port2_string(), port2->object()) .Check(); } static void BroadcastChannel(const FunctionCallbackInfo& args) { CHECK(args[0]->IsString()); Environment* env = Environment::GetCurrent(args); Context::Scope context_scope(env->context()); Utf8Value name(env->isolate(), args[0]); MessagePort* port = MessagePort::New(env, env->context(), {}, SiblingGroup::Get(*name)); if (port != nullptr) { args.GetReturnValue().Set(port->object()); } } static void InitMessaging(Local target, Local unused, Local context, void* priv) { Environment* env = Environment::GetCurrent(context); { Local message_channel_string = FIXED_ONE_BYTE_STRING(env->isolate(), "MessageChannel"); Local templ = env->NewFunctionTemplate(MessageChannel); templ->SetClassName(message_channel_string); target->Set(context, message_channel_string, templ->GetFunction(context).ToLocalChecked()).Check(); } { Local js_transferable_string = FIXED_ONE_BYTE_STRING(env->isolate(), "JSTransferable"); Local t = env->NewFunctionTemplate(JSTransferable::New); t->Inherit(BaseObject::GetConstructorTemplate(env)); t->SetClassName(js_transferable_string); t->InstanceTemplate()->SetInternalFieldCount( JSTransferable::kInternalFieldCount); target->Set(context, js_transferable_string, t->GetFunction(context).ToLocalChecked()).Check(); } target->Set(context, env->message_port_constructor_string(), GetMessagePortConstructorTemplate(env) ->GetFunction(context).ToLocalChecked()).Check(); // These are not methods on the MessagePort prototype, because // the browser equivalents do not provide them. env->SetMethod(target, "stopMessagePort", MessagePort::Stop); env->SetMethod(target, "checkMessagePort", MessagePort::CheckType); env->SetMethod(target, "drainMessagePort", MessagePort::Drain); env->SetMethod(target, "receiveMessageOnPort", MessagePort::ReceiveMessage); env->SetMethod(target, "moveMessagePortToContext", MessagePort::MoveToContext); env->SetMethod(target, "setDeserializerCreateObjectFunction", SetDeserializerCreateObjectFunction); env->SetMethod(target, "broadcastChannel", BroadcastChannel); { Local domexception = GetDOMException(context).ToLocalChecked(); target ->Set(context, FIXED_ONE_BYTE_STRING(env->isolate(), "DOMException"), domexception) .Check(); } } static void RegisterExternalReferences(ExternalReferenceRegistry* registry) { registry->Register(MessageChannel); registry->Register(BroadcastChannel); registry->Register(JSTransferable::New); registry->Register(MessagePort::New); registry->Register(MessagePort::PostMessage); registry->Register(MessagePort::Start); registry->Register(MessagePort::Stop); registry->Register(MessagePort::CheckType); registry->Register(MessagePort::Drain); registry->Register(MessagePort::ReceiveMessage); registry->Register(MessagePort::MoveToContext); registry->Register(SetDeserializerCreateObjectFunction); } } // anonymous namespace } // namespace worker } // namespace node NODE_MODULE_CONTEXT_AWARE_INTERNAL(messaging, node::worker::InitMessaging) NODE_MODULE_EXTERNAL_REFERENCE(messaging, node::worker::RegisterExternalReferences)