-
Notifications
You must be signed in to change notification settings - Fork 780
Expand file tree
/
Copy pathtest_thread.py
More file actions
352 lines (262 loc) · 11.6 KB
/
Copy pathtest_thread.py
File metadata and controls
352 lines (262 loc) · 11.6 KB
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
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
# -*- coding: utf-8 -*-
"""Test CLR bridge threading and GIL handling."""
import sys
import threading
import time
import pytest
import _thread as thread
from .utils import dprint
def _gil_enabled():
"""True on every CPython that has a GIL. Always True before 3.13."""
return getattr(sys, "_is_gil_enabled", lambda: True)()
def test_simple_callback_to_python():
"""Test a call to managed code that then calls back into Python."""
from Python.Test import ThreadTest
dprint("thread %s SimpleCallBack" % thread.get_ident())
result = ThreadTest.CallEchoString("spam")
assert result == "spam"
dprint("thread %s SimpleCallBack ret" % thread.get_ident())
def test_double_callback_to_python():
"""Test a call to managed code that then calls back into Python
that then calls managed code that then calls Python again."""
from Python.Test import ThreadTest
dprint("thread %s DoubleCallBack" % thread.get_ident())
result = ThreadTest.CallEchoString2("spam")
assert result == "spam"
dprint("thread %s DoubleCallBack ret" % thread.get_ident())
def test_python_thread_calls_to_clr():
"""Test calls by Python-spawned threads into managed code."""
# This test is very likely to hang if something is wrong ;)
import System
done = []
def run_thread():
for i in range(10):
time.sleep(0.1)
dprint("thread %s %d" % (thread.get_ident(), i))
mstr = System.String("thread %s %d" % (thread.get_ident(), i))
dprint(mstr.ToString())
done.append(None)
dprint("thread %s %d done" % (thread.get_ident(), i))
threads = [threading.Thread(target=run_thread) for _ in range(5)]
for t in threads:
t.start()
while len(done) < 50:
dprint(len(done))
time.sleep(0.1)
# Join the workers so they cannot outlive this test and fire
# threading.excepthook from background activity (visible under FT).
for t in threads:
t.join()
# Free-threaded / refcount tests below. Run on every interpreter; the GIL
# builds exercise the same code paths in single-threaded form while the FT
# builds (Py_GIL_DISABLED) actually stress the concurrent paths.
def test_runtime_refcount_matches_sys_getrefcount():
"""Refcount tracks sys.getrefcount on both GIL and FT builds."""
obj = object()
rc_before = sys.getrefcount(obj)
extra = [obj, obj, obj]
assert sys.getrefcount(obj) - rc_before == 3
del extra
def test_is_gil_enabled_attribute_present_on_3_13_plus():
"""sys._is_gil_enabled is present from 3.13 — used by ABI.DetectFreeThreaded."""
if sys.version_info < (3, 13):
assert not hasattr(sys, "_is_gil_enabled")
else:
assert isinstance(sys._is_gil_enabled(), bool)
def test_module_dunder_all_added_once():
"""Module.__all__ adds each name exactly once.
Exercises ModuleObject.allNames (ConcurrentDictionary) — the per-name
"have we surfaced this in __all__ yet" guard. A torn HashSet would let
duplicates slip through here on free-threaded builds.
"""
import System
names = list(System.__all__)
assert len(names) == len(set(names))
def _run_in_threads(target, n_threads, *args, **kwargs):
"""Run target() in n_threads threads, return results in start order, raise on first error."""
results = [None] * n_threads
errors = [None] * n_threads
def worker(i):
try:
results[i] = target(i, *args, **kwargs)
except BaseException as e:
errors[i] = e
threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)]
for t in threads:
t.start()
for t in threads:
t.join()
for e in errors:
if e is not None:
raise e
return results
def test_concurrent_clr_method_calls():
"""Concurrent CLR method invocation across threads."""
from Python.Test import ThreadTest
def call(_):
return [ThreadTest.CallEchoString("ping") for _ in range(200)]
for r in _run_in_threads(call, n_threads=8):
assert all(x == "ping" for x in r)
def test_concurrent_attribute_access():
"""Concurrent attribute access — exercises the ConcurrentDictionary InternString cache."""
import System
from System.Collections.Generic import List
def access(_):
for _ in range(500):
_ = System.String.Empty
_ = System.Int32.MaxValue
_ = List[int]
_ = List[str]
return True
assert all(_run_in_threads(access, n_threads=8))
def test_concurrent_module_attribute_access():
"""Concurrent CLR-namespace attribute access — exercises ModuleObject.cache.
Each lookup of `System.X` either hits ModuleObject.cache or populates it
on first miss. A plain Dictionary tore on simultaneous TryGetValue/Add
from multiple threads; the test reads many distinct names per worker.
"""
import System
names = (
"String", "Int32", "Int64", "Double", "Boolean", "Object",
"DateTime", "TimeSpan", "Type", "Array", "Console", "Math",
)
def lookup(_):
for _ in range(200):
for n in names:
getattr(System, n)
return True
assert all(_run_in_threads(lookup, n_threads=8))
@pytest.mark.skipif(_gil_enabled(), reason="Only meaningful on free-threaded Python (Py_GIL_DISABLED).")
def test_concurrent_clr_object_creation():
"""Concurrent CLR object alloc/free — exercises reflectedObjects + loadedExtensions.
FT-only: under the GIL this high-contention pattern hits a pre-existing
pythonnet crash (also reproducible on master) outside this branch's scope.
"""
from System.Collections.Generic import List
LI = List[int]
def make_lists(_):
for _ in range(200):
l = LI()
for j in range(5):
l.Add(j)
assert l.Count == 5
return True
assert all(_run_in_threads(make_lists, n_threads=8))
@pytest.mark.skipif(_gil_enabled(), reason="Only meaningful on free-threaded Python (Py_GIL_DISABLED).")
def test_concurrent_python_subclass_of_clr_type():
"""Concurrent dynamic-subclass creation — exercises ClassDerived's builder lock.
FT-only because the GIL-build code path triggers a pre-existing CLR
object lifecycle crash under high contention.
"""
import System
def derive(i):
cls = type(f"Derived_{i}_{threading.get_ident()}", (System.Object,), {})
cls()
return cls.__name__
names = _run_in_threads(derive, n_threads=8)
assert len(set(names)) == len(names)
@pytest.mark.skipif(_gil_enabled(), reason="Only meaningful on free-threaded Python (Py_GIL_DISABLED).")
def test_concurrent_delegate_creation():
"""Concurrent CLR delegate dispatcher creation — exercises DelegateManager.
Each new (delegate-type, callable) pair runs Reflection.Emit to build a
dispatcher subclass. Without a lock, concurrent DefineType raises
"Duplicate type name within an assembly" or corrupts the IL stream.
FT-only because high-rate Reflection.Emit interacts badly with the
CPython 3.11/3.12/3.13 GIL-build GC under cumulative test state
(same pre-existing crash as test_concurrent_clr_object_creation).
"""
from Python.Runtime import PythonEngine
handler = PythonEngine.ShutdownHandler
def build(_):
for _ in range(50):
handler(lambda: None)
return True
assert all(_run_in_threads(build, n_threads=8))
@pytest.mark.skipif(_gil_enabled(), reason="Only meaningful on free-threaded Python (Py_GIL_DISABLED).")
def test_concurrent_clr_delegate_invocation_from_python():
"""Python callables wrapped as distinct CLR delegate types, invoked concurrently.
Real-world: QuantConnect/Lean and similar embedders pass Python callables
where C# expects a delegate; under FT the dispatcher emit + Invoke run from
multiple threads. Hits DelegateManager.GetDispatcher (Reflection.Emit lock)
and Dispatcher.Dispatch (Py.GIL reacquisition).
"""
from Python.Test import (
PublicDelegate, StringDelegate, BoolDelegate,
)
delegates = (
PublicDelegate(lambda: None),
StringDelegate(lambda: "ok"),
BoolDelegate(lambda: True),
)
def fire(i):
d = delegates[i % len(delegates)]
for _ in range(200):
d()
return True
assert all(_run_in_threads(fire, n_threads=8))
@pytest.mark.skipif(_gil_enabled(), reason="Only meaningful on free-threaded Python (Py_GIL_DISABLED).")
def test_concurrent_generic_type_binding():
"""Concurrent `Dictionary[K, V]` with many distinct type-arg pairs.
Real-world: pythonnet/pythonnet#2269, #1407, #821 — concurrent ToPython /
GenericByName from N threads. Exercises ClassManager.cache,
TypeManager.cache, GenericUtil.mapping, and the generic-type binding
fast path together.
FT-only: the cumulative state under the full pytest suite trips the same
pre-existing CPython 3.11/3.12/3.13 GIL-build crash that gates the other
high-contention tests in this file.
"""
from System import Int32, Int64, String, Double, Single, Byte
from System.Collections.Generic import Dictionary, List
arg_types = (Int32, Int64, String, Double, Single, Byte)
pairs = [(k, v) for k in arg_types for v in arg_types]
def bind(_):
for _ in range(50):
for k, v in pairs:
_ = Dictionary[k, v]
_ = List[k]
return True
assert all(_run_in_threads(bind, n_threads=8))
@pytest.mark.skipif(_gil_enabled(), reason="Only meaningful on free-threaded Python (Py_GIL_DISABLED).")
def test_concurrent_shutdown_handler_register():
"""Concurrent AddShutdownHandler/RemoveShutdownHandler — exercises ShutdownHandlers list.
FT-only because Python<->CLR delegate marshalling at this rate trips
the same pre-existing CPython 3.11/3.12/3.13 GIL-build crash as the
other high-contention tests in this file.
"""
from Python.Runtime import PythonEngine
handlers = [PythonEngine.ShutdownHandler(lambda: None) for _ in range(32)]
def churn(i):
h = handlers[i % len(handlers)]
for _ in range(500):
PythonEngine.AddShutdownHandler(h)
PythonEngine.RemoveShutdownHandler(h)
return True
assert all(_run_in_threads(churn, n_threads=8))
@pytest.mark.skipif(_gil_enabled(), reason="Only meaningful on free-threaded Python (Py_GIL_DISABLED).")
def test_concurrent_gc_collect_on_clr_cycles():
"""Concurrent gc.collect on cyclic CLR-derived objects — exercises
ClassBase.ClearVisited + ManagedType.TryFreeGCHandle atomic slot.
Each worker builds short cycles holding a Python subclass of System.Object,
then calls gc.collect() while other workers do the same. Hits the
tp_clear/tp_dealloc race path on the per-object GCHandle slot.
Also covers ClassDerivedObject.tp_dealloc's strong→weak slot demotion
and MethodBinder.GetMethods lazy-init: a torn slot or torn init both
surface as "No method matches given arguments for Cycle..ctor".
"""
import gc
import System
class Cycle(System.Object):
__namespace__ = "test_concurrent_gc_collect_on_clr_cycles"
def churn(_):
# Sporadic repro is intentional: the NewObjectToPython race only fires
# when .NET GC happens to fire during construction. Across the CI
# matrix this catches regressions reliably without the heavyweight
# CLR GC.Collect that deadlocks Mono on x64 Ubuntu.
for _ in range(100):
a, b = Cycle(), Cycle()
a.peer = b
b.peer = a
del a, b
gc.collect()
return True
assert all(_run_in_threads(churn, n_threads=8))