# 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 #------------------------------------------------------------------------------- import copy, time from types import FunctionType as function import zope.interface as zi, string from twisted.internet import defer, reactor from twisted.python import components, log, failure # from IPython.genutils import time 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 def canTask(task): t = copy.copy(task) t.depend = can(t.depend) if t.recovery_task: t.recovery_task = canTask(t.recovery_task) return t def uncanTask(task): t = copy.copy(task) t.depend = uncan(t.depend) if t.recovery_task and t.recovery_task is not task: t.recovery_task = uncanTask(t.recovery_task) return t time_format = '%Y/%m/%d %H:%M:%S' class Task(object): """Our representation of a task for the `TaskController` interface. The user should create instances of this class to represent a task that needs to be done. :Parameters: expression : str A str that is valid python code that is the task. pull : str or list of str The names of objects to be pulled as results. If not specified, will return {'result', None} push : dict A dict of objects to be pushed into the engines namespace before execution of the expression. clear_before : boolean Should the engine's namespace be cleared before the task is run. Default=False. clear_after : boolean Should the engine's namespace be cleared after the task is run. Default=False. retries : int The number of times to resumbit the task if it fails. Default=0. recovery_task : Task This is the Task to be run when the task has exhausted its retries Default=None. depend : bool function(properties) This is the dependency function for the Task, which determines whether a task can be run on a Worker. `depend` is called with one argument, the worker's properties dict, and should return True if the worker meets the dependencies or False if it does not. Default=None - run on any worker options : dict Any other keyword options for more elaborate uses of tasks Examples -------- >>> t = Task('dostuff(args)') >>> t = Task('a=5', pull='a') >>> t = Task('a=5\nb=4', pull=['a','b']) >>> t = Task('os.kill(os.getpid(),9)', retries=100) # this is a bad idea # A dependency case: >>> def hasMPI(props): ... return props.get('mpi') is not None >>> t = Task('mpi.send(blah,blah)', depend = hasMPI) """ def __init__(self, expression, pull=None, push=None, clear_before=False, clear_after=False, retries=0, recovery_task=None, depend=None, **options): self.expression = expression if isinstance(pull, str): self.pull = [pull] else: self.pull = pull self.push = push self.clear_before = clear_before self.clear_after = clear_after self.retries=retries self.recovery_task = recovery_task self.depend = depend self.options = options self.taskid = None class ResultNS: """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':range(3)} >>> 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. 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 raiseException() 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 in the ipython task system, 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("I am protected!") 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.raiseException() return self.results[key] def raiseException(self): """Re-raise any remote exceptions in the local python session.""" if self.failure is not None: self.failure.raiseException() 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 `TaskResult` object. """ 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. :Parameters: task : a `Task` object :Returns: `Deferred` to a `TaskResult` object. """ if task.clear_before: d = self.queuedEngine.reset() else: d = defer.succeed(None) if task.push is not None: d.addCallback(lambda r: self.queuedEngine.push(task.push)) d.addCallback(lambda r: self.queuedEngine.execute(task.expression)) if task.pull is not None: d.addCallback(lambda r: self.queuedEngine.pull(task.pull)) else: d.addCallback(lambda r: None) def reseter(result): self.queuedEngine.reset() return result if task.clear_after: d.addBoth(reseter) return d.addBoth(self._zipResults, task.pull, time.time(), time.localtime()) def _zipResults(self, result, names, start, start_struct): """Callback for construting the TaskResult object.""" if isinstance(result, failure.Failure): tr = TaskResult(result, self.queuedEngine.id) else: if names is None: resultDict = {} elif len(names) == 1: resultDict = {names[0]:result} else: resultDict = dict(zip(names, result)) tr = TaskResult(resultDict, self.queuedEngine.id) # the time info tr.submitted = time.strftime(time_format, start_struct) tr.completed = time.strftime(time_format) tr.duration = time.time()-start return tr 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 : a `Task` object The task to be queued. flags : dict General keywords for more sophisticated scheduling """ def pop_task(id=None): """Pops a Task object. 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: a `Task` object :Exceptions: IndexError : raised if no taskid in queue """ def add_worker(worker, **flags): """Add a worker to the worker queue. :Parameters: worker : an IWorker implementing object flags : 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 a tuple of the worker and 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 cando = t.depend is None or t.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 (taskid, actualResult) 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(self, 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. """ 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 "We already have one! This should not happen." 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 a TaskResult tuple 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, result, taskid, workerid): """This is the err/callback for a completed task.""" 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 result.failure is not None and isinstance(result.failure, failure.Failure): # we failed 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, Task) 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 result.failure is not None and isinstance(result.failure, failure.Failure): # it failed, penalize worker 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() components.registerAdapter(TaskController, cs.IControllerBase, ITaskController)