The refactoring of the Task system is nearly complete. Now there are

multiple types of tasks including `StringTask` and `MapTask`.  Each
task type is responsible for running itself and processing its own
result.  This makes it much easier for people to create new task types.

Also, the map and parallel function support has been completely 
refactored and improved.  This includes a map and parallel function
implementation for the task controller as well as a @parallel decorator.

docs/examples/kernel/multienginemap.py

@@ -0,0 +1,18 b''
+from IPython.kernel import client
+
+mec = client.MultiEngineClient()
+
+result = mec.map(lambda x: 2*x, range(10))
+print "Simple, default map: ", result
+
+m = mec.mapper(block=False)
+pr = m.map(lambda x: 2*x, range(10))
+print "Submitted map, got PendingResult: ", pr
+result = pr.r
+print "Using a mapper: ", result
+
[e-mail]()
+def f(x): return 2*x
+
+result = f(range(10))
+print "Using a parallel function: ", result
  No newline at end of file

docs/examples/kernel/taskmap.py

@@ -0,0 +1,19 b''
+from IPython.kernel import client
+
+tc = client.TaskClient()
+
+result = tc.map(lambda x: 2*x, range(10))
+print "Simple, default map: ", result
+
+m = tc.mapper(block=False, clear_after=True, clear_before=True)
+tids = m.map(lambda x: 2*x, range(10))
+print "Submitted tasks, got ids: ", tids
+tc.barrier(tids)
+result = [tc.get_task_result(tid) for tid in tids]
+print "Using a mapper: ", result
+
[e-mail]()
+def f(x): return 2*x
+
+result = f(range(10))
+print "Using a parallel function: ", result
  No newline at end of file

IPython/kernel/asyncclient.py

@@ -1,41 +1,41 b''
 # encoding: utf-8
 
 """Asynchronous clients for the IPython controller.
 
 This module has clients for using the various interfaces of the controller 
 in a fully asynchronous manner.  This means that you will need to run the
 Twisted reactor yourself and that all methods of the client classes return
 deferreds to the result.
 
 The main methods are are `get_*_client` and `get_client`.
 """
 
 __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 IPython.kernel import codeutil
 from IPython.kernel.clientconnector import ClientConnector
 
 # Other things that the user will need
-from IPython.kernel.task import Task
+from IPython.kernel.task import MapTask, StringTask
 from IPython.kernel.error import CompositeError
 
 #-------------------------------------------------------------------------------
 # Code
 #-------------------------------------------------------------------------------
 
 _client_tub = ClientConnector()
 get_multiengine_client = _client_tub.get_multiengine_client
 get_task_client = _client_tub.get_task_client
 get_client = _client_tub.get_client
 

IPython/kernel/client.py

@@ -1,96 +1,96 b''
 # encoding: utf-8
 
 """This module contains blocking clients for the controller interfaces.
 
 Unlike the clients in `asyncclient.py`, the clients in this module are fully
 blocking.  This means that methods on the clients return the actual results
 rather than a deferred to the result.  Also, we manage the Twisted reactor
 for you.  This is done by running the reactor in a thread.
 
 The main classes in this module are:
 
     * MultiEngineClient
     * TaskClient
     * Task
     * CompositeError
 """
 
 __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 sys
 
 # from IPython.tools import growl
 # growl.start("IPython1 Client")
 
 
 from twisted.internet import reactor
 from IPython.kernel.clientconnector import ClientConnector
 from IPython.kernel.twistedutil import ReactorInThread
 from IPython.kernel.twistedutil import blockingCallFromThread
 
 # These enable various things 
 from IPython.kernel import codeutil
 import IPython.kernel.magic
 
 # Other things that the user will need
-from IPython.kernel.task import Task
+from IPython.kernel.task import MapTask, StringTask
 from IPython.kernel.error import CompositeError
 
 #-------------------------------------------------------------------------------
 # Code
 #-------------------------------------------------------------------------------
 
 _client_tub = ClientConnector()
 
 
 def get_multiengine_client(furl_or_file=''):
     """Get the blocking MultiEngine client.
     
     :Parameters:
         furl_or_file : str
             A furl or a filename containing a furl.  If empty, the
             default furl_file will be used
             
     :Returns:
         The connected MultiEngineClient instance
     """
     client = blockingCallFromThread(_client_tub.get_multiengine_client, 
         furl_or_file)
     return client.adapt_to_blocking_client()
 
 def get_task_client(furl_or_file=''):
     """Get the blocking Task client.
     
     :Parameters:
         furl_or_file : str
             A furl or a filename containing a furl.  If empty, the
             default furl_file will be used
             
     :Returns:
         The connected TaskClient instance
     """
     client = blockingCallFromThread(_client_tub.get_task_client, 
         furl_or_file)
     return client.adapt_to_blocking_client()
 
 
 MultiEngineClient = get_multiengine_client
 TaskClient = get_task_client
 
 
 
 # Now we start the reactor in a thread
 rit = ReactorInThread()
 rit.setDaemon(True)
 rit.start()
  No newline at end of file

IPython/kernel/magic.py

@@ -1,171 +1,171 b''
 # encoding: utf-8
 
 """Magic command interface for interactive parallel work."""
 
 __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 new
 
 from IPython.iplib import InteractiveShell
 from IPython.Shell import MTInteractiveShell
 
 from twisted.internet.defer import Deferred
 
 
 #-------------------------------------------------------------------------------
 # Definitions of magic functions for use with IPython
 #-------------------------------------------------------------------------------
 
 NO_ACTIVE_CONTROLLER = """
 Error:  No Controller is activated
 Use activate() on a RemoteController object to activate it for magics.
 """
 
 def magic_result(self,parameter_s=''):
     """Print the result of command i on all engines of the active controller.
     
     To activate a controller in IPython, first create it and then call
     the activate() method.
     
     Then you can do the following:
     
     >>> result                                # Print the latest result
     Printing result... 
     [127.0.0.1:0] In [1]: b = 10
     [127.0.0.1:1] In [1]: b = 10
     
     >>> result 0                              # Print result 0
     In [14]: result 0
     Printing result... 
     [127.0.0.1:0] In [0]: a = 5
     [127.0.0.1:1] In [0]: a = 5
     """
     try:
         activeController = __IPYTHON__.activeController
     except AttributeError:
         print NO_ACTIVE_CONTROLLER
     else:
         try:
             index = int(parameter_s)
         except:
             index = None
         result = activeController.get_result(index)
         return result
 
 def magic_px(self,parameter_s=''):
     """Executes the given python command on the active IPython Controller.
     
     To activate a Controller in IPython, first create it and then call
     the activate() method.
     
     Then you can do the following:
      
     >>> %px a = 5       # Runs a = 5 on all nodes
     """
 
     try:
         activeController = __IPYTHON__.activeController
     except AttributeError:
         print NO_ACTIVE_CONTROLLER
     else:
-        print "Executing command on Controller"
+        print "Parallel execution on engines: %s" % activeController.targets
         result = activeController.execute(parameter_s)
         return result
 
 def pxrunsource(self, source, filename="<input>", symbol="single"):
 
     try:
         code = self.compile(source, filename, symbol)
     except (OverflowError, SyntaxError, ValueError):
         # Case 1
         self.showsyntaxerror(filename)
         return None
 
     if code is None:
         # Case 2
         return True
 
     # Case 3
     # Because autopx is enabled, we now call executeAll or disable autopx if
     # %autopx or autopx has been called
     if '_ip.magic("%autopx' in source or '_ip.magic("autopx' in source:
         _disable_autopx(self)
         return False
     else:
         try:
             result = self.activeController.execute(source)
         except:
             self.showtraceback()
         else:
             print result.__repr__()
         return False
         
 def magic_autopx(self, parameter_s=''):
     """Toggles auto parallel mode for the active IPython Controller.
     
     To activate a Controller in IPython, first create it and then call
     the activate() method.
     
     Then you can do the following:
      
     >>> %autopx                    # Now all commands are executed in parallel
     Auto Parallel Enabled
     Type %autopx to disable
     ...
     >>> %autopx                    # Now all commands are locally executed
     Auto Parallel Disabled
     """
   
     if hasattr(self, 'autopx'):
         if self.autopx == True:
             _disable_autopx(self)
         else:
             _enable_autopx(self)
     else:
         _enable_autopx(self)
 
 def _enable_autopx(self):
     """Enable %autopx mode by saving the original runsource and installing 
     pxrunsource.
     """
     try:
         activeController = __IPYTHON__.activeController
     except AttributeError:
         print "No active RemoteController found, use RemoteController.activate()."
     else:
         self._original_runsource = self.runsource
         self.runsource = new.instancemethod(pxrunsource, self, self.__class__)
         self.autopx = True
         print "Auto Parallel Enabled\nType %autopx to disable"
 
 def _disable_autopx(self):
     """Disable %autopx by restoring the original runsource."""
     if hasattr(self, 'autopx'):
         if self.autopx == True:
             self.runsource = self._original_runsource
             self.autopx = False
             print "Auto Parallel Disabled"
             
 # Add the new magic function to the class dict:
 
 InteractiveShell.magic_result = magic_result
 InteractiveShell.magic_px = magic_px
 InteractiveShell.magic_autopx = magic_autopx
 
 # And remove the global name to keep global namespace clean.  Don't worry, the
 # copy bound to IPython stays, we're just removing the global name.
 del magic_result
 del magic_px
 del magic_autopx
 

IPython/kernel/mapper.py

@@ -1,42 +1,233 b''
 # encoding: utf-8
 
 """A parallelized version of Python's builtin map."""
 
 __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 types import FunctionType
 from zope.interface import Interface, implements
+from IPython.kernel.task import MapTask
+from IPython.kernel.twistedutil import DeferredList, gatherBoth
+from IPython.kernel.util import printer
+from IPython.kernel.error import collect_exceptions
+
+#----------------------------------------------------------------------------
+# Code
+#----------------------------------------------------------------------------
 
 class IMapper(Interface):
+    """The basic interface for a Mapper.
+    
+    This defines a generic interface for mapping.  The idea of this is 
+    similar to that of Python's builtin `map` function, which applies a function
+    elementwise to a sequence.
+    """
+    
+    def map(func, *seqs):
+        """Do map in parallel.
+        
+        Equivalent to map(func, *seqs) or:
+        
+        [func(seqs[0][0], seqs[1][0],...), func(seqs[0][1], seqs[1][1],...),...]
+        
+        :Parameters:
+            func : FunctionType
+                The function to apply to the sequence
+            sequences : tuple of iterables
+                A sequence of iterables that are used for sucessive function
+                arguments.  This work just like map
+        """
+
+class IMultiEngineMapperFactory(Interface):
+    """
+    An interface for something that creates `IMapper` instances.
+    """
+    
+    def mapper(dist='b', targets='all', block=True):
+        """
+        Create an `IMapper` implementer with a given set of arguments.
+        
+        The `IMapper` created using a multiengine controller is 
+        not load balanced.
+        """
+
+class ITaskMapperFactory(Interface):
+    """
+    An interface for something that creates `IMapper` instances.
+    """
     
-    def __call__(func, *sequences):
-        """Do map in parallel."""
+    def mapper(clear_before=False, clear_after=False, retries=0, 
+                recovery_task=None, depend=None, block=True):
+        """
+        Create an `IMapper` implementer with a given set of arguments.
+        
+        The `IMapper` created using a task controller is load balanced.
+        
+        See the documentation for `IPython.kernel.task.BaseTask` for 
+        documentation on the arguments to this method.
+        """
 
-class Mapper(object):
+
+class MultiEngineMapper(object):
+    """
+    A Mapper for `IMultiEngine` implementers.
+    """
     
     implements(IMapper)
     
     def __init__(self, multiengine, dist='b', targets='all', block=True):
+        """
+        Create a Mapper for a multiengine.
+        
+        The value of all arguments are used for all calls to `map`.  This
+        class allows these arguemnts to be set for a series of map calls.
+        
+        :Parameters:
+            multiengine : `IMultiEngine` implementer
+                The multiengine to use for running the map commands
+            dist : str
+                The type of decomposition to use.  Only block ('b') is
+                supported currently
+            targets : (str, int, tuple of ints)
+                The engines to use in the map
+            block : boolean
+                Whether to block when the map is applied
+        """
         self.multiengine = multiengine
         self.dist = dist
         self.targets = targets
         self.block = block
-        
-    def __call__(self, func, *sequences):
-        return self.map(func, *sequences)
     
     def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is not load balanced.
