How to use promise.all with a specified number of parallel processing at Node.js-CodePudding

I have a list of promises: [pr1, pr2, pr3, pr4, pr5]. Because each of promise use a large amount of resources, i want at a time, alway, just only a specified number of promises is run.

I tried Promise.allSettled(), i set 2 promises run at a time (pr1, pr2), but i must wait 2 promises are done to start 2 pr next (pr3, pr4). I want if pr1 is finished soon, pr3 will replace pr1 and 2 promises now is (pr2, pr3), alway 2 promises running.

If someone has any solution, it would be appreciate to hear your solution. Thank you so much.

CodePudding user response：

It sounds like you want a buffer and a stream that take from that buffer with a specific number of back pressure. This is not something a simple Promise.all can achieve.

CodePudding user response：

I have a list of promises: [pr1, pr2, pr3, pr4, pr5]

If you have a list of 5 promises you already have 5 things processing in parallel. There is no way to stop it since you have already triggered the processes by creating the promises.

If you want to process only two of them at once you need to NOT CREATE THE PROMISES. Therefore what you need is a list of 5 functions that return promises instead of 5 promises.

What you need is an array of [f1, f2, f3, f4, f5] where f1 will return pr1, f2 will return pr2 etc.

Once you have this all you need to do is to Promise.all() two promises at a time:

const tasks = [f1, f2, f3, f4, f5];

const BATCH_IN_PARALLEL = 2;

async function batchTasks() {
    for (let i=0; i<tasks.length;) {
        let promises = [];

        // create two promises at a time:
        for (let j=0; j<BATCH_IN_PARALLEL && i<tasks.length; i  ,j  ) {
            let t = tasks[i];
            promises.push(t()); // create the promise here!
        }

        await Promise.all(promises); // wait for the two promises
    }
}

If you need the result of the promises just collect them in an array:

async function batchTasks() {
    let result = [];

    for (let i=0; i<tasks.length;) {
        let promises = [];

        // create two promises at a time:
        for (let j=0; j<BATCH_IN_PARALLEL && i<tasks.length; i  ,j  ) {
            let t = tasks[i];
            promises.push(t()); // create the promise here!
        }

        result.push(await Promise.all(promises));
    }

    return result;
}

The above is a basic implementation of batching. It only processes two async functions at a time but it waits for both to complete before processing two more. You can get creative and process another function as soon as one is done but the code for that is a bit more involved.

The async-q library has a function that does just this: asyncq.parallelLimit:

const asyncq = require('async-q');

const tasks = [f1, f2, f3, f4, f5];

let result = asyncq.parallelLimit(tasks, 2);

Additional answer

Here is some old code I found in one of my projects that continuously processes two tasks in parallel. It uses a recursive function to process the tasks array until empty. As you can see, the code is a bit more complicated but not too difficult to understand:

function batch (tasks, batch_in_parallel) {
    let len = tasks.length;
    
    return new Promise((resolve, reject)=>{
        let counter = len;

        function looper () {
            if (tasks.length != 0) {
                // remove task from the front of the array:
                // note: alternatively you can use .pop()
                //       to process tasks from the back
                tasks.shift()().then(()=>{
                    counter--;
                    if (counter) { // if we still have tasks
                        looper();  // process another task
                    }
                    else {
                        // if there are no tasks left we are
                        // done so resolve the promise:
                        resolve();
                    }
                });
            }
        }

        // Start parallel tasks:
        for (let i=0; i<batch_in_parallel; i  ) {
            looper();
        }
    });
}

// Run two tasks in parallel:
batch([f1, f2, f3, f4, f5], 2).then(console.log('done');

Note that the above function does not return results. You can modify it to collect the result in an array then return the result by passing it to resolve(result) but making sure the result is in the same order as the tasks is not trivial.

Nowdays I'd just use asyncq.parallelLimit() unless I really don't want to import the entire async-q library or my boss/client does not trust it.