IPython/kernel/task.py

# encoding: utf-8
# -*- test-case-name: IPython.kernel.tests.test_task -*-

"""Task farming representation of the ControllerService."""

__docformat__ = "restructuredtext en"

#-----------------------------------------------------------------------------
#  Copyright (C) 2008  The IPython Development Team
#
#  Distributed under the terms of the BSD License.  The full license is in
#  the file COPYING, distributed as part of this software.
#-----------------------------------------------------------------------------

#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------

# Tell nose to skip the testing of this module
__test__ = {}

import copy, time
from types import FunctionType

import zope.interface as zi, string
from twisted.internet import defer, reactor
from twisted.python import components, log, failure

from IPython.kernel.util import printer
from IPython.kernel import engineservice as es, error
from IPython.kernel import controllerservice as cs
from IPython.kernel.twistedutil import gatherBoth, DeferredList

from IPython.kernel.pickleutil import can, uncan, CannedFunction

#-----------------------------------------------------------------------------
# Definition of the Task objects
#-----------------------------------------------------------------------------

time_format = '%Y/%m/%d %H:%M:%S'

class ITask(zi.Interface):
    """
    This interface provides a generic definition of what constitutes a task.
    
    There are two sides to a task.  First a task needs to take input from
    a user to determine what work is performed by the task.  Second, the
    task needs to have the logic that knows how to turn that information
    info specific calls to a worker, through the `IQueuedEngine` interface.
    
    Many method in this class get two things passed to them: a Deferred
    and an IQueuedEngine implementer.  Such methods should register callbacks
    on the Deferred that use the IQueuedEngine to accomplish something.  See
    the existing task objects for examples.
    """
    
    zi.Attribute('retries','How many times to retry the task')
    zi.Attribute('recovery_task','A task to try if the initial one fails')
    zi.Attribute('taskid','the id of the task')
    
    def start_time(result):
        """
        Do anything needed to start the timing of the task.
        
        Must simply return the result after starting the timers.
        """
    
    def stop_time(result):
        """
        Do anything needed to stop the timing of the task.
        
        Must simply return the result after stopping the timers.  This 
        method will usually set attributes that are used by `process_result`
        in building result of the task.
        """
    
    def pre_task(d, queued_engine):
        """Do something with the queued_engine before the task is run.
        
        This method should simply add callbacks to the input Deferred
        that do something with the `queued_engine` before the task is run.
        
        :Parameters:
            d : Deferred
                The deferred that actions should be attached to
            queued_engine : IQueuedEngine implementer
                The worker that has been allocated to perform the task
        """
    
    def post_task(d, queued_engine):
        """Do something with the queued_engine after the task is run.
        
        This method should simply add callbacks to the input Deferred
        that do something with the `queued_engine` before the task is run.
        
        :Parameters:
            d : Deferred
                The deferred that actions should be attached to
            queued_engine : IQueuedEngine implementer
                The worker that has been allocated to perform the task
        """
    
    def submit_task(d, queued_engine):
        """Submit a task using the `queued_engine` we have been allocated.
        
        When a task is ready to run, this method is called.  This method
        must take the internal information of the task and make suitable
        calls on the queued_engine to have the actual work done.
        
        This method should simply add callbacks to the input Deferred
        that do something with the `queued_engine` before the task is run.
        
        :Parameters:
            d : Deferred
                The deferred that actions should be attached to
            queued_engine : IQueuedEngine implementer
                The worker that has been allocated to perform the task
        """

    def process_result(d, result, engine_id):
        """Take a raw task result.
        
        Objects that implement `ITask` can choose how the result of running
        the task is presented.  This method takes the raw result and 
        does this logic.  Two example are the `MapTask` which simply returns
        the raw result or a `Failure` object and the `StringTask` which
        returns a `TaskResult` object.
        
        :Parameters:
            d : Deferred
                The deferred that actions should be attached to
            result : object
                The raw task result that needs to be wrapped
            engine_id : int
                The id of the engine that did the task
        
        :Returns:
            The result, as a tuple of the form: (success, result).
            Here, success is a boolean indicating if the task
            succeeded or failed and result is the result.
        """
        
    def check_depend(properties):
        """Check properties to see if the task should be run.
        
        :Parameters:
            properties : dict
                A dictionary of properties that an engine has set
        
        :Returns:
            True if the task should be run, False otherwise
        """
    
    def can_task(self):
        """Serialize (can) any functions in the task for pickling.
        
        Subclasses must override this method and make sure that all 
        functions in the task are canned by calling `can` on the 
        function.
        """
    
    def uncan_task(self):
        """Unserialize (uncan) any canned function in the task."""

