nodeJS/asyncJS parallel processing a dynamic queue

Question

I have one process generating data files in a folder, about one new file every 10 seconds.

I have another nodeJS watcher, monitoring the directory, as new files coming in.

        const watcher = chokidar.watch(['data_folder']);

        watcher.on('add', (path, stats)=>{
            if (stats && stats.size > 0){
                console.log(path);
                //spawn child_process to do further processing
                spawn_child_process_to_run(path);
            }
        });

The new files are then further processed by child_process, which can take quite a long time to finish.

The question is how to queue the files, so that they can be processed in parallel, without hitting the number limits of nodeJS child process.

Community · Accepted Answer · 2020-06-20 09:12:55Z

You can use async library.

async.cargo will be useful , more info here & here

Creates a cargo object with the specified payload. Tasks added to the cargo will be processed altogether (up to the payload limit). If the worker is in progress, the task is queued until it becomes available. Once the worker has completed some tasks, each callback of those tasks is called. Check out these animations for how cargo and queue work.

While queue passes only one task to one of a group of workers at a time, cargo passes an array of tasks to a single worker, repeating when the worker is finished.

var chokidar = require('chokidar');
var async = require('async')

var cargo = async.cargo(function (tasks, callback) {
    async.map(tasks,function(task,cb){
        console.log('spawn_child_process_to_run(path);',task);
        cb();
    },callback);
}, 2);// Number of tasks in parallel 

const watcher = chokidar.watch(['data_folder']);

watcher.on('add', (path, stats)=>{
    if (stats && stats.size > 0){
        cargo.push(path);//Push payload
    }
});

Zhe Hu · Accepted Answer · 2016-05-14 18:02:18Z

0

With the help of async.queue

var async = require('async');    
var exec = require('child_process').exec;


var q = async.queue(function (path, callback) {
    console.log('hello ' + path);
    exec('ping 127.0.0.1 -n 6 >nul ', (err, stdout, stderr)=>{console.log(stdout);callback()});    //simulate 6sec processing time
   }, 4);


  // assign a callback
  q.drain = function() {
     console.log('all items have been processed');
 }



 q.push([1,2,3,4,5,6,7,8],function(){console.log("done");})

edited May 14, 2016 at 18:02

answered May 14, 2016 at 2:22

Zhe Hu

4,0474 gold badges34 silver badges46 bronze badges

Collectives™ on Stack Overflow

nodeJS/asyncJS parallel processing a dynamic queue

2 Answers 2

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related