process: add threadCpuUsage

lib/internal/bootstrap/node.js

@@ -172,6 +172,7 @@ const rawMethods = internalBinding('process_methods');
   process.loadEnvFile = wrapped.loadEnvFile;
   process._rawDebug = wrapped._rawDebug;
   process.cpuUsage = wrapped.cpuUsage;
+  process.threadCpuUsage = wrapped.threadCpuUsage;
   process.resourceUsage = wrapped.resourceUsage;
   process.memoryUsage = wrapped.memoryUsage;
   process.constrainedMemory = rawMethods.constrainedMemory;

lib/internal/process/per_thread.js

@@ -97,6 +97,7 @@ function nop() {}
 function wrapProcessMethods(binding) {
   const {
     cpuUsage: _cpuUsage,
+    threadCpuUsage: _threadCpuUsage,
     memoryUsage: _memoryUsage,
     rss,
     resourceUsage: _resourceUsage,
@@ -148,6 +149,46 @@ function wrapProcessMethods(binding) {
     };
   }
 
+  const threadCpuValues = new Float64Array(2);
+
+  // Replace the native function with the JS version that calls the native
+  // function.
+  function threadCpuUsage(prevValue) {
+    // If a previous value was passed in, ensure it has the correct shape.
+    if (prevValue) {
+      if (!previousValueIsValid(prevValue.user)) {
+        validateObject(prevValue, 'prevValue');
+
+        validateNumber(prevValue.user, 'prevValue.user');
+        throw new ERR_INVALID_ARG_VALUE.RangeError('prevValue.user',
+                                                   prevValue.user);
+      }
+
+      if (!previousValueIsValid(prevValue.system)) {
+        validateNumber(prevValue.system, 'prevValue.system');
+        throw new ERR_INVALID_ARG_VALUE.RangeError('prevValue.system',
+                                                   prevValue.system);
+      }
+    }
+
+    // Call the native function to get the current values.
+    _threadCpuUsage(threadCpuValues);
+
+    // If a previous value was passed in, return diff of current from previous.
+    if (prevValue) {
+      return {
+        user: threadCpuValues[0] - prevValue.user,
+        system: threadCpuValues[1] - prevValue.system,
+      };
+    }
+
+    // If no previous value passed in, return current value.
+    return {
+      user: threadCpuValues[0],
+      system: threadCpuValues[1],
+    };
+  }
+
   // Ensure that a previously passed in value is valid. Currently, the native
   // implementation always returns numbers <= Number.MAX_SAFE_INTEGER.
   function previousValueIsValid(num) {
@@ -263,6 +304,7 @@ function wrapProcessMethods(binding) {
   return {
     _rawDebug,
     cpuUsage,
+    threadCpuUsage,
     resourceUsage,
     memoryUsage,
     kill,

src/node_errors.h

@@ -111,7 +111,9 @@ void OOMErrorHandler(const char* location, const v8::OOMDetails& details);
   V(ERR_WASI_NOT_STARTED, Error)                                               \
   V(ERR_ZLIB_INITIALIZATION_FAILED, Error)                                     \
   V(ERR_WORKER_INIT_FAILED, Error)                                             \
-  V(ERR_PROTO_ACCESS, Error)
+  V(ERR_PROTO_ACCESS, Error)                                                   \
+  V(ERR_THREAD_CPU_USAGE_UNSUPPORTED, Error)                                   \
+  V(ERR_THREAD_CPU_USAGE_FAILED, Error)
 
 #define V(code, type)                                                          \
   template <typename... Args>                                                  \
@@ -214,7 +216,10 @@ ERRORS_WITH_CODE(V)
   V(ERR_WORKER_INIT_FAILED, "Worker initialization failure")                   \
   V(ERR_PROTO_ACCESS,                                                          \
     "Accessing Object.prototype.__proto__ has been "                           \
-    "disallowed with --disable-proto=throw")
+    "disallowed with --disable-proto=throw")                                   \
+  V(ERR_THREAD_CPU_USAGE_UNSUPPORTED,                                          \
+    "process.threadCpuUsage is not available on this platform")                \
+  V(ERR_THREAD_CPU_USAGE_FAILED, "Failed to get thread CPU usage")
 
 #define V(code, message)                                                       \
   inline v8::Local<v8::Object> code(v8::Isolate* isolate) {                    \

src/node_process_methods.cc

@@ -130,6 +130,71 @@ static void CPUUsage(const FunctionCallbackInfo<Value>& args) {
   fields[1] = MICROS_PER_SEC * rusage.ru_stime.tv_sec + rusage.ru_stime.tv_usec;
 }
 
+// ThreadCPUUsage use system dependent implementation to get the current thread
+// cpu times:
+//
+// - On Unix, it uses getrusage(2) with RUSAGE_THREAD.
+// - On macOS, it uses thread_info(2) with THREAD_BASIC_INFO.
+// - On Windows, it uses GetThreadTimes(2).
+//
+// Returns those values as Float64 microseconds in the elements of the array
+// passed to the function.
+static void ThreadCPUUsage(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+
+  // Get the double array pointer from the Float64Array argument.
+  Local<ArrayBuffer> ab = get_fields_array_buffer(args, 0, 2);
+  double* fields = static_cast<double*>(ab->Data());
+
+#ifdef __linux__
+  clockid_t clockid;
+  pthread_getcpuclockid(pthread_self(), &clockid);
+
+  struct rusage usage;
+
+  if (getrusage(RUSAGE_THREAD, &usage) == -1) {
+    return THROW_ERR_THREAD_CPU_USAGE_FAILED(env);
+  }
+
+  fields[0] = MICROS_PER_SEC * usage.ru_utime.tv_sec + usage.ru_utime.tv_usec;
+  fields[1] = MICROS_PER_SEC * usage.ru_stime.tv_sec + usage.ru_stime.tv_usec;
+#elif __APPLE__
+  mach_port_t thread = mach_thread_self();
+  mach_msg_type_number_t count = THREAD_BASIC_INFO_COUNT;
+  thread_basic_info_data_t usage;
+  kern_return_t kr =
+      thread_info(thread, THREAD_BASIC_INFO, (thread_info_t)&usage, &count);
+
+  if (kr != KERN_SUCCESS) {
+    return THROW_ERR_THREAD_CPU_USAGE_FAILED(env);
+  }
+
+  fields[0] =
+      MICROS_PER_SEC * usage.user_time.seconds + usage.user_time.microseconds;
+  fields[1] = MICROS_PER_SEC * usage.system_time.seconds +
+              usage.system_time.microseconds;
+#elif _WIN32
+  HANDLE thread_ = GetCurrentThread();
+  FILETIME _creation_time, _exit_time, kernel_time, user_time;
+
+  if (!GetThreadTimes(
+          thread_, &_creation_time, &_exit_time, &kernel_time, &user_time)) {
+    return THROW_ERR_THREAD_CPU_USAGE_FAILED(env);
+  }
+
+  fields[0] = ((((uint64_t)user_time.dwHighDateTime << 32) |
+                (uint64_t)user_time.dwLowDateTime) *
+               100) /
+              1000;
+  fields[1] = ((((uint64_t)system_ns.dwHighDateTime << 32) |
+                (uint64_t)system_ns.dwLowDateTime) *
+               100) /
+              1000;
+#else
+  return THROW_ERR_THREAD_CPU_USAGE_UNSUPPORTED(env);
+#endif
+}
+
 static void Cwd(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   CHECK(env->has_run_bootstrapping_code());
@@ -650,6 +715,7 @@ static void CreatePerIsolateProperties(IsolateData* isolate_data,
   SetMethod(isolate, target, "availableMemory", GetAvailableMemory);
   SetMethod(isolate, target, "rss", Rss);
   SetMethod(isolate, target, "cpuUsage", CPUUsage);
+  SetMethod(isolate, target, "threadCpuUsage", ThreadCPUUsage);
   SetMethod(isolate, target, "resourceUsage", ResourceUsage);
 
   SetMethod(isolate, target, "_debugEnd", DebugEnd);
@@ -694,6 +760,7 @@ void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(GetAvailableMemory);
   registry->Register(Rss);
   registry->Register(CPUUsage);
+  registry->Register(ThreadCPUUsage);
   registry->Register(ResourceUsage);
 
   registry->Register(GetActiveRequests);

test/parallel/test-process-threadCpuUsage-main-thread.js

@@ -0,0 +1,76 @@
+'use strict';
+
+require('../common');
+
+const { ok, throws, notStrictEqual } = require('assert');
+
+function validateResult(result) {
+  notStrictEqual(result, null);
+
+  ok(Number.isFinite(result.user));
+  ok(Number.isFinite(result.system));
+
+  ok(result.user >= 0);
+  ok(result.system >= 0);
+}
+
+// Test that process.threadCpuUsage() works on the main thread
+{
+  const result = process.threadCpuUsage();
+
+  // Validate the result of calling with no previous value argument.
+  validateResult(process.threadCpuUsage());
+
+  // Validate the result of calling with a previous value argument.
+  validateResult(process.threadCpuUsage(result));
+
+  // Ensure the results are >= the previous.
+  let thisUsage;
+  let lastUsage = process.threadCpuUsage();
+  for (let i = 0; i < 10; i++) {
+    thisUsage = process.threadCpuUsage();
+    validateResult(thisUsage);
+    ok(thisUsage.user >= lastUsage.user);
+    ok(thisUsage.system >= lastUsage.system);
+    lastUsage = thisUsage;
+  }
+}
+
+// Test argument validaton
+{
+  throws(
+    () => process.threadCpuUsage(123),
+    {
+      code: 'ERR_INVALID_ARG_TYPE',
+      name: 'TypeError',
+      message: 'The "prevValue" argument must be of type object. Received type number (123)'
+    }
+  );
+
+  throws(
+    () => process.threadCpuUsage([]),
+    {
+      code: 'ERR_INVALID_ARG_TYPE',
+      name: 'TypeError',
+      message: 'The "prevValue" argument must be of type object. Received an instance of Array'
+    }
+  );
+
+  throws(
+    () => process.threadCpuUsage({ user: -123 }),
+    {
+      code: 'ERR_INVALID_ARG_VALUE',
+      name: 'RangeError',
+      message: "The property 'prevValue.user' is invalid. Received -123"
+    }
+  );
+
+  throws(
+    () => process.threadCpuUsage({ user: 0, system: 'bar' }),
+    {
+      code: 'ERR_INVALID_ARG_TYPE',
+      name: 'TypeError',
+      message: "The \"prevValue.system\" property must be of type number. Received type string ('bar')"
+    }
+  );
+}

test/parallel/test-process-threadCpuUsage-worker-threads.js

@@ -0,0 +1,90 @@
+'use strict';
+
+const { mustCall, platformTimeout, hasCrypto, skip } = require('../common');
+const { ok, deepStrictEqual } = require('assert');
+const { randomBytes, createHash } = require('crypto');
+const { once } = require('events');
+const { Worker, isMainThread, parentPort, threadId } = require('worker_threads');
+
+if (!hasCrypto) {
+  skip('missing crypto');
+};
+
+function performLoad() {
+  const buffer = randomBytes(1e8);
+  const index = threadId + 1;
+
+  // Do some work
+  return setInterval(() => {
+    createHash('sha256').update(buffer).end(buffer);
+  }, platformTimeout(index ** 2 * 100));
+}
+
+function getUsages() {
+  return { threadId, process: process.cpuUsage(), thread: process.threadCpuUsage() };
+}
+
+function validateResults(results) {
+  for (let i = 0; i < 4; i++) {
+    deepStrictEqual(results[i].threadId, i);
+  }
+
+  for (let i = 0; i < 3; i++) {
+    const processDifference = results[i].process.user / results[i + 1].process.user;
+    const threadDifference = results[i].thread.user / results[i + 1].thread.user;
+
+    //
+    // All process CPU usages should be the same. Technically they should have returned the same
+    // value but since we measure it at different times they vary a little bit.
+    // Let's allow a tolerance of 5%
+    //
+    ok(processDifference > 0.95);
+    ok(processDifference < 1.05);
+
+    //
+    // Each thread is configured so that the performLoad schedules a new hash with an interval two times bigger of the
+    // previous thread. In theory this should give each thread a load about half of the previous one.
+    // But since we can't really predict CPU scheduling, we just verify a minimum difference of 20% and a monotonic
+    // increasing sequence.
+    //
+    ok(threadDifference > 1.2);
+  }
+}
+
+
+// The main thread will spawn three more threads, then after a while it will ask all of them to
+// report the thread CPU usage and exit.
+if (isMainThread) {
+  const workers = [];
+  for (let i = 0; i < 3; i++) {
+    workers.push(new Worker(__filename));
+  }
+
+  setTimeout(mustCall(async () => {
+    clearInterval(interval);
+
+    const results = [getUsages()];
+
+    for (const worker of workers) {
+      const statusPromise = once(worker, 'message');
+      const exitPromise = once(worker, 'exit');
+
+      worker.postMessage('done');
+      const [status] = await statusPromise;
+      results.push(status);
+      await exitPromise;
+    }
+
+    validateResults(results);
+  }), platformTimeout(5000));
+
+} else {
+  parentPort.on('message', () => {
+    clearInterval(interval);
+    parentPort.postMessage(getUsages());
+    process.exit(0);
+  });
+}
+
+// Perform load on each thread
+const interval = performLoad();