diagnostics_channel: add threadpool work event

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);
+      }
+    });
+  }
+}

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
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> 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

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<diagnostics_channel::Channel> channel =
+      diagnostics_channel::Channel::Get(this, "threadpool.work");
+  if (!channel) return;
+
+  auto* binding =
+      principal_realm()->GetBindingData<diagnostics_channel::BindingData>();
+  CHECK_NOT_NULL(binding);
+  binding->SetChannelStatusCallback(channel->index(), [this](bool active) {
+    threadpool_work_channel_active_ = active;
+  });
+  threadpool_work_channel_ =
+      BaseObjectWeakPtr<diagnostics_channel::Channel>(channel.get());
+  threadpool_work_channel_active_ = channel->HasSubscribers();
+}
+
 void Environment::InitializeLibuv() {
   HandleScope handle_scope(isolate());
   Context::Scope context_scope(context());

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<diagnostics_channel::Channel>&
+  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<diagnostics_channel::Channel> threadpool_work_channel_;
+  bool threadpool_work_channel_active_ = false;
   const uint64_t timer_base_;
   std::shared_ptr<KVStore> env_vars_;
   bool printed_error_ = false;

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)                               \

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<Value>& args) {
       }
     }
   }
+  realm->env()->InitializeThreadPoolWorkChannel();
 }
 
 bool BindingData::PrepareForSerialization(Local<Context> context,
@@ -102,6 +104,7 @@ bool BindingData::PrepareForSerialization(Local<Context> context,
   internal_field_info_ = InternalFieldInfoBase::New<InternalFieldInfo>(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();

src/node_diagnostics_channel.h

@@ -85,6 +85,8 @@ class Channel : public BaseObject {
 
   static BaseObjectPtr<Channel> 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;
   }

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

src/threadpoolwork-inl.h

@@ -24,6 +24,9 @@
 
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
+#include <string_view>
+#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<v8::Context> context = env_->context();
+
+  // Match performance.now() without losing precision.
+  auto to_milliseconds = [isolate, origin = env_->time_origin()](
+                             uint64_t mark) -> v8::Local<v8::Value> {
+    if (mark == 0) return v8::Null(isolate);
+    return v8::Number::New(isolate, static_cast<double>(mark - origin) / 1e6);
+  };
+
+  v8::Local<v8::DictionaryTemplate> 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<v8::Value> values[] = {
+      OneByteString(isolate, type_, -1, v8::NewStringType::kInternalized),
+      to_milliseconds(enqueued_at_),
+      to_milliseconds(work_start_),
+      to_milliseconds(work_end_),
+  };
+
+  v8::Local<v8::Object> message;
+  if (!NewDictionaryInstance(context, tmpl, values).ToLocal(&message)) return;
+  channel.Publish(env_, message);
+}
+
 int ThreadPoolWork::CancelWork() {
   return uv_cancel(reinterpret_cast<uv_req_t*>(&work_req_));
 }

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);

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);
+}));