Used deathQueue to kill the unresponsive worker and create a new successor

Changed the name convention
Added test case and modified test/lib/server.js to support the test case

Author: zhuwang

examples/test-samples/nanny-feature-test.js

@@ -82,7 +82,7 @@ Cluster.prototype.listen = function(createApp, cb) {
             connThreshold: options.connThreshold || 10000, // recycle workers after this many connections
             uptimeThreshold: options.uptimeThreshold || 3600 * 24, // 24 hours (uptimeThreshold is in seconds)
             heartbeatInterval: options.heartbeatInterval,
-            maxHbInterval: options.maxHbInterval,
+            maxHeartbeatDelay: options.maxHeartbeatDelay,
             emitter: self
         });
 

lib/index.js

@@ -49,7 +49,7 @@ var Process = module.exports = function Process(options) {
         workersKilled: 0
     };
 
-    this.options.maxHbInterval = options.maxHbInterval || 3*60000; //default 3 mins
+    this.options.maxHeartbeatDelay = options.maxHeartbeatDelay || 3*60000; //default 3 mins
 
     this._heartbeats = [];
 
@@ -157,7 +157,7 @@ var Process = module.exports = function Process(options) {
                     // update the last heartbeat time for the worker
                     var workerStats = self.stats.workers[message.pid];
                     //console.log('heartbeat ' + process.pid);
-                    workerStats.lastHbTime = Date.now();
+                    workerStats.lastHeartbeatAt= Date.now();
                 }
 
                 self._heartbeatScheduler = self._heartbeatScheduler || setInterval(function () {
@@ -203,20 +203,16 @@ var Process = module.exports = function Process(options) {
                     // Check the last heartbeat time of all the workers
                     _.each(self.stats.workers, function (workerStats, pid) {
                         var now = Date.now();
-                        //console.log(workerStats.lastHbTime + ', ' + self.options.maxHbInterval);
-                        if (now - workerStats.lastHbTime > self.options.maxHbInterval) {
-                            // this worker hasn't been sending heartbeat for maxHbInterval
-                            console.log(util.format('[Cluster2] Detected worker%d is not responsive for %d', pid, now - workerStats.lastHbTime));
-                            // create a replacement immediatly
-                            var replacement = self.createWorker();
-                            self.workers[replacement.pid] = replacement;
-                            console.log(util.format('[Cluster2] Created a new worker with pid %d', replacement.pid));
-                            // kill the dead worker
-                            var workerFileName = util.format('%s/worker.%d.pid', self.options.pids, pid);
-                            console.log(util.format('[Cluster2] Kill the orignial worker %s', workerFileName));
-                            self.kill(workerFileName, 'SIGINT', function deleteEntry() {
-                                delete self.workers[pid];
-                                delete self.stats.workers[pid];
+                        if (now - workerStats.lastHeartbeatAt> self.options.maxHeartbeatDelay) {
+                            // this worker hasn't been sending heartbeat for maxHeartbeatDelay 
+                            log(util.format('[Cluster2] Detected worker%d is not responsive for %d', pid, now - workerStats.lastHeartbeatAt));
+                            var deathQueue = require('./misc').deathQueue;
+                            deathQueue(self.workers[pid], self.emitter, function () {
+                                // create a successor
+                                var successor = self.createWorker();
+                                self.workers[successor.pid] = successor;
+                                log(util.format('[Cluster2] Created a new worker with pid %d', successor.pid));
+                                return successor;
                             });
                         }
                     });

lib/process.js

@@ -26,6 +26,17 @@ server.get('/', function(req, res) {
     }
 });
 
+// test nanny feature
+server.get('/nanny-feature-test', function (req, res) {
+    // the first worker that handles the request will run away
+    res.send(process.pid + '');
+    console.log('worker ' + process.pid + ' will run away after 2s');
+    setTimeout(function () {
+        console.log('worker ' + process.pid + ' runs away');
+        clearInterval(process.heartbeat);
+    }, 2000);
+});
+
 server.on('close', function() {
     serving = false;
 })
@@ -40,12 +51,14 @@ var c = new Cluster({
     ecv: {
         control: true
     },
-    heartbeatInterval: process.env["heartbeatInterval"] || 1000
+    heartbeatInterval: process.env["heartbeatInterval"] || 1000,
+    maxHeartbeatDelay: process.env["maxHeartbeatDelay"] || 3000
 });
 
 c.on('died', function(pid) {
-    console.log('Worker ' + pid + ' died');
+    //console.log('Worker ' + pid + ' died');
     process.send({
+        pid: pid,
         dead: true
     })
 });

test/lib/server.js

@@ -0,0 +1,60 @@
+'use strict';
+
+var should = require('should'),
+    request = require('request'),
+    _ = require('underscore'),
+    spawn = require('child_process').spawn,
+    EventEmitter = require('events').EventEmitter,
+    emitter = new EventEmitter();
+
+describe('Test Cluster2 Nanny Feature', function () {
+
+    var childProc;
+
+    before(function (done) {
+        this.timeout(10000);
+        var env = {};
+        _.extend(env, process.env);
+        _.extend(env, {
+            host: '127.0.0.1',
+            port: 3000,
+            noWorkers: 2,
+            heartbeatInterval: 500,
+            maxHeartbeatDelay: 1500
+        });
+
+        childProc = spawn('node', ['test/lib/server.js'], {
+            env: env,
+            stdio: ['pipe', 1, 2, 'ipc']
+        });
+
+        childProc.once('message', function (msg) {
+            if (msg.ready) {
+                return done();
+            }
+        });
+        //setTimeout(done, 5000);
+    });
+
+    it('Should get the run away worker id and get notified when the worker died', function (done) {
+        var pid;
+        this.timeout(10000);
+        request.get('http://127.0.0.1:3000/nanny-feature-test', function (err, res, body) {
+            should.not.exist(err);
+            body.should.be.ok;
+            pid = parseInt(body);
+            //console.log('body: ' + pid);
+        });
+        childProc.on('message', function (msg) {
+            if (msg.dead) {
+                msg.pid.should.equal(pid);
+                return done();
+            }
+        });
+    });
+    
+    after(function (done) {
+        childProc.kill('SIGKILL');
+        done();
+    });
+});