async_hooks initial implementation

Makefile

@@ -196,7 +196,7 @@ test: all
  $(MAKE) build-addons-napi
  $(MAKE) cctest
  $(PYTHON) tools/test.py --mode=release -J \
- addons addons-napi doctool inspector known_issues message pseudo-tty parallel sequential
+ addons addons-napi doctool inspector known_issues message pseudo-tty parallel sequential async-hooks
  $(MAKE) lint
 
 test-parallel: all
@@ -323,7 +323,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
@@ -415,6 +415,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);
@@ -162,6 +165,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
@@ -176,7 +181,7 @@ Agent.prototype.addRequest = function addRequest(req, options){
  // 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;
@@ -289,7 +294,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\-._~!$&'()*+,;=/:@]/
@@ -577,9 +578,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').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;
@@ -265,8 +267,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.
@@ -624,7 +627,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;
  }