+        """
+        max_len = max(len(s) for s in sequences)
+        for s in sequences:
+            if len(s)!=max_len:
+                raise ValueError('all sequences must have equal length')
         assert isinstance(func, (str, FunctionType)), "func must be a fuction or str"
-        return self.multiengine._map(func, sequences, dist=self.dist,
-            targets=self.targets, block=self.block)
  No newline at end of file
+        return self.multiengine.raw_map(func, sequences, dist=self.dist,
+            targets=self.targets, block=self.block)
+
+class TaskMapper(object):
+    """
+    Make an `ITaskController` look like an `IMapper`.
+    
+    This class provides a load balanced version of `map`.
+    """
+    
+    def __init__(self, task_controller, clear_before=False, clear_after=False, retries=0, 
+            recovery_task=None, depend=None, block=True):
+        """
+        Create a `IMapper` given a `TaskController` and arguments.
+        
+        The additional arguments are those that are common to all types of 
+        tasks and are described in the documentation for 
+        `IPython.kernel.task.BaseTask`.
+        
+        :Parameters:
+            task_controller : an `IBlockingTaskClient` implementer
+                The `TaskController` to use for calls to `map`
+        """
+        self.task_controller = task_controller
+        self.clear_before = clear_before
+        self.clear_after = clear_after
+        self.retries = retries
+        self.recovery_task = recovery_task
+        self.depend = depend
+        self.block = block
+    
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is load balanced.
+        """
+        max_len = max(len(s) for s in sequences)
+        for s in sequences:
+            if len(s)!=max_len:
+                raise ValueError('all sequences must have equal length')
+        task_args = zip(*sequences)
+        task_ids = []
+        dlist = []
+        for ta in task_args:
+            task = MapTask(func, ta, clear_before=self.clear_before,
+                clear_after=self.clear_after, retries=self.retries,
+                recovery_task=self.recovery_task, depend=self.depend)
+            dlist.append(self.task_controller.run(task))
+        dlist = gatherBoth(dlist, consumeErrors=1)
+        dlist.addCallback(collect_exceptions,'map')
+        if self.block:
+            def get_results(task_ids):
+                d = self.task_controller.barrier(task_ids)
+                d.addCallback(lambda _: gatherBoth([self.task_controller.get_task_result(tid) for tid in task_ids], consumeErrors=1))
+                d.addCallback(collect_exceptions, 'map')
+                return d
+            dlist.addCallback(get_results)
+        return dlist
+
+class SynchronousTaskMapper(object):
+    """
+    Make an `IBlockingTaskClient` look like an `IMapper`.
+    
+    This class provides a load balanced version of `map`.
+    """
+    
+    def __init__(self, task_controller, clear_before=False, clear_after=False, retries=0, 
+            recovery_task=None, depend=None, block=True):
+        """
+        Create a `IMapper` given a `IBlockingTaskClient` and arguments.
+        
+        The additional arguments are those that are common to all types of 
+        tasks and are described in the documentation for 
+        `IPython.kernel.task.BaseTask`.
+        
+        :Parameters:
+            task_controller : an `IBlockingTaskClient` implementer
+                The `TaskController` to use for calls to `map`
+        """
+        self.task_controller = task_controller
+        self.clear_before = clear_before
+        self.clear_after = clear_after
+        self.retries = retries
+        self.recovery_task = recovery_task
+        self.depend = depend
+        self.block = block
+    
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is load balanced.
+        """
+        max_len = max(len(s) for s in sequences)
+        for s in sequences:
+            if len(s)!=max_len:
+                raise ValueError('all sequences must have equal length')
+        task_args = zip(*sequences)
+        task_ids = []
+        for ta in task_args:
+            task = MapTask(func, ta, clear_before=self.clear_before,
+                clear_after=self.clear_after, retries=self.retries,
+                recovery_task=self.recovery_task, depend=self.depend)
+            task_ids.append(self.task_controller.run(task))
+        if self.block:
+            self.task_controller.barrier(task_ids)
+            task_results = [self.task_controller.get_task_result(tid) for tid in task_ids]
+            return task_results
+        else:
+            return task_ids
  No newline at end of file

IPython/kernel/multiengine.py

@@ -1,743 +1,753 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_multiengine -*-
 
 """Adapt the IPython ControllerServer to IMultiEngine.
 
 This module provides classes that adapt a ControllerService to the 
 IMultiEngine interface.  This interface is a basic interactive interface
 for working with a set of engines where it is desired to have explicit 
 access to each registered engine.  
 
 The classes here are exposed to the network in files like:
 
 * multienginevanilla.py
 * multienginepb.py
 """
 
 __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 new import instancemethod
 from types import FunctionType
 
 from twisted.application import service
 from twisted.internet import defer, reactor
 from twisted.python import log, components, failure
 from zope.interface import Interface, implements, Attribute
 
 from IPython.tools import growl
 from IPython.kernel.util import printer
 from IPython.kernel.twistedutil import gatherBoth
 from IPython.kernel import map as Map
 from IPython.kernel import error
 from IPython.kernel.pendingdeferred import PendingDeferredManager, two_phase
 from IPython.kernel.controllerservice import \
     ControllerAdapterBase, \
     ControllerService, \
     IControllerBase
 
 
 #-------------------------------------------------------------------------------
 # Interfaces for the MultiEngine representation of a controller
 #-------------------------------------------------------------------------------
 
 class IEngineMultiplexer(Interface):
     """Interface to multiple engines implementing IEngineCore/Serialized/Queued.
     
     This class simply acts as a multiplexer of methods that are in the 
     various IEngines* interfaces.  Thus the methods here are jut like those 
     in the IEngine* interfaces, but with an extra first argument, targets.  
     The targets argument can have the following forms:
     
     * targets = 10            # Engines are indexed by ints
     * targets = [0,1,2,3]     # A list of ints
     * targets = 'all'         # A string to indicate all targets
     
     If targets is bad in any way, an InvalidEngineID will be raised.  This
     includes engines not being registered.
     
     All IEngineMultiplexer multiplexer methods must return a Deferred to a list 
     with length equal to the number of targets.  The elements of the list will 
     correspond to the return of the corresponding IEngine method.
     
     Failures are aggressive, meaning that if an action fails for any target, 
     the overall action will fail immediately with that Failure.
     
     :Parameters:
         targets : int, list of ints, or 'all'
             Engine ids the action will apply to.
     
     :Returns: Deferred to a list of results for each engine.
     
     :Exception:
         InvalidEngineID
             If the targets argument is bad or engines aren't registered.
         NoEnginesRegistered
             If there are no engines registered and targets='all'
     """
         
     #---------------------------------------------------------------------------
     # Mutiplexed methods
     #---------------------------------------------------------------------------
      
     def execute(lines, targets='all'):
         """Execute lines of Python code on targets.
               
         See the class docstring for information about targets and possible
         exceptions this method can raise.
               
         :Parameters:
             lines : str
                 String of python code to be executed on targets.
         """
             
     def push(namespace, targets='all'):
         """Push dict namespace into the user's namespace on targets.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             namspace : dict
                 Dict of key value pairs to be put into the users namspace.
         """
         
     def pull(keys, targets='all'):
         """Pull values out of the user's namespace on targets by keys.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             keys : tuple of strings
                 Sequence of keys to be pulled from user's namespace.
         """
           
     def push_function(namespace, targets='all'):
         """"""
         
     def pull_function(keys, targets='all'):
         """"""
                 
     def get_result(i=None, targets='all'):
         """Get the result for command i from targets.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             i : int or None
                 Command index or None to indicate most recent command.                
         """
         
     def reset(targets='all'):
         """Reset targets.
         
         This clears the users namespace of the Engines, but won't cause
         modules to be reloaded.
         """
         
     def keys(targets='all'):
         """Get variable names defined in user's namespace on targets."""
         
     def kill(controller=False, targets='all'):
         """Kill the targets Engines and possibly the controller.
         
         :Parameters:
             controller : boolean
                 Should the controller be killed as well.  If so all the 
                 engines will be killed first no matter what targets is.
         """
         
     def push_serialized(namespace, targets='all'):
         """Push a namespace of Serialized objects to targets.
         
         :Parameters:
             namespace : dict
                 A dict whose keys are the variable names and whose values
                 are serialized version of the objects.
         """
         
     def pull_serialized(keys, targets='all'):
         """Pull Serialized objects by keys from targets.
         
         :Parameters:
             keys : tuple of strings
                 Sequence of variable names to pull as serialized objects.
         """
         
     def clear_queue(targets='all'):
         """Clear the queue of pending command for targets."""
         
     def queue_status(targets='all'):
         """Get the status of the queue on the targets."""
     
     def set_properties(properties, targets='all'):
         """set properties by key and value"""
     
     def get_properties(keys=None, targets='all'):
         """get a list of properties by `keys`, if no keys specified, get all"""
     
     def del_properties(keys, targets='all'):
         """delete properties by `keys`"""
     
     def has_properties(keys, targets='all'):
         """get a list of bool values for whether `properties` has `keys`"""
     
     def clear_properties(targets='all'):
         """clear the properties dict"""
 
 
 class IMultiEngine(IEngineMultiplexer):
     """A controller that exposes an explicit interface to all of its engines.
     
     This is the primary inteface for interactive usage.
     """
     
     def get_ids():
         """Return list of currently registered ids.
                 
         :Returns:  A Deferred to a list of registered engine ids.
         """
 
 
 
 #-------------------------------------------------------------------------------
 # Implementation of the core MultiEngine classes
 #-------------------------------------------------------------------------------
 
 class MultiEngine(ControllerAdapterBase):
     """The representation of a ControllerService as a IMultiEngine.
     
     Although it is not implemented currently, this class would be where a
     client/notification API is implemented.  It could inherit from something
     like results.NotifierParent and then use the notify method to send
     notifications.
     """
     
     implements(IMultiEngine)
     
     def __init(self, controller):
         ControllerAdapterBase.__init__(self, controller)
     
     #---------------------------------------------------------------------------
     # Helper methods
     #---------------------------------------------------------------------------
     
     def engineList(self, targets):
         """Parse the targets argument into a list of valid engine objects.
         
         :Parameters:
             targets : int, list of ints or 'all'
                 The targets argument to be parsed.
                 
         :Returns: List of engine objects.
         
         :Exception:
             InvalidEngineID
                 If targets is not valid or if an engine is not registered.
         """
         if isinstance(targets, int):
             if targets not in self.engines.keys():
                 log.msg("Engine with id %i is not registered" % targets)
                 raise error.InvalidEngineID("Engine with id %i is not registered" % targets)
             else: 
                 return [self.engines[targets]]
         elif isinstance(targets, (list, tuple)):
             for id in targets:
                 if id not in self.engines.keys():
                     log.msg("Engine with id %r is not registered" % id)
                     raise error.InvalidEngineID("Engine with id %r is not registered" % id)  
             return map(self.engines.get, targets)
         elif targets == 'all':
             eList = self.engines.values()
             if len(eList) == 0:
                 msg = """There are no engines registered.
                      Check the logs in ~/.ipython/log if you think there should have been."""
                 raise error.NoEnginesRegistered(msg)
             else:
                 return eList
         else:
             raise error.InvalidEngineID("targets argument is not an int, list of ints or 'all': %r"%targets)
     
     def _performOnEngines(self, methodName, *args, **kwargs):
         """Calls a method on engines and returns deferred to list of results.
         
         :Parameters:
             methodName : str
                 Name of the method to be called.
             targets : int, list of ints, 'all'
                 The targets argument to be parsed into a list of engine objects.
             args
                 The positional keyword arguments to be passed to the engines.
             kwargs
                 The keyword arguments passed to the method
                 
         :Returns: List of deferreds to the results on each engine
         
         :Exception:
             InvalidEngineID
                 If the targets argument is bad in any way.
             AttributeError
                 If the method doesn't exist on one of the engines.
         """
         targets = kwargs.pop('targets')
         log.msg("Performing %s on %r" % (methodName, targets))
         # log.msg("Performing %s(%r, %r) on %r" % (methodName, args, kwargs, targets))
         # This will and should raise if targets is not valid!
         engines = self.engineList(targets)
         dList = []
         for e in engines:
             meth = getattr(e, methodName, None)
             if meth is not None:
                 dList.append(meth(*args, **kwargs))
             else:
                 raise AttributeError("Engine %i does not have method %s" % (e.id, methodName))
         return dList
         
     def _performOnEnginesAndGatherBoth(self, methodName, *args, **kwargs):
         """Called _performOnEngines and wraps result/exception into deferred."""
         try:
             dList = self._performOnEngines(methodName, *args, **kwargs)
         except (error.InvalidEngineID, AttributeError, KeyError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())
         else:
             # Having fireOnOneErrback is causing problems with the determinacy
             # of the system.  Basically, once a single engine has errbacked, this
             # method returns.  In some cases, this will cause client to submit
             # another command.  Because the previous command is still running
             # on some engines, this command will be queued.  When those commands
             # then errback, the second command will raise QueueCleared.  Ahhh!
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, methodName)
             return d
               
     #---------------------------------------------------------------------------
     # General IMultiEngine methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         return defer.succeed(self.engines.keys())
     
     #---------------------------------------------------------------------------
     # IEngineMultiplexer methods
     #---------------------------------------------------------------------------
         
     def execute(self, lines, targets='all'):
         return self._performOnEnginesAndGatherBoth('execute', lines, targets=targets)
     
     def push(self, ns, targets='all'):
         return self._performOnEnginesAndGatherBoth('push', ns, targets=targets)
         
     def pull(self, keys, targets='all'):
         return self._performOnEnginesAndGatherBoth('pull', keys, targets=targets)
     
     def push_function(self, ns, targets='all'):
         return self._performOnEnginesAndGatherBoth('push_function', ns, targets=targets)
         
     def pull_function(self, keys, targets='all'):
         return self._performOnEnginesAndGatherBoth('pull_function', keys, targets=targets)
     
     def get_result(self, i=None, targets='all'):
         return self._performOnEnginesAndGatherBoth('get_result', i, targets=targets)
     
     def reset(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('reset', targets=targets)
     
     def keys(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('keys', targets=targets)
     
     def kill(self, controller=False, targets='all'):
         if controller:
             targets = 'all'
         d = self._performOnEnginesAndGatherBoth('kill', targets=targets)
         if controller:
             log.msg("Killing controller")
             d.addCallback(lambda _: reactor.callLater(2.0, reactor.stop))
             # Consume any weird stuff coming back
             d.addBoth(lambda _: None)
         return d
     
     def push_serialized(self, namespace, targets='all'):
         for k, v in namespace.iteritems():
             log.msg("Pushed object %s is %f MB" % (k, v.getDataSize()))
         d = self._performOnEnginesAndGatherBoth('push_serialized', namespace, targets=targets)      
         return d
         
     def pull_serialized(self, keys, targets='all'):
         try:
             dList = self._performOnEngines('pull_serialized', keys, targets=targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())
         else:
             for d in dList:
                 d.addCallback(self._logSizes)
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'pull_serialized')
             return d  
                               
     def _logSizes(self, listOfSerialized):
         if isinstance(listOfSerialized, (list, tuple)):
             for s in listOfSerialized:
                 log.msg("Pulled object is %f MB" % s.getDataSize())
         else:
             log.msg("Pulled object is %f MB" % listOfSerialized.getDataSize())
         return listOfSerialized
     
     def clear_queue(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('clear_queue', targets=targets)         
     
     def queue_status(self, targets='all'):
         log.msg("Getting queue status on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = []
             for e in engines:
                 dList.append(e.queue_status().addCallback(lambda s:(e.id, s)))
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'queue_status')
             return d 
     
     def get_properties(self, keys=None, targets='all'):
         log.msg("Getting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.get_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'get_properties')
             return d
     
     def set_properties(self, properties, targets='all'):
         log.msg("Setting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.set_properties(properties) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'set_properties')
             return d
     
     def has_properties(self, keys, targets='all'):
         log.msg("Checking properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.has_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'has_properties')
             return d
     
     def del_properties(self, keys, targets='all'):
         log.msg("Deleting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.del_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'del_properties')
             return d
     
     def clear_properties(self, targets='all'):
         log.msg("Clearing properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.clear_properties() for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'clear_properties')
             return d
 
 
 components.registerAdapter(MultiEngine,
                            IControllerBase, 
                            IMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # Interfaces for the Synchronous MultiEngine
 #-------------------------------------------------------------------------------
 
 class ISynchronousEngineMultiplexer(Interface):
     pass
 
 
 class ISynchronousMultiEngine(ISynchronousEngineMultiplexer):
     """Synchronous, two-phase version of IMultiEngine.
     
     Methods in this interface are identical to those of IMultiEngine, but they
     take one additional argument:
     
     execute(lines, targets='all') -> execute(lines, targets='all, block=True)
     
     :Parameters:
         block : boolean
             Should the method return a deferred to a deferredID or the 
             actual result.  If block=False a deferred to a deferredID is 
             returned and the user must call `get_pending_deferred` at a later
             point.  If block=True, a deferred to the actual result comes back.
     """
     def get_pending_deferred(deferredID, block=True):
         """"""
     
     def clear_pending_deferreds():
         """"""
 
 
 #-------------------------------------------------------------------------------
 # Implementation of the Synchronous MultiEngine
 #-------------------------------------------------------------------------------
 
 class SynchronousMultiEngine(PendingDeferredManager):
     """Adapt an `IMultiEngine` -> `ISynchronousMultiEngine`
     
     Warning, this class uses a decorator that currently uses **kwargs.  
     Because of this block must be passed as a kwarg, not positionally.
     """
     
     implements(ISynchronousMultiEngine)
     
     def __init__(self, multiengine):
         self.multiengine = multiengine
         PendingDeferredManager.__init__(self)
     
     #---------------------------------------------------------------------------
     # Decorated pending deferred methods
     #---------------------------------------------------------------------------
     
     @two_phase
     def execute(self, lines, targets='all'):
         d = self.multiengine.execute(lines, targets)
         return d
     
     @two_phase
     def push(self, namespace, targets='all'):
         return self.multiengine.push(namespace, targets)
     
     @two_phase
     def pull(self, keys, targets='all'):
         d = self.multiengine.pull(keys, targets)
         return d
     
     @two_phase
     def push_function(self, namespace, targets='all'):
         return self.multiengine.push_function(namespace, targets)
     
     @two_phase
     def pull_function(self, keys, targets='all'):
         d = self.multiengine.pull_function(keys, targets)
         return d
     
     @two_phase
     def get_result(self, i=None, targets='all'):
         return self.multiengine.get_result(i, targets='all')
     
     @two_phase
     def reset(self, targets='all'):
         return self.multiengine.reset(targets)
     
     @two_phase
     def keys(self, targets='all'):
         return self.multiengine.keys(targets)
     
     @two_phase
     def kill(self, controller=False, targets='all'):
         return self.multiengine.kill(controller, targets)
     
     @two_phase
     def push_serialized(self, namespace, targets='all'):
         return self.multiengine.push_serialized(namespace, targets)
     
     @two_phase
     def pull_serialized(self, keys, targets='all'):
         return self.multiengine.pull_serialized(keys, targets)
     
     @two_phase
     def clear_queue(self, targets='all'):
         return self.multiengine.clear_queue(targets)
     
     @two_phase
     def queue_status(self, targets='all'):
         return self.multiengine.queue_status(targets)
     
     @two_phase
     def set_properties(self, properties, targets='all'):
         return self.multiengine.set_properties(properties, targets)
     
     @two_phase
     def get_properties(self, keys=None, targets='all'):
         return self.multiengine.get_properties(keys, targets)
     
     @two_phase
     def has_properties(self, keys, targets='all'):
         return self.multiengine.has_properties(keys, targets)
     
     @two_phase
     def del_properties(self, keys, targets='all'):
         return self.multiengine.del_properties(keys, targets)
     
     @two_phase
     def clear_properties(self, targets='all'):
         return self.multiengine.clear_properties(targets)
     
     #---------------------------------------------------------------------------
     # IMultiEngine methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         """Return a list of registered engine ids.
         
         Never use the two phase block/non-block stuff for this.
         """
         return self.multiengine.get_ids()
 
 
 components.registerAdapter(SynchronousMultiEngine, IMultiEngine, ISynchronousMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # Various high-level interfaces that can be used as MultiEngine mix-ins
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # IMultiEngineCoordinator
 #-------------------------------------------------------------------------------
 
 class IMultiEngineCoordinator(Interface):
     """Methods that work on multiple engines explicitly."""
        
     def scatter(key, seq, dist='b', flatten=False, targets='all'):
         """Partition and distribute a sequence to targets."""
         
     def gather(key, dist='b', targets='all'):
         """Gather object key from targets."""
     
-    def _map(func, seq, dist='b', targets='all'):
-        """A parallelized version of Python's builtin map."""
-    
-    def map(func, *sequences):
-        """Do a basic map with default for dist and targets."""
-    
-    def mapper(dist='b', targets='all'):
-        """Create a mapper with dist and targets."""
-    
-    def parallel(dist='b', targets='all'):
-        """A decorator that build a parallel function."""
+    def raw_map(func, seqs, dist='b', targets='all'):
+        """
+        A parallelized version of Python's builtin `map` function.
+        
+        This has a slightly different syntax than the builtin `map`.
+        This is needed because we need to have keyword arguments and thus
+        can't use *args to capture all the sequences.  Instead, they must
+        be passed in a list or tuple.
+        
+        The equivalence is:
+        
+        raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
+        
+        Most users will want to use parallel functions or the `mapper`
+        and `map` methods for an API that follows that of the builtin
+        `map`.
+        """
 
 
 class ISynchronousMultiEngineCoordinator(IMultiEngineCoordinator):
     """Methods that work on multiple engines explicitly."""
 
     def scatter(key, seq, dist='b', flatten=False, targets='all', block=True):
         """Partition and distribute a sequence to targets."""
         
     def gather(key, dist='b', targets='all', block=True):
         """Gather object key from targets"""
     
-    def _map(func, sequences, dist='b', targets='all', block=True):
-        """Perform an actual map."""
-    
-    def map(func, *sequences):
-        """Do a basic map with default for dist and targets."""
-    
-    def mapper(dist='b', targets='all', block=True):
-        """Create a mapper with dist, targets and block."""
-    
-    def parallel(dist='b', targets='all', block=True):
-        """A decorator that build a parallel function."""
+    def raw_map(func, seqs, dist='b', targets='all', block=True):
+        """
+        A parallelized version of Python's builtin map.
+        
+        This has a slightly different syntax than the builtin `map`.
+        This is needed because we need to have keyword arguments and thus
+        can't use *args to capture all the sequences.  Instead, they must
+        be passed in a list or tuple.
+        
+        raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
+        
+        Most users will want to use parallel functions or the `mapper`
+        and `map` methods for an API that follows that of the builtin
+        `map`.
+        """
 
 
 #-------------------------------------------------------------------------------
 # IMultiEngineExtras
 #-------------------------------------------------------------------------------
 
 class IMultiEngineExtras(Interface):
     
     def zip_pull(targets, keys):
         """
         Pull, but return results in a different format from `pull`.
         
         This method basically returns zip(pull(targets, *keys)), with a few 
         edge cases handled differently.  Users of chainsaw will find this format 
         familiar.
         """
     
     def run(targets, fname):
         """Run a .py file on targets."""
 
 
 class ISynchronousMultiEngineExtras(IMultiEngineExtras):
     def zip_pull(targets, keys, block=True):
         """
         Pull, but return results in a different format from `pull`.
         
         This method basically returns zip(pull(targets, *keys)), with a few 
         edge cases handled differently.  Users of chainsaw will find this format 
         familiar.
         """
     
     def run(targets, fname, block=True):
         """Run a .py file on targets."""
 
 #-------------------------------------------------------------------------------
 # The full MultiEngine interface
 #-------------------------------------------------------------------------------
 
 class IFullMultiEngine(IMultiEngine, 
     IMultiEngineCoordinator, 
     IMultiEngineExtras):
     pass
 
 
 class IFullSynchronousMultiEngine(ISynchronousMultiEngine, 
     ISynchronousMultiEngineCoordinator,
     ISynchronousMultiEngineExtras):
     pass
 

IPython/kernel/multiengineclient.py

@@ -1,852 +1,896 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_multiengineclient -*-
 
 """General Classes for IMultiEngine clients."""
 
 __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 sys
 import cPickle as pickle
 from types import FunctionType
 import linecache
 
 from twisted.internet import reactor
 from twisted.python import components, log
 from twisted.python.failure import Failure
 from zope.interface import Interface, implements, Attribute
 
 from IPython.ColorANSI import TermColors
 
 from IPython.kernel.twistedutil import blockingCallFromThread
 from IPython.kernel import error
 from IPython.kernel.parallelfunction import ParallelFunction
-from IPython.kernel.mapper import Mapper
+from IPython.kernel.mapper import (
+    MultiEngineMapper, 
+    IMultiEngineMapperFactory,
+    IMapper
+)
 from IPython.kernel import map as Map
 from IPython.kernel import multiengine as me
 from IPython.kernel.multiengine import (IFullMultiEngine,
     IFullSynchronousMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # Pending Result things
 #-------------------------------------------------------------------------------
 
 class IPendingResult(Interface):
     """A representation of a result that is pending.
     
     This class is similar to Twisted's `Deferred` object, but is designed to be
     used in a synchronous context.
     """
     
     result_id=Attribute("ID of the deferred on the other side")
     client=Attribute("A client that I came from")
     r=Attribute("An attribute that is a property that calls and returns get_result")
     
     def get_result(default=None, block=True):
         """
         Get a result that is pending.
                 
         :Parameters:
             default
                 The value to return if the result is not ready.
             block : boolean
                 Should I block for the result.
                 
         :Returns: The actual result or the default value.
         """
         
     def add_callback(f, *args, **kwargs):
         """
         Add a callback that is called with the result.
         
         If the original result is foo, adding a callback will cause
         f(foo, *args, **kwargs) to be returned instead.  If multiple
         callbacks are registered, they are chained together: the result of
         one is passed to the next and so on.  
         
         Unlike Twisted's Deferred object, there is no errback chain.  Thus
         any exception raised will not be caught and handled.  User must 
         catch these by hand when calling `get_result`.
         """
 
 
 class PendingResult(object):
     """A representation of a result that is not yet ready.
     
     A user should not create a `PendingResult` instance by hand.
     
     Methods
     =======
     
     * `get_result`
     * `add_callback`
     
     Properties
     ==========
     * `r`
     """
     
     def __init__(self, client, result_id):
         """Create a PendingResult with a result_id and a client instance.
         
         The client should implement `_getPendingResult(result_id, block)`.
         """
         self.client = client
         self.result_id = result_id
         self.called = False
         self.raised = False
         self.callbacks = []
         
     def get_result(self, default=None, block=True):
         """Get a result that is pending.
                 
         This method will connect to an IMultiEngine adapted controller
         and see if the result is ready.  If the action triggers an exception
         raise it and record it.  This method records the result/exception once it is 
         retrieved.  Calling `get_result` again will get this cached result or will
         re-raise the exception.  The .r attribute is a property that calls
         `get_result` with block=True.
         
         :Parameters:
             default
                 The value to return if the result is not ready.
             block : boolean
                 Should I block for the result.
                 
         :Returns: The actual result or the default value.
         """
         
         if self.called:
             if self.raised:
                 raise self.result[0], self.result[1], self.result[2]
             else:
                 return self.result
         try:
             result = self.client.get_pending_deferred(self.result_id, block)
         except error.ResultNotCompleted:
             return default
         except:
             # Reraise other error, but first record them so they can be reraised
             # later if .r or get_result is called again.
             self.result = sys.exc_info()
             self.called = True
             self.raised = True
             raise
         else:
             for cb in self.callbacks:
                 result = cb[0](result, *cb[1], **cb[2])
             self.result = result
             self.called = True
             return result
         
     def add_callback(self, f, *args, **kwargs):
         """Add a callback that is called with the result.
         
         If the original result is result, adding a callback will cause
         f(result, *args, **kwargs) to be returned instead.  If multiple
         callbacks are registered, they are chained together: the result of
         one is passed to the next and so on.  
         
         Unlike Twisted's Deferred object, there is no errback chain.  Thus
         any exception raised will not be caught and handled.  User must 
         catch these by hand when calling `get_result`.
         """
         assert callable(f)
         self.callbacks.append((f, args, kwargs))
         
     def __cmp__(self, other):
         if self.result_id < other.result_id:
             return -1
         else:
             return 1
             
     def _get_r(self):
         return self.get_result(block=True)
     
     r = property(_get_r)
     """This property is a shortcut to a `get_result(block=True)`."""
 
 
 #-------------------------------------------------------------------------------
 # Pretty printing wrappers for certain lists
 #-------------------------------------------------------------------------------    
     
 class ResultList(list):
     """A subclass of list that pretty prints the output of `execute`/`get_result`."""
     
     def __repr__(self):
         output = []
-        blue = TermColors.Blue
-        normal = TermColors.Normal
-        red = TermColors.Red
-        green = TermColors.Green
+        # These colored prompts were not working on Windows
+        if sys.platform == 'win32':
+            blue = normal = red = green = ''
+        else:
+            blue = TermColors.Blue
+            normal = TermColors.Normal
+            red = TermColors.Red
+            green = TermColors.Green
         output.append("<Results List>\n")
         for cmd in self:
             if isinstance(cmd, Failure):
                 output.append(cmd)
             else:
                 target = cmd.get('id',None)
                 cmd_num = cmd.get('number',None)
                 cmd_stdin = cmd.get('input',{}).get('translated','No Input')
                 cmd_stdout = cmd.get('stdout', None)
                 cmd_stderr = cmd.get('stderr', None)
                 output.append("%s[%i]%s In [%i]:%s %s\n" % \
                     (green, target,
                     blue, cmd_num, normal, cmd_stdin))
                 if cmd_stdout:
                     output.append("%s[%i]%s Out[%i]:%s %s\n" % \
                         (green, target,
                         red, cmd_num, normal, cmd_stdout))
                 if cmd_stderr:
                     output.append("%s[%i]%s Err[%i]:\n%s %s" % \
                         (green, target,
                         red, cmd_num, normal, cmd_stderr))
         return ''.join(output)
 
 
 def wrapResultList(result):
     """A function that wraps the output of `execute`/`get_result` -> `ResultList`."""
     if len(result) == 0:
         result = [result]
     return ResultList(result)
 
 
 class QueueStatusList(list):
     """A subclass of list that pretty prints the output of `queue_status`."""
     
     def __repr__(self):
         output = []
         output.append("<Queue Status List>\n")
         for e in self:
             output.append("Engine: %s\n" % repr(e[0]))
             output.append("    Pending: %s\n" % repr(e[1]['pending']))
             for q in e[1]['queue']:
                 output.append("    Command: %s\n" % repr(q))
         return ''.join(output)
 
 
 #-------------------------------------------------------------------------------
 # InteractiveMultiEngineClient
 #-------------------------------------------------------------------------------    
 
 class InteractiveMultiEngineClient(object):
     """A mixin class that add a few methods to a multiengine client.
     
     The methods in this mixin class are designed for interactive usage.
     """
                                         
     def activate(self):
         """Make this `MultiEngineClient` active for parallel magic commands.
         
         IPython has a magic command syntax to work with `MultiEngineClient` objects.
         In a given IPython session there is a single active one.  While
         there can be many `MultiEngineClient` created and used by the user, 
         there is only one active one.  The active `MultiEngineClient` is used whenever 
         the magic commands %px and %autopx are used.
         
         The activate() method is called on a given `MultiEngineClient` to make it 
         active.  Once this has been done, the magic commands can be used.
         """
         
         try:
             __IPYTHON__.activeController = self
         except NameError:
             print "The IPython Controller magics only work within IPython."
                     
     def __setitem__(self, key, value):
         """Add a dictionary interface for pushing/pulling.
         
         This functions as a shorthand for `push`.
         
         :Parameters:
             key : str 
                 What to call the remote object.
             value : object
                 The local Python object to push.
         """
         targets, block = self._findTargetsAndBlock()
         return self.push({key:value}, targets=targets, block=block)
     
     def __getitem__(self, key):
         """Add a dictionary interface for pushing/pulling.
         
         This functions as a shorthand to `pull`.
         
         :Parameters:
          - `key`: A string representing the key.
         """
         if isinstance(key, str):
             targets, block = self._findTargetsAndBlock()
             return self.pull(key, targets=targets, block=block)
         else:
             raise TypeError("__getitem__ only takes strs")
             
     def __len__(self):
         """Return the number of available engines."""
         return len(self.get_ids())
-        
-    def parallelize(self, func, targets=None, block=None):
-        """Build a `ParallelFunction` object for functionName on engines.
-        
-        The returned object will implement a parallel version of functionName
-        that takes a local sequence as its only argument and calls (in 
-        parallel) functionName on each element of that sequence.  The
-        `ParallelFunction` object has a `targets` attribute that controls
-        which engines the function is run on.
-        
-        :Parameters:
-            targets : int, list or 'all'
-                The engine ids the action will apply to.  Call `get_ids` to see
-                a list of currently available engines.
-            functionName : str
-                A Python string that names a callable defined on the engines.
-        
-        :Returns:  A `ParallelFunction` object.                
-        
-        Examples
-        ========
-        
-        >>> psin = rc.parallelize('all','lambda x:sin(x)')
-        >>> psin(range(10000))
-        [0,2,4,9,25,36,...]
-        """
-        targets, block = self._findTargetsAndBlock(targets, block)
-        return ParallelFunction(func, self, targets, block)
-
+    
     #---------------------------------------------------------------------------
     # Make this a context manager for with
     #---------------------------------------------------------------------------
     
     def findsource_file(self,f):
         linecache.checkcache()
         s = findsource(f.f_code)
         lnum = f.f_lineno
         wsource = s[0][f.f_lineno:]
         return strip_whitespace(wsource)
 
     def findsource_ipython(self,f):
         from IPython import ipapi
         self.ip = ipapi.get()
         wsource = [l+'\n' for l in
                    self.ip.IP.input_hist_raw[-1].splitlines()[1:]] 
         return strip_whitespace(wsource)
         
     def __enter__(self):
         f = sys._getframe(1)
         local_ns = f.f_locals
         global_ns = f.f_globals
         if f.f_code.co_filename == '<ipython console>':
             s = self.findsource_ipython(f)
         else:
             s = self.findsource_file(f)
 
         self._with_context_result = self.execute(s)
 
     def __exit__ (self, etype, value, tb):
         if issubclass(etype,error.StopLocalExecution):
             return True
 
 
 def remote():
     m = 'Special exception to stop local execution of parallel code.'
     raise error.StopLocalExecution(m)
 
 def strip_whitespace(source):
     # Expand tabs to avoid any confusion.
     wsource = [l.expandtabs(4) for l in source]
     # Detect the indentation level
     done = False
     for line in wsource:
         if line.isspace():
             continue
         for col,char in enumerate(line):
             if char != ' ':
                 done = True
                 break
         if done:
             break
     # Now we know how much leading space there is in the code.  Next, we
     # extract up to the first line that has less indentation.
     # WARNINGS: we skip comments that may be misindented, but we do NOT yet
     # detect triple quoted strings that may have flush left text.
     for lno,line in enumerate(wsource):
         lead = line[:col]
         if lead.isspace():
             continue
         else:
             if not lead.lstrip().startswith('#'):
                 break
     # The real 'with' source is up to lno
     src_lines = [l[col:] for l in wsource[:lno+1]]
 
     # Finally, check that the source's first non-comment line begins with the
     # special call 'remote()'
     for nline,line in enumerate(src_lines):
         if line.isspace() or line.startswith('#'):
             continue
         if 'remote()' in line:
             break
         else:
             raise ValueError('remote() call missing at the start of code')
     src = ''.join(src_lines[nline+1:])
     #print 'SRC:\n<<<<<<<>>>>>>>\n%s<<<<<>>>>>>' % src  # dbg
     return src
 
 
 #-------------------------------------------------------------------------------
 # The top-level MultiEngine client adaptor
 #-------------------------------------------------------------------------------
 
 
 class IFullBlockingMultiEngineClient(Interface):
     pass
 
 
 class FullBlockingMultiEngineClient(InteractiveMultiEngineClient):
     """
     A blocking client to the `IMultiEngine` controller interface.
     
     This class allows users to use a set of engines for a parallel
     computation through the `IMultiEngine` interface.  In this interface,
     each engine has a specific id (an int) that is used to refer to the
     engine, run code on it, etc.
     """
     
-    implements(IFullBlockingMultiEngineClient)
+    implements(
+        IFullBlockingMultiEngineClient,
+        IMultiEngineMapperFactory,
+        IMapper
+    )
     
     def __init__(self, smultiengine):
         self.smultiengine = smultiengine
         self.block = True
         self.targets = 'all'
     
     def _findBlock(self, block=None):
         if block is None:
             return self.block
         else:
             if block in (True, False):
                 return block
             else:
                 raise ValueError("block must be True or False")
     
     def _findTargets(self, targets=None):
         if targets is None:
             return self.targets
         else:
             if not isinstance(targets, (str,list,tuple,int)):
                 raise ValueError("targets must be a str, list, tuple or int")
             return targets
     
     def _findTargetsAndBlock(self, targets=None, block=None):
         return self._findTargets(targets), self._findBlock(block) 
     
     def _blockFromThread(self, function, *args, **kwargs):
         block = kwargs.get('block', None)
         if block is None:
             raise error.MissingBlockArgument("'block' keyword argument is missing")
         result = blockingCallFromThread(function, *args, **kwargs)
         if not block:
             result = PendingResult(self, result)
         return result
     
     def get_pending_deferred(self, deferredID, block):
         return blockingCallFromThread(self.smultiengine.get_pending_deferred, deferredID, block)
     
     def barrier(self, pendingResults):
         """Synchronize a set of `PendingResults`.
         
         This method is a synchronization primitive that waits for a set of
         `PendingResult` objects to complete.  More specifically, barier does
         the following.
         
         * The `PendingResult`s are sorted by result_id.
         * The `get_result` method is called for each `PendingResult` sequentially
           with block=True.
         * If a `PendingResult` gets a result that is an exception, it is 
           trapped and can be re-raised later by calling `get_result` again.
         * The `PendingResult`s are flushed from the controller.
                 
         After barrier has been called on a `PendingResult`, its results can 
         be retrieved by calling `get_result` again or accesing the `r` attribute
         of the instance.
         """
         
         # Convert to list for sorting and check class type 
         prList = list(pendingResults)
         for pr in prList:
             if not isinstance(pr, PendingResult):
                 raise error.NotAPendingResult("Objects passed to barrier must be PendingResult instances")
                             
         # Sort the PendingResults so they are in order
         prList.sort()
         # Block on each PendingResult object
         for pr in prList:
             try:
                 result = pr.get_result(block=True)
             except Exception:
                 pass
     
     def flush(self):
         """
         Clear all pending deferreds/results from the controller.
         
         For each `PendingResult` that is created by this client, the controller
         holds on to the result for that `PendingResult`.  This can be a problem
         if there are a large number of `PendingResult` objects that are created.
         
         Once the result of the `PendingResult` has been retrieved, the result
         is removed from the controller, but if a user doesn't get a result (
         they just ignore the `PendingResult`) the result is kept forever on the
         controller.  This method allows the user to clear out all un-retrieved
         results on the controller. 
         """
         r = blockingCallFromThread(self.smultiengine.clear_pending_deferreds)
         return r
     
     clear_pending_results = flush
     
     #---------------------------------------------------------------------------
     # IEngineMultiplexer related methods
     #---------------------------------------------------------------------------
     
     def execute(self, lines, targets=None, block=None):
         """
         Execute code on a set of engines.
         
         :Parameters:
             lines : str
                 The Python code to execute as a string
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         result = blockingCallFromThread(self.smultiengine.execute, lines,
             targets=targets, block=block)
         if block:
             result = ResultList(result)
         else:
             result = PendingResult(self, result)
             result.add_callback(wrapResultList)
         return result
     
     def push(self, namespace, targets=None, block=None):
         """
         Push a dictionary of keys and values to engines namespace.
         
         Each engine has a persistent namespace.  This method is used to push
         Python objects into that namespace.
         
         The objects in the namespace must be pickleable.
         
         :Parameters:
             namespace : dict
                 A dict that contains Python objects to be injected into
                 the engine persistent namespace.
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push, namespace,
             targets=targets, block=block)
     
     def pull(self, keys, targets=None, block=None):
         """
         Pull Python objects by key out of engines namespaces.
         
         :Parameters:
             keys : str or list of str
                 The names of the variables to be pulled
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull, keys, targets=targets, block=block)
     
     def push_function(self, namespace, targets=None, block=None):
         """
         Push a Python function to an engine.
         
         This method is used to push a Python function to an engine.  This
         method can then be used in code on the engines.  Closures are not supported.
         
         :Parameters:
             namespace : dict
                 A dict whose values are the functions to be pushed.  The keys give
                 that names that the function will appear as in the engines
                 namespace.
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push_function, namespace, targets=targets, block=block)
     
     def pull_function(self, keys, targets=None, block=None):
         """
         Pull a Python function from an engine.
         
         This method is used to pull a Python function from an engine.
         Closures are not supported.
         
         :Parameters:
             keys : str or list of str
                 The names of the functions to be pulled
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull_function, keys, targets=targets, block=block)
     
     def push_serialized(self, namespace, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push_serialized, namespace, targets=targets, block=block)
     
     def pull_serialized(self, keys, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull_serialized, keys, targets=targets, block=block)
     
     def get_result(self, i=None, targets=None, block=None):
         """
         Get a previous result.
         
         When code is executed in an engine, a dict is created and returned.  This
         method retrieves that dict for previous commands.
         
         :Parameters:
             i : int
                 The number of the result to get
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         result = blockingCallFromThread(self.smultiengine.get_result, i, targets=targets, block=block)
         if block:
             result = ResultList(result)
         else:
             result = PendingResult(self, result)
             result.add_callback(wrapResultList)
         return result
     
     def reset(self, targets=None, block=None):
         """
         Reset an engine.
         
         This method clears out the namespace of an engine.
 
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.reset, targets=targets, block=block)
     
     def keys(self, targets=None, block=None):
         """
         Get a list of all the variables in an engine's namespace.
 
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.keys, targets=targets, block=block)
     
     def kill(self, controller=False, targets=None, block=None):
         """
         Kill the engines and controller.
         
         This method is used to stop the engine and controller by calling
         `reactor.stop`.
         
         :Parameters:
             controller : boolean
                 If True, kill the engines and controller.  If False, just the 
                 engines
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.kill, controller, targets=targets, block=block)
     
     def clear_queue(self, targets=None, block=None):
         """
         Clear out the controller's queue for an engine.
         
         The controller maintains a queue for each engine.  This clear it out.
         
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.clear_queue, targets=targets, block=block)
     
     def queue_status(self, targets=None, block=None):
         """
         Get the status of an engines queue.
         
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.queue_status, targets=targets, block=block)
     
     def set_properties(self, properties, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.set_properties, properties, targets=targets, block=block)
     
     def get_properties(self, keys=None, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.get_properties, keys, targets=targets, block=block)
     
     def has_properties(self, keys, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.has_properties, keys, targets=targets, block=block)
     
     def del_properties(self, keys, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.del_properties, keys, targets=targets, block=block)
     
     def clear_properties(self, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.clear_properties, targets=targets, block=block)
     
     #---------------------------------------------------------------------------
     # IMultiEngine related methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         """
         Returns the ids of currently registered engines.
         """
         result = blockingCallFromThread(self.smultiengine.get_ids)
         return result
         
     #---------------------------------------------------------------------------
     # IMultiEngineCoordinator
     #---------------------------------------------------------------------------
              
     def scatter(self, key, seq, dist='b', flatten=False, targets=None, block=None):
         """
         Partition a Python sequence and send the partitions to a set of engines.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.scatter, key, seq, 
             dist, flatten, targets=targets, block=block)
     
     def gather(self, key, dist='b', targets=None, block=None):
         """
         Gather a partitioned sequence on a set of engines as a single local seq.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.gather, key, dist, 
             targets=targets, block=block)
     
-    def _map(self, func, seq, dist='b', targets=None, block=None):
+    def raw_map(self, func, seq, dist='b', targets=None, block=None):
         """
-        A parallelized version of Python's builtin map
+        A parallelized version of Python's builtin map.
+        
+        This has a slightly different syntax than the builtin `map`.
+        This is needed because we need to have keyword arguments and thus
+        can't use *args to capture all the sequences.  Instead, they must
+        be passed in a list or tuple.
+        
+        raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
+        
+        Most users will want to use parallel functions or the `mapper`
+        and `map` methods for an API that follows that of the builtin
+        `map`.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
-        return self._blockFromThread(self.smultiengine._map, func, seq, 
+        return self._blockFromThread(self.smultiengine.raw_map, func, seq, 
             dist, targets=targets, block=block)
     
     def map(self, func, *sequences):
-        return self.mapper()(func, *sequences)
+        """
+        A parallel version of Python's builtin `map` function.
+        
+        This method applies a function to sequences of arguments.  It 
+        follows the same syntax as the builtin `map`.
+        
+        This method creates a mapper objects by calling `self.mapper` with
+        no arguments and then uses that mapper to do the mapping.  See
+        the documentation of `mapper` for more details.
+        """
+        return self.mapper().map(func, *sequences)
     
     def mapper(self, dist='b', targets='all', block=None):
-        return Mapper(self, dist, targets, block)
+        """
+        Create a mapper object that has a `map` method.
+        
+        This method returns an object that implements the `IMapper` 
+        interface.  This method is a factory that is used to control how 
+        the map happens.
+        
+        :Parameters:
+            dist : str
+                What decomposition to use, 'b' is the only one supported
+                currently
+            targets : str, int, sequence of ints
+                Which engines to use for the map
+            block : boolean
+                Should calls to `map` block or not
+        """
+        return MultiEngineMapper(self, dist, targets, block)
     
     def parallel(self, dist='b', targets=None, block=None):
+        """
+        A decorator that turns a function into a parallel function.
+        
+        This can be used as:
+        
+        @parallel()
+        def f(x, y)
+            ...
+        
+        f(range(10), range(10))
+        
+        This causes f(0,0), f(1,1), ... to be called in parallel.
+        
+        :Parameters:
+            dist : str
+                What decomposition to use, 'b' is the only one supported
+                currently
+            targets : str, int, sequence of ints
+                Which engines to use for the map
+            block : boolean
+                Should calls to `map` block or not
+        """
         targets, block = self._findTargetsAndBlock(targets, block)       
-        pf = ParallelFunction(self, dist=dist, targets=targets, block=block)
+        mapper = self.mapper(dist, targets, block)
+        pf = ParallelFunction(mapper)
         return pf
     
     #---------------------------------------------------------------------------
     # IMultiEngineExtras
     #---------------------------------------------------------------------------
     
     def zip_pull(self, keys, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.zip_pull, keys, 
             targets=targets, block=block)
     
     def run(self, filename, targets=None, block=None):
         """
         Run a Python code in a file on the engines.
         
         :Parameters:
             filename : str
                 The name of the local file to run
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.run, filename,
             targets=targets, block=block)
 
 
 
 components.registerAdapter(FullBlockingMultiEngineClient,
             IFullSynchronousMultiEngine, IFullBlockingMultiEngineClient)
 
 
 
 

IPython/kernel/multienginefc.py

@@ -1,688 +1,757 b''
 # encoding: utf-8
 
 """
 Expose the multiengine controller over the Foolscap network protocol.
 """
 
 __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 cPickle as pickle
 from types import FunctionType
 
 from zope.interface import Interface, implements
 from twisted.internet import defer
 from twisted.python import components, failure, log
 
 from foolscap import Referenceable
 
 from IPython.kernel import error 
 from IPython.kernel.util import printer
 from IPython.kernel import map as Map
 from IPython.kernel.parallelfunction import ParallelFunction
-from IPython.kernel.mapper import Mapper
+from IPython.kernel.mapper import (
+    MultiEngineMapper, 
+    IMultiEngineMapperFactory,
+    IMapper
+)
 from IPython.kernel.twistedutil import gatherBoth
 from IPython.kernel.multiengine import (MultiEngine,
     IMultiEngine,
     IFullSynchronousMultiEngine,
     ISynchronousMultiEngine)
 from IPython.kernel.multiengineclient import wrapResultList
 from IPython.kernel.pendingdeferred import PendingDeferredManager
 from IPython.kernel.pickleutil import (can, canDict,
     canSequence, uncan, uncanDict, uncanSequence)
 
 from IPython.kernel.clientinterfaces import (
     IFCClientInterfaceProvider, 
     IBlockingClientAdaptor
 )
 
 # Needed to access the true globals from __main__.__dict__ 
 import __main__
 
 #-------------------------------------------------------------------------------
 # The Controller side of things
 #-------------------------------------------------------------------------------
 
 def packageResult(wrappedMethod):
     
     def wrappedPackageResult(self, *args, **kwargs):
         d = wrappedMethod(self, *args, **kwargs)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     return wrappedPackageResult
 
 
 class IFCSynchronousMultiEngine(Interface):
     """Foolscap interface to `ISynchronousMultiEngine`.  
     
     The methods in this interface are similar to those of 
     `ISynchronousMultiEngine`, but their arguments and return values are pickled
     if they are not already simple Python types that can be send over XML-RPC.
     
     See the documentation of `ISynchronousMultiEngine` and `IMultiEngine` for 
     documentation about the methods.
     
     Most methods in this interface act like the `ISynchronousMultiEngine`
     versions and can be called in blocking or non-blocking mode.
     """
     pass
 
 
 class FCSynchronousMultiEngineFromMultiEngine(Referenceable):
     """Adapt `IMultiEngine` -> `ISynchronousMultiEngine` -> `IFCSynchronousMultiEngine`.
     """
     
     implements(IFCSynchronousMultiEngine, IFCClientInterfaceProvider)
     
     addSlash = True
     
     def __init__(self, multiengine):
         # Adapt the raw multiengine to `ISynchronousMultiEngine` before saving
         # it.  This allow this class to do two adaptation steps.
         self.smultiengine = ISynchronousMultiEngine(multiengine)
         self._deferredIDCallbacks = {}
     
     #---------------------------------------------------------------------------
     # Non interface methods
     #---------------------------------------------------------------------------
     
     def packageFailure(self, f):
         f.cleanFailure()
         return self.packageSuccess(f)
     
     def packageSuccess(self, obj):
         serial = pickle.dumps(obj, 2)
         return serial
     
     #---------------------------------------------------------------------------
     # Things related to PendingDeferredManager
     #---------------------------------------------------------------------------
     
     @packageResult
     def remote_get_pending_deferred(self, deferredID, block):
         d = self.smultiengine.get_pending_deferred(deferredID, block)
         try:
             callback = self._deferredIDCallbacks.pop(deferredID)
         except KeyError:
             callback = None
         if callback is not None:
             d.addCallback(callback[0], *callback[1], **callback[2])
         return d
        
     @packageResult
     def remote_clear_pending_deferreds(self):
         return defer.maybeDeferred(self.smultiengine.clear_pending_deferreds)
     
     def _addDeferredIDCallback(self, did, callback, *args, **kwargs):
         self._deferredIDCallbacks[did] = (callback, args, kwargs)
         return did
         
     #---------------------------------------------------------------------------
     # IEngineMultiplexer related methods
     #---------------------------------------------------------------------------
     
     @packageResult
     def remote_execute(self, lines, targets, block):
         return self.smultiengine.execute(lines, targets=targets, block=block)
     
     @packageResult    
     def remote_push(self, binaryNS, targets, block):
         try:
             namespace = pickle.loads(binaryNS)
         except:
             d = defer.fail(failure.Failure())
         else:
             d = self.smultiengine.push(namespace, targets=targets, block=block)
         return d
     
     @packageResult
     def remote_pull(self, keys, targets, block):
         d = self.smultiengine.pull(keys, targets=targets, block=block)
         return d
     
     @packageResult    
     def remote_push_function(self, binaryNS, targets, block):
         try:
             namespace = pickle.loads(binaryNS)
         except:
             d = defer.fail(failure.Failure())
         else:
             namespace = uncanDict(namespace)
             d = self.smultiengine.push_function(namespace, targets=targets, block=block)
         return d
     
     def _canMultipleKeys(self, result):
         return [canSequence(r) for r in result]
     
     @packageResult
     def remote_pull_function(self, keys, targets, block):
         def can_functions(r, keys):
             if len(keys)==1 or isinstance(keys, str):
                 result = canSequence(r)
             elif len(keys)>1:
                 result = [canSequence(s) for s in r]
             return result
         d = self.smultiengine.pull_function(keys, targets=targets, block=block)
         if block:
             d.addCallback(can_functions, keys)
         else:
             d.addCallback(lambda did: self._addDeferredIDCallback(did, can_functions, keys))
         return d
     
     @packageResult    
     def remote_push_serialized(self, binaryNS, targets, block):
         try:
             namespace = pickle.loads(binaryNS)
         except:
             d = defer.fail(failure.Failure())
         else:
             d = self.smultiengine.push_serialized(namespace, targets=targets, block=block)
         return d
     
     @packageResult
     def remote_pull_serialized(self, keys, targets, block):
         d = self.smultiengine.pull_serialized(keys, targets=targets, block=block)
         return d
     
     @packageResult
     def remote_get_result(self, i, targets, block):
         if i == 'None':
             i = None
         return self.smultiengine.get_result(i, targets=targets, block=block)
     
     @packageResult
     def remote_reset(self, targets, block):
         return self.smultiengine.reset(targets=targets, block=block)
     
     @packageResult
     def remote_keys(self, targets, block):
         return self.smultiengine.keys(targets=targets, block=block)
     
     @packageResult
     def remote_kill(self, controller, targets, block):
         return self.smultiengine.kill(controller, targets=targets, block=block)
     
     @packageResult
     def remote_clear_queue(self, targets, block):
         return self.smultiengine.clear_queue(targets=targets, block=block)
     
     @packageResult
     def remote_queue_status(self, targets, block):
         return self.smultiengine.queue_status(targets=targets, block=block)
     
     @packageResult
     def remote_set_properties(self, binaryNS, targets, block):
         try:
             ns = pickle.loads(binaryNS)
         except:
             d = defer.fail(failure.Failure())
         else:
             d = self.smultiengine.set_properties(ns, targets=targets, block=block)
         return d
     
     @packageResult
     def remote_get_properties(self, keys, targets, block):
         if keys=='None':
             keys=None
         return self.smultiengine.get_properties(keys, targets=targets, block=block)
     
     @packageResult
     def remote_has_properties(self, keys, targets, block):
         return self.smultiengine.has_properties(keys, targets=targets, block=block)
     
     @packageResult
     def remote_del_properties(self, keys, targets, block):
         return self.smultiengine.del_properties(keys, targets=targets, block=block)
     
     @packageResult
     def remote_clear_properties(self, targets, block):
         return self.smultiengine.clear_properties(targets=targets, block=block)
     
     #---------------------------------------------------------------------------
     # IMultiEngine related methods
     #---------------------------------------------------------------------------
     
     def remote_get_ids(self):
         """Get the ids of the registered engines.
         
         This method always blocks.
         """
         return self.smultiengine.get_ids()
     
     #---------------------------------------------------------------------------
     # IFCClientInterfaceProvider related methods
     #---------------------------------------------------------------------------
     
     def remote_get_client_name(self):
         return 'IPython.kernel.multienginefc.FCFullSynchronousMultiEngineClient'
 
 
 # The __init__ method of `FCMultiEngineFromMultiEngine` first adapts the
 # `IMultiEngine` to `ISynchronousMultiEngine` so this is actually doing a
 # two phase adaptation.
 components.registerAdapter(FCSynchronousMultiEngineFromMultiEngine,
             IMultiEngine, IFCSynchronousMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # The Client side of things
 #-------------------------------------------------------------------------------
 
 
 class FCFullSynchronousMultiEngineClient(object):
     
-    implements(IFullSynchronousMultiEngine, IBlockingClientAdaptor)
+    implements(
+        IFullSynchronousMultiEngine, 
+        IBlockingClientAdaptor,
+        IMultiEngineMapperFactory,
+        IMapper
+    )
     
     def __init__(self, remote_reference):
         self.remote_reference = remote_reference
         self._deferredIDCallbacks = {}
         # This class manages some pending deferreds through this instance.  This
         # is required for methods like gather/scatter as it enables us to
         # create our own pending deferreds for composite operations.
         self.pdm = PendingDeferredManager()
     
     #---------------------------------------------------------------------------
     # Non interface methods
     #---------------------------------------------------------------------------
                  
     def unpackage(self, r):
         return pickle.loads(r)
     
     #---------------------------------------------------------------------------
     # Things related to PendingDeferredManager
     #---------------------------------------------------------------------------
     
     def get_pending_deferred(self, deferredID, block=True):
         
         # Because we are managing some pending deferreds locally (through
         # self.pdm) and some remotely (on the controller), we first try the 
         # local one and then the remote one.
         if self.pdm.quick_has_id(deferredID):
             d = self.pdm.get_pending_deferred(deferredID, block)
             return d
         else:
             d = self.remote_reference.callRemote('get_pending_deferred', deferredID, block)
             d.addCallback(self.unpackage)
             try:
                 callback = self._deferredIDCallbacks.pop(deferredID)
             except KeyError:
                 callback = None
             if callback is not None:
                 d.addCallback(callback[0], *callback[1], **callback[2])
             return d
     
     def clear_pending_deferreds(self):
         
         # This clear both the local (self.pdm) and remote pending deferreds
         self.pdm.clear_pending_deferreds()
         d2 = self.remote_reference.callRemote('clear_pending_deferreds')
         d2.addCallback(self.unpackage)
         return d2
     
     def _addDeferredIDCallback(self, did, callback, *args, **kwargs):
         self._deferredIDCallbacks[did] = (callback, args, kwargs)
         return did
        
     #---------------------------------------------------------------------------
     # IEngineMultiplexer related methods
     #---------------------------------------------------------------------------
         
     def execute(self, lines, targets='all', block=True):
         d = self.remote_reference.callRemote('execute', lines, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def push(self, namespace, targets='all', block=True):
         serial = pickle.dumps(namespace, 2)
         d =  self.remote_reference.callRemote('push', serial, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def pull(self, keys, targets='all', block=True):
         d = self.remote_reference.callRemote('pull', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def push_function(self, namespace, targets='all', block=True):
         cannedNamespace = canDict(namespace)
         serial = pickle.dumps(cannedNamespace, 2)
         d = self.remote_reference.callRemote('push_function', serial, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def pull_function(self, keys, targets='all', block=True):
         def uncan_functions(r, keys):
             if len(keys)==1 or isinstance(keys, str):
                 return uncanSequence(r)
             elif len(keys)>1:
                 return [uncanSequence(s) for s in r]
         d = self.remote_reference.callRemote('pull_function', keys, targets, block)
         if block:
             d.addCallback(self.unpackage)
             d.addCallback(uncan_functions, keys)
         else:
             d.addCallback(self.unpackage)
             d.addCallback(lambda did: self._addDeferredIDCallback(did, uncan_functions, keys))
         return d
     
     def push_serialized(self, namespace, targets='all', block=True):
         cannedNamespace = canDict(namespace)
         serial = pickle.dumps(cannedNamespace, 2)
         d =  self.remote_reference.callRemote('push_serialized', serial, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def pull_serialized(self, keys, targets='all', block=True):
         d = self.remote_reference.callRemote('pull_serialized', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
         
     def get_result(self, i=None, targets='all', block=True):
         if i is None: # This is because None cannot be marshalled by xml-rpc
             i = 'None'
         d = self.remote_reference.callRemote('get_result', i, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def reset(self, targets='all', block=True):
         d = self.remote_reference.callRemote('reset', targets, block)
         d.addCallback(self.unpackage)
         return d        
     
     def keys(self, targets='all', block=True):
         d = self.remote_reference.callRemote('keys', targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def kill(self, controller=False, targets='all', block=True):
         d = self.remote_reference.callRemote('kill', controller, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def clear_queue(self, targets='all', block=True):
         d = self.remote_reference.callRemote('clear_queue', targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def queue_status(self, targets='all', block=True):
         d = self.remote_reference.callRemote('queue_status', targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def set_properties(self, properties, targets='all', block=True):
         serial = pickle.dumps(properties, 2)
         d = self.remote_reference.callRemote('set_properties', serial, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def get_properties(self, keys=None, targets='all', block=True):
         if keys==None:
             keys='None'
         d = self.remote_reference.callRemote('get_properties', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def has_properties(self, keys, targets='all', block=True):
         d = self.remote_reference.callRemote('has_properties', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def del_properties(self, keys, targets='all', block=True):
         d = self.remote_reference.callRemote('del_properties', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def clear_properties(self, targets='all', block=True):
         d = self.remote_reference.callRemote('clear_properties', targets, block)
         d.addCallback(self.unpackage)
         return d
     
     #---------------------------------------------------------------------------
     # IMultiEngine related methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         d = self.remote_reference.callRemote('get_ids')
         return d
     
     #---------------------------------------------------------------------------
     # ISynchronousMultiEngineCoordinator related methods
     #---------------------------------------------------------------------------
 
     def _process_targets(self, targets):
         def create_targets(ids):
             if isinstance(targets, int):
                 engines = [targets]
             elif targets=='all':
                 engines = ids
             elif isinstance(targets, (list, tuple)):
                 engines = targets
             for t in engines:
                 if not t in ids:
                     raise error.InvalidEngineID("engine with id %r does not exist"%t)
             return engines
         
         d = self.get_ids()
         d.addCallback(create_targets)
         return d
     
     def scatter(self, key, seq, dist='b', flatten=False, targets='all', block=True):
         
         # Note: scatter and gather handle pending deferreds locally through self.pdm.
         # This enables us to collect a bunch fo deferred ids and make a secondary 
         # deferred id that corresponds to the entire group.  This logic is extremely
         # difficult to get right though.
         def do_scatter(engines):
             nEngines = len(engines)
             mapClass = Map.dists[dist]
             mapObject = mapClass()
             d_list = []
             # Loop through and push to each engine in non-blocking mode.
             # This returns a set of deferreds to deferred_ids
             for index, engineid in enumerate(engines):
                 partition = mapObject.getPartition(seq, index, nEngines)
                 if flatten and len(partition) == 1:
                     d = self.push({key: partition[0]}, targets=engineid, block=False)
                 else:
                     d = self.push({key: partition}, targets=engineid, block=False)
                 d_list.append(d)
             # Collect the deferred to deferred_ids
             d = gatherBoth(d_list,
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             # Now d has a list of deferred_ids or Failures coming
             d.addCallback(error.collect_exceptions, 'scatter')
             def process_did_list(did_list):
                 """Turn a list of deferred_ids into a final result or failure."""
                 new_d_list = [self.get_pending_deferred(did, True) for did in did_list]
                 final_d = gatherBoth(new_d_list,
                                      fireOnOneErrback=0,
                                      consumeErrors=1,
                                      logErrors=0)
                 final_d.addCallback(error.collect_exceptions, 'scatter')
                 final_d.addCallback(lambda lop: [i[0] for i in lop])
                 return final_d
             # Now, depending on block, we need to handle the list deferred_ids
             # coming down the pipe diferently.
             if block:
                 # If we are blocking register a callback that will transform the
                 # list of deferred_ids into the final result.
                 d.addCallback(process_did_list)
                 return d
             else:
                 # Here we are going to use a _local_ PendingDeferredManager.
                 deferred_id = self.pdm.get_deferred_id()
                 # This is the deferred we will return to the user that will fire
                 # with the local deferred_id AFTER we have received the list of 
                 # primary deferred_ids
                 d_to_return = defer.Deferred()
                 def do_it(did_list):
                     """Produce a deferred to the final result, but first fire the
                     deferred we will return to the user that has the local
                     deferred id."""
                     d_to_return.callback(deferred_id)
                     return process_did_list(did_list)
                 d.addCallback(do_it)
                 # Now save the deferred to the final result
                 self.pdm.save_pending_deferred(d, deferred_id)
                 return d_to_return
 
         d = self._process_targets(targets)
         d.addCallback(do_scatter)
         return d
 
     def gather(self, key, dist='b', targets='all', block=True):
         
         # Note: scatter and gather handle pending deferreds locally through self.pdm.
         # This enables us to collect a bunch fo deferred ids and make a secondary 
         # deferred id that corresponds to the entire group.  This logic is extremely
         # difficult to get right though.
         def do_gather(engines):
             nEngines = len(engines)
             mapClass = Map.dists[dist]
             mapObject = mapClass()
             d_list = []
             # Loop through and push to each engine in non-blocking mode.
             # This returns a set of deferreds to deferred_ids
             for index, engineid in enumerate(engines):
                 d = self.pull(key, targets=engineid, block=False)
                 d_list.append(d)
             # Collect the deferred to deferred_ids
             d = gatherBoth(d_list,
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             # Now d has a list of deferred_ids or Failures coming
             d.addCallback(error.collect_exceptions, 'scatter')
             def process_did_list(did_list):
                 """Turn a list of deferred_ids into a final result or failure."""
                 new_d_list = [self.get_pending_deferred(did, True) for did in did_list]
                 final_d = gatherBoth(new_d_list,
                                      fireOnOneErrback=0,
                                      consumeErrors=1,
                                      logErrors=0)
                 final_d.addCallback(error.collect_exceptions, 'gather')
                 final_d.addCallback(lambda lop: [i[0] for i in lop])
                 final_d.addCallback(mapObject.joinPartitions)
                 return final_d
             # Now, depending on block, we need to handle the list deferred_ids
             # coming down the pipe diferently.
             if block:
                 # If we are blocking register a callback that will transform the
                 # list of deferred_ids into the final result.
                 d.addCallback(process_did_list)
                 return d
             else:
                 # Here we are going to use a _local_ PendingDeferredManager.
                 deferred_id = self.pdm.get_deferred_id()
                 # This is the deferred we will return to the user that will fire
                 # with the local deferred_id AFTER we have received the list of 
                 # primary deferred_ids
                 d_to_return = defer.Deferred()
                 def do_it(did_list):
                     """Produce a deferred to the final result, but first fire the
                     deferred we will return to the user that has the local
                     deferred id."""
                     d_to_return.callback(deferred_id)
                     return process_did_list(did_list)
                 d.addCallback(do_it)
                 # Now save the deferred to the final result
                 self.pdm.save_pending_deferred(d, deferred_id)
                 return d_to_return
 
         d = self._process_targets(targets)
         d.addCallback(do_gather)
         return d
 
-    def _map(self, func, sequences, dist='b', targets='all', block=True):
+    def raw_map(self, func, sequences, dist='b', targets='all', block=True):
         """
-        Call a callable on elements of a sequence.
+        A parallelized version of Python's builtin map.
+        
+        This has a slightly different syntax than the builtin `map`.
+        This is needed because we need to have keyword arguments and thus
+        can't use *args to capture all the sequences.  Instead, they must
+        be passed in a list or tuple.
+        
+        raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
+        
+        Most users will want to use parallel functions or the `mapper`
+        and `map` methods for an API that follows that of the builtin
+        `map`.
         """
         if not isinstance(sequences, (list, tuple)):
             raise TypeError('sequences must be a list or tuple')
         max_len = max(len(s) for s in sequences)
         for s in sequences:
             if len(s)!=max_len:
                 raise ValueError('all sequences must have equal length')
         if isinstance(func, FunctionType):
             d = self.push_function(dict(_ipython_map_func=func), targets=targets, block=False)
             d.addCallback(lambda did: self.get_pending_deferred(did, True))
             sourceToRun = '_ipython_map_seq_result = map(_ipython_map_func, *zip(*_ipython_map_seq))'
         elif isinstance(func, str):
             d = defer.succeed(None)
             sourceToRun = \
                 '_ipython_map_seq_result = map(%s, *zip(*_ipython_map_seq))' % func
         else:
             raise TypeError("func must be a function or str")
         
         d.addCallback(lambda _: self.scatter('_ipython_map_seq', zip(*sequences), dist, targets=targets))
         d.addCallback(lambda _: self.execute(sourceToRun, targets=targets, block=False))
         d.addCallback(lambda did: self.get_pending_deferred(did, True))
         d.addCallback(lambda _: self.gather('_ipython_map_seq_result', dist, targets=targets, block=block))
         return d
 
     def map(self, func, *sequences):
-        return self.mapper()(func, *sequences)
+        """
+        A parallel version of Python's builtin `map` function.
+        
+        This method applies a function to sequences of arguments.  It 
+        follows the same syntax as the builtin `map`.
+        
+        This method creates a mapper objects by calling `self.mapper` with
+        no arguments and then uses that mapper to do the mapping.  See
+        the documentation of `mapper` for more details.
+        """
+        return self.mapper().map(func, *sequences)
     
     def mapper(self, dist='b', targets='all', block=True):
-        return Mapper(self, dist, targets, block)
+        """
+        Create a mapper object that has a `map` method.
+        
+        This method returns an object that implements the `IMapper` 
+        interface.  This method is a factory that is used to control how 
+        the map happens.
+        
+        :Parameters:
+            dist : str
+                What decomposition to use, 'b' is the only one supported
+                currently
+            targets : str, int, sequence of ints
+                Which engines to use for the map
+            block : boolean
+                Should calls to `map` block or not
+        """
+        return MultiEngineMapper(self, dist, targets, block)
 
     def parallel(self, dist='b', targets='all', block=True):
-        pf = ParallelFunction(self, dist=dist, targets=targets, block=True)
+        """
+        A decorator that turns a function into a parallel function.
+        
+        This can be used as:
+        
+        @parallel()
+        def f(x, y)
+            ...
+        
+        f(range(10), range(10))
+        
+        This causes f(0,0), f(1,1), ... to be called in parallel.
+        
+        :Parameters:
+            dist : str
+                What decomposition to use, 'b' is the only one supported
+                currently
+            targets : str, int, sequence of ints
+                Which engines to use for the map
+            block : boolean
+                Should calls to `map` block or not
+        """
+        mapper = self.mapper(dist, targets, block)
+        pf = ParallelFunction(mapper)
         return pf
     
     #---------------------------------------------------------------------------
     # ISynchronousMultiEngineExtras related methods
     #---------------------------------------------------------------------------
     
     def _transformPullResult(self, pushResult, multitargets, lenKeys):
         if not multitargets:
             result = pushResult[0]
         elif lenKeys > 1:
             result = zip(*pushResult)
         elif lenKeys is 1:
             result = list(pushResult)
         return result
     
     def zip_pull(self, keys, targets='all', block=True):
         multitargets = not isinstance(targets, int) and len(targets) > 1
         lenKeys = len(keys)
         d = self.pull(keys, targets=targets, block=block)
         if block:
             d.addCallback(self._transformPullResult, multitargets, lenKeys)
         else:
             d.addCallback(lambda did: self._addDeferredIDCallback(did, self._transformPullResult, multitargets, lenKeys))
         return d
     
     def run(self, fname, targets='all', block=True):
         fileobj = open(fname,'r')
         source = fileobj.read()
         fileobj.close()
         # if the compilation blows, we get a local error right away
         try:
             code = compile(source,fname,'exec')
         except:
             return defer.fail(failure.Failure()) 
         # Now run the code
         d = self.execute(source, targets=targets, block=block)
         return d
     
     #---------------------------------------------------------------------------
     # IBlockingClientAdaptor related methods
     #---------------------------------------------------------------------------
     
     def adapt_to_blocking_client(self):
         from IPython.kernel.multiengineclient import IFullBlockingMultiEngineClient
         return IFullBlockingMultiEngineClient(self)

IPython/kernel/parallelfunction.py

@@ -1,47 +1,107 b''
 # encoding: utf-8
 
 """A parallelized function that does scatter/execute/gather."""
 
 __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 types import FunctionType
 from zope.interface import Interface, implements
 
 
+class IMultiEngineParallelDecorator(Interface):
+    """A decorator that creates a parallel function."""
+    
+    def parallel(dist='b', targets=None, block=None):
+        """
+        A decorator that turns a function into a parallel function.
+        
+        This can be used as:
+        
+        @parallel()
+        def f(x, y)
+            ...
+        
+        f(range(10), range(10))
+        
+        This causes f(0,0), f(1,1), ... to be called in parallel.
+        
+        :Parameters:
+            dist : str
+                What decomposition to use, 'b' is the only one supported
+                currently
+            targets : str, int, sequence of ints
+                Which engines to use for the map
+            block : boolean
+                Should calls to `map` block or not
+        """
+
+class ITaskParallelDecorator(Interface):
+    """A decorator that creates a parallel function."""
+    
+    def parallel(clear_before=False, clear_after=False, retries=0, 
+        recovery_task=None, depend=None, block=True):
+        """
+        A decorator that turns a function into a parallel function.
+        
+        This can be used as:
+        
+        @parallel()
+        def f(x, y)
+            ...
+        
+        f(range(10), range(10))
+        
+        This causes f(0,0), f(1,1), ... to be called in parallel.
+        
+        See the documentation for `IPython.kernel.task.BaseTask` for 
+        documentation on the arguments to this method.
+        """
+
+class IParallelFunction(Interface):
+    pass
+
 class ParallelFunction(object):
     """
-    A decorator for building parallel functions.
+    The implementation of a parallel function.
+    
+    A parallel function is similar to Python's map function: 
+    
+    map(func, *sequences) -> pfunc(*sequences)
+    
+    Parallel functions should be created by using the @parallel decorator.
     """
     
-    def __init__(self, multiengine, dist='b', targets='all', block=True):
+    implements(IParallelFunction)
+    
+    def __init__(self, mapper):
         """
-        Create a `ParallelFunction decorator`.
+        Create a parallel function from an `IMapper`.
+        
+        :Parameters:
+            mapper : an `IMapper` implementer.
+                The mapper to use for the parallel function
         """
-        self.multiengine = multiengine
-        self.dist = dist
-        self.targets = targets
-        self.block = block
+        self.mapper = mapper
         
     def __call__(self, func):
         """
-        Decorate the function to make it run in parallel.
+        Decorate a function to make it run in parallel.
         """
         assert isinstance(func, (str, FunctionType)), "func must be a fuction or str"
         self.func = func
         def call_function(*sequences):
-            return self.multiengine._map(self.func, sequences, dist=self.dist,
-                targets=self.targets, block=self.block)
+            return self.mapper.map(self.func, *sequences)
         return call_function
 
     
  No newline at end of file

IPython/kernel/task.py

@@ -1,830 +1,1113 b''
 # 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
 
 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.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
-
-def can_task(task):
-    t = copy.copy(task)
-    t.depend = can(t.depend)
-    t.expression = can(t.expression)
-    if t.recovery_task:
-        t.recovery_task = can_task(t.recovery_task)
-    return t
+from IPython.kernel.pickleutil import can, uncan, CannedFunction
 
-def uncan_task(task):
-    t = copy.copy(task)
-    t.depend = uncan(t.depend)
-    t.expression = uncan(t.expression)
-    if t.recovery_task and t.recovery_task is not task:
-        t.recovery_task = uncan_task(t.recovery_task)
-    return t
+#-----------------------------------------------------------------------------
+# Definition of the Task objects
+#-----------------------------------------------------------------------------
 
 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
-    --------
+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.
     
-    >>> 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)
+    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.
     """
     
-    def __init__(self, expression, args=None, kwargs=None, pull=None, push=None,
-            clear_before=False, clear_after=False, retries=0, 
-            recovery_task=None, depend=None, **options):
-        self.expression = expression
+    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 isinstance(pull, str):
-            self.pull = [pull]
-        else:
+        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
-        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
+        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:
-    """The result namespace object for use in TaskResult objects as tr.ns.
+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':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.
+    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 raiseException() method of 
+    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
+    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.
+    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
+    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
+    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()
+            self.raise_exception()
         return self.results[key]
     
-    def raiseException(self):
+    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 `TaskResult` 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 `TaskResult` 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.
         """
-        if task.clear_before:
-            d = self.queuedEngine.reset()
-        else:
-            d = defer.succeed(None)
-        
-        if isinstance(task.expression, FunctionType):
-            d.addCallback(lambda r: self.queuedEngine.push_function(
-                dict(_ipython_task_function=task.expression))
-            )
-            d.addCallback(lambda r: self.queuedEngine.push(
-                dict(_ipython_task_args=task.args,_ipython_task_kwargs=task.kwargs))
-            )
-            d.addCallback(lambda r: self.queuedEngine.execute(
-                '_ipython_task_result = _ipython_task_function(*_ipython_task_args,**_ipython_task_kwargs)')
-            )
-            d.addCallback(lambda r: self.queuedEngine.pull('_ipython_task_result'))
-        elif isinstance(task.expression, str):
-            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)
-        else:
-            raise TypeError("task expression must be a str or function")
-        
-        def reseter(result):
-            self.queuedEngine.reset()
-            return result
-            
-        if task.clear_after:
-            d.addBoth(reseter)
-        
-        if isinstance(task.expression, FunctionType):
-            return d.addBoth(self._zipResults, None, time.time(), time.localtime())
-        else:
-            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)
-            if names is None:
-                tr.result = result
-            else:
-                tr.result = None
-        # the time info
-        tr.submitted = time.strftime(time_format, start_struct)
-        tr.completed = time.strftime(time_format)
-        tr.duration = time.time()-start
-        return tr
-    
+        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 : a `Task` object
+            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.
+        """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: a `Task` object
+        :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 implementing object
-            flags : General keywords for more sophisticated scheduling
+            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 a tuple of the worker and task pair for the next
-        task to be run.
-        """
+        """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.
+    """
+    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)
+                    # 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.
+    """
+    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
+    """
+    The Task based interface to a `ControllerService` object
     
     This adapts a `ControllerService` to the ITaskController interface.
     """
     
     def run(task):
-        """Run a 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.
+        """
+        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
+        :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.
+        """
+        Block until the list of taskids are completed.
         
         Returns None on success.
         """
     
     def spin():
-        """touch the scheduler, to resume scheduling without submitting
-        a task.
+        """
+        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.
+    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 "We already have one!  This should not happen."
+            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."""
+        """
+        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)
+        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)
+        """
+        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."""
+        """
+        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)
+        """
+        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
+        # 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")
+        """
+        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))
+            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)
+                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):
+    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 result.failure is not None and isinstance(result.failure, failure.Failure): # we failed
+            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, Task) and \
+                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)
+                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
+        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.  
+        """
+        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)

IPython/kernel/taskclient.py

@@ -1,161 +1,180 b''
 # 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.
+"""
+A blocking version of the task client.
 """
 
 __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
+from IPython.kernel.mapper import (
+    SynchronousTaskMapper,
+    ITaskMapperFactory,
+    IMapper
+)
+from IPython.kernel.parallelfunction import (
+    ParallelFunction, 
+    ITaskParallelDecorator
+)
 
 #-------------------------------------------------------------------------------
-# Connecting Task Client
+# The 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.
+    A vague interface of the blocking task client
     """
     pass
 
