tests/test_gil_scoped.py

from __future__ import annotations

import multiprocessing
import queue
import sys
import sysconfig
import threading
import time

import pytest

import env
from pybind11_tests import gil_scoped as m

# Test collection seems to hold the gil
# These tests have rare flakes in nogil; since they
# are testing the gil, they are skipped at the moment.
skipif_free_threaded = pytest.mark.skipif(
    sysconfig.get_config_var("Py_GIL_DISABLED"),
    reason="Flaky without the GIL",
)


class ExtendedVirtClass(m.VirtClass):
    def virtual_func(self):
        pass

    def pure_virtual_func(self):
        pass


def test_callback_py_obj():
    m.test_callback_py_obj(lambda: None)


def test_callback_std_func():
    m.test_callback_std_func(lambda: None)


def test_callback_virtual_func():
    extended = ExtendedVirtClass()
    m.test_callback_virtual_func(extended)


def test_callback_pure_virtual_func():
    extended = ExtendedVirtClass()
    m.test_callback_pure_virtual_func(extended)


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


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


def test_cross_module_gil_inner_custom_released():
    """Makes sure that the GIL can be acquired/released by another module
    from a GIL-released state using custom locking logic."""
    m.test_cross_module_gil_inner_custom_released()


def test_cross_module_gil_inner_custom_acquired():
    """Makes sure that the GIL can be acquired/acquired by another module
    from a GIL-acquired state using custom locking logic."""
    m.test_cross_module_gil_inner_custom_acquired()


def test_cross_module_gil_inner_pybind11_released():
    """Makes sure that the GIL can be acquired/released by another module
    from a GIL-released state using pybind11 locking logic."""
    m.test_cross_module_gil_inner_pybind11_released()


def test_cross_module_gil_inner_pybind11_acquired():
    """Makes sure that the GIL can be acquired/acquired by another module
    from a GIL-acquired state using pybind11 locking logic."""
    m.test_cross_module_gil_inner_pybind11_acquired()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_cross_module_gil_nested_custom_released():
    """Makes sure that the GIL can be nested acquired/released by another module
    from a GIL-released state using custom locking logic."""
    m.test_cross_module_gil_nested_custom_released()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_cross_module_gil_nested_custom_acquired():
    """Makes sure that the GIL can be nested acquired/acquired by another module
    from a GIL-acquired state using custom locking logic."""
    m.test_cross_module_gil_nested_custom_acquired()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_cross_module_gil_nested_pybind11_released():
    """Makes sure that the GIL can be nested acquired/released by another module
    from a GIL-released state using pybind11 locking logic."""
    m.test_cross_module_gil_nested_pybind11_released()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_cross_module_gil_nested_pybind11_acquired():
    """Makes sure that the GIL can be nested acquired/acquired by another module
    from a GIL-acquired state using pybind11 locking logic."""
    m.test_cross_module_gil_nested_pybind11_acquired()


def test_release_acquire():
    assert m.test_release_acquire(0xAB) == "171"


def test_nested_acquire():
    assert m.test_nested_acquire(0xAB) == "171"


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.skipif(
    env.GRAALPY and sys.platform == "darwin",
    reason="Transiently crashes on GraalPy on OS X",
)
def test_multi_acquire_release_cross_module():
    for bits in range(16 * 8):
        internals_ids = m.test_multi_acquire_release_cross_module(bits)
        assert len(internals_ids) == (2 if bits % 8 else 1)


# Intentionally putting human review in the loop here, to guard against accidents.
VARS_BEFORE_ALL_BASIC_TESTS = dict(vars())  # Make a copy of the dict (critical).
ALL_BASIC_TESTS = (
    test_callback_py_obj,
    test_callback_std_func,
    test_callback_virtual_func,
    test_callback_pure_virtual_func,
    test_cross_module_gil_released,
    test_cross_module_gil_acquired,
    test_cross_module_gil_inner_custom_released,
    test_cross_module_gil_inner_custom_acquired,
    test_cross_module_gil_inner_pybind11_released,
    test_cross_module_gil_inner_pybind11_acquired,
    test_cross_module_gil_nested_custom_released,
    test_cross_module_gil_nested_custom_acquired,
    test_cross_module_gil_nested_pybind11_released,
    test_cross_module_gil_nested_pybind11_acquired,
    test_release_acquire,
    test_nested_acquire,
    test_multi_acquire_release_cross_module,
)


