cluster: refactor module into multiple files

This commit splits the existing cluster module into several
internal modules. More specifically, the cluster master and
worker implementations are separated, and the various data
structures are separated.

PR-URL: https://github.com/nodejs/node/pull/10746
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Sam Roberts <vieuxtech@gmail.com>

1 files changed, 224 insertions, 0 deletions

diff --git a/lib/internal/cluster/child.js b/lib/internal/cluster/child.js
new file mode 100644
index 0000000000..3275eec720
--- /dev/null
+++ b/lib/internal/cluster/child.js
@@ -0,0 +1,224 @@
+'use strict';
+const assert = require('assert');
+const util = require('util');
+const EventEmitter = require('events');
+const Worker = require('internal/cluster/worker');
+const { internal, sendHelper } = require('internal/cluster/utils');
+const cluster = new EventEmitter();
+const handles = {};
+const indexes = {};
+const noop = () => {};
+
+module.exports = cluster;
+
+cluster.isWorker = true;
+cluster.isMaster = false;
+cluster.worker = null;
+cluster.Worker = Worker;
+
+cluster._setupWorker = function() {
+  const worker = new Worker({
+    id: +process.env.NODE_UNIQUE_ID | 0,
+    process: process,
+    state: 'online'
+  });
+
+  cluster.worker = worker;
+
+  process.once('disconnect', () => {
+    worker.emit('disconnect');
+
+    if (!worker.exitedAfterDisconnect) {
+      // Unexpected disconnect, master exited, or some such nastiness, so
+      // worker exits immediately.
+      process.exit(0);
+    }
+  });
+
+  process.on('internalMessage', internal(worker, onmessage));
+  send({ act: 'online' });
+
+  function onmessage(message, handle) {
+    if (message.act === 'newconn')
+      onconnection(message, handle);
+    else if (message.act === 'disconnect')
+      _disconnect.call(worker, true);
+  }
+};
+
+// obj is a net#Server or a dgram#Socket object.
+cluster._getServer = function(obj, options, cb) {
+  const indexesKey = [options.address,
+                      options.port,
+                      options.addressType,
+                      options.fd ].join(':');
+
+  if (indexes[indexesKey] === undefined)
+    indexes[indexesKey] = 0;
+  else
+    indexes[indexesKey]++;
+
+  const message = util._extend({
+    act: 'queryServer',
+    index: indexes[indexesKey],
+    data: null
+  }, options);
+
+  // Set custom data on handle (i.e. tls tickets key)
+  if (obj._getServerData)
+    message.data = obj._getServerData();
+
+  send(message, (reply, handle) => {
+    if (typeof obj._setServerData === 'function')
+      obj._setServerData(reply.data);
+
+    if (handle)
+      shared(reply, handle, indexesKey, cb);  // Shared listen socket.
+    else
+      rr(reply, indexesKey, cb);              // Round-robin.
+  });
+
+  obj.once('listening', () => {
+    cluster.worker.state = 'listening';
+    const address = obj.address();
+    message.act = 'listening';
+    message.port = address && address.port || options.port;
+    send(message);
+  });
+};
+
+// Shared listen socket.
+function shared(message, handle, indexesKey, cb) {
+  const key = message.key;
+  // Monkey-patch the close() method so we can keep track of when it's
+  // closed. Avoids resource leaks when the handle is short-lived.
+  const close = handle.close;
+
+  handle.close = function() {
+    send({ act: 'close', key });
+    delete handles[key];
+    delete indexes[indexesKey];
+    return close.apply(this, arguments);
+  };
+  assert(handles[key] === undefined);
+  handles[key] = handle;
+  cb(message.errno, handle);
+}
+
+// Round-robin. Master distributes handles across workers.
+function rr(message, indexesKey, cb) {
+  if (message.errno)
+    return cb(message.errno, null);
+
+  var key = message.key;
+
+  function listen(backlog) {
+    // TODO(bnoordhuis) Send a message to the master that tells it to
+    // update the backlog size. The actual backlog should probably be
+    // the largest requested size by any worker.
+    return 0;
+  }
+
+  function close() {
+    // lib/net.js treats server._handle.close() as effectively synchronous.
+    // That means there is a time window between the call to close() and
+    // the ack by the master process in which we can still receive handles.
+    // onconnection() below handles that by sending those handles back to
+    // the master.
+    if (key === undefined)
+      return;
+
+    send({ act: 'close', key });
+    delete handles[key];
+    delete indexes[indexesKey];
+    key = undefined;
+  }
+
+  function getsockname(out) {
+    if (key)
+      util._extend(out, message.sockname);
+
+    return 0;
+  }
+
+  // Faux handle. Mimics a TCPWrap with just enough fidelity to get away
+  // 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 };
+
+  if (message.sockname) {
+    handle.getsockname = getsockname;  // TCP handles only.
+  }
+
+  assert(handles[key] === undefined);
+  handles[key] = handle;
+  cb(0, handle);
+}
+
+// Round-robin connection.
+function onconnection(message, handle) {
+  const key = message.key;
+  const server = handles[key];
+  const accepted = server !== undefined;
+
+  send({ ack: message.seq, accepted });
+
+  if (accepted)
+    server.onconnection(0, handle);
+}
+
+function send(message, cb) {
+  return sendHelper(process, message, null, cb);
+}
+
+function _disconnect(masterInitiated) {
+  this.exitedAfterDisconnect = true;
+  let waitingCount = 1;
+
+  function checkWaitingCount() {
+    waitingCount--;
+
+    if (waitingCount === 0) {
+      // If disconnect is worker initiated, wait for ack to be sure
+      // exitedAfterDisconnect is properly set in the master, otherwise, if
+      // it's master initiated there's no need to send the
+      // exitedAfterDisconnect message
+      if (masterInitiated) {
+        process.disconnect();
+      } else {
+        send({ act: 'exitedAfterDisconnect' }, () => process.disconnect());
+      }
+    }
+  }
+
+  for (const key in handles) {
+    const handle = handles[key];
+    delete handles[key];
+    waitingCount++;
+
+    if (handle.owner)
+      handle.owner.close(checkWaitingCount);
+    else
+      handle.close(checkWaitingCount);
+  }
+
+  checkWaitingCount();
+}
+
+// Extend generic Worker with methods specific to worker processes.
+Worker.prototype.disconnect = function() {
+  _disconnect.call(this);
+  return this;
+};
+
+Worker.prototype.destroy = function() {
+  this.exitedAfterDisconnect = true;
+
+  if (!this.isConnected()) {
+    process.exit(0);
+  } else {
+    send({ act: 'exitedAfterDisconnect' }, () => process.disconnect());
+    process.once('disconnect', () => process.exit(0));
+  }
+};