-
-class BlockingTaskClient(InteractiveTaskClient):
+class BlockingTaskClient(object):
     """
-    This class provides a blocking task client.
+    A blocking task client that adapts a non-blocking one.
     """
     
-    implements(IBlockingTaskClient)
+    implements(
+        IBlockingTaskClient, 
+        ITaskMapperFactory,
+        IMapper,
+        ITaskParallelDecorator
+    )
     
     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.
+    def run(self, task, block=False):
+        """Run a task on the `TaskController`.
+        
+        See the documentation of the `MapTask` and `StringTask` classes for 
+        details on how to build a task of different types.
         
         :Parameters:
-            task : `Task`
-                The `Task` object to run
+            task : an `ITask` implementer
+        
+        :Returns: The int taskid of the submitted task.  Pass this to 
+            `get_task_result` to get the `TaskResult` object.
         """
-        return blockingCallFromThread(self.task_controller.run, task)
+        tid = blockingCallFromThread(self.task_controller.run, task)
+        if block:
+            return self.get_task_result(tid, block=True)
+        else:
+            return tid
     
     def get_task_result(self, taskid, block=False):
         """
-        Get or poll for a task result.
+        Get a task result by taskid.
         
         :Parameters:
             taskid : int
-                The id of the task whose result to get
+                The taskid of the task to be retrieved.
             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.
+                Should I block until the task is done?
+        
+        :Returns: A `TaskResult` object that encapsulates the task result.
         """
         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.
