Merge pull request #920 from minrk/parallel

Unordered iteration of AsyncMapResults (+ a couple fixes)

Feature:
* AsyncMapResults now support unordered iteration, in case your map is inhomogeneous, and you want to get the quick results first (including new tests).

Fix:
* fixed '*' as an ip argument in the kernel/qtconsole and ipcontroller.  It was not always getting converted to a connectable interface.

Docs and Examples:
* added a couple more parallel examples, based on recent discussions (monitoring engine stdout/err, more advanced handling of results as they come in, AsyncMapResult iteration).
* merged in some cleanup / fixes to the parallel docs from my SciPy 2011 tutorial.
* moved parallel doc figures into figs subdir

docs/examples/parallel/customresults.py

@@ -0,0 +1,61 b''
+"""An example for handling results in a way that AsyncMapResult doesn't provide
+
+Specifically, out-of-order results with some special handing of metadata.
+
+This just submits a bunch of jobs, waits on the results, and prints the stdout
+and results of each as they finish.
+
+Authors
+-------
+* MinRK
+"""
+import time
+import random
+
+from IPython import parallel
+
+# create client & views
+rc = parallel.Client()
+dv = rc[:]
+v = rc.load_balanced_view()
+
+
+# scatter 'id', so id=0,1,2 on engines 0,1,2
+dv.scatter('id', rc.ids, flatten=True)
+print dv['id']
+
+
+def sleep_here(count, t):
+    """simple function that takes args, prints a short message, sleeps for a time, and returns the same args"""
+    import time,sys
+    print "hi from engine %i" % id
+    sys.stdout.flush()
+    time.sleep(t)
+    return count,t
+
+amr = v.map(sleep_here, range(100), [ random.random() for i in range(100) ], chunksize=2)
+
+pending = set(amr.msg_ids)
+while pending:
+    try:
+        rc.wait(pending, 1e-3)
+    except parallel.TimeoutError:
+        # ignore timeouterrors, since they only mean that at least one isn't done
+        pass
+    # finished is the set of msg_ids that are complete
+    finished = pending.difference(rc.outstanding)
+    # update pending to exclude those that just finished
+    pending = pending.difference(finished)
+    for msg_id in finished:
+        # we know these are done, so don't worry about blocking
+        ar = rc.get_result(msg_id)
+        print "job id %s finished on engine %i" % (msg_id, ar.engine_id)
+        print "with stdout:"
+        print '    ' + ar.stdout.replace('\n', '\n    ').rstrip()
+        print "and results:"
+        
+        # note that each job in a map always returns a list of length chunksize
+        # even if chunksize == 1
+        for (count,t) in ar.result:
+            print "  item %i: slept for %.2fs" % (count, t)
+

docs/examples/parallel/iopubwatcher.py

@@ -0,0 +1,83 b''
+"""A script for watching all traffic on the IOPub channel (stdout/stderr/pyerr) of engines.
+
+This connects to the default cluster, or you can pass the path to your ipcontroller-client.json
+
+Try running this script, and then running a few jobs that print (and call sys.stdout.flush),
+and you will see the print statements as they arrive, notably not waiting for the results
+to finish.
+
+You can use the zeromq SUBSCRIBE mechanism to only receive information from specific engines,
+and easily filter by message type.
+
+Authors
+-------
+* MinRK
+"""
+
+import os
+import sys
+import json
+import zmq
+
+from IPython.zmq.session import Session
+from IPython.parallel.util import disambiguate_url
+from IPython.utils.py3compat import str_to_bytes
+from IPython.utils.path import get_security_file
+
+def main(connection_file):
+    """watch iopub channel, and print messages"""
+    
+    ctx = zmq.Context.instance()
+    
+    with open(connection_file) as f:
+        cfg = json.loads(f.read())
+    
+    location = cfg['location']
+    reg_url = cfg['url']
+    session = Session(key=str_to_bytes(cfg['exec_key']))
+    
+    query = ctx.socket(zmq.DEALER)
+    query.connect(disambiguate_url(cfg['url'], location))
+    session.send(query, "connection_request")
+    idents,msg = session.recv(query, mode=0)
+    c = msg['content']
+    iopub_url = disambiguate_url(c['iopub'], location)
+    sub = ctx.socket(zmq.SUB)
+    # This will subscribe to all messages:
+    sub.setsockopt(zmq.SUBSCRIBE, b'')
+    # replace with b'' with b'engine.1.stdout' to subscribe only to engine 1's stdout
+    # 0MQ subscriptions are simple 'foo*' matches, so 'engine.1.' subscribes
+    # to everything from engine 1, but there is no way to subscribe to
+    # just stdout from everyone.
+    # multiple calls to subscribe will add subscriptions, e.g. to subscribe to
+    # engine 1's stderr and engine 2's stdout:
+    # sub.setsockopt(zmq.SUBSCRIBE, b'engine.1.stderr')
+    # sub.setsockopt(zmq.SUBSCRIBE, b'engine.2.stdout')
+    sub.connect(iopub_url)
+    while True:
+        try:
+            idents,msg = session.recv(sub, mode=0)
+        except KeyboardInterrupt:
+            return
+        # ident always length 1 here
+        topic = idents[0]
+        if msg['msg_type'] == 'stream':
+            # stdout/stderr
+            # stream names are in msg['content']['name'], if you want to handle
+            # them differently
+            print "%s: %s" % (topic, msg['content']['data'])
+        elif msg['msg_type'] == 'pyerr':
+            # Python traceback
+            c = msg['content']
+            print topic + ':'
+            for line in c['traceback']:
+                # indent lines
+                print '    ' + line
+
+if __name__ == '__main__':
+    if len(sys.argv) > 1:
+        cf = sys.argv[1]
+    else:
+        # This gets the security file for the default profile:
+        cf = get_security_file('ipcontroller-client.json')
+    main(cf)
  No newline at end of file

docs/examples/parallel/itermapresult.py

@@ -0,0 +1,65 b''
+"""Example of iteration through AsyncMapResults, without waiting for all results
+
+When you call view.map(func, sequence), you will receive a special AsyncMapResult
+object.  These objects are used to reconstruct the results of the split call.
+One feature AsyncResults provide is that they are iterable *immediately*, so
+you can iterate through the actual results as they complete.
+
+This is useful if you submit a large number of tasks that may take some time,
+but want to perform logic on elements in the result, or even abort subsequent
+tasks in cases where you are searching for the first affirmative result.
+
+By default, the results will match the ordering of the submitted sequence, but
+if you call `map(...ordered=False)`, then results will be provided to the iterator
+on a first come first serve basis.
+
+Authors
+-------
+* MinRK
+"""
+import time
+
+from IPython import parallel
+
+# create client & view
+rc = parallel.Client()
+dv = rc[:]
+v = rc.load_balanced_view()
+
+# scatter 'id', so id=0,1,2 on engines 0,1,2
+dv.scatter('id', rc.ids, flatten=True)
+print "Engine IDs: ", dv['id']
+
+# create a Reference to `id`. This will be a different value on each engine
+ref = parallel.Reference('id')
+print "sleeping for `id` seconds on each engine"
+tic = time.time()
+ar = dv.apply(time.sleep, ref)
+for i,r in enumerate(ar):
+    print "%i: %.3f"%(i, time.time()-tic)
+
+def sleep_here(t):
+    import time
+    time.sleep(t)
+    return id,t
+
+# one call per task
+print "running with one call per task"
+amr = v.map(sleep_here, [.01*t for t in range(100)])
+tic = time.time()
+for i,r in enumerate(amr):
+    print "task %i on engine %i: %.3f" % (i, r[0], time.time()-tic)
+
+print "running with four calls per task"
+# with chunksize, we can have four calls per task
+amr = v.map(sleep_here, [.01*t for t in range(100)], chunksize=4)
+tic = time.time()
+for i,r in enumerate(amr):
+    print "task %i on engine %i: %.3f" % (i, r[0], time.time()-tic)
+
+print "running with two calls per task, with unordered results"
+# We can even iterate through faster results first, with ordered=False
+amr = v.map(sleep_here, [.01*t for t in range(100,0,-1)], ordered=False, chunksize=2)
+tic = time.time()
+for i,r in enumerate(amr):
+    print "slept %.2fs on engine %i: %.3f" % (r[1], r[0], time.time()-tic)

