Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
SIGTERM issue
// Copyright Joyent, Inc. and other Node contributors.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the
// "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish,
// distribute, sublicense, and/or sell copies of the Software, and to permit
// persons to whom the Software is furnished to do so, subject to the
// following conditions:
//
// The above copyright notice and this permission notice shall be included
// in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
// USE OR OTHER DEALINGS IN THE SOFTWARE.
//TODO: emit a finalized event when all workers are listening on all ports
var assert = require('assert');
var os = require('os');
var fork = require('child_process').fork;
var net = require('net');
var EventEmitter = require('events').EventEmitter;
var util = require('util');
var cluster = module.exports = new EventEmitter();
// Events threre exists:
// death, disconnect, fork, spawn, online, listening;
var debug;
if (process.env.NODE_DEBUG && /cluster/.test(process.env.NODE_DEBUG)) {
debug = function(x) {
var prefix = process.pid + ',' +
(process.env.NODE_WORKER_ID ? 'Worker' : 'Master');
console.error(prefix, x);
};
} else {
debug = function() { };
}
// Used in the master:
var masterStarted = false;
var ids = 0;
var serverHandlers = {};
var workerFilename;
var workerArgs;
var workerTotal;
// Used in the worker:
var serverLisenters = {};
var queryIds = 0;
var queryCallbacks = {};
cluster.isWorker = 'NODE_WORKER_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 oly used in the naster
cluster.workers = cluster.isMaster ? [] : null;
//Simple function there call a function on each worker
cluster.eachWorker = function(cb) {
// This can only be called from the master.
assert(cluster.isMaster);
// Go througe all workers
for (var id in cluster.workers) {
if (cluster.workers[id]) {
cb(cluster.workers[id]);
}
}
};
// Call this from the master process. It will start child workers.
cluster.setupMaster = function(options) {
// This can only be called from the master.
assert(cluster.isMaster);
//Don't allow this function to run more that once
if (masterStarted) return;
masterStarted = true;
//Get filename and arguments
options = options || {};
workerFilename = options.exec || process.argv[1];
workerArgs = options.args || process.argv.slice(2);
workerTotal = options.workers || os.cpus().length;
// Signal handling
process.on('SIGINT', function () {
console.log('master: SIGINT');
cluster.destroy(function () {
process.exit(0);
});
});
process.on('SIGTERM', function () {
console.log('master: SIGTERM');
cluster.destroy(function () {
process.exit(0);
});
});
process.on('SIGQUIT', function () {
console.log('master: SIGQUIT');
cluster.disconnect(function () {
process.exit(0);
});
});
process.on('SIGUSR2', cluster.restart.bind(cluster));
process.on('SIGCHLD', cluster.autoFork.bind(cluster));
/*
//This is really bad
process.on('uncaughtException', function(e) {
// Quickly try to kill all the workers.
// TODO: be session leader - will cause auto SIGHUP to the children.
cluster.eachWorker(function(worker) {
debug('kill worker ' + worker.pid);
worker.kill();
});
console.error('Exception in cluster master process: ' +
e.message + '\n' + e.stack);
console.error('Please report this bug.');
process.exit(1);
});
*/
};
//Return the number of alive workers
cluster.workersOnline = function () {
var online = 0;
//Kill all workers
cluster.eachWorker(function(worker) {
if ({'online': 1, 'listening': 1}[worker.state]) {
online += 1;
}
});
return online;
};
//Destroy all workers and itself
cluster.destroy = function(callback) {
// This can only be called from the master.
assert(cluster.isMaster);
//Kill all workers
cluster.eachWorker(function(worker) {
console.log('kill worker');
worker.kill();
});
//Call callback
if(callback) callback();
};
//disconnect all workers
cluster.disconnect = function(callback) {
// This can only be called from the master.
assert(cluster.isMaster);
//How many workers are online
var total = cluster.workersOnline();
var disconnected = 0;
cluster.eachWorker(function(worker) {
//When all workers are disconnected call callback
worker.once('disconnect', function () {
disconnected += 1;
if (disconnected === total) {
if(callback) callback();
}
});
//Disconnect all workers
worker.disconnect();
});
};
cluster.restart = function() {
// This can only be called from the master.
assert(cluster.isMaster);
//First disconnect then fork
cluster.disconnect(function () {
cluster.autoFork();
});
};
// Check if a message is internal only
function isInternalMessage(message) {
return (message !== null &&
typeof message === 'object' &&
message._internal === true);
}
// Check if a message require echo
function requireEcho(message) {
return (message !== null &&
typeof message === 'object' &&
message.hasOwnProperty('_requestEcho'));
}
// Checi if a echo is recived
function receiveEcho(message) {
return (message !== null &&
typeof message === 'object' &&
message.hasOwnProperty('_queryEcho'));
}
// Create a internal message object
function internalMessage(sending, receiving) {
sending = sending || {};
receiving = receiving || {};
sending._internal = true;
if (requireEcho(receiving)) {
sending._queryEcho = receiving._requestEcho;
}
return sending;
}
// Handle messages from both master and workers
var messageHandingObject = {};
function handleMessage(message, handle, worker) {
//Run handler if it exist
if (message.cmd && typeof messageHandingObject[message.cmd] === 'function') {
messageHandingObject[message.cmd](message, worker);
}
//Else check for callback request
else {
if (requireEcho(message)) {
worker.send(internalMessage({}, message));
}
if (receiveEcho(message)) {
queryCallbacks[message._queryEcho](message, handle);
delete queryCallbacks[message._queryEcho];
}
}
}
//Messages to the master will be handled using this methods
if (cluster.isMaster) {
//Handle online messages from workers
messageHandingObject.online = function(message, worker) {
worker.state = 'online';
debug('Worker ' + worker.process.pid + ' online');
cluster.emit('online', worker);
//Send echo if requested
if (requireEcho(message)) {
worker.send(internalMessage({}, message));
}
};
//Handle queryServer messages form workers
messageHandingObject.queryServer = function(message, worker) {
//This sequence of infomation is unique to the connection but not the worker
var args = [message.address, message.port, message.addressType];
var key = args.join(':');
var handler;
if (serverHandlers.hasOwnProperty(key)) {
handler = serverHandlers[key];
} else {
handler = serverHandlers[key] = net._createServerHandle.apply(net, args);
}
//echo callback id, with the fd handler associated with it
worker.send(internalMessage({}, message), handler);
};
//Handle listening messages from workers
messageHandingObject.listening = function(message, worker) {
worker.state = 'listening';
//Emit a listining now that we know the worker is listning
cluster.emit('listening', worker, {
address: message.address,
port: message.port,
addressType: message.addressType
});
//Send echo if requested
if (requireEcho(message)) {
worker.send(internalMessage({}, message));
}
};
//Handle suicide messages from workers
messageHandingObject.suicide = function(message, worker) {
worker.suicide = true;
//Send echo if requested
if (requireEcho(message)) {
worker.send(internalMessage({}, message));
}
};
//Handle disconnect messages from workers
messageHandingObject.disconnect = function(message, worker) {
if (message.state === 'setState') {
worker.state = 'disconnect';
worker.suicide = true;
//Send echo if requested
if (requireEcho(message)) {
worker.send(internalMessage({}, message));
}
}
else if (message.state === 'done') {
//Send echo if requested before closing channel
if (requireEcho(message)) {
worker.send(internalMessage({}, message));
}
worker.process._channel.close();
worker.emit('disconnect', worker);
cluster.emit('disconnect', worker);
}
};
}
//Messages to a worker will be handled using this methods
else if (cluster.isWorker) {
//Handle disconnect messages from master
messageHandingObject.disconnect = function(message, worker) {
//Run disconnect
worker.disconnect();
//Send echo if requested
if (requireEcho(message)) {
worker.send(internalMessage({}, message));
}
};
}
// Create a worker call there works both for master and worker
function Worker(env) {
if (!(this instanceof Worker)) return new Worker();
var self = this;
//Assign an id and state
Object.defineProperty(this, 'workerID', {
value: (cluster.isMaster ? ++ids : parseInt(process.env.NODE_WORKER_ID, 10))
});
this.state = 'none';
//Create or get process
if (cluster.isMaster) {
var envCopy = env || {};
for (var x in process.env) {
envCopy[x] = process.env[x];
}
envCopy['NODE_WORKER_ID'] = this.workerID;
self.process = fork(workerFilename, workerArgs, { env: envCopy });
} else {
self.process = process;
}
//Handle message
this.process.on('message', function(message, handle) {
debug('recived: ', message);
//If this is an internal message handle it and ignore the rest
if (isInternalMessage(message)) {
handleMessage(message, handle, self);
return undefined;
}
//Check if a echo is required
if (requireEcho(message)) {
self.send({ _queryEcho: message._requestEcho });
}
//Check if a echo is recived
if (queryEcho(message)) {
queryCallbacks[msg._queryEcho](msg, handle);
delete queryCallbacks[msg._queryEcho];
}
//Emit message
self.emit('message', message, self);
});
//Handle exit
self.process.on('exit', function() {
debug('worker id=' + self.workerID + ' died');
//set state to dead so the user can check for old worker objects
self.state = 'dead';
//Make suicide a boolean
self.suicide = !!self.suicide;
//Remove from workers in the master
if (cluster.isMaster) {
delete cluster.workers[self.workerID];
}
//Emit exit and death
self.emit('exit', self);
cluster.emit('death', self);
});
//Handle disconnect
self.on('disconnect', function() {
//set state to disconnect
self.disconnect = 'disconnect';
//Make suicide a boolean
self.suicide = !!self.suicide;
//Remove from workers in the master
if (cluster.isMaster) {
delete cluster.workers[self.workerID];
}
//Emit exit and death
self.emit('exit', self);
cluster.emit('death', self);
});
}
util.inherits(Worker, EventEmitter);
//Send message to worker or master
Worker.prototype.send = function(/*message, handler, callback*/) {
debug('send ' + JSON.stringify(message));
//Exist callback
var callback = arguments[arguments.length - 1];
if (typeof callback !== 'function') {
callback = undefined;
}
//Get message and handler as array
var slice = Array.prototype.slice;
var sliceTo = callback ? arguments.length - 1 : arguments.length;
var message = slice.call(arguments, 0, sliceTo);
// Store callback for later.
if (callback) {
message[0] = message[0] || {};
// Grab some random requestEcho string
message[0]._requestEcho = this.workerID + ':' + (++queryIds);
queryCallbacks[message[0]._requestEcho] = callback;
}
// Send message
this.process.send.apply(this.process, message);
};
// Kill the worker without restarting
Worker.prototype.kill = function() {
var self = this;
this.suicide = true;
if (cluster.isMaster) {
this.process.kill();
} else {
//Inform master that is is suicide and then kill
this.send(internalMessage({cmd: 'suicide'}), function() {
self.process.kill();
});
}
};
// Kill the worker without restarting
Worker.prototype.disconnect = function() {
var self = this;
if (cluster.isMaster) {
//Inform worker that is should disconnect from the master
this.send(internalMessage({cmd: 'disconnect'}));
} else {
//Inform master that about state and suicide and make it emit disconnect
this.send(internalMessage({
cmd: 'disconnect', state: 'setState'
}), function() {
var item;
//Predefine closeState
var closeState = {master: false};
for (item in serverLisenters) {
if (serverLisenters.hasOwnProperty(item)) {
closeState[item] = false;
}
}
//Check closeState
var setState = function(name) {
return function() {
//Set State
closeState[name] = true;
//Check all closeStates
var state;
for (state in closeState) {
if (closeState.hasOwnProperty(state) && closeState === false) {
return undefined;
}
}
//Emit a disconnect if all closeState are true
self.emit('disconnect', self);
};
};
//Close TCP connections
for (item in serverLisenters) {
if (serverLisenters.hasOwnProperty(item)) {
//Close TCP connection and set closeState when done
serverLisenters[item].once('end', setState(item));
serverLisenters[item].close();
}
}
//Make master emit a disconnect event
self.send(internalMessage({
cmd: 'disconnect', state: 'done'
}), function() {
//Close connection to the master
setState('master')();
});
});
}
};
// 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 initalized
cluster.setupMaster();
// Create and store worker
var worker = new Worker(env);
cluster.workers[worker.workerID] = worker;
//Emit a fork event
cluster.emit('fork', worker);
return worker;
};
// Spawn all necessary workers
cluster.autoFork = function() {
// This can only be called from the master.
assert(cluster.isMaster);
//Make sure that the master is inialized
cluster.setupMaster();
//Make sure that there hasn't been spawned any workers
if (cluster.workersOnline() >= workerTotal) {
return;
}
//Spawn workers
var i = workerTotal;
while (i--) {
cluster.fork();
}
//Restart workers when they die
cluster.on('death', function(worker) {
if (worker.suicide === false) {
debug('worker ' + worker.process.pid + ' died. restart...');
cluster.fork();
}
});
};
// Internal function. Called from src/node.js when worker process starts.
cluster._setupWorker = function() {
// This can only be called from a worker.
assert(cluster.isWorker);
// Get worker class
var worker = cluster.worker = new Worker();
// signal handlers
process.on('SIGINT', function () {
console.log("worker: SIGINT");
worker.kill.bind(worker);
});
process.on('SIGTERM', function () {
console.log("worker: SIGTERM");
worker.kill.bind(worker);
});
process.on('SIGQUIT', worker.disconnect.bind(worker));
//Tell master that the worker is online
worker.state = 'online';
worker.send(internalMessage({ cmd: 'online' }));
};
// Internal function. Called by lib/net.js when attempting to bind a server.
cluster._getServer = function(tcpSelf, address, port, addressType, cb) {
// This can only be called from a worker.
assert(cluster.isWorker);
//Store tcp instance for later use
var key = [address, port, addressType].join(':');
serverLisenters[key] = tcpSelf;
//Send a listening message to the master
tcpSelf.once('listening', function() {
cluster.worker.state = 'listening';
cluster.worker.send(internalMessage({
cmd: 'listening',
address: address,
port: port,
addressType: addressType
}));
});
//Request the fd handler from the master process
var message = internalMessage({
cmd: 'queryServer',
address: address,
port: port,
addressType: addressType
});
// The callback will be stored until the master has responed
cluster.worker.send(message, function(msg, handle) {
cb(handle);
});
};
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.