+        Abort a task by taskid.
+        
+        :Parameters:
+            taskid : int
+                The taskid of the task to be aborted.
         """
         return blockingCallFromThread(self.task_controller.abort, taskid)
     
     def barrier(self, taskids):
-        """
-        Wait for a set of tasks to finish.
+        """Block until a set of tasks are completed.
         
         :Parameters:
-            taskids : list of ints
-                A list of task ids to wait for.
+            taskids : list, tuple
+                A sequence of taskids to block on.
         """
         return blockingCallFromThread(self.task_controller.barrier, taskids)
     
     def spin(self):
         """
-        Cause the scheduler to schedule tasks.
+        Touch the scheduler, to resume scheduling without submitting a task.
         
         This method only needs to be called in unusual situations where the
-        scheduler is idle for some reason.
+        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)
+    
+    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.
+        """
+        return blockingCallFromThread(self.task_controller.clear)
+    
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is load balanced.
+        """
+        return self.mapper().map(func, *sequences)
 
+    def mapper(self, clear_before=False, clear_after=False, retries=0, 
+                recovery_task=None, depend=None, block=True):
+        """
+        Create an `IMapper` implementer with a given set of arguments.
+        
+        The `IMapper` created using a task controller is load balanced.
+        
+        See the documentation for `IPython.kernel.task.BaseTask` for 
+        documentation on the arguments to this method.
+        """
+        return SynchronousTaskMapper(self, clear_before=clear_before, 
+            clear_after=clear_after, retries=retries, 
+            recovery_task=recovery_task, depend=depend, block=block)
+    
+    def parallel(self, clear_before=False, clear_after=False, retries=0, 
+        recovery_task=None, depend=None, block=True):
+        mapper = self.mapper(clear_before, clear_after, retries,
+            recovery_task, depend, block)
+        pf = ParallelFunction(mapper)
+        return pf
 
 components.registerAdapter(BlockingTaskClient,
             task.ITaskController, IBlockingTaskClient)
 
 

IPython/kernel/taskfc.py

@@ -1,267 +1,329 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.tests.test_taskxmlrpc -*-
 """A Foolscap interface to a TaskController.
 
 This class lets Foolscap clients talk to a TaskController.
 """
 
 __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 cPickle as pickle
 import xmlrpclib, copy
 
 from zope.interface import Interface, implements
 from twisted.internet import defer
 from twisted.python import components, failure
 
 from foolscap import Referenceable
 
 from IPython.kernel.twistedutil import blockingCallFromThread
 from IPython.kernel import error, task as taskmodule, taskclient
 from IPython.kernel.pickleutil import can, uncan
 from IPython.kernel.clientinterfaces import (
     IFCClientInterfaceProvider, 
     IBlockingClientAdaptor
 )
+from IPython.kernel.mapper import (
+    TaskMapper,
+    ITaskMapperFactory,
+    IMapper
+)
+from IPython.kernel.parallelfunction import (
+    ParallelFunction, 
+    ITaskParallelDecorator
+)
 
 #-------------------------------------------------------------------------------
 # The Controller side of things
 #-------------------------------------------------------------------------------
 
 
 class IFCTaskController(Interface):
     """Foolscap interface to task controller.
         