IPython/parallel/apps/ipclusterapp.py

@@ -38,7 +38,7 b' from IPython.core.profiledir import ProfileDir'
 from IPython.utils.daemonize import daemonize
 from IPython.utils.importstring import import_item
 from IPython.utils.sysinfo import num_cpus
-from IPython.utils.traitlets import (Int, Unicode, Bool, CFloat, Dict, List,
+from IPython.utils.traitlets import (Int, Unicode, Bool, CFloat, Dict, List, Any,
                                         DottedObjectName)
 
 from IPython.parallel.apps.baseapp import (
@@ -233,6 +233,12 b' class IPClusterEngines(BaseParallelApplication):'
         help="""The number of engines to start. The default is to use one for each
         CPU on your machine""")
 
+    engine_launcher = Any(config=True, help="Deprecated, use engine_launcher_class")
+    def _engine_launcher_changed(self, name, old, new):
+        if isinstance(new, basestring):
+            self.log.warn("WARNING: %s.engine_launcher is deprecated as of 0.12,"
+                    " use engine_launcher_class" % self.__class__.__name__)
+            self.engine_launcher_class = new
     engine_launcher_class = DottedObjectName('LocalEngineSetLauncher',
         config=True,
         help="""The class for launching a set of Engines. Change this value
@@ -249,11 +255,22 b' class IPClusterEngines(BaseParallelApplication):'
             MPIExecEngineSetLauncher : use mpiexec to launch in an MPI environment
             PBSEngineSetLauncher : use PBS (qsub) to submit engines to a batch queue
             SGEEngineSetLauncher : use SGE (qsub) to submit engines to a batch queue
+            LSFEngineSetLauncher : use LSF (bsub) to submit engines to a batch queue
             SSHEngineSetLauncher : use SSH to start the controller
                                 Note that SSH does *not* move the connection files
                                 around, so you will likely have to do this manually
                                 unless the machines are on a shared file system.
             WindowsHPCEngineSetLauncher : use Windows HPC
+
+        If you are using one of IPython's builtin launchers, you can specify just the
+        prefix, e.g:
+
+            c.IPClusterEngines.engine_launcher_class = 'SSH'
+
+        or:
+
+            ipcluster start --engines 'MPIExec'
+
         """
         )
     daemonize = Bool(False, config=True,
@@ -265,9 +282,11 b' class IPClusterEngines(BaseParallelApplication):'
         if new:
             self.log_to_file = True
 
+    early_shutdown = Int(30, config=True, help="The timeout (in seconds)")
+    _stopping = False
+    
     aliases = Dict(engine_aliases)
     flags = Dict(engine_flags)
-    _stopping = False
 
     @catch_config_error
     def initialize(self, argv=None):
@@ -277,7 +296,6 b' class IPClusterEngines(BaseParallelApplication):'
 
     def init_launchers(self):
         self.engine_launcher = self.build_launcher(self.engine_launcher_class, 'EngineSet')
-        self.engine_launcher.on_stop(lambda r: self.loop.stop())
 
     def init_signal(self):
         # Setup signals
@@ -304,13 +322,42 b' class IPClusterEngines(BaseParallelApplication):'
         )
         return launcher
 
+    def engines_started_ok(self):
+        self.log.info("Engines appear to have started successfully")
+        self.early_shutdown = 0
+    
     def start_engines(self):
         self.log.info("Starting %i engines"%self.n)
         self.engine_launcher.start(self.n)
+        self.engine_launcher.on_stop(self.engines_stopped_early)
+        if self.early_shutdown:
+            ioloop.DelayedCallback(self.engines_started_ok, self.early_shutdown*1000, self.loop).start()
+
+    def engines_stopped_early(self, r):
+        if self.early_shutdown and not self._stopping:
+            self.log.error("""
+            Engines shutdown early, they probably failed to connect.
+            
+            Check the engine log files for output.
+            
+            If your controller and engines are not on the same machine, you probably
+            have to instruct the controller to listen on an interface other than localhost.
+            
+            You can set this by adding "--ip='*'" to your ControllerLauncher.controller_args.
+            
+            Be sure to read our security docs before instructing your controller to listen on
+            a public interface.
+            """)
+            self.stop_launchers()
+        
+        return self.engines_stopped(r)
+    
+    def engines_stopped(self, r):
+        return self.loop.stop()
 
     def stop_engines(self):
-        self.log.info("Stopping Engines...")
         if self.engine_launcher.running:
+            self.log.info("Stopping Engines...")
             d = self.engine_launcher.stop()
             return d
         else:
@@ -322,7 +369,7 b' class IPClusterEngines(BaseParallelApplication):'
             self.log.error("IPython cluster: stopping")
             self.stop_engines()
             # Wait a few seconds to let things shut down.
-            dc = ioloop.DelayedCallback(self.loop.stop, 4000, self.loop)
+            dc = ioloop.DelayedCallback(self.loop.stop, 3000, self.loop)
             dc.start()
 
     def sigint_handler(self, signum, frame):
@@ -394,6 +441,13 b' class IPClusterStart(IPClusterEngines):'
     delay = CFloat(1., config=True,
         help="delay (in s) between starting the controller and the engines")
 
+    controller_launcher = Any(config=True, help="Deprecated, use controller_launcher_class")
+    def _controller_launcher_changed(self, name, old, new):
+        if isinstance(new, basestring):
+            # old 0.11-style config
+            self.log.warn("WARNING: %s.controller_launcher is deprecated as of 0.12,"
+                    " use controller_launcher_class" % self.__class__.__name__)
+            self.controller_launcher_class = new
     controller_launcher_class = DottedObjectName('LocalControllerLauncher',
         config=True,
         help="""The class for launching a Controller. Change this value if you want
@@ -408,8 +462,19 b' class IPClusterStart(IPClusterEngines):'
             MPIExecControllerLauncher : use mpiexec to launch engines in an MPI universe
             PBSControllerLauncher : use PBS (qsub) to submit engines to a batch queue
             SGEControllerLauncher : use SGE (qsub) to submit engines to a batch queue
+            LSFControllerLauncher : use LSF (bsub) to submit engines to a batch queue
             SSHControllerLauncher : use SSH to start the controller
             WindowsHPCControllerLauncher : use Windows HPC
+
+        If you are using one of IPython's builtin launchers, you can specify just the
+        prefix, e.g:
+
+            c.IPClusterStart.controller_launcher_class = 'SSH'
+
+        or:
+
+            ipcluster start --controller 'MPIExec'
+
         """
         )
     reset = Bool(False, config=True,
@@ -423,7 +488,11 b' class IPClusterStart(IPClusterEngines):'
         self.controller_launcher = self.build_launcher(self.controller_launcher_class, 'Controller')
         self.engine_launcher = self.build_launcher(self.engine_launcher_class, 'EngineSet')
         self.controller_launcher.on_stop(self.stop_launchers)
-
+    
+    def engines_stopped(self, r):
+        """prevent parent.engines_stopped from stopping everything on engine shutdown"""
+        pass
+    
     def start_controller(self):
         self.controller_launcher.start()
 

IPython/parallel/apps/ipcontrollerapp.py

@@ -55,7 +55,7 b' from IPython.parallel.controller.hub import HubFactory'
 from IPython.parallel.controller.scheduler import TaskScheduler,launch_scheduler
 from IPython.parallel.controller.sqlitedb import SQLiteDB
 
-from IPython.parallel.util import signal_children, split_url, asbytes
+from IPython.parallel.util import signal_children, split_url, asbytes, disambiguate_url
 
 # conditional import of MongoDB backend class
 
@@ -290,13 +290,15 b' class IPControllerApp(BaseParallelApplication):'
         mq = import_item(str(self.mq_class))
         
         hub = self.factory
-        # maybe_inproc = 'inproc://monitor' if self.use_threads else self.monitor_url
+        # disambiguate url, in case of *
+        monitor_url = disambiguate_url(hub.monitor_url)
+        # maybe_inproc = 'inproc://monitor' if self.use_threads else monitor_url
         # IOPub relay (in a Process)
         q = mq(zmq.PUB, zmq.SUB, zmq.PUB, b'N/A',b'iopub')
         q.bind_in(hub.client_info['iopub'])
         q.bind_out(hub.engine_info['iopub'])
         q.setsockopt_out(zmq.SUBSCRIBE, b'')
-        q.connect_mon(hub.monitor_url)
+        q.connect_mon(monitor_url)
         q.daemon=True
         children.append(q)
 
@@ -305,7 +307,7 b' class IPControllerApp(BaseParallelApplication):'
         q.bind_in(hub.client_info['mux'])
         q.setsockopt_in(zmq.IDENTITY, b'mux')
         q.bind_out(hub.engine_info['mux'])
-        q.connect_mon(hub.monitor_url)
+        q.connect_mon(monitor_url)
         q.daemon=True
         children.append(q)
 
@@ -314,7 +316,7 b' class IPControllerApp(BaseParallelApplication):'
         q.bind_in(hub.client_info['control'])
         q.setsockopt_in(zmq.IDENTITY, b'control')
         q.bind_out(hub.engine_info['control'])
-        q.connect_mon(hub.monitor_url)
+        q.connect_mon(monitor_url)
         q.daemon=True
         children.append(q)
         try:
@@ -329,7 +331,7 b' class IPControllerApp(BaseParallelApplication):'
             q.bind_in(hub.client_info['task'][1])
             q.setsockopt_in(zmq.IDENTITY, b'task')
             q.bind_out(hub.engine_info['task'])
-            q.connect_mon(hub.monitor_url)
+            q.connect_mon(monitor_url)
             q.daemon=True
             children.append(q)
         elif scheme == 'none':
@@ -338,7 +340,7 b' class IPControllerApp(BaseParallelApplication):'
         else:
             self.log.info("task::using Python %s Task scheduler"%scheme)
             sargs = (hub.client_info['task'][1], hub.engine_info['task'],
-                                hub.monitor_url, hub.client_info['notification'])
+                                monitor_url, disambiguate_url(hub.client_info['notification']))
             kwargs = dict(logname='scheduler', loglevel=self.log_level,
                             log_url = self.log_url, config=dict(self.config))
             if 'Process' in self.mq_class:

IPython/parallel/apps/launcher.py

@@ -633,7 +633,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
                 d = el.start(user=user, hostname=host)
                 if i==0:
                     self.log.info("Starting SSHEngineSetLauncher: %r" % el.args)
-                self.launchers[host+str(i)] = el
+                self.launchers[ "%s/%i" % (host,i) ] = el
                 dlist.append(d)
         self.notify_start(dlist)
         return dlist

IPython/parallel/client/asyncresult.py

@@ -62,6 +62,7 b' class AsyncResult(object):'
         self._tracker = tracker
         self._ready = False
         self._success = None
+        self._metadata = None
         if len(msg_ids) == 1:
             self._single_result = not isinstance(targets, (list, tuple))
         else:
@@ -231,13 +232,13 b' class AsyncResult(object):'
         else:
             raise TypeError("Invalid key type %r, must be 'int','slice', or 'str'"%type(key))
 
-    @check_ready
     def __getattr__(self, key):
         """getattr maps to getitem for convenient attr access to metadata."""
-        if key not in self._metadata[0].keys():
+        try:
+            return self.__getitem__(key)
+        except (error.TimeoutError, KeyError):
             raise AttributeError("%r object has no attribute %r"%(
                     self.__class__.__name__, key))
-        return self.__getitem__(key)
 
     # asynchronous iterator:
     def __iter__(self):
@@ -261,12 +262,19 b' class AsyncMapResult(AsyncResult):'
     """Class for representing results of non-blocking gathers.
 
     This will properly reconstruct the gather.
+    
+    This class is iterable at any time, and will wait on results as they come.
+    
+    If ordered=False, then the first results to arrive will come first, otherwise
+    results will be yielded in the order they were submitted.
+    
     """
 
-    def __init__(self, client, msg_ids, mapObject, fname=''):
+    def __init__(self, client, msg_ids, mapObject, fname='', ordered=True):
         AsyncResult.__init__(self, client, msg_ids, fname=fname)
         self._mapObject = mapObject
         self._single_result = False
+        self.ordered = ordered
 
     def _reconstruct_result(self, res):
         """Perform the gather on the actual results."""
@@ -274,6 +282,13 b' class AsyncMapResult(AsyncResult):'
 
     # asynchronous iterator:
     def __iter__(self):
+        it = self._ordered_iter if self.ordered else self._unordered_iter
+        for r in it():
+            yield r
+
+    # asynchronous ordered iterator:
+    def _ordered_iter(self):
+        """iterator for results *as they arrive*, preserving submission order."""
         try:
             rlist = self.get(0)
         except error.TimeoutError:
@@ -294,6 +309,42 b' class AsyncMapResult(AsyncResult):'
             for r in rlist:
                 yield r
 
+    # asynchronous unordered iterator:
+    def _unordered_iter(self):
+        """iterator for results *as they arrive*, on FCFS basis, ignoring submission order."""
+        try:
+            rlist = self.get(0)
+        except error.TimeoutError:
+            pending = set(self.msg_ids)
+            while pending:
+                try:
+                    self._client.wait(pending, 1e-3)
+                except error.TimeoutError:
+                    # ignore timeout error, because that only means
+                    # *some* jobs are outstanding
+                    pass
+                # update ready set with those no longer outstanding:
+                ready = pending.difference(self._client.outstanding)
+                # update pending to exclude those that are finished
+                pending = pending.difference(ready)
+                while ready:
+                    msg_id = ready.pop()
+                    ar = AsyncResult(self._client, msg_id, self._fname)
+                    rlist = ar.get()
+                    try:
+                        for r in rlist:
+                            yield r
+                    except TypeError:
+                        # flattened, not a list
+                        # this could get broken by flattened data that returns iterables
+                        # but most calls to map do not expose the `flatten` argument
+                        yield rlist
+        else:
+            # already done
+            for r in rlist:
+                yield r
+
+
 
 class AsyncHubResult(AsyncResult):
     """Class to wrap pending results that must be requested from the Hub.

IPython/parallel/client/remotefunction.py

@@ -46,7 +46,7 b' def remote(view, block=None, **flags):'
     return remote_function
 
 @skip_doctest
-def parallel(view, dist='b', block=None, **flags):
+def parallel(view, dist='b', block=None, ordered=True, **flags):
     """Turn a function into a parallel remote function.
 
     This method can be used for map:
@@ -57,7 +57,7 b" def parallel(view, dist='b', block=None, **flags):"
     """
 
     def parallel_function(f):
-        return ParallelFunction(view, f, dist=dist, block=block, **flags)
+        return ParallelFunction(view, f, dist=dist, block=block, ordered=ordered, **flags)
     return parallel_function
 
 #--------------------------------------------------------------------------
@@ -122,15 +122,19 b' class ParallelFunction(RemoteFunction):'
         to use the current `block` attribute of `view`
     chunksize : int or None
         The size of chunk to use when breaking up sequences in a load-balanced manner
+    ordered : bool [default: True]
+        Whether 
     **flags : remaining kwargs are passed to View.temp_flags
     """
 
     chunksize=None
+    ordered=None
     mapObject=None
 
-    def __init__(self, view, f, dist='b', block=None, chunksize=None, **flags):
+    def __init__(self, view, f, dist='b', block=None, chunksize=None, ordered=True, **flags):
         super(ParallelFunction, self).__init__(view, f, block=block, **flags)
         self.chunksize = chunksize
+        self.ordered = ordered
 
         mapClass = Map.dists[dist]
         self.mapObject = mapClass()
@@ -186,7 +190,10 b' class ParallelFunction(RemoteFunction):'
 
             msg_ids.append(ar.msg_ids[0])
 
-        r = AsyncMapResult(self.view.client, msg_ids, self.mapObject, fname=self.func.__name__)
+        r = AsyncMapResult(self.view.client, msg_ids, self.mapObject, 
+                            fname=self.func.__name__,
+                            ordered=self.ordered
+                        )
 
         if self.block:
             try:

IPython/parallel/client/view.py

@@ -992,7 +992,7 b' class LoadBalancedView(View):'
     @spin_after
     @save_ids
     def map(self, f, *sequences, **kwargs):
-        """view.map(f, *sequences, block=self.block, chunksize=1) => list|AsyncMapResult
+        """view.map(f, *sequences, block=self.block, chunksize=1, ordered=True) => list|AsyncMapResult
 
         Parallel version of builtin `map`, load-balanced by this View.
 
@@ -1009,14 +1009,20 b' class LoadBalancedView(View):'
             function to be mapped
         *sequences: one or more sequences of matching length
             the sequences to be distributed and passed to `f`
-        block : bool
-            whether to wait for the result or not [default self.block]
+        block : bool [default self.block]
+            whether to wait for the result or not
         track : bool
             whether to create a MessageTracker to allow the user to
             safely edit after arrays and buffers during non-copying
             sends.
-        chunksize : int
-            how many elements should be in each task [default 1]
+        chunksize : int [default 1]
+            how many elements should be in each task.
+        ordered : bool [default True]
+            Whether the results should be gathered as they arrive, or enforce
+            the order of submission.
+            
+            Only applies when iterating through AsyncMapResult as results arrive.
+            Has no effect when block=True.
 
         Returns
         -------
@@ -1034,6 +1040,7 b' class LoadBalancedView(View):'
         # default
         block = kwargs.get('block', self.block)
         chunksize = kwargs.get('chunksize', 1)
+        ordered = kwargs.get('ordered', True)
 
         keyset = set(kwargs.keys())
         extra_keys = keyset.difference_update(set(['block', 'chunksize']))
@@ -1042,7 +1049,7 b' class LoadBalancedView(View):'
 
         assert len(sequences) > 0, "must have some sequences to map onto!"
 
-        pf = ParallelFunction(self, f, block=block,  chunksize=chunksize)
+        pf = ParallelFunction(self, f, block=block, chunksize=chunksize, ordered=ordered)
         return pf.map(*sequences)
 
 __all__ = ['LoadBalancedView', 'DirectView']

IPython/parallel/engine/engine.py

@@ -214,6 +214,14 b' class EngineFactory(RegistrationFactory):'
 
     def abort(self):
         self.log.fatal("Registration timed out after %.1f seconds"%self.timeout)
+        if self.url.startswith('127.'):
+            self.log.fatal("""
+            If the controller and engines are not on the same machine,
+            you will have to instruct the controller to listen on an external IP (in ipcontroller_config.py):
+                c.HubFactory.ip='*' # for all interfaces, internal and external
+                c.HubFactory.ip='192.168.1.101' # or any interface that the engines can see
+            or tunnel connections via ssh.
+            """)
         self.session.send(self.registrar, "unregistration_request", content=dict(id=self.id))
         time.sleep(1)
         sys.exit(255)

IPython/parallel/tests/test_asyncresult.py

@@ -70,4 +70,46 b' class AsyncResultTest(ClusterTestCase):'
         self.assertEquals(sorted(d.keys()), sorted(self.client.ids))
         for eid,r in d.iteritems():
             self.assertEquals(r, 5)
+    
+    def test_list_amr(self):
+        ar = self.client.load_balanced_view().map_async(wait, [0.1]*5)
+        rlist = list(ar)
+    
+    def test_getattr(self):
+        ar = self.client[:].apply_async(wait, 0.5)
+        self.assertRaises(AttributeError, lambda : ar._foo)
+        self.assertRaises(AttributeError, lambda : ar.__length_hint__())
+        self.assertRaises(AttributeError, lambda : ar.foo)
+        self.assertRaises(AttributeError, lambda : ar.engine_id)
+        self.assertFalse(hasattr(ar, '__length_hint__'))
+        self.assertFalse(hasattr(ar, 'foo'))
+        self.assertFalse(hasattr(ar, 'engine_id'))
+        ar.get(5)
+        self.assertRaises(AttributeError, lambda : ar._foo)
+        self.assertRaises(AttributeError, lambda : ar.__length_hint__())
+        self.assertRaises(AttributeError, lambda : ar.foo)
+        self.assertTrue(isinstance(ar.engine_id, list))
+        self.assertEquals(ar.engine_id, ar['engine_id'])
+        self.assertFalse(hasattr(ar, '__length_hint__'))
+        self.assertFalse(hasattr(ar, 'foo'))
+        self.assertTrue(hasattr(ar, 'engine_id'))
+
+    def test_getitem(self):
+        ar = self.client[:].apply_async(wait, 0.5)
+        self.assertRaises(TimeoutError, lambda : ar['foo'])
+        self.assertRaises(TimeoutError, lambda : ar['engine_id'])
+        ar.get(5)
+        self.assertRaises(KeyError, lambda : ar['foo'])
+        self.assertTrue(isinstance(ar['engine_id'], list))
+        self.assertEquals(ar.engine_id, ar['engine_id'])
+    
+    def test_single_result(self):
+        ar = self.client[-1].apply_async(wait, 0.5)
+        self.assertRaises(TimeoutError, lambda : ar['foo'])
+        self.assertRaises(TimeoutError, lambda : ar['engine_id'])
+        self.assertTrue(ar.get(5) == 0.5)
+        self.assertTrue(isinstance(ar['engine_id'], int))
+        self.assertTrue(isinstance(ar.engine_id, int))
+        self.assertEquals(ar.engine_id, ar['engine_id'])
+
 

IPython/parallel/tests/test_lbview.py

@@ -58,6 +58,44 b' class TestLoadBalancedView(ClusterTestCase):'
         r = self.view.map_sync(f, data)
         self.assertEquals(r, map(f, data))
 
+    def test_map_unordered(self):
+        def f(x):
+            return x**2
+        def slow_f(x):
+            import time
+            time.sleep(0.05*x)
+            return x**2
+        data = range(16,0,-1)
+        reference = map(f, data)
+        
+        amr = self.view.map_async(slow_f, data, ordered=False)
+        self.assertTrue(isinstance(amr, pmod.AsyncMapResult))
+        # check individual elements, retrieved as they come
+        # list comprehension uses __iter__
+        astheycame = [ r for r in amr ]
+        # Ensure that at least one result came out of order:
+        self.assertNotEquals(astheycame, reference, "should not have preserved order")
+        self.assertEquals(sorted(astheycame, reverse=True), reference, "result corrupted")
+
+    def test_map_ordered(self):
+        def f(x):
+            return x**2
+        def slow_f(x):
+            import time
+            time.sleep(0.05*x)
+            return x**2
+        data = range(16,0,-1)
+        reference = map(f, data)
+        
+        amr = self.view.map_async(slow_f, data)
+        self.assertTrue(isinstance(amr, pmod.AsyncMapResult))
+        # check individual elements, retrieved as they come
+        # list(amr) uses __iter__
+        astheycame = list(amr)
+        # Ensure that results came in order
+        self.assertEquals(astheycame, reference)
+        self.assertEquals(amr.result, reference)
+    
     def test_abort(self):
         view = self.view
         ar = self.client[:].apply_async(time.sleep, .5)

IPython/zmq/heartbeat.py

@@ -31,15 +31,14 b' class Heartbeat(Thread):'
     def __init__(self, context, addr=(LOCALHOST, 0)):
         Thread.__init__(self)
         self.context = context
-        self.addr = addr
-        self.ip = addr[0]
-        self.port = addr[1]
+        self.ip, self.port = addr
         if self.port == 0:
             s = socket.socket()
-            s.bind(self.addr)
+            # '*' means all interfaces to 0MQ, which is '' to socket.socket
+            s.bind(('' if self.ip == '*' else self.ip, 0))
             self.port = s.getsockname()[1]
             s.close()
-            self.addr = (self.ip, self.port)
+        self.addr = (self.ip, self.port)
         self.daemon = True
 
     def run(self):

IPython/zmq/kernelmanager.py

@@ -710,6 +710,9 b' class KernelManager(HasTraits):'
     # The addresses for the communication channels.
     connection_file = Unicode('')
     ip = Unicode(LOCALHOST)
+    def _ip_changed(self, name, old, new):
+        if new == '*':
+            self.ip = '0.0.0.0'
     shell_port = Int(0)
     iopub_port = Int(0)
     stdin_port = Int(0)

docs/source/parallel/dag_dependencies.txt

@@ -10,7 +10,7 b' for working with Graphs is NetworkX_. Here, we will walk through a demo mapping'
 a nx DAG to task dependencies.
 
 The full script that runs this demo can be found in
-:file:`docs/examples/newparallel/dagdeps.py`.
+:file:`docs/examples/parallel/dagdeps.py`.
 
 Why are DAGs good for task dependencies?
 ----------------------------------------
@@ -30,7 +30,8 b' A Sample DAG'
 
 Here, we have a very simple 5-node DAG:
 
-.. figure:: simpledag.*
+.. figure:: figs/simpledag.*
+    :width: 600px
 
 With NetworkX, an arrow is just a fattened bit on the edge. Here, we can see that task 0
 depends on nothing, and can run immediately. 1 and 2 depend on 0; 3 depends on
@@ -80,7 +81,7 b' The code to generate the simple DAG:'
 For demonstration purposes, we have a function that generates a random DAG with a given
 number of nodes and edges.
 
-.. literalinclude:: ../../examples/newparallel/dagdeps.py
+.. literalinclude:: ../../examples/parallel/dagdeps.py
     :language: python
     :lines: 20-36
 
@@ -117,11 +118,13 b' on which it depends:'
     In [6]: results = {}
     
     In [7]: for node in G.topological_sort():
-       ...:     # get list of AsyncResult objects from nodes
-       ...:     # leading into this one as dependencies
-       ...:     deps = [ results[n] for n in G.predecessors(node) ]
-       ...:     # submit and store AsyncResult object
-       ...:     results[node] = view.apply_with_flags(jobs[node], after=deps, block=False)
+       ...:    # get list of AsyncResult objects from nodes
+       ...:    # leading into this one as dependencies
+       ...:    deps = [ results[n] for n in G.predecessors(node) ]
+       ...:    # submit and store AsyncResult object
+       ...:    with view.temp_flags(after=deps, block=False):
+       ...:         results[node] = view.apply_with_flags(jobs[node])
+
 
 Now that we have submitted all the jobs, we can wait for the results:
 
@@ -137,7 +140,7 b' These objects store a variety of metadata about each task, including various tim'
 We can validate that the dependencies were respected by checking that each task was
 started after all of its predecessors were completed:
 
-.. literalinclude:: ../../examples/newparallel/dagdeps.py
+.. literalinclude:: ../../examples/parallel/dagdeps.py
     :language: python
     :lines: 64-70
 
@@ -155,16 +158,17 b' will be at the top, and quick, small tasks will be at the bottom.'
     In [12]: pos = {}; colors = {}
     
     In [12]: for node in G:
-        ...:    md = results[node].metadata
-        ...:    start = date2num(md.started)
-        ...:    runtime = date2num(md.completed) - start
-        ...:    pos[node] = (start, runtime)
-        ...:    colors[node] = md.engine_id
+       ....:    md = results[node].metadata
+       ....:    start = date2num(md.started)
+       ....:    runtime = date2num(md.completed) - start
+       ....:    pos[node] = (start, runtime)
+       ....:    colors[node] = md.engine_id
     
     In [13]: nx.draw(G, pos, node_list=colors.keys(), node_color=colors.values(),
-        ...:    cmap=gist_rainbow)
+       ....:    cmap=gist_rainbow)
 