class BaseTask(object):
    """
    Common fuctionality for all objects implementing `ITask`.
    """
    
    zi.implements(ITask)
    
    def __init__(self, clear_before=False, clear_after=False, retries=0,
            recovery_task=None, depend=None):
        """
        Make a generic task.
        
        :Parameters:
            clear_before : boolean
                Should the engines namespace be cleared before the task
                is run
            clear_after : boolean
                Should the engines namespace be clear after the task is run
            retries : int
                The number of times a task should be retries upon failure
            recovery_task : any task object
                If a task fails and it has a recovery_task, that is run
                upon a retry
            depend : FunctionType
                A function that is called to test for properties.  This function
                must take one argument, the properties dict and return a boolean
        """
        self.clear_before = clear_before
        self.clear_after = clear_after
        self.retries = retries
        self.recovery_task = recovery_task
        self.depend = depend
        self.taskid = None
    
    def start_time(self, result):
        """
        Start the basic timers.
        """
        self.start = time.time()
        self.start_struct = time.localtime()
        return result
    
    def stop_time(self, result):
        """
        Stop the basic timers.
        """
        self.stop = time.time()
        self.stop_struct = time.localtime()
        self.duration = self.stop - self.start 
        self.submitted = time.strftime(time_format, self.start_struct)
        self.completed = time.strftime(time_format)
        return result
    
    def pre_task(self, d, queued_engine):
        """
        Clear the engine before running the task if clear_before is set.
        """
        if self.clear_before:
            d.addCallback(lambda r: queued_engine.reset())
    
    def post_task(self, d, queued_engine):
        """
        Clear the engine after running the task if clear_after is set.
        """
        def reseter(result):
            queued_engine.reset()
            return result
        if self.clear_after:
            d.addBoth(reseter)
    
    def submit_task(self, d, queued_engine):
        raise NotImplementedError('submit_task must be implemented in a subclass')
    
    def process_result(self, result, engine_id):
        """
        Process a task result.
        
        This is the default `process_result` that just returns the raw
        result or a `Failure`.
        """
        if isinstance(result, failure.Failure):
            return (False, result)
        else:
            return (True, result)
    
    def check_depend(self, properties):
        """
        Calls self.depend(properties) to see if a task should be run.
        """
        if self.depend is not None:
            return self.depend(properties)
        else:
            return True

    def can_task(self):
        self.depend = can(self.depend)
        if isinstance(self.recovery_task, BaseTask):
            self.recovery_task.can_task()
            
    def uncan_task(self):
        self.depend = uncan(self.depend)
        if isinstance(self.recovery_task, BaseTask):
            self.recovery_task.uncan_task()