-    See the documentation of ITaskController for documentation about the methods.
+    See the documentation of `ITaskController` for more information.
     """
-    def remote_run(request, binTask):
+    def remote_run(binTask):
         """"""
     
-    def remote_abort(request, taskid):
+    def remote_abort(taskid):
         """"""
         
-    def remote_get_task_result(request, taskid, block=False):
+    def remote_get_task_result(taskid, block=False):
         """"""
         
-    def remote_barrier(request, taskids):
+    def remote_barrier(taskids):
+        """"""
+    
+    def remote_spin():
         """"""
     
-    def remote_spin(request):
+    def remote_queue_status(verbose):
         """"""
     
-    def remote_queue_status(request, verbose):
+    def remote_clear():
         """"""
 
 
 class FCTaskControllerFromTaskController(Referenceable):
-    """XML-RPC attachmeot for controller.
-        
-    See IXMLRPCTaskController and ITaskController (and its children) for documentation. 
     """
+    Adapt a `TaskController` to an `IFCTaskController`
+    
+    This class is used to expose a `TaskController` over the wire using
+    the Foolscap network protocol.
+    """
+    
     implements(IFCTaskController, IFCClientInterfaceProvider)
     
     def __init__(self, taskController):
         self.taskController = taskController
     
     #---------------------------------------------------------------------------
     # Non interface methods
     #---------------------------------------------------------------------------
     
     def packageFailure(self, f):
         f.cleanFailure()
         return self.packageSuccess(f)
     
     def packageSuccess(self, obj):
         serial = pickle.dumps(obj, 2)
         return serial
     
     #---------------------------------------------------------------------------
     # ITaskController related methods
     #---------------------------------------------------------------------------
     
     def remote_run(self, ptask):
         try:
