Choose proper async method for batch processing for max requests/sec

const N = 10; # threads count const D = 1000; # delay after each execution var processUser = function(user, callback){ someBusinessLogicProc(user, function(err) { setTimeout(function() { return callback(err); }, D); }); } var async = require('async') ; var people = new Array(900); async.batchMethod(people, processUser, N, finalCallback);

Putting a delay on the results is not really what you want. Instead, you want to keep track of what you've sent and when you sent it so as soon as you fall under the requests per second boundary, you can send another request.

Here's a general concept for a function that will control rate limiting for you to a fixed number of requests per second. This uses promises and requires that you supply a request function that returns a promise (if you aren't using promises now, you just need to wrap your request function in a promise).

// pass the following arguments:
//   array - array of values to iterate
//   requestsPerSec - max requests per second to send (integer)
//   maxInFlight - max number of requests in process at a time
//   fn - function to process an array value
//        function is passed array element as first argument
//        function returns a promise that is resolved/rejected when async operation is done
// Returns: promise that is resolved with an array of resolves values
//          or rejected with first error that occurs
function rateLimitMap(array, requestsPerSec, maxInFlight, fn) {
    return new Promise(function(resolve, reject) {
        var index = 0;
        var inFlightCntr = 0;
        var doneCntr = 0;
        var launchTimes = [];
        var results = new Array(array.length);

        // calculate num requests in last second
        function calcRequestsInLastSecond() {
            var now = Date.now();
            // look backwards in launchTimes to see how many were launched within the last second
            var cnt = 0;
            for (var i = launchTimes.length - 1; i >= 0; i--) {
                if (now - launchTimes[i] < 1000) {
                    ++cnt;
                } else {
                    break;
                }
            }
            return cnt;            
        }

        function runMore() {
            while (index < array.length && inFlightCntr < maxInFlight && calcRequestsInLastSecond() < requestsPerSec) {
                (function(i) {
                    ++inFlightCntr;
                    launchTimes.push(Date.now());
                    fn(array[i]).then(function(val) {
                        results[i] = val;
                        --inFlightCntr;
                        ++doneCntr;
                        runMore();
                    }, reject);
                })(index);
                ++index;
            }
            // see if we're done
            if (doneCntr === array.length) {
                resolve(results);
            } else if (launchTimes.length >= requestsPerSec) {
                // calc how long we have to wait before sending more
                var delta = 1000 - (Date.now() - launchTimes[launchTimes.length - requestsPerSec]);
                if (delta >= 0) {
                    setTimeout(runMore, ++delta);
                }

            }
        }
        runMore();
    });
}

Example Usage:

rateLimitMap(inputArrayToProcess, 9, 20, myRequestFunc).then(function(results) {
    // process array of results here
}, function(err) {
    // process error here
});

A more advanced version of this function called rateMap() is here on Github.

The general idea behind this code is this:

You pass in an array to iterate through
It returns a promise who's resolved value is an array of results (in order)
You pass a max number of requestsPerSec to ever hit
You pass a max number of requests in flight at the same time
You pass a function that will be passed an element from the array that is being iterated and must return a promise
It keeps an array of timestamps when a request was last sent.
To see if another request can be sent, it looks backwards in the array and counts how many requests were sent in the last second.
If that number is lower than the threshold, then it sends another one.
If that number meets the threshold, then it calciulates how long you have to wait to send another one and set a timer for that amount of time.
Upon completion of each request, it checks to see if it can send more
If any request rejects its promise, then the returned promise rejects immediately. If you don't want it to stop upon first error, then modify your passed in function to not reject, but to resolve with some value that you can identify as a failed request later when processing the results.

Here's a working simulation: https://jsfiddle.net/jfriend00/3gr0tq7k/

Note: If the maxInFlight value you pass in is higher than the requestsPerSec value, then this function will basically just send requestsPerSec requests and then one second later, send another requestsPerSec requests since that's the quickest way to stay under the requestsPerSec boundary. If the maxInFlight value is the same or lower than requestsPerSec then it will send requestsPerSec and then as each request finishes, it will see if it can send another one.

Recommended topics

Hot tags