Process array with x parallel executions

Question

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.

Please help me stating problems in my new code and improving it.

/**
* 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
– cramopy, Commented Aug 29, 2017 at 20:22

Gerrit0 · Accepted Answer · 2017-08-30 01:38:01Z

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.

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).
camelCase variable names. percentageDebug is much easier to read than percentagedebug.
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.
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');

    threads = parseInt(threads);

    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>

Stack Exchange Network

Process array with x parallel executions

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Process array with x parallel executions

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions