diff --git a/lib/diagnostics_channel.js b/lib/diagnostics_channel.js index 7b78851208df66..c33908e59406e1 100644 --- a/lib/diagnostics_channel.js +++ b/lib/diagnostics_channel.js @@ -31,7 +31,6 @@ const { const { triggerUncaughtException } = internalBinding('errors'); const dc_binding = internalBinding('diagnostics_channel'); -const { subscribers: subscriberCounts } = dc_binding; const { WeakReference, kEmptyObject } = require('internal/util'); const { isPromise } = require('internal/util/types'); @@ -132,7 +131,7 @@ class ActiveChannel { this._subscribers = ArrayPrototypeSlice(this._subscribers); ArrayPrototypePush(this._subscribers, subscription); channels.incRef(this.name); - if (this._index !== undefined) subscriberCounts[this._index]++; + if (this._index !== undefined) dc_binding.subscribers[this._index]++; } unsubscribe(subscription) { @@ -145,7 +144,7 @@ class ActiveChannel { ArrayPrototypePushApply(this._subscribers, after); channels.decRef(this.name); - if (this._index !== undefined) subscriberCounts[this._index]--; + if (this._index !== undefined) dc_binding.subscribers[this._index]--; maybeMarkInactive(this); return true; @@ -155,7 +154,7 @@ class ActiveChannel { const replacing = this._stores.has(store); if (!replacing) { channels.incRef(this.name); - if (this._index !== undefined) subscriberCounts[this._index]++; + if (this._index !== undefined) dc_binding.subscribers[this._index]++; } this._stores.set(store, transform); } @@ -168,7 +167,7 @@ class ActiveChannel { this._stores.delete(store); channels.decRef(this.name); - if (this._index !== undefined) subscriberCounts[this._index]--; + if (this._index !== undefined) dc_binding.subscribers[this._index]--; maybeMarkInactive(this); return true; @@ -209,7 +208,7 @@ class Channel { this._stores = undefined; this.name = name; if (typeof name === 'string') { - this._index = dc_binding.getOrCreateChannelIndex(name); + this._index = undefined; } channels.set(name, this); @@ -640,7 +639,14 @@ function tracingChannel(nameOrChannels) { return new TracingChannel(nameOrChannels); } -dc_binding.linkNativeChannel((name) => channel(name)); +dc_binding.linkNativeChannel((name, index) => { + const linkedChannel = channel(name); + linkedChannel._index = index; + dc_binding.subscribers[index] = + (linkedChannel._subscribers?.length || 0) + + (linkedChannel._stores?.size || 0); + return linkedChannel; +}); module.exports = { channel, diff --git a/lib/internal/process/pre_execution.js b/lib/internal/process/pre_execution.js index 38ea6675928ad8..90474aba88052f 100644 --- a/lib/internal/process/pre_execution.js +++ b/lib/internal/process/pre_execution.js @@ -669,7 +669,14 @@ function setupDiagnosticsChannel() { // JS references are cleared during serialization. const dc = require('diagnostics_channel'); const dc_binding = internalBinding('diagnostics_channel'); - dc_binding.linkNativeChannel((name) => dc.channel(name)); + dc_binding.linkNativeChannel((name, index) => { + const channel = dc.channel(name); + channel._index = index; + dc_binding.subscribers[index] = + (channel._subscribers?.length || 0) + + (channel._stores?.size || 0); + return channel; + }); } function initializePermission() { diff --git a/src/node_diagnostics_channel.cc b/src/node_diagnostics_channel.cc index 450a124c86959a..cfae019da62f02 100644 --- a/src/node_diagnostics_channel.cc +++ b/src/node_diagnostics_channel.cc @@ -16,6 +16,7 @@ using v8::Function; using v8::FunctionCallbackInfo; using v8::FunctionTemplate; using v8::HandleScope; +using v8::Integer; using v8::Isolate; using v8::Local; using v8::Object; @@ -28,8 +29,10 @@ BindingData::BindingData(Realm* realm, Local wrap, InternalFieldInfo* info) : SnapshotableObject(realm, wrap, type_int), - subscribers_( - realm->isolate(), kMaxChannels, MAYBE_FIELD_PTR(info, subscribers)) { + subscribers_(realm->isolate(), + info == nullptr ? kInitialChannelCapacity + : info->subscribers_capacity, + MAYBE_FIELD_PTR(info, subscribers)) { if (info == nullptr) { wrap->Set(realm->context(), FIXED_ONE_BYTE_STRING(realm->isolate(), "subscribers"), @@ -50,25 +53,20 @@ uint32_t BindingData::GetOrCreateChannelIndex(const std::string& name) { if (it != channel_indices_.end()) { return it->second; } - CHECK_LT(next_channel_index_, kMaxChannels); + if (next_channel_index_ == subscribers_.Length()) { + subscribers_.reserve(subscribers_.Length() * 2); + object() + ->Set(realm()->context(), + FIXED_ONE_BYTE_STRING(realm()->isolate(), "subscribers"), + subscribers_.GetJSArray()) + .Check(); + subscribers_.MakeWeak(); + } uint32_t index = next_channel_index_++; channel_indices_.emplace(name, index); return index; } -void BindingData::GetOrCreateChannelIndex( - const FunctionCallbackInfo& args) { - Realm* realm = Realm::GetCurrent(args); - BindingData* binding = realm->GetBindingData(); - CHECK_NOT_NULL(binding); - - CHECK(args[0]->IsString()); - Utf8Value name(realm->isolate(), args[0]); - - uint32_t index = binding->GetOrCreateChannelIndex(*name); - args.GetReturnValue().Set(index); -} - void BindingData::LinkNativeChannel(const FunctionCallbackInfo& args) { Realm* realm = Realm::GetCurrent(args); BindingData* binding = realm->GetBindingData(); @@ -85,10 +83,11 @@ void BindingData::LinkNativeChannel(const FunctionCallbackInfo& args) { Local name = String::NewFromUtf8(isolate, channel_ptr->name_.c_str()) .ToLocalChecked(); - Local argv[] = {name}; + Local argv[] = { + name, Integer::NewFromUnsigned(isolate, channel_ptr->index_)}; Local result; if (binding->link_callback_.Get(isolate) - ->Call(context, v8::Undefined(isolate), 1, argv) + ->Call(context, v8::Undefined(isolate), arraysize(argv), argv) .ToLocal(&result) && result->IsObject()) { channel_ptr->Link(isolate, result.As()); @@ -102,6 +101,7 @@ bool BindingData::PrepareForSerialization(Local context, DCHECK_NULL(internal_field_info_); internal_field_info_ = InternalFieldInfoBase::New(type()); internal_field_info_->subscribers = subscribers_.Serialize(context, creator); + internal_field_info_->subscribers_capacity = subscribers_.Length(); link_callback_.Reset(); channel_wrap_template_.Reset(); channels_.clear(); @@ -130,8 +130,6 @@ void BindingData::Deserialize(Local context, void BindingData::CreatePerIsolateProperties(IsolateData* isolate_data, Local target) { Isolate* isolate = isolate_data->isolate(); - SetMethod( - isolate, target, "getOrCreateChannelIndex", GetOrCreateChannelIndex); SetMethod(isolate, target, "linkNativeChannel", LinkNativeChannel); } @@ -146,7 +144,6 @@ void BindingData::CreatePerContextProperties(Local target, void BindingData::RegisterExternalReferences( ExternalReferenceRegistry* registry) { - registry->Register(GetOrCreateChannelIndex); registry->Register(LinkNativeChannel); } @@ -226,10 +223,10 @@ Channel* Channel::Get(Environment* env, const char* name) { HandleScope handle_scope(isolate); Local context = env->context(); Local js_name = String::NewFromUtf8(isolate, name).ToLocalChecked(); - Local argv[] = {js_name}; + Local argv[] = {js_name, Integer::NewFromUnsigned(isolate, index)}; Local result; if (binding->link_callback_.Get(isolate) - ->Call(context, v8::Undefined(isolate), 1, argv) + ->Call(context, v8::Undefined(isolate), arraysize(argv), argv) .ToLocal(&result) && result->IsObject()) { channel->Link(isolate, result.As()); diff --git a/src/node_diagnostics_channel.h b/src/node_diagnostics_channel.h index 1c1831a0f9e45f..073e4e4f273bdf 100644 --- a/src/node_diagnostics_channel.h +++ b/src/node_diagnostics_channel.h @@ -20,10 +20,11 @@ class Channel; class BindingData : public SnapshotableObject { public: - static constexpr size_t kMaxChannels = 1024; + static constexpr size_t kInitialChannelCapacity = 1024; struct InternalFieldInfo : public node::InternalFieldInfoBase { AliasedBufferIndex subscribers; + size_t subscribers_capacity; }; BindingData(Realm* realm, @@ -48,8 +49,6 @@ class BindingData : public SnapshotableObject { v8::Global channel_wrap_template_; std::vector> channels_; - static void GetOrCreateChannelIndex( - const v8::FunctionCallbackInfo& args); static void LinkNativeChannel( const v8::FunctionCallbackInfo& args); diff --git a/test/cctest/test_diagnostics_channel.cc b/test/cctest/test_diagnostics_channel.cc index b456f6aaf9eaab..f4295d3ba8b130 100644 --- a/test/cctest/test_diagnostics_channel.cc +++ b/test/cctest/test_diagnostics_channel.cc @@ -257,3 +257,45 @@ TEST_F(DiagnosticsChannelTest, JSChannelVisibleFromCpp) { EXPECT_TRUE(js_has_subs->IsTrue()); EXPECT_TRUE(ch->HasSubscribers()); } + +// Native channels grow the shared subscriber storage past its initial +// capacity without losing the state of channels that were already linked. +TEST_F(DiagnosticsChannelTest, NativeChannelsGrowSubscriberStorage) { + const v8::HandleScope handle_scope(isolate_); + Argv argv; + Env env{handle_scope, argv}; + + SetProcessExitHandler(*env, [&](node::Environment* env_, int exit_code) { + EXPECT_EQ(exit_code, 0); + node::Stop(*env); + }); + + node::LoadEnvironment( + *env, + "globalThis.__dc = require('diagnostics_channel');" + "globalThis.__firstSubscriber = () => {};" + "globalThis.__dc.subscribe('test:cctest:grow:0', " + " globalThis.__firstSubscriber);"); + + Channel* first = Channel::Get(*env, "test:cctest:grow:0"); + ASSERT_NE(first, nullptr); + ASSERT_TRUE(first->HasSubscribers()); + + Channel* last = nullptr; + for (size_t i = 1; i <= 1024; i++) { + std::string name = "test:cctest:grow:" + std::to_string(i); + last = Channel::Get(*env, name.c_str()); + ASSERT_NE(last, nullptr); + } + + RunJS(isolate_, + "globalThis.__dc.unsubscribe('test:cctest:grow:0', " + " globalThis.__firstSubscriber);"); + EXPECT_FALSE(first->HasSubscribers()); + + RunJS(isolate_, + "globalThis.__lastSubscriber = () => {};" + "globalThis.__dc.subscribe('test:cctest:grow:1024', " + " globalThis.__lastSubscriber);"); + EXPECT_TRUE(last->HasSubscribers()); +} diff --git a/test/parallel/test-diagnostics-channel-many-channels.js b/test/parallel/test-diagnostics-channel-many-channels.js new file mode 100644 index 00000000000000..3183bc10f485c1 --- /dev/null +++ b/test/parallel/test-diagnostics-channel-many-channels.js @@ -0,0 +1,19 @@ +'use strict'; + +const common = require('../common'); +const assert = require('node:assert'); +const dc = require('node:diagnostics_channel'); + +let last; +for (let i = 0; i < 1024 * 10 + 1; i++) { + last = dc.channel(`test:many-channels:${i}`); +} + +const onMessage = common.mustCall((message, name) => { + assert.strictEqual(message, 'message'); + assert.strictEqual(name, last.name); +}); + +last.subscribe(onMessage); +last.publish('message'); +assert.strictEqual(last.unsubscribe(onMessage), true); diff --git a/test/parallel/test-permission-diagnostics-channel.js b/test/parallel/test-permission-diagnostics-channel.js index add99d5f8f0970..3593305e552c9d 100644 --- a/test/parallel/test-permission-diagnostics-channel.js +++ b/test/parallel/test-permission-diagnostics-channel.js @@ -12,6 +12,12 @@ const assert = require('node:assert'); const dc = require('node:diagnostics_channel'); const fs = require('node:fs'); +// JS-only channels must not consume the native subscriber storage used by the +// permission audit publisher. +for (let i = 0; i < 1024 * 10 + 1; i++) { + dc.channel(`test:permission:unrelated:${i}`); +} + const messages = []; dc.subscribe('node:permission-model:fs', (msg) => { messages.push(msg);