9 from pybind11_tests
import gil_scoped
as m
21 m.test_callback_py_obj(
lambda:
None)
25 m.test_callback_std_func(
lambda:
None)
30 m.test_callback_virtual_func(extended)
35 m.test_callback_pure_virtual_func(extended)
39 """Makes sure that the GIL can be acquired by another module from a GIL-released state."""
40 m.test_cross_module_gil_released()
44 """Makes sure that the GIL can be acquired by another module from a GIL-acquired state."""
45 m.test_cross_module_gil_acquired()
49 """Makes sure that the GIL can be acquired/released by another module
50 from a GIL-released state using custom locking logic."""
51 m.test_cross_module_gil_inner_custom_released()
55 """Makes sure that the GIL can be acquired/acquired by another module
56 from a GIL-acquired state using custom locking logic."""
57 m.test_cross_module_gil_inner_custom_acquired()
61 """Makes sure that the GIL can be acquired/released by another module
62 from a GIL-released state using pybind11 locking logic."""
63 m.test_cross_module_gil_inner_pybind11_released()
67 """Makes sure that the GIL can be acquired/acquired by another module
68 from a GIL-acquired state using pybind11 locking logic."""
69 m.test_cross_module_gil_inner_pybind11_acquired()
73 """Makes sure that the GIL can be nested acquired/released by another module
74 from a GIL-released state using custom locking logic."""
75 m.test_cross_module_gil_nested_custom_released()
79 """Makes sure that the GIL can be nested acquired/acquired by another module
80 from a GIL-acquired state using custom locking logic."""
81 m.test_cross_module_gil_nested_custom_acquired()
85 """Makes sure that the GIL can be nested acquired/released by another module
86 from a GIL-released state using pybind11 locking logic."""
87 m.test_cross_module_gil_nested_pybind11_released()
91 """Makes sure that the GIL can be nested acquired/acquired by another module
92 from a GIL-acquired state using pybind11 locking logic."""
93 m.test_cross_module_gil_nested_pybind11_acquired()
97 assert m.test_release_acquire(0xAB) ==
"171"
101 assert m.test_nested_acquire(0xAB) ==
"171"
105 for bits
in range(16 * 8):
106 internals_ids = m.test_multi_acquire_release_cross_module(bits)
107 assert len(internals_ids) == 2
if bits % 8
else 1
111 VARS_BEFORE_ALL_BASIC_TESTS =
dict(vars())
113 test_callback_py_obj,
114 test_callback_std_func,
115 test_callback_virtual_func,
116 test_callback_pure_virtual_func,
117 test_cross_module_gil_released,
118 test_cross_module_gil_acquired,
119 test_cross_module_gil_inner_custom_released,
120 test_cross_module_gil_inner_custom_acquired,
121 test_cross_module_gil_inner_pybind11_released,
122 test_cross_module_gil_inner_pybind11_acquired,
123 test_cross_module_gil_nested_custom_released,
124 test_cross_module_gil_nested_custom_acquired,
125 test_cross_module_gil_nested_pybind11_released,
126 test_cross_module_gil_nested_pybind11_acquired,
127 test_release_acquire,
129 test_multi_acquire_release_cross_module,
135 for key, value
in VARS_BEFORE_ALL_BASIC_TESTS.items():
136 if not key.startswith(
"test_"):
138 assert value
in ALL_BASIC_TESTS
140 assert len(ALL_BASIC_TESTS) == num_found
144 m.intentional_deadlock()
147 ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK = ALL_BASIC_TESTS + (_intentional_deadlock,)
151 test_fn = target
if len(args) == 0
else args[0]
153 timeout = 0.1
if test_fn
is _intentional_deadlock
else 10
154 process = multiprocessing.Process(target=target, args=args, kwargs=kwargs)
155 process.daemon =
True
157 t_start = time.time()
161 "\nprocess.pid STARTED", process.pid, (sys.argv, target, args, kwargs)
163 print(f
"COPY-PASTE-THIS: gdb {sys.argv[0]} -p {process.pid}", flush=
True)
164 process.join(timeout=timeout)
166 print(
"\nprocess.pid JOINED", process.pid, flush=
True)
167 t_delta = time.time() - t_start
168 if process.exitcode == 66
and m.defined_THREAD_SANITIZER:
173 "ThreadSanitizer: starting new threads after multi-threaded fork is not supported."
175 elif test_fn
is _intentional_deadlock:
176 assert process.exitcode
is None
179 if process.exitcode
is None:
180 assert t_delta > 0.9 * timeout
181 msg =
"DEADLOCK, most likely, exactly what this test is meant to detect."
182 if env.PYPY
and env.WIN:
184 raise RuntimeError(msg)
185 return process.exitcode
187 if process.is_alive():
193 for _
in range(num_threads):
194 thread = threading.Thread(target=test_fn)
198 threads.append(thread)
201 for thread
in threads:
206 @pytest.mark.parametrize(
"test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
208 """Makes sure there is no GIL deadlock when running in a thread.
210 It runs in a separate process to be able to stop and assert if it deadlocks.
212 assert _run_in_process(_run_in_threads, test_fn, num_threads=1, parallel=
False) == 0
216 @pytest.mark.parametrize(
"test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
218 """Makes sure there is no GIL deadlock when running in a thread multiple times in parallel.
220 It runs in a separate process to be able to stop and assert if it deadlocks.
222 assert _run_in_process(_run_in_threads, test_fn, num_threads=8, parallel=
True) == 0
226 @pytest.mark.parametrize(
"test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
228 """Makes sure there is no GIL deadlock when running in a thread multiple times sequentially.
230 It runs in a separate process to be able to stop and assert if it deadlocks.
232 assert _run_in_process(_run_in_threads, test_fn, num_threads=8, parallel=
False) == 0
236 @pytest.mark.parametrize(
"test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
238 """Makes sure there is no GIL deadlock when using processes.
240 This test is for completion, but it was never an issue.