Skip to content
Open
Show file tree
Hide file tree
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/diagnostics_channel.js
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down Expand Up @@ -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) {
Expand All @@ -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;
Expand All @@ -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);
}
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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,
Expand Down
9 changes: 8 additions & 1 deletion lib/internal/process/pre_execution.js
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
43 changes: 20 additions & 23 deletions src/node_diagnostics_channel.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -28,8 +29,10 @@ BindingData::BindingData(Realm* realm,
Local<Object> 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"),
Expand All @@ -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<Value>& args) {
Realm* realm = Realm::GetCurrent(args);
BindingData* binding = realm->GetBindingData<BindingData>();
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<Value>& args) {
Realm* realm = Realm::GetCurrent(args);
BindingData* binding = realm->GetBindingData<BindingData>();
Expand All @@ -85,10 +83,11 @@ void BindingData::LinkNativeChannel(const FunctionCallbackInfo<Value>& args) {
Local<String> name =
String::NewFromUtf8(isolate, channel_ptr->name_.c_str())
.ToLocalChecked();
Local<Value> argv[] = {name};
Local<Value> argv[] = {
name, Integer::NewFromUnsigned(isolate, channel_ptr->index_)};
Local<Value> 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<Object>());
Expand All @@ -102,6 +101,7 @@ bool BindingData::PrepareForSerialization(Local<Context> context,
DCHECK_NULL(internal_field_info_);
internal_field_info_ = InternalFieldInfoBase::New<InternalFieldInfo>(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();
Expand Down Expand Up @@ -130,8 +130,6 @@ void BindingData::Deserialize(Local<Context> context,
void BindingData::CreatePerIsolateProperties(IsolateData* isolate_data,
Local<ObjectTemplate> target) {
Isolate* isolate = isolate_data->isolate();
SetMethod(
isolate, target, "getOrCreateChannelIndex", GetOrCreateChannelIndex);
SetMethod(isolate, target, "linkNativeChannel", LinkNativeChannel);
}

Expand All @@ -146,7 +144,6 @@ void BindingData::CreatePerContextProperties(Local<Object> target,

void BindingData::RegisterExternalReferences(
ExternalReferenceRegistry* registry) {
registry->Register(GetOrCreateChannelIndex);
registry->Register(LinkNativeChannel);
}

Expand Down Expand Up @@ -226,10 +223,10 @@ Channel* Channel::Get(Environment* env, const char* name) {
HandleScope handle_scope(isolate);
Local<Context> context = env->context();
Local<String> js_name = String::NewFromUtf8(isolate, name).ToLocalChecked();
Local<Value> argv[] = {js_name};
Local<Value> argv[] = {js_name, Integer::NewFromUnsigned(isolate, index)};
Local<Value> 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<Object>());
Expand Down
5 changes: 2 additions & 3 deletions src/node_diagnostics_channel.h
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -48,8 +49,6 @@ class BindingData : public SnapshotableObject {
v8::Global<v8::FunctionTemplate> channel_wrap_template_;
std::vector<BaseObjectPtr<Channel>> channels_;

static void GetOrCreateChannelIndex(
const v8::FunctionCallbackInfo<v8::Value>& args);
static void LinkNativeChannel(
const v8::FunctionCallbackInfo<v8::Value>& args);

Expand Down
42 changes: 42 additions & 0 deletions test/cctest/test_diagnostics_channel.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
19 changes: 19 additions & 0 deletions test/parallel/test-diagnostics-channel-many-channels.js
Original file line number Diff line number Diff line change
@@ -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);
6 changes: 6 additions & 0 deletions test/parallel/test-permission-diagnostics-channel.js
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading