Throttled Group of Asynchronous Workers

throttled-worker-group.js

/**
 * Executes no more than maxConcurrency workers at once and executes the
 * given afterFinish callback when all workers have completed. Workers are
 * expected to signal when they are finished.
 *
 * @example
 *
 *   WorkerGroup(
 *     [ function(whenDone) { $.ajax({url: 'http://jquery.com/a/', error: whenDone, success: whenDone}); }
 *     , function(whenDone) { $.ajax({url: 'http://jquery.com/b/', error: whenDone, success: whenDone}); }
 *     , function(whenDone) { $.ajax({url: 'http://jquery.com/c/', error: whenDone, success: whenDone}); } ],
 *     , function(whenDone) { $.ajax({url: 'http://jquery.com/d/', error: whenDone, success: whenDone}); } ],
 *     function(returnVals) { console.log('results are in:', returnVals); }, 2);
 *
 *   // WorkerGroup returns immediately and execution resumes here.
 *   console.log('collecting data');
 */
function WorkerGroup(workers, afterFinish, maxConcurrency) {
  var pendingCount = workers.length;
  var returnVals   = [];

  var observeWorker = function(workerId) {
    // Called by a worker when it's finished
    return function(returnVal) {
      returnVals[workerId] = returnVal;

      // Start the next worker
      if (-- pendingCount > 0)
        startWorker(workers.length - pendingCount);

      if (pendingCount == 0 && afterFinish)
        afterFinish(returnVals);
    };
  };

  var startWorker = function(workerId) {
    setTimeout(function() {
      // Hmm... this worker may punt by calling setTimeout,
      // so we can't really know if he finishes unless he
      // calls the given observeWorker value
      workers[workerId](observeWorker(workerId));
    }, 0);
  }

  for (var n = 0; n < (maxConcurrency || workers.length); n ++)
    startWorker(n);
}