-.. figure:: dagdeps.*
+.. figure:: figs/dagdeps.*
+    :width: 600px
 
     Time started on x, runtime on y, and color-coded by engine-id (in this case there
     were four engines). Edges denote dependencies.

docs/source/parallel/parallel_demos.txt

@@ -4,7 +4,7 b' Parallel examples'
 
 .. note::
 
-    Performance numbers from ``IPython.kernel``, not newparallel.
+    Performance numbers from ``IPython.kernel``, not new ``IPython.parallel``.
 
 In this section we describe two more involved examples of using an IPython
 cluster to perform a parallel computation. In these examples, we will be using
@@ -27,7 +27,7 b' million digits.'
 
 In both the serial and parallel calculation we will be using functions defined
 in the :file:`pidigits.py` file, which is available in the
-:file:`docs/examples/newparallel` directory of the IPython source distribution.
+:file:`docs/examples/parallel` directory of the IPython source distribution.
 These functions provide basic facilities for working with the digits of pi and
 can be loaded into IPython by putting :file:`pidigits.py` in your current
 working directory and then doing:
@@ -75,7 +75,7 b' The resulting plot of the single digit counts shows that each digit occurs'
 approximately 1,000 times, but that with only 10,000 digits the
 statistical fluctuations are still rather large:
 