class MapTask(BaseTask):
    """
    A task that consists of a function and arguments.
    """
    
    zi.implements(ITask)
    
    def __init__(self, function, args=None, kwargs=None, clear_before=False, 
            clear_after=False, retries=0, recovery_task=None, depend=None):
        """
        Create a task based on a function, args and kwargs.
        
        This is a simple type of task that consists of calling:
        function(*args, **kwargs) and wrapping the result in a `TaskResult`.
        
        The return value of the function, or a `Failure` wrapping an 
        exception is the task result for this type of task.
        """
        BaseTask.__init__(self, clear_before, clear_after, retries, 
            recovery_task, depend)
        if not isinstance(function, FunctionType):
            raise TypeError('a task function must be a FunctionType')
        self.function = function
        if args is None:
            self.args = ()
        else:
            self.args = args
        if not isinstance(self.args, (list, tuple)):
            raise TypeError('a task args must be a list or tuple')
        if kwargs is None:
            self.kwargs = {}
        else:
            self.kwargs = kwargs
        if not isinstance(self.kwargs, dict):
            raise TypeError('a task kwargs must be a dict')
    
    def submit_task(self, d, queued_engine):
        d.addCallback(lambda r: queued_engine.push_function(
            dict(_ipython_task_function=self.function))
        )
        d.addCallback(lambda r: queued_engine.push(
            dict(_ipython_task_args=self.args,_ipython_task_kwargs=self.kwargs))
        )
        d.addCallback(lambda r: queued_engine.execute(
            '_ipython_task_result = _ipython_task_function(*_ipython_task_args,**_ipython_task_kwargs)')
        )
        d.addCallback(lambda r: queued_engine.pull('_ipython_task_result'))
    
    def can_task(self):
        self.function = can(self.function)
        BaseTask.can_task(self)
    
    def uncan_task(self):
        self.function = uncan(self.function)
        BaseTask.uncan_task(self)


class StringTask(BaseTask):
    """
    A task that consists of a string of Python code to run.
    """

    def __init__(self, expression, pull=None, push=None,
            clear_before=False, clear_after=False, retries=0, 
            recovery_task=None, depend=None):
        """
        Create a task based on a Python expression and variables
        
        This type of task lets you push a set of variables to the engines
        namespace, run a Python string in that namespace and then bring back
        a different set of Python variables as the result.
        
        Because this type of task can return many results (through the
        `pull` keyword argument) it returns a special `TaskResult` object
        that wraps the pulled variables, statistics about the run and
        any exceptions raised.
        """
        if not isinstance(expression, str):
            raise TypeError('a task expression must be a string')
        self.expression = expression
        
        if pull==None:
            self.pull = ()
        elif isinstance(pull, str):
            self.pull = (pull,)
        elif isinstance(pull, (list, tuple)):
            self.pull = pull
        else:
            raise TypeError('pull must be str or a sequence of strs')
        
        if push==None:
            self.push = {}
        elif isinstance(push, dict):
            self.push = push
        else:
            raise TypeError('push must be a dict')
        
        BaseTask.__init__(self, clear_before, clear_after, retries, 
            recovery_task, depend)

    def submit_task(self, d, queued_engine):
        if self.push is not None:
            d.addCallback(lambda r: queued_engine.push(self.push))
        
        d.addCallback(lambda r: queued_engine.execute(self.expression))
        
        if self.pull is not None:
            d.addCallback(lambda r: queued_engine.pull(self.pull))
        else:
            d.addCallback(lambda r: None)    
    
    def process_result(self, result, engine_id):
        if isinstance(result, failure.Failure):
            tr = TaskResult(result, engine_id)
        else:
            if self.pull is None:
                resultDict = {}
            elif len(self.pull) == 1:
                resultDict = {self.pull[0]:result}
            else:
                resultDict = dict(zip(self.pull, result))
            tr = TaskResult(resultDict, engine_id)
        # Assign task attributes
        tr.submitted = self.submitted
        tr.completed = self.completed
        tr.duration = self.duration
        if hasattr(self,'taskid'):
            tr.taskid = self.taskid
        else:
            tr.taskid = None
        if isinstance(result, failure.Failure):
            return (False, tr)
        else:
            return (True, tr)

class ResultNS(object):
    """
    A dict like object for holding the results of a task.
    
    The result namespace object for use in `TaskResult` objects as tr.ns.
    It builds an object from a dictionary, such that it has attributes
    according to the key,value pairs of the dictionary.
    
    This works by calling setattr on ALL key,value pairs in the dict.  If a user
    chooses to overwrite the `__repr__` or `__getattr__` attributes, they can.
    This can be a bad idea, as it may corrupt standard behavior of the
    ns object.
    
    Example
    --------
    
    >>> ns = ResultNS({'a':17,'foo':range(3)})
    >>> print ns
    NS{'a': 17, 'foo': [0, 1, 2]}
    >>> ns.a
    17
    >>> ns['foo']
    [0, 1, 2]
    """
    def __init__(self, dikt):
        for k,v in dikt.iteritems():
            setattr(self,k,v)
    
    def __repr__(self):
        l = dir(self)
        d = {}
        for k in l:
            # do not print private objects
            if k[:2] != '__' and k[-2:] != '__':
                d[k] = getattr(self, k)
        return "NS"+repr(d)
    
    def __getitem__(self, key):
        return getattr(self, key)

