1import multiprocessing 2import threading 3from pybind11_tests import gil_scoped as m 4 5 6def _run_in_process(target, *args, **kwargs): 7 """Runs target in process and returns its exitcode after 10s (None if still alive).""" 8 process = multiprocessing.Process(target=target, args=args, kwargs=kwargs) 9 process.daemon = True 10 try: 11 process.start() 12 # Do not need to wait much, 10s should be more than enough. 13 process.join(timeout=10) 14 return process.exitcode 15 finally: 16 if process.is_alive(): 17 process.terminate() 18 19 20def _python_to_cpp_to_python(): 21 """Calls different C++ functions that come back to Python.""" 22 class ExtendedVirtClass(m.VirtClass): 23 def virtual_func(self): 24 pass 25 26 def pure_virtual_func(self): 27 pass 28 29 extended = ExtendedVirtClass() 30 m.test_callback_py_obj(lambda: None) 31 m.test_callback_std_func(lambda: None) 32 m.test_callback_virtual_func(extended) 33 m.test_callback_pure_virtual_func(extended) 34 35 36def _python_to_cpp_to_python_from_threads(num_threads, parallel=False): 37 """Calls different C++ functions that come back to Python, from Python threads.""" 38 threads = [] 39 for _ in range(num_threads): 40 thread = threading.Thread(target=_python_to_cpp_to_python) 41 thread.daemon = True 42 thread.start() 43 if parallel: 44 threads.append(thread) 45 else: 46 thread.join() 47 for thread in threads: 48 thread.join() 49 50 51def test_python_to_cpp_to_python_from_thread(): 52 """Makes sure there is no GIL deadlock when running in a thread. 53 54 It runs in a separate process to be able to stop and assert if it deadlocks. 55 """ 56 assert _run_in_process(_python_to_cpp_to_python_from_threads, 1) == 0 57 58 59def test_python_to_cpp_to_python_from_thread_multiple_parallel(): 60 """Makes sure there is no GIL deadlock when running in a thread multiple times in parallel. 61 62 It runs in a separate process to be able to stop and assert if it deadlocks. 63 """ 64 assert _run_in_process(_python_to_cpp_to_python_from_threads, 8, parallel=True) == 0 65 66 67def test_python_to_cpp_to_python_from_thread_multiple_sequential(): 68 """Makes sure there is no GIL deadlock when running in a thread multiple times sequentially. 69 70 It runs in a separate process to be able to stop and assert if it deadlocks. 71 """ 72 assert _run_in_process(_python_to_cpp_to_python_from_threads, 8, parallel=False) == 0 73 74 75def test_python_to_cpp_to_python_from_process(): 76 """Makes sure there is no GIL deadlock when using processes. 77 78 This test is for completion, but it was never an issue. 79 """ 80 assert _run_in_process(_python_to_cpp_to_python) == 0 81 82 83def test_cross_module_gil(): 84 """Makes sure that the GIL can be acquired by another module from a GIL-released state.""" 85 m.test_cross_module_gil() # Should not raise a SIGSEGV 86