updated newparallel examples, moved into docs

docs/examples/newparallel/demo/dependencies.py

@@ -0,0 +1,118 b''
+from IPython.zmq.parallel import error
+from IPython.zmq.parallel.dependency import Dependency
+from IPython.zmq.parallel.client import *
+
+client = Client()
+
+# this will only run on machines that can import numpy:
+@require('numpy')
+def norm(A):
+    from numpy.linalg import norm
+    return norm(A,2)
+
+def checkpid(pid):
+    """return the pid of the engine"""
+    import os
+    return os.getpid() == pid
+
+def checkhostname(host):
+    import socket
+    return socket.gethostname() == host
+
+def getpid():
+    import os
+    return os.getpid()
+
+pid0 = client[0].apply_sync(getpid)
+
+# this will depend on the pid being that of target 0:
+@depend(checkpid, pid0)
+def getpid2():
+    import os
+    return os.getpid()
+
+view = client[None]
+view.block=True
+
+# will run on anything:
+pids1 = [ view.apply(getpid) for i in range(len(client.ids)) ]
+print pids1
+# will only run on e0:
+pids2 = [ view.apply(getpid2) for i in range(len(client.ids)) ]
+print pids2
+
+print "now test some dependency behaviors"
+
+def wait(t):
+    import time
+    time.sleep(t)
+    return t
+
+# fail after some time:
+def wait_and_fail(t):
+    import time
+    time.sleep(t)
+    return 1/0
+
+successes = [ view.apply_async(wait, 1).msg_ids[0] for i in range(len(client.ids)) ]
+failures = [ view.apply_async(wait_and_fail, 1).msg_ids[0] for i in range(len(client.ids)) ]
+
+mixed = [failures[0],successes[0]]
+d1a = Dependency(mixed, mode='any', success_only=False) # yes
+d1b = Dependency(mixed, mode='any', success_only=True) # yes
+d2a = Dependency(mixed, mode='all', success_only=False) # yes after / no follow
+d2b = Dependency(mixed, mode='all', success_only=True) # no
+d3 = Dependency(failures, mode='any', success_only=True) # no
+d4 = Dependency(failures, mode='any', success_only=False) # yes
+d5 = Dependency(failures, mode='all', success_only=False) # yes after / no follow
+d6 = Dependency(successes, mode='all', success_only=False) # yes after / no follow
+
+client.block = False
+
+r1a = client.apply(getpid, after=d1a)
+r1b = client.apply(getpid, follow=d1b)
+r2a = client.apply(getpid, after=d2b, follow=d2a)
+r2b = client.apply(getpid, after=d2a, follow=d2b)
+r3 = client.apply(getpid, after=d3)
+r4a = client.apply(getpid, after=d4)
+r4b = client.apply(getpid, follow=d4)
+r4c = client.apply(getpid, after=d3, follow=d4)
+r5 = client.apply(getpid, after=d5)
+r5b = client.apply(getpid, follow=d5, after=d3)
+r6 = client.apply(getpid, follow=d6)
+r6b = client.apply(getpid, after=d6, follow=d2b)
+
+def should_fail(f):
+    try:
+        f()
+    except error.KernelError:
+        pass
+    else:
+        print 'should have raised'
+        # raise Exception("should have raised")
+
+# print r1a.msg_ids
+r1a.get()
+# print r1b.msg_ids
+r1b.get()
+# print r2a.msg_ids
+should_fail(r2a.get)
+# print r2b.msg_ids
+should_fail(r2b.get)
+# print r3.msg_ids
+should_fail(r3.get)
+# print r4a.msg_ids
+r4a.get()
+# print r4b.msg_ids
+r4b.get()
+# print r4c.msg_ids
+should_fail(r4c.get)
+# print r5.msg_ids
+r5.get()
+# print r5b.msg_ids
+should_fail(r5b.get)
+# print r6.msg_ids
+should_fail(r6.get) # assuming > 1 engine
+# print r6b.msg_ids
+should_fail(r6b.get)
+print 'done'

docs/examples/newparallel/demo/map.py

@@ -0,0 +1,36 b''
+from IPython.zmq.parallel.client import *
+
+client = Client()
+
+@remote(client, block=True)
+def square(a):
+    """return square of a number"""
+    return a*a
+
+squares = map(square, range(42))
+
+# but that blocked between each result; not exactly useful
+
+square.block = False
+
+arlist = map(square, range(42))
+# submitted very fast
+
+# wait for the results:
+squares2 = [ r.get() for r in arlist ]
+
+# now the more convenient @parallel decorator, which has a map method:
+
+@parallel(client, block=False)
+def psquare(a):
+    """return square of a number"""
+    return a*a
+
+# this chunks the data into n-negines jobs, not 42 jobs:
+ar = psquare.map(range(42))
+
+# wait for the results to be done:
+squares3 = ar.get()
+
+print squares == squares2, squares3==squares
+# True
  No newline at end of file

