child_process: fix channel disconnect logic

lib/internal/child_process.js

@@ -301,6 +301,24 @@ function closePendingHandle(target) {
   target._pendingMessage = null;
 }
 
+function clearHandleQueue(target) {
+  const queue = target._handleQueue;
+  target._handleQueue = null;
+
+  for (var i = 0; i < queue.length; i++) {
+    const args = queue[i];
+
+    if (!args.options.swallowErrors) {
+      const ex = new ERR_IPC_CHANNEL_CLOSED();
+
+      if (args.callback) {
+        process.nextTick(args.callback, ex);
+      } else {
+        process.nextTick(() => target.emit('error', ex));
+      }
+    }
+  }
+}
 
 ChildProcess.prototype.spawn = function(options) {
   var ipc;
@@ -544,11 +562,14 @@ function setupChannel(target, channel) {
 
     } else {
       this.buffering = false;
+
+      // The channel is closed, so get rid of remaining handles manually.
+      if (target._pendingMessage)
+        closePendingHandle(target);
+      if (target._handleQueue)
+        clearHandleQueue(target);
+
       target.disconnect();
-      channel.onread = nop;
-      channel.close();
-      target.channel = null;
-      maybeClose(target);
     }
   };
 
@@ -800,16 +821,17 @@ function setupChannel(target, channel) {
     // This marks the fact that the channel is actually disconnected.
     this.channel = null;
 
-    if (this._pendingMessage)
-      closePendingHandle(this);
-
     var fired = false;
     function finish() {
       if (fired) return;
       fired = true;
 
+      channel.onread = nop;
       channel.close();
+
       target.emit('disconnect');
+
+      maybeClose(target);
     }
 
     // If a message is being read, then wait for it to complete.

test/parallel/test-child-process-close-handle-queue.js

@@ -0,0 +1,86 @@
+// Tests that a child's internal handle queue and pending handle get cleared
+// when the child exits, calling all provided callbacks (or emitting errors).
+
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const net = require('net');
+const cp = require('child_process');
+const fixtures = require('../common/fixtures');
+
+const fixture = fixtures.path('exit.js');
+
+function testErrorCallOnce() {
+  return common.expectsError({
+    type: Error,
+    message: 'Channel closed',
+    code: 'ERR_IPC_CHANNEL_CLOSED'
+  }, 1);
+}
+
+const server = net.createServer((s) => {
+  if (common.isWindows) {
+    s.on('error', function(err) {
+      // Prevent possible ECONNRESET errors from popping up
+      if (err.code !== 'ECONNRESET')
+        throw err;
+    });
+  }
+  setTimeout(function() {
+    s.destroy();
+  }, 100);
+}).listen(0, common.mustCall((error) => {
+  const child = cp.fork(fixture);
+
+  let gotExit = false;
+
+  function send(callback) {
+    const s = net.connect(server.address().port, function() {
+      child.send({}, s, callback);
+    });
+
+    // https://github.com/nodejs/node/issues/3635#issuecomment-157714683
+    // ECONNREFUSED or ECONNRESET errors can happen if this connection is still
+    // establishing while the server has already closed.
+    // EMFILE can happen if the worker __and__ the server had already closed.
+    s.on('error', function(err) {
+      if ((err.code !== 'ECONNRESET') &&
+          (err.code !== 'ECONNREFUSED') &&
+          (err.code !== 'EMFILE')) {
+        throw err;
+      }
+    });
+  }
+
+  // Pending handle get written and callback gets
+  // called immediately (before 'exit')
+  send(common.mustCall((error) => {
+    assert.strictEqual(error, null);
+    assert.strictEqual(gotExit, false);
+  }));
+
+  // Handle gets put into handle queue
+  send(testErrorCallOnce());
+
+  // Handle also gets put into handle queue, but without
+  // a callback, so the error should be emitted instead
+  send();
+
+  child.on('error', testErrorCallOnce());
+
+  child.on('exit', common.mustCall((code, signal) => {
+    gotExit = true;
+    assert.strictEqual(code, 1);
+    assert.strictEqual(signal, null);
+  }));
+
+  child.on('close', common.mustCall((code, signal) => {
+    assert.strictEqual(code, 1);
+    assert.strictEqual(signal, null);
+
+    assert.strictEqual(child._handleQueue, null);
+    assert.strictEqual(child._pendingMessage, null);
+
+    server.close();
+  }));
+}));

test/parallel/test-child-process-disconnect.js

@@ -68,48 +68,55 @@ if (process.argv[2] === 'child') {
   // testcase
   const child = fork(process.argv[1], ['child']);
 
-  let childFlag = false;
-  let parentFlag = false;
+  let exitEmitted = false;
+  let disconnectEmitted = false;
+  let gotMessage = false;
 
   // when calling .disconnect the event should emit
   // and the disconnected flag should be true.
   child.on('disconnect', common.mustCall(function() {
-    parentFlag = child.connected;
+    assert.strictEqual(child.connected, false);
+
+    disconnectEmitted = true;
   }));
 
   // the process should also self terminate without using signals
-  child.on('exit', common.mustCall());
+  child.on('exit', common.mustCall(function(code, signal) {
+    assert.strictEqual(disconnectEmitted, true);
+    assert.strictEqual(code, 0);
+    assert.strictEqual(signal, null);
+
+    exitEmitted = true;
+  }));
 
   // when child is listening
   child.on('message', function(obj) {
-    if (obj && obj.msg === 'ready') {
-
-      // connect to child using TCP to know if disconnect was emitted
-      const socket = net.connect(obj.port);
-
-      socket.on('data', function(data) {
-        data = data.toString();
-
-        // ready to be disconnected
-        if (data === 'ready') {
-          child.disconnect();
-          assert.throws(
-            child.disconnect.bind(child),
-            {
-              code: 'ERR_IPC_DISCONNECTED'
-            });
-          return;
-        }
-
-        // disconnect is emitted
-        childFlag = (data === 'true');
-      });
-
-    }
-  });
+    assert.ok(obj && obj.msg === 'ready');
+
+    gotMessage = true;
+
+    // connect to child using TCP to know if disconnect was emitted
+    const socket = net.connect(obj.port);
 
-  process.on('exit', function() {
-    assert.strictEqual(childFlag, false);
-    assert.strictEqual(parentFlag, false);
+    socket.setEncoding('utf-8');
+
+    socket.on('data', function(data) {
+      // ready to be disconnected
+      if (data === 'ready') {
+        child.disconnect();
+        assert.throws(child.disconnect.bind(child),
+                      { code: 'ERR_IPC_DISCONNECTED' });
+        return;
+      }
+
+      assert.strictEqual(data, 'false');
+    });
   });
+
+  // The process emits 'close' after all streams have flushed and it has exited.
+  child.on('close', common.mustCall(function() {
+    assert.strictEqual(disconnectEmitted, true);
+    assert.strictEqual(exitEmitted, true);
+    assert.strictEqual(gotMessage, true);
+  }));
 }