/**
* 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;
};