Merge with upstream

IPython/kernel/mapper.py

@@ -0,0 +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 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 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 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.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

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/frontend/tests/test_frontendbase.py

@@ -1,151 +1,151 b''
 # encoding: utf-8
 
 """This file contains unittests for the frontendbase module."""
 
 __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 unittest
 from IPython.frontend import frontendbase
 from IPython.kernel.engineservice import EngineService
 
 class FrontEndCallbackChecker(frontendbase.AsyncFrontEndBase):
     """FrontEndBase subclass for checking callbacks"""
     def __init__(self, engine=None, history=None):
         super(FrontEndCallbackChecker, self).__init__(engine=engine, 
                                                     history=history)
         self.updateCalled = False
         self.renderResultCalled = False
         self.renderErrorCalled = False
     
     def update_cell_prompt(self, result, blockID=None):
         self.updateCalled = True
         return result
     
     def render_result(self, result):
         self.renderResultCalled = True
         return result
     
     
     def render_error(self, failure):
         self.renderErrorCalled = True
         return failure
     
 
 
     
 class TestAsyncFrontendBase(unittest.TestCase):
     def setUp(self):
         """Setup the EngineService and FrontEndBase"""
         
         self.fb = FrontEndCallbackChecker(engine=EngineService())
     
     
     def test_implements_IFrontEnd(self):
         assert(frontendbase.IFrontEnd.implementedBy(
                                     frontendbase.AsyncFrontEndBase))
     
     
     def test_is_complete_returns_False_for_incomplete_block(self):
         """"""
         
         block = """def test(a):"""
         
         assert(self.fb.is_complete(block) == False)
     
     def test_is_complete_returns_True_for_complete_block(self):
         """"""
         
         block = """def test(a): pass"""
             
         assert(self.fb.is_complete(block))
         
         block = """a=3"""
         
         assert(self.fb.is_complete(block))
     
     
     def test_blockID_added_to_result(self):
         block = """3+3"""
         
         d = self.fb.execute(block, blockID='TEST_ID')
         
         d.addCallback(self.checkBlockID, expected='TEST_ID')
     
     def test_blockID_added_to_failure(self):
-        block = "raise  Exception()"
+        block = "raise Exception()"
         
         d = self.fb.execute(block,blockID='TEST_ID')
         d.addErrback(self.checkFailureID, expected='TEST_ID')
     
     def checkBlockID(self, result, expected=""):
         assert(result['blockID'] == expected)
     
     
     def checkFailureID(self, failure, expected=""):
         assert(failure.blockID == expected)
     
     
     def test_callbacks_added_to_execute(self):
         """test that
         update_cell_prompt
         render_result
         
         are added to execute request
         """
         
         d = self.fb.execute("10+10")
         d.addCallback(self.checkCallbacks)
     
     
     def checkCallbacks(self, result):
         assert(self.fb.updateCalled)
         assert(self.fb.renderResultCalled)
     
     
     def test_error_callback_added_to_execute(self):
         """test that render_error called on execution error"""
         
         d = self.fb.execute("raise Exception()")
         d.addCallback(self.checkRenderError)
     
     def checkRenderError(self, result):
         assert(self.fb.renderErrorCalled)
     
     def test_history_returns_expected_block(self):
         """Make sure history browsing doesn't fail"""
         
         blocks = ["a=1","a=2","a=3"]
         for b in blocks:
             d = self.fb.execute(b)
         
         # d is now the deferred for the last executed block
         d.addCallback(self.historyTests, blocks)
         
     
     def historyTests(self, result, blocks):
         """historyTests"""
         
         assert(len(blocks) >= 3)
         assert(self.fb.get_history_previous("") == blocks[-2])
         assert(self.fb.get_history_previous("") == blocks[-3])
         assert(self.fb.get_history_next() == blocks[-2])
     
     
     def test_history_returns_none_at_startup(self):
         """test_history_returns_none_at_startup"""
         
         assert(self.fb.get_history_previous("")==None)
         assert(self.fb.get_history_next()==None)
     
     

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/contexts.py

@@ -1,186 +1,143 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_contexts -*-
 """Context managers for IPython.
 
 Python 2.5 introduced the `with` statement, which is based on the context
 manager protocol.  This module offers a few context managers for common cases,
 which can also be useful as templates for writing new, application-specific
 managers.
 """
 
 from __future__ import with_statement
 
 __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 linecache
 import sys
 
 from twisted.internet.error import ConnectionRefusedError
 
 from IPython.ultraTB import _fixed_getinnerframes, findsource
 from IPython import ipapi
 
 from IPython.kernel import error
 
 #---------------------------------------------------------------------------
 # Utility functions needed by all context managers.
 #---------------------------------------------------------------------------
 
 def remote():
     """Raises a special exception meant to be caught by context managers.
     """
     m = 'Special exception to stop local execution of parallel code.'
     raise error.StopLocalExecution(m)
 
 
 def strip_whitespace(source,require_remote=True):
     """strip leading whitespace from input source.
 
     :Parameters:
 
     """
     remote_mark = 'remote()'
     # 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()'
     if require_remote:
         for nline,line in enumerate(src_lines):
             if line.isspace() or line.startswith('#'):
                 continue
             if line.startswith(remote_mark):
                 break
             else:
                 raise ValueError('%s call missing at the start of code' %
                                  remote_mark)
         out_lines = src_lines[nline+1:]
     else:
         # If the user specified that the remote() call wasn't mandatory
         out_lines = src_lines
 
     # src = ''.join(out_lines)  # dbg
     #print 'SRC:\n<<<<<<<>>>>>>>\n%s<<<<<>>>>>>' % src  # dbg
     return ''.join(out_lines)
 
 class RemoteContextBase(object):
     def __init__(self):
         self.ip = ipapi.get()
 
     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()
         buf = self.ip.IP.input_hist_raw[-1].splitlines()[1:]
         wsource = [l+'\n' for l in buf ]
 
         return strip_whitespace(wsource)
 
     def findsource(self,frame):
         local_ns = frame.f_locals
         global_ns = frame.f_globals
         if frame.f_code.co_filename == '<ipython console>':
             src = self._findsource_ipython(frame)
         else:
             src = self._findsource_file(frame)
         return src
 
     def __enter__(self):
         raise NotImplementedError
 
     def __exit__ (self, etype, value, tb):
         if issubclass(etype,error.StopLocalExecution):
             return True
 
 class RemoteMultiEngine(RemoteContextBase):
     def __init__(self,mec):
         self.mec = mec
         RemoteContextBase.__init__(self)
 
     def __enter__(self):
         src = self.findsource(sys._getframe(1))
         return self.mec.execute(src)
-
-
-# XXX - Temporary hackish testing, we'll move this into proper tests right
-# away
-
-if __name__ == '__main__':
-
-    # XXX - for now, we need a running cluster to be started separately.  The
-    # daemon work is almost finished, and will make much of this unnecessary.
-    from IPython.kernel import client
-    mec = client.MultiEngineClient(('127.0.0.1',10105))
-
-    try:
-        mec.get_ids()
-    except ConnectionRefusedError:
-        import os, time
-        os.system('ipcluster -n 2 &')
-        time.sleep(2)
-        mec = client.MultiEngineClient(('127.0.0.1',10105))
-
-    mec.block = False
-
-    import itertools
-    c = itertools.count()
-
-    parallel = RemoteMultiEngine(mec)
-
-    mec.pushAll()
-
-    with parallel as pr:
-        # A comment
-        remote()  # this means the code below only runs remotely
-        print 'Hello remote world'
-        x = range(10)
-        # Comments are OK
-        # Even misindented.
-        y = x+1
-
-
-    with pfor('i',sequence) as pr:
-        print x[i]
-
-    print pr.x + pr.y

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/map.py

@@ -1,121 +1,121 b''
 # encoding: utf-8
 
 """Classes used in scattering and gathering sequences.
 
 Scattering consists of partitioning a sequence and sending the various
 pieces to individual nodes in a cluster.
 """
 
 __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 types
 
 from IPython.genutils import flatten as genutil_flatten
 
 #-------------------------------------------------------------------------------
 # Figure out which array packages are present and their array types
 #-------------------------------------------------------------------------------
 
 arrayModules = []
 try:
     import Numeric
 except ImportError:
     pass
 else:
     arrayModules.append({'module':Numeric, 'type':Numeric.arraytype})
 try:
     import numpy
 except ImportError:
     pass
 else:
     arrayModules.append({'module':numpy, 'type':numpy.ndarray})
 try:
     import numarray
 except ImportError:
     pass
 else:
     arrayModules.append({'module':numarray,
         'type':numarray.numarraycore.NumArray})
 
 class Map:
     """A class for partitioning a sequence using a map."""
             
     def getPartition(self, seq, p, q):
         """Returns the pth partition of q partitions of seq."""
         
         # Test for error conditions here
         if p<0 or p>=q:
           print "No partition exists."
           return
           
         remainder = len(seq)%q
         basesize = len(seq)/q
         hi = []
         lo = []
         for n in range(q):
             if n < remainder:
                 lo.append(n * (basesize + 1))
                 hi.append(lo[-1] + basesize + 1)
             else:
                 lo.append(n*basesize + remainder)
                 hi.append(lo[-1] + basesize)
 
         
         result = seq[lo[p]:hi[p]]
         return result
            
     def joinPartitions(self, listOfPartitions):
         return self.concatenate(listOfPartitions)
                     
     def concatenate(self, listOfPartitions):
         testObject = listOfPartitions[0]
         # First see if we have a known array type
         for m in arrayModules:
             #print m
             if isinstance(testObject, m['type']):
                 return m['module'].concatenate(listOfPartitions)
         # Next try for Python sequence types
         if isinstance(testObject, (types.ListType, types.TupleType)):
             return genutil_flatten(listOfPartitions)
         # If we have scalars, just return listOfPartitions
         return listOfPartitions
 
 class RoundRobinMap(Map):
     """Partitions a sequence in a roun robin fashion.
     
     This currently does not work!
     """
 
     def getPartition(self, seq, p, q):
         return seq[p:len(seq):q]
         #result = []
         #for i in range(p,len(seq),q):
         #    result.append(seq[i])
         #return result
 
     def joinPartitions(self, listOfPartitions):
         #lengths = [len(x) for x in listOfPartitions]
         #maxPartitionLength = len(listOfPartitions[0])
         #numberOfPartitions = len(listOfPartitions)
         #concat = self.concatenate(listOfPartitions)
         #totalLength = len(concat)
         #result = []
         #for i in range(maxPartitionLength):
         #    result.append(concat[i:totalLength:maxPartitionLength])
         return self.concatenate(listOfPartitions)
         
-styles = {'basic':Map}
+dists = {'b':Map}
 
     
     

IPython/kernel/multiengine.py

@@ -1,780 +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, style='basic', flatten=False, targets='all'):
-        """Partition and distribute a sequence to targets.
+    def scatter(key, seq, dist='b', flatten=False, targets='all'):
+        """Partition and distribute a sequence to targets."""
         
-        :Parameters:
-            key : str
-                The variable name to call the scattered sequence.
-            seq : list, tuple, array
-                The sequence to scatter.  The type should be preserved.
-            style : string
-                A specification of how the sequence is partitioned.  Currently 
-                only 'basic' is implemented.
-            flatten : boolean
-                Should single element sequences be converted to scalars.
-        """
-        
-    def gather(key, style='basic', targets='all'):
-        """Gather object key from targets.
+    def gather(key, dist='b', targets='all'):
+        """Gather object key from targets."""
     
-        :Parameters:
-            key : string
-                The name of a sequence on the targets to gather.
-            style : string
-                A specification of how the sequence is partitioned.  Currently 
-                only 'basic' is implemented.                
+    def raw_map(func, seqs, dist='b', targets='all'):
         """
