Async Hooks - Initial Implementation

Makefile

@@ -196,7 +196,8 @@ test: all
 	$(MAKE) build-addons-napi
 	$(MAKE) cctest
 	$(PYTHON) tools/test.py --mode=release -J \
-		doctool inspector known_issues message pseudo-tty parallel sequential $(CI_NATIVE_SUITES)
+		doctool inspector known_issues message pseudo-tty parallel sequential \
+		async-hooks $(CI_NATIVE_SUITES)
 	$(MAKE) lint
 
 test-parallel: all
@@ -326,7 +327,7 @@ test-all-valgrind: test-build
 	$(PYTHON) tools/test.py --mode=debug,release --valgrind
 
 CI_NATIVE_SUITES := addons addons-napi
-CI_JS_SUITES := doctool inspector known_issues message parallel pseudo-tty sequential
+CI_JS_SUITES := doctool inspector known_issues message parallel pseudo-tty sequential async-hooks
 
 # Build and test addons without building anything else
 test-ci-native: LOGLEVEL := info
@@ -418,6 +419,9 @@ test-timers:
 test-timers-clean:
 	$(MAKE) --directory=tools clean
 
+test-async-hooks:
+	$(PYTHON) tools/test.py --mode=release async-hooks
+
 
 ifneq ("","$(wildcard deps/v8/tools/run-tests.py)")
 test-v8: v8

lib/_http_agent.js

@@ -25,6 +25,8 @@ const net = require('net');
 const util = require('util');
 const EventEmitter = require('events');
 const debug = util.debuglog('http');
+const async_id_symbol = process.binding('async_wrap').async_id_symbol;
+const nextTick = require('internal/process/next_tick').nextTick;
 
 // New Agent code.
 
@@ -93,6 +95,7 @@ function Agent(options) {
           self.freeSockets[name] = freeSockets;
           socket.setKeepAlive(true, self.keepAliveMsecs);
           socket.unref();
+          socket[async_id_symbol] = -1;
           socket._httpMessage = null;
           self.removeSocket(socket, options);
           freeSockets.push(socket);
@@ -163,6 +166,8 @@ Agent.prototype.addRequest = function addRequest(req, options, port/*legacy*/,
   if (freeLen) {
     // we have a free socket, so use that.
     var socket = this.freeSockets[name].shift();
+    // Assign the handle a new asyncId and run any init() hooks.
+    socket._handle.asyncReset();
     debug('have free socket');
 
     // don't leak
@@ -177,7 +182,7 @@ Agent.prototype.addRequest = function addRequest(req, options, port/*legacy*/,
     // If we are under maxSockets create a new one.
     this.createSocket(req, options, function(err, newSocket) {
       if (err) {
-        process.nextTick(function() {
+        nextTick(newSocket._handle.getAsyncId(), function() {
           req.emit('error', err);
         });
         return;
@@ -290,7 +295,7 @@ Agent.prototype.removeSocket = function removeSocket(s, options) {
     // If we have pending requests and a socket gets closed make a new one
     this.createSocket(req, options, function(err, newSocket) {
       if (err) {
-        process.nextTick(function() {
+        nextTick(newSocket._handle.getAsyncId(), function() {
           req.emit('error', err);
         });
         return;

lib/_http_client.js

@@ -36,6 +36,7 @@ const Agent = require('_http_agent');
 const Buffer = require('buffer').Buffer;
 const urlToOptions = require('internal/url').urlToOptions;
 const outHeadersKey = require('internal/http').outHeadersKey;
+const nextTick = require('internal/process/next_tick').nextTick;
 
 // The actual list of disallowed characters in regexp form is more like:
 //    /[^A-Za-z0-9\-._~!$&'()*+,;=/:@]/
@@ -587,9 +588,12 @@ function responseKeepAlive(res, req) {
     socket.removeListener('close', socketCloseListener);
     socket.removeListener('error', socketErrorListener);
     socket.once('error', freeSocketErrorListener);
+    // There are cases where _handle === null. Avoid those. Passing null to
+    // nextTick() will call initTriggerId() to retrieve the id.
+    const asyncId = socket._handle ? socket._handle.getAsyncId() : null;
     // Mark this socket as available, AFTER user-added end
     // handlers have a chance to run.
-    process.nextTick(emitFreeNT, socket);
+    nextTick(asyncId, emitFreeNT, socket);
   }
 }
 

lib/_http_common.js

@@ -28,6 +28,7 @@ const HTTPParser = binding.HTTPParser;
 const FreeList = require('internal/freelist');
 const ondrain = require('internal/http').ondrain;
 const incoming = require('_http_incoming');
+const emitDestroy = require('async_hooks').emitDestroy;
 const IncomingMessage = incoming.IncomingMessage;
 const readStart = incoming.readStart;
 const readStop = incoming.readStop;
@@ -211,8 +212,13 @@ function freeParser(parser, req, socket) {
     parser.incoming = null;
     parser.outgoing = null;
     parser[kOnExecute] = null;
-    if (parsers.free(parser) === false)
+    if (parsers.free(parser) === false) {
       parser.close();
+    } else {
+      // Since the Parser destructor isn't going to run the destroy() callbacks
+      // it needs to be triggered manually.
+      emitDestroy(parser.getAsyncId());
+    }
   }
   if (req) {
     req.parser = null;

lib/_http_outgoing.js

@@ -31,6 +31,8 @@ const common = require('_http_common');
 const checkIsHttpToken = common._checkIsHttpToken;
 const checkInvalidHeaderChar = common._checkInvalidHeaderChar;
 const outHeadersKey = require('internal/http').outHeadersKey;
+const async_id_symbol = process.binding('async_wrap').async_id_symbol;
+const nextTick = require('internal/process/next_tick').nextTick;
 
 const CRLF = common.CRLF;
 const debug = common.debug;
@@ -264,8 +266,9 @@ function _writeRaw(data, encoding, callback) {
     if (this.output.length) {
       this._flushOutput(conn);
     } else if (!data.length) {
-      if (typeof callback === 'function')
-        process.nextTick(callback);
+      if (typeof callback === 'function') {
+        nextTick(this.socket[async_id_symbol], callback);
+      }
       return true;
     }
     // Directly write to socket.
@@ -623,7 +626,10 @@ const crlf_buf = Buffer.from('\r\n');
 OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {
   if (this.finished) {
     var err = new Error('write after end');
-    process.nextTick(writeAfterEndNT.bind(this), err, callback);
+    nextTick(this.socket[async_id_symbol],
+             writeAfterEndNT.bind(this),
+             err,
+             callback);
 
     return true;
   }