multi-tasks

Multi-tasks is a toolkit to manage long-term and large-scale parallel computing tasks. It manages progress and tasks based on the file system, which means that even if the host crashes, tasks can be resumed based on file records.

Install:

npm install multi-tasks

How to use:

//see examples/example0
let multiTasks = require('multi-tasks').multiTasks;

//Step1, create your tasks that need to be executed simultaneously as an array.
let alltasks = [];
for(let i=0;i<50;i++){
    alltasks.push({
        name: `task-${i}`,
        data: `This prop is for a subtask`
    });
};

//Step2, provide a function to process a certain sub-task and return the result data
let processTask = (task, helper)=>{
    let {taskCount} = task;//get your task data

    //Run your processing logic here...

    //then return the result as a plain javascript object or an async promise
    return 'result data'; 
};

//Step3, run!
multiTasks({
    initialTasks: alltasks, 
    processTask, 
    taskRootFolder: `../examples-tmp-data/example0`, //a directory to store progress and results files, you can check the progress here
    taskId: 'my-task',
    numberOfWorkers: 3, //how many workers are working in parallel
    //autoCloseAfterCompletion: true, //if you have dynamically generated new tasks, put this as false
    shouldTerminate:(info)=>{
        //return true if you need to terminate the whole process
    },
    onFinish: (report)=>{
        console.log('finish callback', report);
    }
});

More about the task processing function

//Example1, return a promise for async processes,
//          you can return a promise or a non-promise result, 
//          all result data can be found in the results/succ folder
let processTask = (task, helper)=>{
    let {taskCount} = task;

    return new Promise((resolve, reject)=>{
        resolve({
            data:`task${taskCount} complete`
        })
    })
};

//Example2, dynamically create a new task while processing
let processTask = (task, helper)=>{
    let {taskCount} = task;
    
    if(taskCount % 2 === 0){
        //create a new task if needed
        helper.createNewTasks({
            msg:'a new task'
        });
        return;
    }

    return new Promise((resolve, reject)=>{
        resolve({
            data:`task${taskCount} complete`
        })
    })
};

//Example3, generate/throw exceptions in a task method
let processTask = (task, helper)=>{
    let {taskCount} = task;

    //This is the demo of exceptions/errors, they will be captured and saved in the results/errors folder
    if(taskCount===3) throw 'exception';
    if(taskCount===4) return Promise.reject({err:'a test error'});//use Promise.reject method
    if(taskCount===5) aaa =  bbb;//this undefined exception will be captured by multi-tasks

    return {data:'succ'};
}

Resuming

Sometimes the task execution is interrupted due to some reasons (such as power outage), you can resume the execution like this

multiTasks({
    initialTasks: `/myworks/my_scan_tasks/`, //Point 'initialTasks' to the interrupted task directory, multi-tasks will read the tasks in the 'new' folder and initialize them to 'initialTasks' and then continue execution
    ...
    ...//Other configurations remain unchanged
    ...
});

Changelog:

• 2.2.0 Support Resuming from an interrupted task
• 2.1.2 Update readme
• 2.1.1 Fix: recreate a new one when a worker collapsed unexpectedly.
• 2.1.0 Add a new output directory "log" where you can view the process time of each subtask
• 2.0.9 Update readme examples
• 2.0.8 Update readme examples
• 2.0.7 Update changelog
• 2.0.6 Update readme, remove failed examples
• 2.0.5 Update readme examples
• 2.0.4 Support resume from a failed task
• 2.0.3 Fix: mkdir bug on windows
• 2.0.2 new feature: support shouldTerminate
• 2.0.1 Avoid possible I/O conflicts.
• 2.0.0 Rewritten with a new architecture to support dynamic tasks.
• 1.2.8 Fix: create task folder failed on MacOS
• 1.2.7 Small updates
• 1.2.6 Support onFinish event
• 1.2.5 Rename numberOfWorks to numberOfWorkers, the old one are still supported ;-)
• 1.2.4 Fix: opt.numberOfWorkers not work
• 1.2.3 Update README
• 1.2.2 Update README and examples
• 1.2.1 Handle exceptions and errors in subtasks
• 1.2.0 Simplified usage by providing the function way and support return Promise
• 1.1.4 Remove make-dir
• 1.1.3 Simplified usage, see example0
• 1.1.2
• 1.1.1 Rename files, updated changelog
• 1.1.0 Simplified the usage of a customized Consumer, see example0
• 1.0.8 Fix examples
• 1.0.7 Remove moment
• 1.0.6 Performance optimization

Github:

https://github.com/zhanglei923/multi-tasks

Support / Bug Report:

[email protected]