From 6bfb24c8086b59bacacb798eccf00abbabd5d293 Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Tue, 14 Jul 2026 14:10:35 -0700 Subject: [PATCH 1/7] cuda.core: defer CUDA user-object payload reclamation Move graph attachment destruction off CUDA's internal callback thread through a coalesced Py_AddPendingCall drain. Queue payloads lock-free, retry scheduling failures from later safe entries, and leak intact payloads once Python finalization begins. --- cuda_core/cuda/core/_cpp/DESIGN.md | 14 ++ cuda_core/cuda/core/_cpp/resource_handles.cpp | 153 +++++++++++++++++- cuda_core/cuda/core/_cpp/resource_handles.hpp | 4 + cuda_core/cuda/core/_resource_handles.pyx | 2 + 4 files changed, 170 insertions(+), 3 deletions(-) diff --git a/cuda_core/cuda/core/_cpp/DESIGN.md b/cuda_core/cuda/core/_cpp/DESIGN.md index 713d64e00fe..a8b1ed5b1f4 100644 --- a/cuda_core/cuda/core/_cpp/DESIGN.md +++ b/cuda_core/cuda/core/_cpp/DESIGN.md @@ -217,6 +217,20 @@ Handle destructors may run from any thread. The implementation includes RAII gua The handle API functions are safe to call with or without the GIL held. They will release the GIL (if necessary) before calling CUDA driver API functions. +### CUDA User-Object Reclamation + +CUDA user-object destructors may run on an internal driver thread where CUDA +API calls are forbidden. Graph attachment payloads can release handles whose +deleters call CUDA or run Python finalizers, so their CUDA callback must not +delete the payload directly. + +The callback pushes a preallocated intrusive node onto a process-lifetime +lock-free queue and schedules one coalesced `Py_AddPendingCall`. The pending +callback drains all queued payloads from Python's main thread. If scheduling +fails, payloads remain intact for a later retry; they are intentionally leaked +once interpreter finalization begins. A single pending call is shared by all +queued payloads because CPython's main-thread pending-call queue is bounded. + ### Static Initialization and Deadlock Hazards When writing C++ code that interacts with Python, a subtle deadlock can occur diff --git a/cuda_core/cuda/core/_cpp/resource_handles.cpp b/cuda_core/cuda/core/_cpp/resource_handles.cpp index b97b207d2a4..3a090747ae2 100644 --- a/cuda_core/cuda/core/_cpp/resource_handles.cpp +++ b/cuda_core/cuda/core/_cpp/resource_handles.cpp @@ -6,6 +6,7 @@ #include "resource_handles.hpp" #include +#include #include #include #include @@ -185,6 +186,145 @@ class GILAcquireGuard { } // namespace +// ============================================================================ +// CUDA user-object deferred reclamation +// +// CUDA may invoke a user-object destructor on an internal thread where CUDA +// calls are forbidden. Payload destruction can release resource handles whose +// deleters call CUDA, so the callback only transfers a preallocated intrusive +// node to this process-lifetime queue. One coalesced pending call drains all +// queued payloads from Python's main thread. +// ============================================================================ + +namespace { + +class DeferredReclaimer; + +using DeferredDestroyFn = void (*)(void*) noexcept; + +struct DeferredDelete { + std::atomic next{nullptr}; + void* payload; + DeferredDestroyFn destroy; + DeferredReclaimer* reclaimer; + + DeferredDelete(void* payload_, DeferredDestroyFn destroy_, + DeferredReclaimer* reclaimer_) noexcept + : payload(payload_), destroy(destroy_), reclaimer(reclaimer_) {} +}; + +class DeferredReclaimer { +public: + void retire(DeferredDelete* item) noexcept { + DeferredDelete* head = head_.load(std::memory_order_relaxed); + do { + item->next.store(head, std::memory_order_relaxed); + } while (!head_.compare_exchange_weak( + head, item, std::memory_order_release, std::memory_order_relaxed)); + try_schedule(); + } + + void stop() noexcept { + accepting_.store(false, std::memory_order_release); + } + + void retry() noexcept { + try_schedule(); + } + +private: + static int pending_call(void* arg) noexcept { + static_cast(arg)->drain(); + return 0; + } + + void try_schedule() noexcept { + if (!accepting_.load(std::memory_order_acquire) || + !head_.load(std::memory_order_acquire)) { + return; + } + bool expected = false; + if (!scheduled_.compare_exchange_strong( + expected, true, std::memory_order_acq_rel, + std::memory_order_relaxed)) { + return; + } + if (Py_AddPendingCall(&DeferredReclaimer::pending_call, this) != 0) { + // Keep every payload queued. A later retirement or safe cuda-core + // entry can retry without blocking CUDA's callback thread. + scheduled_.store(false, std::memory_order_release); + } + } + + void drain() noexcept { + if (!Py_IsInitialized() || py_is_finalizing()) { + stop(); + scheduled_.store(false, std::memory_order_release); + return; // Intentionally leak intact payloads during shutdown. + } + + while (DeferredDelete* list = + head_.exchange(nullptr, std::memory_order_acquire)) { + while (list) { + DeferredDelete* next = + list->next.load(std::memory_order_relaxed); + list->destroy(list->payload); + delete list; + list = next; + } + } + + scheduled_.store(false, std::memory_order_release); + if (head_.load(std::memory_order_acquire)) { + try_schedule(); + } + } + + std::atomic head_{nullptr}; + std::atomic scheduled_{false}; + std::atomic accepting_{true}; +}; + +DeferredReclaimer* deferred_reclaimer = nullptr; // Intentionally never freed. + +void stop_deferred_reclaimer() { + if (deferred_reclaimer) { + deferred_reclaimer->stop(); + } +} + +DeferredDelete* make_deferred_delete( + void* payload, DeferredDestroyFn destroy) { + DeferredReclaimer* reclaimer = deferred_reclaimer; + if (!reclaimer) { + throw std::runtime_error("deferred reclaimer is not initialized"); + } + reclaimer->retry(); + return new DeferredDelete(payload, destroy, reclaimer); +} + +void destroy_deferred_delete_now(DeferredDelete* item) noexcept { + item->destroy(item->payload); + delete item; +} + +void enqueue_deferred_delete(void* item) noexcept { + auto* deferred = static_cast(item); + deferred->reclaimer->retire(deferred); +} + +} // namespace + +void initialize_deferred_reclaimer() { + if (deferred_reclaimer) { + return; + } + deferred_reclaimer = new DeferredReclaimer(); + if (Py_AtExit(stop_deferred_reclaimer) != 0) { + deferred_reclaimer->stop(); + } +} + // ============================================================================ // Handle reverse-lookup registry // @@ -1176,13 +1316,20 @@ GraphSlotTable* ensure_slot_table(const GraphBox* box) { return nullptr; } auto* table = new GraphSlotTable(); + DeferredDelete* deferred = nullptr; + try { + deferred = make_deferred_delete(table, destroy_graph_slot_table); + } catch (...) { + delete table; + throw; + } CUuserObject user_obj = nullptr; { GILReleaseGuard gil; - if (p_cuUserObjectCreate(&user_obj, table, - reinterpret_cast(destroy_graph_slot_table), + if (p_cuUserObjectCreate(&user_obj, deferred, + reinterpret_cast(enqueue_deferred_delete), 1, CU_USER_OBJECT_NO_DESTRUCTOR_SYNC) != CUDA_SUCCESS) { - delete table; // no user object created; nothing else owns the table + destroy_deferred_delete_now(deferred); return nullptr; } if (p_cuGraphRetainUserObject(box->resource, user_obj, 1, diff --git a/cuda_core/cuda/core/_cpp/resource_handles.hpp b/cuda_core/cuda/core/_cpp/resource_handles.hpp index e8f4e83ecaf..a8dc72c2922 100644 --- a/cuda_core/cuda/core/_cpp/resource_handles.hpp +++ b/cuda_core/cuda/core/_cpp/resource_handles.hpp @@ -256,6 +256,10 @@ StreamHandle create_stream_handle_with_owner(CUstream stream, PyObject* owner); // If Python is finalized or finalizing, the object is intentionally leaked. void py_object_user_object_destroy(void* py_object) noexcept; +// Initialize the process-lifetime CUDA user-object reclamation queue. Called +// once from module initialization while Python is fully initialized. +void initialize_deferred_reclaimer(); + // Return the context dependency associated with a stream handle, if any. ContextHandle get_stream_context(const StreamHandle& h) noexcept; diff --git a/cuda_core/cuda/core/_resource_handles.pyx b/cuda_core/cuda/core/_resource_handles.pyx index 9867f5cdcb4..b0a8c04c66a 100644 --- a/cuda_core/cuda/core/_resource_handles.pyx +++ b/cuda_core/cuda/core/_resource_handles.pyx @@ -61,6 +61,7 @@ cdef extern from "_cpp/resource_handles.hpp" namespace "cuda_core": cydriver.CUstream stream, object owner) except+ nogil void py_object_user_object_destroy "cuda_core::py_object_user_object_destroy" ( void* py_object) noexcept nogil + void initialize_deferred_reclaimer "cuda_core::initialize_deferred_reclaimer" () except+ ContextHandle get_stream_context "cuda_core::get_stream_context" ( const StreamHandle& h) noexcept nogil StreamHandle get_legacy_stream "cuda_core::get_legacy_stream" () except+ nogil @@ -461,6 +462,7 @@ cdef void _init_driver_fn_pointers() noexcept: p_cuDevSmResourceSplit = _get_optional_driver_fn("cuDevSmResourceSplit") _init_driver_fn_pointers() +initialize_deferred_reclaimer() # ============================================================================= # NVRTC function pointer initialization From 78fcbb87c987bc395663848a0f945a25ebde3c0f Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Tue, 14 Jul 2026 14:11:28 -0700 Subject: [PATCH 2/7] test(cuda.core): stress deferred graph reclamation Verify more than 32 CUDA callbacks coalesce onto Python's main thread, recover after Py_AddPendingCall queue saturation, and remain safe during interpreter shutdown. --- .../graph/test_graph_definition_lifetime.py | 121 ++++++++++++++++-- 1 file changed, 113 insertions(+), 8 deletions(-) diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index f8e1b030e4e..957d5a00247 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -5,6 +5,10 @@ import ctypes import gc +import subprocess +import sys +import textwrap +import threading import time import weakref @@ -15,10 +19,9 @@ from conftest import xfail_on_graph_mempool_oom from cuda_python_test_helpers import under_compute_sanitizer -# Resource finalization triggered by graph destruction is not strictly -# synchronous: the graph's slot table is freed through a CUDA user-object -# destructor that the driver may run on its own thread, after which each owner -# is released (a shared_ptr decrement, or Py_DECREF under the GIL). Release is +# Resource finalization triggered by graph destruction is not synchronous. A +# CUDA user-object callback transfers the slot table to a pending-call +# reclaimer, which releases each owner from Python's main thread. Release is # deterministic at the reference-count level, so the predicate normally flips # within milliseconds; this budget only bounds a slow/loaded runner. It stays a # hard failure rather than a warning so a real leak still fails the suite. @@ -35,13 +38,23 @@ class _Sentinel: """ +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, timeout=None, interval=0.02): """Poll ``predicate()`` until true, or raise AssertionError on timeout. - Each iteration drives ``gc.collect()`` and yields the main thread (which - releases the GIL) so the driver's asynchronous user-object destructor -- - and the ``Py_DECREF`` it triggers -- can make progress. Used for resource - cleanup that lags graph destruction; see ``_FINALIZE_TIMEOUT``. + Each iteration drives ``gc.collect()`` and reaches bytecode boundaries so + pending reclamation can run. Used for resource cleanup that lags graph + destruction; see ``_FINALIZE_TIMEOUT``. """ if timeout is None: timeout = _FINALIZE_TIMEOUT @@ -352,6 +365,98 @@ def test_event_survives_graph_clone_and_execution(init_cuda): # ============================================================================= +@pytest.mark.agent_authored(model="gpt-5.6") +def test_user_object_reclamation_is_coalesced_on_python_thread(init_cuda): + """More than 32 CUDA callbacks drain through one main-thread pending call.""" + finalized_threads = [] + main_thread = threading.get_ident() + graphs = [] + + for _ in range(64): + callback = _ThreadRecordingCallback(finalized_threads) + graph = GraphDefinition() + graph.callback(callback) + graphs.append(graph) + + del callback, graph, graphs + _wait_until(lambda: len(finalized_threads) == 64) + assert set(finalized_threads) == {main_thread} + + +@pytest.mark.agent_authored(model="gpt-5.6") +def test_pending_call_scheduling_failure_retries_later(init_cuda): + """A full CPython queue delays reclamation until a later safe retry.""" + 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 + + @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] + assert finalized_threads == [] + + # Preparing another graph attachment is a safe cuda-core entry point that + # retries scheduling the first graph's still-intact queued payload. + 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} + + +@pytest.mark.agent_authored(model="gpt-5.6") +def test_pending_reclamation_is_safe_during_python_shutdown(init_cuda): + """Outstanding graph attachments neither call Python nor hang at shutdown.""" + code = textwrap.dedent( + """ + from cuda.core import Device + from cuda.core.graph import GraphDefinition + + class Callback: + def __call__(self): + pass + + Device() + graph = GraphDefinition() + graph.callback(Callback()) + """ + ) + result = subprocess.run( # noqa: S603 - controlled interpreter probe + [sys.executable, "-c", code], + capture_output=True, + text=True, + timeout=20, + ) + assert result.returncode == 0, result.stderr + + def test_python_callable_callback_survives_del(init_cuda): """Python callable is kept alive by the graph after Python ref is dropped.""" called = [False] From f454e13ae64394a07f91a4bb5fe5ebaa67e8b6ac Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Tue, 14 Jul 2026 15:41:57 -0700 Subject: [PATCH 3/7] docs(cuda.core): clarify deferred reclaimer roles Add concise descriptions for each reclaimer operation and state field, and explain the typed synchronous helper versus CUDA's void-pointer callback ABI. --- cuda_core/cuda/core/_cpp/resource_handles.cpp | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/cuda_core/cuda/core/_cpp/resource_handles.cpp b/cuda_core/cuda/core/_cpp/resource_handles.cpp index 3a090747ae2..8d1bd4e5347 100644 --- a/cuda_core/cuda/core/_cpp/resource_handles.cpp +++ b/cuda_core/cuda/core/_cpp/resource_handles.cpp @@ -202,6 +202,7 @@ class DeferredReclaimer; using DeferredDestroyFn = void (*)(void*) noexcept; +// Preallocated envelope that transfers one payload from CUDA to the reclaimer. struct DeferredDelete { std::atomic next{nullptr}; void* payload; @@ -213,8 +214,10 @@ struct DeferredDelete { : payload(payload_), destroy(destroy_), reclaimer(reclaimer_) {} }; +// Process-lifetime MPSC queue that drains payloads from Python's main thread. class DeferredReclaimer { public: + // Transfer one preallocated payload envelope from a producer to the queue. void retire(DeferredDelete* item) noexcept { DeferredDelete* head = head_.load(std::memory_order_relaxed); do { @@ -224,20 +227,24 @@ class DeferredReclaimer { try_schedule(); } + // Permanently disable pending-call scheduling during interpreter shutdown. void stop() noexcept { accepting_.store(false, std::memory_order_release); } + // Reattempt scheduling for payloads left queued after an earlier failure. void retry() noexcept { try_schedule(); } private: + // Adapt the reclaimer drain to CPython's int (*)(void*) callback ABI. static int pending_call(void* arg) noexcept { static_cast(arg)->drain(); return 0; } + // Coalesce all queued work behind at most one CPython pending call. void try_schedule() noexcept { if (!accepting_.load(std::memory_order_acquire) || !head_.load(std::memory_order_acquire)) { @@ -256,6 +263,7 @@ class DeferredReclaimer { } } + // Detach and destroy all queued payloads from Python's main thread. void drain() noexcept { if (!Py_IsInitialized() || py_is_finalizing()) { stop(); @@ -280,8 +288,11 @@ class DeferredReclaimer { } } + // Head of the intrusive multi-producer, single-consumer payload stack. std::atomic head_{nullptr}; + // True while one cuda-core drain callback is pending or executing. std::atomic scheduled_{false}; + // False once shutdown begins, causing later payloads to be leaked safely. std::atomic accepting_{true}; }; @@ -303,11 +314,13 @@ DeferredDelete* make_deferred_delete( return new DeferredDelete(payload, destroy, reclaimer); } +// Destroy an envelope synchronously before CUDA has acquired ownership. void destroy_deferred_delete_now(DeferredDelete* item) noexcept { item->destroy(item->payload); delete item; } +// CUDA's CUhostFn ABI is void (*)(void*); recover the typed envelope and queue it. void enqueue_deferred_delete(void* item) noexcept { auto* deferred = static_cast(item); deferred->reclaimer->retire(deferred); From 6816b96866348e9e72aa3d8c6ea7f72d1d976925 Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Tue, 14 Jul 2026 16:02:32 -0700 Subject: [PATCH 4/7] refactor(cuda.core): simplify deferred queue envelopes Use the process-wide reclaimer directly instead of storing the same pointer in every envelope, publish the singleton atomically, and make the intrusive next link non-atomic under the queue's release/acquire handoff. --- cuda_core/cuda/core/_cpp/resource_handles.cpp | 38 ++++++++++--------- 1 file changed, 21 insertions(+), 17 deletions(-) diff --git a/cuda_core/cuda/core/_cpp/resource_handles.cpp b/cuda_core/cuda/core/_cpp/resource_handles.cpp index 8d1bd4e5347..4ba5330307b 100644 --- a/cuda_core/cuda/core/_cpp/resource_handles.cpp +++ b/cuda_core/cuda/core/_cpp/resource_handles.cpp @@ -204,14 +204,12 @@ using DeferredDestroyFn = void (*)(void*) noexcept; // Preallocated envelope that transfers one payload from CUDA to the reclaimer. struct DeferredDelete { - std::atomic next{nullptr}; + DeferredDelete* next = nullptr; void* payload; DeferredDestroyFn destroy; - DeferredReclaimer* reclaimer; - DeferredDelete(void* payload_, DeferredDestroyFn destroy_, - DeferredReclaimer* reclaimer_) noexcept - : payload(payload_), destroy(destroy_), reclaimer(reclaimer_) {} + DeferredDelete(void* payload_, DeferredDestroyFn destroy_) noexcept + : payload(payload_), destroy(destroy_) {} }; // Process-lifetime MPSC queue that drains payloads from Python's main thread. @@ -221,7 +219,7 @@ class DeferredReclaimer { void retire(DeferredDelete* item) noexcept { DeferredDelete* head = head_.load(std::memory_order_relaxed); do { - item->next.store(head, std::memory_order_relaxed); + item->next = head; } while (!head_.compare_exchange_weak( head, item, std::memory_order_release, std::memory_order_relaxed)); try_schedule(); @@ -274,8 +272,7 @@ class DeferredReclaimer { while (DeferredDelete* list = head_.exchange(nullptr, std::memory_order_acquire)) { while (list) { - DeferredDelete* next = - list->next.load(std::memory_order_relaxed); + DeferredDelete* next = list->next; list->destroy(list->payload); delete list; list = next; @@ -296,22 +293,25 @@ class DeferredReclaimer { std::atomic accepting_{true}; }; -DeferredReclaimer* deferred_reclaimer = nullptr; // Intentionally never freed. +// Published once at module initialization and intentionally never freed. +std::atomic deferred_reclaimer{nullptr}; void stop_deferred_reclaimer() { - if (deferred_reclaimer) { - deferred_reclaimer->stop(); + if (DeferredReclaimer* reclaimer = + deferred_reclaimer.load(std::memory_order_acquire)) { + reclaimer->stop(); } } DeferredDelete* make_deferred_delete( void* payload, DeferredDestroyFn destroy) { - DeferredReclaimer* reclaimer = deferred_reclaimer; + DeferredReclaimer* reclaimer = + deferred_reclaimer.load(std::memory_order_acquire); if (!reclaimer) { throw std::runtime_error("deferred reclaimer is not initialized"); } reclaimer->retry(); - return new DeferredDelete(payload, destroy, reclaimer); + return new DeferredDelete(payload, destroy); } // Destroy an envelope synchronously before CUDA has acquired ownership. @@ -323,18 +323,22 @@ void destroy_deferred_delete_now(DeferredDelete* item) noexcept { // CUDA's CUhostFn ABI is void (*)(void*); recover the typed envelope and queue it. void enqueue_deferred_delete(void* item) noexcept { auto* deferred = static_cast(item); - deferred->reclaimer->retire(deferred); + if (DeferredReclaimer* reclaimer = + deferred_reclaimer.load(std::memory_order_acquire)) { + reclaimer->retire(deferred); + } } } // namespace void initialize_deferred_reclaimer() { - if (deferred_reclaimer) { + if (deferred_reclaimer.load(std::memory_order_acquire)) { return; } - deferred_reclaimer = new DeferredReclaimer(); + auto* reclaimer = new DeferredReclaimer(); + deferred_reclaimer.store(reclaimer, std::memory_order_release); if (Py_AtExit(stop_deferred_reclaimer) != 0) { - deferred_reclaimer->stop(); + reclaimer->stop(); } } From d5a956c86e4769c748edd7ba63868f51fe3df54d Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Tue, 14 Jul 2026 16:16:01 -0700 Subject: [PATCH 5/7] refactor(cuda.core): unify deferred cleanup terminology Use payload, cleanup item, cleanup queue, enqueue, drain, and cleanup consistently across implementation, documentation, and tests. --- cuda_core/cuda/core/_cpp/DESIGN.md | 15 ++- cuda_core/cuda/core/_cpp/resource_handles.cpp | 122 +++++++++--------- cuda_core/cuda/core/_cpp/resource_handles.hpp | 6 +- cuda_core/cuda/core/_resource_handles.pyx | 4 +- .../graph/test_graph_definition_lifetime.py | 8 +- 5 files changed, 78 insertions(+), 77 deletions(-) diff --git a/cuda_core/cuda/core/_cpp/DESIGN.md b/cuda_core/cuda/core/_cpp/DESIGN.md index a8b1ed5b1f4..3aafe68e2fd 100644 --- a/cuda_core/cuda/core/_cpp/DESIGN.md +++ b/cuda_core/cuda/core/_cpp/DESIGN.md @@ -217,19 +217,20 @@ Handle destructors may run from any thread. The implementation includes RAII gua The handle API functions are safe to call with or without the GIL held. They will release the GIL (if necessary) before calling CUDA driver API functions. -### CUDA User-Object Reclamation +### CUDA User-Object Cleanup CUDA user-object destructors may run on an internal driver thread where CUDA API calls are forbidden. Graph attachment payloads can release handles whose deleters call CUDA or run Python finalizers, so their CUDA callback must not delete the payload directly. -The callback pushes a preallocated intrusive node onto a process-lifetime -lock-free queue and schedules one coalesced `Py_AddPendingCall`. The pending -callback drains all queued payloads from Python's main thread. If scheduling -fails, payloads remain intact for a later retry; they are intentionally leaked -once interpreter finalization begins. A single pending call is shared by all -queued payloads because CPython's main-thread pending-call queue is bounded. +The callback enqueues a preallocated `DeferredCleanupItem` on a +process-lifetime lock-free `DeferredCleanupQueue` and schedules one coalesced +`Py_AddPendingCall`. The pending callback drains all queued payloads from +Python's main thread. If scheduling fails, payloads remain intact for a later +retry; they are intentionally leaked once interpreter finalization begins. A +single pending call is shared by all queued payloads because CPython's +main-thread pending-call queue is bounded. ### Static Initialization and Deadlock Hazards diff --git a/cuda_core/cuda/core/_cpp/resource_handles.cpp b/cuda_core/cuda/core/_cpp/resource_handles.cpp index 4ba5330307b..56e20b9572c 100644 --- a/cuda_core/cuda/core/_cpp/resource_handles.cpp +++ b/cuda_core/cuda/core/_cpp/resource_handles.cpp @@ -187,10 +187,10 @@ class GILAcquireGuard { } // namespace // ============================================================================ -// CUDA user-object deferred reclamation +// CUDA user-object deferred cleanup // // CUDA may invoke a user-object destructor on an internal thread where CUDA -// calls are forbidden. Payload destruction can release resource handles whose +// calls are forbidden. Payload cleanup can release resource handles whose // deleters call CUDA, so the callback only transfers a preallocated intrusive // node to this process-lifetime queue. One coalesced pending call drains all // queued payloads from Python's main thread. @@ -198,31 +198,31 @@ class GILAcquireGuard { namespace { -class DeferredReclaimer; +class DeferredCleanupQueue; -using DeferredDestroyFn = void (*)(void*) noexcept; +using CleanupFn = void (*)(void*) noexcept; -// Preallocated envelope that transfers one payload from CUDA to the reclaimer. -struct DeferredDelete { - DeferredDelete* next = nullptr; +// Preallocated queue item that transfers one payload out of CUDA's callback. +struct DeferredCleanupItem { + DeferredCleanupItem* next = nullptr; void* payload; - DeferredDestroyFn destroy; + CleanupFn cleanup; - DeferredDelete(void* payload_, DeferredDestroyFn destroy_) noexcept - : payload(payload_), destroy(destroy_) {} + DeferredCleanupItem(void* payload_, CleanupFn cleanup_) noexcept + : payload(payload_), cleanup(cleanup_) {} }; // Process-lifetime MPSC queue that drains payloads from Python's main thread. -class DeferredReclaimer { +class DeferredCleanupQueue { public: - // Transfer one preallocated payload envelope from a producer to the queue. - void retire(DeferredDelete* item) noexcept { - DeferredDelete* head = head_.load(std::memory_order_relaxed); + // Transfer one preallocated cleanup item from a producer to the queue. + void enqueue(DeferredCleanupItem* item) noexcept { + DeferredCleanupItem* head = head_.load(std::memory_order_relaxed); do { item->next = head; } while (!head_.compare_exchange_weak( head, item, std::memory_order_release, std::memory_order_relaxed)); - try_schedule(); + schedule(); } // Permanently disable pending-call scheduling during interpreter shutdown. @@ -231,19 +231,19 @@ class DeferredReclaimer { } // Reattempt scheduling for payloads left queued after an earlier failure. - void retry() noexcept { - try_schedule(); + void retry_schedule() noexcept { + schedule(); } private: - // Adapt the reclaimer drain to CPython's int (*)(void*) callback ABI. + // Adapt queue draining to CPython's int (*)(void*) callback ABI. static int pending_call(void* arg) noexcept { - static_cast(arg)->drain(); + static_cast(arg)->drain(); return 0; } // Coalesce all queued work behind at most one CPython pending call. - void try_schedule() noexcept { + void schedule() noexcept { if (!accepting_.load(std::memory_order_acquire) || !head_.load(std::memory_order_acquire)) { return; @@ -254,8 +254,8 @@ class DeferredReclaimer { std::memory_order_relaxed)) { return; } - if (Py_AddPendingCall(&DeferredReclaimer::pending_call, this) != 0) { - // Keep every payload queued. A later retirement or safe cuda-core + if (Py_AddPendingCall(&DeferredCleanupQueue::pending_call, this) != 0) { + // Keep every payload queued. A later enqueue or safe cuda-core // entry can retry without blocking CUDA's callback thread. scheduled_.store(false, std::memory_order_release); } @@ -269,11 +269,11 @@ class DeferredReclaimer { return; // Intentionally leak intact payloads during shutdown. } - while (DeferredDelete* list = + while (DeferredCleanupItem* list = head_.exchange(nullptr, std::memory_order_acquire)) { while (list) { - DeferredDelete* next = list->next; - list->destroy(list->payload); + DeferredCleanupItem* next = list->next; + list->cleanup(list->payload); delete list; list = next; } @@ -281,12 +281,12 @@ class DeferredReclaimer { scheduled_.store(false, std::memory_order_release); if (head_.load(std::memory_order_acquire)) { - try_schedule(); + schedule(); } } // Head of the intrusive multi-producer, single-consumer payload stack. - std::atomic head_{nullptr}; + std::atomic head_{nullptr}; // True while one cuda-core drain callback is pending or executing. std::atomic scheduled_{false}; // False once shutdown begins, causing later payloads to be leaked safely. @@ -294,51 +294,51 @@ class DeferredReclaimer { }; // Published once at module initialization and intentionally never freed. -std::atomic deferred_reclaimer{nullptr}; +std::atomic deferred_cleanup_queue{nullptr}; -void stop_deferred_reclaimer() { - if (DeferredReclaimer* reclaimer = - deferred_reclaimer.load(std::memory_order_acquire)) { - reclaimer->stop(); +void stop_deferred_cleanup() { + if (DeferredCleanupQueue* queue = + deferred_cleanup_queue.load(std::memory_order_acquire)) { + queue->stop(); } } -DeferredDelete* make_deferred_delete( - void* payload, DeferredDestroyFn destroy) { - DeferredReclaimer* reclaimer = - deferred_reclaimer.load(std::memory_order_acquire); - if (!reclaimer) { - throw std::runtime_error("deferred reclaimer is not initialized"); +DeferredCleanupItem* make_cleanup_item( + void* payload, CleanupFn cleanup) { + DeferredCleanupQueue* queue = + deferred_cleanup_queue.load(std::memory_order_acquire); + if (!queue) { + throw std::runtime_error("deferred cleanup is not initialized"); } - reclaimer->retry(); - return new DeferredDelete(payload, destroy); + queue->retry_schedule(); + return new DeferredCleanupItem(payload, cleanup); } -// Destroy an envelope synchronously before CUDA has acquired ownership. -void destroy_deferred_delete_now(DeferredDelete* item) noexcept { - item->destroy(item->payload); +// Execute an item's cleanup synchronously before CUDA has acquired ownership. +void execute_cleanup_now(DeferredCleanupItem* item) noexcept { + item->cleanup(item->payload); delete item; } -// CUDA's CUhostFn ABI is void (*)(void*); recover the typed envelope and queue it. -void enqueue_deferred_delete(void* item) noexcept { - auto* deferred = static_cast(item); - if (DeferredReclaimer* reclaimer = - deferred_reclaimer.load(std::memory_order_acquire)) { - reclaimer->retire(deferred); +// CUDA's CUhostFn ABI is void (*)(void*); recover and enqueue the cleanup item. +void enqueue_cleanup(void* item) noexcept { + auto* cleanup = static_cast(item); + if (DeferredCleanupQueue* queue = + deferred_cleanup_queue.load(std::memory_order_acquire)) { + queue->enqueue(cleanup); } } } // namespace -void initialize_deferred_reclaimer() { - if (deferred_reclaimer.load(std::memory_order_acquire)) { +void initialize_deferred_cleanup() { + if (deferred_cleanup_queue.load(std::memory_order_acquire)) { return; } - auto* reclaimer = new DeferredReclaimer(); - deferred_reclaimer.store(reclaimer, std::memory_order_release); - if (Py_AtExit(stop_deferred_reclaimer) != 0) { - reclaimer->stop(); + auto* queue = new DeferredCleanupQueue(); + deferred_cleanup_queue.store(queue, std::memory_order_release); + if (Py_AtExit(stop_deferred_cleanup) != 0) { + queue->stop(); } } @@ -1303,7 +1303,7 @@ void free_deleter(const void* p) noexcept { std::free(const_cast(p)); } -void destroy_graph_slot_table(void* table) noexcept { +void cleanup_graph_slot_table(void* table) noexcept { delete static_cast(table); } @@ -1333,9 +1333,9 @@ GraphSlotTable* ensure_slot_table(const GraphBox* box) { return nullptr; } auto* table = new GraphSlotTable(); - DeferredDelete* deferred = nullptr; + DeferredCleanupItem* cleanup_item = nullptr; try { - deferred = make_deferred_delete(table, destroy_graph_slot_table); + cleanup_item = make_cleanup_item(table, cleanup_graph_slot_table); } catch (...) { delete table; throw; @@ -1343,10 +1343,10 @@ GraphSlotTable* ensure_slot_table(const GraphBox* box) { CUuserObject user_obj = nullptr; { GILReleaseGuard gil; - if (p_cuUserObjectCreate(&user_obj, deferred, - reinterpret_cast(enqueue_deferred_delete), + if (p_cuUserObjectCreate(&user_obj, cleanup_item, + reinterpret_cast(enqueue_cleanup), 1, CU_USER_OBJECT_NO_DESTRUCTOR_SYNC) != CUDA_SUCCESS) { - destroy_deferred_delete_now(deferred); + execute_cleanup_now(cleanup_item); return nullptr; } if (p_cuGraphRetainUserObject(box->resource, user_obj, 1, diff --git a/cuda_core/cuda/core/_cpp/resource_handles.hpp b/cuda_core/cuda/core/_cpp/resource_handles.hpp index a8dc72c2922..bb6bb3aa84f 100644 --- a/cuda_core/cuda/core/_cpp/resource_handles.hpp +++ b/cuda_core/cuda/core/_cpp/resource_handles.hpp @@ -256,9 +256,9 @@ StreamHandle create_stream_handle_with_owner(CUstream stream, PyObject* owner); // If Python is finalized or finalizing, the object is intentionally leaked. void py_object_user_object_destroy(void* py_object) noexcept; -// Initialize the process-lifetime CUDA user-object reclamation queue. Called -// once from module initialization while Python is fully initialized. -void initialize_deferred_reclaimer(); +// Initialize the process-lifetime CUDA user-object cleanup queue. Called once +// from module initialization while Python is fully initialized. +void initialize_deferred_cleanup(); // Return the context dependency associated with a stream handle, if any. ContextHandle get_stream_context(const StreamHandle& h) noexcept; diff --git a/cuda_core/cuda/core/_resource_handles.pyx b/cuda_core/cuda/core/_resource_handles.pyx index b0a8c04c66a..7407da45564 100644 --- a/cuda_core/cuda/core/_resource_handles.pyx +++ b/cuda_core/cuda/core/_resource_handles.pyx @@ -61,7 +61,7 @@ cdef extern from "_cpp/resource_handles.hpp" namespace "cuda_core": cydriver.CUstream stream, object owner) except+ nogil void py_object_user_object_destroy "cuda_core::py_object_user_object_destroy" ( void* py_object) noexcept nogil - void initialize_deferred_reclaimer "cuda_core::initialize_deferred_reclaimer" () except+ + void initialize_deferred_cleanup "cuda_core::initialize_deferred_cleanup" () except+ ContextHandle get_stream_context "cuda_core::get_stream_context" ( const StreamHandle& h) noexcept nogil StreamHandle get_legacy_stream "cuda_core::get_legacy_stream" () except+ nogil @@ -462,7 +462,7 @@ cdef void _init_driver_fn_pointers() noexcept: p_cuDevSmResourceSplit = _get_optional_driver_fn("cuDevSmResourceSplit") _init_driver_fn_pointers() -initialize_deferred_reclaimer() +initialize_deferred_cleanup() # ============================================================================= # NVRTC function pointer initialization diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index 957d5a00247..88c4164040c 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -21,7 +21,7 @@ # Resource finalization triggered by graph destruction is not synchronous. A # CUDA user-object callback transfers the slot table to a pending-call -# reclaimer, which releases each owner from Python's main thread. Release is +# cleanup queue, which releases each owner from Python's main thread. Release is # deterministic at the reference-count level, so the predicate normally flips # within milliseconds; this budget only bounds a slow/loaded runner. It stays a # hard failure rather than a warning so a real leak still fails the suite. @@ -53,7 +53,7 @@ def _wait_until(predicate, timeout=None, interval=0.02): """Poll ``predicate()`` until true, or raise AssertionError on timeout. Each iteration drives ``gc.collect()`` and reaches bytecode boundaries so - pending reclamation can run. Used for resource cleanup that lags graph + pending cleanup can run. Used for resource cleanup that lags graph destruction; see ``_FINALIZE_TIMEOUT``. """ if timeout is None: @@ -366,7 +366,7 @@ def test_event_survives_graph_clone_and_execution(init_cuda): @pytest.mark.agent_authored(model="gpt-5.6") -def test_user_object_reclamation_is_coalesced_on_python_thread(init_cuda): +def test_user_object_cleanup_is_coalesced_on_python_thread(init_cuda): """More than 32 CUDA callbacks drain through one main-thread pending call.""" finalized_threads = [] main_thread = threading.get_ident() @@ -432,7 +432,7 @@ def fill_queue_and_destroy(): @pytest.mark.agent_authored(model="gpt-5.6") -def test_pending_reclamation_is_safe_during_python_shutdown(init_cuda): +def test_pending_cleanup_is_safe_during_python_shutdown(init_cuda): """Outstanding graph attachments neither call Python nor hang at shutdown.""" code = textwrap.dedent( """ From d340969286436fea59ecdf5450227edb629e71b1 Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Tue, 14 Jul 2026 17:00:59 -0700 Subject: [PATCH 6/7] test(cuda.core): isolate shutdown subprocess Run the interpreter-shutdown probe outside the source tree so CI imports the installed package with generated modules. --- cuda_core/tests/graph/test_graph_definition_lifetime.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index 88c4164040c..3057f2b71ba 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -432,7 +432,7 @@ def fill_queue_and_destroy(): @pytest.mark.agent_authored(model="gpt-5.6") -def test_pending_cleanup_is_safe_during_python_shutdown(init_cuda): +def test_pending_cleanup_is_safe_during_python_shutdown(init_cuda, tmp_path): """Outstanding graph attachments neither call Python nor hang at shutdown.""" code = textwrap.dedent( """ @@ -453,6 +453,8 @@ def __call__(self): capture_output=True, text=True, timeout=20, + # Avoid shadowing the installed package with cuda_core/cuda/core. + cwd=tmp_path, ) assert result.returncode == 0, result.stderr From 8958da0f3ae2ec5d3272c023b22d335132e98cd6 Mon Sep 17 00:00:00 2001 From: Andy Jost Date: Tue, 14 Jul 2026 17:33:11 -0700 Subject: [PATCH 7/7] test(cuda.core): remove pending-call timing race Validate queue-saturation safety without assuming when CUDA invokes its asynchronous user-object destructor. --- .../tests/graph/test_graph_definition_lifetime.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index 3057f2b71ba..cc3872bb827 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -384,8 +384,8 @@ def test_user_object_cleanup_is_coalesced_on_python_thread(init_cuda): @pytest.mark.agent_authored(model="gpt-5.6") -def test_pending_call_scheduling_failure_retries_later(init_cuda): - """A full CPython queue delays reclamation until a later safe retry.""" +def test_pending_call_queue_saturation_preserves_cleanup(init_cuda): + """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] @@ -418,10 +418,10 @@ def fill_queue_and_destroy(): assert worker_done.wait(timeout=5) worker.join() assert queue_was_full == [True] - assert finalized_threads == [] - # Preparing another graph attachment is a safe cuda-core entry point that - # retries scheduling the first graph's still-intact queued payload. + # 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)