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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
|
import py
import pytest
try:
import _continuation
except ImportError:
py.test.skip("to run on top of a translated pypy-c")
py.test.skip("convert to an apptest")
import sys, random
from rpython.tool.udir import udir
# ____________________________________________________________
STATUS_MAX = 50000
CONTINULETS = 50
def set_fast_mode():
global STATUS_MAX, CONTINULETS
STATUS_MAX = 100
CONTINULETS = 5
# ____________________________________________________________
class Done(Exception):
pass
class Runner(object):
def __init__(self):
self.foobar = 12345
self.conts = {} # {continulet: parent-or-None}
self.contlist = []
def run_test(self):
self.start_continulets()
self.n = 0
try:
while True:
self.do_switch(src=None)
assert self.target is None
except Done:
self.check_traceback(sys.exc_info()[2])
def do_switch(self, src):
assert src not in self.conts.values()
c = random.choice(self.contlist)
self.target = self.conts[c]
self.conts[c] = src
c.switch()
assert self.target is src
def run_continulet(self, c, i):
while True:
assert self.target is c
assert self.contlist[i] is c
self.do_switch(c)
assert self.foobar == 12345
self.n += 1
if self.n >= STATUS_MAX:
raise Done
def start_continulets(self, i=0):
c = _continuation.continulet(self.run_continulet, i)
self.contlist.append(c)
if i < CONTINULETS:
self.start_continulets(i + 1)
# ^^^ start each continulet with a different base stack
self.conts[c] = c # initially (i.e. not started) there are all loops
def check_traceback(self, tb):
found = []
tb = tb.tb_next
while tb:
if tb.tb_frame.f_code.co_name != 'do_switch':
assert tb.tb_frame.f_code.co_name == 'run_continulet', (
"got %r" % (tb.tb_frame.f_code.co_name,))
found.append(tb.tb_frame.f_locals['c'])
tb = tb.tb_next
found.reverse()
#
expected = []
c = self.target
while c is not None:
expected.append(c)
c = self.conts[c]
#
assert found == expected, "%r == %r" % (found, expected)
# ____________________________________________________________
class AppTestWrapper:
def setup_class(cls):
"Run test_various_depths() when we are run with 'pypy py.test -A'."
from pypy.conftest import option
if not option.runappdirect:
py.test.skip("meant only for -A run")
cls.w_vmprof_file = cls.space.wrap(str(udir.join('profile.vmprof')))
def test_vmprof(self):
"""
The point of this test is to check that we do NOT segfault. In
particular, we need to ensure that vmprof does not sample the stack in
the middle of a switch, else we read nonsense.
"""
_vmprof = pytest.importorskip('_vmprof')
def switch_forever(c):
while True:
c.switch()
#
f = open(self.vmprof_file, 'w+b')
_vmprof.enable(f.fileno(), 1/250.0, False, False, False, False)
c = _continuation.continulet(switch_forever)
for i in range(10**7):
if i % 100000 == 0:
print i
c.switch()
_vmprof.disable()
f.close()
def test_thread_switch_to_sub(self):
try:
import thread, time
except ImportError:
py.test.skip("no threads")
c_list = []
lock = thread.allocate_lock()
lock.acquire()
lock2 = thread.allocate_lock()
lock2.acquire()
#
def fn():
c = _continuation.continulet(lambda c_main: c_main.switch())
c.switch()
c_list.append(c)
lock.release()
lock2.acquire()
#
thread.start_new_thread(fn, ())
lock.acquire()
[c] = c_list
py.test.raises(_continuation.error, c.switch)
#
lock2.release()
time.sleep(0.5)
py.test.raises(_continuation.error, c.switch)
def test_thread_switch_to_sub_nonstarted(self):
try:
import thread, time
except ImportError:
py.test.skip("no threads")
c_list = []
lock = thread.allocate_lock()
lock.acquire()
lock2 = thread.allocate_lock()
lock2.acquire()
#
def fn():
c = _continuation.continulet(lambda c_main: None)
c_list.append(c)
lock.release()
lock2.acquire()
#
thread.start_new_thread(fn, ())
lock.acquire()
[c] = c_list
py.test.raises(_continuation.error, c.switch)
#
lock2.release()
time.sleep(0.5)
py.test.raises(_continuation.error, c.switch)
def test_thread_switch_to_main(self):
try:
import thread, time
except ImportError:
py.test.skip("no threads")
c_list = []
lock = thread.allocate_lock()
lock.acquire()
lock2 = thread.allocate_lock()
lock2.acquire()
#
def fn():
def in_continulet(c_main):
c_list.append(c_main)
lock.release()
lock2.acquire()
c = _continuation.continulet(in_continulet)
c.switch()
#
thread.start_new_thread(fn, ())
lock.acquire()
[c] = c_list
py.test.raises(_continuation.error, c.switch)
#
lock2.release()
time.sleep(0.5)
py.test.raises(_continuation.error, c.switch)
def _setup():
for _i in range(20):
def test_single_threaded(self):
Runner().run_test()
test_single_threaded.func_name = 'test_single_threaded_%d' % _i
setattr(AppTestWrapper, test_single_threaded.func_name,
test_single_threaded)
for _i in range(5):
def test_multi_threaded(self):
multithreaded_test()
test_multi_threaded.func_name = 'test_multi_threaded_%d' % _i
setattr(AppTestWrapper, test_multi_threaded.func_name,
test_multi_threaded)
_setup()
class ThreadTest(object):
def __init__(self, lock):
self.lock = lock
self.ok = False
lock.acquire()
def run(self):
try:
Runner().run_test()
self.ok = True
finally:
self.lock.release()
def multithreaded_test():
try:
import thread
except ImportError:
py.test.skip("no threads")
ts = [ThreadTest(thread.allocate_lock()) for i in range(5)]
for t in ts:
thread.start_new_thread(t.run, ())
for t in ts:
t.lock.acquire()
for t in ts:
assert t.ok
# ____________________________________________________________
if __name__ == '__main__':
Runner().run_test()
|
