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/kernel/asyncclient.py

@@ -27,7 +27,7 b' 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
 
 #-------------------------------------------------------------------------------

IPython/kernel/client.py

@@ -44,7 +44,7 b' 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
 
 #-------------------------------------------------------------------------------

IPython/kernel/contexts.py

@@ -141,46 +141,3 b' class RemoteMultiEngine(RemoteContextBase):'
     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

@@ -79,7 +79,7 b" def magic_px(self,parameter_s=''):"
     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
 

IPython/kernel/map.py

@@ -115,7 +115,7 b' class RoundRobinMap(Map):'
         #    result.append(concat[i:totalLength:maxPartitionLength])
         return self.concatenate(listOfPartitions)
         
-styles = {'basic':Map}
+dists = {'b':Map}
 
     
     

IPython/kernel/multiengine.py

@@ -653,67 +653,55 b' components.registerAdapter(SynchronousMultiEngine, IMultiEngine, ISynchronousMul'
 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, dist='b', targets='all'):
+        """Gather object key from targets."""
         
-    def gather(key, style='basic', 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'):
         """
+        A parallelized version of Python's builtin `map` function.
     
-    def map(func, seq, style='basic', targets='all'):
-        """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.
         
-        This function implements the following pattern:
+        The equivalence is:
         
-        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.
+        raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
                 
-        :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
-        =======
-        
-        >>> rc.mapAll('lambda x: x*x', range(10000))
-        [0,2,4,9,25,36,...]
+        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`.
+        """
 
 
 #-------------------------------------------------------------------------------
@@ -722,46 +710,31 b' class ISynchronousMultiEngineCoordinator(IMultiEngineCoordinator):'
 
 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

IPython/kernel/multiengineclient.py

@@ -31,6 +31,11 b' 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,
@@ -186,6 +191,10 b' class ResultList(list):'
     
     def __repr__(self):
         output = []
+        # 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
@@ -295,27 +304,6 b' class InteractiveMultiEngineClient(object):'
         """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
     #---------------------------------------------------------------------------
@@ -415,7 +403,11 b' class FullBlockingMultiEngineClient(InteractiveMultiEngineClient):'
     engine, run code on it, etc.
     """
     
-    implements(IFullBlockingMultiEngineClient)
+    implements(
+        IFullBlockingMultiEngineClient,
+        IMultiEngineMapperFactory,
+        IMapper
+    )
     
     def __init__(self, smultiengine):
         self.smultiengine = smultiengine
@@ -772,29 +764,100 b' class FullBlockingMultiEngineClient(InteractiveMultiEngineClient):'
     # 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.
+        
+        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.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):
         """
-        A parallelized version of Python's builtin map
+        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)
-        return self._blockFromThread(self.smultiengine.map, func, seq, 
-            style, targets=targets, block=block)
+        mapper = self.mapper(dist, targets, block)
+        pf = ParallelFunction(mapper)
+        return pf
     
     #---------------------------------------------------------------------------
     # IMultiEngineExtras

IPython/kernel/multienginefc.py

@@ -29,6 +29,12 b' 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,
@@ -280,7 +286,12 b' components.registerAdapter(FCSynchronousMultiEngineFromMultiEngine,'
 
 class FCFullSynchronousMultiEngineClient(object):
     
-    implements(IFullSynchronousMultiEngine, IBlockingClientAdaptor)
+    implements(
+        IFullSynchronousMultiEngine, 
+        IBlockingClientAdaptor,
+        IMultiEngineMapperFactory,
+        IMapper
+    )
     
     def __init__(self, remote_reference):
         self.remote_reference = remote_reference
@@ -475,7 +486,7 b' class FCFullSynchronousMultiEngineClient(object):'
         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 
@@ -483,7 +494,7 b' class FCFullSynchronousMultiEngineClient(object):'
         # 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.
@@ -541,7 +552,7 b' class FCFullSynchronousMultiEngineClient(object):'
         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 
@@ -549,7 +560,7 b' class FCFullSynchronousMultiEngineClient(object):'
         # 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.
@@ -604,25 +615,103 b' class FCFullSynchronousMultiEngineClient(object):'
         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
     #---------------------------------------------------------------------------

IPython/kernel/parallelfunction.py

@@ -16,17 +16,92 b' __docformat__ = "restructuredtext en"'
 #-------------------------------------------------------------------------------
 
 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_function(*sequences):
+            return self.mapper.map(self.func, *sequences)
+        return call_function
+
         
  No newline at end of file
-    def __call__(self, sequence):
-        return self.multiengine.map(self.func, sequence, targets=self.targets, block=self.block)
  No newline at end of file

IPython/kernel/task.py

@@ -5,117 +5,404 b''
 
 __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
-
-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.
+class ITask(zi.Interface):
+    """
+    This interface provides a generic definition of what constitutes a task.
+    
+    There are two sides to a task.  First a task needs to take input from
+    a user to determine what work is performed by the task.  Second, the
+    task needs to have the logic that knows how to turn that information
+    info specific calls to a worker, through the `IQueuedEngine` interface.
+    
+    Many method in this class get two things passed to them: a Deferred
+    and an IQueuedEngine implementer.  Such methods should register callbacks
+    on the Deferred that use the IQueuedEngine to accomplish something.  See
+    the existing task objects for examples.
+    """
+    
+    zi.Attribute('retries','How many times to retry the task')
+    zi.Attribute('recovery_task','A task to try if the initial one fails')
+    zi.Attribute('taskid','the id of the task')
+    
+    def start_time(result):
+        """
+        Do anything needed to start the timing of the task.
+        
+        Must simply return the result after starting the timers.
+        """
+    
+    def stop_time(result):
+        """
+        Do anything needed to stop the timing of the task.
+        
+        Must simply return the result after stopping the timers.  This 
+        method will usually set attributes that are used by `process_result`
+        in building result of the task.
+        """
+    
+    def pre_task(d, queued_engine):
+        """Do something with the queued_engine before the task is run.
+        
+        This method should simply add callbacks to the input Deferred
+        that do something with the `queued_engine` before the task is run.
+        
+        :Parameters:
+            d : Deferred
+                The deferred that actions should be attached to
+            queued_engine : IQueuedEngine implementer
+                The worker that has been allocated to perform the task
+        """
+    
+    def post_task(d, queued_engine):
+        """Do something with the queued_engine after the task is run.
+        
+        This method should simply add callbacks to the input Deferred
+        that do something with the `queued_engine` before the task is run.
+        
+        :Parameters:
+            d : Deferred
+                The deferred that actions should be attached to
+            queued_engine : IQueuedEngine implementer
+                The worker that has been allocated to perform the task
+        """
+    
+    def submit_task(d, queued_engine):
+        """Submit a task using the `queued_engine` we have been allocated.
+        
+        When a task is ready to run, this method is called.  This method
+        must take the internal information of the task and make suitable
+        calls on the queued_engine to have the actual work done.
+        
+        This method should simply add callbacks to the input Deferred
+        that do something with the `queued_engine` before the task is run.
+        
+        :Parameters:
+            d : Deferred
+                The deferred that actions should be attached to
+            queued_engine : IQueuedEngine implementer
+                The worker that has been allocated to perform the task
+        """
+
+    def process_result(d, result, engine_id):
+        """Take a raw task result.
+        
+        Objects that implement `ITask` can choose how the result of running
+        the task is presented.  This method takes the raw result and 
+        does this logic.  Two example are the `MapTask` which simply returns
+        the raw result or a `Failure` object and the `StringTask` which
+        returns a `TaskResult` object.
+        
+        :Parameters:
+            d : Deferred
+                The deferred that actions should be attached to
+            result : object
+                The raw task result that needs to be wrapped
+            engine_id : int
+                The id of the engine that did the task
+        
+        :Returns:
+            The result, as a tuple of the form: (success, result).
+            Here, success is a boolean indicating if the task
+            succeeded or failed and result is the result.
+        """
+        
+    def check_depend(properties):
+        """Check properties to see if the task should be run.
+        
+        :Parameters:
+            properties : dict
+                A dictionary of properties that an engine has set
+        
+        :Returns:
+            True if the task should be run, False otherwise
+        """
+    
+    def can_task(self):
+        """Serialize (can) any functions in the task for pickling.
+        
+        Subclasses must override this method and make sure that all 
+        functions in the task are canned by calling `can` on the 
+        function.
+        """
+    
+    def uncan_task(self):
+        """Unserialize (uncan) any canned function in the task."""
 
-    The user should create instances of this class to represent a task that 
-    needs to be done.
+class BaseTask(object):
+    """
+    Common fuctionality for all objects implementing `ITask`.
+    """
+    
+    zi.implements(ITask)
+    
+    def __init__(self, clear_before=False, clear_after=False, retries=0,
+            recovery_task=None, depend=None):
+        """
+        Make a generic task.
     
     :Parameters:
-        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.
+                Should the engines namespace be cleared before the task
+                is run
         clear_after : boolean 
-            Should the engine's namespace be cleared after the task is run.
-            Default=False.
+                Should the engines namespace be clear after the task is run
         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
-    --------
+                The number of times a task should be retries upon failure
+            recovery_task : any task object
+                If a task fails and it has a recovery_task, that is run
+                upon a retry
+            depend : FunctionType
+                A function that is called to test for properties.  This function
+                must take one argument, the properties dict and return a boolean
+        """
+        self.clear_before = clear_before
+        self.clear_after = clear_after
+        self.retries = retries
+        self.recovery_task = recovery_task
+        self.depend = depend
+        self.taskid = None
+    
+    def start_time(self, result):
+        """
+        Start the basic timers.
+        """
+        self.start = time.time()
+        self.start_struct = time.localtime()
+        return result
+    
+    def stop_time(self, result):
+        """
+        Stop the basic timers.
+        """
+        self.stop = time.time()
+        self.stop_struct = time.localtime()
+        self.duration = self.stop - self.start 
+        self.submitted = time.strftime(time_format, self.start_struct)
+        self.completed = time.strftime(time_format)
+        return result
+    
+    def pre_task(self, d, queued_engine):
+        """
+        Clear the engine before running the task if clear_before is set.
+        """
+        if self.clear_before:
+            d.addCallback(lambda r: queued_engine.reset())
+    
+    def post_task(self, d, queued_engine):
+        """
+        Clear the engine after running the task if clear_after is set.
+        """
+        def reseter(result):
+            queued_engine.reset()
+            return result
+        if self.clear_after:
+            d.addBoth(reseter)
+    
+    def submit_task(self, d, queued_engine):
+        raise NotImplementedError('submit_task must be implemented in a subclass')
+    
+    def process_result(self, result, engine_id):
+        """
+        Process a task result.
+        
+        This is the default `process_result` that just returns the raw
+        result or a `Failure`.
+        """
+        if isinstance(result, failure.Failure):
+            return (False, result)
+        else:
+            return (True, result)
+    
+    def check_depend(self, properties):
+        """
+        Calls self.depend(properties) to see if a task should be run.
+        """
+        if self.depend is not None:
+            return self.depend(properties)
+        else:
+            return True
+
+    def can_task(self):
+        self.depend = can(self.depend)
+        if isinstance(self.recovery_task, BaseTask):
+            self.recovery_task.can_task()
+            
+    def uncan_task(self):
+        self.depend = uncan(self.depend)
+        if isinstance(self.recovery_task, BaseTask):
+            self.recovery_task.uncan_task()
+
+class MapTask(BaseTask):
+    """
+    A task that consists of a function and arguments.
+    """
+    
+    zi.implements(ITask)
     
-    >>> 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 __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`.
 
-    A dependency case:
-    >>> def hasMPI(props):
-    ...     return props.get('mpi') is not None
-    >>> t = Task('mpi.send(blah,blah)', depend = hasMPI)
+        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, **options):
+            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 isinstance(pull, str):
-            self.pull = [pull]
-        else:
+        
+        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
-        self.clear_before = clear_before
-        self.clear_after = clear_after
-        self.retries=retries
-        self.recovery_task = recovery_task
-        self.depend = depend
-        self.options = options
-        self.taskid = None
+        else:
+            raise TypeError('push must be a dict')
+        
+        BaseTask.__init__(self, clear_before, clear_after, retries, 
+            recovery_task, depend)
+
+    def submit_task(self, d, queued_engine):
+        if self.push is not None:
+            d.addCallback(lambda r: queued_engine.push(self.push))
+        
+        d.addCallback(lambda r: queued_engine.execute(self.expression))
 