-.. image:: single_digits.*
+.. image:: figs/single_digits.*
 
 It is clear that to reduce the relative fluctuations in the counts, we need
 to look at many more digits of pi. That brings us to the parallel calculation.
@@ -101,13 +101,13 b' compute the two digit counts for the digits in a single file. Then in a final'
 step the counts from each engine will be added up. To perform this
 calculation, we will need two top-level functions from :file:`pidigits.py`:
 
-.. literalinclude:: ../../examples/newparallel/pidigits.py
+.. literalinclude:: ../../examples/parallel/pi/pidigits.py
    :language: python
    :lines: 47-62
 
 We will also use the :func:`plot_two_digit_freqs` function to plot the
 results. The code to run this calculation in parallel is contained in
-:file:`docs/examples/newparallel/parallelpi.py`. This code can be run in parallel
+:file:`docs/examples/parallel/parallelpi.py`. This code can be run in parallel
 using IPython by following these steps:
 
 1. Use :command:`ipcluster` to start 15 engines. We used an 8 core (2 quad 
@@ -188,7 +188,7 b' most likely and that "06" and "07" are least likely. Further analysis would'
 show that the relative size of the statistical fluctuations have decreased
 compared to the 10,000 digit calculation.
 
-.. image:: two_digit_counts.*
+.. image:: figs/two_digit_counts.*
 
 
 Parallel options pricing
@@ -209,12 +209,12 b' simulation of the underlying asset price. In this example we use this approach'
 to price both European and Asian (path dependent) options for various strike
 prices and volatilities.
 
-The code for this example can be found in the :file:`docs/examples/newparallel`
+The code for this example can be found in the :file:`docs/examples/parallel`
 directory of the IPython source. The function :func:`price_options` in
 :file:`mcpricer.py` implements the basic Monte Carlo pricing algorithm using
 the NumPy package and is shown here:
 
-.. literalinclude:: ../../examples/newparallel/mcpricer.py
+.. literalinclude:: ../../examples/parallel/options/mcpricer.py
    :language: python
 
 To run this code in parallel, we will use IPython's :class:`LoadBalancedView` class,
@@ -222,21 +222,21 b' which distributes work to the engines using dynamic load balancing. This'
 view is a wrapper of the :class:`Client` class shown in
 the previous example. The parallel calculation using :class:`LoadBalancedView` can
 be found in the file :file:`mcpricer.py`. The code in this file creates a
-:class:`TaskClient` instance and then submits a set of tasks using
-:meth:`TaskClient.run` that calculate the option prices for different
+:class:`LoadBalancedView` instance and then submits a set of tasks using
+:meth:`LoadBalancedView.apply` that calculate the option prices for different
 volatilities and strike prices. The results are then plotted as a 2D contour
 plot using Matplotlib.
 
-.. literalinclude:: ../../examples/newparallel/mcdriver.py
+.. literalinclude:: ../../examples/parallel/options/mckernel.py
    :language: python
 
 To use this code, start an IPython cluster using :command:`ipcluster`, open
-IPython in the pylab mode with the file :file:`mcdriver.py` in your current
+IPython in the pylab mode with the file :file:`mckernel.py` in your current
 working directory and then type:
 
 .. sourcecode:: ipython
 
-    In [7]: run mcdriver.py
+    In [7]: run mckernel.py
     Submitted tasks:  [0, 1, 2, ...]
 
 Once all the tasks have finished, the results can be plotted using the
@@ -257,9 +257,9 b' entire calculation (10 strike prices, 10 volatilities, 100,000 paths for each)'
 took 30 seconds in parallel, giving a speedup of 7.7x, which is comparable
 to the speedup observed in our previous example.
 
-.. image:: asian_call.*
+.. image:: figs/asian_call.*
 
-.. image:: asian_put.*
+.. image:: figs/asian_put.*
 
 Conclusion
 ==========
@@ -280,5 +280,5 b' parallel architecture that have been demonstrated:'
 
 .. note::
 
-    The newparallel code has never been run on Windows HPC Server, so the last
+    The new parallel code has never been run on Windows HPC Server, so the last
     conclusion is untested.

docs/source/parallel/parallel_intro.txt

@@ -60,6 +60,10 b' the ``I`` in IPython. The following are some example usage cases for IPython:'
 Architecture overview
 =====================
 
+.. figure:: figs/wideView.png
+    :width: 300px
+
+
 The IPython architecture consists of four components:
 
 * The IPython engine.
@@ -99,7 +103,7 b' same machine as the Hub, but can be run anywhere from local threads or on remote'
 The controller also provides a single point of contact for users who wish to
 utilize the engines connected to the controller. There are different ways of
 working with a controller. In IPython, all of these models are implemented via
-the client's :meth:`.View.apply` method, with various arguments, or
+the :meth:`.View.apply` method, after
 constructing :class:`.View` objects to represent subsets of engines. The two
 primary models for interacting with engines are:
 
@@ -181,6 +185,34 b' ipcontroller-client.json'
     but since the controller may listen on different ports for clients and
     engines, it is stored separately.
 
+ipcontroller-client.json will look something like this, under default localhost
+circumstances:
+
+.. sourcecode:: python
+
+    {
+      "url":"tcp:\/\/127.0.0.1:54424",
+      "exec_key":"a361fe89-92fc-4762-9767-e2f0a05e3130",
+      "ssh":"",
+      "location":"10.19.1.135"
+    }
+
+If, however, you are running the controller on a work node on a cluster, you will likely
+need to use ssh tunnels to connect clients from your laptop to it.  You will also
+probably need to instruct the controller to listen for engines coming from other work nodes
+on the cluster.  An example of ipcontroller-client.json, as created by::
+
+    $> ipcontroller --ip=0.0.0.0 --ssh=login.mycluster.com
+
+
+.. sourcecode:: python
+
+    {
+      "url":"tcp:\/\/*:54424",
+      "exec_key":"a361fe89-92fc-4762-9767-e2f0a05e3130",
+      "ssh":"login.mycluster.com",
+      "location":"10.0.0.2"
+    }
 More details of how these JSON files are used are given below.
 
 A detailed description of the security model and its implementation in IPython
@@ -248,7 +280,7 b' then you would connect to it with:'
 
 .. sourcecode:: ipython
 
-    In [2]: c = Client(sshserver='myhub.example.com')
+    In [2]: c = Client('/path/to/my/ipcontroller-client.json', sshserver='[e-mail]')
 
 Where 'myhub.example.com' is the url or IP address of the machine on
 which the Hub process is running (or another machine that has direct access to the Hub's ports).

docs/source/parallel/parallel_multiengine.txt

@@ -24,8 +24,8 b' the :command:`ipcluster` command::'
 For more detailed information about starting the controller and engines, see
 our :ref:`introduction <parallel_overview>` to using IPython for parallel computing.
 
-Creating a ``Client`` instance
-==============================
+Creating a ``DirectView`` instance
+==================================
 
 The first step is to import the IPython :mod:`IPython.parallel`
 module and then create a :class:`.Client` instance:
@@ -117,10 +117,10 b' two decorators:'
 .. sourcecode:: ipython
 
     In [10]: @dview.remote(block=True)
-        ...: def getpid():
-        ...:     import os
-        ...:     return os.getpid()
-        ...: 
+       ....: def getpid():
+       ....:     import os
+       ....:     return os.getpid()
+       ....: 
 
     In [11]: getpid()
     Out[11]: [12345, 12346, 12347, 12348]
@@ -135,8 +135,8 b' operations and distribute them, reconstructing the result.'
     In [13]: A = np.random.random((64,48))
     
     In [14]: @dview.parallel(block=True)
-        ...: def pmul(A,B):
-        ...:     return A*B
+       ....: def pmul(A,B):
+       ....:     return A*B
     
     In [15]: C_local = A*A
     
@@ -183,6 +183,8 b' dv.track : bool'
     This is primarily useful for non-copying sends of numpy arrays that you plan to
     edit in-place.  You need to know when it becomes safe to edit the buffer
     without corrupting the message.
+dv.targets : int, list of ints
+    which targets this view is associated with.
 
 
 Creating a view is simple: index-access on a client creates a :class:`.DirectView`.
@@ -260,10 +262,10 b' local Python/IPython session:'
     
     # define our function
     In [6]: def wait(t):
-       ...:     import time
-       ...:     tic = time.time()
-       ...:     time.sleep(t)
-       ...:     return time.time()-tic
+      ....:     import time
+      ....:     tic = time.time()
+      ....:     time.sleep(t)
+      ....:     return time.time()-tic
     
     # In non-blocking mode
     In [7]: ar = dview.apply_async(wait, 2)
@@ -326,7 +328,7 b' and blocks until all of the associated results are ready:'
 The ``block`` and ``targets`` keyword arguments and attributes
 --------------------------------------------------------------
 
-Most DirectView methods (excluding :meth:`apply` and :meth:`map`) accept ``block`` and
+Most DirectView methods (excluding :meth:`apply`) accept ``block`` and
 ``targets`` as keyword arguments. As we have seen above, these keyword arguments control the
 blocking mode and which engines the command is applied to. The :class:`View` class also has
 :attr:`block` and :attr:`targets` attributes that control the default behavior when the keyword
@@ -362,11 +364,6 b' The :attr:`block` and :attr:`targets` instance attributes of the'
 Parallel magic commands
 -----------------------
 
-.. warning::
-
-    The magics have not been changed to work with the zeromq system. The
-    magics do work, but *do not* print stdin/out like they used to in IPython.kernel.
-
 We provide a few IPython magic commands (``%px``, ``%autopx`` and ``%result``)
 that make it more pleasant to execute Python commands on the engines
 interactively. These are simply shortcuts to :meth:`execute` and
@@ -376,9 +373,6 b' Python command on the engines specified by the :attr:`targets` attribute of the'
 
 .. sourcecode:: ipython
 
-    # load the parallel magic extension:
-    In [21]: %load_ext parallelmagic
-    
     # Create a DirectView for all targets
     In [22]: dv = rc[:]
     
@@ -387,10 +381,10 b' Python command on the engines specified by the :attr:`targets` attribute of the'
 
     In [24]: dv.block=True
 
-    In [25]: import numpy
-
-    In [26]: %px import numpy
-    Parallel execution on engines: [0, 1, 2, 3]
+    # import numpy here and everywhere
+    In [25]: with dv.sync_imports():
+       ....:    import numpy
+    importing numpy on engine(s)
 
     In [27]: %px a = numpy.random.rand(2,2)
     Parallel execution on engines: [0, 1, 2, 3]
@@ -400,10 +394,10 b' Python command on the engines specified by the :attr:`targets` attribute of the'
 
     In [28]: dv['ev']
     Out[28]: [ array([ 1.09522024, -0.09645227]), 
-               array([ 1.21435496, -0.35546712]), 
-               array([ 0.72180653,  0.07133042]), 
-               array([  1.46384341e+00,   1.04353244e-04])
-             ]
+       ....:   array([ 1.21435496, -0.35546712]), 
+       ....:   array([ 0.72180653,  0.07133042]), 
+       ....:   array([  1.46384341,   1.04353244e-04])
+       ....: ]
 
 The ``%result`` magic gets the most recent result, or takes an argument
 specifying the index of the result to be requested. It is simply a shortcut to the
@@ -415,9 +409,9 b' specifying the index of the result to be requested. It is simply a shortcut to t'
     
     In [30]: %result
     Out[30]: [ [ 1.28167017  0.14197338],
-                [-0.14093616  1.27877273],
-                [-0.37023573  1.06779409],
-                [ 0.83664764 -0.25602658] ]
+       ....:    [-0.14093616  1.27877273],
+       ....:    [-0.37023573  1.06779409],
+       ....:    [ 0.83664764 -0.25602658] ]
 
 The ``%autopx`` magic switches to a mode where everything you type is executed
 on the engines given by the :attr:`targets` attribute:
