From a41699c5e57d318addf4e3e2184f6f418a761102 Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Wed, 22 Jul 2026 08:18:56 -0700 Subject: [PATCH 1/2] test(cuda.core): isolate pending-call saturation cleanup Run the process-global queue probe in a subprocess and explicitly drain its synthetic callbacks so parallel free-threaded tests cannot contaminate one another. --- .../graph/test_graph_definition_lifetime.py | 135 ++++++++++++------ 1 file changed, 93 insertions(+), 42 deletions(-) diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index a01ff8162f3..8a97279b38f 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -384,51 +384,102 @@ def test_user_object_cleanup_is_coalesced_on_python_thread(init_cuda): @pytest.mark.agent_authored(model="gpt-5.6") -def test_pending_call_queue_saturation_preserves_cleanup(init_cuda): +def test_pending_call_queue_saturation_preserves_cleanup(tmp_path): """A full CPython queue neither strands nor mis-threads cleanup.""" - pending_callback_type = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p) - add_pending_call = ctypes.pythonapi.Py_AddPendingCall - add_pending_call.argtypes = [pending_callback_type, ctypes.c_void_p] - add_pending_call.restype = ctypes.c_int + code = textwrap.dedent( + """ + import ctypes + import gc + import threading + import time - @pending_callback_type - def noop_pending_call(_): - return 0 + from cuda.core import Device + from cuda.core.graph import GraphDefinition + from cuda_python_test_helpers import under_compute_sanitizer - finalized_threads = [] - main_thread = threading.get_ident() - first_callback = _ThreadRecordingCallback(finalized_threads) - first_graph = GraphDefinition() - first_graph.callback(first_callback) - graph_holder = [first_graph] - worker_done = threading.Event() - queue_was_full = [] - - del first_callback, first_graph - - def fill_queue_and_destroy(): - while add_pending_call(noop_pending_call, None) == 0: - pass - queue_was_full.append(True) - graph_holder.clear() - worker_done.set() - - worker = threading.Thread(target=fill_queue_and_destroy) - worker.start() - assert worker_done.wait(timeout=5) - worker.join() - assert queue_was_full == [True] - - # Preparing another attachment retries the first payload if its scheduling - # attempt observed the full queue. CUDA may also invoke its destructor - # later, after pending-call queue space is available. - second_callback = _ThreadRecordingCallback(finalized_threads) - second_graph = GraphDefinition() - second_graph.callback(second_callback) - del second_callback, second_graph - - _wait_until(lambda: len(finalized_threads) == 2) - assert set(finalized_threads) == {main_thread} + timeout = 30.0 if under_compute_sanitizer() else 5.0 + + class ThreadRecordingCallback: + def __init__(self, finalized_threads): + self.finalized_threads = finalized_threads + + def __call__(self): + pass + + def __del__(self): + self.finalized_threads.append(threading.get_ident()) + + def wait_until(predicate): + deadline = time.monotonic() + timeout + while True: + gc.collect() + if predicate(): + return + if time.monotonic() >= deadline: + break + time.sleep(0) + time.sleep(0.02) + raise AssertionError(f"condition not satisfied within {timeout}s") + + Device(0).set_current() + + pending_callback_type = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p) + add_pending_call = ctypes.pythonapi.Py_AddPendingCall + add_pending_call.argtypes = [pending_callback_type, ctypes.c_void_p] + add_pending_call.restype = ctypes.c_int + make_pending_calls = ctypes.pythonapi.Py_MakePendingCalls + make_pending_calls.argtypes = [] + make_pending_calls.restype = ctypes.c_int + + @pending_callback_type + def noop_pending_call(_): + return 0 + + finalized_threads = [] + main_thread = threading.get_ident() + first_callback = ThreadRecordingCallback(finalized_threads) + first_graph = GraphDefinition() + first_graph.callback(first_callback) + graph_holder = [first_graph] + worker_done = threading.Event() + queue_was_full = [] + + del first_callback, first_graph + + def fill_queue_and_destroy(): + while add_pending_call(noop_pending_call, None) == 0: + pass + queue_was_full.append(True) + graph_holder.clear() + worker_done.set() + + worker = threading.Thread(target=fill_queue_and_destroy) + worker.start() + assert worker_done.wait(timeout=5) + worker.join() + assert queue_was_full == [True] + + # Free space before the cuda-core entry that retries cleanup scheduling. + assert make_pending_calls() == 0 + + second_callback = ThreadRecordingCallback(finalized_threads) + second_graph = GraphDefinition() + second_graph.callback(second_callback) + del second_callback, second_graph + + wait_until(lambda: len(finalized_threads) == 2) + assert set(finalized_threads) == {main_thread} + """ + ) + result = subprocess.run( # noqa: S603 - controlled interpreter probe + [sys.executable, "-c", code], + capture_output=True, + text=True, + timeout=60, + # Isolate the process-global pending-call queue from parallel tests. + cwd=tmp_path, + ) + assert result.returncode == 0, result.stderr @pytest.mark.agent_authored(model="gpt-5.6") From e26e969201b9a70f1ea4166435ee2784aac5a4e2 Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Wed, 22 Jul 2026 11:32:20 -0700 Subject: [PATCH 2/2] test(cuda.core): keep subprocess probe self-contained Pass the finalization timeout into the child so process isolation does not depend on importing test helpers from the temporary working directory. --- cuda_core/tests/graph/test_graph_definition_lifetime.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index 8a97279b38f..d3420e35edb 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -386,7 +386,7 @@ def test_user_object_cleanup_is_coalesced_on_python_thread(init_cuda): @pytest.mark.agent_authored(model="gpt-5.6") def test_pending_call_queue_saturation_preserves_cleanup(tmp_path): """A full CPython queue neither strands nor mis-threads cleanup.""" - code = textwrap.dedent( + code = f"timeout = {_FINALIZE_TIMEOUT!r}\n" + textwrap.dedent( """ import ctypes import gc @@ -395,9 +395,6 @@ def test_pending_call_queue_saturation_preserves_cleanup(tmp_path): from cuda.core import Device from cuda.core.graph import GraphDefinition - from cuda_python_test_helpers import under_compute_sanitizer - - timeout = 30.0 if under_compute_sanitizer() else 5.0 class ThreadRecordingCallback: def __init__(self, finalized_threads):