Changeset 3021

Timestamp:

02/05/08 22:16:56 (10 months ago)

Author:

bgranger

Message:

Merging changes from ipython1-data-r3017 branch

Files:

• ipython1/trunk/docs/ChangeLog (modified) (1 diff)
• ipython1/trunk/docs/examples/rmt.ipy (modified) (1 diff)
• ipython1/trunk/ipython1/core/interpreter.py (modified) (2 diffs)
• ipython1/trunk/ipython1/kernel/enginepb.py (modified) (13 diffs)
• ipython1/trunk/ipython1/kernel/engineservice.py (modified) (9 diffs)
• ipython1/trunk/ipython1/kernel/map.py (modified) (1 diff)
• ipython1/trunk/ipython1/kernel/multiengine.py (modified) (7 diffs)
• ipython1/trunk/ipython1/kernel/multiengineclient.py (modified) (4 diffs)
• ipython1/trunk/ipython1/kernel/multienginehttp.py (modified) (4 diffs)
• ipython1/trunk/ipython1/kernel/multienginexmlrpc.py (modified) (13 diffs)
• ipython1/trunk/ipython1/kernel/pbutil.py (modified) (2 diffs)
• ipython1/trunk/ipython1/kernel/pendingdeferred.py (modified) (1 diff)
• ipython1/trunk/ipython1/kernel/pickleutil.py (modified) (1 diff)
• ipython1/trunk/ipython1/kernel/task.py (modified) (1 diff)
• ipython1/trunk/ipython1/kernel/tests/engineservicetest.py (modified) (4 diffs)
• ipython1/trunk/ipython1/kernel/tests/multienginetest.py (modified) (3 diffs)
• ipython1/trunk/ipython1/kernel/tests/test_multiengine.py (modified) (1 diff)
• ipython1/trunk/ipython1/kernel/twistedutil.py (copied) (copied from ipython1/branches/ipython1-data-r3016/ipython1/kernel/twistedutil.py)
• ipython1/trunk/ipython1/kernel/util.py (modified) (1 diff)

ipython1/trunk/docs/ChangeLog

@@ +1 @@
+2008-02-05  Brian Granger <ellisonbg@gmail.com>
+
+    * ipython1.kernel.enginepb/multienginexmlrpc: Simplified certain 
+    interfaces so that they are simply a pass.  This was done for
+    interfaces that 1) are not part of the public API and 2)
+    only have one implementation (and always will).
+
+    * ipython1.kernel: Created new pushFunction/pullFunctions in the API
+    of the engines.  These functions now have all the needed logic for
+    pushing/pulling functions.  The logic has been removed from regular
+    push/pull to make things more robust.
+    
+    * ipython1.kernel.multiengine and friends:  Removed the 
+    IMultiEngineCoordinator interface (gather/scatter) entirely from the
+    controller.  These things are now handled by the client using push/pull.
+    This was done to make the load on the controller lighter.
+
-2008-01-28  Brian Granger <ellisonbg@gmail.com>
-

ipython1/trunk/docs/examples/rmt.ipy

