async_hooks: add currentResource

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

@@ -0,0 +1,165 @@
+'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,
+  currentResource,
+  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: ['current-resource', 'destroy'],
+  method: ['callbacks', 'async'],
+  n: [1e6]
+});
+
+function buildCurrentResource(getServe) {
+  const server = createServer(getServe(getCLS, setCLS));
+  const hook = createHook({ init });
+  const cls = Symbol('cls');
+  let closed = false;
+  hook.enable();
+
+  return {
+    server,
+    close
+  };
+
+  function getCLS() {
+    // we need to protect this, as once the hook is
+    // disabled currentResource will return null
+    if (closed) {
+      return;
+    }
+
+    const resource = currentResource();
+    if (!resource[cls]) {
+      return null;
+    }
+    return resource[cls].state;
+  }
+
+  function setCLS(state) {
+    // we need to protect this, as once the hook is
+    // disabled currentResource will return null
+    if (closed) {
+      return;
+    }
+    const resource = currentResource();
+    if (!resource[cls]) {
+      resource[cls] = { state };
+    } else {
+      resource[cls].state = state;
+    }
+  }
+
+  function init(asyncId, type, triggerAsyncId, resource) {
+    if (type === 'TIMERWRAP') return;
+
+    var cr = currentResource();
+    if (cr) {
+      resource[cls] = cr[cls];
+    }
+  }
+
+  function close() {
+    closed = true;
+    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) {
+    if (type === 'TIMERWRAP') return;
+
+    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(function() {
+      readFile(__filename, function() {
+        res.setHeader('content-type', 'application/json');
+        res.end(JSON.stringify({ cls: getCLS() }));
+      });
+    }, 10);
+  };
+}
+
+const types = {
+  'current-resource': buildCurrentResource,
+  'destroy': buildDestroy
+};
+
+const asyncMethod = {
+  'callbacks': getServeCallbacks,
+  'async': getServeAwait
+};
+
+function main({ type, method }) {
+  const { server, close } = types[type](asyncMethod[method]);
+
+  server
+    .listen(common.PORT)
+    .on('listening', function() {
+
+      bench.http({
+        path,
+        connections
+      }, function() {
+        close();
+      });
+    });
+}

doc/api/async_hooks.md

