net: add initial async_hooks net support

cluster is still slightly messed up, as I figure out how to handle the "faux" handles passed to net.
nodejs · trevnorris · Oct 31, 2016 · Oct 31, 2016 · Sep 28, 2016 · Nov 2, 2016
commit 955d390734535a4e81d02117b64fc4853b97e7da
diff --git a/lib/internal/cluster/child.js b/lib/internal/cluster/child.js
@@ -145,7 +145,13 @@ function rr(message, indexesKey, cb) {
   // with it. Fools net.Server into thinking that it's backed by a real
   // handle. Use a noop function for ref() and unref() because the control
   // channel is going to keep the worker alive anyway.
-  const handle = { close, listen, ref: noop, unref: noop };
+  //
+  // Emulate AsyncWrap::GetAsyncId() by returning a value from getAsyncId().
+  // TODO(trevnorris): Right now just return -2 so locations where this is
+  // propagating can be identified, if there are any. Since this is a faux
+  // handle not sure how to treat it (i.e. should it receive its own uid?).
+  const handle = {
+    close, listen, ref: noop, unref: noop, getAsyncId: () => -2 };
 
   if (message.sockname) {
     handle.getsockname = getsockname;  // TCP handles only.

diff --git a/lib/internal/cluster/shared_handle.js b/lib/internal/cluster/shared_handle.js
@@ -19,10 +19,12 @@ function SharedHandle(key, address, port, addressType, fd, flags) {
   else
     rval = net._createServerHandle(address, port, addressType, fd);
 
-  if (typeof rval === 'number')
+  if (typeof rval === 'number') {
     this.errno = rval;
-  else
+  } else {
     this.handle = rval;
+    this.getAsyncId = this.handle.getAsyncId;
+  }
 }
 
 SharedHandle.prototype.add = function(worker, send) {

diff --git a/lib/net.js b/lib/net.js
@@ -19,6 +19,7 @@ const PipeConnectWrap = process.binding('pipe_wrap').PipeConnectWrap;
 const ShutdownWrap = process.binding('stream_wrap').ShutdownWrap;
 const WriteWrap = process.binding('stream_wrap').WriteWrap;
 
+const setInitTriggerId = require('async_hooks').setInitTriggerId;
 
 var cluster;
 const errnoException = util._errnoException;
@@ -114,6 +115,7 @@ function initSocketHandle(self) {
   if (self._handle) {
     self._handle.owner = self;
     self._handle.onread = onread;
+    self._asyncId = self._handle.getAsyncId();
 
     // If handle doesn't support writev - neither do we
     if (!self._handle.writev)
@@ -129,6 +131,7 @@ function Socket(options) {
   if (!(this instanceof Socket)) return new Socket(options);
 
   this.connecting = false;
+  this._asyncId = -1;
   this._hadError = false;
   this._handle = null;
   this._parent = null;
@@ -143,9 +146,11 @@ function Socket(options) {
 
   if (options.handle) {
     this._handle = options.handle; // private
+    this._asyncId = this._handle.getAsyncId();
   } else if (options.fd !== undefined) {
     this._handle = createHandle(options.fd);
     this._handle.open(options.fd);
+    this._asyncId = this._handle.getAsyncId();
     if ((options.fd == 1 || options.fd == 2) &&
         (this._handle instanceof Pipe) &&
         process.platform === 'win32') {
@@ -228,6 +233,7 @@ function onSocketFinish() {
   if (!this._handle || !this._handle.shutdown)
     return this.destroy();
 
+  setInitTriggerId(this._asyncId);
   var req = new ShutdownWrap();
   req.oncomplete = afterShutdown;
   req.handle = this._handle;
@@ -296,6 +302,7 @@ function writeAfterFIN(chunk, encoding, cb) {
   // TODO: defer error events consistently everywhere, not just the cb
   this.emit('error', er);
   if (typeof cb === 'function') {
+    setInitTriggerId(this._asyncId);
     process.nextTick(cb, er);
   }
 }
@@ -854,6 +861,7 @@ function connect(self, address, port, addressType, localAddress, localPort) {
   }
 
   if (addressType === 6 || addressType === 4) {
+    setInitTriggerId(self._asyncId);
     const req = new TCPConnectWrap();
     req.oncomplete = afterConnect;
     req.address = address;
@@ -867,6 +875,7 @@ function connect(self, address, port, addressType, localAddress, localPort) {
       err = self._handle.connect6(req, address, port);
 
   } else {
+    setInitTriggerId(self._asyncId);
     const req = new PipeConnectWrap();
     req.address = address;
     req.oncomplete = afterConnect;
@@ -968,6 +977,7 @@ function lookupAndConnect(self, options) {
   // If host is an IP, skip performing a lookup
   var addressType = exports.isIP(host);
   if (addressType) {
+    setInitTriggerId(self._asyncId);
     process.nextTick(function() {
       if (self.connecting)
         connect(self, host, port, addressType, localAddress, localPort);
@@ -1061,6 +1071,7 @@ function afterConnect(status, handle, req, readable, writable) {
 
   // Update handle if it was wrapped
   // TODO(indutny): assert that the handle is actually an ancestor of old one
+  // TODO(trevnorris): Doesn't look like this does anything. Verify and remove.
   handle = self._handle;
 
   debug('afterConnect');
@@ -1137,6 +1148,7 @@ function Server(options, connectionListener) {
     configurable: true, enumerable: false
   });
 
+  this._asyncId = -1;
   this._handle = null;
   this._usingSlaves = false;
   this._slaves = [];
@@ -1252,6 +1264,7 @@ Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
     this._handle = rval;
   }
 
+  this._asyncId = this._handle.getAsyncId();
   this._handle.onconnection = onconnection;
   this._handle.owner = this;
 
@@ -1261,6 +1274,7 @@ Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
     var ex = exceptionWithHostPort(err, 'listen', address, port);
     this._handle.close();
     this._handle = null;
+    setInitTriggerId(this._asyncId);
     process.nextTick(emitErrorNT, this, ex);
     return;
   }
@@ -1272,6 +1286,7 @@ Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
   if (this._unref)
     this.unref();
 
+  setInitTriggerId(this._asyncId);
   process.nextTick(emitListeningNT, this);
 };
 
@@ -1357,6 +1372,7 @@ Server.prototype.listen = function() {
 
   if (options instanceof TCP) {
     this._handle = options;
+    this._asyncId = this._handle.getAsyncId();
     listen(this, null, -1, -1, backlog);
   } else if (typeof options.fd === 'number' && options.fd >= 0) {
     listen(this, null, null, null, backlog, options.fd);
@@ -1455,7 +1471,10 @@ function onconnection(err, clientHandle) {
 
 
 Server.prototype.getConnections = function(cb) {
+  const self = this;
+
   function end(err, connections) {
+    setInitTriggerId(self._asyncId);
     process.nextTick(cb, err, connections);
   }
 
@@ -1534,6 +1553,7 @@ Server.prototype._emitCloseIfDrained = function() {
     return;
   }
 
+  setInitTriggerId(this._asyncId);
   process.nextTick(emitCloseNT, this);
 };
 

diff --git a/test/parallel/test-net-end-close.js b/test/parallel/test-net-end-close.js
@@ -10,7 +10,8 @@ const s = new net.Socket({
     readStart: function() {
       process.nextTick(() => this.onread(uv.UV_EOF, null));
     },
-    close: (cb) => process.nextTick(cb)
+    close: (cb) => process.nextTick(cb),
+    getAsyncId: () => 0,
   },
   writable: false
 });