~ 0.4.23 much more rigorous deathQueue

Author: huzhou

lib/misc.js

@@ -14,10 +14,12 @@
  * limitations under the License.
  */
 
-var fs = require('fs'),
-    assert = require('assert'),
+var _ = require('underscore'),
+    fs = require('fs'),
+    util = require('util'),
     when = require('when'),
-    timeout = require('when/timeout');
+    timeout = require('when/timeout'),
+    assert = require('assert');
 
 // Utility to ensure that certain directories exist
 exports.ensureDir = function(dir, clean) {
@@ -53,7 +55,7 @@ exports.forMemoryNum = function(memory) {
 };
 
 
-var tillPrevDeath = null;
+/*var tillPrevDeath = null;
 
 exports.deathQueue = function deathQueue(worker, emitter, success){
 
@@ -105,4 +107,144 @@ exports.deathQueue = function deathQueue(worker, emitter, success){
         //some one in the queue already, wait till prev death and then start `die`
         afterDeath = tillPrevDeath = timeout(60000, tillPrevDeath.ensure(die));
     }
+};*/
+
+exports.safeKill = function(pid, signal, logger){
+
+    try{
+        process.kill(pid, signal);
+        return false;
+    }
+    catch(e){
+        //verify error is Error: ESRCH
+        logger.debug('[shutdown] safeKill received:%j', e);
+        return e.errno === 'ESRCH'; //no such process
+    }
 };
+
+exports.deathQueueGenerator = function(options){
+
+    options = options || {};
+
+    var tillPrevDeath = null,
+        queue = options.queue || [],
+        wait = options.timeout || 60000,
+        retry = options.retry || 3,
+        logger = options.logger || {
+            'debug' : function(){
+                console.log.apply(console, arguments);
+            }
+        };
+
+    return function deathQueue(worker, emitter, success){
+
+        assert.ok(worker);
+        assert.ok(emitter);
+        assert.ok(success);
+
+        var pid = worker.pid,
+            death = util.format('worker-%d-died', pid);
+
+        if(!_.contains(queue, pid)){
+
+            queue.push(pid);
+
+            var tillDeath = when.defer(),
+                afterDeath = null,
+                die = function die(retry){
+
+                    if(!retry){
+                        if(tillPrevDeath){
+                            tillPrevDeath.reject(new Error('[deathQueue] failed after retries'));
+                        }
+                        tillPrevDeath = null;//reset
+                    }
+                    
+                    var successor = success(),
+                        successorPid = successor.process.pid,
+                        successorGuard = setTimeout(function onSuccessorTimeout(){
+                            //handle error case of successor not 'listening' after started
+                            exports.safeKill(pid, 'SIGTERM', logger);
+                            
+                            logger.debug('[deathQueue] successor:%d did not start listening, kill by SIGTERM', successorPid);
+                            //cancel onListening event handler of the dead successor
+                            emitter.removeListener('listening', onSuccessorListening);
+                            //retry of the 'die' process
+                            die(retry - 1);
+                            
+                        }, wait);
+
+                    //when successor is in place, the old worker could be discontinued finally
+                    emitter.on('listening', function onSuccessorListening(onboard){
+
+                        if(successorPid !== onboard){
+                            return; //noop
+                        }
+                        else{
+                            emitter.removeListener('listening', onSuccessorListening);
+                        }
+
+                        clearTimeout(successorGuard);
+                        logger.debug('[deathQueue] successor:%d of %d is ready, wait for %s and timeout in:%dms', successorPid, pid, death, wait);
+
+                        var deathGuard = setTimeout(function(){
+
+                            if(!exports.safeKill(pid, 'SIGTERM', logger)){
+                                //worker still there, should emit 'exit' eventually
+                                logger.debug('[deathQueue] worker:%d did not report death by:%d, kill by SIGTERM', pid, wait);
+                                //remove the redundant exit listener
+                                emitter.removeListener('died', onDeath);
+                            }
+                            else{//suicide or accident already happended, process has run away
+                                //we emit this from master on behalf of the run away process.
+                                logger.debug('[deathQueue] worker:%d probably ran away, emit:%s on behalf', death);
+                                //immediately report death to the master
+                                emitter.emit('died', pid);
+                            }
+
+                        }, wait);
+
+                        worker.kill('SIGINT');
+                        
+                        emitter.on('died', function onDeath(dismiss){
+
+                            if(pid !== dismiss){
+                                return;
+                            }
+                            else{
+                                emitter.removeListener('died', onDeath);
+                            }
+
+                            logger.debug('[deathQueue] %d died', pid);
+
+                            clearTimeout(deathGuard);//release the deathGuard
+                            
+                            tillDeath.resolve(pid);
+
+                            if(tillPrevDeath === afterDeath){//last of dyingQueue resolved, clean up the dyingQueue
+
+                                logger.debug('[deathQueue] death queue cleaned up');
+
+                                tillPrevDeath = null;
+                                
+                                queue.splice(0, queue.length);
+                            }
+                        });
+                    });
+                };
+
+            if(!tillPrevDeath){//1st in the dying queue,
+                afterDeath = tillPrevDeath = tillDeath.promise;//1 min
+                die(retry);
+            }
+            else{
+                afterDeath = tillPrevDeath = tillPrevDeath.ensure(_.bind(die, null, retry));
+            }
+        }
+    };
+    
+};
+
+exports.deathQueue = exports.deathQueueGenerator({
+    'timeout': 60000
+});

package.json

@@ -3,12 +3,16 @@
     "contributors": [{
         "name": "Subbu Allamaraju",
         "email": "subbu@ebaysf.com"
+    }, 
+    {
+        "name": "Roy Zhou",
+        "email": "huzhou@ebay.com"
     }],
     "name": "cluster2",
-    "version": "0.4.17",
+    "version": "0.4.23",
     "repository": {
         "type": "git",
-        "url": "https://github.com/ql-io/cluster2"
+        "url": "https://github.com/cubejs/cluster2"
     },
     "engines": {
         "node": ">= 0.8.0"
@@ -34,5 +38,10 @@
     "scripts": {
         "test": "nodeunit test"
     },
-    "optionalDependencies": {}
+    "optionalDependencies": {
+
+    },
+    "publishConfig": {
+        "registry": "https://registry.npmjs.org"
+    }
 }