diff --git a/benchmark/diagnostics_channel/threadpool-work.js b/benchmark/diagnostics_channel/threadpool-work.js new file mode 100644 index 000000000000..f038d49f56a0 --- /dev/null +++ b/benchmark/diagnostics_channel/threadpool-work.js @@ -0,0 +1,61 @@ +'use strict'; + +const assert = require('node:assert'); +const crypto = require('node:crypto'); +const dc = require('node:diagnostics_channel'); +const fs = require('node:fs'); +const zlib = require('node:zlib'); +const common = require('../common.js'); +const tmpdir = require('../../test/common/tmpdir'); + +const bench = common.createBenchmark(main, { + n: [1e3, 1e4], + mode: ['subscribed', 'unsubscribed'], + operation: ['crypto', 'zlib', 'readFile', 'writeFile'], +}); + +function main({ n, mode, operation }) { + const subscriber = () => {}; + if (mode === 'subscribed') { + dc.subscribe('threadpool.work', subscriber); + } else if (mode === 'unsubscribed') { + dc.subscribe('threadpool.work', subscriber); + dc.unsubscribe('threadpool.work', subscriber); + } + + tmpdir.refresh(); + const file = tmpdir.resolve('file'); + const data = Buffer.alloc(1024, 'x'); + fs.writeFileSync(file, data); + + const operations = { + crypto(callback) { + crypto.pbkdf2('secret', 'salt', 1, 32, 'sha256', callback); + }, + zlib(callback) { + zlib.gzip(data, callback); + }, + readFile(callback) { + fs.readFile(file, callback); + }, + writeFile(callback) { + fs.writeFile(file, data, callback); + }, + }; + + let completed = 0; + bench.start(); + run(); + + function run() { + operations[operation]((err) => { + assert.ifError(err); + if (++completed < n) return run(); + + bench.end(n); + if (mode === 'subscribed') { + dc.unsubscribe('threadpool.work', subscriber); + } + }); + } +} diff --git a/doc/api/diagnostics_channel.md b/doc/api/diagnostics_channel.md index e63f23829f90..2f4c02c6ba9f 100644 --- a/doc/api/diagnostics_channel.md +++ b/doc/api/diagnostics_channel.md @@ -2026,6 +2026,27 @@ statement is garbage collected. Subscribers must not close the database or the statement, since both are still in use while the event is being delivered; see [`database.close()`][] and [`statement.close()`][]. +#### Thread Pool + + + +> Stability: 1 - Experimental + +##### Event: `'threadpool.work'` + +* `type` {string} The kind of work that ran. Values include `'zlib'`, + `'crypto'`, `'node_api'`, `'fs.readfile'`, `'fs.writefile'`, `'fs.cp'`, + `'readdir_recursive'`, and `'node_sqlite3.BackupJob'`. +* `enqueued` {number} When the work was submitted to the pool. +* `started` {number|null} When execution started, or `null` if cancelled. +* `ended` {number|null} When execution ended, or `null` if cancelled. + +Emitted after the work finishes and before its completion callback. Timestamps +use the [`performance.now()`][] timeline. `started - enqueued` is queue time; +`ended - started` is execution time. + [BoundedChannel Channels]: #boundedchannel-channels [TracingChannel Channels]: #tracingchannel-channels [`'uncaughtException'`]: process.md#event-uncaughtexception @@ -2051,6 +2072,7 @@ statement, since both are still in use while the event is being delivered; see [`error` event]: #errorevent [`locks.request()`]: worker_threads.md#locksrequestname-options-callback [`net.Server.listen()`]: net.md#serverlisten +[`performance.now()`]: perf_hooks.md#performancenow [`process.execve()`]: process.md#processexecvefile-args-env [`start` event]: #startevent [`statement.close()`]: sqlite.md#statementclose diff --git a/src/env.cc b/src/env.cc index d35c28fcf7ed..ff27b3acc223 100644 --- a/src/env.cc +++ b/src/env.cc @@ -8,6 +8,7 @@ #include "node_buffer.h" #include "node_context_data.h" #include "node_contextify.h" +#include "node_diagnostics_channel.h" #include "node_errors.h" #include "node_file_utils.h" #include "node_internals.h" @@ -1275,6 +1276,27 @@ Environment::~Environment() { } } +void Environment::InitializeThreadPoolWorkChannel() { + if (threadpool_work_channel_.get() != nullptr || + isolate_data()->is_building_snapshot()) { + return; + } + + BaseObjectPtr channel = + diagnostics_channel::Channel::Get(this, "threadpool.work"); + if (!channel) return; + + auto* binding = + principal_realm()->GetBindingData(); + CHECK_NOT_NULL(binding); + binding->SetChannelStatusCallback(channel->index(), [this](bool active) { + threadpool_work_channel_active_ = active; + }); + threadpool_work_channel_ = + BaseObjectWeakPtr(channel.get()); + threadpool_work_channel_active_ = channel->HasSubscribers(); +} + void Environment::InitializeLibuv() { HandleScope handle_scope(isolate()); Context::Scope context_scope(context()); diff --git a/src/env.h b/src/env.h index 29754ddb6371..4bc0b7a85714 100644 --- a/src/env.h +++ b/src/env.h @@ -76,6 +76,10 @@ class MacCache; namespace node { +namespace diagnostics_channel { +class Channel; +} + namespace shadow_realm { class ShadowRealm; } @@ -714,6 +718,14 @@ class Environment final : public MemoryRetainer { void RunDeserializeRequests(); // Should be called before InitializeInspector() void InitializeDiagnostics(); + void InitializeThreadPoolWorkChannel(); + inline bool has_threadpool_work_subscribers() const { + return threadpool_work_channel_active_; + } + inline const BaseObjectWeakPtr& + threadpool_work_channel() const { + return threadpool_work_channel_; + } #if HAVE_INSPECTOR // If the environment is created for a worker, pass parent_handle and @@ -1216,6 +1228,8 @@ class Environment final : public MemoryRetainer { AliasedInt32Array timeout_info_; TickInfo tick_info_; permission::Permission permission_; + BaseObjectWeakPtr threadpool_work_channel_; + bool threadpool_work_channel_active_ = false; const uint64_t timer_base_; std::shared_ptr env_vars_; bool printed_error_ = false; diff --git a/src/env_properties.h b/src/env_properties.h index bfb981d1c89b..14298d3d6e88 100644 --- a/src/env_properties.h +++ b/src/env_properties.h @@ -487,6 +487,7 @@ V(srv_record_template, v8::DictionaryTemplate) \ V(streambaseoutputstream_constructor_template, v8::ObjectTemplate) \ V(tcp_constructor_template, v8::FunctionTemplate) \ + V(threadpool_work_template, v8::DictionaryTemplate) \ V(tlsa_record_template, v8::DictionaryTemplate) \ V(tty_constructor_template, v8::FunctionTemplate) \ V(txt_record_template, v8::DictionaryTemplate) \ diff --git a/src/node_diagnostics_channel.cc b/src/node_diagnostics_channel.cc index 2593f6eab90f..c438add3aa60 100644 --- a/src/node_diagnostics_channel.cc +++ b/src/node_diagnostics_channel.cc @@ -3,6 +3,7 @@ #include "base_object-inl.h" #include "env-inl.h" #include "node_external_reference.h" +#include "node_internals.h" #include "util-inl.h" #include "v8.h" @@ -94,6 +95,7 @@ void BindingData::LinkNativeChannel(const FunctionCallbackInfo& args) { } } } + realm->env()->InitializeThreadPoolWorkChannel(); } bool BindingData::PrepareForSerialization(Local context, @@ -102,6 +104,7 @@ bool BindingData::PrepareForSerialization(Local context, internal_field_info_ = InternalFieldInfoBase::New(type()); internal_field_info_->subscribers = subscribers_.Serialize(context, creator); internal_field_info_->subscribers_capacity = subscribers_.Length(); + channel_status_callbacks_.clear(); link_callback_.Reset(); channel_wrap_template_.Reset(); channels_.clear(); diff --git a/src/node_diagnostics_channel.h b/src/node_diagnostics_channel.h index c8c1a79994b2..6f06754719a2 100644 --- a/src/node_diagnostics_channel.h +++ b/src/node_diagnostics_channel.h @@ -85,6 +85,8 @@ class Channel : public BaseObject { static BaseObjectPtr Get(Environment* env, std::string_view name); + uint32_t index() const { return index_; } + inline bool HasSubscribers() const { return binding_data_ != nullptr && binding_data_->subscribers_[index_] > 0; } diff --git a/src/node_internals.h b/src/node_internals.h index 17e23ce61d64..2096281dd511 100644 --- a/src/node_internals.h +++ b/src/node_internals.h @@ -45,6 +45,10 @@ struct sockaddr; namespace node { +namespace diagnostics_channel { +class Channel; +} + namespace builtins { class BuiltinLoader; } @@ -312,9 +316,18 @@ class ThreadPoolWork { Environment* env() const { return env_; } private: + inline bool IsObserved() const { return enqueued_at_ != 0; } + + inline void PublishDiagnostics(diagnostics_channel::Channel& channel); + Environment* env_; uv_work_t work_req_; const char* type_; + + // libuv synchronizes these marks between the loop and worker threads. + uint64_t enqueued_at_ = 0; + uint64_t work_start_ = 0; + uint64_t work_end_ = 0; }; // Functions defined in node.cc that are exposed via the bootstrapper object diff --git a/src/threadpoolwork-inl.h b/src/threadpoolwork-inl.h index 729e3de8eeaa..dd26a4069bd8 100644 --- a/src/threadpoolwork-inl.h +++ b/src/threadpoolwork-inl.h @@ -24,6 +24,9 @@ #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS +#include +#include "env-inl.h" +#include "node_diagnostics_channel.h" #include "node_internals.h" #include "tracing/trace_event.h" #include "util-inl.h" @@ -34,14 +37,26 @@ void ThreadPoolWork::ScheduleWork() { env_->IncreaseWaitingRequestCounter(); TRACE_EVENT_NESTABLE_ASYNC_BEGIN0( TRACING_CATEGORY_NODE2(threadpoolwork, async), type_, this); + + // Some ThreadPoolWork instances are submitted more than once. + enqueued_at_ = 0; + work_start_ = 0; + work_end_ = 0; + + if (env_->has_threadpool_work_subscribers()) [[unlikely]] { + enqueued_at_ = uv_hrtime(); + } + int status = uv_queue_work( env_->event_loop(), &work_req_, [](uv_work_t* req) { ThreadPoolWork* self = ContainerOf(&ThreadPoolWork::work_req_, req); + if (self->IsObserved()) self->work_start_ = uv_hrtime(); TRACE_EVENT_BEGIN0(TRACING_CATEGORY_NODE2(threadpoolwork, sync), self->type_); self->DoThreadPoolWork(); + if (self->IsObserved()) self->work_end_ = uv_hrtime(); TRACE_EVENT_END0(TRACING_CATEGORY_NODE2(threadpoolwork, sync), self->type_); }, @@ -54,11 +69,50 @@ void ThreadPoolWork::ScheduleWork() { self, "result", status); + // AfterThreadPoolWork() may unsubscribe or delete `self`. + if (self->IsObserved()) { + auto* channel = self->env_->threadpool_work_channel().get(); + if (channel != nullptr) self->PublishDiagnostics(*channel); + } self->AfterThreadPoolWork(status); }); CHECK_EQ(status, 0); } +void ThreadPoolWork::PublishDiagnostics(diagnostics_channel::Channel& channel) { + if (!env_->can_call_into_js() || !channel.HasSubscribers()) return; + + v8::Isolate* isolate = env_->isolate(); + v8::HandleScope handle_scope(isolate); + v8::Local context = env_->context(); + + // Match performance.now() without losing precision. + auto to_milliseconds = [isolate, origin = env_->time_origin()]( + uint64_t mark) -> v8::Local { + if (mark == 0) return v8::Null(isolate); + return v8::Number::New(isolate, static_cast(mark - origin) / 1e6); + }; + + v8::Local tmpl = env_->threadpool_work_template(); + if (tmpl.IsEmpty()) { + static constexpr std::string_view names[] = { + "type", "enqueued", "started", "ended"}; + tmpl = v8::DictionaryTemplate::New(isolate, names); + env_->set_threadpool_work_template(tmpl); + } + + v8::MaybeLocal values[] = { + OneByteString(isolate, type_, -1, v8::NewStringType::kInternalized), + to_milliseconds(enqueued_at_), + to_milliseconds(work_start_), + to_milliseconds(work_end_), + }; + + v8::Local message; + if (!NewDictionaryInstance(context, tmpl, values).ToLocal(&message)) return; + channel.Publish(env_, message); +} + int ThreadPoolWork::CancelWork() { return uv_cancel(reinterpret_cast(&work_req_)); } diff --git a/test/node-api/test_async/test-loop.js b/test/node-api/test_async/test-loop.js index 1b05cbad4a50..ae9ab07ca84e 100644 --- a/test/node-api/test_async/test-loop.js +++ b/test/node-api/test_async/test-loop.js @@ -1,14 +1,24 @@ 'use strict'; const common = require('../../common'); const assert = require('assert'); +const dc = require('diagnostics_channel'); const test_async = require(`./build/${common.buildType}/test_async`); const iterations = 500; let x = 0; +const events = []; +const onThreadPoolWork = (event) => events.push(event); +const napiEvents = () => events.filter((event) => event.type === 'node_api'); const workDone = common.mustCall((status) => { assert.strictEqual(status, 0); + if (x === 0) assert.strictEqual(napiEvents().length, 0); if (++x < iterations) { setImmediate(() => test_async.DoRepeatedWork(workDone)); + } else { + dc.unsubscribe('threadpool.work', onThreadPoolWork); + assert.strictEqual(napiEvents().length, iterations - 1); } }, iterations); +// Subscribe after submission to verify subscription latching. test_async.DoRepeatedWork(workDone); +dc.subscribe('threadpool.work', onThreadPoolWork); diff --git a/test/node-api/test_async/test.js b/test/node-api/test_async/test.js index 34ecae08e67f..1d761b0f733b 100644 --- a/test/node-api/test_async/test.js +++ b/test/node-api/test_async/test.js @@ -2,6 +2,7 @@ const common = require('../../common'); const assert = require('assert'); const child_process = require('child_process'); +const dc = require('diagnostics_channel'); const test_async = require(`./build/${common.buildType}/test_async`); const testException = 'test_async_cb_exception'; @@ -27,4 +28,12 @@ test_async.Test(5, {}, common.mustCall(function(err, val) { })); // Async work item cancellation with callback. -test_async.TestCancel(common.mustCall()); +const events = []; +const onThreadPoolWork = (event) => events.push(event); +dc.subscribe('threadpool.work', onThreadPoolWork); +test_async.TestCancel(common.mustCall(() => { + dc.unsubscribe('threadpool.work', onThreadPoolWork); + const event = events.find(({ type, started, ended }) => + type === 'node_api' && started === null && ended === null); + assert.ok(event); +})); diff --git a/test/parallel/test-diagnostics-channel-threadpool-work.js b/test/parallel/test-diagnostics-channel-threadpool-work.js new file mode 100644 index 000000000000..6f94c601f0b8 --- /dev/null +++ b/test/parallel/test-diagnostics-channel-threadpool-work.js @@ -0,0 +1,130 @@ +'use strict'; + +const common = require('../common'); + +const assert = require('node:assert'); +const crypto = common.hasCrypto ? require('node:crypto') : null; +const dc = require('node:diagnostics_channel'); +const fs = require('node:fs'); +const { afterEach, beforeEach, describe, it } = require('node:test'); +const { Worker } = require('node:worker_threads'); +const zlib = require('node:zlib'); + +const tmpdir = require('../common/tmpdir'); +tmpdir.refresh(); + +const CHANNEL = 'threadpool.work'; + +function assertTimestamps(event, now = performance.now()) { + assert.ok(event.enqueued <= event.started); + assert.ok(event.started <= event.ended); + assert.ok(event.ended <= now); +} + +describe('threadpool.work diagnostics channel', () => { + let events; + const handler = (message) => events.push(message); + + beforeEach(() => { + events = []; + dc.subscribe(CHANNEL, handler); + }); + + afterEach(() => dc.unsubscribe(CHANNEL, handler)); + + it('publishes for zlib work', async () => { + await new Promise((resolve, reject) => { + zlib.gzip(Buffer.alloc(64 * 1024, 'x'), (err) => { + if (err) reject(err); + else resolve(); + }); + }); + + assert.ok(events.some((event) => event.type === 'zlib')); + }); + + it('publishes for crypto work', { skip: !common.hasCrypto }, async () => { + await new Promise((resolve, reject) => { + crypto.pbkdf2('secret', 'salt', 100, 32, 'sha256', (err) => { + if (err) reject(err); + else resolve(); + }); + }); + + const event = events.find((event) => event.type === 'crypto'); + assert.ok(event); + assert.deepStrictEqual(Object.keys(event), + ['type', 'enqueued', 'started', 'ended']); + assertTimestamps(event); + }); + + it('publishes for fs work', async () => { + await Promise.all([ + fs.promises.readFile(__filename), + fs.promises.writeFile(tmpdir.resolve('threadpool-work'), 'data'), + ]); + + assert.ok(events.some((event) => event.type === 'fs.readfile')); + assert.ok(events.some((event) => event.type === 'fs.writefile')); + }); + + it('publishes in worker threads', async () => { + const result = await new Promise((resolve, reject) => { + const worker = new Worker(` + const dc = require('node:diagnostics_channel'); + const { parentPort } = require('node:worker_threads'); + const zlib = require('node:zlib'); + const events = []; + const handler = (event) => events.push(event); + dc.subscribe('threadpool.work', handler); + zlib.gzip('data', (err) => { + if (err) throw err; + dc.unsubscribe('threadpool.work', handler); + parentPort.postMessage({ + event: events.find((event) => event.type === 'zlib'), + now: performance.now(), + }); + }); + `, { eval: true }); + worker.once('message', resolve); + worker.once('error', reject); + }); + + assert.ok(result.event); + assertTimestamps(result.event, result.now); + }); + + it('stays monotonic across repeated submissions of one instance', async () => { + await new Promise((resolve, reject) => { + const gzip = zlib.createGzip(); + gzip.once('error', reject); + gzip.on('end', resolve); + gzip.resume(); + for (let i = 0; i < 20; i++) gzip.write(Buffer.alloc(8192, i)); + gzip.end(); + }); + + const zlibEvents = events.filter((event) => event.type === 'zlib'); + assert.ok(zlibEvents.length > 1); + for (let i = 0; i < zlibEvents.length; i++) { + assertTimestamps(zlibEvents[i]); + if (i > 0) { + assert.ok(zlibEvents[i - 1].ended <= zlibEvents[i].enqueued); + } + } + }); + + it('stops publishing after unsubscribe', async () => { + const pending = new Promise((resolve, reject) => { + zlib.gzip(Buffer.alloc(64 * 1024, 'x'), (err) => { + if (err) reject(err); + else resolve(); + }); + }); + + dc.unsubscribe(CHANNEL, handler); + await pending; + assert.ok(!events.some((event) => event.type === 'zlib')); + }); + +});