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
247
248
|
"""Tests monitoring, sys.settrace, and sys.setprofile in a multi-threaded
environment to verify things are thread-safe in a free-threaded build"""
import sys
import time
import unittest
import weakref
from sys import monitoring
from test.support import threading_helper
from threading import Thread, _PyRLock
from unittest import TestCase
class InstrumentationMultiThreadedMixin:
thread_count = 10
func_count = 50
fib = 12
def after_threads(self):
"""Runs once after all the threads have started"""
pass
def during_threads(self):
"""Runs repeatedly while the threads are still running"""
pass
def work(self, n, funcs):
"""Fibonacci function which also calls a bunch of random functions"""
for func in funcs:
func()
if n < 2:
return n
return self.work(n - 1, funcs) + self.work(n - 2, funcs)
def start_work(self, n, funcs):
# With the GIL builds we need to make sure that the hooks have
# a chance to run as it's possible to run w/o releasing the GIL.
time.sleep(0.1)
self.work(n, funcs)
def after_test(self):
"""Runs once after the test is done"""
pass
def test_instrumentation(self):
# Setup a bunch of functions which will need instrumentation...
funcs = []
for i in range(self.func_count):
x = {}
exec("def f(): pass", x)
funcs.append(x["f"])
threads = []
for i in range(self.thread_count):
# Each thread gets a copy of the func list to avoid contention
t = Thread(target=self.start_work, args=(self.fib, list(funcs)))
t.start()
threads.append(t)
self.after_threads()
while True:
any_alive = False
for t in threads:
if t.is_alive():
any_alive = True
break
if not any_alive:
break
self.during_threads()
self.after_test()
class MonitoringTestMixin:
def setUp(self):
for i in range(6):
if monitoring.get_tool(i) is None:
self.tool_id = i
monitoring.use_tool_id(i, self.__class__.__name__)
break
def tearDown(self):
monitoring.free_tool_id(self.tool_id)
@threading_helper.requires_working_threading()
class SetPreTraceMultiThreaded(InstrumentationMultiThreadedMixin, TestCase):
"""Sets tracing one time after the threads have started"""
def setUp(self):
super().setUp()
self.called = False
def after_test(self):
self.assertTrue(self.called)
def trace_func(self, frame, event, arg):
self.called = True
return self.trace_func
def after_threads(self):
sys.settrace(self.trace_func)
@threading_helper.requires_working_threading()
class MonitoringMultiThreaded(
MonitoringTestMixin, InstrumentationMultiThreadedMixin, TestCase
):
"""Uses sys.monitoring and repeatedly toggles instrumentation on and off"""
def setUp(self):
super().setUp()
self.set = False
self.called = False
monitoring.register_callback(
self.tool_id, monitoring.events.LINE, self.callback
)
def tearDown(self):
monitoring.set_events(self.tool_id, 0)
super().tearDown()
def callback(self, *args):
self.called = True
def after_test(self):
self.assertTrue(self.called)
def during_threads(self):
if self.set:
monitoring.set_events(
self.tool_id, monitoring.events.CALL | monitoring.events.LINE
)
else:
monitoring.set_events(self.tool_id, 0)
self.set = not self.set
@threading_helper.requires_working_threading()
class SetTraceMultiThreaded(InstrumentationMultiThreadedMixin, TestCase):
"""Uses sys.settrace and repeatedly toggles instrumentation on and off"""
def setUp(self):
self.set = False
self.called = False
def after_test(self):
self.assertTrue(self.called)
def tearDown(self):
sys.settrace(None)
def trace_func(self, frame, event, arg):
self.called = True
return self.trace_func
def during_threads(self):
if self.set:
sys.settrace(self.trace_func)
else:
sys.settrace(None)
self.set = not self.set
@threading_helper.requires_working_threading()
class SetProfileMultiThreaded(InstrumentationMultiThreadedMixin, TestCase):
"""Uses sys.setprofile and repeatedly toggles instrumentation on and off"""
def setUp(self):
self.set = False
self.called = False
def after_test(self):
self.assertTrue(self.called)
def tearDown(self):
sys.setprofile(None)
def trace_func(self, frame, event, arg):
self.called = True
return self.trace_func
def during_threads(self):
if self.set:
sys.setprofile(self.trace_func)
else:
sys.setprofile(None)
self.set = not self.set
@threading_helper.requires_working_threading()
class MonitoringMisc(MonitoringTestMixin, TestCase):
def register_callback(self):
def callback(*args):
pass
for i in range(200):
monitoring.register_callback(self.tool_id, monitoring.events.LINE, callback)
self.refs.append(weakref.ref(callback))
def test_register_callback(self):
self.refs = []
threads = []
for i in range(50):
t = Thread(target=self.register_callback)
t.start()
threads.append(t)
for thread in threads:
thread.join()
monitoring.register_callback(self.tool_id, monitoring.events.LINE, None)
for ref in self.refs:
self.assertEqual(ref(), None)
def test_set_local_trace_opcodes(self):
def trace(frame, event, arg):
frame.f_trace_opcodes = True
return trace
loops = 1_000
sys.settrace(trace)
try:
l = _PyRLock()
def f():
for i in range(loops):
with l:
pass
t = Thread(target=f)
t.start()
for i in range(loops):
with l:
pass
t.join()
finally:
sys.settrace(None)
if __name__ == "__main__":
unittest.main()
|