-    
-    def map(func, seq, style='basic', targets='all'):
-        """A parallelized version of Python's builtin map.
+        A parallelized version of Python's builtin `map` function.
         
-        This function implements the following pattern:
+        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.
         
-        1. The sequence seq is scattered to the given targets.
-        2. map(functionSource, seq) is called on each engine.
-        3. The resulting sequences are gathered back to the local machine.
-                
-        :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.
-            func : str, function
-                An actual function object or a Python string that names a 
-                callable defined on the engines.
-            seq : list, tuple or numpy array
-                The local sequence to be scattered.
-            style : str
-                Only 'basic' is supported for now.
-                
-        :Returns: A list of len(seq) with functionSource called on each element
-        of seq.
-                
-        Example
-        =======
+        The equivalence is:
         
-        >>> rc.mapAll('lambda x: x*x', range(10000))
-        [0,2,4,9,25,36,...]
+        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."""
-    pass
+
+    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 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`.
+    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.
-        
-        :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.
-            keys: list or tuple of str
-                A list of variable names as string of the Python objects to be pulled
-                back to the client.
-        
-        :Returns: A list of pulled Python objects for each target.
         """
     
     def run(targets, fname):
-        """Run a .py file on targets.
-        
-        :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.
-            fname : str
-                The filename of a .py file on the local system to be sent to and run
-                on the engines.
-            block : boolean
-                Should I block or not.  If block=True, wait for the action to
-                complete and return the result.  If block=False, return a
-                `PendingResult` object that can be used to later get the
-                result.  If block is not specified, the block attribute 
-                will be used instead. 
-        """
+        """Run a .py file on targets."""
 
 
 class ISynchronousMultiEngineExtras(IMultiEngineExtras):
-    pass
-
+    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,833 +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 (
+    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.                
-        """
-        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, style='basic', flatten=False, targets=None, block=None):
+    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, 
-            style, flatten, targets=targets, block=block)
+            dist, flatten, targets=targets, block=block)
     
-    def gather(self, key, style='basic', targets=None, block=None):
+    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, style, 
+        return self._blockFromThread(self.smultiengine.gather, key, dist, 
             targets=targets, block=block)
     
-    def map(self, func, seq, style='basic', 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, 
-            style, targets=targets, block=block)
+        return self._blockFromThread(self.smultiengine.raw_map, func, seq, 
+            dist, targets=targets, block=block)
+    
+    def map(self, 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):
+        """
+        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)       
+        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,668 +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 (
+    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, style='basic', flatten=False, targets='all', block=True):
+    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.styles[style]
+            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, style='basic', targets='all', block=True):
+    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.styles[style]
+            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, seq, style='basic', targets='all', block=True):
-        d_list = []
+    def raw_map(self, func, sequences, 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`.
+        """
+        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, _ipython_map_seq)'
+            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, _ipython_map_seq)' % func
+                '_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', seq, style, targets=targets))
+        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', style, targets=targets, block=block))
+        d.addCallback(lambda _: self.gather('_ipython_map_seq_result', dist, targets=targets, block=block))
         return d
 
+    def map(self, 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):
+        """
+        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):
+        """
+        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,32 +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 ParallelFunction:
-    """A function that operates in parallel on sequences."""
-    def __init__(self, func, multiengine, targets, block):
-        """Create a `ParallelFunction`.
+
+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):
+    """
+    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.
+    """
+    
+    implements(IParallelFunction)
+    
+    def __init__(self, mapper):
+        """
+        Create a parallel function from an `IMapper`.
+        
+        :Parameters:
+            mapper : an `IMapper` implementer.
+                The mapper to use for the parallel function
+        """
+        self.mapper = mapper
+        
+    def __call__(self, func):
+        """
+        Decorate a function to make it run in parallel.
         """
         assert isinstance(func, (str, FunctionType)), "func must be a fuction or str"
         self.func = func
-        self.multiengine = multiengine
-        self.targets = targets
-        self.block = block
-        
-    def __call__(self, sequence):
-        return self.multiengine.map(self.func, sequence, targets=self.targets, block=self.block)
  No newline at end of file
+        def call_function(*sequences):
+            return self.mapper.map(self.func, *sequences)
+        return call_function
+
+    
  No newline at end of file

IPython/kernel/task.py

@@ -1,803 +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 as function
+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 canTask(task):
-    t = copy.copy(task)
-    t.depend = can(t.depend)
-    if t.recovery_task:
-        t.recovery_task = canTask(t.recovery_task)
-    return t
+from IPython.kernel.pickleutil import can, uncan, CannedFunction
 
-def uncanTask(task):
-    t = copy.copy(task)
-    t.depend = uncan(t.depend)
-    if t.recovery_task and t.recovery_task is not task:
-        t.recovery_task = uncanTask(t.recovery_task)
-    return t
+#-----------------------------------------------------------------------------
+# Definition of the Task objects
+#-----------------------------------------------------------------------------
 
 time_format = '%Y/%m/%d %H:%M:%S'
 
-class Task(object):
-    r"""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.
+    
+    Many method in this class get two things passed to them: a Deferred
+    and an IQueuedEngine implementer.  Such methods should register callbacks
+    on the Deferred that use the IQueuedEngine to accomplish something.  See
+    the existing task objects for examples.
+    """
+    
+    zi.Attribute('retries','How many times to retry the task')
+    zi.Attribute('recovery_task','A task to try if the initial one fails')
+    zi.Attribute('taskid','the id of the task')
+    
+    def start_time(result):
+        """
+        Do anything needed to start the timing of the task.
+        
+        Must simply return the result after starting the timers.
+        """
     
-    >>> 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
+    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
+        """
 
-    A dependency case:
-    >>> def hasMPI(props):
-    ...     return props.get('mpi') is not None
-    >>> t = Task('mpi.send(blah,blah)', depend = hasMPI)
+    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`.
     """
     
-    def __init__(self, expression, pull=None, push=None,
-            clear_before=False, clear_after=False, retries=0, 
-            recovery_task=None, depend=None, **options):
-        self.expression = expression
-        if isinstance(pull, str):
-            self.pull = [pull]
-        else:
-            self.pull = pull
-        self.push = push
+    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.retries = retries
         self.recovery_task = recovery_task
         self.depend = depend
-        self.options = options
         self.taskid = None
+    
+    def start_time(self, result):
+        """
+        Start the basic timers.
+        """
+        self.start = time.time()
+        self.start_struct = time.localtime()
+        return result
+    
+    def stop_time(self, result):
+        """
+        Stop the basic timers.
+        """
+        self.stop = time.time()
+        self.stop_struct = time.localtime()
+        self.duration = self.stop - self.start 
+        self.submitted = time.strftime(time_format, self.start_struct)
+        self.completed = time.strftime(time_format)
+        return result
+    
+    def pre_task(self, d, queued_engine):
+        """
+        Clear the engine before running the task if clear_before is set.
+        """
+        if self.clear_before:
+            d.addCallback(lambda r: queued_engine.reset())
+    
+    def post_task(self, d, queued_engine):
+        """
+        Clear the engine after running the task if clear_after is set.
+        """
+        def reseter(result):
+            queued_engine.reset()
+            return result
+        if self.clear_after:
+            d.addBoth(reseter)
+    
+    def submit_task(self, d, queued_engine):
+        raise NotImplementedError('submit_task must be implemented in a subclass')
+    
+    def process_result(self, result, engine_id):
+        """
+        Process a task result.
+        
+        This is the default `process_result` that just returns the raw
+        result or a `Failure`.
+        """
+        if isinstance(result, failure.Failure):
+            return (False, result)
+        else:
+            return (True, result)
+    
+    def check_depend(self, properties):
+        """
+        Calls self.depend(properties) to see if a task should be run.
+        """
+        if self.depend is not None:
+            return self.depend(properties)
+        else:
+            return True
+
+    def can_task(self):
+        self.depend = can(self.depend)
+        if isinstance(self.recovery_task, BaseTask):
+            self.recovery_task.can_task()
+            
+    def uncan_task(self):
+        self.depend = uncan(self.depend)
+        if isinstance(self.recovery_task, BaseTask):
+            self.recovery_task.uncan_task()
+
+class MapTask(BaseTask):
+    """
+    A task that consists of a function and arguments.
+    """
+    
+    zi.implements(ITask)
+    
+    def __init__(self, function, args=None, kwargs=None, clear_before=False, 
+            clear_after=False, retries=0, recovery_task=None, depend=None):
+        """
+        Create a task based on a function, args and kwargs.
+        
+        This is a simple type of task that consists of calling:
+        function(*args, **kwargs) and wrapping the result in a `TaskResult`.
+        
+        The return value of the function, or a `Failure` wrapping an 
+        exception is the task result for this type of task.
+        """
+        BaseTask.__init__(self, clear_before, clear_after, retries, 
+            recovery_task, depend)
+        if not isinstance(function, FunctionType):
+            raise TypeError('a task function must be a FunctionType')
+        self.function = function
+        if args is None:
+            self.args = ()
+        else:
+            self.args = args
+        if not isinstance(self.args, (list, tuple)):
+            raise TypeError('a task args must be a list or tuple')
+        if kwargs is None:
+            self.kwargs = {}
+        else:
+            self.kwargs = kwargs
+        if not isinstance(self.kwargs, dict):
+            raise TypeError('a task kwargs must be a dict')
+    
+    def submit_task(self, d, queued_engine):
+        d.addCallback(lambda r: queued_engine.push_function(
+            dict(_ipython_task_function=self.function))
+        )
+        d.addCallback(lambda r: queued_engine.push(
+            dict(_ipython_task_args=self.args,_ipython_task_kwargs=self.kwargs))
+        )
+        d.addCallback(lambda r: queued_engine.execute(
+            '_ipython_task_result = _ipython_task_function(*_ipython_task_args,**_ipython_task_kwargs)')
+        )
+        d.addCallback(lambda r: queued_engine.pull('_ipython_task_result'))
+    
+    def can_task(self):
+        self.function = can(self.function)
+        BaseTask.can_task(self)
+    
+    def uncan_task(self):
+        self.function = uncan(self.function)
+        BaseTask.uncan_task(self)
+
+
+class StringTask(BaseTask):
+    """
+    A task that consists of a string of Python code to run.
+    """
+
+    def __init__(self, expression, pull=None, push=None,
+            clear_before=False, clear_after=False, retries=0, 
+            recovery_task=None, depend=None):
+        """
+        Create a task based on a Python expression and variables
+        
+        This type of task lets you push a set of variables to the engines
+        namespace, run a Python string in that namespace and then bring back
+        a different set of Python variables as the result.
+        
+        Because this type of task can return many results (through the
+        `pull` keyword argument) it returns a special `TaskResult` object
+        that wraps the pulled variables, statistics about the run and
+        any exceptions raised.
+        """
+        if not isinstance(expression, str):
+            raise TypeError('a task expression must be a string')
+        self.expression = expression
+        
+        if pull==None:
+            self.pull = ()
+        elif isinstance(pull, str):
+            self.pull = (pull,)
+        elif isinstance(pull, (list, tuple)):
+            self.pull = pull
+        else:
+            raise TypeError('pull must be str or a sequence of strs')
+        
+        if push==None:
+            self.push = {}
+        elif isinstance(push, dict):
+            self.push = push
+        else:
+            raise TypeError('push must be a dict')
+        
+        BaseTask.__init__(self, clear_before, clear_after, retries, 
+            recovery_task, depend)
 
-class ResultNS:
-    """The result namespace object for use in TaskResult objects as tr.ns.
+    def submit_task(self, d, queued_engine):
+        if self.push is not None:
+            d.addCallback(lambda r: queued_engine.push(self.push))
+        
+        d.addCallback(lambda r: queued_engine.execute(self.expression))
+        
+        if self.pull is not None:
+            d.addCallback(lambda r: queued_engine.pull(self.pull))
+        else:
+            d.addCallback(lambda r: None)    
+    
+    def process_result(self, result, engine_id):
+        if isinstance(result, failure.Failure):
+            tr = TaskResult(result, engine_id)
+        else:
+            if self.pull is None:
+                resultDict = {}
+            elif len(self.pull) == 1:
+                resultDict = {self.pull[0]:result}
+            else:
+                resultDict = dict(zip(self.pull, result))
+            tr = TaskResult(resultDict, engine_id)
+        # Assign task attributes
+        tr.submitted = self.submitted
+        tr.completed = self.completed
+        tr.duration = self.duration
+        if hasattr(self,'taskid'):
+            tr.taskid = self.taskid
+        else:
+            tr.taskid = None
+        if isinstance(result, failure.Failure):
+            return (False, tr)
+        else:
+            return (True, tr)
+
+class ResultNS(object):
+    """
+    A dict like object for holding the results of a task.
+    
+    The result namespace object for use in `TaskResult` objects as tr.ns.
     It builds an object from a dictionary, such that it has attributes
     according to the key,value pairs of the dictionary.
     
     This works by calling setattr on ALL key,value pairs in the dict.  If a user
     chooses to overwrite the `__repr__` or `__getattr__` attributes, they can.
     This can be a bad idea, as it may corrupt standard behavior of the
     ns object.
     
     Example
     --------
     
     >>> ns = ResultNS({'a':17,'foo':range(3)})
-
     >>> print ns
-    NS{'a': 17, 'foo': [0, 1, 2]}
-
+        NS{'a':17,'foo':range(3)}
     >>> ns.a
-    17
-
+        17
     >>> ns['foo']
