diagnostics_channel: add threadpool work event

benchmark/diagnostics_channel/threadpool-work.js

@@ -0,0 +1,69 @@
+'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 = () => {};
+  const type = {
+    crypto: 'crypto',
+    zlib: 'zlib',
+    readFile: 'fs.readfile',
+    writeFile: 'fs.writefile',
+  }[operation];
+  assert(type);
+  const channel = `threadpool.work.${type}`;
+  if (mode === 'subscribed') {
+    dc.subscribe(channel, subscriber);
+  } else if (mode === 'unsubscribed') {
+    dc.subscribe(channel, subscriber);
+    dc.unsubscribe(channel, 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(channel, subscriber);
+      }
+    });
+  }
+}

doc/api/diagnostics_channel.md

@@ -2026,6 +2026,53 @@ 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>'`
+
+Each kind of thread pool work has a separate channel named by appending its
+`type` to `'threadpool.work.'`.
+
+Supported values for `<type>` are:
+
+* `'crypto'`
+
+* `'fs.cp'`
+
+* `'fs.readfile'`
+
+* `'fs.writefile'`
+
+* `'glob'`
+
+* `'histogram.qrde'`
+
+* `'node_api'`
+
+* `'readdir_recursive'`
+
+* `'node_sqlite3.BackupJob'`
+
+* `'zlib'`
+
+Messages contain:
+
+* `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 +2098,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/crypto/crypto_util.h

@@ -343,7 +343,7 @@ class CryptoJob : public AsyncWrap, public ThreadPoolWork {
                      CryptoJobMode mode,
                      AdditionalParams&& params)
       : AsyncWrap(env, object, type),
-        ThreadPoolWork(env, "crypto"),
+        ThreadPoolWork(env, ThreadPoolWorkType::kCrypto),
         mode_(mode),
         params_(std::move(params)) {
     // If the CryptoJob is async, then the instance will be

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_dotenv.h"
 #include "node_errors.h"
 #include "node_file_utils.h"
@@ -1311,6 +1312,30 @@ Environment::~Environment() {
   isolate_data_->RemoveEnvironment();
 }
 
+void Environment::InitializeThreadPoolWorkChannels() {
+  if (isolate_data()->is_building_snapshot()) return;
+
+  auto* binding =
+      principal_realm()->GetBindingData<diagnostics_channel::BindingData>();
+  CHECK_NOT_NULL(binding);
+  for (size_t i = 0; i < threadpool_work_channels_.size(); i++) {
+    ThreadPoolWorkChannel* entry = &threadpool_work_channels_[i];
+    if (entry->channel.get() != nullptr) continue;
+
+    std::string name = "threadpool.work.";
+    name += kThreadPoolWorkNames[i];
+    BaseObjectPtr<diagnostics_channel::Channel> channel =
+        diagnostics_channel::Channel::Get(this, name);
+    if (!channel) continue;
+
+    binding->SetChannelStatusCallback(
+        channel->index(), [entry](bool active) { entry->active = active; });
+    entry->channel =
+        BaseObjectWeakPtr<diagnostics_channel::Channel>(channel.get());
+    entry->active = channel->HasSubscribers();
+  }
+}
+
 void Environment::InitializeLibuv() {
   HandleScope handle_scope(isolate());
   Context::Scope context_scope(context());

src/env.h

@@ -63,6 +63,7 @@
 #include <ostream>
 #include <set>
 #include <string>
+#include <string_view>
 #include <unordered_map>
 #include <unordered_set>
 #include <variant>
@@ -76,6 +77,42 @@ class MacCache;
 
 namespace node {
 
+namespace diagnostics_channel {
+class Channel;
+}
+
+struct ThreadPoolWorkChannel {
+  BaseObjectWeakPtr<diagnostics_channel::Channel> channel;
+  bool active = false;
+};
+
+enum class ThreadPoolWorkType : uint8_t {
+  kCrypto,
+  kFsCp,
+  kFsReadFile,
+  kFsWriteFile,
+  kGlob,
+  kHistogramQrde,
+  kNodeApi,
+  kReadDirRecursive,
+  kSQLiteBackup,
+  kZlib,
+  kCount,
+};
+
+inline constexpr std::array<std::string_view,
+                            static_cast<size_t>(ThreadPoolWorkType::kCount)>
+    kThreadPoolWorkNames = {"crypto",
+                            "fs.cp",
+                            "fs.readfile",
+                            "fs.writefile",
+                            "glob",
+                            "histogram.qrde",
+                            "node_api",
+                            "readdir_recursive",
+                            "node_sqlite3.BackupJob",
+                            "zlib"};
+
 namespace shadow_realm {
 class ShadowRealm;
 }
@@ -737,6 +774,11 @@ class Environment final : public MemoryRetainer {
   void RunDeserializeRequests();
   // Should be called before InitializeInspector()
   void InitializeDiagnostics();
+  void InitializeThreadPoolWorkChannels();
+  inline ThreadPoolWorkChannel* threadpool_work_channel(
+      ThreadPoolWorkType type) {
+    return &threadpool_work_channels_[static_cast<size_t>(type)];
+  }
 
 #if HAVE_INSPECTOR
   // If the environment is created for a worker, pass parent_handle and
@@ -1254,6 +1296,9 @@ class Environment final : public MemoryRetainer {
   AliasedInt32Array timeout_info_;
   TickInfo tick_info_;
   permission::Permission permission_;
+  std::array<ThreadPoolWorkChannel,
+             static_cast<size_t>(ThreadPoolWorkType::kCount)>
+      threadpool_work_channels_;
   const uint64_t timer_base_;
   std::shared_ptr<KVStore> env_vars_;
   int ipc_channel_fd_ = -1;

src/env_properties.h

@@ -459,6 +459,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/glob/node_glob.cc

@@ -431,7 +431,7 @@ GlobRequest::GlobRequest(Environment* env,
                          const std::vector<CompiledPatternPtr>& excludes,
                          std::unique_ptr<JsExcludeFilter> filter)
     : AsyncWrap(env, object, AsyncWrap::PROVIDER_GLOBREQUEST),
-      ThreadPoolWork(env, "glob"),
+      ThreadPoolWork(env, ThreadPoolWorkType::kGlob),
       filter_(std::move(filter)),
       walk_(env, options, includes, excludes),
       with_file_types_(options.with_file_types) {

src/histogram.cc

@@ -970,7 +970,7 @@ class QrdeJob final : public ThreadPoolWork {
           Histogram::RecordedSnapshotSource snapshot_source,
           std::vector<double> probabilities,
           QrdeDequantization dequantization)
-      : ThreadPoolWork(env, "histogram.qrde"),
+      : ThreadPoolWork(env, ThreadPoolWorkType::kHistogramQrde),
         histogram_(std::move(histogram)),
         snapshot_source_(std::move(snapshot_source)),
         probabilities_(std::move(probabilities)),

src/node_api.cc

@@ -1211,7 +1211,7 @@ class Work : public node::AsyncResource, public node::ThreadPoolWork {
             env->isolate,
             async_resource,
             node::Utf8Value(env->isolate, async_resource_name).ToStringView()),
-        ThreadPoolWork(env->node_env(), "node_api"),
+        ThreadPoolWork(env->node_env(), node::ThreadPoolWorkType::kNodeApi),
         _env(env),
         _data(data),
         _execute(execute),

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()->InitializeThreadPoolWorkChannels();
 }
 
 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_file.cc

@@ -2732,7 +2732,7 @@ class ReadDirRecursiveWork final : public ThreadPoolWork {
  public:
   ReadDirRecursiveWork(Environment* env,
                        std::shared_ptr<ReadDirRecursiveRequest> request)
-      : ThreadPoolWork(env, "readdir_recursive"),
+      : ThreadPoolWork(env, ThreadPoolWorkType::kReadDirRecursive),
         request_(std::move(request)) {}
 
   void DoThreadPoolWork() override { request_->walk()->Run(); }
@@ -3777,7 +3777,7 @@ class ReadFileJob final : public AsyncWrap, public ThreadPoolWork {
               uint64_t limit,
               bool track_fd)
       : AsyncWrap(env, object, AsyncWrap::PROVIDER_FSREQCALLBACK),
-        ThreadPoolWork(env, "fs.readfile"),
+        ThreadPoolWork(env, ThreadPoolWorkType::kFsReadFile),
         path_(std::move(path)),
         limit_(limit),
         flags_(flags),
@@ -3965,7 +3965,7 @@ class WriteFileJob final : public AsyncWrap, public ThreadPoolWork {
                int mode,
                Local<ArrayBufferView> view)
       : AsyncWrap(env, object, AsyncWrap::PROVIDER_FSREQCALLBACK),
-        ThreadPoolWork(env, "fs.writefile"),
+        ThreadPoolWork(env, ThreadPoolWorkType::kFsWriteFile),
         path_(std::move(path)),
         flags_(flags),
         mode_(mode) {
@@ -5299,7 +5299,7 @@ class CpDirJob final : public AsyncWrap, public ThreadPoolWork {
            std::string&& dest_display,
            CpDirOptions options)
       : AsyncWrap(env, object, AsyncWrap::PROVIDER_FSREQCALLBACK),
-        ThreadPoolWork(env, "fs.cp"),
+        ThreadPoolWork(env, ThreadPoolWorkType::kFsCp),
         src_(std::move(src)),
         dest_(std::move(dest)),
         dest_display_(std::move(dest_display)),