src: trace threadpool event

PR-URL: #44458
Reviewed-By: Chengzhong Wu <[email protected]>

Author: theanarkh

doc/api/tracing.md

@@ -22,6 +22,10 @@ The available categories are:
 * `node.bootstrap`: Enables capture of Node.js bootstrap milestones.
 * `node.console`: Enables capture of `console.time()` and `console.count()`
   output.
+* `node.threadpoolwork.sync`: Enables capture of trace data for threadpool
+  synchronous operations, such as `blob`, `zlib`, `crypto` and `node_api`.
+* `node.threadpoolwork.async`: Enables capture of trace data for threadpool
+  asynchronous operations, such as `blob`, `zlib`, `crypto` and `node_api`.
 * `node.dns.native`: Enables capture of trace data for DNS queries.
 * `node.net.native`: Enables capture of trace data for network.
 * `node.environment`: Enables capture of Node.js Environment milestones.

src/crypto/crypto_util.h

@@ -333,14 +333,13 @@ class CryptoJob : public AsyncWrap, public ThreadPoolWork {
  public:
   using AdditionalParams = typename CryptoJobTraits::AdditionalParameters;
 
-  explicit CryptoJob(
-      Environment* env,
-      v8::Local<v8::Object> object,
-      AsyncWrap::ProviderType type,
-      CryptoJobMode mode,
-      AdditionalParams&& params)
+  explicit CryptoJob(Environment* env,
+                     v8::Local<v8::Object> object,
+                     AsyncWrap::ProviderType type,
+                     CryptoJobMode mode,
+                     AdditionalParams&& params)
       : AsyncWrap(env, object, type),
-        ThreadPoolWork(env),
+        ThreadPoolWork(env, "crypto"),
         mode_(mode),
         params_(std::move(params)) {
     // If the CryptoJob is async, then the instance will be

src/node_api.cc

@@ -1069,7 +1069,7 @@ class Work : public node::AsyncResource, public node::ThreadPoolWork {
             env->isolate,
             async_resource,
             *v8::String::Utf8Value(env->isolate, async_resource_name)),
-        ThreadPoolWork(env->node_env()),
+        ThreadPoolWork(env->node_env(), "node_api"),
         _env(env),
         _data(data),
         _execute(execute),

src/node_blob.cc

@@ -306,13 +306,12 @@ void Blob::GetDataObject(const v8::FunctionCallbackInfo<v8::Value>& args) {
   }
 }
 
-FixedSizeBlobCopyJob::FixedSizeBlobCopyJob(
-    Environment* env,
-    Local<Object> object,
-    Blob* blob,
-    FixedSizeBlobCopyJob::Mode mode)
+FixedSizeBlobCopyJob::FixedSizeBlobCopyJob(Environment* env,
+                                           Local<Object> object,
+                                           Blob* blob,
+                                           FixedSizeBlobCopyJob::Mode mode)
     : AsyncWrap(env, object, AsyncWrap::PROVIDER_FIXEDSIZEBLOBCOPY),
-      ThreadPoolWork(env),
+      ThreadPoolWork(env, "blob"),
       mode_(mode) {
   if (mode == FixedSizeBlobCopyJob::Mode::SYNC) MakeWeak();
   source_ = blob->entries();

src/node_internals.h

@@ -258,7 +258,8 @@ class DebugSealHandleScope {
 
 class ThreadPoolWork {
  public:
-  explicit inline ThreadPoolWork(Environment* env) : env_(env) {
+  explicit inline ThreadPoolWork(Environment* env, const char* type)
+      : env_(env), type_(type) {
     CHECK_NOT_NULL(env);
   }
   inline virtual ~ThreadPoolWork() = default;
@@ -274,6 +275,7 @@ class ThreadPoolWork {
  private:
   Environment* env_;
   uv_work_t work_req_;
+  const char* type_;
 };
 
 #define TRACING_CATEGORY_NODE "node"

src/node_v8_platform-inl.h

@@ -109,14 +109,15 @@ struct V8Platform {
     if (!initialized_)
       return;
     initialized_ = false;
-
+    node::tracing::TraceEventHelper::SetAgent(nullptr);
     StopTracingAgent();
     platform_->Shutdown();
     delete platform_;
     platform_ = nullptr;
     // Destroy tracing after the platform (and platform threads) have been
     // stopped.
     tracing_agent_.reset(nullptr);
+    // The observer remove itself in OnTraceEnabled
     trace_state_observer_.reset(nullptr);
   }
 

src/node_zlib.cc

@@ -259,7 +259,7 @@ class CompressionStream : public AsyncWrap, public ThreadPoolWork {
 
   CompressionStream(Environment* env, Local<Object> wrap)
       : AsyncWrap(env, wrap, AsyncWrap::PROVIDER_ZLIB),
-        ThreadPoolWork(env),
+        ThreadPoolWork(env, "zlib"),
         write_result_(nullptr) {
     MakeWeak();
   }

src/threadpoolwork-inl.h

@@ -24,23 +24,36 @@
 
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
-#include "util-inl.h"
 #include "node_internals.h"
+#include "tracing/trace_event.h"
+#include "util-inl.h"
 
 namespace node {
 
 void ThreadPoolWork::ScheduleWork() {
   env_->IncreaseWaitingRequestCounter();
+  TRACE_EVENT_NESTABLE_ASYNC_BEGIN0(
+      TRACING_CATEGORY_NODE2(threadpoolwork, async), type_, this);
   int status = uv_queue_work(
       env_->event_loop(),
       &work_req_,
       [](uv_work_t* req) {
         ThreadPoolWork* self = ContainerOf(&ThreadPoolWork::work_req_, req);
+        TRACE_EVENT_BEGIN0(TRACING_CATEGORY_NODE2(threadpoolwork, sync),
+                           self->type_);
         self->DoThreadPoolWork();
+        TRACE_EVENT_END0(TRACING_CATEGORY_NODE2(threadpoolwork, sync),
+                         self->type_);
       },
       [](uv_work_t* req, int status) {
         ThreadPoolWork* self = ContainerOf(&ThreadPoolWork::work_req_, req);
         self->env_->DecreaseWaitingRequestCounter();
+        TRACE_EVENT_NESTABLE_ASYNC_END1(
+            TRACING_CATEGORY_NODE2(threadpoolwork, async),
+            self->type_,
+            self,
+            "result",
+            status);
         self->AfterThreadPoolWork(status);
       });
   CHECK_EQ(status, 0);

src/tracing/trace_event.cc

@@ -8,8 +8,13 @@ Agent* g_agent = nullptr;
 v8::TracingController* g_controller = nullptr;
 
 void TraceEventHelper::SetAgent(Agent* agent) {
-  g_agent = agent;
-  g_controller = agent->GetTracingController();
+  if (agent) {
+    g_agent = agent;
+    g_controller = agent->GetTracingController();
+  } else {
+    g_agent = nullptr;
+    g_controller = nullptr;
+  }
 }
 
 Agent* TraceEventHelper::GetAgent() {

test/parallel/test-trace-events-threadpool.js

@@ -0,0 +1,71 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const cp = require('child_process');
+const fs = require('fs');
+const path = require('path');
+const tmpdir = require('../common/tmpdir');
+
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+const { hkdf } = require('crypto');
+const { deflate } = require('zlib');
+const { Blob } = require('buffer');
+
+if (process.env.isChild === '1') {
+  hkdf('sha512', 'key', 'salt', 'info', 64, () => {});
+  deflate('hello', () => {});
+  // Make async call
+  const blob = new Blob(['h'.repeat(4096 * 2)]);
+  blob.arrayBuffer();
+  return;
+}
+
+tmpdir.refresh();
+const FILE_NAME = path.join(tmpdir.path, 'node_trace.1.log');
+
+cp.spawnSync(process.execPath,
+             [
+               '--trace-events-enabled',
+               '--trace-event-categories',
+               'node.threadpoolwork.sync,node.threadpoolwork.async',
+               __filename,
+             ],
+             {
+               cwd: tmpdir.path,
+               env: {
+                 ...process.env,
+                 isChild: '1',
+               },
+             });
+
+assert(fs.existsSync(FILE_NAME));
+const data = fs.readFileSync(FILE_NAME);
+const traces = JSON.parse(data.toString()).traceEvents;
+
+assert(traces.length > 0);
+
+let blobCount = 0;
+let zlibCount = 0;
+let cryptoCount = 0;
+
+traces.forEach((item) => {
+  if ([
+    'node,node.threadpoolwork,node.threadpoolwork.sync',
+    'node,node.threadpoolwork,node.threadpoolwork.async',
+  ].includes(item.cat)) {
+    if (item.name === 'blob') {
+      blobCount++;
+    } else if (item.name === 'zlib') {
+      zlibCount++;
+    } else if (item.name === 'crypto') {
+      cryptoCount++;
+    }
+  }
+});
+
+// There are three types, each type has two async events and sync events at least
+assert.ok(blobCount >= 4);
+assert.ok(zlibCount >= 4);
+assert.ok(cryptoCount >= 4);