class TaskResult(object):
    """
    An object for returning task results for certain types of tasks.
        
    This object encapsulates the results of a task.  On task
    success it will have a keys attribute that will have a list
    of the variables that have been pulled back.  These variables
    are accessible as attributes of this class as well.  On 
    success the failure attribute will be None.
        
    In task failure, keys will be empty, but failure will contain
    the failure object that encapsulates the remote exception.
    One can also simply call the `raise_exception` method of 
    this class to re-raise any remote exception in the local
    session.
        
    The `TaskResult` has a `.ns` member, which is a property for access
    to the results.  If the Task had pull=['a', 'b'], then the 
    Task Result will have attributes `tr.ns.a`, `tr.ns.b` for those values.
    Accessing `tr.ns` will raise the remote failure if the task failed.
        
    The `engineid` attribute should have the `engineid` of the engine
    that ran the task.  But, because engines can come and go, 
    the `engineid` may not continue to be
    valid or accurate.
        
    The `taskid` attribute simply gives the `taskid` that the task
    is tracked under.
    """
    taskid = None
    
    def _getNS(self):
        if isinstance(self.failure, failure.Failure):
            return self.failure.raiseException()
        else:
            return self._ns
    
    def _setNS(self, v):
        raise Exception("the ns attribute cannot be changed")
    
    ns = property(_getNS, _setNS)
    
    def __init__(self, results, engineid):
        self.engineid = engineid
        if isinstance(results, failure.Failure):
            self.failure = results
            self.results = {}
        else:
            self.results = results
            self.failure = None
        
        self._ns = ResultNS(self.results)
        
        self.keys = self.results.keys()
    
    def __repr__(self):
        if self.failure is not None:
            contents = self.failure
        else:
            contents = self.results
        return "TaskResult[ID:%r]:%r"%(self.taskid, contents)
    
    def __getitem__(self, key):
        if self.failure is not None:
            self.raise_exception()
        return self.results[key]
    
    def raise_exception(self):
        """Re-raise any remote exceptions in the local python session."""
        if self.failure is not None:
            self.failure.raiseException()


#-----------------------------------------------------------------------------
# The controller side of things
#-----------------------------------------------------------------------------

class IWorker(zi.Interface):
    """The Basic Worker Interface. 
    
    A worked is a representation of an Engine that is ready to run tasks.
    """
    
    zi.Attribute("workerid", "the id of the worker")
    
    def run(task):
        """Run task in worker's namespace.
        
        :Parameters:
            task : a `Task` object
        
        :Returns: `Deferred` to a tuple of (success, result) where
            success if a boolean that signifies success or failure
            and result is the task result.
        """


class WorkerFromQueuedEngine(object):
    """Adapt an `IQueuedEngine` to an `IWorker` object"""
    
    zi.implements(IWorker)
    
    def __init__(self, qe):
        self.queuedEngine = qe
        self.workerid = None
    
    def _get_properties(self):
        return self.queuedEngine.properties
    
    properties = property(_get_properties, lambda self, _:None)
    
    def run(self, task):
        """Run task in worker's namespace.
        
        This takes a task and calls methods on the task that actually
        cause `self.queuedEngine` to do the task.  See the methods of
        `ITask` for more information about how these methods are called.
        
        :Parameters:
            task : a `Task` object
        
        :Returns: `Deferred` to a tuple of (success, result) where
            success if a boolean that signifies success or failure
            and result is the task result.
        """
        d = defer.succeed(None)
        d.addCallback(task.start_time)
        task.pre_task(d, self.queuedEngine)
        task.submit_task(d, self.queuedEngine)
        task.post_task(d, self.queuedEngine)
        d.addBoth(task.stop_time)
        d.addBoth(task.process_result, self.queuedEngine.id)
        # At this point, there will be (success, result) coming down the line
        return d