# Keep first-use initialization of the helper module's private internals isolated,
# both with and without the GIL held. Batched cases intentionally share an interpreter;
# their first call into the helper is test_cross_module_gil_released.
PROCESS_TEST_CASES = (
    pytest.param((test_cross_module_gil_released,), id="first_use_gil_released"),
    pytest.param((test_cross_module_gil_acquired,), id="first_use_gil_acquired"),
    pytest.param(ALL_BASIC_TESTS, id="all_basic_tests"),
)


def test_all_basic_tests_completeness():
    num_found = 0
    for key, value in VARS_BEFORE_ALL_BASIC_TESTS.items():
        if not key.startswith("test_"):
            continue
        assert value in ALL_BASIC_TESTS
        num_found += 1
    assert len(ALL_BASIC_TESTS) == num_found


# Defined after ALL_BASIC_TESTS on purpose: this test is a regression for the
# `gil_scoped_release` + factory `py::init` path, not a deadlock check, so it should not
# run in the subprocess batches (whose cases get at least a 10s execution allowance;
# on Windows this test is much slower due to sleep timer granularity and GIL handoff costs).
@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_init_factory_gil_released_concurrent_construction():
    """Concurrent construction via a factory `py::init` with `call_guard<gil_scoped_release>`.

    `init_instance` runs while the GIL is released and must internally acquire the GIL
    before touching the instance map. Without that fix this aborts with
    "pybind11_object_dealloc(): Tried to deallocate unregistered instance!" (races on
    `internals.registered_instances`, which is unguarded on GIL builds). On free-threaded
    builds the detached thread state instead segfaults in `PyCriticalSection_BeginMutex`
    (via `get_type_info`), even without concurrency.
    """
    num_threads = 8
    iterations = 100

    def construct_many():
        for _ in range(iterations):
            instance = m.SlowInit(0)
            del instance  # Destructor runs with the GIL held (deregistration).

    threads = [threading.Thread(target=construct_many) for _ in range(num_threads)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()


def _intentional_deadlock():
    m.intentional_deadlock()


def _run_in_process_target(target, args, kwargs, status_writer):
    with status_writer:
        status_writer.send("started")
        target(*args, **kwargs)
        status_writer.send("finished")


def _run_in_process(target, *args, timeout=10, get_progress=None, **kwargs):
    if env.ANDROID or env.IOS or sys.platform.startswith("emscripten"):
        pytest.skip("Requires subprocess support")

    # Events allocate named semaphores whose delayed cleanup can fail on PyPy.
    status_reader, status_writer = multiprocessing.Pipe(duplex=False)
    # Keep the parent's writer open through cleanup so EOF cannot look like a signal.
    with status_reader, status_writer:
        process = multiprocessing.Process(
            target=_run_in_process_target,
            args=(target, args, kwargs, status_writer),
        )
        process.daemon = True
        try:
            t_start = time.time()
            process.start()
            if timeout >= 100:  # For debugging.
                print(
                    "\nprocess.pid STARTED",
                    process.pid,
                    (sys.argv, target, args, kwargs),
                )
                print(
                    f"COPY-PASTE-THIS: gdb {sys.argv[0]} -p {process.pid}", flush=True
                )
            # Imports can take longer than the execution timeout, especially with spawn.
            child_started = (
                status_reader.poll(timeout=10) and status_reader.recv() == "started"
            )
            child_finished = False
            if child_started:
                while True:
                    previous_progress = None if get_progress is None else get_progress()
                    child_finished = (
                        status_reader.poll(timeout=timeout)
                        and status_reader.recv() == "finished"
                    )
                    if (
                        child_finished
                        or process.exitcode is not None
                        or get_progress is None
                        or get_progress() == previous_progress
                    ):
                        break
                    # Give a newly started case a full execution allowance.
                    # A deadlock may take one extra wait interval to detect.
            # A returned test must not look deadlocked while the process shuts down.
            process.join(timeout=10 if child_finished else 0)
            if timeout >= 100:
                print("\nprocess.pid JOINED", process.pid, flush=True)
            t_delta = time.time() - t_start
            if process.exitcode == 66 and m.defined_THREAD_SANITIZER:  # Issue #2754
                # WOULD-BE-NICE-TO-HAVE: Check that the message below is actually in the output.
                # Maybe this could work:
                # https://gist.github.com/alexeygrigorev/01ce847f2e721b513b42ea4a6c96905e
                pytest.skip(
                    "ThreadSanitizer: starting new threads after multi-threaded fork is not supported."
                )
            if not child_started:
                raise RuntimeError("Subprocess did not start within 10 seconds.")
            if child_finished and process.exitcode is None:
                raise RuntimeError(
                    "Subprocess did not exit within 10 seconds after the test returned."
                )

            if process.exitcode is None:
                assert t_delta > 0.9 * timeout
                msg = (
                    "DEADLOCK, most likely, exactly what this test is meant to detect."
                )
                soabi = sysconfig.get_config_var("SOABI")
                if env.WIN and env.PYPY:
                    pytest.xfail(f"[TEST-GIL-SCOPED] {soabi} PyPy: " + msg)
                if env.MACOS:
                    if not env.sys_is_gil_enabled():
                        pytest.xfail(
                            f"[TEST-GIL-SCOPED] {soabi} with GIL disabled: " + msg
                        )
                    if env.PY_GIL_DISABLED:
                        pytest.xfail(f"[TEST-GIL-SCOPED] {soabi}: " + msg)
                raise RuntimeError(msg)
            return process.exitcode
        finally:
            if process.is_alive():
                process.terminate()
            if process.pid is not None:
                process.join()


def _run_in_threads(test_fn, num_threads, parallel):
    exceptions = queue.SimpleQueue()

    def run_test():
        try:
            test_fn()
        except BaseException as error:
            exceptions.put(error)

    threads = []
    for _ in range(num_threads):
        thread = threading.Thread(target=run_test)
        thread.daemon = True
        thread.start()
        if parallel:
            threads.append(thread)
        else:
            thread.join()
    for thread in threads:
        thread.join()
    # Thread exceptions otherwise only print a traceback, leaving a successful process.
    if not exceptions.empty():
        raise exceptions.get()


def _run_test_cases(test_fns, progress_writer, num_threads, parallel):
    for index, test_fn in enumerate(test_fns):
        progress_writer.send(index)
        if num_threads is None:
            test_fn()
        else:
            # Fresh threads per case preserve thread-state creation and cleanup.
            _run_in_threads(test_fn, num_threads=num_threads, parallel=parallel)


def _run_test_cases_in_process(test_fns, *, num_threads=None, parallel=False):
    if env.ANDROID or env.IOS or sys.platform.startswith("emscripten"):
        pytest.skip("Requires subprocess support")

    # Shared ctypes can fail on Windows PyPy: https://github.com/pypy/pypy/issues/3488.
    progress_reader, progress_writer = multiprocessing.Pipe(duplex=False)
    with progress_reader, progress_writer:
        current_test = -1

        def get_progress():
            nonlocal current_test
            try:
                while progress_reader.poll():
                    current_test = progress_reader.recv()
            except EOFError:
                pass
            return current_test

        def current_test_name():
            index = get_progress()
            return test_fns[index].__name__ if index >= 0 else "before the first case"

        try:
            exitcode = _run_in_process(
                _run_test_cases,
                test_fns,
                progress_writer,
                num_threads,
                parallel,
                get_progress=get_progress,
            )
        except RuntimeError as error:
            raise RuntimeError(
                f"{error} (while running {current_test_name()})"
            ) from error
        assert exitcode == 0, (
            f"Subprocess exited with {exitcode} while running {current_test_name()}"
        )


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.parametrize("parallel", [False, True], ids=["sequential", "parallel"])
def test_run_in_threads_propagates_exception(parallel):
    def fail():
        raise AssertionError("worker failure")

    with pytest.raises(AssertionError, match="worker failure"):
        _run_in_threads(fail, num_threads=2, parallel=parallel)


@skipif_free_threaded
@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.parametrize("test_fns", PROCESS_TEST_CASES)
@pytest.mark.skipif(
    "env.GRAALPY",
    reason="GraalPy transiently complains about unfinished threads at process exit",
)
def test_run_in_process_multiple_threads_parallel(test_fns):
    """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.
    """
    _run_test_cases_in_process(test_fns, num_threads=8, parallel=True)


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.parametrize("test_fns", PROCESS_TEST_CASES)
@pytest.mark.skipif(
    "env.GRAALPY",
    reason="GraalPy transiently complains about unfinished threads at process exit",
)
def test_run_in_process_multiple_threads_sequential(test_fns):
    """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.
    """
    # Its first worker also covers the former one-thread parametrization.
    _run_test_cases_in_process(test_fns, num_threads=8, parallel=False)


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.parametrize("test_fns", PROCESS_TEST_CASES)
@pytest.mark.skipif(
    "env.GRAALPY",
    reason="GraalPy transiently complains about unfinished threads at process exit",
)
def test_run_in_process_direct(test_fns):
    """Exercise the main thread of a fresh interpreter, including its initial GIL state."""
    _run_test_cases_in_process(test_fns)


@pytest.mark.skipif(
    env.GRAALPY,
    reason="GraalPy transiently complains about unfinished threads at process exit",
)
def test_run_in_process_batch_without_sharedctypes(monkeypatch):
    def unavailable_value(*_args, **_kwargs):
        raise FileExistsError("Cannot find name for new mmap")

    monkeypatch.setattr(multiprocessing, "Value", unavailable_value)
    _run_test_cases_in_process((test_callback_py_obj,))


@pytest.mark.skipif(
    env.GRAALPY, reason="GraalPy can leave unfinished threads at process exit"
)
def test_run_in_process_without_events(monkeypatch):
    def unavailable_event(*_args, **_kwargs):
        raise RuntimeError("multiprocessing.Event is unavailable")

    monkeypatch.setattr(multiprocessing, "Event", unavailable_event)
    assert _run_in_process(test_callback_py_obj) == 0


@pytest.mark.skipif(env.PY_GIL_DISABLED, reason="Requires the GIL to deadlock")
@pytest.mark.skipif(
    env.GRAALPY, reason="GraalPy can leave unfinished threads at process exit"
)
def test_run_in_process_detects_deadlock():
    # Known platform exceptions still indicate that the deadlock was detected.
    with pytest.raises((RuntimeError, pytest.xfail.Exception), match="DEADLOCK"):
        _run_in_process(_intentional_deadlock, timeout=0.1)


def _run_with_delay(target, args, delay_before):
    if delay_before:
        time.sleep(0.2)
    target(*args)
    if not delay_before:
        time.sleep(0.2)


@pytest.mark.parametrize("delay_before", [True, False], ids=["startup", "shutdown"])
def test_run_in_process_slow_start_or_exit(monkeypatch, delay_before):
    """Slow startup and shutdown must not look like a deadlock in the test itself."""
    original_process = multiprocessing.Process

    def delayed_process(*, target, args):
        return original_process(
            target=_run_with_delay, args=(target, args, delay_before)
        )

    monkeypatch.setattr(multiprocessing, "Process", delayed_process)
    assert _run_in_process(test_callback_py_obj, timeout=0.1) == 0