lukechilds
/
node
mirror of https://github.com/lukechilds/node.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

cluster: clean up lib/cluster.js 

Clean up and DRY the cluster source code. Fix a few bugs while we're
here:

* Short-lived handles in long-lived worker processes were never
  reclaimed, resulting in resource leaks.

* Handles in the master process are now closed when the last worker
  that holds a reference to them quits. Previously, they were only
  closed at cluster shutdown.

* The cluster object no longer exposes functions/properties that are
  only valid in the 'other' process, e.g. cluster.fork() is no longer
  exported in worker processes.

So much goodness and still manages to reduce the line count from 590
to 320.
v0.11.2-release
Ben Noordhuis 12 years ago
parent
c77312405e
commit
41b75ca926
1 changed files with 237 additions and 505 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 718
      lib/cluster.js

718
lib/cluster.js
View File

@ -19,572 +19,304 @@
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE // OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
// USE OR OTHER DEALINGS IN THE SOFTWARE. // USE OR OTHER DEALINGS IN THE SOFTWARE.
var EventEmitter = require('events').EventEmitter;
var assert = require('assert'); var assert = require('assert');
var dgram = require('dgram');
var fork = require('child_process').fork; var fork = require('child_process').fork;
var net = require('net'); var net = require('net');
var EventEmitter = require('events').EventEmitter;
var util = require('util'); var util = require('util');
function isObject(o) { var cluster = new EventEmitter;
return (typeof o === 'object' && o !== null); module.exports = cluster;
} cluster.Worker = Worker;
cluster.isWorker = ('NODE_UNIQUE_ID' in process.env);
cluster.isMaster = (cluster.isWorker === false);
var debug;
if (process.env.NODE_DEBUG && /cluster/.test(process.env.NODE_DEBUG)) {
debug = function(x) {
var prefix = process.pid + ',' +
(process.env.NODE_UNIQUE_ID ? 'Worker' : 'Master');
console.error(prefix, x);
};
} else {
debug = function() { };
}
// cluster object: function Worker() {
function Cluster() { if (!(this instanceof Worker)) return new Worker;
EventEmitter.call(this); EventEmitter.call(this);
this.suicide = undefined;
this.state = 'none';
this.id = 0;
} }
util.inherits(Worker, EventEmitter);
util.inherits(Cluster, EventEmitter); Worker.prototype.kill = function() {
this.destroy.apply(this, arguments);
var cluster = module.exports = new Cluster();
// Used in the master:
var masterStarted = false;
var ids = 0;
var serverHandlers = {};
// Used in the worker:
var serverListeners = {};
var queryIds = 0;
var queryCallbacks = {};
// Define isWorker and isMaster
cluster.isWorker = 'NODE_UNIQUE_ID' in process.env;
cluster.isMaster = ! cluster.isWorker;
// The worker object is only used in a worker
cluster.worker = cluster.isWorker ? {} : null;
// The workers array is only used in the master
cluster.workers = cluster.isMaster ? {} : null;
// Settings object
var settings = cluster.settings = {};
// Simple function to call a function on each worker
function eachWorker(cb) {
// Go through all workers
for (var id in cluster.workers) {
if (cluster.workers.hasOwnProperty(id)) {
cb(cluster.workers[id]);
}
}
}
// Extremely simple progress tracker
function ProgressTracker(missing, callback) {
this.missing = missing;
this.callback = callback;
}
ProgressTracker.prototype.done = function() {
this.missing -= 1;
this.check();
}; };
ProgressTracker.prototype.check = function() {
if (this.missing === 0) this.callback(); Worker.prototype.send = function() {
this.process.send.apply(this.process, arguments);
}; };
cluster.setupMaster = function(options) { // Master/worker specific methods are defined in the *Init() functions.
// This can only be called from the master.
assert(cluster.isMaster);
// Don't allow this function to run more than once
if (masterStarted) return;
masterStarted = true;
// Get filename and arguments if (cluster.isMaster)
options = options || {}; masterInit();
else
workerInit();
// By default, V8 writes the profile data of all processes to a single
// v8.log.
//
// Running that log file through a tick processor produces bogus numbers
// because many events won't match up with the recorded memory mappings
// and you end up with graphs where 80+% of ticks is unaccounted for.
//
// Fixing the tick processor to deal with multi-process output is not very
// useful because the processes may be running wildly disparate workloads.
//
// That's why we fix up the command line arguments to include
// a "--logfile=v8-%p.log" argument (where %p is expanded to the PID)
// unless it already contains a --logfile argument.
var execArgv = options.execArgv || process.execArgv;
if (execArgv.some(function(s) { return /^--prof/.test(s); }) &&
!execArgv.some(function(s) { return /^--logfile=/.test(s); }))
{
execArgv = execArgv.slice();
execArgv.push('--logfile=v8-%p.log');
}
// Set settings object function masterInit() {
settings = cluster.settings = { cluster.workers = {};
exec: options.exec || process.argv[1],
execArgv: execArgv,
args: options.args || process.argv.slice(2),
silent: options.silent || false
};
// emit setup event var intercom = new EventEmitter;
cluster.emit('setup'); var settings = {
args: process.argv.slice(2),
exec: process.argv[1],
execArgv: process.execArgv,
silent: false
}; };
cluster.settings = settings;
// Check if a message is internal only // Indexed by address:port:etc key. Its entries are dicts with handle and
var INTERNAL_PREFIX = 'NODE_CLUSTER_'; // workers keys. That second one is a list of workers that hold a reference
function isInternalMessage(message) { // to the handle. When a worker dies, we scan the dicts and close the handle
return isObject(message) && // when its reference count drops to zero. Yes, that means we're doing an
typeof message.cmd === 'string' && // O(n*m) scan but n and m are small and worker deaths are rare events anyway.
message.cmd.length > INTERNAL_PREFIX.length && var handles = {};
message.cmd.slice(0, INTERNAL_PREFIX.length) === INTERNAL_PREFIX;
}
// Modify message object to be internal
function internalMessage(inMessage) {
var outMessage = util._extend({}, inMessage);
// Add internal prefix to cmd
outMessage.cmd = INTERNAL_PREFIX + (outMessage.cmd || '');
return outMessage; var initialized = false;
cluster.setupMaster = function(options) {
if (initialized === true) return;
initialized = true;
settings = util._extend(settings, options || {});
// Tell V8 to write profile data for each process to a separate file.
// Without --logfile=v8-%p.log, everything ends up in a single, unusable
// file. (Unusable because what V8 logs are memory addresses and each
// process has its own memory mappings.)
if (settings.execArgv.some(function(s) { return /^--prof/.test(s); }) &&
!settings.execArgv.some(function(s) { return /^--logfile=/.test(s); }))
{
settings.execArgv = settings.execArgv.concat(['--logfile=v8-%p.log']);
} }
cluster.settings = settings;
cluster.emit('setup');
};
// Handle callback messages var ids = 0;
function handleResponse(outMessage, outHandle, inMessage, inHandle, worker) { cluster.fork = function(env) {
cluster.setupMaster();
// The message there will be sent var worker = new Worker;
var message = internalMessage(outMessage); worker.id = ++ids;
var workerEnv = util._extend({}, process.env);
// callback id - will be undefined if not set workerEnv = util._extend(workerEnv, env);
message._queryEcho = inMessage._requestEcho; workerEnv.NODE_UNIQUE_ID = '' + worker.id;
worker.process = fork(settings.exec, settings.args, {
env: workerEnv,
silent: settings.silent,
execArgv: settings.execArgv
});
worker.process.once('exit', function(exitCode, signalCode) {
worker.suicide = !!worker.suicide;
worker.state = 'dead';
worker.emit('exit', exitCode, signalCode);
cluster.emit('exit', worker, exitCode, signalCode);
delete cluster.workers[worker.id];
});
worker.process.once('disconnect', function() {
worker.suicide = !!worker.suicide;
worker.state = 'disconnected';
worker.emit('disconnect');
cluster.emit('disconnect', worker);
delete cluster.workers[worker.id];
});
worker.process.on('error', worker.emit.bind(worker, 'error'));
worker.process.on('message', worker.emit.bind(worker, 'message'));
worker.process.on('internalMessage', internal(worker, onmessage));
process.nextTick(function() {
cluster.emit('fork', worker);
});
cluster.workers[worker.id] = worker;
return worker;
};
// Call callback if a query echo is received cluster.disconnect = function(cb) {
if (inMessage._queryEcho) { for (var key in cluster.workers) {
queryCallbacks[inMessage._queryEcho](inMessage.content, inHandle); var worker = cluster.workers[key];
delete queryCallbacks[inMessage._queryEcho]; worker.disconnect();
} }
if (cb) intercom.once('disconnect', cb);
};
// Send if outWrap contains something useful cluster.on('disconnect', function(worker) {
if (!(outMessage === undefined && message._queryEcho === undefined)) { delete cluster.workers[worker.id];
sendInternalMessage(worker, message, outHandle); // O(n*m) scan but for small values of n and m.
} for (var key in handles) {
var e = handles[key];
var i = e.workers.indexOf(worker);
if (i === -1) continue;
e.workers.splice(i, 1);
if (e.workers.length !== 0) continue;
e.handle.close();
delete handles[key];
}
if (Object.keys(handles).length === 0) {
intercom.emit('disconnect');
} }
});
// Handle messages from both master and workers Worker.prototype.disconnect = function() {
var messageHandler = {}; this.suicide = true;
function handleMessage(worker, inMessage, inHandle) { send(this, { act: 'disconnect' });
};
// Remove internal prefix
var message = util._extend({}, inMessage);
message.cmd = inMessage.cmd.substr(INTERNAL_PREFIX.length);
var respondUsed = false;
function respond(outMessage, outHandler) {
respondUsed = true;
handleResponse(outMessage, outHandler, inMessage, inHandle, worker);
}
// Run handler if it exists Worker.prototype.destroy = function(signo) {
if (messageHandler[message.cmd]) { signo = signo || 'SIGTERM';
messageHandler[message.cmd](message, worker, respond); var proc = this.process;
if (proc.connected) {
proc.once('disconnect', proc.kill.bind(proc, signo));
proc.disconnect();
return;
} }
proc.kill(signo);
};
// Send respond if it hasn't been called yet function onmessage(message, handle) {
if (respondUsed === false) { var worker = this;
respond(); if (message.act === 'online')
} online(worker);
else if (message.act === 'queryServer')
queryServer(worker, message);
else if (message.act === 'listening')
listening(worker, message);
else if (message.act === 'suicide')
worker.suicide = true;
} }
// Messages to the master will be handled using these methods function online(worker) {
if (cluster.isMaster) {
// Handle online messages from workers
messageHandler.online = function(message, worker) {
worker.state = 'online'; worker.state = 'online';
debug('Worker ' + worker.process.pid + ' online');
worker.emit('online'); worker.emit('online');
cluster.emit('online', worker); cluster.emit('online', worker);
}; }
// Handle queryServer messages from workers
messageHandler.queryServer = function(message, worker, send) {
// This sequence of information is unique to the connection function queryServer(worker, message) {
// but not to the worker
var args = [message.address, var args = [message.address,
message.port, message.port,
message.addressType, message.addressType,
message.fd]; message.fd];
var key = args.join(':'); var key = args.join(':');
var handler; var e = handles[key];
if (typeof e === 'undefined') {
if (serverHandlers.hasOwnProperty(key)) { e = { workers: [] };
handler = serverHandlers[key]; if (message.addressType === 'udp4' || message.addressType === 'udp6')
} else if (message.addressType === 'udp4' || e.handle = dgram._createSocketHandle.apply(null, args);
message.addressType === 'udp6') { else
var dgram = require('dgram'); e.handle = net._createServerHandle.apply(null, args);
handler = dgram._createSocketHandle.apply(net, args); handles[key] = e;
serverHandlers[key] = handler; }
} else { e.workers.push(worker);
handler = net._createServerHandle.apply(net, args); send(worker, { ack: message.seq }, e.handle);
serverHandlers[key] = handler;
} }
// echo callback with the fd handler associated with it function listening(worker, message) {
send({}, handler); var info = {
};
// Handle listening messages from workers
messageHandler.listening = function(message, worker) {
worker.state = 'listening';
// Emit listening, now that we know the worker is listening
worker.emit('listening', {
address: message.address,
port: message.port,
addressType: message.addressType, addressType: message.addressType,
fd: message.fd
});
cluster.emit('listening', worker, {
address: message.address, address: message.address,
port: message.port, port: message.port,
addressType: message.addressType,
fd: message.fd fd: message.fd
});
}; };
worker.state = 'listening';
// Handle suicide messages from workers worker.emit('listening', info);
messageHandler.suicide = function(message, worker) { cluster.emit('listening', worker, info);
worker.suicide = true;
};
}
// Messages to a worker will be handled using these methods
else if (cluster.isWorker) {
// Handle worker.disconnect from master
messageHandler.disconnect = function(message, worker) {
worker.disconnect();
};
}
function toDecInt(value) {
value = parseInt(value, 10);
return isNaN(value) ? null : value;
}
// Create a worker object, that works both for master and worker
function Worker(customEnv) {
if (!(this instanceof Worker)) return new Worker();
EventEmitter.call(this);
var self = this;
var env = process.env;
// Assign a unique id, default null
this.id = cluster.isMaster ? ++ids : toDecInt(env.NODE_UNIQUE_ID);
// XXX: Legacy. Remove in 0.9
this.workerID = this.uniqueID = this.id;
// Assign state
this.state = 'none';
// Create or get process
if (cluster.isMaster) {
// Create env object
// first: copy and add id property
var envCopy = util._extend({}, env);
envCopy['NODE_UNIQUE_ID'] = this.id;
// second: extend envCopy with the env argument
if (isObject(customEnv)) {
envCopy = util._extend(envCopy, customEnv);
}
// fork worker
this.process = fork(settings.exec, settings.args, {
'env': envCopy,
'silent': settings.silent,
'execArgv': settings.execArgv
});
} else {
this.process = process;
}
if (cluster.isMaster) {
// Save worker in the cluster.workers array
cluster.workers[this.id] = this;
// Emit a fork event, on next tick
// There is no worker.fork event since this has no real purpose
process.nextTick(function() {
cluster.emit('fork', self);
});
}
// handle internalMessage, exit and disconnect event
this.process.on('internalMessage', handleMessage.bind(null, this));
this.process.once('exit', function(exitCode, signalCode) {
prepareExit(self, 'dead');
self.emit('exit', exitCode, signalCode);
cluster.emit('exit', self, exitCode, signalCode);
});
this.process.once('disconnect', function() {
prepareExit(self, 'disconnected');
self.emit('disconnect');
cluster.emit('disconnect', self);
});
// relay message and error
this.process.on('message', this.emit.bind(this, 'message'));
this.process.on('error', this.emit.bind(this, 'error'));
}
util.inherits(Worker, EventEmitter);
cluster.Worker = Worker;
function prepareExit(worker, state) {
// set state to disconnect
worker.state = state;
// Make suicide a boolean
worker.suicide = !!worker.suicide;
// Remove from workers in the master
if (cluster.isMaster) {
delete cluster.workers[worker.id];
}
} }
// Send internal message function send(worker, message, handle, cb) {
function sendInternalMessage(worker, message/*, handler, callback*/) { sendHelper(worker.process, message, handle, cb);
// Exist callback
var callback = arguments[arguments.length - 1];
if (typeof callback !== 'function') {
callback = undefined;
} }
// exist handler
var handler = arguments[2] !== callback ? arguments[2] : undefined;
if (!isInternalMessage(message)) {
message = internalMessage(message);
} }
// Store callback for later
if (callback) {
message._requestEcho = worker.id + ':' + (++queryIds);
queryCallbacks[message._requestEcho] = callback;
}
function workerInit() {
var handles = [];
worker.send(message, handler); // Called from src/node.js
cluster._setupWorker = function() {
var worker = new Worker;
cluster.worker = worker;
worker.id = +process.env.NODE_UNIQUE_ID | 0;
worker.state = 'online';
worker.process = process;
process.once('disconnect', process.exit.bind(null, 0));
process.on('internalMessage', internal(worker, onmessage));
send({ act: 'online' });
function onmessage(message, handle) {
if (message.act === 'disconnect') worker.disconnect();
} }
// Send message to worker or master
Worker.prototype.send = function() {
// You could also just use process.send in a worker
this.process.send.apply(this.process, arguments);
}; };
// Kill the worker without restarting // obj is a net#Server or a dgram#Socket object.
Worker.prototype.kill = Worker.prototype.destroy = function(signal) { cluster._getServer = function(obj, address, port, addressType, fd, cb) {
if (!signal) var message = {
signal = 'SIGTERM'; addressType: addressType,
address: address,
var self = this; port: port,
act: 'queryServer',
this.suicide = true; fd: fd
};
if (cluster.isMaster) { send(message, function(_, handle) {
// Disconnect IPC channel // Monkey-patch the close() method so we can keep track of when it's
// this way the worker won't need to propagate suicide state to master // closed. Avoids resource leaks when the handle is short-lived.
if (self.process.connected) { var close = handle.close;
self.process.once('disconnect', function() { handle.close = function() {
self.process.kill(signal); var index = handles.indexOf(handle);
}); if (index !== -1) handles.splice(index, 1);
self.process.disconnect(); return close.apply(this, arguments);
} else { };
self.process.kill(signal); handles.push(handle);
} cb(handle);
} else {
// Channel is open
if (this.process.connected) {
// Inform master to suicide and then kill
sendInternalMessage(this, {cmd: 'suicide'}, function() {
process.exit(0);
}); });
obj.once('listening', function() {
// When channel is closed, terminate the process cluster.worker.state = 'listening';
this.process.once('disconnect', function() { message.act = 'listening';
process.exit(0); message.port = obj.address().port || port;
send(message);
}); });
} else {
process.exit(0);
}
}
}; };
// The .disconnect function will close all servers
// and then disconnect the IPC channel.
if (cluster.isMaster) {
// Used in master
Worker.prototype.disconnect = function() { Worker.prototype.disconnect = function() {
this.suicide = true; for (var handle; handle = handles.shift(); handle.close());
process.disconnect();
sendInternalMessage(this, {cmd: 'disconnect'});
}; };
} else { Worker.prototype.destroy = function() {
// Used in workers if (!process.connected) process.exit(0);
Worker.prototype.disconnect = function() { var exit = process.exit.bind(null, 0);
var self = this; send({ act: 'suicide' }, exit);
process.once('disconnect', exit);
this.suicide = true; process.disconnect();
};
// keep track of open servers
var servers = Object.keys(serverListeners).length;
var progress = new ProgressTracker(servers, function() {
// There are no more servers open so we will close the IPC channel.
// Closing the IPC channel will emit a disconnect event
// in both master and worker on the process object.
// This event will be handled by prepareExit.
self.process.disconnect();
});
// depending on where this function was called from (master or worker)
// The suicide state has already been set,
// but it doesn't really matter if we set it again.
sendInternalMessage(this, {cmd: 'suicide'}, function() {
// in case there are no servers
progress.check();
// closing all servers gracefully
var server;
for (var key in serverListeners) {
server = serverListeners[key];
// in case the server is closed we won't close it again
if (server._handle === null) {
progress.done();
continue;
}
server.on('close', progress.done.bind(progress)); function send(message, cb) {
server.close(); sendHelper(process, message, null, cb);
} }
});
};
} }
// Fork a new worker
cluster.fork = function(env) {
// This can only be called from the master.
assert(cluster.isMaster);
// Make sure that the master has been initialized
cluster.setupMaster();
return (new cluster.Worker(env)); var seq = 0;
}; var callbacks = {};
function sendHelper(proc, message, handle, cb) {
// execute .disconnect on all workers and close handlers when done // Mark message as internal. See INTERNAL_PREFIX in lib/child_process.js
cluster.disconnect = function(callback) { message = util._extend({ cmd: 'NODE_CLUSTER' }, message);
// This can only be called from the master. if (cb) callbacks[seq] = cb;
assert(cluster.isMaster); message.seq = seq;
seq += 1;
// Close all TCP handlers when all workers are disconnected proc.send(message, handle);
var workers = Object.keys(cluster.workers).length;
var progress = new ProgressTracker(workers, function() {
for (var key in serverHandlers) {
serverHandlers[key].close();
delete serverHandlers[key];
} }
// call callback when done
if (callback) callback();
});
// begin disconnecting all workers
eachWorker(function(worker) {
worker.once('disconnect', progress.done.bind(progress));
worker.disconnect();
});
// in case there weren't any workers
progress.check();
};
// Internal function. Called from src/node.js when worker process starts.
cluster._setupWorker = function() {
// Get worker class
var worker = cluster.worker = new Worker();
// we will terminate the worker // Returns an internalMessage listener that hands off normal messages
// when the worker is disconnected from the parent accidentally // to the callback but intercepts and redirects ACK messages.
process.once('disconnect', function() { function internal(worker, cb) {
if (worker.suicide !== true) { return function(message, handle) {
process.exit(0); if (message.cmd !== 'NODE_CLUSTER') return;
var fn = cb;
if (typeof message.ack !== 'undefined') {
fn = callbacks[message.ack];
delete callbacks[message.ack];
} }
}); fn.apply(worker, arguments);
// Tell master that the worker is online
worker.state = 'online';
sendInternalMessage(worker, { cmd: 'online' });
};
// Internal function. Called by net.js and dgram.js when attempting to bind a
// TCP server or UDP socket.
cluster._getServer = function(tcpSelf, address, port, addressType, fd, cb) {
// This can only be called from a worker.
assert(cluster.isWorker);
// Store tcp instance for later use
var key = [address, port, addressType, fd].join(':');
serverListeners[key] = tcpSelf;
// Send a listening message to the master
tcpSelf.once('listening', function() {
cluster.worker.state = 'listening';
sendInternalMessage(cluster.worker, {
cmd: 'listening',
address: address,
port: tcpSelf.address().port || port,
addressType: addressType,
fd: fd
});
});
// Request the fd handler from the master process
var message = {
cmd: 'queryServer',
address: address,
port: port,
addressType: addressType,
fd: fd
};
// The callback will be stored until the master has responded
sendInternalMessage(cluster.worker, message, function(msg, handle) {
cb(handle);
});
}; };
}

Write Preview
Loading…
Cancel
Save