task_runner.js

var MAX_ITEMS = 360;
var PAGE_SIZE = 10;
var PAGE_LOAD_THRESHOLD = 1 * 1000;

function AsyncRunner(options) {
    var self = this;
    
    var tasks = [];
    
    var intervalIndex = -1;
    
    var currentOwnerIndex = 0;
    
    var runningInterval = options.runningInterval || 100;
    
    var jobTimeout = options.jobTimeout || 5000;
    
    var job = options.job; // should not be null
    
    var data = options.data; // should not be null
    
    var endCallback = options.endCallback; // should not be null
    
    var jobEndCallback = options.jobEndCallback; // should not be null

    // initialize the tasks array
    data.forEach(function(options) {
        tasks.push({
            options: options,
            target: options.target,
            inProcess: false,
            ownerIndex: -1,
            processBegin: -1,
            done: false,
            result: null
        });
    });
    
    /* a task is outdated if it started more than 'jobTimeout' ms ago */
    function isTaskOutdated(task) {
        var currentTime = (new Date()).getTime();
        return (currentTime - task.processBegin) >= jobTimeout;
    }
    
    /* reset the task's owner, and set to not inProcess */
    function resetTask(task) {
        task.inProcess = false;
        task.ownerIndex = -1;
        task.processBegin = -  1;
    }
    
    /* reset all outdated tasks */
    function resetOutdatedTasks() {
        var task = null;
        for(var i = 0; i<tasks.length; i++) {
           task = tasks[i];
           if(!task.done && task.inProcess && isTaskOutdated(task)){
               console.log('task n°' + i + ' is outdated, reseting it.');
               resetTask(task);
           }
        }
    }
    
    /* we're done only if all tasks are done */
    function isDone() {
        for(var i = 0; i<tasks.length; i++) {
            if(tasks[i].done === false) {
                return false;
            }
        }
        
        return true;
    }
    
    /* prepare tasks outputs for user, call the callback and clears the interval */
    function onDone() {
        var data = tasks.reduce(function(res, task) {
            res[task.target] = task.result;
            return res;
        }, {});
        
        console.log('All tasks are done, calling final callback.');
        
        clearInterval(intervalIndex);
        endCallback(data);
    }
    
    /* gets the index of the next task */
    function nextTask() {
        for(var i = 0; i<tasks.length; i++) {            
            if(tasks[i].inProcess === false && tasks[i].done === false) {
                return i;
            }
        }
        
        console.log('no task available.');
        return -1;
    }
    
    /* fills a task with the result of computation */
    function completeTask(taskIndex, ownerIndex, data) {
        var task = tasks[taskIndex];
        
        // ignore the data if the owner changed (due
        // to a timeout for example)
        if(task.ownerIndex !== ownerIndex) {
            return;
        }
        
        task.inProcess = false;
        task.done = true;
        task.result = data;
        
        jobEndCallback(task.target, data);
    }
    
    /* find a task and assigns it to a runner */
    function handleTasks() {
        var nextTaskIndex = nextTask();
        
        if(nextTaskIndex === -1) {
            return; // no task available...
        }
        
        var task = tasks[nextTaskIndex];
        var ownerIndex = currentOwnerIndex++;
        
        function done(data) {
            completeTask(nextTaskIndex, ownerIndex, data);
        }
        
        task.ownerIndex = ownerIndex;
        task.inProcess = true;
        task.processBegin = (new Date()).getTime();
        
        console.log('starting task n°' + nextTaskIndex);
        job(task.options, done);
    }
    
    function stepInterval() {
        resetOutdatedTasks();
        
        if(isDone()) {
            onDone();
        }
        else {
            handleTasks();
        }
    }
    
    self.start = function() {
        console.log('starting runner');
        intervalIndex = setInterval(stepInterval, runningInterval);
    };
    
    self.stop = function() {
        console.log('stopping runner');
        clearInterval(intervalIndex);
    };
}

function range(min, max) {
    var x = [];
    for(;min<max;min++) x.push(min);
    return x;
}

function page(n, s) {
    var min=s*n;
    var max=s*(n+1);
    
    if(min >= MAX_ITEMS) return [];
    if(max > MAX_ITEMS) max=MAX_ITEMS;
    
    return range(min, max);
}

var total_pages = Math.ceil(MAX_ITEMS / PAGE_SIZE);
var data = [];
for(var i = 0; i<total_pages; i++) {
    data.push(
        /* options for the runner */
        { 
            target: i,
            page: i,
            pageSize: PAGE_SIZE
        } 
    );
}

var options = {};
options.runningInterval = 100;
options.jobTimeout = 1000;
options.data = data;

options.job = function(options, done) {
    /* 20% chances to fail */
    setTimeout(function() {
        if(Math.random() > 0.2) {
            done(page(options.page, PAGE_SIZE));
        }
    }, 500);
};

options.endCallback = function(data) {
    console.log('----------');
    console.log('Result: ');
    console.log(data);
    console.log('----------');
};

options.jobEndCallback = function(target, data) {
    console.log('----------');
    console.log('computed: ' + target);
    console.log('output: ' + data);
    console.log('----------');
};

var runner = new AsyncRunner(options);

runner.start();