Skip to content
This repository has been archived by the owner on Apr 22, 2023. It is now read-only.

Commit

Permalink
cluster: add graceful disconnect support
Browse files Browse the repository at this point in the history
This patch add a worker.disconnect() method there will stop the worker from accepting
new connections and then stop the IPC. This allow the worker to die graceful.
When the IPC has been disconnected a 'disconnect' event will emit.

The patch also add a cluster.disconnect() method, this will call worker.disconnect() on
all connected workers. When the workers are disconneted it will then close all server
handlers. This allow the cluster itself to self terminate in a graceful way.
  • Loading branch information
AndreasMadsen authored and isaacs committed Mar 19, 2012
1 parent ab32e9e commit d927fbc
Show file tree
Hide file tree
Showing 4 changed files with 426 additions and 15 deletions.
90 changes: 87 additions & 3 deletions doc/api/cluster.markdown
Expand Up @@ -118,6 +118,21 @@ where the 'listening' event is emitted.
console.log("We are now connected");
});

## Event: 'disconnect'

* `worker` {Worker object}

When a workers IPC channel has disconnected this event is emitted. This will happen
when the worker die, usually after calling `.destroy()`.

But also when calling `.disconnect()`, in this case it is possible there is delay
between the `disconnect` and `death` and the event can be used to detect if the
process is stuck in a cleanup or if there are long living connection.

cluster.on('disconnect', function(worker) {
console.log('The worker #' + worker.uniqueID + ' has disconnected');
});

## Event: 'death'

* `worker` {Worker object}
Expand Down Expand Up @@ -179,6 +194,16 @@ Spawn a new worker process. This can only be called from the master process.
All settings set by the `.setupMaster` is stored in this settings object.
This object is not supposed to be change or set manually.

## cluster.disconnect([callback])

* `callback` {Function} called when all workers are disconnected and handlers are closed

When calling this method all workers will commit a graceful suicide. When they are
disconnected all internal handlers will be closed, allowing the master process to
die graceful if no other event is waiting.

The method takes an optional callback argument there will be called when finished.

## cluster.workers

* {Object}
Expand Down Expand Up @@ -232,9 +257,8 @@ See: [Child Process module](child_process.html)

* {Boolean}

This property is a boolean. It is set when a worker dies, until then it is
`undefined`. It is true if the worker was killed using the `.destroy()`
method, and false otherwise.
This property is a boolean. It is set when a worker dies after calling `.destroy()`
or immediately after calling the `.disconnect()` method. Until then it is `undefined`.

### worker.send(message, [sendHandle])

Expand Down Expand Up @@ -273,6 +297,55 @@ a suicide boolean is set to true.
// destroy worker
worker.destroy();


## Worker.disconnect()

When calling this function the worker will no longer accept new connections, but
they will be handled by any other listening worker. Existing connection will be
allowed to exit as usual. When no more connections exist, the IPC channel to the worker
will close allowing it to die graceful. When the IPC channel is closed the `disconnect`
event will emit, this is then followed by the `death` event, there is emitted when
the worker finally die.

Because there might be long living connections, it is useful to implement a timeout.
This example ask the worker to disconnect and after 2 seconds it will destroy the
server. An alternative wound be to execute `worker.destroy()` after 2 seconds, but
that would normally not allow the worker to do any cleanup if needed.

if (cluster.isMaster) {
var worker = cluser.fork();
var timeout;

worker.on('listening', function () {
worker.disconnect();
timeout = setTimeout(function () {
worker.send('force kill');
}, 2000);
});

worker.on('disconnect', function () {
clearTimeout(timeout);
});

} else if (cluster.isWorker) {
var net = require('net');
var server = net.createServer(function (socket) {
// connection never end
});

server.listen(8000);

server.on('close', function () {
// cleanup
});

process.on('message', function (msg) {
if (msg === 'force kill') {
server.destroy();
}
});
}

### Event: 'message'

* `message` {Object}
Expand Down Expand Up @@ -342,6 +415,17 @@ on the specified worker.
// Worker is listening
};

## Event: 'disconnect'

* `worker` {Worker object}

Same as the `cluster.on('disconnect')` event, but emits only when the state change
on the specified worker.