@@ -452,9 +446,9 b' on the engines given by the :attr:`targets` attribute:'
     
     In [37]: dv['ans']
     Out[37]: [ 'Average max eigenvalue is:  10.1387247332',
-               'Average max eigenvalue is:  10.2076902286',
-               'Average max eigenvalue is:  10.1891484655',
-               'Average max eigenvalue is:  10.1158837784',]
+       ....:   'Average max eigenvalue is:  10.2076902286',
+       ....:   'Average max eigenvalue is:  10.1891484655',
+       ....:   'Average max eigenvalue is:  10.1158837784',]
 
 
 Moving Python objects around
@@ -522,7 +516,7 b' follow that terminology. However, it is important to remember that in'
 IPython's :class:`Client` class, :meth:`scatter` is from the
 interactive IPython session to the engines and :meth:`gather` is from the
 engines back to the interactive IPython session. For scatter/gather operations
-between engines, MPI should be used:
+between engines, MPI, pyzmq, or some other direct interconnect should be used.
 
 .. sourcecode:: ipython
 
@@ -568,7 +562,7 b" created by a DirectView's :meth:`sync_imports` method:"
 .. sourcecode:: ipython
 
     In [69]: with dview.sync_imports():
-        ...:     import numpy
+       ....:     import numpy
     importing numpy on engine(s)
 
 Any imports made inside the block will also be performed on the view's engines.