-class ResultNS:
-    """The result namespace object for use in TaskResult objects as tr.ns.
+        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.
     
@@ -128,13 +415,10 b' class ResultNS:'
     --------
     
     >>> 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
-
     >>> ns['foo']
     [0, 1, 2]
     """
@@ -156,7 +440,7 b' class ResultNS:'
 
 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
@@ -166,21 +450,21 b' class TaskResult(object):'
         
     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
@@ -192,7 +476,7 b' class TaskResult(object):'
             return self._ns
     
     def _setNS(self, v):
-        raise Exception("I am protected!")
+        raise Exception("the ns attribute cannot be changed")
     
     ns = property(_getNS, _setNS)
     
@@ -218,15 +502,19 b' class TaskResult(object):'
     
     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. 
     
@@ -241,12 +529,15 b' class IWorker(zi.Interface):'
         :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):
@@ -261,52 +552,26 b' class WorkerFromQueuedEngine(object):'
     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.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)
@@ -323,14 +588,14 b' class IScheduler(zi.Interface):'
         """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.
@@ -340,7 +605,7 b' class IScheduler(zi.Interface):'
                 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
@@ -350,8 +615,9 b' class IScheduler(zi.Interface):'
         """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):
@@ -374,15 +640,15 b' class IScheduler(zi.Interface):'
         """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)
@@ -439,7 +705,9 b' class FIFOScheduler(object):'
         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:
@@ -449,9 +717,12 b' class FIFOScheduler(object):'
 
 
 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):
@@ -466,13 +737,15 b' class LIFOScheduler(FIFOScheduler):'
     
 
 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
@@ -481,13 +754,14 b' class ITaskController(cs.IControllerBase):'
         """
     
     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:
@@ -512,23 +786,35 b' class ITaskController(cs.IControllerBase):'
         """
     
     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.
@@ -565,7 +851,7 b' class TaskController(cs.ControllerAdapterBase):'
     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
@@ -590,21 +876,25 b' class TaskController(cs.ControllerAdapterBase):'
     #---------------------------------------------------------------------------
     
     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)
@@ -619,7 +909,9 b' class TaskController(cs.ControllerAdapterBase):'
             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:
@@ -678,8 +970,10 b' class TaskController(cs.ControllerAdapterBase):'
     #---------------------------------------------------------------------------
     
     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:
@@ -687,14 +981,16 b' class TaskController(cs.ControllerAdapterBase):'
     
     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
@@ -709,7 +1005,7 b' class TaskController(cs.ControllerAdapterBase):'
             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:
@@ -731,14 +1027,15 b' class TaskController(cs.ControllerAdapterBase):'
                 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:
@@ -755,7 +1052,7 b' class TaskController(cs.ControllerAdapterBase):'
             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
@@ -763,7 +1060,7 b' class TaskController(cs.ControllerAdapterBase):'
                     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 
@@ -779,17 +1076,18 b' class TaskController(cs.ControllerAdapterBase):'
                 # 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
+            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`.
@@ -799,5 +1097,17 b' class TaskController(cs.ControllerAdapterBase):'
             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,9 +1,8 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"
@@ -24,116 +23,97 b' 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.
@@ -154,6 +134,45 b' class BlockingTaskClient(InteractiveTaskClient):'
         """
         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