components.registerAdapter(WorkerFromQueuedEngine, es.IEngineQueued, IWorker)

class IScheduler(zi.Interface):
    """The interface for a Scheduler.
    """
    zi.Attribute("nworkers", "the number of unassigned workers")
    zi.Attribute("ntasks", "the number of unscheduled tasks")
    zi.Attribute("workerids", "a list of the worker ids")
    zi.Attribute("taskids", "a list of the task ids")
    
    def add_task(task, **flags):
        """Add a task to the queue of the Scheduler.
        
        :Parameters:
            task : an `ITask` implementer
                The task to be queued.
            flags : dict
                General keywords for more sophisticated scheduling
        """
    
    def pop_task(id=None):
        """Pops a task object from the queue.
        
        This gets the next task to be run.  If no `id` is requested, the highest priority
        task is returned.
        
        :Parameters:
            id
                The id of the task to be popped.  The default (None) is to return 
                the highest priority task.
                
        :Returns: an `ITask` implementer
        
        :Exceptions:
            IndexError : raised if no taskid in queue
        """
    
    def add_worker(worker, **flags):
        """Add a worker to the worker queue.
        
        :Parameters:
            worker : an `IWorker` implementer
            flags : dict
                General keywords for more sophisticated scheduling
        """
    
    def pop_worker(id=None):
        """Pops an IWorker object that is ready to do work.
        
        This gets the next IWorker that is ready to do work. 
        
        :Parameters:
            id : if specified, will pop worker with workerid=id, else pops
                 highest priority worker.  Defaults to None.
        
        :Returns:
            an IWorker object
        
        :Exceptions:
            IndexError : raised if no workerid in queue
        """
    
    def ready():
        """Returns True if there is something to do, False otherwise"""
    
    def schedule():
        """Returns (worker,task) pair for the next task to be run."""
    

class FIFOScheduler(object):
    """
    A basic First-In-First-Out (Queue) Scheduler.
    
    This is the default Scheduler for the `TaskController`.
    See the docstrings for `IScheduler` for interface details.
    """
    
    zi.implements(IScheduler)
    
    def __init__(self):
        self.tasks = []
        self.workers = []
    
    def _ntasks(self):
        return len(self.tasks)
    
    def _nworkers(self):
        return len(self.workers)
    
    ntasks = property(_ntasks, lambda self, _:None)
    nworkers = property(_nworkers, lambda self, _:None)
    
    def _taskids(self):
        return [t.taskid for t in self.tasks]
    
    def _workerids(self):
        return [w.workerid for w in self.workers]
    
    taskids = property(_taskids, lambda self,_:None)
    workerids = property(_workerids, lambda self,_:None)
    
    def add_task(self, task, **flags):    
        self.tasks.append(task)
    
    def pop_task(self, id=None):
        if id is None:
            return self.tasks.pop(0)
        else:
            for i in range(len(self.tasks)):
                taskid = self.tasks[i].taskid
                if id == taskid:
                    return self.tasks.pop(i)
            raise IndexError("No task #%i"%id)
    
    def add_worker(self, worker, **flags):
        self.workers.append(worker)
    
    def pop_worker(self, id=None):
        if id is None:
            return self.workers.pop(0)
        else:
            for i in range(len(self.workers)):
                workerid = self.workers[i].workerid
                if id == workerid:
                    return self.workers.pop(i)
            raise IndexError("No worker #%i"%id)
    
    def schedule(self):
        for t in self.tasks:
            for w in self.workers:
                try:# do not allow exceptions to break this
                    # Allow the task to check itself using its
                    # check_depend method.
                    cando = t.check_depend(w.properties)
                except:
                    cando = False
                if cando:
                    return self.pop_worker(w.workerid), self.pop_task(t.taskid)
        return None, None
    


