IPython/kernel/taskclient.py

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

"""The Generic Task Client object.  

This must be subclassed based on your connection method.
"""

__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
#-------------------------------------------------------------------------------

from zope.interface import Interface, implements
from twisted.python import components, log

from IPython.kernel.twistedutil import blockingCallFromThread
from IPython.kernel import task, error

#-------------------------------------------------------------------------------
# Connecting Task Client
#-------------------------------------------------------------------------------

class InteractiveTaskClient(object):
    
    def irun(self, *args, **kwargs):
        """Run a task on the `TaskController`.
        
        This method is a shorthand for run(task) and its arguments are simply
        passed onto a `Task` object:
        
        irun(*args, **kwargs) -> run(Task(*args, **kwargs))

        :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.
            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.
            options : dict
                Any other keyword options for more elaborate uses of tasks
            
        :Returns: A `TaskResult` object.      
        """
        block = kwargs.pop('block', False)
        if len(args) == 1 and isinstance(args[0], task.Task):
            t = args[0]
        else:
            t = task.Task(*args, **kwargs)
        taskid = self.run(t)
        print "TaskID = %i"%taskid
        if block:
            return self.get_task_result(taskid, block)
        else:
            return taskid

class IBlockingTaskClient(Interface):
    """
    An interface for blocking task clients.
    """
    pass


class BlockingTaskClient(InteractiveTaskClient):
    """
    This class provides a blocking task client.
    """
    
    implements(IBlockingTaskClient)
    
    def __init__(self, task_controller):
        self.task_controller = task_controller
        self.block = True
        
    def run(self, task):
        """
        Run a task and return a task id that can be used to get the task result.
        
        :Parameters:
            task : `Task`
                The `Task` object to run
        """
        return blockingCallFromThread(self.task_controller.run, task)
    
    def get_task_result(self, taskid, block=False):
        """
        Get or poll for a task result.
        
        :Parameters:
            taskid : int
                The id of the task whose result to get
            block : boolean
                If True, wait until the task is done and then result the
                `TaskResult` object.  If False, just poll for the result and
                return None if the task is not done.
        """
        return blockingCallFromThread(self.task_controller.get_task_result,
            taskid, block)
    
    def abort(self, taskid):
        """
        Abort a task by task id if it has not been started.
        """
        return blockingCallFromThread(self.task_controller.abort, taskid)
    
    def barrier(self, taskids):
        """
        Wait for a set of tasks to finish.
        
        :Parameters:
            taskids : list of ints
                A list of task ids to wait for.
        """
        return blockingCallFromThread(self.task_controller.barrier, taskids)
    
    def spin(self):
        """
        Cause the scheduler to schedule tasks.
        
        This method only needs to be called in unusual situations where the
        scheduler is idle for some reason.
        """
        return blockingCallFromThread(self.task_controller.spin)
    
    def queue_status(self, verbose=False):
        """
        Get a dictionary with the current state of the task queue.
        
        :Parameters:
            verbose : boolean
                If True, return a list of taskids.  If False, simply give
                the number of tasks with each status.
        
        :Returns:
            A dict with the queue status.
        """
        return blockingCallFromThread(self.task_controller.queue_status, verbose)


components.registerAdapter(BlockingTaskClient,
            task.ITaskController, IBlockingTaskClient)