# -*- 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 PythonCLR 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 strongweak 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))