async_hooks: add executionAsyncResource

Remove the need for the destroy hook in the basic APM case.

Co-authored-by: Stephen Belanger <[email protected]>
PR-URL: #30959
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: Vladimir de Turckheim <[email protected]>
Reviewed-By: Chengzhong Wu <[email protected]>
Reviewed-By: James M Snell <[email protected]>

Author: 2 people

benchmark/async_hooks/async-resource-vs-destroy.js

@@ -0,0 +1,151 @@
+'use strict';
+
+const { promisify } = require('util');
+const { readFile } = require('fs');
+const sleep = promisify(setTimeout);
+const read = promisify(readFile);
+const common = require('../common.js');
+const {
+  createHook,
+  executionAsyncResource,
+  executionAsyncId
+} = require('async_hooks');
+const { createServer } = require('http');
+
+// Configuration for the http server
+// there is no need for parameters in this test
+const connections = 500;
+const path = '/';
+
+const bench = common.createBenchmark(main, {
+  type: ['async-resource', 'destroy'],
+  asyncMethod: ['callbacks', 'async'],
+  n: [1e6]
+});
+
+function buildCurrentResource(getServe) {
+  const server = createServer(getServe(getCLS, setCLS));
+  const hook = createHook({ init });
+  const cls = Symbol('cls');
+  hook.enable();
+
+  return {
+    server,
+    close
+  };
+
+  function getCLS() {
+    const resource = executionAsyncResource();
+    if (resource === null || !resource[cls]) {
+      return null;
+    }
+    return resource[cls].state;
+  }
+
+  function setCLS(state) {
+    const resource = executionAsyncResource();
+    if (resource === null) {
+      return;
+    }
+    if (!resource[cls]) {
+      resource[cls] = { state };
+    } else {
+      resource[cls].state = state;
+    }
+  }
+
+  function init(asyncId, type, triggerAsyncId, resource) {
+    var cr = executionAsyncResource();
+    if (cr !== null) {
+      resource[cls] = cr[cls];
+    }
+  }
+
+  function close() {
+    hook.disable();
+    server.close();
+  }
+}
+
+function buildDestroy(getServe) {
+  const transactions = new Map();
+  const server = createServer(getServe(getCLS, setCLS));
+  const hook = createHook({ init, destroy });
+  hook.enable();
+
+  return {
+    server,
+    close
+  };
+
+  function getCLS() {
+    const asyncId = executionAsyncId();
+    return transactions.has(asyncId) ? transactions.get(asyncId) : null;
+  }
+
+  function setCLS(value) {
+    const asyncId = executionAsyncId();
+    transactions.set(asyncId, value);
+  }
+
+  function init(asyncId, type, triggerAsyncId, resource) {
+    transactions.set(asyncId, getCLS());
+  }
+
+  function destroy(asyncId) {
+    transactions.delete(asyncId);
+  }
+
+  function close() {
+    hook.disable();
+    server.close();
+  }
+}
+
+function getServeAwait(getCLS, setCLS) {
+  return async function serve(req, res) {
+    setCLS(Math.random());
+    await sleep(10);
+    await read(__filename);
+    res.setHeader('content-type', 'application/json');
+    res.end(JSON.stringify({ cls: getCLS() }));
+  };
+}
+
+function getServeCallbacks(getCLS, setCLS) {
+  return function serve(req, res) {
+    setCLS(Math.random());
+    setTimeout(() => {
+      readFile(__filename, () => {
+        res.setHeader('content-type', 'application/json');
+        res.end(JSON.stringify({ cls: getCLS() }));
+      });
+    }, 10);
+  };
+}
+
+const types = {
+  'async-resource': buildCurrentResource,
+  'destroy': buildDestroy
+};
+
+const asyncMethods = {
+  'callbacks': getServeCallbacks,
+  'async': getServeAwait
+};
+
+function main({ type, asyncMethod }) {
+  const { server, close } = types[type](asyncMethods[asyncMethod]);
+
+  server
+    .listen(common.PORT)
+    .on('listening', () => {
+
+      bench.http({
+        path,
+        connections
+      }, () => {
+        close();
+      });
+    });
+}

doc/api/async_hooks.md

