# Process array with x parallel executions

I wanted to create a function in JavaScript using jQuery's Deferred which executes a given function for each element in an array. The function is intended for operations which take approximately the same amount of time for any element. It is also only intended for items which need to be processed through the same function (i.e. only a single one). The provided function must be executed for each array item. These executions do not have to return results but must succeed (i.e. stop) in some way.

/**
* Executes a function {func} for each item in the array {arr} with the inital amount of parallel calculations as specified in {par}
* @method runAsyncXtimesParallel
* @param {Array} arr (required) - the array containing the items to be processed
* @param {Function} func (required) - the function which should be executed for each array element
* @param {Number} parallel (required) - the amount of initial parallel executions
* @param {Boolean} taskdebug (optional) - indicates if the "task #x finished" message should be written into console
* @param {Boolean} percentagedebug (optional) - indicates if the current percentage progress should be written into console
* @return {Object} masterdeferred -  a jQuery Deferred representing the function
*/
function runAsyncXtimesParallel(arr, func, par, taskdebug, percentagedebug)
{
//type checks
if (!(arr instanceof Array)) { throw new TypeError('first parameter is expected to be an array.'); }
if (!(func instanceof Function)) { throw new TypeError('second parameter is expected to be a function.'); }
if (typeof par !== "number") { throw new TypeError('third parameter is expected to be a number.'); }
if (typeof (taskdebug || true) !== "boolean") { throw new TypeError('fourth parameter is expected to be a boolean.'); }
if (typeof (percentagedebug || true) !== "boolean") { throw new TypeError('fifth parameter is expected to be a boolean.'); }

//change par to a non floating point number
par = parseInt(par);

//create the master deferred for the function
var masterdeferred = new $.Deferred(); //start promising the deferred masterdeferred.promise(); //array for saving the task deferreds var deferreds = []; //array zum speichern der ergebnisse var results = undefined; //variable for storing the current processed count var processedcount = 0; //variables for storing the current progress percentage var percentage = 0; var tmppercentage = 0; //loop through the first array items as long as there are some left and it does not exceed the given parallel count for (var i = 0; i < par && i < arr.length; i++) { //create a new sub-deferred for each parallel task var def = new$.Deferred();
//promise the deferred
def.promise();
//save the background data for the deferred item
def.options = {
//property containing the current array item
args: arr[i],
//property containing the deferred itself
deferred: def,
//the index of the deferred (i.e. time of creation and index in deferreds array
deferredindex: 0 + i,
//the current array item index
parindex: 0 + i,
//the function to be executed
work: func,
//function to check if
next: function(result) {
//can only proceed if this is set
if (this === undefined) return;
//check if a result was provided at function call, if so add it to the results array
if (result !== undefined) { results = results || []; results.push(result); }
//check and show the current percentage progress of all items if the user wants to see it
if (percentagedebug === true)
{
//increase the processed counter
processedcount++;
//temporarily calculate the progress in percent
tmppercentage = Math.floor(processedcount * 100 / arr.length);
//check if the progess increased since the last calculation
if (tmppercentage > percentage)
{
//output the current progress to the console
console.log("state: still working, currently at " + tmppercentage + " percent");
}
//recalculate and store the current progress
percentage = Math.floor(processedcount * 100 / arr.length);
}
//check if there are still items left to process
if (this.parindex + par < arr.length)
{
//increase the current processed item index with the parallelcount
this.parindex += par;
//get the new item to be processed
this.args = arr[this.parindex];
//process the new item
this.work(this.args, this);
}
else
{
//check and show if the user wants to see when the single task has finished
if (taskdebug === true) {console.log("task #" + this.deferredindex + " finished!"); }
//no more items for this "task" to process thus resolve it
this.deferred.resolve();
}
}
};
//push the deferred into the array
deferreds.push(def);
//start working at the deferred
def.options.work(def.options.args, def.options);
}
//use jQuery to wait until all sub-deferred get resolved
$.when.apply($, deferreds).then(function() {
//save the result to the deferred object
masterdeferred.results = results;
//resolve the master deferred with the results
masterdeferred.resolve(results);
});
//return the master deferred
return masterdeferred;
}


