docs/examples/newparallel/demo/dependencies.py

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'