@@ -459,6 +459,62 @@ init for PROMISE with id 6, trigger id: 5  # the Promise returned by then()
   after 6
 ```
 
+#### `async_hooks.executionAsyncResource()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Returns: {Object} The resource representing the current execution.
+  Useful to store data within the resource.
+
+Resource objects returned by `executionAsyncResource()` are most often internal
+Node.js handle objects with undocumented APIs. Using any functions or properties
+on the object is likely to crash your application and should be avoided.
+
+Using `executionAsyncResource()` in the top-level execution context will
+return an empty object as there is no handle or request object to use,
+but having an object representing the top-level can be helpful.
+
+```js
+const { open } = require('fs');
+const { executionAsyncId, executionAsyncResource } = require('async_hooks');
+
+console.log(executionAsyncId(), executionAsyncResource());  // 1 {}
+open(__filename, 'r', (err, fd) => {
+  console.log(executionAsyncId(), executionAsyncResource());  // 7 FSReqWrap
+});
+```
+
+This can be used to implement continuation local storage without the
+use of a tracking `Map` to store the metadata:
+
+```js
+const { createServer } = require('http');
+const {
+  executionAsyncId,
+  executionAsyncResource,
+  createHook
+} = require('async_hooks');
+const sym = Symbol('state'); // Private symbol to avoid pollution
+
+createHook({
+  init(asyncId, type, triggerAsyncId, resource) {
+    const cr = executionAsyncResource();
+    if (cr) {
+      resource[sym] = cr[sym];
+    }
+  }
+}).enable();
+
+const server = createServer(function(req, res) {
+  executionAsyncResource()[sym] = { state: req.url };
+  setTimeout(function() {
+    res.end(JSON.stringify(executionAsyncResource()[sym]));
+  }, 100);
+}).listen(3000);
+```
+
 #### `async_hooks.executionAsyncId()`
 
 <!-- YAML

lib/async_hooks.js