@@ -34,6 +34,15 b' 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
@@ -43,32 +52,38 b' from IPython.kernel.clientinterfaces import ('
 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(request):
+    def remote_spin():
         """"""
     
-    def remote_queue_status(request, verbose):
+    def remote_queue_status(verbose):
+        """"""
+    
+    def remote_clear():
         """"""
 
 
 class FCTaskControllerFromTaskController(Referenceable):
-    """XML-RPC attachmeot for controller.
+    """
+    Adapt a `TaskController` to an `IFCTaskController`
         
-    See IXMLRPCTaskController and ITaskController (and its children) for documentation. 
+    This class is used to expose a `TaskController` over the wire using
+    the Foolscap network protocol.
     """
+    
     implements(IFCTaskController, IFCClientInterfaceProvider)
     
     def __init__(self, taskController):
@@ -92,8 +107,8 b' class FCTaskControllerFromTaskController(Referenceable):'
     
     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:
@@ -132,6 +147,9 b' class FCTaskControllerFromTaskController(Referenceable):'
         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'
     
@@ -144,13 +162,23 b' components.registerAdapter(FCTaskControllerFromTaskController,'
 #-------------------------------------------------------------------------------
 
 class FCTaskClient(object):
-    """XML-RPC based TaskController client that implements ITaskController.
+    """
+    Client class for Foolscap exposed `TaskController`.
         
-    :Parameters:
-        addr : (ip, port)
-            The ip (str) and port (int) tuple of the `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)
+    
+    implements(
+        taskmodule.ITaskController, 
+        IBlockingClientAdaptor,
+        ITaskMapperFactory,
+        IMapper,
+        ITaskParallelDecorator
+    )
     
     def __init__(self, remote_reference):
         self.remote_reference = remote_reference
@@ -168,48 +196,26 b' class FCTaskClient(object):'
     def run(self, task):
         """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 : 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):)