This is how an example array will look:

//create an array to store the user items
var workitems = [];
//for the users 0 to 9999
for (var i = 0; i < 10000; i ++)
{
//create a new object representing the user, the list and some rights
workitems.push({
list: "exampleList",
user: "user." + i,
rights: [ 1, 2, 3 ]
});
}


The executing function will look like this:

function myFunc(workingitem, options) {
//issue an ajax post to the server
$.ajax({ type: "POST", url: "https://example.com/upload.php", data: workingitem, success: function(text) { //call succeeded, continue execution options.next({ status: "ok", data: text }); }, error: function(xhr, text, error) { //call failed but continue execution options.next({ status: "error", data: text }); } }); }  The main call by myself will be in the following way: //run the function for each array item with 10 parallel executions var raxp = runAsyncXtimesParallel(workitems, myFunc, 10); //wait until the function processed all array items$.when(raxp).done(function(results) {
//results are available in the {results} argument and as a property of the master deferred
//work with the result
doFurtherProcessing(raxp.results);
});

• @Gerrit0 yeak, you seem right 👌 had a typo because I changed it to this instead of def. Thanks for pointing out. Please feel free to try again :) – cramopy Aug 29 '17 at 20:22

For the most part, this looks pretty good! I do see a few issues though. I'll cover general recommendations first and then walk through the function itself.

1. taskdebug and percentagedebug should not be implemented by this function. I would instead recommend using deferred.progress and deferred.notify to let the caller of this function handle progress updates. This has two main advantages. First, it is more flexible. With this change it would be possible to update a progress bar on the page with a listener. Second, it reduces the complexity of this function. 100 lines is very long for a single function (even with comments).
2. camelCase variable names. percentageDebug is much easier to read than percentagedebug.
3. Avoid passing in more information than the function needs. What items, besides next in the options object does the work function actually need? I can't think of any instance in which the args would be needed as the argument is passed in already. Passing deferred exposes an internal object that the function shouldn't modify. deferredindex and parindex serve no purpose for the function itself as they can't be used for anything meaningful. work is just a self-reference that can be achieved simply through naming the function that is passed in to the function. All that the work function really needs is the arguments and the next function.
4. By requiring func to call the options.next function, this function mixes callbacks and deferreds. It is possible to significantly simplify the use and implementation if functions are expected to return a deferred object instead of calling options.next.

On to the function itself.

• First up, the name. runAsyncXtimesParallel is rather confusing. A more descriptive name might just be forEachParallel as this function just loops over all of the items of the array, running tasks in parallel on each item.

• Another name that could be improved is par. Without the JSDoc description, it is rather difficult to guess what this means. I would recommend threads as an alternative.

• !(arr instanceof Array) can be improved by using the Array.isArray or the $.isArray method. • !(func instanceof Function) can be improved by using $.isFunction

• Since the function calls parseInt on threads, it doesn't actually matter if the passed parameter is a number. All it needs to be is numeric. jQuery provides a function to check this. $.isNumeric • The type checks for taskDebug and percentageDebug can be improved. Since these parameters are optional, its a good idea to use default parameters to set a default value to simplify this check. function forEachParallel(arr, func, threads, taskDebug = false, percentageDebug = false) • results should not be initialized to undefined. If it is instead initialized to an empty array the results = results || [] can be dropped later on resulting in cleaner code. • There is no point in defining tmppercentage here as it does not need to be preserved between next calls. It would be better to move it into the next function. • The condition i < threads && i < arr.length can be simplified to i < Math.min(threads, arr.length). Skipping over the next function for a moment. • $.when.apply($, deferreds) can be simplified to $.when(...deferreds) by using the spread operator.

Now let's take a look at the next function.

• If this is undefined then the user tried to call the next function after assigning it to a variable. It would be better to immediately warn the user of their mistake by throwing an error as otherwise function will never resolve and the user will be left without any indication of what went wrong.

• It would be better to not assume that all threads will complete their processing at the same rate. If items 1, 3, 5, and 7 take 1 second to complete and items 0, 2, 4, 6, and 8 take 100 seconds to complete it will take 500 seconds to complete processing instead of 300.

• I don't believe it is unreasonable to assume that the user would expect the results of the parallel processing to resolve in the same order that they are passed in. Recycling the above example, the returned results array would be in this order: [1, 3, 5, 7, 0, 2, 4, 6, 8].

With all of the above in mind, here is how I would implement this function, taking advantage of ES6 features. The one thing that it does not do that the original function did is provide a progress update when a thread exists. However adding this in if you really must have that capability should be pretty simple.

function forEachParallel(arr, func, threads) {
if (!$.isArray(arr)) throw new TypeError('First parameter must be an array'); if (!$.isFunction(func)) throw new TypeError('Second parameter must be a function');
if (!$.isNumeric(threads)) throw new TypeError('Third parameter must be a number'); // The number of threads must be an integer threads = parseInt(threads); let masterDeferred = new$.Deferred();
// To hold the result of each func(arr[i]) call
let results = [];
// To hold the deferreds that must be resolved before resolving masterDeferred
let processes = [];
let percentComplete = 0;

// Map the input arr into an array of objects to preserve the index information.
let queue = arr.map((value, index) => ({index, value}))

// Create a new "process" for each item in the queue, up to the thread limit
for (let i = 0; i < Math.min(queue.length, threads); i++) {
// Note: Don't blindly change let to var here or this will break
// this depends on block scoping.
let process = new $.Deferred(); processes.push(process.promise()); (function next() { // Get the next item in the queue let item = queue.shift(); if (!item) { // If no items were found, this process is done. process.resolve(); return; } // Call the function with the value at this index func(item.value) // Then update the results with the result .done(result => results[item.index] = result) .done(() => { // Update percentage, calling any progress listeners let newPercentage = Math.floor((arr.length - queue.length) * 100 / arr.length); if (newPercentage > percentComplete) { percentComplete = newPercentage; masterDeferred.notify(percentComplete); } // Loop next(); }); }()); } // Resolve the returned deferred value once processing is complete$.when(...processes).done(() => masterDeferred.resolve(results))

return masterDeferred.promise();
}


The provided example use case changes only slightly:

function myFunc(workingitem) {
return $.ajax({ type: "POST", url: "https://example.com/upload.php", data: workingitem, }) .done(text => ({status: "ok", data: text})) .fail((_xhr, text) => ({ status: "error", data: text})) } let process = forEachParallel(workItems, myFunc, 10) process.done(results => { doFurtherProcessing(results) });  Demo: function forEachParallel(arr, func, threads) { if (!$.isArray(arr)) throw new TypeError('First parameter must be an array');
if (!$.isFunction(func)) throw new TypeError('Second parameter must be a function'); if (!$.isNumeric(threads)) throw new TypeError('Third parameter must be a number');

let masterDeferred = new $.Deferred(); let results = []; let processes = []; let percentComplete = 0; let queue = arr.map((value, index) => ({index, value})) // Create a new "process" for each item in the queue, up to the thread limit for (let i = 0; i < Math.min(queue.length, threads); i++) { let process = new$.Deferred();
processes.push(process.promise());

(function next() {
let item = queue.shift();
if (!item) {
process.resolve();
return;
}

func(item.value)
.done(result => results[item.index] = result)
.done(() => {
let newPercentage = Math.floor((arr.length - queue.length) * 100 / arr.length);
if (newPercentage > percentComplete) {
percentComplete = newPercentage;
masterDeferred.notify(percentComplete);
}
next();
});
}());
}

$.when(...processes).done(() => masterDeferred.resolve(results)) return masterDeferred.promise(); } let numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]; function myFunc(num) { let d = new$.Deferred();
// Resolve in a random order.
setTimeout(() => {
console.log('Resolving for:', num);
d.resolve(num);
}, Math.random() * 1000);
return d.promise();
}

let d = forEachParallel(numbers, myFunc, 2);
d.progress(percent => console.log('Progress:', percent));
d.done(results => console.log(results));
<script src="https://ajax.googleapis.com/ajax/libs/jquery/2.1.1/jquery.min.js"></script>