async_hooks initial implementation

lib/_http_agent.js

@@ -4,6 +4,7 @@ const net = require('net');
 const util = require('util');
 const EventEmitter = require('events');
 const debug = util.debuglog('http');
+const setInitTriggerId = require('async_hooks').setInitTriggerId;
 
 // New Agent code.
 
@@ -72,6 +73,7 @@ function Agent(options) {
           self.freeSockets[name] = freeSockets;
           socket.setKeepAlive(true, self.keepAliveMsecs);
           socket.unref();
+          socket._asyncId = -1;
           socket._httpMessage = null;
           self.removeSocket(socket, options);
           freeSockets.push(socket);
@@ -142,6 +144,8 @@ Agent.prototype.addRequest = function addRequest(req, options) {
   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
@@ -154,8 +158,11 @@ Agent.prototype.addRequest = function addRequest(req, options) {
   } else if (sockLen < this.maxSockets) {
     debug('call onSocket', sockLen, freeLen);
     // If we are under maxSockets create a new one.
+    // Agent#addRequest() is only called from the ClientRequest constructor. So
+    // it's unnecessary to set the parent id before calling createSocket().
     this.createSocket(req, options, function(err, newSocket) {
       if (err) {
+        setInitTriggerId(newSocket._handle.getAsyncId());
         process.nextTick(function() {
           req.emit('error', err);
         });
@@ -269,6 +276,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) {
+        setInitTriggerId(newSocket._handle.getAsyncId());
         process.nextTick(function() {
           req.emit('error', err);
         });

lib/_http_client.js

@@ -13,6 +13,7 @@ const debug = common.debug;
 const OutgoingMessage = require('_http_outgoing').OutgoingMessage;
 const Agent = require('_http_agent');
 const Buffer = require('buffer').Buffer;
+const setInitTriggerId = require('async_hooks').setInitTriggerId;
 const urlToOptions = require('internal/url').urlToOptions;
 
 // The actual list of disallowed characters in regexp form is more like:
@@ -563,6 +564,9 @@ function responseKeepAlive(res, req) {
     socket.removeListener('close', socketCloseListener);
     socket.removeListener('error', socketErrorListener);
     socket.once('error', freeSocketErrorListener);
+    // There are cases where _handle === null. Avoid those.
+    if (socket._handle)
+      setInitTriggerId(socket._handle.getAsyncId());
     // Mark this socket as available, AFTER user-added end
     // handlers have a chance to run.
     process.nextTick(emitFreeNT, socket);

lib/_http_common.js

@@ -18,6 +18,8 @@ exports.chunkExpression = /(?:^|\W)chunked(?:$|\W)/i;
 exports.continueExpression = /(?:^|\W)100-continue(?:$|\W)/i;
 exports.methods = methods;
 
+const emitDestroy = require('async_hooks').emitDestroy;
+
 const kOnHeaders = HTTPParser.kOnHeaders | 0;
 const kOnHeadersComplete = HTTPParser.kOnHeadersComplete | 0;
 const kOnBody = HTTPParser.kOnBody | 0;
@@ -196,8 +198,14 @@ 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
+      // need to be triggered manually.
+      emitDestroy(parser.getAsyncId());
+    }
+    parser = null;
   }
   if (req) {
     req.parser = null;

lib/_http_outgoing.js

@@ -9,6 +9,7 @@ const Buffer = require('buffer').Buffer;
 const common = require('_http_common');
 const checkIsHttpToken = common._checkIsHttpToken;
 const checkInvalidHeaderChar = common._checkInvalidHeaderChar;
+const setInitTriggerId = require('async_hooks').setInitTriggerId;
 
 const CRLF = common.CRLF;
 const debug = common.debug;
@@ -153,8 +154,10 @@ function _writeRaw(data, encoding, callback) {
     if (this.output.length) {
       this._flushOutput(conn);
     } else if (!data.length) {
-      if (typeof callback === 'function')
+      if (typeof callback === 'function') {
+        setInitTriggerId(this.socket._asyncId);
         process.nextTick(callback);
+      }
       return true;
     }
     // Directly write to socket.
@@ -477,6 +480,7 @@ 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');
+    setInitTriggerId(this.socket._asyncId);
     process.nextTick(writeAfterEndNT, this, err, callback);
 
     return true;