@@ -34 +34 @@
-    rc.pushAll(num_per_engine=num_per_engine, N=N)
-    rc.executeAll('diffs = ensembleDiffs(num_per_engine, N)')
-
-pd.getResult(block=True
+
+rc.gatherAll('diffs'
-
-

ipython1/trunk/ipython1/core/interpreter.py

@@ -437 +437 @@
-        ----------
-        **kwds
-
-
+
+
+
+
+
-        # First set the func_globals for all functions to self.user_ns
+        new_kwds = {}
-        for k, v in kwds.iteritems():
-
-kwds[k] = FunctionType(v.func_code, self.user_ns
-
-kwds)
+not 
+raise TypeError("function object expected"
+
+new_kwds)        
-
-    def pack_exception(self,message,exc):
-        message['exception'] = exc.__class__
-        message['exception_value'] = \
-
+    
-
-    def feed_block(self, source, filename='<input>', symbol='single'):
@@ -530 +534 @@
-            return result
-
+    def pullFunction(self, key):
+        return self.pull(key)
-
-    #### Interactive user API ##################################################

ipython1/trunk/ipython1/kernel/enginepb.py

@@ -51 +51 @@
-from ipython1.kernel.pbutil import packageFailure, unpackageFailure, checkMessageSize
-from ipython1.kernel.pbconfig import CHUNK_SIZE
-
+
+
-from ipython1.kernel import newserialized
-from ipython1.kernel.error import PBMessageSizeError, ProtocolError
-, protocols
+
-from ipython1.kernel.controllerservice import IControllerBase
-from ipython1.kernel.engineservice import \
@@ -212 +213 @@
-    will be returned instead.
-    """
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
-
-class PBEngineReferenceFromService(pb.Referenceable, object):
@@ -316 +244 @@
-    
-    def remote_execute(self, lines):
+        """Execute lines of Python code.
+        
+        Returns a deferred to a result dict.
+        
+        Upon failure returns a pickled Failure.
+        """
-        d = self.service.execute(lines)
-        d.addErrback(packageFailure)
@@ -322 +256 @@
-        #d.addCallback(lambda r: log.msg("Got result: " + str(r)))
-        return d
-
-
+
+
+
+
+
+
+
+
+
+
-        try:
-            namespace = pickle.loads(pNamespace)
@@ -329 +271 @@
-            return defer.fail(failure.Failure()).addErrback(packageFailure)
-        else:
-            return self.service.setProperties(**namespace).addErrback(packageFailure)
-    
-    def remote_getProperties(self, *keys):
-        d = self.service.getProperties(*keys)
-        d.addCallback(pickle.dumps, 2)
-        d.addCallback(checkMessageSize, repr(keys))
-        d.addErrback(packageFailure)
-        return d
-    
-    def remote_hasProperties(self, *keys):
-        d = self.service.hasProperties(*keys)
-        d.addCallback(pickle.dumps, 2)
-        d.addCallback(checkMessageSize, repr(keys))
-        d.addErrback(packageFailure)
-        return d
-    
-    def remote_delProperties(self, *keys):
-        d = self.service.delProperties(*keys)
-        # d.addCallback(pickle.dumps, 2)
-        # d.addCallback(checkMessageSize, repr(keys))
-        d.addErrback(packageFailure)
-        return d
-    
-    def remote_clearProperties(self):
-        d = self.service.clearProperties()
-        # d.addCallback(pickle.dumps, 2)
-        # d.addCallback(checkMessageSize, repr(keys))
-        d.addErrback(packageFailure)
-        return d
-    
-    
-    #---------------------------------------------------------------------------
-    # Old version of push
-    #---------------------------------------------------------------------------
-        
-    def remote_push(self, pNamespace):
-        try:
-            namespace = pickle.loads(pNamespace)
-        except:
-            return defer.fail(failure.Failure()).addErrback(packageFailure)
-        else:
-            # The usage of globals() here is an attempt to bind any pickled functions
-            # to the globals of this module.  What we really want is to have it bound
-            # to the globals of the callers module.  This will require walking the 
-            # stack.  BG 10/3/07.
-            namespace = uncanDict(namespace, globals())
-            return self.service.push(**namespace).addErrback(packageFailure)
-    
@@ -417 +313 @@
-    # pull
-    #---------------------------------------------------------------------------     
-
+        
-    def remote_pull(self, *keys):
+        """Pull objects from a users namespace by keys.
+        
+        Returns a deferred to a pickled tuple of objects.  If any single
+        key has a problem, the Failure of that will be returned.
+        """
-        d = self.service.pull(*keys)
+        d.addCallback(pickle.dumps, 2)
+        d.addCallback(checkMessageSize, repr(keys))
+        d.addErrback(packageFailure)
+        return d
+    
+    # NOTE:  The paging version of pull is not being used right now  (BG 1/15/07).
+    
+    def remote_pullPaging(self, key, collector):
+        d = self.service.pull(key)
+        d.addCallback(newserialized.serialize)
+        d.addCallbacks(self._startPaging, packageFailure, callbackArgs=(collector,))
+        return d
+    
+    def _startPaging(self, serial, collector):
+        pager = SerializedPager(collector, serial, chunkSize=CHUNK_SIZE)
+
+    #---------------------------------------------------------------------------
+    # push/pullFuction
+    #---------------------------------------------------------------------------
+    
+    def remote_pushFunction(self, pNamespace):
+        try:
+            namespace = pickle.loads(pNamespace)
+        except:
+            return defer.fail(failure.Failure()).addErrback(packageFailure)
+        else:
+            # The usage of globals() here is an attempt to bind any pickled functions
+            # to the globals of this module.  What we really want is to have it bound
+            # to the globals of the callers module.  This will require walking the 
+            # stack.  BG 10/3/07.
+            namespace = uncanDict(namespace, globals())
+            return self.service.pushFunction(**namespace).addErrback(packageFailure)
+    
+    def remote_pullFunction(self, *keys):
+        d = self.service.pullFunction(*keys)
-        if len(keys)>1:
-            d.addCallback(canSequence)
@@ -428 +364 @@
-        d.addErrback(packageFailure)
-        return d
-
-
-
-
-
-
-
-
-
-
-
-
+
-    #---------------------------------------------------------------------------
-    # Other methods
@@ -445 +370 @@
-    
-    def remote_getResult(self, i=None):
+        """Get result i.
+        
+        Returns a deferred to a pickled result dict.
+        """
-        return self.service.getResult(i).addErrback(packageFailure)
-    
-    def remote_reset(self):
+        """Reset the Engine."""
-        return self.service.reset().addErrback(packageFailure)
-    
-    def remote_kill(self):
+        """Stop the Engines reactor."""
-        return self.service.kill().addErrback(packageFailure)
-    
-    def remote_keys(self):
+        """Get variable names that are currently defined in the user's namespace.
+        
+        Returns a deferred to a tuple of keys.
+        """
-        return self.service.keys().addErrback(packageFailure)
-    
@@ -461 +396 @@
-    
-    def remote_pushSerialized(self, pNamespace):
+        """Push a dict of keys and serialized objects into users namespace.
+        
+        @arg pNamespace: a pickle namespace of keys and serialized objects.
+        """
-        try:
-            namespace = pickle.loads(pNamespace)
@@ -470 +409 @@
-    
-    def remote_pullSerialized(self, *keys):
+        """Pull objects from users namespace by key as Serialized.
+        
+        Returns a deferred to a pickled dict of key, Serialized pairs.
+        """
-        d = self.service.pullSerialized(*keys)
-        d.addCallback(pickle.dumps, 2)
-        d.addCallback(checkMessageSize, repr(keys))
+        d.addErrback(packageFailure)
+        return d
+    
+    #---------------------------------------------------------------------------
+    # Properties interface
+    #---------------------------------------------------------------------------
+    
+    def remote_setProperties(self, pNamespace):
+        """Update the properties for this engine"""
+        try:
+            namespace = pickle.loads(pNamespace)
+        except:
+            return defer.fail(failure.Failure()).addErrback(packageFailure)
+        else:
+            return self.service.setProperties(**namespace).addErrback(packageFailure)
+    
+    def remote_getProperties(self, *keys):
+        """Pull a subdict of the properties for this engine by keys"""
+        d = self.service.getProperties(*keys)
+        d.addCallback(pickle.dumps, 2)
+        d.addCallback(checkMessageSize, repr(keys))
+        d.addErrback(packageFailure)
+        return d
+    
+    def remote_hasProperties(self, *keys):
+        """Check for keys in the properties for this engine."""
+        d = self.service.hasProperties(*keys)
+        d.addCallback(pickle.dumps, 2)
+        d.addCallback(checkMessageSize, repr(keys))
+        d.addErrback(packageFailure)
+        return d
+    
+    def remote_delProperties(self, *keys):
+        """Remove values of the properties for this engine by keys."""
+        d = self.service.delProperties(*keys)
+        # d.addCallback(pickle.dumps, 2)
+        # d.addCallback(checkMessageSize, repr(keys))
+        d.addErrback(packageFailure)
+        return d
+    
+    def remote_clearProperties(self):
+        """Clear the properties for this engine."""
+        d = self.service.clearProperties()
+        # d.addCallback(pickle.dumps, 2)
+        # d.addCallback(checkMessageSize, repr(keys))
-        d.addErrback(packageFailure)
-        return d
@@ -571 +559 @@
-    def pushOld(self, **namespace):
-        try:
-canDict(namespace)
+namespace
-        except:
-            return defer.fail(failure.Failure())
@@ -619 +607 @@
-    # pull
-    #---------------------------------------------------------------------------
-
+    
-    def pullOld(self, *keys):
-        d = self.callRemote('pull', *keys)
-        d.addCallback(self.checkReturnForFailure)
-        d.addCallback(pickle.loads)
-        # The usage of globals() here is an attempt to bind any pickled functions
-        # to the globals of this module.  What we really want is to have it bound
-        # to the globals of the callers module.  This will require walking the 
-        # stack.  BG 10/3/07.
-        if len(keys)>1:
-            d.addCallback(uncanSequence, globals())
-        elif len(keys)>0:
-            d.addCallback(uncan, globals())
-        return d
-    
@@ -658 +638 @@
-    
-    pull = pullOld
+
+    #---------------------------------------------------------------------------
+    # push/pullFunction
+    #---------------------------------------------------------------------------
+    
+    def pushFunction(self, **namespace):
+        try:
+            package = pickle.dumps(canDict(namespace), 2)
+        except:
+            return defer.fail(failure.Failure())
+        else:
+            package = checkMessageSize(package, namespace.keys())
+            if isinstance(package, failure.Failure):
+                return defer.fail(package)
+            else:
+                d = self.callRemote('pushFunction', package)
+                return d.addCallback(self.checkReturnForFailure)    
+    
+    def pullFunction(self, *keys):
+        d = self.callRemote('pullFunction', *keys)
+        d.addCallback(self.checkReturnForFailure)
+        d.addCallback(pickle.loads)
+        # The usage of globals() here is an attempt to bind any pickled functions
+        # to the globals of this module.  What we really want is to have it bound
+        # to the globals of the callers module.  This will require walking the 
+        # stack.  BG 10/3/07.
+        if len(keys)==1:
+            d.addCallback(uncan, globals())
+        elif len(keys)>1:
+            d.addCallback(uncanSequence, globals())            
+        return d
-    
-    #---------------------------------------------------------------------------

ipython1/trunk/ipython1/kernel/engineservice.py

@@ -45 +45 @@
-from ipython1.core.interpreter import Interpreter
-from ipython1.kernel import newserialized, error, util
-
+twisted
-from ipython1.kernel import codeutil
-from ipython1.kernel.pickleutil import can, uncan
-
-
@@ -93 +92 @@
-        """Pulls values out of the user's namespace by keys.
-        
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        """
-    
@@ -205 +217 @@
-
-
-
-
-#-------------------------------------------------------------------------------
-# Functions and classes to implement the EngineService
-#-------------------------------------------------------------------------------
+
+
-class StrictDict(dict):
-    """This is a strict copying dictionary for use as the interface to the 
@@ -403 +415 @@
-method: pull(*keys)
-keys = %r""" % (self.id, keys)
-
+
+
+
+
-            pulledDeferreds = []
-            for key in keys:
@@ -414 +429 @@
-                           consumeErrors=1)
-            return dTotal
-        elif len(keys) > 0:
-            return self.executeAndRaise(msg, self.shell.pull, keys[0])
-        else:
-            return self.executeAndRaise(msg, self.shell.pull, None)
+    
+    def pushFunction(self, **namespace):
+        msg = """engine: %r
+method: pushFunction(**namespace)
+namespace.keys() = %r""" % (self.id, namespace.keys())
+        d = self.executeAndRaise(msg, self.shell.pushFunction, **namespace)
+        return d
+    
+    def pullFunction(self, *keys):
+        msg = """engine %r
+method: pullFunction(*keys)
+keys = %r""" % (self.id, keys)
+        if len(keys)==1:
+            return self.executeAndRaise(msg, self.shell.pullFunction, keys[0])
+        elif len(keys) > 1:
+            pulledDeferreds = []
+            for key in keys:
+                d = self.executeAndRaise(msg, self.shell.pullFunction, key)
+                pulledDeferreds.append(d)
+            # This will fire on the first failure and log the rest.
+            dTotal = gatherBoth(pulledDeferreds, 
+                           fireOnOneErrback=1,
+                           logErrors=1, 
+                           consumeErrors=1)
+            return dTotal
+        else:
+            return self.executeAndRaise(msg, self.shell.pullFunction, None)        
-    
-    def getResult(self, i=None):
@@ -507 +547 @@
-            try:
-                unserialized = newserialized.IUnSerialized(v)
-
-
-
-
-
+
-            except:
-                return defer.fail()
@@ -520 +556 @@
-method: pullSerialized(*keys)
-keys = %r""" % (self.id, keys)
-
+
+
+
+
+
+
-            pulledDeferreds = []
-            for key in keys:
-
+ 
-                pulledDeferreds.append(d)
-            # This will fire on the first failure and log the rest.
@@ -535 +576 @@
-                for v in values:
-                    try:
- = newserialized.serialize(v
+.append(newserialized.serialize(v)
-                    except:
-                        return defer.fail(failure.Failure())
-value
+serial
-            return packThemUp
-
-
-
-
-
-
-
+
+
-def queue(methodToQueue):
-    def queuedMethod(this, *args, **kwargs):
@@ -685 +721 @@
-        pass
-        
+    @queue
+    def pushFunction(self, **namespace):
+        pass      
+    
+    @queue
+    def pullFunction(self, *keys):
+        pass        
+
-    def getResult(self, i=None):
-        if i is None:

ipython1/trunk/ipython1/kernel/map.py

@@ -89 +89 @@
-            return genutil_flatten(listOfPartitions)
-        # If we have scalars, just return listOfPartitions
-    
+
-
-class RoundRobinMap(Map):

ipython1/trunk/ipython1/kernel/multiengine.py

@@ -33 +33 @@
-from zope.interface import Interface, implements, Attribute
-
-gatherBoth
- import map as Map
+printer
+.twistedutil import gatherBoth
-from ipython1.kernel import error
-from ipython1.kernel.controllerservice import \
@@ -126 +126 @@
-    def pullAll(*keys):
-        """Pull from all targets."""
+          
+    def pushFunction(targets, **namespace):
+        """"""
+        
+    def pushFunctionAll(**namespace):
+        """"""
+        
+    def pullFunction(targets, *keys):
+        """"""
+        
+    def pullFunctionAll(*keys):
+        """"""
-                
-    def getResult(targets, i=None):
@@ -210 +222 @@
-        """get all the properties dicts."""
-    
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
-    """A controller that exposes an explicit interface to all of its engines.
-    
@@ -400 +371 @@
-        return self.pull('all', *keys)
-    
+    def pushFunction(self, targets, **ns):
+        return self._performOnEnginesAndGatherBoth('pushFunction', targets, **ns)
+        
+    def pushFunctionAll(self, **ns):
+        return self.pushFunction('all', **ns)
+        
+    def pullFunction(self, targets, *keys):
+        return self._performOnEnginesAndGatherBoth('pullFunction', targets, *keys)
+    
+    def pullFunctionAll(self, *keys):
+        return self.pullFunction('all', *keys)
+    
-    def getResult(self, targets, i=None):
-        return self._performOnEnginesAndGatherBoth('getResult', targets, i)
@@ -569 +552 @@
-        return self.getProperties('all')
-    
-    #---------------------------------------------------------------------------
-    # IEngineCoordinator methods
-    #---------------------------------------------------------------------------
-
-    def scatter(self, targets, key, seq, style='basic', flatten=False):
-        log.msg("Scattering %r to %r" % (key, targets))
-        try:
-            engines = self.engineList(targets)
-        except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
-            return defer.fail(failure.Failure())
-        else:
-            nEngines = len(engines)    
-            mapClass = Map.styles[style]
-            mapObject = mapClass()
-            dList = []
-            for index, engine in enumerate(engines):
-                partition = mapObject.getPartition(seq, index, nEngines)
-                if flatten and len(partition) == 1:    
-                    dList.append(engine.push(**{key: partition[0]}))
-                else:
-                    dList.append(engine.push(**{key: partition}))
-            return gatherBoth(dList, 
-                              fireOnOneErrback=1,
-                              consumeErrors=1,
-                              logErrors=0)  
-                              
-    def scatterAll(self, key, seq, style='basic', flatten=False):
-        return self.scatter('all', key, seq, style, flatten)
-    
-    def gather(self, targets, key, style='basic'):
-        """gather a distributed object, and reassemble it"""
-        log.msg("Gathering %s from %r" % (key, targets))
-        try:
-             engines = self.engineList(targets)
-        except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
-            return defer.fail(failure.Failure())
-        else:
-            nEngines = len(engines)    
-            dList = []
-            for e in engines:
-                dList.append(e.pull(key))    
-            mapClass = Map.styles[style]
-            mapObject = mapClass()
-            d = gatherBoth(dList, 
-                           fireOnOneErrback=1,
-                           consumeErrors=1,
-                           logErrors=0)  
-            return d.addCallback(mapObject.joinPartitions)
-    
-    def gatherAll(self, key, style='basic'):
-        return self.gather('all', key, style)
-
-
-components.registerAdapter(MultiEngine, 
@@ -674 +605 @@
-    @twoPhase
-    def pull(self, targets, *keys):
-
-
+
+
+
+
+
+
+
+
+
+
+
+
-    @twoPhase
-    def getResult(self, targets, i=None):
@@ -728 +669 @@
-        return self.multiengine.clearProperties(targets, *keys)
-    
-    @twoPhase
-    def scatter(self, targets, key, seq, style='basic', flatten=False):
-        return self.multiengine.scatter(targets, key, seq, style, flatten)
-    
-    @twoPhase
-    def gather(self, targets, key, style='basic'):
-        return self.multiengine.gather(targets, key, style)
-    
-    #---------------------------------------------------------------------------
-    # IMultiEngine methods

ipython1/trunk/ipython1/kernel/multiengineclient.py

@@ -30 +30 @@
-from ipython1.kernel import error
-from ipython1.kernel.parallelfunction import ParallelFunction
+from ipython1.kernel import map as Map
-
-
@@ -422 +423 @@
-        if self.block:
-            if isinstance(func, FunctionType):
-
+Function
-                sourceToRun = '_ipython_map_seq_result = map(_ipython_map_func, _ipython_map_seq)'
-            elif isinstance(func, str):
@@ -435 +436 @@
-        else:
-            if isinstance(func, FunctionType):
-
+Function
-                sourceToRun = '_ipython_map_seq_result = map(_ipython_map_func, _ipython_map_seq)'
-            elif isinstance(func, str):
@@ -489 +490 @@
-        """
-        return self.parallelize('all', functionName)
+
+
+#-------------------------------------------------------------------------------
+# MultiEngineCoordinator
+#-------------------------------------------------------------------------------
+
+class MultiEngineCoordinator(object):
+    
+    def scatter(self, targets, key, seq, style='basic', flatten=False):
+        """Partition and distribute a sequence to a set of targets/engines.
+        
+        This method partitions a Python sequence and then pushes the partitions
+        to a set of engines.
+        
+        :Parameters:
+            targets : int, list or 'all'
+                The engine ids the action will apply to.  Call `getIDs` to see
+                a list of currently available engines.
+            key : str
+                What to call the partitions on the engines.
+            seq : list, tuple or numpy array
+                The sequence to be partitioned and pushed.
+            style : str
+                The style of partitioning to use.  Only 'basic' is supported
+            flatten : boolean
+                Should length 1 partitions be flattened to scalars upon pushing.
+        """
+        self._checkClientID()
+        localBlock = self._reallyBlock()
+        if targets=='all':
+            engines = self.getIDs()
+        else:
+            engines = targets
+        nEngines = len(engines)    
+        mapClass = Map.styles[style]
+        mapObject = mapClass()
+        results = []
+        for index, engine in enumerate(engines):
+            partition = mapObject.getPartition(seq, index, nEngines)
+            if flatten and len(partition) == 1:    
+                results.append(self.push(engine, **{key: partition[0]}))
+            else:
+                results.append(self.push(engine, **{key: partition}))           
+        if not localBlock:
+            self.barrier(*results)
+        return nEngines*[None]
+        
+    def scatterAll(self, key, seq, style='basic', flatten=False):
+        """Partition and distribute a sequence to all targets/engines.
+        
+        See the docstring for `scatter` for full details.
+        """
+        return self.scatter('all', key, seq, style, flatten)
+    
+    scatter_all = scatterAll
+    
+    def gather(self, targets, key, style='basic'):
+        """Gather a set of sequence partitions that are distributed on targets.
+        
+        This method is the inverse of `scatter`