cluster.fork().on('disconnect', function (worker) {
// Worker has disconnected
};

## Event: 'death'

* `worker` {Worker object}
Expand Down
119 changes: 107 additions & 12 deletions lib/cluster.js
Expand Up @@ -77,6 +77,19 @@ function eachWorker(cb) {
}
}

// 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();
};

cluster.setupMaster = function(options) {
// This can only be called from the master.
assert(cluster.isMaster);
Expand Down Expand Up @@ -239,7 +252,10 @@ if (cluster.isMaster) {
// Messages to a worker will be handled using this methods
else if (cluster.isWorker) {

// TODO: the disconnect step will use this
// Handle worker.disconnect from master
messageHandingObject.disconnect = function(message, worker) {
worker.disconnect();
};
}

function toDecInt(value) {
Expand Down Expand Up @@ -293,9 +309,11 @@ function Worker(customEnv) {
});
}

// handle internalMessage and exit event
// handle internalMessage, exit and disconnect event
this.process.on('internalMessage', handleMessage.bind(null, this));
this.process.on('exit', prepareDeath.bind(null, this, 'dead', 'death'));
this.process.on('disconnect',
prepareDeath.bind(null, this, 'disconnected', 'disconnect'));

// relay message and error
this.process.on('message', this.emit.bind(this, 'message'));
Expand Down Expand Up @@ -356,14 +374,6 @@ Worker.prototype.send = function() {
this.process.send.apply(this.process, arguments);
};


function closeWorkerChannel(worker, callback) {
//Apparently the .close method is async, but do not have a callback
worker.process._channel.close();
worker.process._channel = null;
process.nextTick(callback);
}

// Kill the worker without restarting
Worker.prototype.destroy = function() {
var self = this;
Expand All @@ -373,9 +383,14 @@ Worker.prototype.destroy = function() {
if (cluster.isMaster) {
// Disconnect IPC channel
// this way the worker won't need to propagate suicide state to master
closeWorkerChannel(this, function() {
if (self.process.connected) {
self.process.once('disconnect', function() {
self.process.kill();
});
self.process.disconnect();
} else {
self.process.kill();
});
}

} else {
// Channel is open
Expand Down Expand Up @@ -403,6 +418,59 @@ Worker.prototype.destroy = function() {
}
};

// The .disconnect function will close all server and then disconnect
// the IPC channel.
if (cluster.isMaster) {
// Used in master
Worker.prototype.disconnect = function() {
this.suicide = true;

sendInternalMessage(this, {cmd: 'disconnect'});
};

} else {
// Used in workers
Worker.prototype.disconnect = function() {
var self = this;

this.suicide = true;

// keep track of open servers
var servers = Object.keys(serverLisenters).length;
var progress = new ProgressTracker(servers, function() {
// there are no more servers open so we will close the IPC channel.
// Closeing the IPC channel will emit emit a disconnect event
// in both master and worker on the process object.
// This event will be handled by prepearDeath.
self.process.disconnect();
});

// depending on where this function was called from (master or worker)
// the suicide state has allready been set.
// But it dosn't really matter if we set it again.
sendInternalMessage(this, {cmd: 'suicide'}, function() {
// in case there are no servers
progress.check();

// closeing all servers graceful
var server;
for (var key in serverLisenters) {
server = serverLisenters[key];

// in case the server is closed we wont close it again
if (server._handle === null) {
progress.done();
continue;
}

server.on('close', progress.done.bind(progress));
server.close();
}
});

};
}

// Fork a new worker
cluster.fork = function(env) {
// This can only be called from the master.
Expand All @@ -414,6 +482,33 @@ cluster.fork = function(env) {
return (new cluster.Worker(env));
};

// execute .disconnect on all workers and close handlers when done
cluster.disconnect = function(callback) {
// This can only be called from the master.
assert(cluster.isMaster);

// Close all TCP handlers when all workers are disconnected
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 wasn't any workers
progress.check();
};

// Sync way to quickly kill all cluster workers
// However the workers may not die instantly
function quickDestroyCluster() {
Expand Down

0 comments on commit d927fbc

Please sign in to comment.