forked from ipython/ipython
-
Notifications
You must be signed in to change notification settings - Fork 1
/
dependencies.py
128 lines (111 loc) · 3.03 KB
/
dependencies.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
from IPython.parallel 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.load_balanced_view()
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, all=False, failure=True) # yes
d1b = Dependency(mixed, all=False) # yes
d2a = Dependency(mixed, all=True, failure=True) # yes after / no follow
d2b = Dependency(mixed, all=True) # no
d3 = Dependency(failures, all=False) # no
d4 = Dependency(failures, all=False, failure=True) # yes
d5 = Dependency(failures, all=True, failure=True) # yes after / no follow
d6 = Dependency(successes, all=True, failure=True) # yes after / no follow
view.block = False
flags = view.temp_flags
with flags(after=d1a):
r1a = view.apply(getpid)
with flags(follow=d1b):
r1b = view.apply(getpid)
with flags(after=d2b, follow=d2a):
r2a = view.apply(getpid)
with flags(after=d2a, follow=d2b):
r2b = view.apply(getpid)
with flags(after=d3):
r3 = view.apply(getpid)
with flags(after=d4):
r4a = view.apply(getpid)
with flags(follow=d4):
r4b = view.apply(getpid)
with flags(after=d3, follow=d4):
r4c = view.apply(getpid)
with flags(after=d5):
r5 = view.apply(getpid)
with flags(follow=d5, after=d3):
r5b = view.apply(getpid)
with flags(follow=d6):
r6 = view.apply(getpid)
with flags(after=d6, follow=d2b):
r6b = view.apply(getpid)
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')