class LIFOScheduler(FIFOScheduler):
    """
    A Last-In-First-Out (Stack) Scheduler.
    
    This scheduler should naively reward fast engines by giving 
    them more jobs.  This risks starvation, but only in cases with 
    low load, where starvation does not really matter.
    """
    
    def add_task(self, task, **flags):
        # self.tasks.reverse()
        self.tasks.insert(0, task)
        # self.tasks.reverse()
    
    def add_worker(self, worker, **flags):
        # self.workers.reverse()
        self.workers.insert(0, worker)
        # self.workers.reverse()
    

class ITaskController(cs.IControllerBase):
    """
    The Task based interface to a `ControllerService` object
    
    This adapts a `ControllerService` to the ITaskController interface.
    """
    
    def run(task):
        """
        Run a task.
        
        :Parameters:
            task : an IPython `Task` object
        
        :Returns: the integer ID of the task
        """
    
    def get_task_result(taskid, block=False):
        """
        Get the result of a task by its ID.
        
        :Parameters:
            taskid : int
                the id of the task whose result is requested
        
        :Returns: `Deferred` to the task result if the task is done, and None
            if not.
        
        :Exceptions:
            actualResult will be an `IndexError` if no such task has been submitted
        """
    
    def abort(taskid):
        """Remove task from queue if task is has not been submitted.
        
        If the task has already been submitted, wait for it to finish and discard 
        results and prevent resubmission.
        
        :Parameters:
            taskid : the id of the task to be aborted
        
        :Returns:
            `Deferred` to abort attempt completion.  Will be None on success.
        
        :Exceptions:
            deferred will fail with `IndexError` if no such task has been submitted
            or the task has already completed.
        """
    
    def barrier(taskids):
        """
        Block until the list of taskids are completed.
        
        Returns None on success.
        """
    
    def spin():
        """
        Touch the scheduler, to resume scheduling without submitting a task.
        """
    
    def queue_status(verbose=False):
        """
        Get a dictionary with the current state of the task queue.
        
        If verbose is True, then return lists of taskids, otherwise, 
        return the number of tasks with each status.
        """
    
    def clear():
        """
        Clear all previously run tasks from the task controller.
        
        This is needed because the task controller keep all task results
        in memory.  This can be a problem is there are many completed
        tasks.  Users should call this periodically to clean out these
        cached task results.
        """
    