docs/examples/newparallel/demo/dag/dagdeps.py

@@ -1,106 +1,106 b''
 """Example for generating an arbitrary DAG as a dependency map.
 
 This demo uses networkx to generate the graph.
 
 Authors
 -------
 * MinRK
 """
 import networkx as nx
 from random import randint, random
 from IPython.zmq.parallel import client as cmod
 
 def randomwait():
     import time
     from random import random
     time.sleep(random())
     return time.time()
 
 
 def random_dag(nodes, edges):
     """Generate a random Directed Acyclic Graph (DAG) with a given number of nodes and edges."""
     G = nx.DiGraph()
     for i in range(nodes):
         G.add_node(i)
     while edges > 0:
         a = randint(0,nodes-1)
         b=a
         while b==a:
             b = randint(0,nodes-1)
         G.add_edge(a,b)
         if nx.is_directed_acyclic_graph(G):
             edges -= 1
         else:
             # we closed a loop!
             G.remove_edge(a,b)
     return G
 
 def add_children(G, parent, level, n=2):
     """Add children recursively to a binary tree."""
     if level == 0:
         return
     for i in range(n):
         child = parent+str(i)
         G.add_node(child)
         G.add_edge(parent,child)
         add_children(G, child, level-1, n)
 
 def make_bintree(levels):
     """Make a symmetrical binary tree with @levels"""
     G = nx.DiGraph()
     root = '0'
     G.add_node(root)
     add_children(G, root, levels, 2)
     return G
 
 def submit_jobs(client, G, jobs):
     """Submit jobs via client where G describes the time dependencies."""
     results = {}
     for node in nx.topological_sort(G):
         deps = [ results[n].msg_ids[0] for n in G.predecessors(node) ]
         results[node] = client.apply(jobs[node], after=deps)
     return results
 
-def validate_tree(G, times):
+def validate_tree(G, results):
     """Validate that jobs executed after their dependencies."""
     for node in G:
-        t = times[node]
+        started = results[node].metadata.started
         for parent in G.predecessors(node):
-            pt = times[parent]
-            assert t > pt, "%s should have happened after %s"%(node, parent)
+            finished = results[parent].metadata.completed
+            assert started > finished, "%s should have happened after %s"%(node, parent)
 
 def main(nodes, edges):
     """Generate a random graph, submit jobs, then validate that the
     dependency order was enforced.
     Finally, plot the graph, with time on the x-axis, and
     in-degree on the y (just for spread).  All arrows must
     point at least slightly to the right if the graph is valid.
     """
+    from matplotlib.dates import date2num
     print "building DAG"
     G = random_dag(nodes, edges)
     jobs = {}
-    msg_ids = {}
-    times = {}
     pos = {}
     for node in G:
         jobs[node] = randomwait
     
-    client = cmod.Client('tcp://127.0.0.1:10101')
+    client = cmod.Client()
     print "submitting tasks"
     results = submit_jobs(client, G, jobs)
     print "waiting for results"
     client.barrier()
     print "done"
     for node in G:
-        times[node] = results[node].get()
-        pos[node] = (times[node], G.in_degree(node)+random())
+        # times[node] = results[node].get()
+        t = date2num(results[node].metadata.started)
+        pos[node] = (t, G.in_degree(node)+random())
     
-    validate_tree(G, times)
+    validate_tree(G, results)
     nx.draw(G, pos)
-    return G,times,msg_ids
+    return G,results
 
 if __name__ == '__main__':
     import pylab
     main(32,128)
     pylab.show()
     
  No newline at end of file

docs/examples/newparallel/demo/remotefunction.py

@@ -1,22 +1,22 b''
 from IPython.zmq.parallel.client import *
 
-client = Client('tcp://127.0.0.1:10101')
+client = Client()
 
 @remote(client, bound=True)
 def getkey(name):
     """fetch something from globals"""
     return globals().get(name)
 
 @remote(client, bound=True, targets='all')
 def setpids():
     import os
     globals()['pid'] = os.getpid()
 
 # set pid in the globals
 setpids()
 getkey('pid')
 getkey.targets=[1,2]
 getkey('pid')
 getkey.bound=False
 getkey('pid') is None
 

docs/examples/newparallel/demo/throughput.py

@@ -1,89 +1,86 b''
 import time
 import numpy as np
 from IPython.zmq.parallel import client as clientmod
 
 nlist = map(int, np.logspace(2,9,16,base=2))
 nlist2 = map(int, np.logspace(2,8,15,base=2))
 tlist = map(int, np.logspace(7,22,16,base=2))
 nt = 16
 def wait(t=0):
     import time
     time.sleep(t)
 
 def echo(s=''):
     return s
 
 def time_throughput(nmessages, t=0, f=wait):
