ParallelQueuePartitioner

Round-Robin parallel task dispatcher that executes tasks in sequence if related. The relationship is defined through the "partitionId". The partitioner works on top of any queue/broker.

Be aware that the version 2 works only with Node.js v4+

Installation

$ npm install parallel-queue-partitioner
For node 0.X (no longer supported)
$ npm install parallel-queue-partitioner@1.0.1

Usage

In a new js file, first reference the partitioner:

varparallelQueuePartitioner=require('parallel-queue-partitioner');varPartitioner=parallelQueuePartitioner.Partitioner;

then reference cluster

varcluster=require('cluster');

We need to register the jobs in each cluster. Each job must contain a parameter and return a promise!

if(cluster.isWorker){parallelQueuePartitioner.registerJob('sum',job=>{returnnewPromise(resolve=>{varsum=job.data.one+job.data.twoconsole.log("partition: %d, pid: %d, sum: %d",job.partitionId,process.pid,sum)resolve()})})parallelQueuePartitioner.registerJob('delayedSum',job=>{returnnewPromise(resolve=>{setTimeout(()=>{varsum=job.data.one+job.data.twoconsole.log("partition: %d, pid: %d, sum: %d",job.partitionId,process.pid,sum)resolve()},1000)})})}

We have to produce the messages only from the master. A job message must contain:

• An id that identifies the job
• A partitionId, jobs with the same partitionId will be executed by one process in sequence; jobs with different partitionId will be executed in parallel
• A type that defines the job to run, in our case 'sum' or 'delayedSum'
• Optional data

letstart=()=>{varpartitioner=newPartitioner({numberOfWorkers: 8//number of process workers});setTimeout(()=>{for(leti=1;i<50;i++){partitioner.enqueueJob({id: i,partitionId: i%8,//Spreading across the workerstype: "sum",//job to rundata: {one: i,two: i+1}},err=>{//Optional callback, will be executed once the job is completed, useful to send acks to a broker.console.log("sum job ended");});}for(leti=1;i<50;i++){partitioner.enqueueJob({id: i,partitionId: 10,//Only one process will execute these 50 messages in sequencetype: "delayedSum",data: {one: i,two: i*i}},err=>{console.log("sequential job ended");});}},2000);//Arbitrary delayer to wait all forks are completed}if(cluster.isMaster)start();

Configuration

The partitioner cab be configured as follow:

• numberOfWorkers, is the number of processes we want to run (default 1)
• cleanIdlePartitionsAfterMinutes, after the given minutes, a job will clean the partition that were not used during that interval (default 15 minutes)
• loggerLevel, 'debug', 'info', 'warn', or 'error' (default 'error')
• consoleLogger, true or false, enable or disable the console logger (default: true)
• fileLogger, true or false, enable or disable the file logger (default: true)
• fileLoggerPath, the path where the logs will be saved (default: "./logger")

constpartitioner=newPartitioner({numberOfWorkers: 4,cleanIdlePartitionsAfterMinutes: 30,loggerLevel: 'info',consoleLogger: false,fileLogger: false,fileLoggerPath: "./bin/partitionerLogger"});

Logger

If enabled, the logger will create the file in the folder specified by the user or, if not specified, in "./logger". The master and each worker will create its own log. The filenames are:

• master-pid-{PIDID}-partitioner.log
• worker-pid-{PIDID}-partitioner.log

Exemple how to use the partitioner with redis

"use strict"constparallelQueuePartitioner=require('parallel-queue-partitioner');constcluster=require("cluster")constprocess=require('process')constkue=require('kue')constqueue=kue.createQueue({prefix: 'queue',redis: {host: process.env.IP}})if(cluster.isWorker){parallelQueuePartitioner.registerJob('test',job=>{returnnewPromise(resolve=>{console.log("the job has been executed by %d",process.pid)resolve()})})parallelQueuePartitioner.registerJob('sequential',job=>{returnnewPromise(resolve=>{console.log("delayed in-sequence job started. Id: %d, Partition: %d, pid: %d, sequence: %d",job.id,job.partitionId,process.pid,job.data.sequence)setTimeout(()=>{console.log("delayed in-sequence job completed. Id: %d, Partition: %d, pid: %d, sequence: %d",job.id,job.partitionId,process.pid,job.data.sequence)resolve()},1000)})})}letstart=()=>{varpartitioner=newPartitioner({numberOfWorkers: 4,loggerLevel: 'debug',consoleLogger: true,fileLogger: true,fileLoggerPath: "./bin/logger"})setTimeout(()=>{queue.process('jobs',256,(job,done)=>{if(job.data.type=="test"){partitioner.enqueueJob({id: job.id,partitionId: job.data.partitionId,type: job.data.type,data: {}},err=>{if(err)console.log(err)else{console.log("test job %d done",job.id)done()}})}if(job.data.type=="sequential"){partitioner.enqueueJob({id: job.id,partitionId: job.data.partitionId,type: job.data.type,data: {sequence: job.data.sequence}},function(err){if(err===undefined){console.log("sequential job %d done",job.id)done()}else{console.log(err)}})}})},2000)}letid=0if(cluster.isMaster){console.log('pushing messages')for(leti=0;i<50;i++){queue.create('jobs',{partitionId: 0,type: "sequential",sequence: i}).save(err=>{if(err)console.log(err)})}for(leti=0;i<50;i++){queue.create('jobs',{partitionId: 1,type: "sequential",sequence: i}).save(err=>{if(err)console.log(err)})}for(leti=0;i<150;i++){queue.create('jobs',{partitionId: i%5,type: "test"}).save(err=>{if(err)console.log(err)})}start()}