@@ -588,15 +582,15 b' execution, and will fail with an UnmetDependencyError.'
     In [69]: from IPython.parallel import require
 
     In [70]: @require('re'):
-        ...: def findall(pat, x):
-        ...:     # re is guaranteed to be available
-        ...:     return re.findall(pat, x)
+       ....: def findall(pat, x):
+       ....:     # re is guaranteed to be available
+       ....:     return re.findall(pat, x)
           
     # you can also pass modules themselves, that you already have locally:
     In [71]: @require(time):
-        ...: def wait(t):
-        ...:     time.sleep(t)
-        ...:     return t
+       ....: def wait(t):
+       ....:     time.sleep(t)
+       ....:     return t
 
 .. _parallel_exceptions:
 

docs/source/parallel/parallel_process.txt

@@ -141,9 +141,39 b' Using various batch systems with :command:`ipcluster`'
 
 :command:`ipcluster` has a notion of Launchers that can start controllers
 and engines with various remote execution schemes.  Currently supported
-models include :command:`ssh`, :command:`mpiexec`, PBS-style (Torque, SGE),
+models include :command:`ssh`, :command:`mpiexec`, PBS-style (Torque, SGE, LSF),
 and Windows HPC Server.
 
+In general, these are configured by the :attr:`IPClusterEngines.engine_set_launcher_class`,
+and :attr:`IPClusterStart.controller_launcher_class` configurables, which can be the
+fully specified object name (e.g. ``'IPython.parallel.apps.launcher.LocalControllerLauncher'``),
+but if you are using IPython's builtin launchers, you can specify just the class name,
+or even just the prefix e.g:
+
+.. sourcecode:: python
+
+    c.IPClusterEngines.engine_launcher_class = 'SSH'
+    # equivalent to
+    c.IPClusterEngines.engine_launcher_class = 'SSHEngineSetLauncher'
+    # both of which expand to
+    c.IPClusterEngines.engine_launcher_class = 'IPython.parallel.apps.launcher.SSHEngineSetLauncher'
+
+The shortest form being of particular use on the command line, where all you need to do to
+get an IPython cluster running with engines started with MPI is:
+
+.. sourcecode:: bash
+
+    $> ipcluster start --engines=MPIExec
+
+Assuming that the default MPI config is sufficient.
+
+.. note::
+
+    shortcuts for builtin launcher names were added in 0.12, as was the ``_class`` suffix
+    on the configurable names.  If you use the old 0.11 names (e.g. ``engine_set_launcher``),
+    they will still work, but you will get a deprecation warning that the name has changed.
+
+
 .. note::
 
     The Launchers and configuration are designed in such a way that advanced
