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;
  }