-            ctask = pickle.loads(ptask)
-            task = taskmodule.uncan_task(ctask)
+            task = pickle.loads(ptask)
+            task.uncan_task()
         except:
             d = defer.fail(pickle.UnpickleableError("Could not unmarshal task"))
         else:
             d = self.taskController.run(task)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
     def remote_abort(self, taskid):
         d = self.taskController.abort(taskid)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
     def remote_get_task_result(self, taskid, block=False):
         d = self.taskController.get_task_result(taskid, block)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
     def remote_barrier(self, taskids):
         d = self.taskController.barrier(taskids)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d        
     
     def remote_spin(self):
         d = self.taskController.spin()
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d        
     
     def remote_queue_status(self, verbose):
         d = self.taskController.queue_status(verbose)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
+    def remote_clear(self):
+        return self.taskController.clear()
+    
     def remote_get_client_name(self):
         return 'IPython.kernel.taskfc.FCTaskClient'
     
 components.registerAdapter(FCTaskControllerFromTaskController,
             taskmodule.ITaskController, IFCTaskController)
 
 
 #-------------------------------------------------------------------------------
 # The Client side of things
 #-------------------------------------------------------------------------------
 
 class FCTaskClient(object):
-    """XML-RPC based TaskController client that implements ITaskController.
-        
-    :Parameters:
-        addr : (ip, port)
-            The ip (str) and port (int) tuple of the `TaskController`.  
     """