@@ -25,6 +25,7 @@ const {
   getHookArrays,
   enableHooks,
   disableHooks,
+  executionAsyncResource,
   // Internal Embedder API
   newAsyncId,
   getDefaultTriggerAsyncId,
@@ -176,7 +177,7 @@ class AsyncResource {
 
   runInAsyncScope(fn, thisArg, ...args) {
     const asyncId = this[async_id_symbol];
-    emitBefore(asyncId, this[trigger_async_id_symbol]);
+    emitBefore(asyncId, this[trigger_async_id_symbol], this);
 
     const ret = thisArg === undefined ?
       fn(...args) :
@@ -211,6 +212,7 @@ module.exports = {
   createHook,
   executionAsyncId,
   triggerAsyncId,
+  executionAsyncResource,
   // Embedder API
   AsyncResource,
 };

lib/internal/async_hooks.js

@@ -28,18 +28,26 @@ const async_wrap = internalBinding('async_wrap');
  *    3. executionAsyncId of the current resource.
  *
  * async_ids_stack is a Float64Array that contains part of the async ID
- * stack. Each pushAsyncIds() call adds two doubles to it, and each
- * popAsyncIds() call removes two doubles from it.
+ * stack. Each pushAsyncContext() call adds two doubles to it, and each
+ * popAsyncContext() call removes two doubles from it.
  * It has a fixed size, so if that is exceeded, calls to the native
- * side are used instead in pushAsyncIds() and popAsyncIds().
+ * side are used instead in pushAsyncContext() and popAsyncContext().
  */
-const { async_hook_fields, async_id_fields, owner_symbol } = async_wrap;
+const {
+  async_hook_fields,
+  async_id_fields,
+  execution_async_resources,
+  owner_symbol
+} = async_wrap;
 // Store the pair executionAsyncId and triggerAsyncId in a std::stack on
 // Environment::AsyncHooks::async_ids_stack_ tracks the resource responsible for
 // the current execution stack. This is unwound as each resource exits. In the
 // case of a fatal exception this stack is emptied after calling each hook's
 // after() callback.
-const { pushAsyncIds: pushAsyncIds_, popAsyncIds: popAsyncIds_ } = async_wrap;
+const {
+  pushAsyncContext: pushAsyncContext_,
+  popAsyncContext: popAsyncContext_
+} = async_wrap;
 // For performance reasons, only track Promises when a hook is enabled.
 const { enablePromiseHook, disablePromiseHook } = async_wrap;
 // Properties in active_hooks are used to keep track of the set of hooks being
@@ -92,6 +100,15 @@ const emitDestroyNative = emitHookFactory(destroy_symbol, 'emitDestroyNative');
 const emitPromiseResolveNative =
     emitHookFactory(promise_resolve_symbol, 'emitPromiseResolveNative');
 
+const topLevelResource = {};
+
+function executionAsyncResource() {
+  const index = async_hook_fields[kStackLength] - 1;
+  if (index === -1) return topLevelResource;
+  const resource = execution_async_resources[index];
+  return resource;
+}
+
 // Used to fatally abort the process if a callback throws.
 function fatalError(e) {
   if (typeof e.stack === 'string') {
@@ -330,8 +347,8 @@ function emitInitScript(asyncId, type, triggerAsyncId, resource) {
 }
 
 
-function emitBeforeScript(asyncId, triggerAsyncId) {
-  pushAsyncIds(asyncId, triggerAsyncId);
+function emitBeforeScript(asyncId, triggerAsyncId, resource) {
+  pushAsyncContext(asyncId, triggerAsyncId, resource);
 
   if (async_hook_fields[kBefore] > 0)
     emitBeforeNative(asyncId);
@@ -342,7 +359,7 @@ function emitAfterScript(asyncId) {
   if (async_hook_fields[kAfter] > 0)
     emitAfterNative(asyncId);
 
-  popAsyncIds(asyncId);
+  popAsyncContext(asyncId);
 }
 
 
@@ -360,6 +377,7 @@ function clearAsyncIdStack() {
   async_id_fields[kExecutionAsyncId] = 0;
   async_id_fields[kTriggerAsyncId] = 0;
   async_hook_fields[kStackLength] = 0;
+  execution_async_resources.splice(0, execution_async_resources.length);
 }
 
 
@@ -369,31 +387,33 @@ function hasAsyncIdStack() {
 
 
 // This is the equivalent of the native push_async_ids() call.
-function pushAsyncIds(asyncId, triggerAsyncId) {
+function pushAsyncContext(asyncId, triggerAsyncId, resource) {
   const offset = async_hook_fields[kStackLength];
   if (offset * 2 >= async_wrap.async_ids_stack.length)
-    return pushAsyncIds_(asyncId, triggerAsyncId);
+    return pushAsyncContext_(asyncId, triggerAsyncId, resource);
   async_wrap.async_ids_stack[offset * 2] = async_id_fields[kExecutionAsyncId];
   async_wrap.async_ids_stack[offset * 2 + 1] = async_id_fields[kTriggerAsyncId];
+  execution_async_resources[offset] = resource;
   async_hook_fields[kStackLength]++;
   async_id_fields[kExecutionAsyncId] = asyncId;
   async_id_fields[kTriggerAsyncId] = triggerAsyncId;
 }
 
 
 // This is the equivalent of the native pop_async_ids() call.
-function popAsyncIds(asyncId) {
+function popAsyncContext(asyncId) {
   const stackLength = async_hook_fields[kStackLength];
   if (stackLength === 0) return false;
 
   if (enabledHooksExist() && async_id_fields[kExecutionAsyncId] !== asyncId) {
     // Do the same thing as the native code (i.e. crash hard).
-    return popAsyncIds_(asyncId);
+    return popAsyncContext_(asyncId);
   }
 
   const offset = stackLength - 1;
   async_id_fields[kExecutionAsyncId] = async_wrap.async_ids_stack[2 * offset];
   async_id_fields[kTriggerAsyncId] = async_wrap.async_ids_stack[2 * offset + 1];
+  execution_async_resources.pop();
   async_hook_fields[kStackLength] = offset;
   return offset > 0;
 }
@@ -426,6 +446,7 @@ module.exports = {
   clearDefaultTriggerAsyncId,
   clearAsyncIdStack,
   hasAsyncIdStack,
+  executionAsyncResource,
   // Internal Embedder API
   newAsyncId,
   getOrSetAsyncId,

lib/internal/process/task_queues.js

@@ -71,7 +71,7 @@ function processTicksAndRejections() {
   do {
     while (tock = queue.shift()) {
       const asyncId = tock[async_id_symbol];
-      emitBefore(asyncId, tock[trigger_async_id_symbol]);
+      emitBefore(asyncId, tock[trigger_async_id_symbol], tock);
 
       try {
         const callback = tock.callback;

lib/internal/timers.js

@@ -96,7 +96,7 @@ const {
   emitInit,
   emitBefore,
   emitAfter,
-  emitDestroy
+  emitDestroy,
 } = require('internal/async_hooks');
 
 // Symbols for storing async id state.
@@ -448,7 +448,7 @@ function getTimerCallbacks(runNextTicks) {
       prevImmediate = immediate;
 
       const asyncId = immediate[async_id_symbol];
-      emitBefore(asyncId, immediate[trigger_async_id_symbol]);
+      emitBefore(asyncId, immediate[trigger_async_id_symbol], immediate);
 
       try {
         const argv = immediate._argv;
@@ -537,7 +537,7 @@ function getTimerCallbacks(runNextTicks) {
         continue;
       }
 
-      emitBefore(asyncId, timer[trigger_async_id_symbol]);
+      emitBefore(asyncId, timer[trigger_async_id_symbol], timer);
 
       let start;
       if (timer._repeat)