update parallel docs with some changes from scipy tutorial

also added a couple more parallel examples:

* iopubwatcher for watching stdout/stderr of a cluster
* itermapresult for iterating through AsyncMapResults without waiting for all to complete
* customresults for advanced handling of results not provided by AsyncResult objects

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,52 b''
+"""Example of iteration through AsyncMapResult, without waiting for all results
+
+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)

docs/source/parallel/dag_dependencies.txt

@@ -30,7 +30,8 b' A Sample DAG'
 
 Here, we have a very simple 5-node DAG:
 
-.. figure:: figs/ 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
@@ -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:
 
@@ -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:: figs/ 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_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_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
@@ -113,12 +113,24 b' that turns any Python function into a parallel function:'
 
 .. _parallel_taskmap:
 
-The AsyncMapResult
-==================
+Map results are iterable!
+-------------------------
 
-When you call ``lview.map_async(f, sequence)``, or just :meth:`map` with `block=True`, then
-what you get in return will be an :class:`~AsyncMapResult` object. These are similar to 
-AsyncResult objects, but with one key difference
+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:
 
@@ -166,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.
@@ -184,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)``
@@ -209,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
@@ -222,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
 ------------------
@@ -287,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::
 
@@ -440,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.