added PROCESSORS_PARALLELISM parameter to allow controlling how many …

…data processors execute in parallel

README.md

@@ -55,6 +55,7 @@ Same storage path used by result upload service must be accessible to result pro
 Supported environment parameters:
 - AMQ_USER, AMQ_PASSWORD for authenticating to RabbitMQ. This is temporary until a better way to retrieve credentials by services is available.
 - PROCESSORS_PATH - path where result processors can be found. If not specified, "processors" directory is used
+- PROCESSORS_PARALLELISM - number of data processors that can be executed in parallel. If not present, number of CPU cores is used.
 - MSGW_PORT - port for metrics gateway. If not specified, 8080 is used
 - METRICS_BATCH_SIZE - batch size to use when sending data to metrics gateway
 

service/notification.js

@@ -10,6 +10,7 @@ var log4js = require('log4js');
 var amqp = require('amqplib');
 var VError = require('verror');
 var when = require('when');
+var os = require('os');
 
 var logger = log4js.getLogger('notification.js');
 
@@ -34,6 +35,13 @@ function getAmqCredentials() {
     };
 }
 
+/**
+ * Returns number of processors that may execute in parallel.
+ */
+function getParallelism() {
+    return Number(process.env.PROCESSORS_PARALLELISM) || os.cpus().length;
+}
+
 /**
  * Returns controller that can be used to ack/nack given message.
  *
@@ -113,6 +121,10 @@ function initQueues(channel, processorDescs, msgConsumer) {
 function initChannel(conn, processorDescs, msgConsumer) {
     // TODO: consider channel per processor, but since Node.js is 1 thread maybe its not needed
     return conn.createChannel().then(function(ch) {
+        var parallelism = getParallelism();
+        logger.info('Allowing up to ' + parallelism + ' parallel data processor executions');
+        ch.prefetch(parallelism, true);
+
         // TODO: handle channel recreation in case of close caused by error
         function onClose() {
             channel = null;

service/processors.js

@@ -214,8 +214,8 @@ function executeProcessor(processorDesc, processingMetadata, contentMetadata) {
     child.once('error', function(err) {
         logger.debug('Execution of processor failed ', err);
     });
-    child.once('exit', function(code) {
-        logger.debug('Processor exited with ' + code);
+    child.once('exit', function() {
+        logger.debug('Processor exited');
     });
     child.once('close', function() {
         readStream.destroy();