-    [0, 1, 2]
+        [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!")
+        raise Exception("the ns attribute cannot be changed")
     
     ns = property(_getNS, _setNS)
     
     def __init__(self, results, engineid):
         self.engineid = engineid
         if isinstance(results, failure.Failure):
             self.failure = results
             self.results = {}
         else:
             self.results = results
             self.failure = None
         
         self._ns = ResultNS(self.results)
         
         self.keys = self.results.keys()
     
     def __repr__(self):
         if self.failure is not None:
             contents = self.failure
         else:
             contents = self.results
         return "TaskResult[ID:%r]:%r"%(self.taskid, contents)
     
     def __getitem__(self, key):
         if self.failure is not None:
-            self.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 task.push is not None:
-            d.addCallback(lambda r: self.queuedEngine.push(task.push))
-        
-        d.addCallback(lambda r: self.queuedEngine.execute(task.expression))
-        
-        if task.pull is not None:
-            d.addCallback(lambda r: self.queuedEngine.pull(task.pull))
-        else:
-            d.addCallback(lambda r: None)
-        
-        def reseter(result):
-            self.queuedEngine.reset()
-            return result
-            
-        if task.clear_after:
-            d.addBoth(reseter)
-        
-        return d.addBoth(self._zipResults, task.pull, time.time(), time.localtime())
-    
-    def _zipResults(self, result, names, start, start_struct):
-        """Callback for construting the TaskResult object."""
-        if isinstance(result, failure.Failure):
-            tr = TaskResult(result, self.queuedEngine.id)
-        else:
-            if names is None:
-                resultDict = {} 
-            elif len(names) == 1:
-                resultDict = {names[0]:result}
-            else:
-                resultDict = dict(zip(names, result))
-            tr = TaskResult(resultDict, self.queuedEngine.id)
-        # the time info
-        tr.submitted = time.strftime(time_format, start_struct)
-        tr.completed = time.strftime(time_format)
-        tr.duration = time.time()-start
-        return tr
-    
+        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.uncanTask(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.canTask(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/engineservicetest.py

@@ -1,373 +1,372 b''
 # encoding: utf-8
 
 """Test template for complete engine object"""
 
 __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 twisted.internet import defer, reactor
 from twisted.python import failure
 from twisted.application import service
 import zope.interface as zi
 
 from IPython.kernel import newserialized
 from IPython.kernel import error
 from IPython.kernel.pickleutil import can, uncan
 import IPython.kernel.engineservice as es
 from IPython.kernel.core.interpreter import Interpreter
 from IPython.testing.parametric import Parametric, parametric
 
 #-------------------------------------------------------------------------------
 # Tests
 #-------------------------------------------------------------------------------
 
     
 # A sequence of valid commands run through execute
 validCommands = ['a=5',
                  'b=10',
                  'a=5; b=10; c=a+b',
                  'import math; 2.0*math.pi',
                  """def f():
     result = 0.0
     for i in range(10):
         result += i
 """,
                  'if 1<2: a=5',
                  """import time
 time.sleep(0.1)""",
                 """from math import cos;
 x = 1.0*cos(0.5)""", # Semicolons lead to Discard ast nodes that should be discarded
                 """from sets import Set
 s = Set()
    """, # Trailing whitespace should be allowed.
                 """import math
 math.cos(1.0)""", # Test a method call with a discarded return value
                 """x=1.0234
 a=5; b=10""", # Test an embedded semicolon
                 """x=1.0234
 a=5; b=10;""" # Test both an embedded and trailing semicolon
                  ]
                  
 # A sequence of commands that raise various exceptions
 invalidCommands = [('a=1/0',ZeroDivisionError),
                    ('print v',NameError),
                    ('l=[];l[0]',IndexError),
                    ("d={};d['a']",KeyError),
                    ("assert 1==0",AssertionError),
                    ("import abababsdbfsbaljasdlja",ImportError),
                    ("raise Exception()",Exception)]
 
 def testf(x):
     return 2.0*x
 
 globala = 99
 
 def testg(x):
     return  globala*x
 
 class IEngineCoreTestCase(object):
     """Test an IEngineCore implementer."""
 
     def createShell(self):
         return Interpreter()
 
     def catchQueueCleared(self, f):
         try:
             f.raiseException()
         except error.QueueCleared:
             pass
         
     def testIEngineCoreInterface(self):
         """Does self.engine claim to implement IEngineCore?"""
         self.assert_(es.IEngineCore.providedBy(self.engine))
         
     def testIEngineCoreInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngineCore."""
         for m in list(es.IEngineCore):
             self.assert_(hasattr(self.engine, m))
     
     def testIEngineCoreDeferreds(self):
         d = self.engine.execute('a=5')
         d.addCallback(lambda _: self.engine.pull('a'))
         d.addCallback(lambda _: self.engine.get_result())
         d.addCallback(lambda _: self.engine.keys())
         d.addCallback(lambda _: self.engine.push(dict(a=10)))
         return d
     
     def runTestExecute(self, cmd):
         self.shell = Interpreter()
         actual = self.shell.execute(cmd)
         def compare(computed):
             actual['id'] = computed['id']
             self.assertEquals(actual, computed)
         d = self.engine.execute(cmd)
         d.addCallback(compare)
         return d
     
     @parametric
     def testExecute(cls):
         return [(cls.runTestExecute, cmd) for cmd in validCommands]
 
     def runTestExecuteFailures(self, cmd, exc):
         def compare(f):
             self.assertRaises(exc, f.raiseException)
         d = self.engine.execute(cmd)
         d.addErrback(compare)
         return d
     
     @parametric
     def testExecuteFailures(cls):
         return [(cls.runTestExecuteFailures, cmd, exc) for cmd, exc in invalidCommands]
     
     def runTestPushPull(self, o):
         d = self.engine.push(dict(a=o))
         d.addCallback(lambda r: self.engine.pull('a'))
         d.addCallback(lambda r: self.assertEquals(o,r))
         return d
     
     @parametric
     def testPushPull(cls):
         objs = [10,"hi there",1.2342354,{"p":(1,2)},None]
         return [(cls.runTestPushPull, o) for o in objs]
         
     def testPullNameError(self):
         d = self.engine.push(dict(a=5))
         d.addCallback(lambda _:self.engine.reset())
         d.addCallback(lambda _: self.engine.pull("a"))
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         return d        
     
     def testPushPullFailures(self):
         d = self.engine.pull('a')
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         d.addCallback(lambda _: self.engine.execute('l = lambda x: x'))
         d.addCallback(lambda _: self.engine.pull('l'))
         d.addErrback(lambda f: self.assertRaises(pickle.PicklingError, f.raiseException))
         d.addCallback(lambda _: self.engine.push(dict(l=lambda x: x)))
         d.addErrback(lambda f: self.assertRaises(pickle.PicklingError, f.raiseException))
         return d
         
     def testPushPullArray(self):
         try:
             import numpy
         except:
-            print 'no numpy, ',
             return
         a = numpy.random.random(1000)
         d = self.engine.push(dict(a=a))
         d.addCallback(lambda _: self.engine.pull('a'))
         d.addCallback(lambda b: b==a)
         d.addCallback(lambda c: c.all())
         return self.assertDeferredEquals(d, True)
         
     def testPushFunction(self):
                     
         d = self.engine.push_function(dict(f=testf))
         d.addCallback(lambda _: self.engine.execute('result = f(10)'))
         d.addCallback(lambda _: self.engine.pull('result'))
         d.addCallback(lambda r: self.assertEquals(r, testf(10)))
         return d
 
     def testPullFunction(self):
         d = self.engine.push_function(dict(f=testf, g=testg))
         d.addCallback(lambda _: self.engine.pull_function(('f','g')))
         d.addCallback(lambda r: self.assertEquals(r[0](10), testf(10)))
         return d
         
     def testPushFunctionGlobal(self):
         """Make sure that pushed functions pick up the user's namespace for globals."""
         d = self.engine.push(dict(globala=globala))
         d.addCallback(lambda _: self.engine.push_function(dict(g=testg)))
         d.addCallback(lambda _: self.engine.execute('result = g(10)'))
         d.addCallback(lambda _: self.engine.pull('result'))
         d.addCallback(lambda r: self.assertEquals(r, testg(10)))
         return d
         
     def testGetResultFailure(self):
         d = self.engine.get_result(None)
         d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
         d.addCallback(lambda _: self.engine.get_result(10))
         d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
         return d
 
     def runTestGetResult(self, cmd):
         self.shell = Interpreter()
         actual = self.shell.execute(cmd)
         def compare(computed):
             actual['id'] = computed['id']
             self.assertEquals(actual, computed)
         d = self.engine.execute(cmd)
         d.addCallback(lambda r: self.engine.get_result(r['number']))
         d.addCallback(compare)
         return d
     
     @parametric
     def testGetResult(cls):
         return [(cls.runTestGetResult, cmd) for cmd in validCommands]
     
     def testGetResultDefault(self):
         cmd = 'a=5'
         shell = self.createShell()
         shellResult = shell.execute(cmd)
         def popit(dikt, key):
             dikt.pop(key)
             return dikt
         d = self.engine.execute(cmd)
         d.addCallback(lambda _: self.engine.get_result())
         d.addCallback(lambda r: self.assertEquals(shellResult, popit(r,'id')))
         return d
 
     def testKeys(self):
         d = self.engine.keys()
         d.addCallback(lambda s: isinstance(s, list))
         d.addCallback(lambda r: self.assertEquals(r, True))
         return d
             
 Parametric(IEngineCoreTestCase)
             
 class IEngineSerializedTestCase(object):
     """Test an IEngineCore implementer."""
         
     def testIEngineSerializedInterface(self):
         """Does self.engine claim to implement IEngineCore?"""
         self.assert_(es.IEngineSerialized.providedBy(self.engine))
         
     def testIEngineSerializedInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngireCore."""
         for m in list(es.IEngineSerialized):
             self.assert_(hasattr(self.engine, m))
        
     def testIEngineSerializedDeferreds(self):
         dList = []
         d = self.engine.push_serialized(dict(key=newserialized.serialize(12345)))
         self.assert_(isinstance(d, defer.Deferred))
         dList.append(d)
         d = self.engine.pull_serialized('key')
         self.assert_(isinstance(d, defer.Deferred))
         dList.append(d)
         D = defer.DeferredList(dList)
         return D
                 
     def testPushPullSerialized(self):
         objs = [10,"hi there",1.2342354,{"p":(1,2)}]
         d = defer.succeed(None)
         for o in objs:
             self.engine.push_serialized(dict(key=newserialized.serialize(o)))
             value = self.engine.pull_serialized('key')
             value.addCallback(lambda serial: newserialized.IUnSerialized(serial).getObject())
             d = self.assertDeferredEquals(value,o,d)
         return d
 
     def testPullSerializedFailures(self):
         d = self.engine.pull_serialized('a')
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         d.addCallback(lambda _: self.engine.execute('l = lambda x: x'))
         d.addCallback(lambda _: self.engine.pull_serialized('l'))
         d.addErrback(lambda f: self.assertRaises(pickle.PicklingError, f.raiseException))
         return d
 
 Parametric(IEngineSerializedTestCase)
 
 class IEngineQueuedTestCase(object):
     """Test an IEngineQueued implementer."""
         
     def testIEngineQueuedInterface(self):
         """Does self.engine claim to implement IEngineQueued?"""
         self.assert_(es.IEngineQueued.providedBy(self.engine))
         
     def testIEngineQueuedInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngireQueued."""
         for m in list(es.IEngineQueued):
             self.assert_(hasattr(self.engine, m))
             
     def testIEngineQueuedDeferreds(self): 
         dList = []
         d = self.engine.clear_queue()
         self.assert_(isinstance(d, defer.Deferred))
         dList.append(d)
         d = self.engine.queue_status()
         self.assert_(isinstance(d, defer.Deferred))
         dList.append(d)
         D = defer.DeferredList(dList)
         return D
             
     def testClearQueue(self):
         result = self.engine.clear_queue()
         d1 = self.assertDeferredEquals(result, None)
         d1.addCallback(lambda _: self.engine.queue_status())
         d2 = self.assertDeferredEquals(d1, {'queue':[], 'pending':'None'})
         return d2
         
     def testQueueStatus(self):
         result = self.engine.queue_status()
         result.addCallback(lambda r: 'queue' in r and 'pending' in r)
         d = self.assertDeferredEquals(result, True)
         return d
 
 Parametric(IEngineQueuedTestCase)
 
 class IEnginePropertiesTestCase(object):
     """Test an IEngineProperties implementor."""
     
     def testIEnginePropertiesInterface(self):
         """Does self.engine claim to implement IEngineProperties?"""
         self.assert_(es.IEngineProperties.providedBy(self.engine))
     
     def testIEnginePropertiesInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngireProperties."""
         for m in list(es.IEngineProperties):
             self.assert_(hasattr(self.engine, m))
     
     def testGetSetProperties(self):
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d = self.engine.set_properties(dikt)
         d.addCallback(lambda r: self.engine.get_properties())
         d = self.assertDeferredEquals(d, dikt)
         d.addCallback(lambda r: self.engine.get_properties(('c',)))
         d = self.assertDeferredEquals(d, {'c': dikt['c']})
         d.addCallback(lambda r: self.engine.set_properties(dict(c=False)))
         d.addCallback(lambda r: self.engine.get_properties(('c', 'd')))
         d = self.assertDeferredEquals(d, dict(c=False, d=None))
         return d
     
     def testClearProperties(self):
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d = self.engine.set_properties(dikt)
         d.addCallback(lambda r: self.engine.clear_properties())
         d.addCallback(lambda r: self.engine.get_properties())
         d = self.assertDeferredEquals(d, {})
         return d
     
     def testDelHasProperties(self):
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d = self.engine.set_properties(dikt)
         d.addCallback(lambda r: self.engine.del_properties(('b','e')))
         d.addCallback(lambda r: self.engine.has_properties(('a','b','c','d','e')))
         d = self.assertDeferredEquals(d, [True, False, True, True, False])
         return d
     
     def testStrictDict(self):
         s = """from IPython.kernel.engineservice import get_engine
 p = get_engine(%s).properties"""%self.engine.id
         d = self.engine.execute(s)
         d.addCallback(lambda r: self.engine.execute("p['a'] = lambda _:None"))
         d = self.assertDeferredRaises(d, error.InvalidProperty)
         d.addCallback(lambda r: self.engine.execute("p['a'] = range(5)"))
         d.addCallback(lambda r: self.engine.execute("p['a'].append(5)"))
         d.addCallback(lambda r: self.engine.get_properties('a'))
         d = self.assertDeferredEquals(d, dict(a=range(5)))
         return d
         
 Parametric(IEnginePropertiesTestCase)

IPython/kernel/tests/multienginetest.py

@@ -1,838 +1,828 b''
 # 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
 #-------------------------------------------------------------------------------
 
 from twisted.internet import defer
 
 from IPython.kernel import engineservice as es
 from IPython.kernel import multiengine as me
 from IPython.kernel import newserialized
 from IPython.kernel.error import NotDefined
 from IPython.testing import util
 from IPython.testing.parametric import parametric, Parametric
 from IPython.kernel import newserialized
 from IPython.kernel.util import printer
 from IPython.kernel.error import (InvalidEngineID,
     NoEnginesRegistered,
     CompositeError,
     InvalidDeferredID)
 from IPython.kernel.tests.engineservicetest import validCommands, invalidCommands
 from IPython.kernel.core.interpreter import Interpreter
 
 
 #-------------------------------------------------------------------------------
 # Base classes and utilities
 #-------------------------------------------------------------------------------
 
 class IMultiEngineBaseTestCase(object):
     """Basic utilities for working with multiengine tests.
 
     Some subclass should define:
 
     * self.multiengine
     * self.engines to keep track of engines for clean up"""
 
     def createShell(self):
         return Interpreter()
 
     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)
 
 
 def testf(x):
     return 2.0*x
 
 
 globala = 99
 
 
 def testg(x):
     return  globala*x
 
 
 def isdid(did):
     if not isinstance(did, str):
         return False
     if not len(did)==40:
         return False
     return True
 
 
 def _raise_it(f):
     try:
         f.raiseException()
     except CompositeError, e:
         e.raise_exception()
 
 #-------------------------------------------------------------------------------
 # IMultiEngineTestCase
 #-------------------------------------------------------------------------------
 
 class IMultiEngineTestCase(IMultiEngineBaseTestCase):
     """A test for any object that implements IEngineMultiplexer.
 
     self.multiengine must be defined and implement IEngineMultiplexer.
     """
 
     def testIMultiEngineInterface(self):
         """Does self.engine claim to implement IEngineCore?"""
         self.assert_(me.IEngineMultiplexer.providedBy(self.multiengine))
         self.assert_(me.IMultiEngine.providedBy(self.multiengine))
 
     def testIEngineMultiplexerInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngineCore."""
         for m in list(me.IEngineMultiplexer):
             self.assert_(hasattr(self.multiengine, m))
 
     def testIEngineMultiplexerDeferreds(self):
         self.addEngine(1)
         d= self.multiengine.execute('a=5', targets=0)
         d.addCallback(lambda _: self.multiengine.push(dict(a=5),targets=0))
         d.addCallback(lambda _: self.multiengine.push(dict(a=5, b='asdf', c=[1,2,3]),targets=0))
         d.addCallback(lambda _: self.multiengine.pull(('a','b','c'),targets=0))
         d.addCallback(lambda _: self.multiengine.get_result(targets=0))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.keys(targets=0))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(a=newserialized.serialize(10)),targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a',targets=0))
         d.addCallback(lambda _: self.multiengine.clear_queue(targets=0))
         d.addCallback(lambda _: self.multiengine.queue_status(targets=0))
         return d
 
     def testInvalidEngineID(self):
          self.addEngine(1)
          badID = 100
          d = self.multiengine.execute('a=5', targets=badID)
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.push(dict(a=5), targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.pull('a', targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.reset(targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.keys(targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.push_serialized(dict(a=newserialized.serialize(10)), targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.queue_status(targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          return d
 
     def testNoEnginesRegistered(self):
         badID = 'all'
         d= self.multiengine.execute('a=5', targets=badID)
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.push(dict(a=5), targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.pull('a', targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_result(targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.reset(targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.keys(targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(a=newserialized.serialize(10)), targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.queue_status(targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         return d
 
     def runExecuteAll(self, d, cmd, shell):
         actual = shell.execute(cmd)
         d.addCallback(lambda _: self.multiengine.execute(cmd))
         def compare(result):
             for r in result:
                 actual['id'] = r['id']
                 self.assertEquals(r, actual)
         d.addCallback(compare)
 
     def testExecuteAll(self):
         self.addEngine(4)
         d= defer.Deferred()
         shell = Interpreter()
         for cmd in validCommands:
                 self.runExecuteAll(d, cmd, shell)
         d.callback(None)
         return d
 
     # The following two methods show how to do parametrized
     # tests.  This is really slick!  Same is used above.
     def runExecuteFailures(self, cmd, exc):
         self.addEngine(4)
         d= self.multiengine.execute(cmd)
         d.addErrback(lambda f: self.assertRaises(exc, _raise_it, f))
         return d
 
     @parametric
     def testExecuteFailures(cls):
         return [(cls.runExecuteFailures,cmd,exc) for
                 cmd,exc in invalidCommands]
 
     def testPushPull(self):
         self.addEngine(1)
         objs = [10,"hi there",1.2342354,{"p":(1,2)}]
         d= self.multiengine.push(dict(key=objs[0]), targets=0)
         d.addCallback(lambda _: self.multiengine.pull('key', targets=0))
         d.addCallback(lambda r: self.assertEquals(r, [objs[0]]))
         d.addCallback(lambda _: self.multiengine.push(dict(key=objs[1]), targets=0))
         d.addCallback(lambda _: self.multiengine.pull('key', targets=0))
         d.addCallback(lambda r: self.assertEquals(r, [objs[1]]))
         d.addCallback(lambda _: self.multiengine.push(dict(key=objs[2]), targets=0))
         d.addCallback(lambda _: self.multiengine.pull('key', targets=0))
         d.addCallback(lambda r: self.assertEquals(r, [objs[2]]))
         d.addCallback(lambda _: self.multiengine.push(dict(key=objs[3]), targets=0))
         d.addCallback(lambda _: self.multiengine.pull('key', targets=0))
         d.addCallback(lambda r: self.assertEquals(r, [objs[3]]))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.pull('a', targets=0))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         d.addCallback(lambda _: self.multiengine.push(dict(a=10,b=20)))
         d.addCallback(lambda _: self.multiengine.pull(('a','b')))
         d.addCallback(lambda r: self.assertEquals(r, [[10,20]]))
         return d
 
     def testPushPullAll(self):
         self.addEngine(4)
         d= self.multiengine.push(dict(a=10))
         d.addCallback(lambda _: self.multiengine.pull('a'))
         d.addCallback(lambda r: self.assert_(r==[10,10,10,10]))
         d.addCallback(lambda _: self.multiengine.push(dict(a=10, b=20)))
         d.addCallback(lambda _: self.multiengine.pull(('a','b')))
         d.addCallback(lambda r: self.assert_(r==4*[[10,20]]))
         d.addCallback(lambda _: self.multiengine.push(dict(a=10, b=20), targets=0))
         d.addCallback(lambda _: self.multiengine.pull(('a','b'), targets=0))
         d.addCallback(lambda r: self.assert_(r==[[10,20]]))
         d.addCallback(lambda _: self.multiengine.push(dict(a=None, b=None), targets=0))
         d.addCallback(lambda _: self.multiengine.pull(('a','b'), targets=0))
         d.addCallback(lambda r: self.assert_(r==[[None,None]]))
         return d
 
     def testPushPullSerialized(self):
         self.addEngine(1)
         objs = [10,"hi there",1.2342354,{"p":(1,2)}]
         d= self.multiengine.push_serialized(dict(key=newserialized.serialize(objs[0])), targets=0)
         d.addCallback(lambda _: self.multiengine.pull_serialized('key', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, objs[0]))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(key=newserialized.serialize(objs[1])), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('key', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, objs[1]))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(key=newserialized.serialize(objs[2])), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('key', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, objs[2]))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(key=newserialized.serialize(objs[3])), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('key', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, objs[3]))
         d.addCallback(lambda _: self.multiengine.push(dict(a=10,b=range(5)), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized(('a','b'), targets=0))
         d.addCallback(lambda serial: [newserialized.IUnSerialized(s).getObject() for s in serial[0]])
         d.addCallback(lambda r: self.assertEquals(r, [10, range(5)]))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         return d
 
         objs = [10,"hi there",1.2342354,{"p":(1,2)}]
         d= defer.succeed(None)
         for o in objs:
             self.multiengine.push_serialized(0, key=newserialized.serialize(o))
             value = self.multiengine.pull_serialized(0, 'key')
             value.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
             d = self.assertDeferredEquals(value,o,d)
         return d
 
     def runGetResultAll(self, d, cmd, shell):
         actual = shell.execute(cmd)
         d.addCallback(lambda _: self.multiengine.execute(cmd))
         d.addCallback(lambda _: self.multiengine.get_result())
         def compare(result):
             for r in result:
                 actual['id'] = r['id']
                 self.assertEquals(r, actual)
         d.addCallback(compare)
 
     def testGetResultAll(self):
         self.addEngine(4)
         d= defer.Deferred()
         shell = Interpreter()
         for cmd in validCommands:
                 self.runGetResultAll(d, cmd, shell)
         d.callback(None)
         return d
 
     def testGetResultDefault(self):
         self.addEngine(1)
         target = 0
         cmd = 'a=5'
         shell = self.createShell()
         shellResult = shell.execute(cmd)
         def popit(dikt, key):
             dikt.pop(key)
             return dikt
         d= self.multiengine.execute(cmd, targets=target)
         d.addCallback(lambda _: self.multiengine.get_result(targets=target))
         d.addCallback(lambda r: self.assertEquals(shellResult, popit(r[0],'id')))
         return d
 
     def testGetResultFailure(self):
         self.addEngine(1)
         d= self.multiengine.get_result(None, targets=0)
         d.addErrback(lambda f: self.assertRaises(IndexError, _raise_it, f))
         d.addCallback(lambda _: self.multiengine.get_result(10, targets=0))
         d.addErrback(lambda f: self.assertRaises(IndexError, _raise_it, f))
         return d
 
     def testPushFunction(self):
         self.addEngine(1)
         d= self.multiengine.push_function(dict(f=testf), targets=0)
         d.addCallback(lambda _: self.multiengine.execute('result = f(10)', targets=0))
         d.addCallback(lambda _: self.multiengine.pull('result', targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], testf(10)))
         d.addCallback(lambda _: self.multiengine.push(dict(globala=globala), targets=0))
         d.addCallback(lambda _: self.multiengine.push_function(dict(g=testg), targets=0))
         d.addCallback(lambda _: self.multiengine.execute('result = g(10)', targets=0))
         d.addCallback(lambda _: self.multiengine.pull('result', targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], testg(10)))
         return d
 
     def testPullFunction(self):
         self.addEngine(1)
         d= self.multiengine.push(dict(a=globala), targets=0)
         d.addCallback(lambda _: self.multiengine.push_function(dict(f=testf), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_function('f', targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0](10), testf(10)))
         d.addCallback(lambda _: self.multiengine.execute("def g(x): return x*x", targets=0))
         d.addCallback(lambda _: self.multiengine.pull_function(('f','g'),targets=0))
         d.addCallback(lambda r: self.assertEquals((r[0][0](10),r[0][1](10)), (testf(10), 100)))
         return d
 
     def testPushFunctionAll(self):
         self.addEngine(4)
         d= self.multiengine.push_function(dict(f=testf))
         d.addCallback(lambda _: self.multiengine.execute('result = f(10)'))
         d.addCallback(lambda _: self.multiengine.pull('result'))
         d.addCallback(lambda r: self.assertEquals(r, 4*[testf(10)]))
         d.addCallback(lambda _: self.multiengine.push(dict(globala=globala)))
         d.addCallback(lambda _: self.multiengine.push_function(dict(testg=testg)))
         d.addCallback(lambda _: self.multiengine.execute('result = testg(10)'))
         d.addCallback(lambda _: self.multiengine.pull('result'))
         d.addCallback(lambda r: self.assertEquals(r, 4*[testg(10)]))
         return d
 
     def testPullFunctionAll(self):
         self.addEngine(4)
         d= self.multiengine.push_function(dict(f=testf))
         d.addCallback(lambda _: self.multiengine.pull_function('f'))
         d.addCallback(lambda r: self.assertEquals([func(10) for func in r], 4*[testf(10)]))
         return d
 
     def testGetIDs(self):
         self.addEngine(1)
         d= self.multiengine.get_ids()
         d.addCallback(lambda r: self.assertEquals(r, [0]))
         d.addCallback(lambda _: self.addEngine(3))
         d.addCallback(lambda _: self.multiengine.get_ids())
         d.addCallback(lambda r: self.assertEquals(r, [0,1,2,3]))
         return d
 
     def testClearQueue(self):
         self.addEngine(4)
         d= self.multiengine.clear_queue()
         d.addCallback(lambda r: self.assertEquals(r,4*[None]))
         return d
 
     def testQueueStatus(self):
         self.addEngine(4)
         d= self.multiengine.queue_status(targets=0)
         d.addCallback(lambda r: self.assert_(isinstance(r[0],tuple)))
         return d
 
     def testGetSetProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.get_properties())
         d.addCallback(lambda r: self.assertEquals(r, 4*[dikt]))
         d.addCallback(lambda r: self.multiengine.get_properties(('c',)))
         d.addCallback(lambda r: self.assertEquals(r, 4*[{'c': dikt['c']}]))
         d.addCallback(lambda r: self.multiengine.set_properties(dict(c=False)))
         d.addCallback(lambda r: self.multiengine.get_properties(('c', 'd')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[dict(c=False, d=None)]))
         return d
 
     def testClearProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.clear_properties())
         d.addCallback(lambda r: self.multiengine.get_properties())
         d.addCallback(lambda r: self.assertEquals(r, 4*[{}]))
         return d
 
     def testDelHasProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.del_properties(('b','e')))
         d.addCallback(lambda r: self.multiengine.has_properties(('a','b','c','d','e')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[True, False, True, True, False]]))
         return d
 
 Parametric(IMultiEngineTestCase)
 
 #-------------------------------------------------------------------------------
 # ISynchronousMultiEngineTestCase
 #-------------------------------------------------------------------------------
 
 class ISynchronousMultiEngineTestCase(IMultiEngineBaseTestCase):
 
     def testISynchronousMultiEngineInterface(self):
         """Does self.engine claim to implement IEngineCore?"""
         self.assert_(me.ISynchronousEngineMultiplexer.providedBy(self.multiengine))
         self.assert_(me.ISynchronousMultiEngine.providedBy(self.multiengine))
 
     def testExecute(self):
         self.addEngine(4)
         execute = self.multiengine.execute
         d= execute('a=5', targets=0, block=True)
         d.addCallback(lambda r: self.assert_(len(r)==1))
         d.addCallback(lambda _: execute('b=10'))
         d.addCallback(lambda r: self.assert_(len(r)==4))
         d.addCallback(lambda _: execute('c=30', block=False))
         d.addCallback(lambda did: self.assert_(isdid(did)))
         d.addCallback(lambda _: execute('d=[0,1,2]', block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assert_(len(r)==4))
         return d
 
     def testPushPull(self):
         data = dict(a=10, b=1.05, c=range(10), d={'e':(1,2),'f':'hi'})
         self.addEngine(4)
         push = self.multiengine.push
         pull = self.multiengine.pull
         d= push({'data':data}, targets=0)
         d.addCallback(lambda r: pull('data', targets=0))
         d.addCallback(lambda r: self.assertEqual(r,[data]))
         d.addCallback(lambda _: push({'data':data}))
         d.addCallback(lambda r: pull('data'))
         d.addCallback(lambda r: self.assertEqual(r,4*[data]))
         d.addCallback(lambda _: push({'data':data}, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: pull('data', block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEqual(r,4*[data]))
         d.addCallback(lambda _: push(dict(a=10,b=20)))
         d.addCallback(lambda _: pull(('a','b')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[10,20]]))
         return d
 
     def testPushPullFunction(self):
         self.addEngine(4)
         pushf = self.multiengine.push_function
         pullf = self.multiengine.pull_function
         push = self.multiengine.push
         pull = self.multiengine.pull
         execute = self.multiengine.execute
         d= pushf({'testf':testf}, targets=0)
         d.addCallback(lambda r: pullf('testf', targets=0))
         d.addCallback(lambda r: self.assertEqual(r[0](1.0), testf(1.0)))
         d.addCallback(lambda _: execute('r = testf(10)', targets=0))
         d.addCallback(lambda _: pull('r', targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], testf(10)))
         d.addCallback(lambda _: pushf({'testf':testf}, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: pullf('testf', block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEqual(r[0](1.0), testf(1.0)))
         d.addCallback(lambda _: execute("def g(x): return x*x", targets=0))
         d.addCallback(lambda _: pullf(('testf','g'),targets=0))
         d.addCallback(lambda r: self.assertEquals((r[0][0](10),r[0][1](10)), (testf(10), 100)))
         return d
 
     def testGetResult(self):
         shell = Interpreter()
         result1 = shell.execute('a=10')
         result1['id'] = 0
         result2 = shell.execute('b=20')
         result2['id'] = 0
         execute= self.multiengine.execute
         get_result = self.multiengine.get_result
         self.addEngine(1)
         d= execute('a=10')
         d.addCallback(lambda _: get_result())
         d.addCallback(lambda r: self.assertEquals(r[0], result1))
         d.addCallback(lambda _: execute('b=20'))
         d.addCallback(lambda _: get_result(1))
         d.addCallback(lambda r: self.assertEquals(r[0], result1))
         d.addCallback(lambda _: get_result(2, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r[0], result2))
         return d
 
     def testResetAndKeys(self):
         self.addEngine(1)
 
         #Blocking mode
         d= self.multiengine.push(dict(a=10, b=20, c=range(10)), targets=0)
         d.addCallback(lambda _: self.multiengine.keys(targets=0))
         def keys_found(keys):
             self.assert_('a' in keys[0])
             self.assert_('b' in keys[0])
             self.assert_('b' in keys[0])
         d.addCallback(keys_found)
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.keys(targets=0))
         def keys_not_found(keys):
             self.assert_('a' not in keys[0])
             self.assert_('b' not in keys[0])
             self.assert_('b' not in keys[0])
         d.addCallback(keys_not_found)
 
         #Non-blocking mode
         d.addCallback(lambda _: self.multiengine.push(dict(a=10, b=20, c=range(10)), targets=0))
         d.addCallback(lambda _: self.multiengine.keys(targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         def keys_found(keys):
             self.assert_('a' in keys[0])
             self.assert_('b' in keys[0])
             self.assert_('b' in keys[0])
         d.addCallback(keys_found)
         d.addCallback(lambda _: self.multiengine.reset(targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: self.multiengine.keys(targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         def keys_not_found(keys):
             self.assert_('a' not in keys[0])
             self.assert_('b' not in keys[0])
             self.assert_('b' not in keys[0])
         d.addCallback(keys_not_found)
 
         return d
 
     def testPushPullSerialized(self):
         self.addEngine(1)
         dikt = dict(a=10,b='hi there',c=1.2345,d={'p':(1,2)})
         sdikt = {}
         for k,v in dikt.iteritems():
             sdikt[k] = newserialized.serialize(v)
         d= self.multiengine.push_serialized(dict(a=sdikt['a']), targets=0)
         d.addCallback(lambda _: self.multiengine.pull('a',targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], dikt['a']))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, dikt['a']))
         d.addCallback(lambda _: self.multiengine.push_serialized(sdikt, targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized(sdikt.keys(), targets=0))
         d.addCallback(lambda serial: [newserialized.IUnSerialized(s).getObject() for s in serial[0]])
         d.addCallback(lambda r: self.assertEquals(r, dikt.values()))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
 
         #Non-blocking mode
         d.addCallback(lambda r: self.multiengine.push_serialized(dict(a=sdikt['a']), targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: self.multiengine.pull('a',targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], dikt['a']))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, dikt['a']))
         d.addCallback(lambda _: self.multiengine.push_serialized(sdikt, targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: self.multiengine.pull_serialized(sdikt.keys(), targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda serial: [newserialized.IUnSerialized(s).getObject() for s in serial[0]])
         d.addCallback(lambda r: self.assertEquals(r, dikt.values()))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         return d
 
     def testClearQueue(self):
         self.addEngine(4)
         d= self.multiengine.clear_queue()
         d.addCallback(lambda r: self.multiengine.clear_queue(block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r,4*[None]))
         return d
 
     def testQueueStatus(self):
         self.addEngine(4)
         d= self.multiengine.queue_status(targets=0)
         d.addCallback(lambda r: self.assert_(isinstance(r[0],tuple)))
         d.addCallback(lambda r: self.multiengine.queue_status(targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assert_(isinstance(r[0],tuple)))
         return d
 
     def testGetIDs(self):
         self.addEngine(1)
         d= self.multiengine.get_ids()
         d.addCallback(lambda r: self.assertEquals(r, [0]))
         d.addCallback(lambda _: self.addEngine(3))
         d.addCallback(lambda _: self.multiengine.get_ids())
         d.addCallback(lambda r: self.assertEquals(r, [0,1,2,3]))
         return d
 
     def testGetSetProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.get_properties())
         d.addCallback(lambda r: self.assertEquals(r, 4*[dikt]))
         d.addCallback(lambda r: self.multiengine.get_properties(('c',)))
         d.addCallback(lambda r: self.assertEquals(r, 4*[{'c': dikt['c']}]))
         d.addCallback(lambda r: self.multiengine.set_properties(dict(c=False)))
         d.addCallback(lambda r: self.multiengine.get_properties(('c', 'd')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[dict(c=False, d=None)]))
 
         #Non-blocking
         d.addCallback(lambda r: self.multiengine.set_properties(dikt, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.get_properties(block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[dikt]))
         d.addCallback(lambda r: self.multiengine.get_properties(('c',), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[{'c': dikt['c']}]))
         d.addCallback(lambda r: self.multiengine.set_properties(dict(c=False), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.get_properties(('c', 'd'), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[dict(c=False, d=None)]))
         return d
 
     def testClearProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.clear_properties())
         d.addCallback(lambda r: self.multiengine.get_properties())
         d.addCallback(lambda r: self.assertEquals(r, 4*[{}]))
 
         #Non-blocking
         d.addCallback(lambda r: self.multiengine.set_properties(dikt, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.clear_properties(block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.get_properties(block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[{}]))
         return d
 
     def testDelHasProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.del_properties(('b','e')))
         d.addCallback(lambda r: self.multiengine.has_properties(('a','b','c','d','e')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[True, False, True, True, False]]))
 
         #Non-blocking
         d.addCallback(lambda r: self.multiengine.set_properties(dikt, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.del_properties(('b','e'), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.has_properties(('a','b','c','d','e'), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[True, False, True, True, False]]))
         return d
 
     def test_clear_pending_deferreds(self):
         self.addEngine(4)
         did_list = []
         d= self.multiengine.execute('a=10',block=False)
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.push(dict(b=10),block=False))
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.pull(('a','b'),block=False))
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.clear_pending_deferreds())
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[0],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[1],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[2],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         return d
 
 #-------------------------------------------------------------------------------
 # Coordinator test cases
 #-------------------------------------------------------------------------------
 
 class IMultiEngineCoordinatorTestCase(object):
 
     def testScatterGather(self):
         self.addEngine(4)
         d= self.multiengine.scatter('a', range(16))
         d.addCallback(lambda r: self.multiengine.gather('a'))
         d.addCallback(lambda r: self.assertEquals(r, range(16)))
         d.addCallback(lambda _: self.multiengine.gather('asdf'))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         return d
 
     def testScatterGatherNumpy(self):
         try:
             import numpy
             from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         except:
             return
         else:
             self.addEngine(4)
             a = numpy.arange(16)
             d = self.multiengine.scatter('a', a)
             d.addCallback(lambda r: self.multiengine.gather('a'))
             d.addCallback(lambda r: assert_array_equal(r, a))
             return d
 
     def testMap(self):
         self.addEngine(4)
         def f(x):
             return x**2
         data = range(16)
         d= self.multiengine.map(f, data)
         d.addCallback(lambda r: self.assertEquals(r,[f(x) for x in data]))
         return d
 
 
 class ISynchronousMultiEngineCoordinatorTestCase(IMultiEngineCoordinatorTestCase):
 
     def testScatterGatherNonblocking(self):
         self.addEngine(4)
         d= self.multiengine.scatter('a', range(16), block=False)
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.gather('a', block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, range(16)))
         return d
-
+    
     def testScatterGatherNumpyNonblocking(self):
         try:
             import numpy
             from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         except:
             return
         else:
             self.addEngine(4)
             a = numpy.arange(16)
             d = self.multiengine.scatter('a', a, block=False)
             d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
             d.addCallback(lambda r: self.multiengine.gather('a', block=False))
             d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
             d.addCallback(lambda r: assert_array_equal(r, a))
             return d
-
-    def testMapNonblocking(self):
-        self.addEngine(4)
-        def f(x):
-            return x**2
-        data = range(16)
-        d= self.multiengine.map(f, data, block=False)
-        d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
-        d.addCallback(lambda r: self.assertEquals(r,[f(x) for x in data]))
-        return d
-
+    
     def test_clear_pending_deferreds(self):
         self.addEngine(4)
         did_list = []
         d= self.multiengine.scatter('a',range(16),block=False)
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.gather('a',block=False))
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.map(lambda x: x, range(16),block=False))
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.clear_pending_deferreds())
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[0],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[1],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[2],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         return d
 
 #-------------------------------------------------------------------------------
 # Extras test cases
 #-------------------------------------------------------------------------------
 
 class IMultiEngineExtrasTestCase(object):
 
     def testZipPull(self):
         self.addEngine(4)
         d= self.multiengine.push(dict(a=10,b=20))
         d.addCallback(lambda r: self.multiengine.zip_pull(('a','b')))
         d.addCallback(lambda r: self.assert_(r, [4*[10],4*[20]]))
         return d
 
     def testRun(self):
         self.addEngine(4)
         import tempfile
         fname = tempfile.mktemp('foo.py')
         f= open(fname, 'w')
         f.write('a = 10\nb=30')
         f.close()
         d= self.multiengine.run(fname)
         d.addCallback(lambda r: self.multiengine.pull(('a','b')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[10,30]]))
         return d
 
 
 class ISynchronousMultiEngineExtrasTestCase(IMultiEngineExtrasTestCase):
 
     def testZipPullNonblocking(self):
         self.addEngine(4)
         d= self.multiengine.push(dict(a=10,b=20))
         d.addCallback(lambda r: self.multiengine.zip_pull(('a','b'), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assert_(r, [4*[10],4*[20]]))
         return d
 
     def testRunNonblocking(self):
         self.addEngine(4)
         import tempfile
         fname = tempfile.mktemp('foo.py')
         f= open(fname, 'w')
         f.write('a = 10\nb=30')
         f.close()
         d= self.multiengine.run(fname, block=False)
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.pull(('a','b')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[10,30]]))
         return d
 
 
 #-------------------------------------------------------------------------------
 # IFullSynchronousMultiEngineTestCase
 #-------------------------------------------------------------------------------
 
 class IFullSynchronousMultiEngineTestCase(ISynchronousMultiEngineTestCase,
     ISynchronousMultiEngineCoordinatorTestCase,
     ISynchronousMultiEngineExtrasTestCase):
     pass

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_enginefc.py

@@ -1,92 +1,92 b''
 # encoding: utf-8
 
 """This file contains unittests for the enginepb.py module."""
 
 __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:
     from twisted.python import components
     from twisted.internet import reactor, defer
     from twisted.spread import pb
     from twisted.internet.base import DelayedCall
     DelayedCall.debug = True
 
     import zope.interface as zi
 
     from IPython.kernel.fcutil import Tub, UnauthenticatedTub
     from IPython.kernel import engineservice as es
     from IPython.testing.util import DeferredTestCase
     from IPython.kernel.controllerservice import IControllerBase
     from IPython.kernel.enginefc import FCRemoteEngineRefFromService, IEngineBase
     from IPython.kernel.engineservice import IEngineQueued
     from IPython.kernel.engineconnector import EngineConnector
     
     from IPython.kernel.tests.engineservicetest import \
         IEngineCoreTestCase, \
         IEngineSerializedTestCase, \
         IEngineQueuedTestCase
 except ImportError:
     print "we got an error!!!"
-    pass
+    raise
 else:
     class EngineFCTest(DeferredTestCase, 
                        IEngineCoreTestCase, 
                        IEngineSerializedTestCase,
                        IEngineQueuedTestCase
                        ):
         
         zi.implements(IControllerBase)
         
         def setUp(self):
         
             # Start a server and append to self.servers
             self.controller_reference = FCRemoteEngineRefFromService(self)
             self.controller_tub = Tub()
             self.controller_tub.listenOn('tcp:10105:interface=127.0.0.1')
             self.controller_tub.setLocation('127.0.0.1:10105')
             
             furl = self.controller_tub.registerReference(self.controller_reference)
             self.controller_tub.startService()
             
             # Start an EngineService and append to services/client
             self.engine_service = es.EngineService()
             self.engine_service.startService()
             self.engine_tub = Tub()
             self.engine_tub.startService()
             engine_connector = EngineConnector(self.engine_tub)
             d = engine_connector.connect_to_controller(self.engine_service, furl)
             # This deferred doesn't fire until after register_engine has returned and
             # thus, self.engine has been defined and the tets can proceed.
             return d
         
         def tearDown(self):
             dlist = []
             # Shut down the engine
             d = self.engine_tub.stopService()
             dlist.append(d)
             # Shut down the controller
             d = self.controller_tub.stopService()
             dlist.append(d)
             return defer.DeferredList(dlist)
         
         #---------------------------------------------------------------------------
         # Make me look like a basic controller
         #---------------------------------------------------------------------------
         
         def register_engine(self, engine_ref, id=None, ip=None, port=None, pid=None):
             self.engine = IEngineQueued(IEngineBase(engine_ref))
             return {'id':id}
         
         def unregister_engine(self, id):
             pass
  No newline at end of file

IPython/kernel/tests/test_multienginefc.py

@@ -1,70 +1,144 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:
     from twisted.internet import defer, reactor
 
     from IPython.kernel.fcutil import Tub, UnauthenticatedTub
 
     from IPython.testing.util import DeferredTestCase
     from IPython.kernel.controllerservice import ControllerService
     from IPython.kernel.multiengine import IMultiEngine
     from IPython.kernel.tests.multienginetest import IFullSynchronousMultiEngineTestCase
     from IPython.kernel.multienginefc import IFCSynchronousMultiEngine
     from IPython.kernel import multiengine as me
     from IPython.kernel.clientconnector import ClientConnector
+    from IPython.kernel.parallelfunction import ParallelFunction
+    from IPython.kernel.error import CompositeError
+    from IPython.kernel.util import printer
 except ImportError:
     pass
 else:
+    
+    def _raise_it(f):
+        try:
+            f.raiseException()
+        except CompositeError, e:
+            e.raise_exception()
+    
+    
     class FullSynchronousMultiEngineTestCase(DeferredTestCase, IFullSynchronousMultiEngineTestCase):
     
         def setUp(self):
         
             self.engines = []
                 
             self.controller = ControllerService()
             self.controller.startService()
             self.imultiengine = IMultiEngine(self.controller)
             self.mec_referenceable = IFCSynchronousMultiEngine(self.imultiengine)
 
             self.controller_tub = Tub()
             self.controller_tub.listenOn('tcp:10105:interface=127.0.0.1')
             self.controller_tub.setLocation('127.0.0.1:10105')
         
             furl = self.controller_tub.registerReference(self.mec_referenceable)
             self.controller_tub.startService()
         
             self.client_tub = ClientConnector()
             d = self.client_tub.get_multiengine_client(furl)
             d.addCallback(self.handle_got_client)
             return d
         
         def handle_got_client(self, client):
             self.multiengine = 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(4)
+            m = self.multiengine.mapper()
+            self.assertEquals(m.multiengine,self.multiengine)
+            self.assertEquals(m.dist,'b')
+            self.assertEquals(m.targets,'all')
+            self.assertEquals(m.block,True)
+        
+        def test_map_default(self):
+            self.addEngine(4)
+            m = self.multiengine.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.multiengine.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(4)
+            m = self.multiengine.mapper(block=False)
+            d = m.map(lambda x: 2*x, range(10))
+            d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+                
+        def test_mapper_fail(self):
+            self.addEngine(4)
+            m = self.multiengine.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(4)
+            p = self.multiengine.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.multiengine.parallel(block=False)
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 2*x
+            d = f(range(10))
+            d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+        
+        def test_parallel_fail(self):
+            self.addEngine(4)
+            p = self.multiengine.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

IPython/kernel/tests/test_pendingdeferred.py

@@ -1,186 +1,186 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 """Tests for pendingdeferred.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
 #-------------------------------------------------------------------------------
 
 try:
     from twisted.internet import defer
     from twisted.python import failure
 
-    from IPython.testing import tcommon
-    #from IPython.testing.tcommon import *
     from IPython.testing.util import DeferredTestCase
     import IPython.kernel.pendingdeferred as pd
     from IPython.kernel import error
     from IPython.kernel.util import printer
 except ImportError:
     pass
 else:
+    
+    class Foo(object):
+    
+        def bar(self, bahz):
+            return defer.succeed('blahblah: %s' % bahz)
 
     class TwoPhaseFoo(pd.PendingDeferredManager):
     
         def __init__(self, foo):
             self.foo = foo
             pd.PendingDeferredManager.__init__(self)
 
         @pd.two_phase
         def bar(self, bahz):
             return self.foo.bar(bahz)
 
     class PendingDeferredManagerTest(DeferredTestCase):
     
         def setUp(self):
             self.pdm = pd.PendingDeferredManager()
         
         def tearDown(self):
             pass
     
         def testBasic(self):
             dDict = {}
             # Create 10 deferreds and save them
             for i in range(10):
                 d = defer.Deferred()
                 did = self.pdm.save_pending_deferred(d)
                 dDict[did] = d
             # Make sure they are begin saved
             for k in dDict.keys():
                 self.assert_(self.pdm.quick_has_id(k))
             # Get the pending deferred (block=True), then callback with 'foo' and compare
             for did in dDict.keys()[0:5]:
                 d = self.pdm.get_pending_deferred(did,block=True)
                 dDict[did].callback('foo')
                 d.addCallback(lambda r: self.assert_(r=='foo'))
             # Get the pending deferreds with (block=False) and make sure ResultNotCompleted is raised
             for did in dDict.keys()[5:10]:
                 d = self.pdm.get_pending_deferred(did,block=False)
                 d.addErrback(lambda f: self.assertRaises(error.ResultNotCompleted, f.raiseException))
             # Now callback the last 5, get them and compare.
             for did in dDict.keys()[5:10]:
                 dDict[did].callback('foo')
                 d = self.pdm.get_pending_deferred(did,block=False)
                 d.addCallback(lambda r: self.assert_(r=='foo'))
     
         def test_save_then_delete(self):
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             self.assert_(self.pdm.quick_has_id(did))
             self.pdm.delete_pending_deferred(did)
             self.assert_(not self.pdm.quick_has_id(did))
     
         def test_save_get_delete(self):
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,True)
             d2.addErrback(lambda f: self.assertRaises(error.AbortedPendingDeferredError, f.raiseException))
             self.pdm.delete_pending_deferred(did)
             return d2
    
         def test_double_get(self):
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,True)
             d3 = self.pdm.get_pending_deferred(did,True)
             d3.addErrback(lambda f: self.assertRaises(error.InvalidDeferredID, f.raiseException))
     
         def test_get_after_callback(self):
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d.callback('foo')
             d2 = self.pdm.get_pending_deferred(did,True)
             d2.addCallback(lambda r: self.assertEquals(r,'foo'))
             self.assert_(not self.pdm.quick_has_id(did))
 
         def test_get_before_callback(self):
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,True)
             d.callback('foo')
             d2.addCallback(lambda r: self.assertEquals(r,'foo'))
             self.assert_(not self.pdm.quick_has_id(did))
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,True)
             d2.addCallback(lambda r: self.assertEquals(r,'foo'))
             d.callback('foo')
             self.assert_(not self.pdm.quick_has_id(did))
     
         def test_get_after_errback(self):
             class MyError(Exception):
                 pass
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d.errback(failure.Failure(MyError('foo')))
             d2 = self.pdm.get_pending_deferred(did,True)
             d2.addErrback(lambda f: self.assertRaises(MyError, f.raiseException))
             self.assert_(not self.pdm.quick_has_id(did))
 
         def test_get_before_errback(self):
             class MyError(Exception):
                 pass
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,True)
             d.errback(failure.Failure(MyError('foo')))
             d2.addErrback(lambda f: self.assertRaises(MyError, f.raiseException))
             self.assert_(not self.pdm.quick_has_id(did))
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,True)
             d2.addErrback(lambda f: self.assertRaises(MyError, f.raiseException))
             d.errback(failure.Failure(MyError('foo')))
             self.assert_(not self.pdm.quick_has_id(did))
         
         def test_noresult_noblock(self):
             d = defer.Deferred()
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,False)
             d2.addErrback(lambda f: self.assertRaises(error.ResultNotCompleted, f.raiseException))
 
         def test_with_callbacks(self):
             d = defer.Deferred()
             d.addCallback(lambda r: r+' foo')
             d.addCallback(lambda r: r+' bar')
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,True)
             d.callback('bam')
             d2.addCallback(lambda r: self.assertEquals(r,'bam foo bar'))
     
         def test_with_errbacks(self):
             class MyError(Exception):
                 pass
             d = defer.Deferred()
             d.addCallback(lambda r: 'foo')
             d.addErrback(lambda f: 'caught error')
             did = self.pdm.save_pending_deferred(d)
             d2 = self.pdm.get_pending_deferred(did,True)
             d.errback(failure.Failure(MyError('bam')))
             d2.addErrback(lambda f: self.assertRaises(MyError, f.raiseException))
     
         def test_nested_deferreds(self):
             d = defer.Deferred()
             d2 = defer.Deferred()
             d.addCallback(lambda r: d2)
             did = self.pdm.save_pending_deferred(d)
             d.callback('foo')
             d3 = self.pdm.get_pending_deferred(did,False)
             d3.addErrback(lambda f: self.assertRaises(error.ResultNotCompleted, f.raiseException))
             d2.callback('bar')
             d3 = self.pdm.get_pending_deferred(did,False)
             d3.addCallback(lambda r: self.assertEquals(r,'bar'))
 
-
-# Global object expected by Twisted's trial
-testSuite = lambda : makeTestSuite(__name__,dt_files,dt_modules)

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/fetchparse.py

@@ -1,90 +1,90 b''
 """
 An exceptionally lousy site spider
 Ken Kinder <[e-mail]>
 
 This module gives an example of how the TaskClient interface to the 
 IPython controller works.  Before running this script start the IPython controller
 and some engines using something like::
 
     ipcluster -n 4
 """
 from twisted.python.failure import Failure
 from IPython.kernel import client
 import time
 
 fetchParse = """
 from twisted.web import microdom
 import urllib2
 import urlparse
 
 def fetchAndParse(url, data=None):
     links = []
     try:
         page = urllib2.urlopen(url, data=data)
     except Exception:
         return links
     else:
         if page.headers.type == 'text/html':
             doc = microdom.parseString(page.read(), beExtremelyLenient=True)
             for node in doc.getElementsByTagName('a'):
                 if node.getAttribute('href'):
                     links.append(urlparse.urljoin(url, node.getAttribute('href')))
         return links
 """
 
 class DistributedSpider(object):
     
     # Time to wait between polling for task results.
     pollingDelay = 0.5
     
     def __init__(self, site):
         self.tc = client.TaskClient()
         self.rc = client.MultiEngineClient()
         self.rc.execute(fetchParse)
         
         self.allLinks = []
         self.linksWorking = {}
         self.linksDone = {}
         
         self.site = site
         
     def visitLink(self, url):
         if url not in self.allLinks:
             self.allLinks.append(url)
             if url.startswith(self.site):
                 print '    ', url
-                self.linksWorking[url] = self.tc.run(client.Task('links = fetchAndParse(url)', pull=['links'], push={'url': url}))
+                self.linksWorking[url] = self.tc.run(client.StringTask('links = fetchAndParse(url)', pull=['links'], push={'url': url}))
         
     def onVisitDone(self, result, url):
         print url, ':'
         self.linksDone[url] = None
         del self.linksWorking[url]
         if isinstance(result.failure, Failure):
             txt = result.failure.getTraceback()
             for line in txt.split('\n'):
                 print '    ', line
         else:
             for link in result.ns.links:
                 self.visitLink(link)
                 
     def run(self):
         self.visitLink(self.site)
         while self.linksWorking:
             print len(self.linksWorking), 'pending...'
             self.synchronize()
             time.sleep(self.pollingDelay)
     
     def synchronize(self):
         for url, taskId in self.linksWorking.items():
             # Calling get_task_result with block=False will return None if the
             # task is not done yet.  This provides a simple way of polling.
             result = self.tc.get_task_result(taskId, block=False)
             if result is not None:
                 self.onVisitDone(result, url)
 
 def main():
     distributedSpider = DistributedSpider(raw_input('Enter site to crawl: '))
     distributedSpider.run()
 
 if __name__ == '__main__':
     main()

docs/examples/kernel/helloworld.py

@@ -1,14 +1,14 b''
 """
 A Distributed Hello world
 Ken Kinder <[e-mail]>
 """
 from IPython.kernel import client
 
 tc = client.TaskClient()
 mec = client.MultiEngineClient()
 
 mec.execute('import time')
-hello_taskid = tc.run(client.Task('time.sleep(3) ; word = "Hello,"', pull=('word')))
-world_taskid = tc.run(client.Task('time.sleep(3) ; word = "World!"', pull=('word')))
+hello_taskid = tc.run(client.StringTask('time.sleep(3) ; word = "Hello,"', pull=('word')))
+world_taskid = tc.run(client.StringTask('time.sleep(3) ; word = "World!"', pull=('word')))
 print "Submitted tasks:", hello_taskid, world_taskid
 print tc.get_task_result(hello_taskid,block=True).ns.word, tc.get_task_result(world_taskid,block=True).ns.word

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()
+tr1.raise_exception()
 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()

docs/source/changes.txt

@@ -1,173 +1,195 b''
 .. _changes:
 
 ==========
 What's new
 ==========
 
 .. contents::
 
 Release 0.9
 ===========
 
 New features
 ------------
 
+	* The notion of a task has been completely reworked.  An `ITask` interface has
+	  been created.  This interface defines the methods that tasks need to implement.
+	  These methods are now responsible for things like submitting tasks and processing
+	  results.  There are two basic task types: :class:`IPython.kernel.task.StringTask`
+	  (this is the old `Task` object, but renamed) and the new 
+	  :class:`IPython.kernel.task.MapTask`, which is based on a function.
+	* A new interface, :class:`IPython.kernel.mapper.IMapper` has been defined to
+	  standardize the idea of a `map` method.  This interface has a single
+	  `map` method that has the same syntax as the built-in `map`.  We have also defined
+	  a `mapper` factory interface that creates objects that implement
+	  :class:`IPython.kernel.mapper.IMapper` for different controllers.  Both
+	  the multiengine and task controller now have mapping capabilties.
+	* The parallel function capabilities have been reworks.  The major changes are that
+	  i) there is now an `@parallel` magic that creates parallel functions, ii)
+	  the syntax for mulitple variable follows that of `map`, iii) both the
+	  multiengine and task controller now have a parallel function implementation.
 	* All of the parallel computing capabilities from `ipython1-dev` have been merged into
 	  IPython proper.  This resulted in the following new subpackages:  
 	  :mod:`IPython.kernel`, :mod:`IPython.kernel.core`, :mod:`IPython.config`,
 	  :mod:`IPython.tools` and :mod:`IPython.testing`.
 	* As part of merging in the `ipython1-dev` stuff, the `setup.py` script and friends
 	  have been completely refactored.  Now we are checking for dependencies using
 	  the approach that matplotlib uses.
 	* The documentation has been completely reorganized to accept the documentation
 	  from `ipython1-dev`.
 	* We have switched to using Foolscap for all of our network protocols in
 	  :mod:`IPython.kernel`.  This gives us secure connections that are both encrypted
 	  and authenticated.
 	* We have a brand new `COPYING.txt` files that describes the IPython license
 	  and copyright.  The biggest change is that we are putting "The IPython
 	  Development Team" as the copyright holder.  We give more details about exactly
 	  what this means in this file.  All developer should read this and use the new
 	  banner in all IPython source code files.
 	* sh profile: ./foo runs foo as system command, no need to do !./foo anymore
 	* String lists now support 'sort(field, nums = True)' method (to easily
 	  sort system command output). Try it with 'a = !ls -l ; a.sort(1, nums=1)'
 	* '%cpaste foo' now assigns the pasted block as string list, instead of string
 	* The ipcluster script now run by default with no security.  This is done because
 	  the main usage of the script is for starting things on localhost.  Eventually
 	  when ipcluster is able to start things on other hosts, we will put security
 	  back.
 
-	
-
 Bug fixes
 ---------
 
+	* The colors escapes in the multiengine client are now turned off on win32 as they
+	  don't print correctly.
 	* The :mod:`IPython.kernel.scripts.ipengine` script was exec'ing mpi_import_statement
 	  incorrectly, which was leading the engine to crash when mpi was enabled.
 	* A few subpackages has missing `__init__.py` files.
 	* The documentation is only created is Sphinx is found.  Previously, the `setup.py`
 	  script would fail if it was missing.
 
 Backwards incompatible changes
 ------------------------------
 
+	* :class:`IPython.kernel.client.Task` has been renamed
+	  :class:`IPython.kernel.client.StringTask` to make way for new task types.
+	* The keyword argument `style` has been renamed `dist` in `scatter`, `gather`
+	  and `map`.
+	* Renamed the values that the rename `dist` keyword argument can have from
+	  `'basic'` to `'b'`.
 	* IPython has a larger set of dependencies if you want all of its capabilities.
 	  See the `setup.py` script for details.
 	* The constructors for :class:`IPython.kernel.client.MultiEngineClient` and 
 	  :class:`IPython.kernel.client.TaskClient` no longer take the (ip,port) tuple.
 	  Instead they take the filename of a file that contains the FURL for that
 	  client.  If the FURL file is in your IPYTHONDIR, it will be found automatically
 	  and the constructor can be left empty.
 	* The asynchronous clients in :mod:`IPython.kernel.asyncclient` are now created 
 	  using the factory functions :func:`get_multiengine_client` and 
 	  :func:`get_task_client`.  These return a `Deferred` to the actual client.
 	* The command line options to `ipcontroller` and `ipengine` have changed to
 	  reflect the new Foolscap network protocol and the FURL files.  Please see the
 	  help for these scripts for details.
 	* The configuration files for the kernel have changed because of the Foolscap stuff.
 	  If you were using custom config files before, you should delete them and regenerate
 	  new ones.
 
 Changes merged in from IPython1
 -------------------------------
 
 New features
 ............
 
 	* Much improved ``setup.py`` and ``setupegg.py`` scripts.  Because Twisted
 	  and zope.interface are now easy installable, we can declare them as dependencies
 	  in our setupegg.py script.
 	* IPython is now compatible with Twisted 2.5.0 and 8.x.
 	* Added a new example of how to use :mod:`ipython1.kernel.asynclient`.
 	* Initial draft of a process daemon in :mod:`ipython1.daemon`.  This has not
 	  been merged into IPython and is still in `ipython1-dev`.
 	* The ``TaskController`` now has methods for getting the queue status.
 	* The ``TaskResult`` objects not have information about how long the task
 	  took to run.
 	* We are attaching additional attributes to exceptions ``(_ipython_*)`` that 
 	  we use to carry additional info around.
 	* New top-level module :mod:`asyncclient` that has asynchronous versions (that
 	  return deferreds) of the client classes.  This is designed to users who want
 	  to run their own Twisted reactor
 	* All the clients in :mod:`client` are now based on Twisted.  This is done by 
 	  running the Twisted reactor in a separate thread and using the
 	  :func:`blockingCallFromThread` function that is in recent versions of Twisted.
 	* Functions can now be pushed/pulled to/from engines using
 	  :meth:`MultiEngineClient.push_function` and :meth:`MultiEngineClient.pull_function`.
 	* Gather/scatter are now implemented in the client to reduce the work load
 	  of the controller and improve performance.
 	* Complete rewrite of the IPython docuementation.  All of the documentation
 	  from the IPython website has been moved into docs/source as restructured
 	  text documents.  PDF and HTML documentation are being generated using 
 	  Sphinx.
 	* New developer oriented documentation: development guidelines and roadmap. 
 	* Traditional ``ChangeLog`` has been changed to a more useful ``changes.txt`` file
 	  that is organized by release and is meant to provide something more relevant
 	  for users.
 
 Bug fixes
 .........
 
 	* Created a proper ``MANIFEST.in`` file to create source distributions.
 	* Fixed a bug in the ``MultiEngine`` interface.  Previously, multi-engine 
 	  actions were being collected with a :class:`DeferredList` with 
 	  ``fireononeerrback=1``.  This meant that methods were returning 
 	  before all engines had given their results.  This was causing extremely odd 
 	  bugs in certain cases. To fix this problem, we have 1) set 
 	  ``fireononeerrback=0`` to make sure all results (or exceptions) are in 
 	  before returning and 2) introduced a :exc:`CompositeError` exception 
 	  that wraps all of the engine exceptions.  This is a huge change as it means 
 	  that users will have to catch :exc:`CompositeError` rather than the actual
 	  exception.
 
 Backwards incompatible changes
 ..............................
 
 	* All names have been renamed to conform to the lowercase_with_underscore
 	  convention.  This will require users to change references to all names like
 	  ``queueStatus`` to ``queue_status``.
 	* Previously, methods like :meth:`MultiEngineClient.push` and    	
 	  :meth:`MultiEngineClient.push` used ``*args`` and ``**kwargs``.  This was 
 	  becoming a problem as we weren't able to introduce new keyword arguments into
 	  the API.  Now these methods simple take a dict or sequence.  This has also allowed
 	  us to get rid of the ``*All`` methods like :meth:`pushAll` and :meth:`pullAll`.
 	  These things are now handled with the ``targets`` keyword argument that defaults
 	  to ``'all'``.
 	* The :attr:`MultiEngineClient.magicTargets` has been renamed to
 	  :attr:`MultiEngineClient.targets`. 
 	* All methods in the MultiEngine interface now accept the optional keyword argument
 	  ``block``.
 	* Renamed :class:`RemoteController` to :class:`MultiEngineClient` and 
 	  :class:`TaskController` to :class:`TaskClient`.
 	* Renamed the top-level module from :mod:`api` to :mod:`client`.
 	* Most methods in the multiengine interface now raise a :exc:`CompositeError` exception
 	  that wraps the user's exceptions, rather than just raising the raw user's exception.
 	* Changed the ``setupNS`` and ``resultNames`` in the ``Task`` class to ``push`` 
 	  and ``pull``.
 
 Release 0.8.4
 =============
 
 Someone needs to describe what went into 0.8.4.
 
 Release 0.8.2
 =============
 
 * %pushd/%popd behave differently; now "pushd /foo" pushes CURRENT directory 
   and jumps to /foo. The current behaviour is closer to the documented 
   behaviour, and should not trip anyone.
   
 Release 0.8.3
 =============
   
 * pydb is now disabled by default (due to %run -d problems). You can enable
   it by passing -pydb command line argument to IPython. Note that setting
   it in config file won't work.
 
 Older releases
 ==============
 
 Changes in earlier releases of IPython are described in the older file ``ChangeLog``.  
 Please refer to this document for details.
 

docs/source/install/index.txt

@@ -1,9 +1,11 b''
+.. _install_index:
+
 ==================
 Installation
 ==================
 
 .. toctree::
    :maxdepth: 2
 
    basic.txt
    advanced.txt

docs/source/overview.txt

@@ -1,189 +1,174 b''
 .. _overview:
 
 ============
 Introduction
 ============
 
-This is the official documentation for IPython 0.x series (i.e. what
-we are used to refer to just as "IPython"). The original text of the
-manual (most of which is still in place) has been authored by Fernando
-Perez, but as recommended usage patterns and new features have
-emerged, this manual has been updated to reflect that fact. Most of
-the additions have been authored by Ville M. Vainio.
-
-The manual has been generated from reStructuredText source markup with
-Sphinx, which should make it much easier to keep it up-to-date in the
-future. Some reST artifacts and bugs may still be apparent in the
-documentation, but this should improve as the toolchain matures.
-
 Overview
 ========
 
 One of Python's most useful features is its interactive interpreter.
 This system allows very fast testing of ideas without the overhead of
 creating test files as is typical in most programming languages.
 However, the interpreter supplied with the standard Python distribution
 is somewhat limited for extended interactive use.
 
-IPython is a free software project (released under the BSD license)
-which tries to:
+The goal of IPython is to create a comprehensive environment for
+interactive and exploratory computing.  To support, this goal, IPython
+has two main components:
+
+	* An enhanced interactive Python shell.
+	* An architecture for interactive parallel computing.
+
+All of IPython is open source (released under the revised BSD license).
+
+Enhanced interactive Python shell
+=================================
+
+IPython's interactive shell (`ipython`), has the following goals:
 
    1. Provide an interactive shell superior to Python's default. IPython
       has many features for object introspection, system shell access,
       and its own special command system for adding functionality when
       working interactively. It tries to be a very efficient environment
       both for Python code development and for exploration of problems
       using Python objects (in situations like data analysis).
    2. Serve as an embeddable, ready to use interpreter for your own
       programs. IPython can be started with a single call from inside
       another program, providing access to the current namespace. This
       can be very useful both for debugging purposes and for situations
       where a blend of batch-processing and interactive exploration are
       needed.
    3. Offer a flexible framework which can be used as the base
       environment for other systems with Python as the underlying
       language. Specifically scientific environments like Mathematica,
       IDL and Matlab inspired its design, but similar ideas can be
       useful in many fields.
    4. Allow interactive testing of threaded graphical toolkits. IPython
       has support for interactive, non-blocking control of GTK, Qt and
       WX applications via special threading flags. The normal Python
       shell can only do this for Tkinter applications.
 
-
-Main features
--------------
-
-* Dynamic object introspection. One can access docstrings, function
-  definition prototypes, source code, source files and other details
-  of any object accessible to the interpreter with a single
-  keystroke ('?', and using '??' provides additional detail).
-* Searching through modules and namespaces with '*' wildcards, both
-  when using the '?' system and via the %psearch command.
-* Completion in the local namespace, by typing TAB at the prompt.
-  This works for keywords, modules, methods, variables and files in the
-  current directory. This is supported via the readline library, and
-  full access to configuring readline's behavior is provided. 
-  Custom completers can be implemented easily for different purposes
-  (system commands, magic arguments etc.)
-* Numbered input/output prompts with command history (persistent
-  across sessions and tied to each profile), full searching in this
-  history and caching of all input and output.
-* User-extensible 'magic' commands. A set of commands prefixed with
-  % is available for controlling IPython itself and provides
-  directory control, namespace information and many aliases to
-  common system shell commands.
-* Alias facility for defining your own system aliases.
-* Complete system shell access. Lines starting with ! are passed
-  directly to the system shell, and using !! or var = !cmd 
-  captures shell output into python variables for further use.
-* Background execution of Python commands in a separate thread.
-  IPython has an internal job manager called jobs, and a
-  conveninence backgrounding magic function called %bg.
-* The ability to expand python variables when calling the system
-  shell. In a shell command, any python variable prefixed with $ is
-  expanded. A double $$ allows passing a literal $ to the shell (for
-  access to shell and environment variables like $PATH).
-* Filesystem navigation, via a magic %cd command, along with a
-  persistent bookmark system (using %bookmark) for fast access to
-  frequently visited directories.
-* A lightweight persistence framework via the %store command, which
-  allows you to save arbitrary Python variables. These get restored
-  automatically when your session restarts.
-* Automatic indentation (optional) of code as you type (through the
-  readline library).
-* Macro system for quickly re-executing multiple lines of previous
-  input with a single name. Macros can be stored persistently via
-  %store and edited via %edit.
-* Session logging (you can then later use these logs as code in your
-  programs). Logs can optionally timestamp all input, and also store
-  session output (marked as comments, so the log remains valid
-  Python source code).
-* Session restoring: logs can be replayed to restore a previous
-  session to the state where you left it.
-* Verbose and colored exception traceback printouts. Easier to parse
-  visually, and in verbose mode they produce a lot of useful
-  debugging information (basically a terminal version of the cgitb
-  module).
-* Auto-parentheses: callable objects can be executed without
-  parentheses: 'sin 3' is automatically converted to 'sin(3)'.
-* Auto-quoting: using ',' or ';' as the first character forces
-  auto-quoting of the rest of the line: ',my_function a b' becomes
-  automatically 'my_function("a","b")', while ';my_function a b'
-  becomes 'my_function("a b")'.
-* Extensible input syntax. You can define filters that pre-process
-  user input to simplify input in special situations. This allows
-  for example pasting multi-line code fragments which start with
-  '>>>' or '...' such as those from other python sessions or the
-  standard Python documentation.
-* Flexible configuration system. It uses a configuration file which
-  allows permanent setting of all command-line options, module
-  loading, code and file execution. The system allows recursive file
-  inclusion, so you can have a base file with defaults and layers
-  which load other customizations for particular projects.
-* Embeddable. You can call IPython as a python shell inside your own
-  python programs. This can be used both for debugging code or for
-  providing interactive abilities to your programs with knowledge
-  about the local namespaces (very useful in debugging and data
-  analysis situations).
-* Easy debugger access. You can set IPython to call up an enhanced
-  version of the Python debugger (pdb) every time there is an
-  uncaught exception. This drops you inside the code which triggered
-  the exception with all the data live and it is possible to
-  navigate the stack to rapidly isolate the source of a bug. The
-  %run magic command -with the -d option- can run any script under
-  pdb's control, automatically setting initial breakpoints for you.
-  This version of pdb has IPython-specific improvements, including
-  tab-completion and traceback coloring support. For even easier
-  debugger access, try %debug after seeing an exception. winpdb is 
-  also supported, see ipy_winpdb extension.
-* Profiler support. You can run single statements (similar to
-  profile.run()) or complete programs under the profiler's control.
-  While this is possible with standard cProfile or profile modules,
-  IPython wraps this functionality with magic commands (see '%prun'
-  and '%run -p') convenient for rapid interactive work.
-* Doctest support. The special %doctest_mode command toggles a mode
-  that allows you to paste existing doctests (with leading '>>>'
-  prompts and whitespace) and uses doctest-compatible prompts and
-  output, so you can use IPython sessions as doctest code.
-
+Main features of the interactive shell
+--------------------------------------
+
+	* Dynamic object introspection. One can access docstrings, function
+	  definition prototypes, source code, source files and other details
+	  of any object accessible to the interpreter with a single
+	  keystroke (:samp:`?`, and using :samp:`??` provides additional detail).
+	* Searching through modules and namespaces with :samp:`*` wildcards, both
+	  when using the :samp:`?` system and via the :samp:`%psearch` command.
+	* Completion in the local namespace, by typing :kbd:`TAB` at the prompt.
+	  This works for keywords, modules, methods, variables and files in the
+	  current directory. This is supported via the readline library, and
+	  full access to configuring readline's behavior is provided. 
+	  Custom completers can be implemented easily for different purposes
+	  (system commands, magic arguments etc.)
+	* Numbered input/output prompts with command history (persistent
+	  across sessions and tied to each profile), full searching in this
+	  history and caching of all input and output.
+	* User-extensible 'magic' commands. A set of commands prefixed with
+	  :samp:`%` is available for controlling IPython itself and provides
+	  directory control, namespace information and many aliases to
+	  common system shell commands.
+	* Alias facility for defining your own system aliases.
+	* Complete system shell access. Lines starting with :samp:`!` are passed
+	  directly to the system shell, and using :samp:`!!` or :samp:`var = !cmd` 
+	  captures shell output into python variables for further use.
+	* Background execution of Python commands in a separate thread.
+	  IPython has an internal job manager called jobs, and a
+	  conveninence backgrounding magic function called :samp:`%bg`.
+	* The ability to expand python variables when calling the system
+	  shell. In a shell command, any python variable prefixed with :samp:`$` is
+	  expanded. A double :samp:`$$` allows passing a literal :samp:`$` to the shell (for
+	  access to shell and environment variables like :envvar:`PATH`).
+	* Filesystem navigation, via a magic :samp:`%cd` command, along with a
+	  persistent bookmark system (using :samp:`%bookmark`) for fast access to
+	  frequently visited directories.
+	* A lightweight persistence framework via the :samp:`%store` command, which
+	  allows you to save arbitrary Python variables. These get restored
+	  automatically when your session restarts.
+	* Automatic indentation (optional) of code as you type (through the
+	  readline library).
+	* Macro system for quickly re-executing multiple lines of previous
+	  input with a single name. Macros can be stored persistently via
+	  :samp:`%store` and edited via :samp:`%edit`.
+	* Session logging (you can then later use these logs as code in your
+	  programs). Logs can optionally timestamp all input, and also store
+	  session output (marked as comments, so the log remains valid
+	  Python source code).
+	* Session restoring: logs can be replayed to restore a previous
+	  session to the state where you left it.
+	* Verbose and colored exception traceback printouts. Easier to parse
+	  visually, and in verbose mode they produce a lot of useful
+	  debugging information (basically a terminal version of the cgitb
+	  module).
+	* Auto-parentheses: callable objects can be executed without
+	  parentheses: :samp:`sin 3` is automatically converted to :samp:`sin(3)`.
+	* Auto-quoting: using :samp:`,`, or :samp:`;` as the first character forces
+	  auto-quoting of the rest of the line: :samp:`,my_function a b` becomes
+	  automatically :samp:`my_function("a","b")`, while :samp:`;my_function a b`
+	  becomes :samp:`my_function("a b")`.
+	* Extensible input syntax. You can define filters that pre-process
+	  user input to simplify input in special situations. This allows
+	  for example pasting multi-line code fragments which start with
+	  :samp:`>>>` or :samp:`...` such as those from other python sessions or the
+	  standard Python documentation.
+	* Flexible configuration system. It uses a configuration file which
+	  allows permanent setting of all command-line options, module
+	  loading, code and file execution. The system allows recursive file
+	  inclusion, so you can have a base file with defaults and layers
+	  which load other customizations for particular projects.
+	* Embeddable. You can call IPython as a python shell inside your own
+	  python programs. This can be used both for debugging code or for
+	  providing interactive abilities to your programs with knowledge
+	  about the local namespaces (very useful in debugging and data
+	  analysis situations).
+	* Easy debugger access. You can set IPython to call up an enhanced
+	  version of the Python debugger (pdb) every time there is an
+	  uncaught exception. This drops you inside the code which triggered
+	  the exception with all the data live and it is possible to
+	  navigate the stack to rapidly isolate the source of a bug. The
+	  :samp:`%run` magic command (with the :samp:`-d` option) can run any script under
+	  pdb's control, automatically setting initial breakpoints for you.
+	  This version of pdb has IPython-specific improvements, including
+	  tab-completion and traceback coloring support. For even easier
+	  debugger access, try :samp:`%debug` after seeing an exception. winpdb is 
+	  also supported, see ipy_winpdb extension.
+	* Profiler support. You can run single statements (similar to
+	  :samp:`profile.run()`) or complete programs under the profiler's control.
+	  While this is possible with standard cProfile or profile modules,
+	  IPython wraps this functionality with magic commands (see :samp:`%prun`
+	  and :samp:`%run -p`) convenient for rapid interactive work.
+	* Doctest support. The special :samp:`%doctest_mode` command toggles a mode
+	  that allows you to paste existing doctests (with leading :samp:`>>>`
+	  prompts and whitespace) and uses doctest-compatible prompts and
+	  output, so you can use IPython sessions as doctest code.
+
+Interactive parallel computing
+==============================
+
+Increasingly, parallel computer hardware, such as multicore CPUs, clusters and supercomputers, is becoming ubiquitous. Over the last 3 years, we have developed an
+architecture within IPython that allows such hardware to be used quickly and easily
+from Python. Moreover, this architecture is designed to support interactive and
+collaborative parallel computing.
+
+For more information, see our :ref:`overview <parallel_index>` of using IPython for
+parallel computing.
 
 Portability and Python requirements
 -----------------------------------
 
-Python requirements: IPython requires with Python version 2.3 or newer.
-If you are still using Python 2.2 and can not upgrade, the last version
-of IPython which worked with Python 2.2 was 0.6.15, so you will have to
-use that.
-
-IPython is developed under Linux, but it should work in any reasonable
-Unix-type system (tested OK under Solaris and the BSD family, for which
-a port exists thanks to Dryice Liu).
-
-Mac OS X: it works, apparently without any problems (thanks to Jim Boyle
-at Lawrence Livermore for the information). Thanks to Andrea Riciputi,
-Fink support is available.
-
-CygWin: it works mostly OK, though some users have reported problems
-with prompt coloring. No satisfactory solution to this has been found so
-far, you may want to disable colors permanently in the ipythonrc
-configuration file if you experience problems. If you have proper color
-support under cygwin, please post to the IPython mailing list so this
-issue can be resolved for all users.
-
-Windows: it works well under Windows Vista/XP/2k, and I suspect NT should
-behave similarly. Section "Installation under windows" describes
-installation details for Windows, including some additional tools needed
-on this platform.
-
-Windows 9x support is present, and has been reported to work fine (at
-least on WinME).
-
-Location
---------
-
-IPython is generously hosted at http://ipython.scipy.org by the
-Enthought, Inc and the SciPy project. This site offers downloads,
-subversion access, mailing lists and a bug tracking system. I am very
-grateful to Enthought (http://www.enthought.com) and all of the SciPy
-team for their contribution.
  No newline at end of file
+As of the 0.9 release, IPython requires Python 2.4 or greater.  We have
+not begun to test IPython on Python 2.6 or 3.0, but we expect it will
+work with some minor changes.
+
+IPython is known to work on the following operating systems:
+
+	* Linux
+	* AIX
+	* Most other Unix-like OSs (Solaris, BSD, etc.)
+	* Mac OS X
+	* Windows (CygWin, XP, Vista, etc.)
+
+See :ref:`here <install_index>` for instructions on how to install IPython.
  No newline at end of file

docs/source/parallel/index.txt

@@ -1,15 +1,17 b''
+.. _parallel_index:
+
 ====================================
 Using IPython for Parallel computing
 ====================================
 
 User Documentation
 ==================
 
 .. toctree::
    :maxdepth: 2
 
    parallel_intro.txt
    parallel_multiengine.txt
    parallel_task.txt
    parallel_mpi.txt