cluster2 - npm Package Compare versions

Comparing version 0.4.22 to 0.4.23

lib/misc.js

@@ -17,6 +17,8 @@ /*
 
-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');
 
@@ -57,3 +59,3 @@ // Utility to ensure that certain directories exist
 
-var tillPrevDeath = null;
+/*var tillPrevDeath = null;
 
@@ -110,2 +112,142 @@ exports.deathQueue = function deathQueue(worker, emitter, success){
     }
+};*/
+
+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.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

 {
-  "author": "ql.io",
-  "contributors": [
+    "author": "ql.io",
+    "contributors": [{
+        "name": "Subbu Allamaraju",
+        "email": "[e-mail]"
+    }, 
     {
-      "name": "Subbu Allamaraju",
-      "email": "[e-mail]"
+        "name": "Roy Zhou",
+        "email": "[e-mail]"
+    }],
+    "name": "cluster2",
+    "version": "0.4.23",
+    "repository": {
+        "type": "git",
+        "url": "https://github.com/cubejs/cluster2"
+    },
+    "engines": {
+        "node": ">= 0.8.0"
+    },
+    "main": "lib/index.js",
+    "dependencies": {
+        "underscore": "~1.4.4",
+        "express": "~2.5.11",
+        "ejs": "~0.8.4",
+        "npm": "~1.3.0",
+        "when": "~2.4.0",
+        "memwatch": "~0.2.2",
+        "usage": "~0.3.8"
+    },
+    "devDependencies": {
+        "websocket": "~1.0.8",
+        "nodeunit": "~0.8.0",
+        "request": "~2.21.0",
+        "mocha": "~1.11.0",
+        "should": "~1.2.2",
+        "harbor": "~0.2.0"
+    },
+    "scripts": {
+        "test": "nodeunit test"
+    },
+    "optionalDependencies": {},
+    "publishConfig": {
+        "registry": "https://registry.npmjs.org/"
     }
-  ],
-  "name": "cluster2",
-  "version": "0.4.22",
-  "repository": {
-    "type": "git",
-    "url": "https://github.com/ql-io/cluster2"
-  },
-  "engines": {
-    "node": ">= 0.8.0"
-  },
-  "main": "lib/index.js",
-  "dependencies": {
-    "underscore": "~1.4.4",
-    "express": "~2.5.11",
-    "ejs": "~0.8.4",
-    "npm": "~1.3.0",
-    "when": "~2.4.0",
-    "memwatch": "~0.2.2",
-    "usage": "~0.3.8"
-  },
-  "devDependencies": {
-    "websocket": "~1.0.8",
-    "nodeunit": "~0.8.0",
-    "request": "~2.21.0",
-    "mocha": "~1.11.0",
-    "should": "~1.2.2",
-    "harbor": "~0.2.0"
-  },
-  "scripts": {
-    "test": "nodeunit test"
-  },
-  "optionalDependencies": {},
-  "publishConfig": {
-    "registry": "http://registry.npmjs.org/"
-  }
 }

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

Possible typosquat attack

Supply chain risk

There is a package with a similar name that is downloaded much more often.

Did you mean
cluster

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

128101

increased by16.06%

Number of package files

40

increased by48.15%

Lines of code

1959

increased by5.83%

Number of critical supply chain risk alerts

0

decreased by-100%

No dependency changes