-        
-        The meaning of the arguments is as follows:
-        
-        :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
+            task : an `ITask` implementer
             
         :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
@@ -224,20 +230,19 b' class FCTaskClient(object):'
         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
@@ -248,20 +253,77 b' class FCTaskClient(object):'
         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

@@ -163,7 +163,6 b' class IEngineCoreTestCase(object):'
         try:
             import numpy
         except:
-            print 'no numpy, ',
             return
         a = numpy.random.random(1000)
         d = self.engine.push(dict(a=a))

IPython/kernel/tests/multienginetest.py

@@ -750,16 +750,6 b' class ISynchronousMultiEngineCoordinatorTestCase(IMultiEngineCoordinatorTestCase'
             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 = []

IPython/kernel/tests/tasktest.py

@@ -43,23 +43,23 b' class TaskTestBase(object):'
 
 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))
@@ -67,15 +67,15 b' class ITaskControllerTestCase(TaskTestBase):'
         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))
@@ -85,10 +85,10 b' class ITaskControllerTestCase(TaskTestBase):'
         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)
@@ -101,10 +101,10 b' class ITaskControllerTestCase(TaskTestBase):'
         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)
@@ -112,47 +112,76 b' class ITaskControllerTestCase(TaskTestBase):'
         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

@@ -38,7 +38,7 b' try:'
         IEngineQueuedTestCase
 except ImportError:
     print "we got an error!!!"
-    pass
+    raise
 else:
     class EngineFCTest(DeferredTestCase, 
                        IEngineCoreTestCase, 

IPython/kernel/tests/test_multienginefc.py

@@ -26,9 +26,20 b' try:'
     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):
@@ -68,3 +79,66 b' else:'
             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

@@ -20,8 +20,6 b' 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
@@ -30,6 +28,11 b' 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):
@@ -181,6 +184,3 b' else:'
             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

@@ -30,6 +30,8 b' try:'
     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:
@@ -38,6 +40,12 b' else:'
     # Tests
     #-------------------------------------------------------------------------------
 
+    def _raise_it(f):
+        try:
+            f.raiseException()
+        except CompositeError, e:
+            e.raise_exception()
+
     class TaskTest(DeferredTestCase, ITaskControllerTestCase):
 
         def setUp(self):
@@ -88,3 +96,66 b' else:'
             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

@@ -53,7 +53,7 b' class DistributedSpider(object):'
             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, ':'

docs/examples/kernel/helloworld.py

@@ -8,7 +8,7 b' 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

@@ -31,7 +31,7 b' sigma_vals = N.linspace(0.0, 0.2,5)'
 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))

docs/examples/kernel/task1.py

@@ -11,8 +11,8 b' 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

@@ -10,7 +10,7 b' 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)
@@ -18,7 +18,7 b' for i in range(6):'
     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)
@@ -26,7 +26,7 b' for i in range(6):'
     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)

docs/examples/kernel/task_profiler.py

@@ -55,7 +55,7 b' def main():'
 
     # 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)

docs/source/changes.txt

@@ -12,6 +12,22 b' 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`,
@@ -38,11 +54,11 b' New features'
 	  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.
@@ -52,6 +68,12 b' Bug fixes'
 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 

docs/source/install/index.txt

@@ -1,3 +1,5 b''
+.. _install_index:
+
 ==================
 Installation
 ==================

docs/source/overview.txt

@@ -4,18 +4,6 b''
 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
 ========
 
@@ -25,8 +13,19 b' 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,
@@ -50,17 +49,16 b' which tries to:'
       WX applications via special threading flags. The normal Python
       shell can only do this for Tkinter applications.
 
-
-Main features
--------------
+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 ('?', 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.
+	  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. 
@@ -70,31 +68,31 b' Main features'
   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
+	  :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 ! are passed
-  directly to the system shell, and using !! or var = !cmd 
+	* 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 %bg.
+	  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 $ 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
+	  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 %store command, which
+	* 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
-  %store and edited via %edit.
+	  :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
@@ -106,15 +104,15 b' Main features'
   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")'.
+	  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
-  '>>>' or '...' such as those from other python sessions or the
+	  :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
@@ -131,59 +129,46 b' Main features'
   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
+	  :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 %debug after seeing an exception. winpdb is 
+	  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
-  profile.run()) or complete programs under the profiler's control.
+	  :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 '%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 '>>>'
+	  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,3 +1,5 b''
+.. _parallel_index:
+
 ====================================
 Using IPython for Parallel computing
 ====================================