A new version and API for map is now working. We also now have an

@parallel decorator that creates a parallel implementation of a function.

IPython/kernel/mapper.py

@@ -0,0 +1,42 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
+
+class IMapper(Interface):
+    
+    def __call__(func, *sequences):
+        """Do map in parallel."""
+
+class Mapper(object):
+    
+    implements(IMapper)
+    
+    def __init__(self, multiengine, dist='b', targets='all', block=True):
+        self.multiengine = multiengine
+        self.dist = dist
+        self.targets = targets
+        self.block = block
+        
+    def __call__(self, func, *sequences):
+        return self.map(func, *sequences)
+    
+    def map(self, func, *sequences):
+        assert isinstance(func, (str, FunctionType)), "func must be a fuction or str"
+        return self.multiengine._map(func, sequences, dist=self.dist,
+            targets=self.targets, block=self.block)
  No newline at end of file

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,45 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, style='basic', targets='all'):
-        """Gather object key from targets.
+    def gather(key, dist='b', targets='all'):
+        """Gather object key from targets."""
     
-        :Parameters:
-            key : string
-                The name of a sequence on the targets to gather.
-            style : string
-                A specification of how the sequence is partitioned.  Currently 
-                only 'basic' is implemented.                
-        """
+    def _map(func, seq, dist='b', targets='all'):
+        """A parallelized version of Python's builtin map."""
     
-    def map(func, seq, style='basic', targets='all'):
-        """A parallelized version of Python's builtin map.
-        
-        This function implements the following pattern:
-        
-        1. The sequence seq is scattered to the given targets.
-        2. map(functionSource, seq) is called on each engine.
-        3. The resulting sequences are gathered back to the local machine.
-                
-        :Parameters:
-            targets : int, list or 'all'
-                The engine ids the action will apply to.  Call `get_ids` to see
-                a list of currently available engines.
-            func : str, function
-                An actual function object or a Python string that names a 
-                callable defined on the engines.
-            seq : list, tuple or numpy array
-                The local sequence to be scattered.
-            style : str
-                Only 'basic' is supported for now.
-                
-        :Returns: A list of len(seq) with functionSource called on each element
-        of seq.
-                
-        Example
-        =======
-        
-        >>> rc.mapAll('lambda x: x*x', range(10000))
-        [0,2,4,9,25,36,...]
-        """
+    def map(func, *sequences):
+        """Do a basic map with default for dist and targets."""
+    
+    def mapper(dist='b', targets='all'):
+        """Create a mapper with dist and targets."""
+    
+    def parallel(dist='b', targets='all'):
+        """A decorator that build a parallel function."""
 
 
 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 _map(func, sequences, dist='b', targets='all', block=True):
+        """Perform an actual map."""
+    
+    def map(func, *sequences):
+        """Do a basic map with default for dist and targets."""
+    
+    def mapper(dist='b', targets='all', block=True):
+        """Create a mapper with dist, targets and block."""
+    
+    def parallel(dist='b', targets='all', block=True):
+        """A decorator that build a parallel function."""
 
 
 #-------------------------------------------------------------------------------
@@ -722,46 +700,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,7 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 Mapper
 from IPython.kernel import map as Map
 from IPython.kernel import multiengine as me
 from IPython.kernel.multiengine import (IFullMultiEngine,
@@ -779,29 +780,40 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 _map(self, func, seq, dist='b', targets=None, block=None):
         """
         A parallelized version of Python's builtin map
         """
         targets, block = self._findTargetsAndBlock(targets, block)
-        return self._blockFromThread(self.smultiengine.map, func, seq, 
-            style, targets=targets, block=block)
+        return self._blockFromThread(self.smultiengine._map, func, seq, 
+            dist, targets=targets, block=block)
+    
+    def map(self, func, *sequences):
+        return self.mapper()(func, *sequences)
+    
+    def mapper(self, dist='b', targets='all', block=None):
+        return Mapper(self, dist, targets, block)
+    
+    def parallel(self, dist='b', targets=None, block=None):
+        targets, block = self._findTargetsAndBlock(targets, block)       
+        pf = ParallelFunction(self, dist=dist, targets=targets, block=block)
+        return pf
     
     #---------------------------------------------------------------------------
     # IMultiEngineExtras

IPython/kernel/multienginefc.py

@@ -29,6 +29,8 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 Mapper
 from IPython.kernel.twistedutil import gatherBoth
 from IPython.kernel.multiengine import (MultiEngine,
     IMultiEngine,
@@ -475,7 +477,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 +485,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 +543,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 +551,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,14 +606,16 b' class FCFullSynchronousMultiEngineClient(object):'
         d.addCallback(do_gather)
         return d
 
-    def map(self, func, seq, style='basic', targets='all', block=True):
+    def _map(self, func, sequences, dist='b', targets='all', block=True):
         """
         Call a callable on elements of a sequence.
-        
-        map(f, range(10)) -> [f(0), f(1), f(2), ...]
-        map(f, zip(range))
         """
-        d_list = []
+        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))
@@ -623,12 +627,22 b' class FCFullSynchronousMultiEngineClient(object):'
         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):
+        return self.mapper()(func, *sequences)
+    
+    def mapper(self, dist='b', targets='all', block=True):
+        return Mapper(self, dist, targets, block)
+
+    def parallel(self, dist='b', targets='all', block=True):
+        pf = ParallelFunction(self, dist=dist, targets=targets, block=True)
+        return pf
+    
     #---------------------------------------------------------------------------
     # ISynchronousMultiEngineExtras related methods
     #---------------------------------------------------------------------------

IPython/kernel/parallelfunction.py

@@ -16,17 +16,32 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 ParallelFunction(object):
+    """
+    A decorator for building parallel functions.
+    """
+    
+    def __init__(self, multiengine, dist='b', targets='all', block=True):
+        """
+        Create a `ParallelFunction decorator`.
         """
-        assert isinstance(func, (str, FunctionType)), "func must be a fuction or str"
-        self.func = func
         self.multiengine = multiengine
+        self.dist = dist
         self.targets = targets
         self.block = block
         
-    def __call__(self, sequence):
-        return self.multiengine.map(self.func, sequence, targets=self.targets, block=self.block)
  No newline at end of file
+    def __call__(self, func):
+        """
+        Decorate the function to make it run in parallel.
+        """
+        assert isinstance(func, (str, FunctionType)), "func must be a fuction or str"
+        self.func = func
+        def call_function(*sequences):
+            return self.multiengine._map(self.func, sequences, dist=self.dist,
+                targets=self.targets, block=self.block)
+        return call_function
+
+    
  No newline at end of file

docs/source/changes.txt

@@ -52,6 +52,10 b' Bug fixes'
 Backwards incompatible changes
 ------------------------------
 
+	* 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