class TaskController(cs.ControllerAdapterBase):
    """The Task based interface to a Controller object.
    
    If you want to use a different scheduler, just subclass this and set
    the `SchedulerClass` member to the *class* of your chosen scheduler.
    """
    
    zi.implements(ITaskController)
    SchedulerClass = FIFOScheduler
    
    timeout = 30
    
    def __init__(self, controller):
        self.controller = controller
        self.controller.on_register_engine_do(self.registerWorker, True)
        self.controller.on_unregister_engine_do(self.unregisterWorker, True)
        self.taskid = 0
        self.failurePenalty = 1 # the time in seconds to penalize
                                # a worker for failing a task
        self.pendingTasks = {} # dict of {workerid:(taskid, task)}
        self.deferredResults = {} # dict of {taskid:deferred}
        self.finishedResults = {} # dict of {taskid:actualResult}
        self.workers = {} # dict of {workerid:worker}
        self.abortPending = [] # dict of {taskid:abortDeferred}
        self.idleLater = None # delayed call object for timeout
        self.scheduler = self.SchedulerClass()
        
        for id in self.controller.engines.keys():
                self.workers[id] = IWorker(self.controller.engines[id])
                self.workers[id].workerid = id
                self.schedule.add_worker(self.workers[id])
    
    def registerWorker(self, id):
        """Called by controller.register_engine."""
        if self.workers.get(id):
            raise ValueError("worker with id %s already exists.  This should not happen." % id)
        self.workers[id] = IWorker(self.controller.engines[id])
        self.workers[id].workerid = id
        if not self.pendingTasks.has_key(id):# if not working
            self.scheduler.add_worker(self.workers[id])
        self.distributeTasks()
    
    def unregisterWorker(self, id):
        """Called by controller.unregister_engine"""
        
        if self.workers.has_key(id):
            try:
                self.scheduler.pop_worker(id)
            except IndexError:
                pass
            self.workers.pop(id)
    
    def _pendingTaskIDs(self):
        return [t.taskid for t in self.pendingTasks.values()]
    
    #---------------------------------------------------------------------------
    # Interface methods
    #---------------------------------------------------------------------------
    
    def run(self, task):
        """
        Run a task and return `Deferred` to its taskid.
        """
        task.taskid = self.taskid
        task.start = time.localtime()
        self.taskid += 1
        d = defer.Deferred()
        self.scheduler.add_task(task)
        log.msg('Queuing task: %i' % task.taskid)
        
        self.deferredResults[task.taskid] = []
        self.distributeTasks()
        return defer.succeed(task.taskid)
    
    def get_task_result(self, taskid, block=False):
        """
        Returns a `Deferred` to the task result, or None.
        """
        log.msg("Getting task result: %i" % taskid)
        if self.finishedResults.has_key(taskid):
            tr = self.finishedResults[taskid]
            return defer.succeed(tr)
        elif self.deferredResults.has_key(taskid):
            if block:
                d = defer.Deferred()
                self.deferredResults[taskid].append(d)
                return d
            else:
                return defer.succeed(None)
        else:
            return defer.fail(IndexError("task ID not registered: %r" % taskid))
    
    def abort(self, taskid):
        """
        Remove a task from the queue if it has not been run already.
        """
        if not isinstance(taskid, int):
            return defer.fail(failure.Failure(TypeError("an integer task id expected: %r" % taskid)))
        try:
            self.scheduler.pop_task(taskid)
        except IndexError, e:
            if taskid in self.finishedResults.keys():
                d = defer.fail(IndexError("Task Already Completed"))
            elif taskid in self.abortPending:
                d = defer.fail(IndexError("Task Already Aborted"))
            elif taskid in self._pendingTaskIDs():# task is pending
                self.abortPending.append(taskid)
                d = defer.succeed(None)
            else:
                d = defer.fail(e)
        else:
            d = defer.execute(self._doAbort, taskid)
        
        return d
    
    def barrier(self, taskids):
        dList = []
        if isinstance(taskids, int):
            taskids = [taskids]
        for id in taskids:
            d = self.get_task_result(id, block=True)
            dList.append(d)
        d = DeferredList(dList, consumeErrors=1)
        d.addCallbacks(lambda r: None)
        return d
    
    def spin(self):
        return defer.succeed(self.distributeTasks())
    
    def queue_status(self, verbose=False):
        pending = self._pendingTaskIDs()
        failed = []
        succeeded = []
        for k,v in self.finishedResults.iteritems():
            if not isinstance(v, failure.Failure):
                if hasattr(v,'failure'):
                    if v.failure is None:
                        succeeded.append(k)
                    else:
                        failed.append(k)
        scheduled = self.scheduler.taskids
        if verbose:
            result = dict(pending=pending, failed=failed, 
                succeeded=succeeded, scheduled=scheduled)
        else:
            result = dict(pending=len(pending),failed=len(failed),
                succeeded=len(succeeded),scheduled=len(scheduled))
        return defer.succeed(result)
    
    #---------------------------------------------------------------------------
    # Queue methods
    #---------------------------------------------------------------------------
    
    def _doAbort(self, taskid):
        """
        Helper function for aborting a pending task.
        """
        log.msg("Task aborted: %i" % taskid)
        result = failure.Failure(error.TaskAborted())
        self._finishTask(taskid, result)
        if taskid in self.abortPending:
            self.abortPending.remove(taskid)
    
    def _finishTask(self, taskid, result):
        dlist = self.deferredResults.pop(taskid)
        # result.taskid = taskid   # The TaskResult should save the taskid
        self.finishedResults[taskid] = result
        for d in dlist:
            d.callback(result)
    
    def distributeTasks(self):
        """
        Distribute tasks while self.scheduler has things to do.
        """
        log.msg("distributing Tasks")
        worker, task = self.scheduler.schedule()
        if not worker and not task:
            if self.idleLater and self.idleLater.called:# we are inside failIdle
                self.idleLater = None
            else:
                self.checkIdle()
            return False
        # else something to do:
        while worker and task:
            # get worker and task
            # add to pending
            self.pendingTasks[worker.workerid] = task
            # run/link callbacks
            d = worker.run(task)
            log.msg("Running task %i on worker %i" %(task.taskid, worker.workerid))
            d.addBoth(self.taskCompleted, task.taskid, worker.workerid)
            worker, task = self.scheduler.schedule()
        # check for idle timeout:
        self.checkIdle()
        return True
    
    def checkIdle(self):
        if self.idleLater and not self.idleLater.called:
            self.idleLater.cancel()
        if self.scheduler.ntasks and self.workers and \
                    self.scheduler.nworkers == len(self.workers):
            self.idleLater = reactor.callLater(self.timeout, self.failIdle)
        else:
            self.idleLater = None
    
    def failIdle(self):
        if not self.distributeTasks():
            while self.scheduler.ntasks:
                t = self.scheduler.pop_task()
                msg = "task %i failed to execute due to unmet dependencies"%t.taskid
                msg += " for %i seconds"%self.timeout
                log.msg("Task aborted by timeout: %i" % t.taskid)
                f = failure.Failure(error.TaskTimeout(msg))
                self._finishTask(t.taskid, f)
        self.idleLater = None
                
    
    def taskCompleted(self, success_and_result, taskid, workerid):
        """This is the err/callback for a completed task."""
        success, result = success_and_result
        try:
            task = self.pendingTasks.pop(workerid)
        except:
            # this should not happen
            log.msg("Tried to pop bad pending task %i from worker %i"%(taskid, workerid))
            log.msg("Result: %r"%result)
            log.msg("Pending tasks: %s"%self.pendingTasks)
            return
        
        # Check if aborted while pending
        aborted = False
        if taskid in self.abortPending:
            self._doAbort(taskid)
            aborted = True
        
        if not aborted:
            if not success:
                log.msg("Task %i failed on worker %i"% (taskid, workerid))
                if task.retries > 0: # resubmit
                    task.retries -= 1
                    self.scheduler.add_task(task)
                    s = "Resubmitting task %i, %i retries remaining" %(taskid, task.retries)
                    log.msg(s)
                    self.distributeTasks()
                elif isinstance(task.recovery_task, BaseTask) and \
                                    task.recovery_task.retries > -1:
                    # retries = -1 is to prevent infinite recovery_task loop
                    task.retries = -1 
                    task.recovery_task.taskid = taskid
                    task = task.recovery_task
                    self.scheduler.add_task(task)
                    s = "Recovering task %i, %i retries remaining" %(taskid, task.retries)
                    log.msg(s)
                    self.distributeTasks()
                else: # done trying
                    self._finishTask(taskid, result)
                # wait a second before readmitting a worker that failed
                # it may have died, and not yet been unregistered
                reactor.callLater(self.failurePenalty, self.readmitWorker, workerid)
            else: # we succeeded
                log.msg("Task completed: %i"% taskid)
                self._finishTask(taskid, result)
                self.readmitWorker(workerid)
        else: # we aborted the task
            if not success:
                reactor.callLater(self.failurePenalty, self.readmitWorker, workerid)
            else:
                self.readmitWorker(workerid)
    
    def readmitWorker(self, workerid):
        """
        Readmit a worker to the scheduler.  
        
        This is outside `taskCompleted` because of the `failurePenalty` being 
        implemented through `reactor.callLater`.
        """
        
        if workerid in self.workers.keys() and workerid not in self.pendingTasks.keys():
            self.scheduler.add_worker(self.workers[workerid])
            self.distributeTasks()
    
    def clear(self):
        """
        Clear all previously run tasks from the task controller.
        
        This is needed because the task controller keep all task results
        in memory.  This can be a problem is there are many completed
        tasks.  Users should call this periodically to clean out these
        cached task results.
        """
        self.finishedResults = {}
        return defer.succeed(None)
        
    
components.registerAdapter(TaskController, cs.IControllerBase, ITaskController)