1 from __future__
import annotations
11 from pybind11_tests
import gil_scoped
as m
23 m.test_callback_py_obj(
lambda:
None)
27 m.test_callback_std_func(
lambda:
None)
32 m.test_callback_virtual_func(extended)
37 m.test_callback_pure_virtual_func(extended)
41 """Makes sure that the GIL can be acquired by another module from a GIL-released state."""
42 m.test_cross_module_gil_released()
46 """Makes sure that the GIL can be acquired by another module from a GIL-acquired state."""
47 m.test_cross_module_gil_acquired()
51 """Makes sure that the GIL can be acquired/released by another module
52 from a GIL-released state using custom locking logic."""
53 m.test_cross_module_gil_inner_custom_released()
57 """Makes sure that the GIL can be acquired/acquired by another module
58 from a GIL-acquired state using custom locking logic."""
59 m.test_cross_module_gil_inner_custom_acquired()
63 """Makes sure that the GIL can be acquired/released by another module
64 from a GIL-released state using pybind11 locking logic."""
65 m.test_cross_module_gil_inner_pybind11_released()
69 """Makes sure that the GIL can be acquired/acquired by another module
70 from a GIL-acquired state using pybind11 locking logic."""
71 m.test_cross_module_gil_inner_pybind11_acquired()
75 """Makes sure that the GIL can be nested acquired/released by another module
76 from a GIL-released state using custom locking logic."""
77 m.test_cross_module_gil_nested_custom_released()
81 """Makes sure that the GIL can be nested acquired/acquired by another module
82 from a GIL-acquired state using custom locking logic."""
83 m.test_cross_module_gil_nested_custom_acquired()
87 """Makes sure that the GIL can be nested acquired/released by another module
88 from a GIL-released state using pybind11 locking logic."""
89 m.test_cross_module_gil_nested_pybind11_released()
93 """Makes sure that the GIL can be nested acquired/acquired by another module
94 from a GIL-acquired state using pybind11 locking logic."""
95 m.test_cross_module_gil_nested_pybind11_acquired()
99 assert m.test_release_acquire(0xAB) ==
"171"
103 assert m.test_nested_acquire(0xAB) ==
"171"
107 for bits
in range(16 * 8):
108 internals_ids = m.test_multi_acquire_release_cross_module(bits)
109 assert len(internals_ids) == 2
if bits % 8
else 1
113 VARS_BEFORE_ALL_BASIC_TESTS =
dict(vars())
115 test_callback_py_obj,
116 test_callback_std_func,
117 test_callback_virtual_func,
118 test_callback_pure_virtual_func,
119 test_cross_module_gil_released,
120 test_cross_module_gil_acquired,
121 test_cross_module_gil_inner_custom_released,
122 test_cross_module_gil_inner_custom_acquired,
123 test_cross_module_gil_inner_pybind11_released,
124 test_cross_module_gil_inner_pybind11_acquired,
125 test_cross_module_gil_nested_custom_released,
126 test_cross_module_gil_nested_custom_acquired,
127 test_cross_module_gil_nested_pybind11_released,
128 test_cross_module_gil_nested_pybind11_acquired,
129 test_release_acquire,
131 test_multi_acquire_release_cross_module,
137 for key, value
in VARS_BEFORE_ALL_BASIC_TESTS.items():
138 if not key.startswith(
"test_"):
140 assert value
in ALL_BASIC_TESTS
142 assert len(ALL_BASIC_TESTS) == num_found
146 m.intentional_deadlock()
149 ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK = ALL_BASIC_TESTS + (_intentional_deadlock,)
153 test_fn = target
if len(args) == 0
else args[0]
155 timeout = 0.1
if test_fn
is _intentional_deadlock
else 10
156 process = multiprocessing.Process(target=target, args=args, kwargs=kwargs)
157 process.daemon =
True
159 t_start = time.time()
163 "\nprocess.pid STARTED", process.pid, (sys.argv, target, args, kwargs)
165 print(f
"COPY-PASTE-THIS: gdb {sys.argv[0]} -p {process.pid}", flush=
True)
166 process.join(timeout=timeout)
168 print(
"\nprocess.pid JOINED", process.pid, flush=
True)
169 t_delta = time.time() - t_start
170 if process.exitcode == 66
and m.defined_THREAD_SANITIZER:
175 "ThreadSanitizer: starting new threads after multi-threaded fork is not supported."
177 elif test_fn
is _intentional_deadlock:
178 assert process.exitcode
is None
181 if process.exitcode
is None:
182 assert t_delta > 0.9 * timeout
183 msg =
"DEADLOCK, most likely, exactly what this test is meant to detect."
184 if env.PYPY
and env.WIN:
186 raise RuntimeError(msg)
187 return process.exitcode
189 if process.is_alive():
195 for _
in range(num_threads):
196 thread = threading.Thread(target=test_fn)
200 threads.append(thread)
203 for thread
in threads:
208 @pytest.mark.parametrize(
"test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
210 """Makes sure there is no GIL deadlock when running in a thread.
212 It runs in a separate process to be able to stop and assert if it deadlocks.
214 assert _run_in_process(_run_in_threads, test_fn, num_threads=1, parallel=
False) == 0
218 @pytest.mark.parametrize(
"test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
220 """Makes sure there is no GIL deadlock when running in a thread multiple times in parallel.
222 It runs in a separate process to be able to stop and assert if it deadlocks.
224 assert _run_in_process(_run_in_threads, test_fn, num_threads=8, parallel=
True) == 0
228 @pytest.mark.parametrize(
"test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
230 """Makes sure there is no GIL deadlock when running in a thread multiple times sequentially.
232 It runs in a separate process to be able to stop and assert if it deadlocks.
234 assert _run_in_process(_run_in_threads, test_fn, num_threads=8, parallel=
False) == 0
238 @pytest.mark.parametrize(
"test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
240 """Makes sure there is no GIL deadlock when using processes.
242 This test is for completion, but it was never an issue.