##// END OF EJS Templates
rework logging connections
rework logging connections

File last commit:

r3609:8d078bcc
r3610:c221efd0
Show More
dagdeps.py
105 lines | 2.9 KiB | text/x-python | PythonLexer
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564 """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."""
MinRK
tweak dagdeps for new AsyncResult objects
r3606 results = {}
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564 for node in nx.topological_sort(G):
MinRK
tweak dagdeps for new AsyncResult objects
r3606 deps = [ results[n].msg_ids[0] for n in G.predecessors(node) ]
results[node] = client.apply(jobs[node], after=deps)
return results
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564
MinRK
updated newparallel examples, moved into docs
r3609 def validate_tree(G, results):
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564 """Validate that jobs executed after their dependencies."""
for node in G:
MinRK
updated newparallel examples, moved into docs
r3609 started = results[node].metadata.started
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564 for parent in G.predecessors(node):
MinRK
updated newparallel examples, moved into docs
r3609 finished = results[parent].metadata.completed
assert started > finished, "%s should have happened after %s"%(node, parent)
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564
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.
"""
MinRK
updated newparallel examples, moved into docs
r3609 from matplotlib.dates import date2num
MinRK
tweak dagdeps for new AsyncResult objects
r3606 print "building DAG"
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564 G = random_dag(nodes, edges)
jobs = {}
pos = {}
for node in G:
jobs[node] = randomwait
MinRK
updated newparallel examples, moved into docs
r3609 client = cmod.Client()
MinRK
tweak dagdeps for new AsyncResult objects
r3606 print "submitting tasks"
results = submit_jobs(client, G, jobs)
print "waiting for results"
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564 client.barrier()
MinRK
tweak dagdeps for new AsyncResult objects
r3606 print "done"
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564 for node in G:
MinRK
updated newparallel examples, moved into docs
r3609 # times[node] = results[node].get()
t = date2num(results[node].metadata.started)
pos[node] = (t, G.in_degree(node)+random())
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564
MinRK
updated newparallel examples, moved into docs
r3609 validate_tree(G, results)
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564 nx.draw(G, pos)
MinRK
updated newparallel examples, moved into docs
r3609 return G,results
MinRK
added py4science demos as examples + NetworkX DAG dependencies
r3564
if __name__ == '__main__':
import pylab
main(32,128)
pylab.show()