Merge pull request #36 from AnyFetch/one-child-to-rule-many-tasks

one child for multiple task

Author: Quentin01

lib/helpers/child-process.js

@@ -16,10 +16,11 @@ var url = require('url');
 var rarity = require('rarity');
 
 var file_path;
-var path = '/tmp/AFH-' + crypto.randomBytes(20).toString('hex');
+var path;
 
 process.on('message', function(task) {
   file_path = task.file_path;
+  path = '/tmp/AFH-' + crypto.randomBytes(20).toString('hex');
   var hydrate = require(task.functionPath);
   async.waterfall([
     /**
@@ -73,8 +74,6 @@ process.on('message', function(task) {
       err: err,
       changes: changes
     });
-    process.disconnect();
-    process.exit(0);
   });
 });
 
@@ -87,4 +86,5 @@ process.on('SIGTERM', function() {
     catch (err) {}
   }
   process.disconnect();
+  process.exit(0);
 });

lib/helpers/hydrater.js

@@ -19,6 +19,28 @@ var lib = require('../index.js');
 
 var HydrationError = lib.HydrationError;
 
+var tasksPerProcess = process.env.TASKS_PER_PROCESS || 100;
+var taskCount = 0;
+var currentChild;
+
+var getOrForkChild = function() {
+  taskCount += 1;
+  if (taskCount < tasksPerProcess && currentChild) {
+    currentChild.removeAllListeners();
+    currentChild.stdout.removeAllListeners();
+    currentChild.stderr.removeAllListeners();
+    return currentChild;
+  }
+  else {
+    taskCount = 0;
+    if(currentChild) {
+      currentChild.kill('SIGKILL');
+    }
+    currentChild = fork(__dirname + '/child-process.js', {silent: true});
+    return currentChild;
+  }
+};
+
 module.exports = function(hydraterFunction, logger, errLogger) {
   if(!errLogger) {
     errLogger = logger;
@@ -39,7 +61,7 @@ module.exports = function(hydraterFunction, logger, errLogger) {
 
     async.waterfall([
       function performHydration(cb) {
-        var child = fork(__dirname + '/child-process.js', {silent: true});
+        var child = getOrForkChild();
         var stderr = "";
         var stdout = "";
         var timeout;
@@ -50,6 +72,10 @@ module.exports = function(hydraterFunction, logger, errLogger) {
         var cleaner = function(err, changes) {
           if(!cleaner.called) {
             cleaner.called = true;
+            if(err) {
+              currentChild.kill('SIGKILL');
+              currentChild = null;
+            }
             cb(err, changes);
           }
           if(stdout !== "") {
@@ -132,6 +158,7 @@ module.exports = function(hydraterFunction, logger, errLogger) {
               if(child.connected) {
                 child.kill('SIGKILL');
               }
+              currentChild = null;
               cleaner(null, changes);
             }, process.env.TIMEOUT / 6 || 10 * 1000);
           }

test/cleaning.js

@@ -57,9 +57,9 @@ describe('Hydration should be cleaned every time', function() {
             hydrate(task, function(err, changes) {
               // + one process when working
               changes.metadata.nodeCount.should.eql(nodeProccessesAtStart + 1);
-              // +0 process after work
+              // + one process after work if it didn't crash
               shellExec('ps aux | grep "[n]ode" -c', function(err, stdout) {
-                parseInt(stdout).should.be.eql(nodeProccessesAtStart);
+                parseInt(stdout).should.be.eql(nodeProccessesAtStart + 1);
                 cb(err);
               });
             });
@@ -79,7 +79,7 @@ describe('Hydration should be cleaned every time', function() {
         shellExec('ps aux | grep "[n]ode" -c', cb);
       },
       function setNodeProcessesNumber(stdout, stderr, cb) {
-        nodeProccessesAtStart = parseInt(stdout);
+        nodeProccessesAtStart = parseInt(stdout) - 1; // Cause we still have previous child
         cb();
       },
       function hydrateManyTimes(cb) {