diff --git a/cuda_core/cuda/core/_cpp/DESIGN.md b/cuda_core/cuda/core/_cpp/DESIGN.md index 713d64e00fe..3aafe68e2fd 100644 --- a/cuda_core/cuda/core/_cpp/DESIGN.md +++ b/cuda_core/cuda/core/_cpp/DESIGN.md @@ -217,6 +217,21 @@ 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 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 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 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..56e20b9572c 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,162 @@ class GILAcquireGuard { } // namespace +// ============================================================================ +// CUDA user-object deferred cleanup +// +// CUDA may invoke a user-object destructor on an internal thread where CUDA +// 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. +// ============================================================================ + +namespace { + +class DeferredCleanupQueue; + +using CleanupFn = void (*)(void*) noexcept; + +// Preallocated queue item that transfers one payload out of CUDA's callback. +struct DeferredCleanupItem { + DeferredCleanupItem* next = nullptr; + void* payload; + CleanupFn cleanup; + + DeferredCleanupItem(void* payload_, CleanupFn cleanup_) noexcept + : payload(payload_), cleanup(cleanup_) {} +}; + +// Process-lifetime MPSC queue that drains payloads from Python's main thread. +class DeferredCleanupQueue { +public: + // 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)); + 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_schedule() noexcept { + schedule(); + } + +private: + // Adapt queue draining 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 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(&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); + } + } + + // Detach and destroy all queued payloads from Python's main thread. + 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 (DeferredCleanupItem* list = + head_.exchange(nullptr, std::memory_order_acquire)) { + while (list) { + DeferredCleanupItem* next = list->next; + list->cleanup(list->payload); + delete list; + list = next; + } + } + + scheduled_.store(false, std::memory_order_release); + if (head_.load(std::memory_order_acquire)) { + schedule(); + } + } + + // 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}; +}; + +// Published once at module initialization and intentionally never freed. +std::atomic deferred_cleanup_queue{nullptr}; + +void stop_deferred_cleanup() { + if (DeferredCleanupQueue* queue = + deferred_cleanup_queue.load(std::memory_order_acquire)) { + queue->stop(); + } +} + +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"); + } + queue->retry_schedule(); + return new DeferredCleanupItem(payload, cleanup); +} + +// 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 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_cleanup() { + if (deferred_cleanup_queue.load(std::memory_order_acquire)) { + return; + } + auto* queue = new DeferredCleanupQueue(); + deferred_cleanup_queue.store(queue, std::memory_order_release); + if (Py_AtExit(stop_deferred_cleanup) != 0) { + queue->stop(); + } +} + // ============================================================================ // Handle reverse-lookup registry // @@ -1146,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); } @@ -1176,13 +1333,20 @@ GraphSlotTable* ensure_slot_table(const GraphBox* box) { return nullptr; } auto* table = new GraphSlotTable(); + DeferredCleanupItem* cleanup_item = nullptr; + try { + cleanup_item = make_cleanup_item(table, cleanup_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, cleanup_item, + reinterpret_cast(enqueue_cleanup), 1, CU_USER_OBJECT_NO_DESTRUCTOR_SYNC) != CUDA_SUCCESS) { - delete table; // no user object created; nothing else owns the table + 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 e8f4e83ecaf..bb6bb3aa84f 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 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 9867f5cdcb4..7407da45564 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_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 @@ -461,6 +462,7 @@ cdef void _init_driver_fn_pointers() noexcept: p_cuDevSmResourceSplit = _get_optional_driver_fn("cuDevSmResourceSplit") _init_driver_fn_pointers() +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 f8e1b030e4e..cc3872bb827 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 +# 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. @@ -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 cleanup can run. Used for resource cleanup that lags graph + destruction; see ``_FINALIZE_TIMEOUT``. """ if timeout is None: timeout = _FINALIZE_TIMEOUT @@ -352,6 +365,100 @@ def test_event_survives_graph_clone_and_execution(init_cuda): # ============================================================================= +@pytest.mark.agent_authored(model="gpt-5.6") +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() + 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_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] + 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] + + # 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} + + +@pytest.mark.agent_authored(model="gpt-5.6") +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( + """ + 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, + # Avoid shadowing the installed package with cuda_core/cuda/core. + cwd=tmp_path, + ) + 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]