Node.js Concurrency Limited Async Function Calls Using an Array of Functions

control-flow-5.js

/* Control Flow 5 - Concurrency Limited Functions Using an Array of Functions and Callbacks
 *
 * Based on:
 *   - http://book.mixu.net/node/ch7.html
 *   - https://github.com/mixu
 *
 * The 'Array.prototype.slice.call(arguments)' usage is a JavaScript trick and explained here:
 *   - http://mariapacana.tumblr.com/post/79170518832/javascript-tricks
 *   - http://stackoverflow.com/questions/7056925/how-does-array-prototype-slice-call-work
 *
 * Characteristics:
 *   - Runs a number of operations concurrently by passing an array of functions
 *   - Starts a limited number of operations concurrently
 *   - No guarantee of execution order, only that all the operations have been completed
 *   - However, results array is populated in order of passed functions
 *
 * Variations:
 *   - The way in which the result is collected (manual or via a “stashing” callback)
 *   - How error handling is done (manually in each sub-function, or via a dedicated, additional function)
 *   - Since execution is sequential, there is no need for a “final” callback
 *
 * Tags:  concurrent, partial concurrency, concurrency limited, array-based work queue
 */

/*
 * 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 and Array of Callbacks for a Work Queue
 */

function limited(results, callbacks, last, limit) {
    var running = 1;
    var task = 0;
    function next() {
        running--;
        console.log(running + " process(es) running");
        if (task == callbacks.length && running == 0) {
            last(results);
        }
        while (running < limit && callbacks[task]) {
            var callback = callbacks[task];
            (function(index) {
                callback(function() {
                    results[index] = Array.prototype.slice.call(arguments);
                    console.log("Results so far: ", results);
                    next();
                });
            })(task);
            task++;
            running++;
            console.log(running + " process(es) running");
        }
    }
    next();
}

/*
 * Main
 */

var start = process.hrtime();
var results = [];
limited(results, [
    function(next) {
        async(1, next);
    },
    function(next) {
        async(2, next);
    },
    function(next) {
        async(3, next);
    },
    function(next) {
        async(4, next);
    },
    function(next) {
        async(5, next);
    },
    function(next) {
        async(6, next);
    },
    function(next) {
        async(7, next);
    },
    function(next) {
        async(8, next);
    },
], final, 3);