-    implements(taskmodule.ITaskController, IBlockingClientAdaptor)
+    Client class for Foolscap exposed `TaskController`.
+    
+    This class is an adapter that makes a `RemoteReference` to a 
+    `TaskController` look like an actual `ITaskController` on the client side.
+
+    This class also implements `IBlockingClientAdaptor` so that clients can 
+    automatically get a blocking version of this class.
+    """
+    
+    implements(
+        taskmodule.ITaskController, 
+        IBlockingClientAdaptor,
+        ITaskMapperFactory,
+        IMapper,
+        ITaskParallelDecorator
+    )
     
     def __init__(self, remote_reference):
         self.remote_reference = remote_reference
     
     #---------------------------------------------------------------------------
     # Non interface methods
     #---------------------------------------------------------------------------
     
     def unpackage(self, r):
         return pickle.loads(r)
     
     #---------------------------------------------------------------------------
     # ITaskController related methods
     #---------------------------------------------------------------------------
     def run(self, task):
         """Run a task on the `TaskController`.
         
-        :Parameters:
-            task : a `Task` object
-        
-        The Task object is created using the following signature:
-        
-        Task(expression, pull=None, push={}, clear_before=False, 
-            clear_after=False, retries=0, **options):)
+        See the documentation of the `MapTask` and `StringTask` classes for 
+        details on how to build a task of different types.
         
-        The meaning of the arguments is as follows:
+        :Parameters:
+            task : an `ITask` implementer
         
-        :Task 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: The int taskid of the submitted task.  Pass this to 
             `get_task_result` to get the `TaskResult` object.
         """
-        assert isinstance(task, taskmodule.Task), "task must be a Task object!"
-        ctask = taskmodule.can_task(task) # handles arbitrary function in .depend
-                                # as well as arbitrary recovery_task chains
-        ptask = pickle.dumps(ctask, 2)
+        assert isinstance(task, taskmodule.BaseTask), "task must be a Task object!"
+        task.can_task()
+        ptask = pickle.dumps(task, 2)
+        task.uncan_task()
         d = self.remote_reference.callRemote('run', ptask)
         d.addCallback(self.unpackage)
         return d
     
     def get_task_result(self, taskid, block=False):
-        """The task result by taskid.
+        """
+        Get a task result by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be retrieved.
             block : boolean
                 Should I block until the task is done?
         
         :Returns: A `TaskResult` object that encapsulates the task result.
         """
         d = self.remote_reference.callRemote('get_task_result', taskid, block)
         d.addCallback(self.unpackage)
         return d 
     
     def abort(self, taskid):
-        """Abort a task by taskid.
+        """
+        Abort a task by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be aborted.
-            block : boolean
-                Should I block until the task is aborted.        
         """
         d = self.remote_reference.callRemote('abort', taskid)
         d.addCallback(self.unpackage)
         return d 
         
     def barrier(self, taskids):
-        """Block until all tasks are completed.
+        """Block until a set of tasks are completed.
         
         :Parameters:
             taskids : list, tuple
                 A sequence of taskids to block on.
         """
         d = self.remote_reference.callRemote('barrier', taskids)
         d.addCallback(self.unpackage)
         return d 
     
     def spin(self):
-        """touch the scheduler, to resume scheduling without submitting
-        a task.
+        """
+        Touch the scheduler, to resume scheduling without submitting a task.
+        
+        This method only needs to be called in unusual situations where the
+        scheduler is idle for some reason.
         """
         d = self.remote_reference.callRemote('spin')
         d.addCallback(self.unpackage)
         return d
     
     def queue_status(self, verbose=False):
-        """Return a dict with the status of the task queue."""
+        """
+        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.
+        """
         d = self.remote_reference.callRemote('queue_status', verbose)
         d.addCallback(self.unpackage)
         return d
     
+    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.
+        """
+        d = self.remote_reference.callRemote('clear')
+        return d
+    
     def adapt_to_blocking_client(self):
+        """
+        Wrap self in a blocking version that implements `IBlockingTaskClient.
+        """
         from IPython.kernel.taskclient import IBlockingTaskClient
         return IBlockingTaskClient(self)
