Node.js Rate Limited Concurrent Execution of an Async Function

control-flow-3.js

/* Control Flow 3 - Rate Limited Concurrent Execution of an Asynchronous Function
 *
 * Based on:
 *   - http://book.mixu.net/node/ch7.html
 *   - https://github.com/mixu
 *
 * Characteristics:
 *   - Runs a number of operations concurrently
 *   - Starts a limited number of operations concurrently (partial concurrency, full concurrency control)
 *   - No guarantee of order, only that all the operations have been completed
 *
 * Variations:
 *   - The way in which the result is collected (manual or via a “stashing” callback)
 *   - How error handling is done (manually in each subfunction, or via a dedicated, additional function)
 *   - Since execution is sequential, there is no need for a “final” callback
 *
 * Tags:  parallel, partial concurrency, concurrency limited
 */

/*
 * A Simulated Asynchronous Function
 */

function async(arg, callback) {
    var delay = Math.floor(1000 * Math.random());
    console.log('Do something with "' + arg + '", and return ' + delay + 'ms later');
    setTimeout(function() {
        result = arg; // keep the result the same as the argument so async impact can be easily observed
        callback(result);
    }, delay);
}

/*
 * A Simulated Completion Function
 */

function final(results) {
    console.log('Done', results);
    console.log(process.hrtime(start)[0] + "." + (process.hrtime(start)[1] / 1000000).toFixed(0) + " sec(s) total time");
}

/*
 * Concurrent Flow Control Construct w/Concurrency Limits
 */

function limited(items, results, limit) {
    var running = 0;
    function dispatch() {
        while (running < limit && items.length > 0) {
            var item = items.shift();
            async(item, function(result) {
                running--;
                results.push(result);
                console.log("Results so far: ", results, ", " + running + " process(es) are running");
                if (items.length > 0) {
                    dispatch();
                } else if (running == 0) {
                    final(results);
                }
            });
            running++;
            console.log(running + " process(es) are running");
        }
    }
    dispatch();
}

/*
 * Main
 */

var items   = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]; // Work array
var results = [];                              // Results array
var start   = process.hrtime();                // Start a timer

limited(items, results, 3);