@@ -462,6 +462,57 @@ init for PROMISE with id 6, trigger id: 5  # the Promise returned by then()
   after 6
 ```
 
+#### async_hooks.currentResource()
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Returns: {Object} The resource that triggered the current
+  execution context.
+  Useful to store data within the resource.
+
+```js
+const { open } = require('fs');
+const { executionAsyncId, currentResource } = require('async_hooks');
+
+console.log(executionAsyncId(), currentResource());  // 1 null
+open(__filename, 'r', (err, fd) => {
+  console.log(executionAsyncId(), currentResource());  // 7 FSReqWrap
+});
+```
+
+This can be used to implement continuation local storage without the
+using of a tracking `Map` to store the metadata:
+
+```js
+const { createServer } = require('http');
+const {
+  executionAsyncId,
+  currentResource,
+  createHook
+} = require('async_hooks');
+const sym = Symbol('state'); // private symbol to avoid pollution
+
+createHook({
+  init(asyncId, type, triggerAsyncId, resource) {
+    const cr = currentResource();
+    if (cr) {
+      resource[sym] = cr[sym];
+    }
+  }
+}).enable();
+
+const server = createServer(function(req, res) {
+  currentResource()[sym] = { state: req.url };
+  setTimeout(function() {
+    res.end(JSON.stringify(currentResource()[sym]));
+  }, 100);
+}).listen(3000);
+```
+
+`currentResource()` will return `null` during application bootstrap.
+
 #### async_hooks.executionAsyncId()
 
 <!-- YAML

lib/async_hooks.js

@@ -25,6 +25,7 @@ const {
   emitBefore,
   emitAfter,
   emitDestroy,
+  currentResource
 } = internal_async_hooks;
 
 // Get symbols
@@ -216,6 +217,7 @@ module.exports = {
   createHook,
   executionAsyncId,
   triggerAsyncId,
+  currentResource,
   // Embedder API
   AsyncResource,
 };

lib/internal/async_hooks.js

@@ -67,6 +67,8 @@ const active_hooks = {
 
 const { registerDestroyHook } = async_wrap;
 
+const { currentResource, setCurrentResource } = async_wrap;
+
 // Each constant tracks how many callbacks there are for any given step of
 // async execution. These are tracked so if the user didn't include callbacks
 // for a given step, that step can bail out early.
@@ -337,13 +339,14 @@ function emitInitScript(asyncId, type, triggerAsyncId, resource) {
 }
 
 
-function emitBeforeScript(asyncId, triggerAsyncId) {
+function emitBeforeScript(asyncId, triggerAsyncId, resource) {
   // Validate the ids. An id of -1 means it was never set and is visible on the
   // call graph. An id < -1 should never happen in any circumstance. Throw
   // on user calls because async state should still be recoverable.
   validateAsyncId(asyncId, 'asyncId');
   validateAsyncId(triggerAsyncId, 'triggerAsyncId');
 
+  setCurrentResource(resource);
   pushAsyncIds(asyncId, triggerAsyncId);
 
   if (async_hook_fields[kBefore] > 0)
@@ -357,6 +360,8 @@ function emitAfterScript(asyncId) {
   if (async_hook_fields[kAfter] > 0)
     emitAfterNative(asyncId);
 
+  setCurrentResource(null);
+
   popAsyncIds(asyncId);
 }
 
@@ -444,6 +449,7 @@ module.exports = {
   clearDefaultTriggerAsyncId,
   clearAsyncIdStack,
   hasAsyncIdStack,
+  currentResource,
   // Internal Embedder API
   newAsyncId,
   getOrSetAsyncId,
@@ -457,4 +463,5 @@ module.exports = {
   emitAfter: emitAfterScript,
   emitDestroy: emitDestroyScript,
   registerDestroyHook,
+  setCurrentResource
 };

lib/internal/process/next_tick.js

@@ -43,7 +43,7 @@ function setupNextTick(_setupNextTick, _setupPromises) {
     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);
         // emitDestroy() places the async_id_symbol into an asynchronous queue
         // that calls the destroy callback in the future. It's called before
         // calling tock.callback so destroy will be called even if the callback

lib/timers.js

@@ -45,7 +45,8 @@ const {
   // The needed emit*() functions.
   emitBefore,
   emitAfter,
-  emitDestroy
+  emitDestroy,
+  setCurrentResource
 } = require('internal/async_hooks');
 
 // *Must* match Environment::ImmediateInfo::Fields in src/env.h.
@@ -310,7 +311,7 @@ function listOnTimeout(list, now) {
       continue;
     }
 
-    emitBefore(asyncId, timer[trigger_async_id_symbol]);
+    emitBefore(asyncId, timer[trigger_async_id_symbol], timer);
 
     tryOnTimeout(timer);
 
@@ -340,6 +341,7 @@ function tryOnTimeout(timer, start) {
   try {
     ontimeout(timer);
   } finally {
+    setCurrentResource(null);
     if (timer._repeat) {
       rearm(timer, start);
     } else {
@@ -621,7 +623,7 @@ function processImmediate() {
     immediate._destroyed = true;
 
     const asyncId = immediate[async_id_symbol];
-    emitBefore(asyncId, immediate[trigger_async_id_symbol]);
+    emitBefore(asyncId, immediate[trigger_async_id_symbol], immediate);
 
     count++;
     if (immediate[kRefed])

src/async_wrap-inl.h

@@ -50,7 +50,7 @@ inline AsyncWrap::AsyncScope::AsyncScope(AsyncWrap* wrap)
   Environment* env = wrap->env();
   if (env->async_hooks()->fields()[Environment::AsyncHooks::kBefore] == 0)
     return;
-  EmitBefore(env, wrap->get_async_id());
+  EmitBefore(env, wrap->get_async_id(), wrap->object());
 }
 
 inline AsyncWrap::AsyncScope::~AsyncScope() {