+    
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is load balanced.
+        """
+        return self.mapper().map(func, *sequences)
+    
+    def mapper(self, clear_before=False, clear_after=False, retries=0, 
+                recovery_task=None, depend=None, block=True):
+        """
+        Create an `IMapper` implementer with a given set of arguments.
+        
+        The `IMapper` created using a task controller is load balanced.
+        
+        See the documentation for `IPython.kernel.task.BaseTask` for 
+        documentation on the arguments to this method.
+        """
+        return TaskMapper(self, clear_before=clear_before, 
+            clear_after=clear_after, retries=retries, 
+            recovery_task=recovery_task, depend=depend, block=block)
+    
+    def parallel(self, clear_before=False, clear_after=False, retries=0, 
+        recovery_task=None, depend=None, block=True):
+        mapper = self.mapper(clear_before, clear_after, retries,
+            recovery_task, depend, block)
+        pf = ParallelFunction(mapper)
+        return pf
 

IPython/kernel/tests/tasktest.py

@@ -1,158 +1,187 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 __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 time
 
 from IPython.kernel import task, engineservice as es
 from IPython.kernel.util import printer
 from IPython.kernel import error
 
 #-------------------------------------------------------------------------------
 # Tests
 #-------------------------------------------------------------------------------
 
 def _raise_it(f):
     try:
         f.raiseException()
     except CompositeError, e:
         e.raise_exception() 
 
 class TaskTestBase(object):
     
     def addEngine(self, n=1):
         for i in range(n):
             e = es.EngineService()
             e.startService()
             regDict = self.controller.register_engine(es.QueuedEngine(e), None)
             e.id = regDict['id']
             self.engines.append(e)
 
 
 class ITaskControllerTestCase(TaskTestBase):
         
-    def testTaskIDs(self):
+    def test_task_ids(self):
         self.addEngine(1)
-        d = self.tc.run(task.Task('a=5'))
+        d = self.tc.run(task.StringTask('a=5'))
         d.addCallback(lambda r: self.assertEquals(r, 0))
-        d.addCallback(lambda r: self.tc.run(task.Task('a=5')))
+        d.addCallback(lambda r: self.tc.run(task.StringTask('a=5')))
         d.addCallback(lambda r: self.assertEquals(r, 1))
-        d.addCallback(lambda r: self.tc.run(task.Task('a=5')))
+        d.addCallback(lambda r: self.tc.run(task.StringTask('a=5')))
         d.addCallback(lambda r: self.assertEquals(r, 2))
-        d.addCallback(lambda r: self.tc.run(task.Task('a=5')))
+        d.addCallback(lambda r: self.tc.run(task.StringTask('a=5')))
         d.addCallback(lambda r: self.assertEquals(r, 3))
         return d
     
-    def testAbort(self):
+    def test_abort(self):
         """Cannot do a proper abort test, because blocking execution prevents
         abort from being called before task completes"""
         self.addEngine(1)
-        t = task.Task('a=5')
+        t = task.StringTask('a=5')
         d = self.tc.abort(0)
         d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
         d.addCallback(lambda _:self.tc.run(t))
         d.addCallback(self.tc.abort)
         d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
         return d
     
-    def testAbortType(self):
+    def test_abort_type(self):
         self.addEngine(1)
         d = self.tc.abort('asdfadsf')
         d.addErrback(lambda f: self.assertRaises(TypeError, f.raiseException))
         return d
     
-    def testClears(self):
+    def test_clear_before_and_after(self):
         self.addEngine(1)
-        t = task.Task('a=1', clear_before=True, pull='b', clear_after=True)
+        t = task.StringTask('a=1', clear_before=True, pull='b', clear_after=True)
         d = self.multiengine.execute('b=1', targets=0)
         d.addCallback(lambda _: self.tc.run(t))
         d.addCallback(lambda tid: self.tc.get_task_result(tid,block=True))
         d.addCallback(lambda tr: tr.failure)
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         d.addCallback(lambda _:self.multiengine.pull('a', targets=0))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         return d
     
-    def testSimpleRetries(self):
+    def test_simple_retries(self):
         self.addEngine(1)
-        t = task.Task("i += 1\nassert i == 16", pull='i',retries=10)
-        t2 = task.Task("i += 1\nassert i == 16", pull='i',retries=10)
+        t = task.StringTask("i += 1\nassert i == 16", pull='i',retries=10)
+        t2 = task.StringTask("i += 1\nassert i == 16", pull='i',retries=10)
         d = self.multiengine.execute('i=0', targets=0)
         d.addCallback(lambda r: self.tc.run(t))
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: tr.ns.i)
         d.addErrback(lambda f: self.assertRaises(AssertionError, f.raiseException))
         
         d.addCallback(lambda r: self.tc.run(t2))
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: tr.ns.i)
         d.addCallback(lambda r: self.assertEquals(r, 16))
         return d
     
-    def testRecoveryTasks(self):
+    def test_recovery_tasks(self):
         self.addEngine(1)
-        t = task.Task("i=16", pull='i')
-        t2 = task.Task("raise Exception", recovery_task=t, retries = 2)
+        t = task.StringTask("i=16", pull='i')
+        t2 = task.StringTask("raise Exception", recovery_task=t, retries = 2)
         
         d = self.tc.run(t2)
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: tr.ns.i)
         d.addCallback(lambda r: self.assertEquals(r, 16))
         return d
     
-    # def testInfiniteRecoveryLoop(self):
-    #     self.addEngine(1)
-    #     t = task.Task("raise Exception", retries = 5)
-    #     t2 = task.Task("assert True", retries = 2, recovery_task = t)
-    #     t.recovery_task = t2
-    #     
-    #     d = self.tc.run(t)
-    #     d.addCallback(self.tc.get_task_result, block=True)
-    #     d.addCallback(lambda tr: tr.ns.i)
-    #     d.addBoth(printer)
-    #     d.addErrback(lambda f: self.assertRaises(AssertionError, f.raiseException))
-    #     return d
-    # 
-    def testSetupNS(self):
+    def test_setup_ns(self):
         self.addEngine(1)
         d = self.multiengine.execute('a=0', targets=0)
         ns = dict(a=1, b=0)
-        t = task.Task("", push=ns, pull=['a','b'])
+        t = task.StringTask("", push=ns, pull=['a','b'])
         d.addCallback(lambda r: self.tc.run(t))
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: {'a':tr.ns.a, 'b':tr['b']})
         d.addCallback(lambda r: self.assertEquals(r, ns))
         return d
     
-    def testTaskResults(self):
+    def test_string_task_results(self):
         self.addEngine(1)
-        t1 = task.Task('a=5', pull='a')
+        t1 = task.StringTask('a=5', pull='a')
         d = self.tc.run(t1)
         d.addCallback(self.tc.get_task_result, block=True)
-        d.addCallback(lambda tr: (tr.ns.a,tr['a'],tr.failure, tr.raiseException()))
+        d.addCallback(lambda tr: (tr.ns.a,tr['a'],tr.failure, tr.raise_exception()))
         d.addCallback(lambda r: self.assertEquals(r, (5,5,None,None)))  
         
-        t2 = task.Task('7=5')
+        t2 = task.StringTask('7=5')
         d.addCallback(lambda r: self.tc.run(t2))
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: tr.ns)
         d.addErrback(lambda f: self.assertRaises(SyntaxError, f.raiseException))
         
-        t3 = task.Task('', pull='b')
+        t3 = task.StringTask('', pull='b')
         d.addCallback(lambda r: self.tc.run(t3))
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: tr.ns)
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         return d
+    
+    def test_map_task(self):
+        self.addEngine(1)
+        t1 = task.MapTask(lambda x: 2*x,(10,))
+        d = self.tc.run(t1)
+        d.addCallback(self.tc.get_task_result, block=True)
+        d.addCallback(lambda r: self.assertEquals(r,20))
+        
+        t2 = task.MapTask(lambda : 20)
+        d.addCallback(lambda _: self.tc.run(t2))
+        d.addCallback(self.tc.get_task_result, block=True)
+        d.addCallback(lambda r: self.assertEquals(r,20))
+
+        t3 = task.MapTask(lambda x: x,(),{'x':20})
+        d.addCallback(lambda _: self.tc.run(t3))
+        d.addCallback(self.tc.get_task_result, block=True)
+        d.addCallback(lambda r: self.assertEquals(r,20))
+        return d
+    
+    def test_map_task_failure(self):
+        self.addEngine(1)
+        t1 = task.MapTask(lambda x: 1/0,(10,))
+        d = self.tc.run(t1)
+        d.addCallback(self.tc.get_task_result, block=True)
+        d.addErrback(lambda f: self.assertRaises(ZeroDivisionError, f.raiseException))
+        return d
+    
+    def test_map_task_args(self):
+        self.assertRaises(TypeError, task.MapTask, 'asdfasdf')
+        self.assertRaises(TypeError, task.MapTask, lambda x: x, 10)
+        self.assertRaises(TypeError, task.MapTask, lambda x: x, (10,),30)
+    
+    def test_clear(self):
+        self.addEngine(1)
+        t1 = task.MapTask(lambda x: 2*x,(10,))
+        d = self.tc.run(t1)
+        d.addCallback(lambda _: self.tc.get_task_result(0, block=True))
+        d.addCallback(lambda r: self.assertEquals(r,20))
+        d.addCallback(lambda _: self.tc.clear())
+        d.addCallback(lambda _: self.tc.get_task_result(0, block=True))
+        d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
+        return d

IPython/kernel/tests/test_taskfc.py

@@ -1,90 +1,161 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 __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
 #-------------------------------------------------------------------------------
 
 try:
     import time
 
     from twisted.internet import defer, reactor
 
     from IPython.kernel.fcutil import Tub, UnauthenticatedTub
 
     from IPython.kernel import task as taskmodule
     from IPython.kernel import controllerservice as cs
     import IPython.kernel.multiengine as me
     from IPython.testing.util import DeferredTestCase
     from IPython.kernel.multienginefc import IFCSynchronousMultiEngine
     from IPython.kernel.taskfc import IFCTaskController
     from IPython.kernel.util import printer
     from IPython.kernel.tests.tasktest import ITaskControllerTestCase
     from IPython.kernel.clientconnector import ClientConnector
+    from IPython.kernel.error import CompositeError
+    from IPython.kernel.parallelfunction import ParallelFunction
 except ImportError:
     pass
 else:
 
     #-------------------------------------------------------------------------------
     # Tests
     #-------------------------------------------------------------------------------
 
+    def _raise_it(f):
+        try:
+            f.raiseException()
+        except CompositeError, e:
+            e.raise_exception()
+
     class TaskTest(DeferredTestCase, ITaskControllerTestCase):
 
         def setUp(self):
         
             self.engines = []
                 
             self.controller = cs.ControllerService()
             self.controller.startService()
             self.imultiengine = me.IMultiEngine(self.controller)
             self.itc = taskmodule.ITaskController(self.controller)
             self.itc.failurePenalty = 0
         
             self.mec_referenceable = IFCSynchronousMultiEngine(self.imultiengine)
             self.tc_referenceable = IFCTaskController(self.itc)
         
             self.controller_tub = Tub()
             self.controller_tub.listenOn('tcp:10105:interface=127.0.0.1')
             self.controller_tub.setLocation('127.0.0.1:10105')
         
             mec_furl = self.controller_tub.registerReference(self.mec_referenceable)
             tc_furl = self.controller_tub.registerReference(self.tc_referenceable)
             self.controller_tub.startService()
         
             self.client_tub = ClientConnector()
             d = self.client_tub.get_multiengine_client(mec_furl)
             d.addCallback(self.handle_mec_client)
             d.addCallback(lambda _: self.client_tub.get_task_client(tc_furl))
             d.addCallback(self.handle_tc_client)
             return d
         
         def handle_mec_client(self, client):
             self.multiengine = client
     
         def handle_tc_client(self, client):
             self.tc = client
     
         def tearDown(self):
             dlist = []
             # Shut down the multiengine client
             d = self.client_tub.tub.stopService()
             dlist.append(d)
             # Shut down the engines
             for e in self.engines:
                 e.stopService()
             # Shut down the controller
             d = self.controller_tub.stopService()
             d.addBoth(lambda _: self.controller.stopService())
             dlist.append(d)
             return defer.DeferredList(dlist)
-
+        
+        def test_mapper(self):
+            self.addEngine(1)
+            m = self.tc.mapper()
+            self.assertEquals(m.task_controller,self.tc)
+            self.assertEquals(m.clear_before,False)
+            self.assertEquals(m.clear_after,False)
+            self.assertEquals(m.retries,0)
+            self.assertEquals(m.recovery_task,None)
+            self.assertEquals(m.depend,None)
+            self.assertEquals(m.block,True)
+        
+        def test_map_default(self):
+            self.addEngine(1)
+            m = self.tc.mapper()
+            d = m.map(lambda x: 2*x, range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            d.addCallback(lambda _: self.tc.map(lambda x: 2*x, range(10)))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+        
+        def test_map_noblock(self):
+            self.addEngine(1)
+            m = self.tc.mapper(block=False)
+            d = m.map(lambda x: 2*x, range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[x for x in range(10)]))
+            return d
+                
+        def test_mapper_fail(self):
+            self.addEngine(1)
+            m = self.tc.mapper()
+            d = m.map(lambda x: 1/0, range(10))
+            d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
+            return d
+        
+        def test_parallel(self):
+            self.addEngine(1)
+            p = self.tc.parallel()
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 2*x
+            d = f(range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+        
+        def test_parallel_noblock(self):
+            self.addEngine(1)
+            p = self.tc.parallel(block=False)
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 2*x
+            d = f(range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[x for x in range(10)]))
+            return d
+        
+        def test_parallel_fail(self):
+            self.addEngine(1)
+            p = self.tc.parallel()
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 1/0
+            d = f(range(10))
+            d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
+            return d
  No newline at end of file

docs/examples/kernel/mcdriver.py

@@ -1,71 +1,71 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """Run a Monte-Carlo options pricer in parallel."""
 
 from IPython.kernel import client
 import numpy as N
 from mcpricer import MCOptionPricer
 
 
 tc = client.TaskClient()
 rc = client.MultiEngineClient()
 
 # Initialize the common code on the engines
 rc.run('mcpricer.py')
 
 # Push the variables that won't change 
 #(stock print, interest rate, days and MC paths)
 rc.push(dict(S=100.0, r=0.05, days=260, paths=10000))
 
 task_string = """\
 op = MCOptionPricer(S,K,sigma,r,days,paths)
 op.run()
 vp, ap, vc, ac = op.vanilla_put, op.asian_put, op.vanilla_call, op.asian_call
 """
 
 # Create arrays of strike prices and volatilities
 K_vals = N.linspace(90.0,100.0,5)
 sigma_vals = N.linspace(0.0, 0.2,5)
 
 # Submit tasks
 taskids = []
 for K in K_vals:
     for sigma in sigma_vals:
-        t = client.Task(task_string, 
+        t = client.StringTask(task_string, 
             push=dict(sigma=sigma,K=K),
             pull=('vp','ap','vc','ac','sigma','K'))
         taskids.append(tc.run(t))
 
 print "Submitted tasks: ", taskids
 
 # Block until tasks are completed
 tc.barrier(taskids)
 
 # Get the results
 results = [tc.get_task_result(tid) for tid in taskids]
 
 # Assemble the result
 vc = N.empty(K_vals.shape[0]*sigma_vals.shape[0],dtype='float64')
 vp = N.empty(K_vals.shape[0]*sigma_vals.shape[0],dtype='float64')
 ac = N.empty(K_vals.shape[0]*sigma_vals.shape[0],dtype='float64')
 ap = N.empty(K_vals.shape[0]*sigma_vals.shape[0],dtype='float64')
 for i, tr in enumerate(results):
     ns = tr.ns
     vc[i] = ns.vc
     vp[i] = ns.vp
     ac[i] = ns.ac
     ap[i] = ns.ap
 vc.shape = (K_vals.shape[0],sigma_vals.shape[0])
 vp.shape = (K_vals.shape[0],sigma_vals.shape[0])
 ac.shape = (K_vals.shape[0],sigma_vals.shape[0])
 ap.shape = (K_vals.shape[0],sigma_vals.shape[0])
 
 
 def plot_options(K_vals, sigma_vals, prices):
     """Make a contour plot of the option prices."""
     import pylab
     pylab.contourf(sigma_vals, K_vals, prices)
     pylab.colorbar()
     pylab.title("Option Price")
     pylab.xlabel("Volatility")
     pylab.ylabel("Strike Price")

docs/examples/kernel/task1.py

@@ -1,18 +1,18 b''
 from IPython.kernel import client
 
 tc = client.TaskClient()
 rc = client.MultiEngineClient()
 
 rc.push(dict(d=30))
 
 cmd1 = """\
 a = 5
 b = 10*d
 c = a*b*d
 """
 
-t1 = client.Task(cmd1, clear_before=False, clear_after=True, pull=['a','b','c'])
+t1 = client.StringTask(cmd1, clear_before=False, clear_after=True, pull=['a','b','c'])
 tid1 = tc.run(t1)
 tr1 = tc.get_task_result(tid1,block=True)
 tr1.raiseException()
 print "a, b: ", tr1.ns.a, tr1.ns.b
  No newline at end of file

docs/examples/kernel/task2.py

@@ -1,44 +1,44 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 from IPython.kernel import client
 import time
 
 tc = client.TaskClient()
 mec = client.MultiEngineClient()
 
 mec.execute('import time')
 
 for i in range(24):
-    tc.irun('time.sleep(1)')
+    tc.run(client.StringTask('time.sleep(1)'))
 
 for i in range(6):
     time.sleep(1.0)
     print "Queue status (vebose=False)"
     print tc.queue_status()
     
 for i in range(24):
-    tc.irun('time.sleep(1)')
+    tc.run(client.StringTask('time.sleep(1)'))
 
 for i in range(6):
     time.sleep(1.0)
     print "Queue status (vebose=True)"
     print tc.queue_status(True)
 
 for i in range(12):
-    tc.irun('time.sleep(2)')
+    tc.run(client.StringTask('time.sleep(2)'))
 
 print "Queue status (vebose=True)"
 print tc.queue_status(True)
 
 qs = tc.queue_status(True)
 sched = qs['scheduled']
 
 for tid in sched[-4:]:
     tc.abort(tid)
 
 for i in range(6):
     time.sleep(1.0)
     print "Queue status (vebose=True)"
     print tc.queue_status(True)
 

docs/examples/kernel/task_profiler.py

@@ -1,77 +1,77 b''
 #!/usr/bin/env python
 """Test the performance of the task farming system.
 
 This script submits a set of tasks to the TaskClient.  The tasks
 are basically just a time.sleep(t), where t is a random number between
 two limits that can be configured at the command line.  To run 
 the script there must first be an IPython controller and engines running::
 
     ipcluster -n 16
 
 A good test to run with 16 engines is::
 
     python task_profiler.py -n 128 -t 0.01 -T 1.0
 
 This should show a speedup of 13-14x.  The limitation here is that the 
 overhead of a single task is about 0.001-0.01 seconds.
 """
 import random, sys
 from optparse import OptionParser
 
 from IPython.genutils import time
 from IPython.kernel import client
 
 def main():
     parser = OptionParser()
     parser.set_defaults(n=100)
     parser.set_defaults(tmin=1)
     parser.set_defaults(tmax=60)
     parser.set_defaults(controller='localhost')
     parser.set_defaults(meport=10105)
     parser.set_defaults(tport=10113)
     
     parser.add_option("-n", type='int', dest='n',
         help='the number of tasks to run')
     parser.add_option("-t", type='float', dest='tmin', 
         help='the minimum task length in seconds')
     parser.add_option("-T", type='float', dest='tmax',
         help='the maximum task length in seconds')
     parser.add_option("-c", type='string', dest='controller',
         help='the address of the controller')
     parser.add_option("-p", type='int', dest='meport',
         help="the port on which the controller listens for the MultiEngine/RemoteController client")
     parser.add_option("-P", type='int', dest='tport',
         help="the port on which the controller listens for the TaskClient client")
     
     (opts, args) = parser.parse_args()
     assert opts.tmax >= opts.tmin, "tmax must not be smaller than tmin"
     
     rc = client.MultiEngineClient()
     tc = client.TaskClient()
     print tc.task_controller
     rc.block=True
     nengines = len(rc.get_ids())
     rc.execute('from IPython.genutils import time')
 
     # the jobs should take a random time within a range
     times = [random.random()*(opts.tmax-opts.tmin)+opts.tmin for i in range(opts.n)]
-    tasks = [client.Task("time.sleep(%f)"%t) for t in times]
+    tasks = [client.StringTask("time.sleep(%f)"%t) for t in times]
     stime = sum(times)
     
     print "executing %i tasks, totalling %.1f secs on %i engines"%(opts.n, stime, nengines)
     time.sleep(1)
     start = time.time()
     taskids = [tc.run(t) for t in tasks]
     tc.barrier(taskids)
     stop = time.time()
 
     ptime = stop-start
     scale = stime/ptime
     
     print "executed %.1f secs in %.1f secs"%(stime, ptime)
     print "%.3fx parallel performance on %i engines"%(scale, nengines)
     print "%.1f%% of theoretical max"%(100*scale/nengines)
 
 
 if __name__ == '__main__':
     main()