@@ -170,7 +200,7 b' There, instruct ipcluster to use the MPIExec launchers by adding the lines:'
 
 .. sourcecode:: python
 
-    c.IPClusterEngines.engine_launcher = 'IPython.parallel.apps.launcher.MPIExecEngineSetLauncher'
+    c.IPClusterEngines.engine_launcher_class = 'MPIExecEngineSetLauncher'
 
 If the default MPI configuration is correct, then you can now start your cluster, with::
 
@@ -185,7 +215,7 b' If you have a reason to also start the Controller with mpi, you can specify:'
 
 .. sourcecode:: python
 
-    c.IPClusterStart.controller_launcher = 'IPython.parallel.apps.launcher.MPIExecControllerLauncher'
+    c.IPClusterStart.controller_launcher_class = 'MPIExecControllerLauncher'
 
 .. note::
 
@@ -226,10 +256,8 b' and engines:'
 
 .. sourcecode:: python
 
-    c.IPClusterStart.controller_launcher = \
-        'IPython.parallel.apps.launcher.PBSControllerLauncher'
-    c.IPClusterEngines.engine_launcher = \
-        'IPython.parallel.apps.launcher.PBSEngineSetLauncher'
+    c.IPClusterStart.controller_launcher_class = 'PBSControllerLauncher'
+    c.IPClusterEngines.engine_launcher_class = 'PBSEngineSetLauncher'
 
 .. note::
 
@@ -355,12 +383,11 b' To use this mode, select the SSH launchers in :file:`ipcluster_config.py`:'
 
 .. sourcecode:: python
 
-    c.IPClusterEngines.engine_launcher = \
-        'IPython.parallel.apps.launcher.SSHEngineSetLauncher'
+    c.IPClusterEngines.engine_launcher_class = 'SSHEngineSetLauncher'
     # and if the Controller is also to be remote:
-    c.IPClusterStart.controller_launcher = \
-        'IPython.parallel.apps.launcher.SSHControllerLauncher'
-    
+    c.IPClusterStart.controller_launcher_class = 'SSHControllerLauncher'
+
+
 
 The controller's remote location and configuration can be specified:
 

docs/source/parallel/parallel_task.txt

