ext/pybind11/tests/test_gil_scoped.py

import multiprocessing
import threading
from pybind11_tests import gil_scoped as m


def _run_in_process(target, *args, **kwargs):
    """Runs target in process and returns its exitcode after 10s (None if still alive)."""
    process = multiprocessing.Process(target=target, args=args, kwargs=kwargs)
    process.daemon = True
    try:
        process.start()
        # Do not need to wait much, 10s should be more than enough.
        process.join(timeout=10)
        return process.exitcode
    finally:
        if process.is_alive():
            process.terminate()


def _python_to_cpp_to_python():
    """Calls different C++ functions that come back to Python."""
    class ExtendedVirtClass(m.VirtClass):
        def virtual_func(self):
            pass

        def pure_virtual_func(self):
            pass

    extended = ExtendedVirtClass()
    m.test_callback_py_obj(lambda: None)
    m.test_callback_std_func(lambda: None)
    m.test_callback_virtual_func(extended)
    m.test_callback_pure_virtual_func(extended)


def _python_to_cpp_to_python_from_threads(num_threads, parallel=False):
    """Calls different C++ functions that come back to Python, from Python threads."""
    threads = []
    for _ in range(num_threads):
        thread = threading.Thread(target=_python_to_cpp_to_python)
        thread.daemon = True
        thread.start()
        if parallel:
            threads.append(thread)
        else:
            thread.join()
    for thread in threads:
        thread.join()


def test_python_to_cpp_to_python_from_thread():
    """Makes sure there is no GIL deadlock when running in a thread.

    It runs in a separate process to be able to stop and assert if it deadlocks.
    """
    assert _run_in_process(_python_to_cpp_to_python_from_threads, 1) == 0


def test_python_to_cpp_to_python_from_thread_multiple_parallel():
    """Makes sure there is no GIL deadlock when running in a thread multiple times in parallel.

    It runs in a separate process to be able to stop and assert if it deadlocks.
    """
    assert _run_in_process(_python_to_cpp_to_python_from_threads, 8, parallel=True) == 0


def test_python_to_cpp_to_python_from_thread_multiple_sequential():
    """Makes sure there is no GIL deadlock when running in a thread multiple times sequentially.

    It runs in a separate process to be able to stop and assert if it deadlocks.
    """
    assert _run_in_process(_python_to_cpp_to_python_from_threads, 8, parallel=False) == 0


def test_python_to_cpp_to_python_from_process():
    """Makes sure there is no GIL deadlock when using processes.

    This test is for completion, but it was never an issue.
    """
    assert _run_in_process(_python_to_cpp_to_python) == 0


def test_cross_module_gil():
    """Makes sure that the GIL can be acquired by another module from a GIL-released state."""
    m.test_cross_module_gil()  # Should not raise a SIGSEGV