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

@@ -61,13 +61,13 b' def submit_jobs(client, G, jobs):'
         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
@@ -76,28 +76,28 b' def main(nodes, edges):'
     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

docs/examples/newparallel/demo/remotefunction.py

@@ -1,6 +1,6 b''
 from IPython.zmq.parallel.client import *
 
-client = Client('tcp://127.0.0.1:10101')
+client = Client()
 
 @remote(client, bound=True)
 def getkey(name):

docs/examples/newparallel/demo/throughput.py

@@ -14,7 +14,7 b" 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:
@@ -84,6 +84,3 b' def do_echo(n,tlist=[0],f=echo, trials=2, runner=time_throughput):'
         print t,A[i]
     return A
 
  No newline at end of file
-def start_cluster(n, scheduler):
-    pass
-    
  No newline at end of file

docs/examples/newparallel/demo/views.py

@@ -1,6 +1,6 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)

docs/examples/newparallel/workflow/job_wrapper.py

@@ -10,12 +10,16 b' 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

@@ -32,7 +32,7 b' def cleanup(controller, engines):'
 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"