@@ -29,8 +29,8 b' the :command:`ipcluster` command::'
 For more detailed information about starting the controller and engines, see
 our :ref:`introduction <parallel_overview>` to using IPython for parallel computing.
 
-Creating a ``Client`` instance
-==============================
+Creating a ``LoadBalancedView`` instance
+========================================
 
 The first step is to import the IPython :mod:`IPython.parallel`
 module and then create a :class:`.Client` instance, and we will also be using
@@ -87,12 +87,12 b' To load-balance :meth:`map`,simply use a LoadBalancedView:'
     
     In [62]: lview.block = True
     
-	In [63]: serial_result = map(lambda x:x**10, range(32))
+    In [63]: serial_result = map(lambda x:x**10, range(32))
 
-	In [64]: parallel_result = lview.map(lambda x:x**10, range(32))
+    In [64]: parallel_result = lview.map(lambda x:x**10, range(32))
 
-	In [65]: serial_result==parallel_result
-	Out[65]: True
+    In [65]: serial_result==parallel_result
+    Out[65]: True
 
 Parallel function decorator
 ---------------------------
@@ -111,6 +111,27 b' that turns any Python function into a parallel function:'
     In [11]: f.map(range(32))    # this is done in parallel
     Out[11]: [0.0,10.0,160.0,...]
 
+.. _parallel_taskmap:
+
+Map results are iterable!
+-------------------------
+
+When an AsyncResult object actually maps multiple results (e.g. the :class:`~AsyncMapResult`
+object), you can actually iterate through them, and act on the results as they arrive:
+
+.. literalinclude:: ../../examples/parallel/itermapresult.py
+    :language: python
+    :lines: 9-34
+
+.. seealso::
+
+    When AsyncResult or the AsyncMapResult don't provide what you need (for instance,
+    handling individual results as they arrive, but with metadata), you can always
+    just split the original result's ``msg_ids`` attribute, and handle them as you like.
+    
+    For an example of this, see :file:`docs/examples/parallel/customresult.py`
+    
+
 .. _parallel_dependencies:
 
 Dependencies
@@ -157,8 +178,8 b' you specify are importable:'
 .. sourcecode:: ipython
 
     In [10]: @require('numpy', 'zmq')
-        ...: def myfunc():
-        ...:     return dostuff()
+       ....: def myfunc():
+       ....:     return dostuff()
 
 Now, any time you apply :func:`myfunc`, the task will only run on a machine that has
 numpy and pyzmq available, and when :func:`myfunc` is called, numpy and zmq will be imported.
@@ -175,16 +196,16 b' will be assigned to another engine. If the dependency returns *anything other th'
 .. sourcecode:: ipython
 
     In [10]: def platform_specific(plat):
-        ...:    import sys
-        ...:    return sys.platform == plat
+       ....:    import sys
+       ....:    return sys.platform == plat
     
     In [11]: @depend(platform_specific, 'darwin')
-        ...: def mactask():
-        ...:    do_mac_stuff()
+       ....: def mactask():
+       ....:    do_mac_stuff()
 
     In [12]: @depend(platform_specific, 'nt')
-        ...: def wintask():
-        ...:    do_windows_stuff()
+       ....: def wintask():
+       ....:    do_windows_stuff()
     
 In this case, any time you apply ``mytask``, it will only run on an OSX machine.
 ``@depend`` is just like ``apply``, in that it has a ``@depend(f,*args,**kwargs)``
@@ -200,7 +221,7 b' the :class:`dependent` object that the decorators use:'
 .. sourcecode::ipython
 
     In [13]: def mytask(*args):
-        ...:    dostuff()
+       ....:    dostuff()
     
     In [14]: mactask = dependent(mytask, platform_specific, 'darwin')
     # this is the same as decorating the declaration of mytask with @depend
@@ -213,8 +234,8 b' the :class:`dependent` object that the decorators use:'
     
     # is equivalent to:
     In [17]: @depend(g, *dargs, **dkwargs)
-        ...: def t(a,b,c):
-        ...:     # contents of f
+       ....: def t(a,b,c):
+       ....:     # contents of f
 
 Graph Dependencies
 ------------------
@@ -278,10 +299,11 b' you can skip using Dependency objects, and just pass msg_ids or AsyncResult obje'
 
     In [16]: ar2 = lview.apply(f2)
 
-    In [17]: ar3 = lview.apply_with_flags(f3, after=[ar,ar2])
-
-    In [17]: ar4 = lview.apply_with_flags(f3, follow=[ar], timeout=2.5)
+    In [17]: with lview.temp_flags(after=[ar,ar2]):
+       ....:    ar3 = lview.apply(f3)
 
+    In [18]: with lview.temp_flags(follow=[ar], timeout=2.5)
+       ....:    ar4 = lview.apply(f3)
 
 .. seealso::
 
@@ -291,8 +313,6 b' you can skip using Dependency objects, and just pass msg_ids or AsyncResult obje'
     onto task dependencies.
 
 
-
-
 Impossible Dependencies
 ***********************
 
@@ -433,7 +453,7 b' The following is an overview of how to use these classes together:'
 2. Define some functions to be run as tasks
 3. Submit your tasks to using the :meth:`apply` method of your
    :class:`LoadBalancedView` instance.
-4. Use :meth:`Client.get_result` to get the results of the
+4. Use :meth:`.Client.get_result` to get the results of the
    tasks, or use the :meth:`AsyncResult.get` method of the results to wait
    for and then receive the results.
 

docs/source/parallel/parallel_winhpc.txt

@@ -120,7 +120,7 b' opening a Windows Command Prompt and typing ``ipython``. This will'
 start IPython's interactive shell and you should see something like the
 following screenshot:
 
-.. image:: ipython_shell.*
+.. image:: figs/ipython_shell.*
 
 Starting an IPython cluster
 ===========================
@@ -168,7 +168,7 b' You should see a number of messages printed to the screen, ending with'
 "IPython cluster: started". The result should look something like the following
 screenshot:
 
-.. image:: ipcluster_start.*
+.. image:: figs/ipcluster_start.*
 
 At this point, the controller and two engines are running on your local host.
 This configuration is useful for testing and for situations where you want to
@@ -210,7 +210,7 b' The output of this command is shown in the screenshot below. Notice how'
 :command:`ipcluster` prints out the location of the newly created cluster
 directory.
 
-.. image:: ipcluster_create.*
+.. image:: figs/ipcluster_create.*
 
 Configuring a cluster profile
 -----------------------------
@@ -232,10 +232,8 b' will need to edit the following attributes in the file'
 
     # Set these at the top of the file to tell ipcluster to use the
     # Windows HPC job scheduler.
-    c.IPClusterStart.controller_launcher = \
-        'IPython.parallel.apps.launcher.WindowsHPCControllerLauncher'
-    c.IPClusterEngines.engine_launcher = \
-        'IPython.parallel.apps.launcher.WindowsHPCEngineSetLauncher'
+    c.IPClusterStart.controller_launcher_class = 'WindowsHPCControllerLauncher'
+    c.IPClusterEngines.engine_launcher_class = 'WindowsHPCEngineSetLauncher'
 
     # Set these to the host name of the scheduler (head node) of your cluster.
     c.WindowsHPCControllerLauncher.scheduler = 'HEADNODE'
@@ -279,7 +277,7 b' must be run again to regenerate the XML job description files. The'
 following screenshot shows what the HPC Job Manager interface looks like
 with a running IPython cluster.
 
-.. image:: hpc_job_manager.*
+.. image:: figs/hpc_job_manager.*
 
 Performing a simple interactive parallel computation
 ====================================================
@@ -330,5 +328,5 b" The :meth:`map` method has the same signature as Python's builtin :func:`map`"
 function, but runs the calculation in parallel. More involved examples of using
 :class:`MultiEngineClient` are provided in the examples that follow.
 
-.. image:: mec_simple.*
+.. image:: figs/mec_simple.*