test_gil_scoped.py revision 14299:2fbea9df56d2 
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