diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index 78ccb3008b0..804cccc1923 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -611,48 +611,97 @@ 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 = f"timeout = {_FINALIZE_TIMEOUT!r}\n" + textwrap.dedent( + """ + import ctypes + import gc + import threading + import time + + from cuda.core import Device + from cuda.core.graph import GraphDefinition - @pending_callback_type - def noop_pending_call(_): - return 0 + class ThreadRecordingCallback: + def __init__(self, finalized_threads): + self.finalized_threads = finalized_threads - 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 = [] + 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() - del first_callback, first_graph + worker = threading.Thread(target=fill_queue_and_destroy) + worker.start() + assert worker_done.wait(timeout=5) + worker.join() + assert queue_was_full == [True] - 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] - - # A later safe cuda-core close retries after the main thread has had an - # opportunity to drain the foreign pending calls. - retry_builder = Device().create_graph_builder() - retry_builder.close() - - _wait_until(lambda: len(finalized_threads) == 1) - assert set(finalized_threads) == {main_thread} + # Free space before the cuda-core entry that retries cleanup scheduling. + assert make_pending_calls() == 0 + + retry_builder = Device().create_graph_builder() + retry_builder.close() + + wait_until(lambda: len(finalized_threads) == 1) + 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")