/**
|
* Support for concurrent task management and synchronization in web
|
* applications.
|
*
|
* @author Dave Longley
|
* @author David I. Lehn <dlehn@digitalbazaar.com>
|
*
|
* Copyright (c) 2009-2013 Digital Bazaar, Inc.
|
*/
|
var forge = require('./forge');
|
require('./debug');
|
require('./log');
|
require('./util');
|
|
// logging category
|
var cat = 'forge.task';
|
|
// verbose level
|
// 0: off, 1: a little, 2: a whole lot
|
// Verbose debug logging is surrounded by a level check to avoid the
|
// performance issues with even calling the logging code regardless if it
|
// is actually logged. For performance reasons this should not be set to 2
|
// for production use.
|
// ex: if(sVL >= 2) forge.log.verbose(....)
|
var sVL = 0;
|
|
// track tasks for debugging
|
var sTasks = {};
|
var sNextTaskId = 0;
|
// debug access
|
forge.debug.set(cat, 'tasks', sTasks);
|
|
// a map of task type to task queue
|
var sTaskQueues = {};
|
// debug access
|
forge.debug.set(cat, 'queues', sTaskQueues);
|
|
// name for unnamed tasks
|
var sNoTaskName = '?';
|
|
// maximum number of doNext() recursions before a context swap occurs
|
// FIXME: might need to tweak this based on the browser
|
var sMaxRecursions = 30;
|
|
// time slice for doing tasks before a context swap occurs
|
// FIXME: might need to tweak this based on the browser
|
var sTimeSlice = 20;
|
|
/**
|
* Task states.
|
*
|
* READY: ready to start processing
|
* RUNNING: task or a subtask is running
|
* BLOCKED: task is waiting to acquire N permits to continue
|
* SLEEPING: task is sleeping for a period of time
|
* DONE: task is done
|
* ERROR: task has an error
|
*/
|
var READY = 'ready';
|
var RUNNING = 'running';
|
var BLOCKED = 'blocked';
|
var SLEEPING = 'sleeping';
|
var DONE = 'done';
|
var ERROR = 'error';
|
|
/**
|
* Task actions. Used to control state transitions.
|
*
|
* STOP: stop processing
|
* START: start processing tasks
|
* BLOCK: block task from continuing until 1 or more permits are released
|
* UNBLOCK: release one or more permits
|
* SLEEP: sleep for a period of time
|
* WAKEUP: wakeup early from SLEEPING state
|
* CANCEL: cancel further tasks
|
* FAIL: a failure occured
|
*/
|
var STOP = 'stop';
|
var START = 'start';
|
var BLOCK = 'block';
|
var UNBLOCK = 'unblock';
|
var SLEEP = 'sleep';
|
var WAKEUP = 'wakeup';
|
var CANCEL = 'cancel';
|
var FAIL = 'fail';
|
|
/**
|
* State transition table.
|
*
|
* nextState = sStateTable[currentState][action]
|
*/
|
var sStateTable = {};
|
|
sStateTable[READY] = {};
|
sStateTable[READY][STOP] = READY;
|
sStateTable[READY][START] = RUNNING;
|
sStateTable[READY][CANCEL] = DONE;
|
sStateTable[READY][FAIL] = ERROR;
|
|
sStateTable[RUNNING] = {};
|
sStateTable[RUNNING][STOP] = READY;
|
sStateTable[RUNNING][START] = RUNNING;
|
sStateTable[RUNNING][BLOCK] = BLOCKED;
|
sStateTable[RUNNING][UNBLOCK] = RUNNING;
|
sStateTable[RUNNING][SLEEP] = SLEEPING;
|
sStateTable[RUNNING][WAKEUP] = RUNNING;
|
sStateTable[RUNNING][CANCEL] = DONE;
|
sStateTable[RUNNING][FAIL] = ERROR;
|
|
sStateTable[BLOCKED] = {};
|
sStateTable[BLOCKED][STOP] = BLOCKED;
|
sStateTable[BLOCKED][START] = BLOCKED;
|
sStateTable[BLOCKED][BLOCK] = BLOCKED;
|
sStateTable[BLOCKED][UNBLOCK] = BLOCKED;
|
sStateTable[BLOCKED][SLEEP] = BLOCKED;
|
sStateTable[BLOCKED][WAKEUP] = BLOCKED;
|
sStateTable[BLOCKED][CANCEL] = DONE;
|
sStateTable[BLOCKED][FAIL] = ERROR;
|
|
sStateTable[SLEEPING] = {};
|
sStateTable[SLEEPING][STOP] = SLEEPING;
|
sStateTable[SLEEPING][START] = SLEEPING;
|
sStateTable[SLEEPING][BLOCK] = SLEEPING;
|
sStateTable[SLEEPING][UNBLOCK] = SLEEPING;
|
sStateTable[SLEEPING][SLEEP] = SLEEPING;
|
sStateTable[SLEEPING][WAKEUP] = SLEEPING;
|
sStateTable[SLEEPING][CANCEL] = DONE;
|
sStateTable[SLEEPING][FAIL] = ERROR;
|
|
sStateTable[DONE] = {};
|
sStateTable[DONE][STOP] = DONE;
|
sStateTable[DONE][START] = DONE;
|
sStateTable[DONE][BLOCK] = DONE;
|
sStateTable[DONE][UNBLOCK] = DONE;
|
sStateTable[DONE][SLEEP] = DONE;
|
sStateTable[DONE][WAKEUP] = DONE;
|
sStateTable[DONE][CANCEL] = DONE;
|
sStateTable[DONE][FAIL] = ERROR;
|
|
sStateTable[ERROR] = {};
|
sStateTable[ERROR][STOP] = ERROR;
|
sStateTable[ERROR][START] = ERROR;
|
sStateTable[ERROR][BLOCK] = ERROR;
|
sStateTable[ERROR][UNBLOCK] = ERROR;
|
sStateTable[ERROR][SLEEP] = ERROR;
|
sStateTable[ERROR][WAKEUP] = ERROR;
|
sStateTable[ERROR][CANCEL] = ERROR;
|
sStateTable[ERROR][FAIL] = ERROR;
|
|
/**
|
* Creates a new task.
|
*
|
* @param options options for this task
|
* run: the run function for the task (required)
|
* name: the run function for the task (optional)
|
* parent: parent of this task (optional)
|
*
|
* @return the empty task.
|
*/
|
var Task = function(options) {
|
// task id
|
this.id = -1;
|
|
// task name
|
this.name = options.name || sNoTaskName;
|
|
// task has no parent
|
this.parent = options.parent || null;
|
|
// save run function
|
this.run = options.run;
|
|
// create a queue of subtasks to run
|
this.subtasks = [];
|
|
// error flag
|
this.error = false;
|
|
// state of the task
|
this.state = READY;
|
|
// number of times the task has been blocked (also the number
|
// of permits needed to be released to continue running)
|
this.blocks = 0;
|
|
// timeout id when sleeping
|
this.timeoutId = null;
|
|
// no swap time yet
|
this.swapTime = null;
|
|
// no user data
|
this.userData = null;
|
|
// initialize task
|
// FIXME: deal with overflow
|
this.id = sNextTaskId++;
|
sTasks[this.id] = this;
|
if(sVL >= 1) {
|
forge.log.verbose(cat, '[%s][%s] init', this.id, this.name, this);
|
}
|
};
|
|
/**
|
* Logs debug information on this task and the system state.
|
*/
|
Task.prototype.debug = function(msg) {
|
msg = msg || '';
|
forge.log.debug(cat, msg,
|
'[%s][%s] task:', this.id, this.name, this,
|
'subtasks:', this.subtasks.length,
|
'queue:', sTaskQueues);
|
};
|
|
/**
|
* Adds a subtask to run after task.doNext() or task.fail() is called.
|
*
|
* @param name human readable name for this task (optional).
|
* @param subrun a function to run that takes the current task as
|
* its first parameter.
|
*
|
* @return the current task (useful for chaining next() calls).
|
*/
|
Task.prototype.next = function(name, subrun) {
|
// juggle parameters if it looks like no name is given
|
if(typeof(name) === 'function') {
|
subrun = name;
|
|
// inherit parent's name
|
name = this.name;
|
}
|
// create subtask, set parent to this task, propagate callbacks
|
var subtask = new Task({
|
run: subrun,
|
name: name,
|
parent: this
|
});
|
// start subtasks running
|
subtask.state = RUNNING;
|
subtask.type = this.type;
|
subtask.successCallback = this.successCallback || null;
|
subtask.failureCallback = this.failureCallback || null;
|
|
// queue a new subtask
|
this.subtasks.push(subtask);
|
|
return this;
|
};
|
|
/**
|
* Adds subtasks to run in parallel after task.doNext() or task.fail()
|
* is called.
|
*
|
* @param name human readable name for this task (optional).
|
* @param subrun functions to run that take the current task as
|
* their first parameter.
|
*
|
* @return the current task (useful for chaining next() calls).
|
*/
|
Task.prototype.parallel = function(name, subrun) {
|
// juggle parameters if it looks like no name is given
|
if(forge.util.isArray(name)) {
|
subrun = name;
|
|
// inherit parent's name
|
name = this.name;
|
}
|
// Wrap parallel tasks in a regular task so they are started at the
|
// proper time.
|
return this.next(name, function(task) {
|
// block waiting for subtasks
|
var ptask = task;
|
ptask.block(subrun.length);
|
|
// we pass the iterator from the loop below as a parameter
|
// to a function because it is otherwise included in the
|
// closure and changes as the loop changes -- causing i
|
// to always be set to its highest value
|
var startParallelTask = function(pname, pi) {
|
forge.task.start({
|
type: pname,
|
run: function(task) {
|
subrun[pi](task);
|
},
|
success: function(task) {
|
ptask.unblock();
|
},
|
failure: function(task) {
|
ptask.unblock();
|
}
|
});
|
};
|
|
for(var i = 0; i < subrun.length; i++) {
|
// Type must be unique so task starts in parallel:
|
// name + private string + task id + sub-task index
|
// start tasks in parallel and unblock when the finish
|
var pname = name + '__parallel-' + task.id + '-' + i;
|
var pi = i;
|
startParallelTask(pname, pi);
|
}
|
});
|
};
|
|
/**
|
* Stops a running task.
|
*/
|
Task.prototype.stop = function() {
|
this.state = sStateTable[this.state][STOP];
|
};
|
|
/**
|
* Starts running a task.
|
*/
|
Task.prototype.start = function() {
|
this.error = false;
|
this.state = sStateTable[this.state][START];
|
|
// try to restart
|
if(this.state === RUNNING) {
|
this.start = new Date();
|
this.run(this);
|
runNext(this, 0);
|
}
|
};
|
|
/**
|
* Blocks a task until it one or more permits have been released. The
|
* task will not resume until the requested number of permits have
|
* been released with call(s) to unblock().
|
*
|
* @param n number of permits to wait for(default: 1).
|
*/
|
Task.prototype.block = function(n) {
|
n = typeof(n) === 'undefined' ? 1 : n;
|
this.blocks += n;
|
if(this.blocks > 0) {
|
this.state = sStateTable[this.state][BLOCK];
|
}
|
};
|
|
/**
|
* Releases a permit to unblock a task. If a task was blocked by
|
* requesting N permits via block(), then it will only continue
|
* running once enough permits have been released via unblock() calls.
|
*
|
* If multiple processes need to synchronize with a single task then
|
* use a condition variable (see forge.task.createCondition). It is
|
* an error to unblock a task more times than it has been blocked.
|
*
|
* @param n number of permits to release (default: 1).
|
*
|
* @return the current block count (task is unblocked when count is 0)
|
*/
|
Task.prototype.unblock = function(n) {
|
n = typeof(n) === 'undefined' ? 1 : n;
|
this.blocks -= n;
|
if(this.blocks === 0 && this.state !== DONE) {
|
this.state = RUNNING;
|
runNext(this, 0);
|
}
|
return this.blocks;
|
};
|
|
/**
|
* Sleep for a period of time before resuming tasks.
|
*
|
* @param n number of milliseconds to sleep (default: 0).
|
*/
|
Task.prototype.sleep = function(n) {
|
n = typeof(n) === 'undefined' ? 0 : n;
|
this.state = sStateTable[this.state][SLEEP];
|
var self = this;
|
this.timeoutId = setTimeout(function() {
|
self.timeoutId = null;
|
self.state = RUNNING;
|
runNext(self, 0);
|
}, n);
|
};
|
|
/**
|
* Waits on a condition variable until notified. The next task will
|
* not be scheduled until notification. A condition variable can be
|
* created with forge.task.createCondition().
|
*
|
* Once cond.notify() is called, the task will continue.
|
*
|
* @param cond the condition variable to wait on.
|
*/
|
Task.prototype.wait = function(cond) {
|
cond.wait(this);
|
};
|
|
/**
|
* If sleeping, wakeup and continue running tasks.
|
*/
|
Task.prototype.wakeup = function() {
|
if(this.state === SLEEPING) {
|
cancelTimeout(this.timeoutId);
|
this.timeoutId = null;
|
this.state = RUNNING;
|
runNext(this, 0);
|
}
|
};
|
|
/**
|
* Cancel all remaining subtasks of this task.
|
*/
|
Task.prototype.cancel = function() {
|
this.state = sStateTable[this.state][CANCEL];
|
// remove permits needed
|
this.permitsNeeded = 0;
|
// cancel timeouts
|
if(this.timeoutId !== null) {
|
cancelTimeout(this.timeoutId);
|
this.timeoutId = null;
|
}
|
// remove subtasks
|
this.subtasks = [];
|
};
|
|
/**
|
* Finishes this task with failure and sets error flag. The entire
|
* task will be aborted unless the next task that should execute
|
* is passed as a parameter. This allows levels of subtasks to be
|
* skipped. For instance, to abort only this tasks's subtasks, then
|
* call fail(task.parent). To abort this task's subtasks and its
|
* parent's subtasks, call fail(task.parent.parent). To abort
|
* all tasks and simply call the task callback, call fail() or
|
* fail(null).
|
*
|
* The task callback (success or failure) will always, eventually, be
|
* called.
|
*
|
* @param next the task to continue at, or null to abort entirely.
|
*/
|
Task.prototype.fail = function(next) {
|
// set error flag
|
this.error = true;
|
|
// finish task
|
finish(this, true);
|
|
if(next) {
|
// propagate task info
|
next.error = this.error;
|
next.swapTime = this.swapTime;
|
next.userData = this.userData;
|
|
// do next task as specified
|
runNext(next, 0);
|
} else {
|
if(this.parent !== null) {
|
// finish root task (ensures it is removed from task queue)
|
var parent = this.parent;
|
while(parent.parent !== null) {
|
// propagate task info
|
parent.error = this.error;
|
parent.swapTime = this.swapTime;
|
parent.userData = this.userData;
|
parent = parent.parent;
|
}
|
finish(parent, true);
|
}
|
|
// call failure callback if one exists
|
if(this.failureCallback) {
|
this.failureCallback(this);
|
}
|
}
|
};
|
|
/**
|
* Asynchronously start a task.
|
*
|
* @param task the task to start.
|
*/
|
var start = function(task) {
|
task.error = false;
|
task.state = sStateTable[task.state][START];
|
setTimeout(function() {
|
if(task.state === RUNNING) {
|
task.swapTime = +new Date();
|
task.run(task);
|
runNext(task, 0);
|
}
|
}, 0);
|
};
|
|
/**
|
* Run the next subtask or finish this task.
|
*
|
* @param task the task to process.
|
* @param recurse the recursion count.
|
*/
|
var runNext = function(task, recurse) {
|
// get time since last context swap (ms), if enough time has passed set
|
// swap to true to indicate that doNext was performed asynchronously
|
// also, if recurse is too high do asynchronously
|
var swap =
|
(recurse > sMaxRecursions) ||
|
(+new Date() - task.swapTime) > sTimeSlice;
|
|
var doNext = function(recurse) {
|
recurse++;
|
if(task.state === RUNNING) {
|
if(swap) {
|
// update swap time
|
task.swapTime = +new Date();
|
}
|
|
if(task.subtasks.length > 0) {
|
// run next subtask
|
var subtask = task.subtasks.shift();
|
subtask.error = task.error;
|
subtask.swapTime = task.swapTime;
|
subtask.userData = task.userData;
|
subtask.run(subtask);
|
if(!subtask.error) {
|
runNext(subtask, recurse);
|
}
|
} else {
|
finish(task);
|
|
if(!task.error) {
|
// chain back up and run parent
|
if(task.parent !== null) {
|
// propagate task info
|
task.parent.error = task.error;
|
task.parent.swapTime = task.swapTime;
|
task.parent.userData = task.userData;
|
|
// no subtasks left, call run next subtask on parent
|
runNext(task.parent, recurse);
|
}
|
}
|
}
|
}
|
};
|
|
if(swap) {
|
// we're swapping, so run asynchronously
|
setTimeout(doNext, 0);
|
} else {
|
// not swapping, so run synchronously
|
doNext(recurse);
|
}
|
};
|
|
/**
|
* Finishes a task and looks for the next task in the queue to start.
|
*
|
* @param task the task to finish.
|
* @param suppressCallbacks true to suppress callbacks.
|
*/
|
var finish = function(task, suppressCallbacks) {
|
// subtask is now done
|
task.state = DONE;
|
|
delete sTasks[task.id];
|
if(sVL >= 1) {
|
forge.log.verbose(cat, '[%s][%s] finish',
|
task.id, task.name, task);
|
}
|
|
// only do queue processing for root tasks
|
if(task.parent === null) {
|
// report error if queue is missing
|
if(!(task.type in sTaskQueues)) {
|
forge.log.error(cat,
|
'[%s][%s] task queue missing [%s]',
|
task.id, task.name, task.type);
|
} else if(sTaskQueues[task.type].length === 0) {
|
// report error if queue is empty
|
forge.log.error(cat,
|
'[%s][%s] task queue empty [%s]',
|
task.id, task.name, task.type);
|
} else if(sTaskQueues[task.type][0] !== task) {
|
// report error if this task isn't the first in the queue
|
forge.log.error(cat,
|
'[%s][%s] task not first in queue [%s]',
|
task.id, task.name, task.type);
|
} else {
|
// remove ourselves from the queue
|
sTaskQueues[task.type].shift();
|
// clean up queue if it is empty
|
if(sTaskQueues[task.type].length === 0) {
|
if(sVL >= 1) {
|
forge.log.verbose(cat, '[%s][%s] delete queue [%s]',
|
task.id, task.name, task.type);
|
}
|
/* Note: Only a task can delete a queue of its own type. This
|
is used as a way to synchronize tasks. If a queue for a certain
|
task type exists, then a task of that type is running.
|
*/
|
delete sTaskQueues[task.type];
|
} else {
|
// dequeue the next task and start it
|
if(sVL >= 1) {
|
forge.log.verbose(cat,
|
'[%s][%s] queue start next [%s] remain:%s',
|
task.id, task.name, task.type,
|
sTaskQueues[task.type].length);
|
}
|
sTaskQueues[task.type][0].start();
|
}
|
}
|
|
if(!suppressCallbacks) {
|
// call final callback if one exists
|
if(task.error && task.failureCallback) {
|
task.failureCallback(task);
|
} else if(!task.error && task.successCallback) {
|
task.successCallback(task);
|
}
|
}
|
}
|
};
|
|
/* Tasks API */
|
module.exports = forge.task = forge.task || {};
|
|
/**
|
* Starts a new task that will run the passed function asynchronously.
|
*
|
* In order to finish the task, either task.doNext() or task.fail()
|
* *must* be called.
|
*
|
* The task must have a type (a string identifier) that can be used to
|
* synchronize it with other tasks of the same type. That type can also
|
* be used to cancel tasks that haven't started yet.
|
*
|
* To start a task, the following object must be provided as a parameter
|
* (each function takes a task object as its first parameter):
|
*
|
* {
|
* type: the type of task.
|
* run: the function to run to execute the task.
|
* success: a callback to call when the task succeeds (optional).
|
* failure: a callback to call when the task fails (optional).
|
* }
|
*
|
* @param options the object as described above.
|
*/
|
forge.task.start = function(options) {
|
// create a new task
|
var task = new Task({
|
run: options.run,
|
name: options.name || sNoTaskName
|
});
|
task.type = options.type;
|
task.successCallback = options.success || null;
|
task.failureCallback = options.failure || null;
|
|
// append the task onto the appropriate queue
|
if(!(task.type in sTaskQueues)) {
|
if(sVL >= 1) {
|
forge.log.verbose(cat, '[%s][%s] create queue [%s]',
|
task.id, task.name, task.type);
|
}
|
// create the queue with the new task
|
sTaskQueues[task.type] = [task];
|
start(task);
|
} else {
|
// push the task onto the queue, it will be run after a task
|
// with the same type completes
|
sTaskQueues[options.type].push(task);
|
}
|
};
|
|
/**
|
* Cancels all tasks of the given type that haven't started yet.
|
*
|
* @param type the type of task to cancel.
|
*/
|
forge.task.cancel = function(type) {
|
// find the task queue
|
if(type in sTaskQueues) {
|
// empty all but the current task from the queue
|
sTaskQueues[type] = [sTaskQueues[type][0]];
|
}
|
};
|
|
/**
|
* Creates a condition variable to synchronize tasks. To make a task wait
|
* on the condition variable, call task.wait(condition). To notify all
|
* tasks that are waiting, call condition.notify().
|
*
|
* @return the condition variable.
|
*/
|
forge.task.createCondition = function() {
|
var cond = {
|
// all tasks that are blocked
|
tasks: {}
|
};
|
|
/**
|
* Causes the given task to block until notify is called. If the task
|
* is already waiting on this condition then this is a no-op.
|
*
|
* @param task the task to cause to wait.
|
*/
|
cond.wait = function(task) {
|
// only block once
|
if(!(task.id in cond.tasks)) {
|
task.block();
|
cond.tasks[task.id] = task;
|
}
|
};
|
|
/**
|
* Notifies all waiting tasks to wake up.
|
*/
|
cond.notify = function() {
|
// since unblock() will run the next task from here, make sure to
|
// clear the condition's blocked task list before unblocking
|
var tmp = cond.tasks;
|
cond.tasks = {};
|
for(var id in tmp) {
|
tmp[id].unblock();
|
}
|
};
|
|
return cond;
|
};
|