-    client = clientmod.Client('tcp://127.0.0.1:10101')
+    client = clientmod.Client()
     view = client[None]
     # do one ping before starting timing
     if f is echo:
         t = np.random.random(t/8)
     view.apply_sync(echo, '')
     client.spin()
     tic = time.time()
     for i in xrange(nmessages):
         view.apply(f, t)
     lap = time.time()
     client.barrier()
     toc = time.time()
     return lap-tic, toc-tic
 
 def time_twisted(nmessages, t=0, f=wait):
     from IPython.kernel import client as kc
     client = kc.TaskClient()
     if f is wait:
         s = "import time; time.sleep(%f)"%t
         task = kc.StringTask(s)
     elif f is echo:
         t = np.random.random(t/8)
         s = "s=t"
         task = kc.StringTask(s, push=dict(t=t), pull=['s'])
     else:
         raise
     # do one ping before starting timing
     client.barrier(client.run(task))
     tic = time.time()
     tids = []
     for i in xrange(nmessages):
         tids.append(client.run(task))
     lap = time.time()
     client.barrier(tids)
     toc = time.time()
     return lap-tic, toc-tic
 
 def do_runs(nlist,t=0,f=wait, trials=2, runner=time_throughput):
     A = np.zeros((len(nlist),2))
     for i,n in enumerate(nlist):
         t1 = t2 = 0
         for _ in range(trials):
             time.sleep(.25)
             ts = runner(n,t,f)
             t1 += ts[0]
             t2 += ts[1]
         t1 /= trials
         t2 /= trials
         A[i] = (t1,t2)
         A[i] = n/A[i]
         print n,A[i]
     return A
 
 def do_echo(n,tlist=[0],f=echo, trials=2, runner=time_throughput):
     A = np.zeros((len(tlist),2))
     for i,t in enumerate(tlist):
         t1 = t2 = 0
         for _ in range(trials):
             time.sleep(.25)
             ts = runner(n,t,f)
             t1 += ts[0]
             t2 += ts[1]
         t1 /= trials
         t2 /= trials
         A[i] = (t1,t2)
         A[i] = n/A[i]
         print t,A[i]
     return A
-
-def start_cluster(n, scheduler):
-    pass
     
  No newline at end of file

docs/examples/newparallel/demo/views.py

@@ -1,15 +1,15 b''
 from IPython.zmq.parallel.client import *
 
-client = Client('tcp://127.0.0.1:10101')
+client = Client()
 
 for id in client.ids:
     client.push(dict(ids=id*id), targets=id)
 
 rns = client[0]
 rns['a'] = 5
 
 print rns['a']
 
 remotes = client[:]
 
 print remotes['ids']
  No newline at end of file

docs/examples/newparallel/workflow/job_wrapper.py

@@ -1,21 +1,25 b''
 #!/usr/bin/env python
 """Python wrapper around a submitted workflow job.
 
 In reality this would be a more sophisticated script, here we only illustrate
 the basic idea by considering that a submitted 'job' is a Python string to be
 executed.
 """
 
 import sys
 
 argv = sys.argv
 
-from IPython.zmq.parallel.engine import main
+from IPython.zmq.parallel.engine import EngineFactory
+from IPython.zmq.parallel.ipengineapp import launch_new_instance
 
 ns = {}
 
 # job
 exec sys.argv[1] in ns
 
+# this should really use Config:
+EngineFactory.user_ns = ns
+
 # start engine with job namespace
-main([], user_ns=ns)
+launch_new_instance()

docs/examples/newparallel/workflow/wmanager.py

@@ -1,44 +1,44 b''
 """Mock workflow manager.
 
 This is a mock work manager whose submitted 'jobs' simply consist of executing
 a python string.  What we want is to see the implementation of the ipython
 controller part.
 """
 
 from __future__ import print_function
 
 import atexit
 import sys
 
 from subprocess import Popen
 
 def cleanup(controller, engines):
     """Cleanup routine to shut down all subprocesses we opened."""
     import signal, time
     
     print('Starting cleanup')
     print('Stopping engines...')
     for e in engines:
         e.send_signal(signal.SIGINT)
     print('Stopping controller...')
     # so it can shut down its queues
     controller.send_signal(signal.SIGINT)
     time.sleep(0.1)
     print('Killing controller...')
     controller.kill()
     print('Cleanup done')
 
 
 if __name__ == '__main__':
 
     # Start controller in separate process
-    cont = Popen(['python', '-m', 'IPython.zmq.parallel.controller'])
+    cont = Popen(['python', '-m', 'IPython.zmq.parallel.ipcontrollerapp'])
     print('Started controller')
 
     # "Submit jobs"
     eng = []
     for i in range(4):
         eng.append(Popen(['python', 'job_wrapper.py','x=%s' % i]))
 
     # Ensure that all subpro
     atexit.register(lambda : cleanup(cont, eng))