From 778656f12bb06701b2601d2711bff358efb525b3 Mon Sep 17 00:00:00 2001 From: "Ralf W. Grosse-Kunstleve" Date: Sat, 24 Oct 2026 08:11:54 -0700 Subject: [PATCH 1/5] test: distinguish GIL deadlocks from subprocess startup and shutdown Wait for explicit child-start and function-return signals so slow imports or interpreter shutdown cannot masquerade as a blocked test. Replace the intentional-deadlock special case with a dedicated GIL-only detector control, and add regressions for delayed startup and shutdown. Validation: CPython 3.14.4 focused GIL suite, 90 passed on regular Python; 72 passed and 18 skipped on free-threaded Python. Spawn checks and the all-files pre-commit hooks passed. Both timing regressions fail when their corresponding safeguards are removed. --- tests/test_gil_scoped.py | 80 ++++++++++++++++++++++++++++++---------- 1 file changed, 60 insertions(+), 20 deletions(-) diff --git a/tests/test_gil_scoped.py b/tests/test_gil_scoped.py index 8c0ac9597d..ff9e8d3fca 100644 --- a/tests/test_gil_scoped.py +++ b/tests/test_gil_scoped.py @@ -195,17 +195,22 @@ def _intentional_deadlock(): m.intentional_deadlock() -ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK = (*ALL_BASIC_TESTS, _intentional_deadlock) +def _run_in_process_target(target, args, kwargs, event_started, event_finished): + event_started.set() + target(*args, **kwargs) + event_finished.set() -def _run_in_process(target, *args, **kwargs): +def _run_in_process(target, *args, timeout=10, **kwargs): if env.ANDROID or env.IOS or sys.platform.startswith("emscripten"): pytest.skip("Requires subprocess support") - test_fn = target if len(args) == 0 else args[0] - # Do not need to wait much, 10s should be more than enough. - timeout = 0.1 if test_fn is _intentional_deadlock else 10 - process = multiprocessing.Process(target=target, args=args, kwargs=kwargs) + event_started = multiprocessing.Event() + event_finished = multiprocessing.Event() + process = multiprocessing.Process( + target=_run_in_process_target, + args=(target, args, kwargs, event_started, event_finished), + ) process.daemon = True try: t_start = time.time() @@ -215,7 +220,11 @@ def _run_in_process(target, *args, **kwargs): "\nprocess.pid STARTED", process.pid, (sys.argv, target, args, kwargs) ) print(f"COPY-PASTE-THIS: gdb {sys.argv[0]} -p {process.pid}", flush=True) - process.join(timeout=timeout) + # Imports can take longer than the execution timeout, especially with spawn. + child_started = event_started.wait(timeout=10) + child_finished = child_started and event_finished.wait(timeout=timeout) + # 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 @@ -226,9 +235,12 @@ def _run_in_process(target, *args, **kwargs): pytest.skip( "ThreadSanitizer: starting new threads after multi-threaded fork is not supported." ) - elif test_fn is _intentional_deadlock: - assert process.exitcode is None - return 0 + 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 @@ -246,6 +258,8 @@ def _run_in_process(target, *args, **kwargs): finally: if process.is_alive(): process.terminate() + if process.pid is not None: + process.join() def _run_in_threads(test_fn, num_threads, parallel): @@ -263,7 +277,7 @@ def _run_in_threads(test_fn, num_threads, parallel): @pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads") -@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK) +@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS) @pytest.mark.skipif( "env.GRAALPY", reason="GraalPy transiently complains about unfinished threads at process exit", @@ -278,7 +292,7 @@ def test_run_in_process_one_thread(test_fn): @skipif_not_free_threaded @pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads") -@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK) +@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS) @pytest.mark.skipif( "env.GRAALPY", reason="GraalPy transiently complains about unfinished threads at process exit", @@ -292,7 +306,7 @@ def test_run_in_process_multiple_threads_parallel(test_fn): @pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads") -@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK) +@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS) @pytest.mark.skipif( "env.GRAALPY", reason="GraalPy transiently complains about unfinished threads at process exit", @@ -306,13 +320,7 @@ def test_run_in_process_multiple_threads_sequential(test_fn): @pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads") -@pytest.mark.parametrize( - "test_fn", - [ - *ALL_BASIC_TESTS, - pytest.param(_intentional_deadlock, marks=skipif_not_free_threaded), - ], -) +@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS) @pytest.mark.skipif( "env.GRAALPY", reason="GraalPy transiently complains about unfinished threads at process exit", @@ -323,3 +331,35 @@ def test_run_in_process_direct(test_fn): This test is for completion, but it was never an issue. """ assert _run_in_process(test_fn) == 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 From c2a4d2e31b706d600fbdc755fc6c1d23b4d62bf5 Mon Sep 17 00:00:00 2001 From: "Ralf W. Grosse-Kunstleve" Date: Sat, 24 Oct 2026 08:36:33 -0700 Subject: [PATCH 2/5] test: rename free-threaded skip marker to match its condition Rename skipif_not_free_threaded to skipif_free_threaded so the name states when pytest skips the test. The predicate and test behavior are unchanged. Validation: all repository pre-commit hooks passed. --- tests/test_gil_scoped.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_gil_scoped.py b/tests/test_gil_scoped.py index ff9e8d3fca..5580aef254 100644 --- a/tests/test_gil_scoped.py +++ b/tests/test_gil_scoped.py @@ -14,7 +14,7 @@ # 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_not_free_threaded = pytest.mark.skipif( +skipif_free_threaded = pytest.mark.skipif( sysconfig.get_config_var("Py_GIL_DISABLED"), reason="Flaky without the GIL", ) @@ -290,7 +290,7 @@ def test_run_in_process_one_thread(test_fn): assert _run_in_process(_run_in_threads, test_fn, num_threads=1, parallel=False) == 0 -@skipif_not_free_threaded +@skipif_free_threaded @pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads") @pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS) @pytest.mark.skipif( From 867e7df79234ba8e0d0f0fbf55e0583124b519af Mon Sep 17 00:00:00 2001 From: "Ralf W. Grosse-Kunstleve" Date: Sat, 24 Oct 2026 09:13:16 -0700 Subject: [PATCH 3/5] test: check the internals count for every GIL operation pattern --- tests/test_gil_scoped.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_gil_scoped.py b/tests/test_gil_scoped.py index 5580aef254..a57b963fbb 100644 --- a/tests/test_gil_scoped.py +++ b/tests/test_gil_scoped.py @@ -124,7 +124,7 @@ def test_nested_acquire(): 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 + assert len(internals_ids) == (2 if bits % 8 else 1) # Intentionally putting human review in the loop here, to guard against accidents. From d11cf366e112d27d77cdf357de817de12e2820a9 Mon Sep 17 00:00:00 2001 From: "Ralf W. Grosse-Kunstleve" Date: Sat, 24 Oct 2026 09:15:17 -0700 Subject: [PATCH 4/5] test: propagate exceptions from GIL test worker threads --- tests/test_gil_scoped.py | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/tests/test_gil_scoped.py b/tests/test_gil_scoped.py index a57b963fbb..e77589274e 100644 --- a/tests/test_gil_scoped.py +++ b/tests/test_gil_scoped.py @@ -1,6 +1,7 @@ from __future__ import annotations import multiprocessing +import queue import sys import sysconfig import threading @@ -263,9 +264,17 @@ def _run_in_process(target, *args, timeout=10, **kwargs): 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=test_fn) + thread = threading.Thread(target=run_test) thread.daemon = True thread.start() if parallel: @@ -274,6 +283,19 @@ def _run_in_threads(test_fn, num_threads, parallel): 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() + + +@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) @pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads") From d50935a3c26627c45f5e15a300b8ee04c0f20178 Mon Sep 17 00:00:00 2001 From: "Ralf W. Grosse-Kunstleve" Date: Sat, 24 Oct 2026 21:20:27 -0700 Subject: [PATCH 5/5] test: avoid semaphore finalizers in GIL subprocess tests Replace the started/finished multiprocessing.Events with ordered messages on a one-way pipe. Each Event allocates named semaphores under forkserver; the per-case subprocess matrix can accumulate hundreds before PyPy garbage collection, triggering reentrant resource_tracker cleanup in later tests. Preserve separate startup, execution, and shutdown allowances. Keep the parent writer open until child cleanup so EOF cannot masquerade as a phase signal, and close all connection endpoints deterministically. Add a regression that disables multiprocessing.Event and runs a real subprocess successfully. It fails with the previous harness. Validation: clean CPython 3.14.4 suites passed (1454 regular, 1438 free-threaded; Catch2 31 cases/1709 assertions each). Linux PyPy 3.11/8.0.1 full serial Python suite passed (1065 passed). Spawn checks and all pre-commit hooks passed. --- tests/test_gil_scoped.py | 142 +++++++++++++++++++++++---------------- 1 file changed, 85 insertions(+), 57 deletions(-) diff --git a/tests/test_gil_scoped.py b/tests/test_gil_scoped.py index e77589274e..2072e2661f 100644 --- a/tests/test_gil_scoped.py +++ b/tests/test_gil_scoped.py @@ -196,71 +196,88 @@ def _intentional_deadlock(): m.intentional_deadlock() -def _run_in_process_target(target, args, kwargs, event_started, event_finished): - event_started.set() - target(*args, **kwargs) - event_finished.set() +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, **kwargs): if env.ANDROID or env.IOS or sys.platform.startswith("emscripten"): pytest.skip("Requires subprocess support") - event_started = multiprocessing.Event() - event_finished = multiprocessing.Event() - process = multiprocessing.Process( - target=_run_in_process_target, - args=(target, args, kwargs, event_started, event_finished), - ) - 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 = event_started.wait(timeout=10) - child_finished = child_started and event_finished.wait(timeout=timeout) - # 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." + # 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" ) - 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." + child_finished = ( + child_started + and status_reader.poll(timeout=timeout) + and status_reader.recv() == "finished" ) - - 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() + # 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): @@ -355,6 +372,17 @@ def test_run_in_process_direct(test_fn): assert _run_in_process(test_fn) == 0 +@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"