From 8b0e96dd97592b6e9285bd8f1d2525a35634203b Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 10 Aug 2026 11:17:48 -0700 Subject: [PATCH 1/7] Migrate host_tasks to a thread pool based approach Inspired by comment in: https://github.com/AdaptiveCpp/AdaptiveCpp/issues/1915 we move away from use of host_task, which enables compatibility with AdaptiveCpp --- dpctl/CMakeLists.txt | 8 +- ..._host_task_util.hpp => _async_dec_ref.hpp} | 34 +++-- dpctl/_sycl_queue.pyx | 18 +-- dpctl/apis/include/detail/keep_alive_pool.hpp | 125 ++++++++++++++++++ dpctl/apis/include/dpctl4pybind11.hpp | 49 +++---- 5 files changed, 177 insertions(+), 57 deletions(-) rename dpctl/{_host_task_util.hpp => _async_dec_ref.hpp} (74%) create mode 100644 dpctl/apis/include/detail/keep_alive_pool.hpp diff --git a/dpctl/CMakeLists.txt b/dpctl/CMakeLists.txt index a4378a03d0..5672445fdf 100644 --- a/dpctl/CMakeLists.txt +++ b/dpctl/CMakeLists.txt @@ -199,8 +199,12 @@ endforeach() set(_cy_file ${CMAKE_CURRENT_SOURCE_DIR}/_sycl_queue.pyx) get_filename_component(_trgt ${_cy_file} NAME_WLE) build_dpctl_ext(${_trgt} ${_cy_file} "dpctl" SYCL) -# _sycl_queue include _host_task_util.hpp -target_include_directories(${_trgt} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}) +# _sycl_queue includes _async_dec_ref.hpp, which includes +# detail/keep_alive_pool.hpp from the public include directory +target_include_directories(${_trgt} PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR} + ${CMAKE_CURRENT_SOURCE_DIR}/apis/include +) target_link_libraries(DpctlCAPI INTERFACE ${_trgt}_headers) add_subdirectory(compiler) diff --git a/dpctl/_host_task_util.hpp b/dpctl/_async_dec_ref.hpp similarity index 74% rename from dpctl/_host_task_util.hpp rename to dpctl/_async_dec_ref.hpp index 6898893bdd..decf1896b7 100644 --- a/dpctl/_host_task_util.hpp +++ b/dpctl/_async_dec_ref.hpp @@ -1,4 +1,4 @@ -//===--- _host_tasl_util.hpp - Implements async DECREF =// +//===--- _async_dec_ref.hpp - Implements async DECREF ---------------------===// // // Data Parallel Control (dpctl) // @@ -19,11 +19,11 @@ //===----------------------------------------------------------------------===// /// /// \file -/// This file implements a utility function to schedule host task to a sycl -/// queue depending on given array of sycl events to decrement reference counts -/// for the given array of Python objects. +/// This file implements a utility function to decrement reference counts for a +/// given array of Python objects once a given array of sycl events has +/// completed. /// -/// N.B.: The host task attempts to acquire GIL, so queue wait, event wait and +/// N.B.: The deferred work acquires the GIL, so queue wait, event wait and /// other synchronization mechanisms should be called after releasing the GIL to /// avoid deadlocks. /// @@ -33,9 +33,12 @@ #include #include #include +#include +#include #include "Python.h" +#include "detail/keep_alive_pool.hpp" #include "syclinterface/dpctl_data_types.h" #include "syclinterface/dpctl_sycl_type_casters.hpp" @@ -49,16 +52,21 @@ DPCTLSyclEventRef async_dec_ref(DPCTLSyclQueueRef QRef, using dpctl::syclinterface::unwrap; using dpctl::syclinterface::wrap; - sycl::queue *q = unwrap(QRef); + // `QRef` is kept in the signature for API compatibility + (void)QRef; std::vector obj_vec(obj_array, obj_array + obj_array_size); try { - sycl::event ht_ev = q->submit([&](sycl::handler &cgh) { - for (size_t ev_id = 0; ev_id < nDepERefs; ++ev_id) { - cgh.depends_on(*(unwrap(depERefs[ev_id]))); - } - cgh.host_task([obj_array_size, obj_vec]() { + std::vector depends; + depends.reserve(nDepERefs); + for (size_t ev_id = 0; ev_id < nDepERefs; ++ev_id) { + depends.push_back(*(unwrap(depERefs[ev_id]))); + } + + dpctl::detail::KeepAlivePool::get().submit( + std::move(depends), + [obj_array_size, obj_vec = std::move(obj_vec)]() { const bool initialized = Py_IsInitialized(); #if PY_VERSION_HEX < 0x30d0000 const bool finalizing = _Py_IsFinalizing(); @@ -75,12 +83,12 @@ DPCTLSyclEventRef async_dec_ref(DPCTLSyclQueueRef QRef, PyGILState_Release(gstate); } }); - }); static constexpr int result_ok = 0; *status = result_ok; - auto e_ptr = new sycl::event(ht_ev); + // return a dummy event for API compatibility + auto e_ptr = new sycl::event(); return wrap(e_ptr); } catch (const std::exception &e) { static constexpr int result_std_exception = 1; diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index 47bd7f2f67..b18f218d69 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -107,7 +107,7 @@ import struct import sys -cdef extern from "_host_task_util.hpp": +cdef extern from "_async_dec_ref.hpp": DPCTLSyclEventRef async_dec_ref( DPCTLSyclQueueRef, PyObject **, size_t, DPCTLSyclEventRef *, size_t, int * @@ -1357,16 +1357,18 @@ cdef class SyclQueue(_SyclQueue): working on Python objects collected in ``args``. Returns: dpctl.SyclEvent - The event associated with the submission of host task. + An already-complete event. No task is submitted to the queue, + so there is nothing to wait for here; ``events`` are what say + when the work reading ``args`` is done. - Increments reference count of ``args`` and schedules asynchronous - ``host_task`` to decrement the count once dependent events are + Increments reference count of ``args`` and schedules the matching + decrement to run on a background thread once dependent events are complete. .. note:: - The ``host_task`` attempts to acquire Python GIL, and it is - known to be unsafe during interpreter shutdown sequence. It is - thus strongly advised to ensure that all submitted ``host_task`` + The deferred decrement attempts to acquire Python GIL, which is + known to be unsafe during the interpreter shutdown sequence. It + is thus strongly advised to ensure that all dependent events complete before the end of the Python script. """ cdef size_t nDE = len(dEvents) @@ -1409,7 +1411,7 @@ cdef class SyclQueue(_SyclQueue): with nogil: DPCTLEvent_Wait(htERef) DPCTLEvent_Delete(htERef) - raise RuntimeError("Could not submit keep_args_alive host_task") + raise RuntimeError("Could not schedule keep_args_alive cleanup") return SyclEvent._create(htERef) diff --git a/dpctl/apis/include/detail/keep_alive_pool.hpp b/dpctl/apis/include/detail/keep_alive_pool.hpp new file mode 100644 index 0000000000..849d46c869 --- /dev/null +++ b/dpctl/apis/include/detail/keep_alive_pool.hpp @@ -0,0 +1,125 @@ +//===--- keep_alive_pool.hpp - keeps owners alive during offload ----------===// +// +// Data Parallel Control (dpctl) +// +// Copyright 2026 Intel Corporation +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +//===----------------------------------------------------------------------===// +/// +/// \file +/// A fixed-size pool of threads that wait on SYCL events and then run a +/// callable for maintaining Python object lifetime during offloaded tasks. +/// +//===----------------------------------------------------------------------===// + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace dpctl +{ +namespace detail +{ + +class KeepAlivePool +{ +public: + /*! + * @brief Number of waiter threads. + */ + static constexpr std::size_t num_threads = 4; + + static KeepAlivePool &get() + { + // deliberately leaked: workers are detached and hold a bare `this`, so + // the pool must outlive them + static KeepAlivePool *instance = new KeepAlivePool(); + return *instance; + } + + /*! + * @brief Run `task` once every event in `depends` has completed. + * + * `task` must own everything it releases -- move USM `shared_ptr` copies, + * `sycl::buffer` handles, or `PyObject *` references into it. It runs on a + * pool thread without the GIL, so it must take the GIL itself if it touches + * Python. + */ + void submit(std::vector depends, std::function task) + { + { + std::lock_guard lock(queue_mutex_); + tasks_.emplace(std::move(depends), std::move(task)); + } + condition_.notify_one(); + } + + KeepAlivePool(const KeepAlivePool &) = delete; + KeepAlivePool &operator=(const KeepAlivePool &) = delete; + ~KeepAlivePool() = delete; + +private: + KeepAlivePool() + { + for (std::size_t i = 0; i < num_threads; ++i) { + std::thread(&KeepAlivePool::run, this).detach(); + } + } + + void run() + { + for (;;) { + std::pair, std::function> item; + { + std::unique_lock lock(queue_mutex_); + condition_.wait(lock, [this] { return !tasks_.empty(); }); + item = std::move(tasks_.front()); + tasks_.pop(); + } + + try { + sycl::event::wait(item.first); + } catch (const std::exception &) { + // run the task anyway: an async error must not strand the + // task or else it may leak + } + + try { + item.second(); + } catch (const std::exception &) { + // a throwing task must not take down the worker or later + // tasks will be lost + } + } + } + + std::queue, std::function>> + tasks_; + std::mutex queue_mutex_; + std::condition_variable condition_; +}; + +} // namespace detail +} // namespace dpctl diff --git a/dpctl/apis/include/dpctl4pybind11.hpp b/dpctl/apis/include/dpctl4pybind11.hpp index ecfc85596f..9a6f009863 100644 --- a/dpctl/apis/include/dpctl4pybind11.hpp +++ b/dpctl/apis/include/dpctl4pybind11.hpp @@ -25,6 +25,7 @@ #pragma once +#include "detail/keep_alive_pool.hpp" #include "dpctl_capi.h" #include @@ -823,46 +824,26 @@ sycl::event keep_args_alive(sycl::queue &q, } } - bool use_depends = true; - sycl::event host_task_ev; - - if (n_usm_owners_held > 0) { - host_task_ev = q.submit([&](sycl::handler &cgh) { - if (use_depends) { - cgh.depends_on(depends); - use_depends = false; - } - else { - cgh.depends_on(host_task_ev); - } - cgh.host_task([shp_usm = std::move(shp_usm)]() { - // no body, but shared pointers are captured in - // the lambda, ensuring that USM allocation is - // kept alive - }); - }); - } + if (n_usm_owners_held > 0 || n_objects_held > 0) { + dpctl::detail::KeepAlivePool::get().submit( + depends, [n_usm_owners_held, shp_usm = std::move(shp_usm), + n_objects_held, shp_arr = std::move(shp_arr)]() mutable { + for (std::size_t i = 0; i < n_usm_owners_held; ++i) { + shp_usm[i].reset(); + } - if (n_objects_held > 0) { - host_task_ev = q.submit([&](sycl::handler &cgh) { - if (use_depends) { - cgh.depends_on(depends); - use_depends = false; - } - else { - cgh.depends_on(host_task_ev); - } - cgh.host_task([n_objects_held, shp_arr = std::move(shp_arr)]() { - py::gil_scoped_acquire acquire; + if (n_objects_held > 0) { + py::gil_scoped_acquire acquire; - for (std::size_t i = 0; i < n_objects_held; ++i) { - shp_arr[i]->dec_ref(); + for (std::size_t i = 0; i < n_objects_held; ++i) { + shp_arr[i]->dec_ref(); + } } }); - }); } - return host_task_ev; + // return dummy event for API compatibility + return sycl::event{}; } /*! @brief Check if all allocation queues are the same as the From 3390e984e1b75245a64846e41e191606bdfb0ce4 Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 10 Aug 2026 12:20:42 -0700 Subject: [PATCH 2/7] Deprecate APIs which require events from Python object management --- CHANGELOG.md | 6 + .../doc_sources/api_reference/dpctl/index.rst | 8 ++ .../doc_sources/api_reference/dpctl/utils.rst | 10 ++ dpctl/__init__.py | 2 + dpctl/_async_dec_ref.hpp | 54 +++++---- dpctl/_sycl_queue.pyx | 103 +++++++++++++++++- dpctl/_sycl_timer.py | 2 +- dpctl/tests/test_sycl_compiler.py | 6 +- dpctl/tests/test_sycl_kernel_submit.py | 3 +- dpctl/tests/test_sycl_queue.py | 24 ++++ dpctl/tests/test_sycl_timer.py | 18 ++- dpctl/tests/test_utils.py | 29 ++++- dpctl/utils/_order_manager.py | 31 ++++++ examples/python/using_order_manager.py | 26 ++--- 14 files changed, 262 insertions(+), 60 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d2230894a..3a57d4573f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,9 +15,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * Added `DPCTLQueue_MemsetWithEvents` C-API function to support `dpctl.SyclQueue.memset_async` [gh-2361](https://github.com/IntelPython/dpctl/pull/2361) * Added `dpctl.SyclQueue.fill` and `dpctl.SyclQueue.fill_async` methods [gh-2365](https://github.com/IntelPython/dpctl/pull/2365) * Added `DPCTLQueue_Fill8/16/32/64/128WithEvents` C-API functions to support `dpctl.SyclQueue.fill_async` [gh-2365](https://github.com/IntelPython/dpctl/pull/2365) +* Added `dpctl.keep_args_alive` free function, and `add_event` method to the order manager + +### Deprecated +* Deprecated `dpctl.SyclQueue._submit_keep_args_alive` in favor of `dpctl.keep_args_alive` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* Deprecated the order manager's `add_event_pair`, `host_task_events` and `num_host_task_events`, as `host_task` is no longer used for managing object lifetimes [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) ### Changed * Bump minimum NumPy version to 1.26 [gh-2192](https://github.com/IntelPython/dpctl/pull/2192) +* Implemented a thread pool to manage object lifetime during offload rather than use `host_task` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) * Rewrote USM Python examples into a single example [gh-2292](https://github.com/IntelPython/dpctl/pull/2292) * Registered `DPCTL_PARTITION_AFFINITY_DOMAIN_UNKNOWN` enumerator when `DPCTLDevice_GetPartitionAffinityDomains` receives an unrecognized value from the SYCL runtime [gh-2324](https://github.com/IntelPython/dpctl/pull/2324) diff --git a/docs/doc_sources/api_reference/dpctl/index.rst b/docs/doc_sources/api_reference/dpctl/index.rst index aa84cb12b9..4fe6344c76 100644 --- a/docs/doc_sources/api_reference/dpctl/index.rst +++ b/docs/doc_sources/api_reference/dpctl/index.rst @@ -88,6 +88,14 @@ SyclQueueCreationError SyclSubDeviceCreationError +.. rubric:: Lifetime management + +.. autosummary:: + :toctree: generated + :nosignatures: + + keep_args_alive + .. rubric:: Utilities .. autosummary:: diff --git a/docs/doc_sources/api_reference/dpctl/utils.rst b/docs/doc_sources/api_reference/dpctl/utils.rst index 093f298ff6..57ac0b8635 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -15,3 +15,13 @@ Thread-local object mapping each :class:`dpctl.SyclQueue` to an order manager, used to ensure sequential ordering of offloaded tasks. + + Record submitted tasks with ``add_event`` and use ``submitted_events`` + as the dependency list of subsequent submissions. To keep Python objects + referenced by a task alive until it completes, use + :func:`dpctl.keep_args_alive`. + + .. deprecated:: + ``add_event_pair``, ``host_task_events`` and ``num_host_task_events`` + are deprecated. Tasks are no longer paired with a host task event, + so ``add_event`` takes the computational event alone. diff --git a/dpctl/__init__.py b/dpctl/__init__.py index 11c4a3281f..2e45479daa 100644 --- a/dpctl/__init__.py +++ b/dpctl/__init__.py @@ -57,6 +57,7 @@ SyclQueue, SyclQueueCreationError, WorkGroupMemory, + keep_args_alive, ) from ._sycl_queue_manager import get_device_cached_queue from ._sycl_timer import SyclTimer @@ -114,6 +115,7 @@ "WorkGroupMemory", "LocalAccessor", "RawKernelArg", + "keep_args_alive", ] __all__ += [ "get_device_cached_queue", diff --git a/dpctl/_async_dec_ref.hpp b/dpctl/_async_dec_ref.hpp index decf1896b7..b6499a13df 100644 --- a/dpctl/_async_dec_ref.hpp +++ b/dpctl/_async_dec_ref.hpp @@ -42,18 +42,18 @@ #include "syclinterface/dpctl_data_types.h" #include "syclinterface/dpctl_sycl_type_casters.hpp" -DPCTLSyclEventRef async_dec_ref(DPCTLSyclQueueRef QRef, - PyObject **obj_array, - size_t obj_array_size, - DPCTLSyclEventRef *depERefs, - size_t nDepERefs, - int *status) +/*! + * @brief Schedule DECREFs of `obj_array` for once `depERefs` have completed. + * + * Sets `*status` to 0 on success and 1 if scheduling threw. + */ +void async_dec_ref(PyObject **obj_array, + size_t obj_array_size, + DPCTLSyclEventRef *depERefs, + size_t nDepERefs, + int *status) { using dpctl::syclinterface::unwrap; - using dpctl::syclinterface::wrap; - - // `QRef` is kept in the signature for API compatibility - (void)QRef; std::vector obj_vec(obj_array, obj_array + obj_array_size); @@ -85,20 +85,36 @@ DPCTLSyclEventRef async_dec_ref(DPCTLSyclQueueRef QRef, }); static constexpr int result_ok = 0; - *status = result_ok; - // return a dummy event for API compatibility - auto e_ptr = new sycl::event(); - return wrap(e_ptr); } catch (const std::exception &e) { static constexpr int result_std_exception = 1; - *status = result_std_exception; - return nullptr; } +} - static constexpr int result_other_abnormal = 2; +/*! + * @brief Event-returning form of `async_dec_ref`. + * + * Returns a default-constructed event. + * Returns nullptr on failure, with `*status` set. + */ +DPCTLSyclEventRef async_dec_ref_event(DPCTLSyclQueueRef QRef, + PyObject **obj_array, + size_t obj_array_size, + DPCTLSyclEventRef *depERefs, + size_t nDepERefs, + int *status) +{ + using dpctl::syclinterface::wrap; + + (void)QRef; + + async_dec_ref(obj_array, obj_array_size, depERefs, nDepERefs, status); + + if (*status != 0) { + return nullptr; + } - *status = result_other_abnormal; - return nullptr; + auto e_ptr = new sycl::event(); + return wrap(e_ptr); } diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index b18f218d69..14dc60575e 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -97,7 +97,7 @@ from cpython.buffer cimport ( PyObject_CheckBuffer, PyObject_GetBuffer, ) -from cpython.ref cimport Py_INCREF, PyObject +from cpython.ref cimport Py_DECREF, Py_INCREF, PyObject from libc.stdint cimport uint8_t, uint16_t, uint32_t, uint64_t from libc.stdlib cimport free, malloc @@ -105,16 +105,22 @@ import collections.abc import logging import struct import sys +import warnings cdef extern from "_async_dec_ref.hpp": - DPCTLSyclEventRef async_dec_ref( + void async_dec_ref( + PyObject **, size_t, DPCTLSyclEventRef *, size_t, int * + ) nogil + # deprecated, retained for the queue-bound _submit_keep_args_alive + DPCTLSyclEventRef async_dec_ref_event( DPCTLSyclQueueRef, PyObject **, size_t, DPCTLSyclEventRef *, size_t, int * ) nogil __all__ = [ + "keep_args_alive", "SyclQueue", "SyclKernelInvalidRangeError", "SyclKernelSubmitError", @@ -1347,6 +1353,9 @@ cdef class SyclQueue(_SyclQueue): Keeps objects in ``args`` alive until tasks associated with events complete. + Deprecated since dpctl 0.23.0. Use :func:`dpctl.keep_args_alive` + instead, which is not bound to a queue and returns nothing. + Args: args(object): Python object to keep alive. @@ -1371,6 +1380,14 @@ cdef class SyclQueue(_SyclQueue): is thus strongly advised to ensure that all dependent events complete before the end of the Python script. """ + warnings.warn( + "dpctl.SyclQueue._submit_keep_args_alive is deprecated and will " + "be removed in a future release. Use dpctl.keep_args_alive " + "instead, which is not bound to a queue and returns nothing.", + DeprecationWarning, + stacklevel=2, + ) + cdef size_t nDE = len(dEvents) cdef DPCTLSyclEventRef *depEvents = NULL cdef PyObject *args_raw = NULL @@ -1400,7 +1417,7 @@ cdef class SyclQueue(_SyclQueue): # schedule decrement args_raw = args - htERef = async_dec_ref( + htERef = async_dec_ref_event( self.get_queue_ref(), &args_raw, 1, depEvents, nDE, &status @@ -1411,7 +1428,7 @@ cdef class SyclQueue(_SyclQueue): with nogil: DPCTLEvent_Wait(htERef) DPCTLEvent_Delete(htERef) - raise RuntimeError("Could not schedule keep_args_alive cleanup") + raise RuntimeError("Could not schedule keep_args_alive") return SyclEvent._create(htERef) @@ -1453,7 +1470,7 @@ cdef class SyclQueue(_SyclQueue): as unified address space pointers. One way of accomplishing this is to use - :meth:`dpctl.SyclQueue._submit_keep_args_alive`. + :func:`dpctl.keep_args_alive`. """ cdef void **kargs = NULL cdef _arg_data_type *kargty = NULL @@ -2456,3 +2473,79 @@ cdef class RawKernelArg: as a ``size_t``. """ return self._arg_ref + + +def keep_args_alive(args, depends): + """keep_args_alive(args, depends) + + Keep objects in ``args`` alive until the tasks associated with + ``depends`` complete. + + Args: + args (object): + Python object to keep alive, typically a tuple of the arguments + passed to an offloaded task. + depends (List[dpctl.SyclEvent]): + Gating events. The objects in ``args`` are released once every + event in ``depends`` has completed. + + Returns: + None + + Increments the reference count of ``args`` and schedules the matching + decrement to run on a background thread once every event in ``depends`` + is complete. The reference is guaranteed to be held for the whole span + in between, so the objects cannot be collected while offloaded tasks are + still reading them. + + This function is not bound to a queue: the gating events fully determine + when the objects may be released. + + :Example: + .. code-block:: python + + import dpctl + + q = dpctl.SyclQueue() + e = q.submit_async(kernel, [x_usm], [n]) + dpctl.keep_args_alive((x_usm,), [e]) + + .. note:: + The deferred decrement attempts to acquire the Python GIL, which is + known to be unsafe during the interpreter shutdown sequence. It is + thus strongly advised to ensure that all events in ``depends`` + complete before the end of the Python script. + """ + cdef size_t nDE = len(depends) + cdef DPCTLSyclEventRef *depEvents = NULL + cdef PyObject *args_raw = NULL + cdef int status = -1 + + if nDE > 0: + depEvents = ( + malloc(nDE*sizeof(DPCTLSyclEventRef)) + ) + if not depEvents: + raise MemoryError() + for idx, de in enumerate(depends): + if isinstance(de, SyclEvent): + depEvents[idx] = (de).get_event_ref() + else: + free(depEvents) + raise TypeError( + "A sequence of dpctl.SyclEvent is expected" + ) + + # increment reference counts to list of arguments + Py_INCREF(args) + args_raw = args + + # schedule decrement + async_dec_ref(&args_raw, 1, depEvents, nDE, &status) + + free(depEvents) + if status != 0: + # the deferred decrement was never scheduled, so undo the increment + # here rather than leak the reference + Py_DECREF(args) + raise RuntimeError("Could not schedule keep_args_alive") diff --git a/dpctl/_sycl_timer.py b/dpctl/_sycl_timer.py index 90f688a637..6f97f70075 100644 --- a/dpctl/_sycl_timer.py +++ b/dpctl/_sycl_timer.py @@ -78,7 +78,7 @@ def get_event(self): ev = self._submit_empty_task_fn( sycl_queue=self.queue, depends=self._order_manager.submitted_events ) - self._order_manager.add_event_pair(ev, ev) + self._order_manager.add_event(ev) return ev diff --git a/dpctl/tests/test_sycl_compiler.py b/dpctl/tests/test_sycl_compiler.py index d91f83a395..70f139827a 100644 --- a/dpctl/tests/test_sycl_compiler.py +++ b/dpctl/tests/test_sycl_compiler.py @@ -305,10 +305,9 @@ def test_create_kernel_bundle_with_spec_const(): e2 = q.submit(kernel, [x_usm, y_usm], [n], dEvents=[e1]) e3 = q.memcpy_async(y, y_usm, y.nbytes, [e2]) - ht_e = q._submit_keep_args_alive([x_usm], [e3]) + dpctl.keep_args_alive([x_usm], [e3]) e3.wait() - ht_e.wait() assert np.all(y == 43) @@ -345,10 +344,9 @@ def test_create_kernel_bundle_with_composite_spec_const(): e2 = q.submit(kernel, [x_usm, y_usm], [n], dEvents=[e1]) e3 = q.memcpy_async(y, y_usm, y.nbytes, [e2]) - ht_e = q._submit_keep_args_alive([x_usm], [e3]) + dpctl.keep_args_alive([x_usm], [e3]) e3.wait() - ht_e.wait() # 1.0 * 10 + 2.5 = 12.5 assert np.all(y == 12.5) diff --git a/dpctl/tests/test_sycl_kernel_submit.py b/dpctl/tests/test_sycl_kernel_submit.py index 654b988a19..e20013158c 100644 --- a/dpctl/tests/test_sycl_kernel_submit.py +++ b/dpctl/tests/test_sycl_kernel_submit.py @@ -244,7 +244,7 @@ def test_submit_async(): e3_st = e3.execution_status e2_st = e2.execution_status e1_st = e1.execution_status - ht_e = q._submit_keep_args_alive([x_usm], [e1, e2, e3]) + dpctl.keep_args_alive([x_usm], [e1, e2, e3]) are_complete = [ e == status_complete for e in ( @@ -254,7 +254,6 @@ def test_submit_async(): ) ] e3.wait() - ht_e.wait() if not all(are_complete): async_detected = True break diff --git a/dpctl/tests/test_sycl_queue.py b/dpctl/tests/test_sycl_queue.py index af0fe71dcd..08e7e6882d 100644 --- a/dpctl/tests/test_sycl_queue.py +++ b/dpctl/tests/test_sycl_queue.py @@ -22,6 +22,7 @@ import pytest import dpctl +import dpctl.memory from .helper import create_invalid_capsule @@ -402,3 +403,26 @@ def test_cython_api(dpctl_cython_extension): except dpctl.SyclDeviceCreationError: pytest.skip("Default-construction of SyclDevice failed") assert q.sycl_device == d + + +def test_keep_args_alive_validates_events(): + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + + usm = dpctl.memory.MemoryUSMDevice(4096, queue=q) + with pytest.raises(TypeError): + dpctl.keep_args_alive((usm,), [None]) + + +def test_submit_keep_args_alive_deprecated(): + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + + usm = dpctl.memory.MemoryUSMDevice(4096, queue=q) + with pytest.warns(DeprecationWarning): + ht_ev = q._submit_keep_args_alive((usm,), []) + ht_ev.wait() diff --git a/dpctl/tests/test_sycl_timer.py b/dpctl/tests/test_sycl_timer.py index 9a99b1f997..5ab89e1d82 100644 --- a/dpctl/tests/test_sycl_timer.py +++ b/dpctl/tests/test_sycl_timer.py @@ -86,19 +86,18 @@ def test_sycl_timer_order_manager(profiling_queue): count=x.nbytes, dEvents=om.submitted_events, ) - ht1 = q._submit_keep_args_alive((x_usm, x), [e1]) - om.add_event_pair(ht1, e1) + dpctl.keep_args_alive((x_usm, x), [e1]) + om.add_event(e1) e2 = q.memcpy_async( dest=res, src=x_usm, count=res.nbytes, dEvents=om.submitted_events, ) - ht2 = q._submit_keep_args_alive((res, x_usm), [e2]) - om.add_event_pair(ht2, e2) + dpctl.keep_args_alive((res, x_usm), [e2]) + om.add_event(e2) e2.wait() - ht2.wait() host_dt, device_dt = timer.dt assert np.all(res == x) @@ -131,18 +130,17 @@ def test_sycl_timer_accumulation(profiling_queue): count=x.nbytes, dEvents=depends, ) - ht1 = q._submit_keep_args_alive((x_usm, x), [e1]) - om.add_event_pair(ht1, e1) + dpctl.keep_args_alive((x_usm, x), [e1]) + om.add_event(e1) e2 = q.memcpy_async( dest=res, src=x_usm, count=res.nbytes, dEvents=[e1], ) - ht2 = q._submit_keep_args_alive((res, x_usm), [e2]) - om.add_event_pair(ht2, e2) + dpctl.keep_args_alive((res, x_usm), [e2]) + om.add_event(e2) e2.wait() - ht2.wait() assert np.all(res == x) dev_dt = timer.dt.device_dt diff --git a/dpctl/tests/test_utils.py b/dpctl/tests/test_utils.py index 03298c3f22..314336b97b 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -73,13 +73,10 @@ def test_order_manager(): pytest.skip("Queue could not be created for default-selected device") _som = dpctl.utils.SequentialOrderManager _mngr = _som[q] - assert isinstance(_mngr.num_host_task_events, int) assert isinstance(_mngr.num_submitted_events, int) assert isinstance(_mngr.submitted_events, list) - assert isinstance(_mngr.host_task_events, list) - _mngr.add_event_pair(dpctl.SyclEvent(), dpctl.SyclEvent()) - _mngr.add_event_pair([dpctl.SyclEvent()], dpctl.SyclEvent()) - _mngr.add_event_pair(dpctl.SyclEvent(), [dpctl.SyclEvent()]) + _mngr.add_event(dpctl.SyclEvent()) + _mngr.add_event([dpctl.SyclEvent(), dpctl.SyclEvent()]) _mngr.wait() cpy = _mngr.__copy__() _som.clear() @@ -92,3 +89,25 @@ def test_order_manager(): _passed = True finally: assert _passed + + +def test_order_manager_deprecated_host_task_api(): + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + + with pytest.warns(DeprecationWarning): + assert isinstance(_mngr.num_host_task_events, int) + with pytest.warns(DeprecationWarning): + assert isinstance(_mngr.host_task_events, list) + with pytest.warns(DeprecationWarning): + _mngr.add_event_pair(dpctl.SyclEvent(), dpctl.SyclEvent()) + with pytest.warns(DeprecationWarning): + _mngr.add_event_pair([dpctl.SyclEvent()], dpctl.SyclEvent()) + with pytest.warns(DeprecationWarning): + _mngr.add_event_pair(dpctl.SyclEvent(), [dpctl.SyclEvent()]) + _mngr.wait() + _som.clear() diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index 7c66d6bcc8..ed18cfaea8 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -1,5 +1,6 @@ import sys import threading +import warnings import weakref from collections import defaultdict @@ -25,6 +26,14 @@ def __del__(self): SyclEvent.wait_for(_local.get_host_task_events()) def add_event_pair(self, host_task_ev, comp_ev): + warnings.warn( + "add_event_pair is deprecated and will be removed in a future " + "release. dpctl no longer submits host tasks, so there is no " + "separate host task event to track. Use add_event(comp_ev) " + "instead.", + DeprecationWarning, + stacklevel=2, + ) _local = self._state if isinstance(host_task_ev, SyclEvent) and isinstance( comp_ev, SyclEvent @@ -37,8 +46,24 @@ def add_event_pair(self, host_task_ev, comp_ev): comp_ev = (comp_ev,) _local.add_vector_to_both_events(host_task_ev, comp_ev) + def add_event(self, comp_ev): + _local = self._state + if isinstance(comp_ev, SyclEvent): + _local.add_to_submitted_events(comp_ev) + else: + if not isinstance(comp_ev, (list, tuple)): + comp_ev = (comp_ev,) + for ev in comp_ev: + _local.add_to_submitted_events(ev) + @property def num_host_task_events(self): + warnings.warn( + "num_host_task_events is deprecated and will be removed in a " + "future release. dpctl no longer submits host tasks.", + DeprecationWarning, + stacklevel=2, + ) _local = self._state return _local.get_num_host_task_events() @@ -49,6 +74,12 @@ def num_submitted_events(self): @property def host_task_events(self): + warnings.warn( + "host_task_events is deprecated and will be removed in a future " + "release. dpctl no longer submits host tasks.", + DeprecationWarning, + stacklevel=2, + ) _local = self._state return _local.get_host_task_events() diff --git a/examples/python/using_order_manager.py b/examples/python/using_order_manager.py index 9fdb1d2841..3ab8f307f9 100644 --- a/examples/python/using_order_manager.py +++ b/examples/python/using_order_manager.py @@ -30,7 +30,7 @@ def _memset_async(q, usm_buf, fill_byte, om): """ Fill ``usm_buf`` with ``fill_byte`` asynchronously and track in ``om``. - ``_submit_keep_args_alive`` prevents the buffer and the target from being + ``dpctl.keep_args_alive`` prevents the buffer and the target from being garbage-collected while the device is still reading/writing. """ n = usm_buf.nbytes @@ -38,8 +38,8 @@ def _memset_async(q, usm_buf, fill_byte, om): comp_ev = q.memcpy_async(usm_buf, data, n, dEvents=om.submitted_events) # keep Python objects alive until the copy finishes - ht_ev = q._submit_keep_args_alive((usm_buf, data), [comp_ev]) - om.add_event_pair(ht_ev, comp_ev) + dpctl.keep_args_alive((usm_buf, data), [comp_ev]) + om.add_event(comp_ev) return comp_ev @@ -101,8 +101,8 @@ def child_fill(thread_id): chunk, dEvents=child_om.submitted_events, ) - ht_ev = q._submit_keep_args_alive((usm_chunk, usm_data), [comp_ev]) - child_om.add_event_pair(ht_ev, comp_ev) + dpctl.keep_args_alive((usm_chunk, usm_data), [comp_ev]) + child_om.add_event(comp_ev) child_om.wait() return usm_chunk @@ -120,8 +120,8 @@ def child_fill(thread_id): comp_ev = q.memcpy_async( part, child_buf, chunk, dEvents=main_om.submitted_events ) - ht_ev = q._submit_keep_args_alive((part, child_buf), [comp_ev]) - main_om.add_event_pair(ht_ev, comp_ev) + dpctl.keep_args_alive((part, child_buf), [comp_ev]) + main_om.add_event(comp_ev) result_parts.append(part) main_om.wait() @@ -154,7 +154,7 @@ def child_prepare(thread_id): _memset_async(q, buf, fill_val, child_om) - return buf, child_om.host_task_events, child_om.submitted_events + return buf, child_om.submitted_events with concurrent.futures.ThreadPoolExecutor( max_workers=n_threads @@ -162,15 +162,13 @@ def child_prepare(thread_id): futures_results = list(executor.map(child_prepare, range(n_threads))) child_buffers = [] - collected_ht_events = [] collected_comp_events = [] - for buf, ht_events, comp_events in futures_results: + for buf, comp_events in futures_results: child_buffers.append(buf) - collected_ht_events.extend(ht_events) collected_comp_events.extend(comp_events) main_om = SequentialOrderManager[q] - main_om.add_event_pair(collected_ht_events, collected_comp_events) + main_om.add_event(collected_comp_events) results = [] for buf in child_buffers: @@ -178,8 +176,8 @@ def child_prepare(thread_id): comp_ev = q.memcpy_async( out, buf, nbytes, dEvents=main_om.submitted_events ) - ht_ev = q._submit_keep_args_alive((out, buf), [comp_ev]) - main_om.add_event_pair(ht_ev, comp_ev) + dpctl.keep_args_alive((out, buf), [comp_ev]) + main_om.add_event(comp_ev) results.append(out) main_om.wait() From c9c82a621b3504d4a37a865b30b8af8d8ca8594e Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Thu, 13 Aug 2026 13:34:43 -0700 Subject: [PATCH 3/7] address review comments --- dpctl/_sycl_queue.pyx | 12 ++++-------- dpctl/apis/include/dpctl4pybind11.hpp | 3 +++ dpctl/utils/_order_manager.py | 1 + 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index 14dc60575e..4823cac57d 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -1353,8 +1353,9 @@ cdef class SyclQueue(_SyclQueue): Keeps objects in ``args`` alive until tasks associated with events complete. - Deprecated since dpctl 0.23.0. Use :func:`dpctl.keep_args_alive` - instead, which is not bound to a queue and returns nothing. + Deprecated. Use :func:`dpctl.keep_args_alive` instead, which is not + bound to a queue and returns nothing. The event returned by this + function is already complete. Args: args(object): @@ -2494,12 +2495,7 @@ def keep_args_alive(args, depends): Increments the reference count of ``args`` and schedules the matching decrement to run on a background thread once every event in ``depends`` - is complete. The reference is guaranteed to be held for the whole span - in between, so the objects cannot be collected while offloaded tasks are - still reading them. - - This function is not bound to a queue: the gating events fully determine - when the objects may be released. + is complete. :Example: .. code-block:: python diff --git a/dpctl/apis/include/dpctl4pybind11.hpp b/dpctl/apis/include/dpctl4pybind11.hpp index 9a6f009863..050419c119 100644 --- a/dpctl/apis/include/dpctl4pybind11.hpp +++ b/dpctl/apis/include/dpctl4pybind11.hpp @@ -803,6 +803,9 @@ sycl::event keep_args_alive(sycl::queue &q, const py::object (&py_objs)[num], const std::vector &depends = {}) { + // q is only retained for API compatibility + (void)q; + std::size_t n_objects_held = 0; std::array, num> shp_arr{}; diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index ed18cfaea8..aef60970d4 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -23,6 +23,7 @@ def __del__(self): return _local = self._state SyclEvent.wait_for(_local.get_submitted_events()) + # TODO: remove once deprecated add_event_pair is removed SyclEvent.wait_for(_local.get_host_task_events()) def add_event_pair(self, host_task_ev, comp_ev): From 637c49084f72c2818f2b853f01280b78cbdea389 Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Tue, 18 Aug 2026 17:28:07 -0700 Subject: [PATCH 4/7] Apply review feedback Also change pool getter to avoid any dpctl4pybind11 including extensions from having their own thread pools --- CHANGELOG.md | 3 +- CMakeLists.txt | 13 +++++ .../doc_sources/api_reference/dpctl/utils.rst | 2 +- .../beginners_guides/installation.rst | 18 +++++++ dpctl/CMakeLists.txt | 4 ++ dpctl/_async_dec_ref.hpp | 20 ++++++-- dpctl/_sycl_queue.pyx | 5 ++ dpctl/apis/include/detail/keep_alive_pool.hpp | 14 +++++- dpctl/apis/include/dpctl4pybind11.hpp | 50 +++++++++++++++---- scripts/_build_helper.py | 5 ++ scripts/build_locally.py | 13 +++++ 11 files changed, 129 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a57d4573f..6636e4d65f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * Added `DPCTLQueue_MemsetWithEvents` C-API function to support `dpctl.SyclQueue.memset_async` [gh-2361](https://github.com/IntelPython/dpctl/pull/2361) * Added `dpctl.SyclQueue.fill` and `dpctl.SyclQueue.fill_async` methods [gh-2365](https://github.com/IntelPython/dpctl/pull/2365) * Added `DPCTLQueue_Fill8/16/32/64/128WithEvents` C-API functions to support `dpctl.SyclQueue.fill_async` [gh-2365](https://github.com/IntelPython/dpctl/pull/2365) -* Added `dpctl.keep_args_alive` free function, and `add_event` method to the order manager +* Added `dpctl.keep_args_alive` free function, and `add_event` method to the order manager [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* Added `DPCTL_KEEP_ALIVE_POOL_SIZE` CMake option for setting the number of threads that keep objects alive during offload [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) ### Deprecated * Deprecated `dpctl.SyclQueue._submit_keep_args_alive` in favor of `dpctl.keep_args_alive` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) diff --git a/CMakeLists.txt b/CMakeLists.txt index 0d27c9d6c3..b6a4ddc319 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -48,6 +48,19 @@ option( size of shared object with offloading sections" OFF ) +set(DPCTL_KEEP_ALIVE_POOL_SIZE + "4" + CACHE STRING + "Number of threads in the pool that keeps Python objects used by \ +offloaded tasks alive until the tasks complete" +) + +if (NOT DPCTL_KEEP_ALIVE_POOL_SIZE MATCHES "^[1-9][0-9]*$") + message(FATAL_ERROR + "Invalid value for DPCTL_KEEP_ALIVE_POOL_SIZE: \ +\"${DPCTL_KEEP_ALIVE_POOL_SIZE}\". Expected a positive integer." + ) +endif() find_package(IntelSYCL REQUIRED PATHS ${CMAKE_SOURCE_DIR}/cmake NO_DEFAULT_PATH) diff --git a/docs/doc_sources/api_reference/dpctl/utils.rst b/docs/doc_sources/api_reference/dpctl/utils.rst index 57ac0b8635..117fbfdaf3 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -21,7 +21,7 @@ referenced by a task alive until it completes, use :func:`dpctl.keep_args_alive`. - .. deprecated:: + .. deprecated:: 0.23.0 ``add_event_pair``, ``host_task_events`` and ``num_host_task_events`` are deprecated. Tasks are no longer paired with a host task event, so ``add_event`` takes the computational event alone. diff --git a/docs/doc_sources/beginners_guides/installation.rst b/docs/doc_sources/beginners_guides/installation.rst index edfa1b4dae..dd50b25d3d 100644 --- a/docs/doc_sources/beginners_guides/installation.rst +++ b/docs/doc_sources/beginners_guides/installation.rst @@ -247,6 +247,24 @@ devices at the same time: python scripts/build_locally.py --verbose --target-cuda --target-hip=gfx1030 +Configuring the object management thread pool +--------------------------------------------- + +To keep Python objects used by an offloaded task alive until tasks complete, +:py:mod:`dpctl` maintains a pool of threads that wait on the task's events. +The pool uses four threads by default. The size is fixed when ``dpctl`` is +compiled and can be changed with the ``--keep-alive-pool-size`` argument: + +.. code-block:: bash + + python scripts/build_locally.py --verbose --keep-alive-pool-size=8 + +Alternatively, you use the ``DPCTL_KEEP_ALIVE_POOL_SIZE`` CMake option: + +.. code-block:: bash + + python scripts/build_locally.py --verbose --cmake-opts="-DDPCTL_KEEP_ALIVE_POOL_SIZE=8" + Running Examples and Tests ========================== diff --git a/dpctl/CMakeLists.txt b/dpctl/CMakeLists.txt index 5672445fdf..a99e72d757 100644 --- a/dpctl/CMakeLists.txt +++ b/dpctl/CMakeLists.txt @@ -205,6 +205,10 @@ target_include_directories(${_trgt} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_SOURCE_DIR}/apis/include ) +# _sycl_queue owns the pool, so only it needs to know the pool size +target_compile_definitions(${_trgt} PRIVATE + DPCTL_KEEP_ALIVE_POOL_SIZE=${DPCTL_KEEP_ALIVE_POOL_SIZE} +) target_link_libraries(DpctlCAPI INTERFACE ${_trgt}_headers) add_subdirectory(compiler) diff --git a/dpctl/_async_dec_ref.hpp b/dpctl/_async_dec_ref.hpp index b6499a13df..21176d0b1a 100644 --- a/dpctl/_async_dec_ref.hpp +++ b/dpctl/_async_dec_ref.hpp @@ -42,6 +42,21 @@ #include "syclinterface/dpctl_data_types.h" #include "syclinterface/dpctl_sycl_type_casters.hpp" +/*! + * @brief Address of the `KeepAlivePool`. + * + * Returns nullptr if the pool could not be created. + */ +void *keep_alive_pool_ptr() +{ + try { + return static_cast( + &dpctl::detail::KeepAlivePool::local_instance()); + } catch (const std::exception &e) { + return nullptr; + } +} + /*! * @brief Schedule DECREFs of `obj_array` for once `depERefs` have completed. * @@ -64,17 +79,16 @@ void async_dec_ref(PyObject **obj_array, depends.push_back(*(unwrap(depERefs[ev_id]))); } - dpctl::detail::KeepAlivePool::get().submit( + dpctl::detail::KeepAlivePool::local_instance().submit( std::move(depends), [obj_array_size, obj_vec = std::move(obj_vec)]() { - const bool initialized = Py_IsInitialized(); #if PY_VERSION_HEX < 0x30d0000 const bool finalizing = _Py_IsFinalizing(); #else const bool finalizing = Py_IsFinalizing(); #endif // if the main thread has not finalized the interpreter yet - if (initialized && !finalizing) { + if (!finalizing) { PyGILState_STATE gstate; gstate = PyGILState_Ensure(); for (size_t i = 0; i < obj_array_size; ++i) { diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index 4823cac57d..18fcd8cec4 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -117,6 +117,7 @@ cdef extern from "_async_dec_ref.hpp": DPCTLSyclQueueRef, PyObject **, size_t, DPCTLSyclEventRef *, size_t, int * ) nogil + void *keep_alive_pool_ptr() nogil __all__ = [ @@ -2281,6 +2282,10 @@ cdef api SyclQueue SyclQueue_Make(DPCTLSyclQueueRef QRef): cdef DPCTLSyclQueueRef copied_QRef = DPCTLQueue_Copy(QRef) return SyclQueue._create(copied_QRef) + +cdef api void *KeepAlivePool_Get() noexcept nogil: + return keep_alive_pool_ptr() + cdef class _WorkGroupMemory: def __dealloc__(self): if(self._mem_ref): diff --git a/dpctl/apis/include/detail/keep_alive_pool.hpp b/dpctl/apis/include/detail/keep_alive_pool.hpp index 849d46c869..d0d958201a 100644 --- a/dpctl/apis/include/detail/keep_alive_pool.hpp +++ b/dpctl/apis/include/detail/keep_alive_pool.hpp @@ -38,6 +38,10 @@ #include +#ifndef DPCTL_KEEP_ALIVE_POOL_SIZE +#define DPCTL_KEEP_ALIVE_POOL_SIZE 4 +#endif + namespace dpctl { namespace detail @@ -49,9 +53,15 @@ class KeepAlivePool /*! * @brief Number of waiter threads. */ - static constexpr std::size_t num_threads = 4; + static constexpr std::size_t num_threads = DPCTL_KEEP_ALIVE_POOL_SIZE; + + static_assert(num_threads > 0, + "DPCTL_KEEP_ALIVE_POOL_SIZE must be greater than zero"); - static KeepAlivePool &get() + /*! + * @brief The instance belonging to this shared object. + */ + static KeepAlivePool &local_instance() { // deliberately leaked: workers are detached and hold a bare `this`, so // the pool must outlive them diff --git a/dpctl/apis/include/dpctl4pybind11.hpp b/dpctl/apis/include/dpctl4pybind11.hpp index 050419c119..a87dc264a9 100644 --- a/dpctl/apis/include/dpctl4pybind11.hpp +++ b/dpctl/apis/include/dpctl4pybind11.hpp @@ -47,6 +47,21 @@ namespace dpctl namespace detail { +/*! + * @brief Whether the interpreter can still be called into. + * + * Acquiring the GIL once finalization has begun does not return, so work + * deferred to a thread must check this before touching Python. + */ +inline bool interpreter_is_live() +{ +#if PY_VERSION_HEX < 0x30d0000 + return !_Py_IsFinalizing(); +#else + return !Py_IsFinalizing(); +#endif +} + class dpctl_capi { public: @@ -174,15 +189,7 @@ class dpctl_capi { void operator()(py::object *p) const { - const bool initialized = Py_IsInitialized(); -#if PY_VERSION_HEX < 0x30d0000 - const bool finalizing = _Py_IsFinalizing(); -#else - const bool finalizing = Py_IsFinalizing(); -#endif - const bool guard = initialized && !finalizing; - - if (guard) { + if (interpreter_is_live()) { delete p; } } @@ -295,6 +302,25 @@ class dpctl_capi dpctl_capi &operator=(dpctl_capi &&) = default; }; // struct dpctl_capi + +/*! + * @brief The `KeepAlivePool` singleton, owned by `dpctl._sycl_queue`. + */ +inline KeepAlivePool &get_keep_alive_pool() +{ + static KeepAlivePool *pool = []() -> KeepAlivePool * { + // get dpctl_capi to prevent nullptr return + static_cast(dpctl_capi::get()); + + return static_cast(KeepAlivePool_Get()); + }(); + + if (!pool) { + throw std::runtime_error("Could not create dpctl's keep-alive pool"); + } + return *pool; +} + } // namespace detail } // namespace dpctl @@ -828,14 +854,16 @@ sycl::event keep_args_alive(sycl::queue &q, } if (n_usm_owners_held > 0 || n_objects_held > 0) { - dpctl::detail::KeepAlivePool::get().submit( + dpctl::detail::get_keep_alive_pool().submit( depends, [n_usm_owners_held, shp_usm = std::move(shp_usm), n_objects_held, shp_arr = std::move(shp_arr)]() mutable { for (std::size_t i = 0; i < n_usm_owners_held; ++i) { shp_usm[i].reset(); } - if (n_objects_held > 0) { + // if the main thread has not finalized the interpreter yet + if (n_objects_held > 0 && + dpctl::detail::interpreter_is_live()) { py::gil_scoped_acquire acquire; for (std::size_t i = 0; i < n_objects_held; ++i) { diff --git a/scripts/_build_helper.py b/scripts/_build_helper.py index dfbc86f46d..086e6382ec 100644 --- a/scripts/_build_helper.py +++ b/scripts/_build_helper.py @@ -93,6 +93,7 @@ def make_cmake_args( glog: bool = False, verbose: bool = False, other_opts: str = None, + keep_alive_pool_size: int = None, ): args = [ f"-DCMAKE_C_COMPILER:PATH={c_compiler}" if c_compiler else "", @@ -102,6 +103,10 @@ def make_cmake_args( f"-DDPCTL_ENABLE_GLOG:BOOL={'ON' if glog else 'OFF'}", ] + if keep_alive_pool_size is not None: + args.append( + f"-DDPCTL_KEEP_ALIVE_POOL_SIZE:STRING={keep_alive_pool_size}" + ) if verbose: args.append("-DCMAKE_VERBOSE_MAKEFILE:BOOL=ON") if other_opts: diff --git a/scripts/build_locally.py b/scripts/build_locally.py index 6918a429e5..a399912ce6 100644 --- a/scripts/build_locally.py +++ b/scripts/build_locally.py @@ -97,6 +97,13 @@ def parse_args(): help="Disable Level Zero backend", ) + p.add_argument( + "--keep-alive-pool-size", + type=int, + default=None, + help="Number of threads used to keep objects alive during offload.", + ) + p.add_argument( "--cmake-opts", type=str, @@ -149,6 +156,11 @@ def main(): # Level Zero state (on unless explicitly disabled) level_zero_enabled = False if args.no_level_zero else True + if args.keep_alive_pool_size is not None and args.keep_alive_pool_size < 1: + err( + "--keep-alive-pool-size must be a positive integer", "build_locally" + ) + cmake_args = make_cmake_args( c_compiler=c_compiler, cxx_compiler=cxx_compiler, @@ -156,6 +168,7 @@ def main(): glog=args.glog, verbose=args.verbose, other_opts=args.cmake_opts, + keep_alive_pool_size=args.keep_alive_pool_size, ) # handle architecture conflicts From 94378f9bdb2cc32d02edfed01766eaf2fff94848 Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 14 Sep 2026 14:28:39 -0700 Subject: [PATCH 5/7] Use a polling thread for KeepAlivePool This drops the argument for setting the size of KeepAlivePool --- CHANGELOG.md | 7 +- CMakeLists.txt | 14 - .../doc_sources/api_reference/dpctl/utils.rst | 14 +- .../beginners_guides/installation.rst | 18 -- dpctl/CMakeLists.txt | 4 - dpctl/_async_dec_ref.hpp | 206 +++++++++++--- dpctl/_sycl_queue.pxd | 2 + dpctl/_sycl_queue.pyx | 63 ++-- dpctl/apis/include/detail/keep_alive_pool.hpp | 268 ++++++++++++++---- dpctl/apis/include/dpctl4pybind11.hpp | 114 ++++++-- dpctl/memory/_memory.pyx | 85 +++--- dpctl/tests/test_sycl_queue.py | 67 +++++ dpctl/tests/test_utils.py | 58 ++++ dpctl/utils/_order_manager.py | 52 +++- dpctl/utils/src/order_keeper.cpp | 8 +- dpctl/utils/src/sequential_order_keeper.hpp | 68 ++--- scripts/_build_helper.py | 5 - scripts/build_locally.py | 13 - 18 files changed, 789 insertions(+), 277 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6636e4d65f..04baf37c84 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,15 +16,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * Added `dpctl.SyclQueue.fill` and `dpctl.SyclQueue.fill_async` methods [gh-2365](https://github.com/IntelPython/dpctl/pull/2365) * Added `DPCTLQueue_Fill8/16/32/64/128WithEvents` C-API functions to support `dpctl.SyclQueue.fill_async` [gh-2365](https://github.com/IntelPython/dpctl/pull/2365) * Added `dpctl.keep_args_alive` free function, and `add_event` method to the order manager [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) -* Added `DPCTL_KEEP_ALIVE_POOL_SIZE` CMake option for setting the number of threads that keep objects alive during offload [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* Added `add_cleanup_event` method and `cleanup_events` and `num_cleanup_events` properties to the order manager, for tracking the events that gate the release of objects used by offloaded tasks [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) ### Deprecated * Deprecated `dpctl.SyclQueue._submit_keep_args_alive` in favor of `dpctl.keep_args_alive` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) -* Deprecated the order manager's `add_event_pair`, `host_task_events` and `num_host_task_events`, as `host_task` is no longer used for managing object lifetimes [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* Deprecated the order manager's `add_event_pair`, `host_task_events` and `num_host_task_events`, as `host_task` is no longer used for managing object lifetimes, in favor of `add_event`, `add_cleanup_event`, `cleanup_events` and `num_cleanup_events` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) ### Changed * Bump minimum NumPy version to 1.26 [gh-2192](https://github.com/IntelPython/dpctl/pull/2192) -* Implemented a thread pool to manage object lifetime during offload rather than use `host_task` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* Implemented a background thread that polls events to manage object lifetime during offload rather than use `host_task` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* The event returned by `dpctl::utils::keep_args_alive` and `dpctl.SyclQueue._submit_keep_args_alive` is now for an empty kernel that gates the deferred release rather than for a `host_task` that performs it, so its completion means that the objects are no longer in use rather than that they were released [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) * Rewrote USM Python examples into a single example [gh-2292](https://github.com/IntelPython/dpctl/pull/2292) * Registered `DPCTL_PARTITION_AFFINITY_DOMAIN_UNKNOWN` enumerator when `DPCTLDevice_GetPartitionAffinityDomains` receives an unrecognized value from the SYCL runtime [gh-2324](https://github.com/IntelPython/dpctl/pull/2324) diff --git a/CMakeLists.txt b/CMakeLists.txt index b6a4ddc319..a9f139e598 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -48,20 +48,6 @@ option( size of shared object with offloading sections" OFF ) -set(DPCTL_KEEP_ALIVE_POOL_SIZE - "4" - CACHE STRING - "Number of threads in the pool that keeps Python objects used by \ -offloaded tasks alive until the tasks complete" -) - -if (NOT DPCTL_KEEP_ALIVE_POOL_SIZE MATCHES "^[1-9][0-9]*$") - message(FATAL_ERROR - "Invalid value for DPCTL_KEEP_ALIVE_POOL_SIZE: \ -\"${DPCTL_KEEP_ALIVE_POOL_SIZE}\". Expected a positive integer." - ) -endif() - find_package(IntelSYCL REQUIRED PATHS ${CMAKE_SOURCE_DIR}/cmake NO_DEFAULT_PATH) set(_dpctl_sycl_target_compile_options) diff --git a/docs/doc_sources/api_reference/dpctl/utils.rst b/docs/doc_sources/api_reference/dpctl/utils.rst index 117fbfdaf3..cd0bea5866 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -21,7 +21,17 @@ referenced by a task alive until it completes, use :func:`dpctl.keep_args_alive`. + Record events that gate the release of objects used by a task with + ``add_cleanup_event``, and find them in ``cleanup_events``. They are + waited on, but never become dependencies of later tasks. + + Waiting with ``wait`` also drops the references that + :func:`dpctl.keep_args_alive` took for tasks that have since completed, + which is otherwise done the next time ``dpctl`` is called into. + .. deprecated:: 0.23.0 ``add_event_pair``, ``host_task_events`` and ``num_host_task_events`` - are deprecated. Tasks are no longer paired with a host task event, - so ``add_event`` takes the computational event alone. + are deprecated. Tasks are no longer paired with a host task event, so + ``add_event`` takes the computational event alone, and cleanup is + tracked by ``add_cleanup_event``, ``cleanup_events`` and + ``num_cleanup_events``. diff --git a/docs/doc_sources/beginners_guides/installation.rst b/docs/doc_sources/beginners_guides/installation.rst index dd50b25d3d..edfa1b4dae 100644 --- a/docs/doc_sources/beginners_guides/installation.rst +++ b/docs/doc_sources/beginners_guides/installation.rst @@ -247,24 +247,6 @@ devices at the same time: python scripts/build_locally.py --verbose --target-cuda --target-hip=gfx1030 -Configuring the object management thread pool ---------------------------------------------- - -To keep Python objects used by an offloaded task alive until tasks complete, -:py:mod:`dpctl` maintains a pool of threads that wait on the task's events. -The pool uses four threads by default. The size is fixed when ``dpctl`` is -compiled and can be changed with the ``--keep-alive-pool-size`` argument: - -.. code-block:: bash - - python scripts/build_locally.py --verbose --keep-alive-pool-size=8 - -Alternatively, you use the ``DPCTL_KEEP_ALIVE_POOL_SIZE`` CMake option: - -.. code-block:: bash - - python scripts/build_locally.py --verbose --cmake-opts="-DDPCTL_KEEP_ALIVE_POOL_SIZE=8" - Running Examples and Tests ========================== diff --git a/dpctl/CMakeLists.txt b/dpctl/CMakeLists.txt index a99e72d757..5672445fdf 100644 --- a/dpctl/CMakeLists.txt +++ b/dpctl/CMakeLists.txt @@ -205,10 +205,6 @@ target_include_directories(${_trgt} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_SOURCE_DIR}/apis/include ) -# _sycl_queue owns the pool, so only it needs to know the pool size -target_compile_definitions(${_trgt} PRIVATE - DPCTL_KEEP_ALIVE_POOL_SIZE=${DPCTL_KEEP_ALIVE_POOL_SIZE} -) target_link_libraries(DpctlCAPI INTERFACE ${_trgt}_headers) add_subdirectory(compiler) diff --git a/dpctl/_async_dec_ref.hpp b/dpctl/_async_dec_ref.hpp index 21176d0b1a..d12b5a63fe 100644 --- a/dpctl/_async_dec_ref.hpp +++ b/dpctl/_async_dec_ref.hpp @@ -2,7 +2,7 @@ // // Data Parallel Control (dpctl) // -// Copyright 2022 Intel Corporation +// Copyright 2026 Intel Corporation // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -23,15 +23,18 @@ /// given array of Python objects once a given array of sycl events has /// completed. /// -/// N.B.: The deferred work acquires the GIL, so queue wait, event wait and -/// other synchronization mechanisms should be called after releasing the GIL to -/// avoid deadlocks. +/// N.B.: The reference counts are dropped by whichever thread next enters +/// `dpctl` from Python, and never by the thread that finds the events complete, +/// which must not touch Python. Waiting for a decrement to happen therefore +/// deadlocks, and nothing in `dpctl` does. /// //===----------------------------------------------------------------------===// #pragma once -#include +#include +#include #include +#include #include #include #include @@ -42,6 +45,41 @@ #include "syclinterface/dpctl_data_types.h" #include "syclinterface/dpctl_sycl_type_casters.hpp" +namespace dpctl +{ +namespace detail +{ + +namespace +{ + +/*! + * @brief The pool, once there is one, for those who must not create it. + */ +std::atomic created_pool{nullptr}; + +} // namespace + +/*! + * @brief The pool of the `dpctl._sycl_queue` module, which everyone shares. + * + * `KeepAlivePool` befriends this, making it the only creator of a pool. + */ +KeepAlivePool &local_keep_alive_pool() +{ + // deliberately leaked: the polling thread is detached and holds a bare + // `this`, so the pool must outlive it + static KeepAlivePool *instance = []() { + KeepAlivePool *pool = new KeepAlivePool(); + created_pool.store(pool, std::memory_order_release); + return pool; + }(); + return *instance; +} + +} // namespace detail +} // namespace dpctl + /*! * @brief Address of the `KeepAlivePool`. * @@ -50,13 +88,86 @@ void *keep_alive_pool_ptr() { try { - return static_cast( - &dpctl::detail::KeepAlivePool::local_instance()); - } catch (const std::exception &e) { + return static_cast(&dpctl::detail::local_keep_alive_pool()); + } catch (...) { + // nothing may escape into the calling Cython code, which is not + // prepared to handle a C++ exception return nullptr; } } +/*! + * @brief Drop the references of the DECREFs that have come due. + * + * Returns whether there were any, so that a caller draining in the hope of + * reclaiming something knows whether anything was given up. + * + * Expects the caller to hold the GIL. + */ +bool drain_retired_references() +{ + // a pool being created right now is left to the next drain + dpctl::detail::KeepAlivePool *pool = + dpctl::detail::created_pool.load(std::memory_order_acquire); + if (!pool) { + return false; + } + + try { + return pool->drain_retired(); + } catch (...) { + // nothing may escape into the calling Cython code, which is not + // prepared to handle a C++ exception + return false; + } +} + +namespace +{ + +/*! + * @brief Copy `nDepERefs` event references into a vector. + */ +std::vector unwrap_events(DPCTLSyclEventRef *depERefs, + size_t nDepERefs) +{ + using dpctl::syclinterface::unwrap; + + std::vector depends; + depends.reserve(nDepERefs); + for (size_t ev_id = 0; ev_id < nDepERefs; ++ev_id) { + depends.push_back(*(unwrap(depERefs[ev_id]))); + } + + return depends; +} + +/*! + * @brief Schedule DECREFs of `obj_vec` for once `depends` have completed. + */ +void submit_dec_ref(std::vector obj_vec, + std::vector depends) +{ + auto &pool = dpctl::detail::local_keep_alive_pool(); + + // the caller holds the GIL, so this is an opportunity to drop the + // references of the DECREFs scheduled before this one + pool.drain_retired(); + + pool.submit(std::move(depends), [obj_vec = std::move(obj_vec)]() mutable { + // handed to a thread that holds the GIL rather than dropped here, as + // the thread running this must not touch Python + dpctl::detail::local_keep_alive_pool().retire( + [obj_vec = std::move(obj_vec)]() { + for (PyObject *obj : obj_vec) { + Py_DECREF(obj); + } + }); + }); +} + +} // namespace + /*! * @brief Schedule DECREFs of `obj_array` for once `depERefs` have completed. * @@ -68,48 +179,30 @@ void async_dec_ref(PyObject **obj_array, size_t nDepERefs, int *status) { - using dpctl::syclinterface::unwrap; - - std::vector obj_vec(obj_array, obj_array + obj_array_size); - try { - std::vector depends; - depends.reserve(nDepERefs); - for (size_t ev_id = 0; ev_id < nDepERefs; ++ev_id) { - depends.push_back(*(unwrap(depERefs[ev_id]))); - } - - dpctl::detail::KeepAlivePool::local_instance().submit( - std::move(depends), - [obj_array_size, obj_vec = std::move(obj_vec)]() { -#if PY_VERSION_HEX < 0x30d0000 - const bool finalizing = _Py_IsFinalizing(); -#else - const bool finalizing = Py_IsFinalizing(); -#endif - // if the main thread has not finalized the interpreter yet - if (!finalizing) { - PyGILState_STATE gstate; - gstate = PyGILState_Ensure(); - for (size_t i = 0; i < obj_array_size; ++i) { - Py_DECREF(obj_vec[i]); - } - PyGILState_Release(gstate); - } - }); + submit_dec_ref( + std::vector(obj_array, obj_array + obj_array_size), + unwrap_events(depERefs, nDepERefs)); static constexpr int result_ok = 0; *status = result_ok; - } catch (const std::exception &e) { - static constexpr int result_std_exception = 1; - *status = result_std_exception; + } catch (...) { + // nothing may escape into the calling Cython code, which is not + // prepared to handle a C++ exception + static constexpr int result_exception = 1; + *status = result_exception; } } /*! - * @brief Event-returning form of `async_dec_ref`. + * @brief Queue-bound form of `async_dec_ref`. + * + * Returns an event for an empty kernel submitted to `QRef` after `depERefs`, + * which is what the DECREFs wait for. It has completed once `obj_array` is no + * longer in use, and, if `QRef` is in-order, once the work already submitted + * to the queue has completed as well. The DECREFs themselves are dropped + * afterwards, by whichever thread next enters `dpctl` from Python. * - * Returns a default-constructed event. * Returns nullptr on failure, with `*status` set. */ DPCTLSyclEventRef async_dec_ref_event(DPCTLSyclQueueRef QRef, @@ -119,16 +212,35 @@ DPCTLSyclEventRef async_dec_ref_event(DPCTLSyclQueueRef QRef, size_t nDepERefs, int *status) { + using dpctl::syclinterface::unwrap; using dpctl::syclinterface::wrap; - (void)QRef; + try { + sycl::queue *q = unwrap(QRef); + if (!q) { + throw std::runtime_error("Queue reference is null"); + } + + const sycl::event marker = dpctl::detail::submit_keep_alive_marker( + *q, unwrap_events(depERefs, nDepERefs)); + + // allocated before scheduling, as failing afterwards could not be + // reported: the caller would drop a reference the scheduled DECREFs own + std::unique_ptr e_ptr(new sycl::event(marker)); - async_dec_ref(obj_array, obj_array_size, depERefs, nDepERefs, status); + submit_dec_ref( + std::vector(obj_array, obj_array + obj_array_size), + {marker}); - if (*status != 0) { + static constexpr int result_ok = 0; + *status = result_ok; + + return wrap(e_ptr.release()); + } catch (...) { + // nothing may escape into the calling Cython code, which is not + // prepared to handle a C++ exception + static constexpr int result_exception = 1; + *status = result_exception; return nullptr; } - - auto e_ptr = new sycl::event(); - return wrap(e_ptr); } diff --git a/dpctl/_sycl_queue.pxd b/dpctl/_sycl_queue.pxd index c72d09faed..260c047bc4 100644 --- a/dpctl/_sycl_queue.pxd +++ b/dpctl/_sycl_queue.pxd @@ -138,3 +138,5 @@ cdef public api class RawKernelArg(_RawKernelArg) [ object PyRawKernelArgObject, type PyRawKernelArgType ]: pass + +cdef bint drain_retired() diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index 18fcd8cec4..256fda803f 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -118,6 +118,7 @@ cdef extern from "_async_dec_ref.hpp": size_t, DPCTLSyclEventRef *, size_t, int * ) nogil void *keep_alive_pool_ptr() nogil + bint drain_retired_references() __all__ = [ @@ -1355,8 +1356,7 @@ cdef class SyclQueue(_SyclQueue): complete. Deprecated. Use :func:`dpctl.keep_args_alive` instead, which is not - bound to a queue and returns nothing. The event returned by this - function is already complete. + bound to a queue and returns nothing. Args: args(object): @@ -1368,19 +1368,24 @@ cdef class SyclQueue(_SyclQueue): working on Python objects collected in ``args``. Returns: dpctl.SyclEvent - An already-complete event. No task is submitted to the queue, - so there is nothing to wait for here; ``events`` are what say - when the work reading ``args`` is done. + An event for an empty kernel submitted to this queue after + ``events``. It says when ``args`` stop being used, not when + they were released, as the reference is dropped once the event + completes. Increments reference count of ``args`` and schedules the matching - decrement to run on a background thread once dependent events are - complete. + decrement for once the returned event is complete. If this queue is + in-order, the decrement is thus also ordered after the tasks already + submitted to it. The decrement runs on a thread that is running Python, + not on the background thread that finds the event complete, so it + happens the next time ``dpctl`` is entered from Python. .. note:: - The deferred decrement attempts to acquire Python GIL, which is - known to be unsafe during the interpreter shutdown sequence. It - is thus strongly advised to ensure that all dependent events - complete before the end of the Python script. + A decrement that never comes due before the interpreter shuts down + is not performed, leaking the reference rather than risking a + decrement the interpreter can no longer support. Ensure that the + dependent events complete before the end of the Python script to + have the references dropped. """ warnings.warn( "dpctl.SyclQueue._submit_keep_args_alive is deprecated and will " @@ -1427,9 +1432,7 @@ cdef class SyclQueue(_SyclQueue): free(depEvents) if (status != 0): - with nogil: - DPCTLEvent_Wait(htERef) - DPCTLEvent_Delete(htERef) + Py_DECREF(args) raise RuntimeError("Could not schedule keep_args_alive") return SyclEvent._create(htERef) @@ -2286,6 +2289,23 @@ cdef api SyclQueue SyclQueue_Make(DPCTLSyclQueueRef QRef): cdef api void *KeepAlivePool_Get() noexcept nogil: return keep_alive_pool_ptr() + +cdef bint drain_retired(): + """Drop the references of the releases that have come due. + + Returns whether there were any. Also Eexpects the caller to hold the GIL. + """ + return drain_retired_references() + + +def _drain_retired_references(): + """_drain_retired_references() + + Drop the references held for offloaded tasks that have completed. + """ + drain_retired() + + cdef class _WorkGroupMemory: def __dealloc__(self): if(self._mem_ref): @@ -2499,8 +2519,10 @@ def keep_args_alive(args, depends): None Increments the reference count of ``args`` and schedules the matching - decrement to run on a background thread once every event in ``depends`` - is complete. + decrement for once every event in ``depends`` is complete. The decrement + runs on a thread that is running Python, not on the background thread that + finds the events complete, so it happens the next time ``dpctl`` is entered + from Python. :Example: .. code-block:: python @@ -2512,10 +2534,11 @@ def keep_args_alive(args, depends): dpctl.keep_args_alive((x_usm,), [e]) .. note:: - The deferred decrement attempts to acquire the Python GIL, which is - known to be unsafe during the interpreter shutdown sequence. It is - thus strongly advised to ensure that all events in ``depends`` - complete before the end of the Python script. + A decrement that never comes due before the interpreter shuts down is + not performed, leaking the reference rather than risking a decrement + the interpreter can no longer support. Ensure that the events in + ``depends`` complete before the end of the Python script to have the + references dropped. """ cdef size_t nDE = len(depends) cdef DPCTLSyclEventRef *depEvents = NULL diff --git a/dpctl/apis/include/detail/keep_alive_pool.hpp b/dpctl/apis/include/detail/keep_alive_pool.hpp index d0d958201a..07a2526c29 100644 --- a/dpctl/apis/include/detail/keep_alive_pool.hpp +++ b/dpctl/apis/include/detail/keep_alive_pool.hpp @@ -19,71 +19,104 @@ //===----------------------------------------------------------------------===// /// /// \file -/// A fixed-size pool of threads that wait on SYCL events and then run a -/// callable for maintaining Python object lifetime during offloaded tasks. +/// A background thread that polls SYCL events and then runs a callable for +/// maintaining Python object lifetime during offloaded tasks, and a list of +/// releases deferred to a thread that can run Python code. /// //===----------------------------------------------------------------------===// #pragma once +#include +#include #include #include -#include #include #include -#include #include #include #include #include -#ifndef DPCTL_KEEP_ALIVE_POOL_SIZE -#define DPCTL_KEEP_ALIVE_POOL_SIZE 4 -#endif - namespace dpctl { namespace detail { +/*! + * @brief A thread that polls SYCL events and then runs a callable. + * + * A single thread polls the events of every task submitted to it, so that + * tasks run as soon as their own events complete, in whatever order that + * happens, and no thread ever blocks on offloaded work. Blocking would be + * costly as well as ordering: `sycl::event::wait` busy-waits on some backends, + * so a thread parked on a long-running task would burn a core to do nothing. + * + * The thread never runs Python code, so that it can never be waited for by a + * thread holding the GIL and can never be caught asking for the GIL as the + * interpreter finalizes. Work that needs the GIL goes to `retire` instead, and + * runs on a thread that already holds it. + * + * `dpctl` owns the single instance of the pool that every user of `dpctl` + * shares. Obtain it with `dpctl::detail::get_keep_alive_pool()`, declared in + * `dpctl4pybind11.hpp`, which is the only supported way of reaching it. + */ class KeepAlivePool { public: /*! - * @brief Number of waiter threads. + * @brief Run `task` once every event in `depends` has completed. + * + * It runs on the polling thread, which has to be kept moving, so `task` + * must not block on offloaded work, and must not touch Python. Pass work + * that needs the GIL to `retire` rather than doing it here. */ - static constexpr std::size_t num_threads = DPCTL_KEEP_ALIVE_POOL_SIZE; - - static_assert(num_threads > 0, - "DPCTL_KEEP_ALIVE_POOL_SIZE must be greater than zero"); + void submit(std::vector depends, std::function task) + { + { + std::lock_guard lock(submitted_mutex_); + submitted_.push_back(Item{std::move(depends), std::move(task)}); + } + condition_.notify_one(); + } /*! - * @brief The instance belonging to this shared object. + * @brief Hand `release` to whichever thread drains next. */ - static KeepAlivePool &local_instance() + void retire(std::function release) { - // deliberately leaked: workers are detached and hold a bare `this`, so - // the pool must outlive them - static KeepAlivePool *instance = new KeepAlivePool(); - return *instance; + std::lock_guard lock(retired_mutex_); + retired_.push_back(std::move(release)); } /*! - * @brief Run `task` once every event in `depends` has completed. + * @brief Run everything retired so far, on the calling thread. * - * `task` must own everything it releases -- move USM `shared_ptr` copies, - * `sycl::buffer` handles, or `PyObject *` references into it. It runs on a - * pool thread without the GIL, so it must take the GIL itself if it touches - * Python. + * Returns a bool representing whether there was anything to run. */ - void submit(std::vector depends, std::function task) + bool drain_retired() { + std::vector> to_run; { - std::lock_guard lock(queue_mutex_); - tasks_.emplace(std::move(depends), std::move(task)); + std::lock_guard lock(retired_mutex_); + // swapped out rather than run under the lock, as dropping the last + // reference to an object can run code that retires more work + to_run.swap(retired_); } - condition_.notify_one(); + + const bool ran_any = !to_run.empty(); + + for (auto &release : to_run) { + try { + release(); + } catch (...) { + // the rest must still run. `release` is caller-provided, so + // anything at all may come out of here. + } + } + + return ran_any; } KeepAlivePool(const KeepAlivePool &) = delete; @@ -91,45 +124,182 @@ class KeepAlivePool ~KeepAlivePool() = delete; private: - KeepAlivePool() + struct Item { - for (std::size_t i = 0; i < num_threads; ++i) { - std::thread(&KeepAlivePool::run, this).detach(); + std::vector depends; + std::function task; + }; + + /*! + * @brief How long the thread waits between passes over what it is watching. + */ + static constexpr std::chrono::microseconds min_poll_interval{50}; + static constexpr std::chrono::microseconds max_poll_interval{10000}; + + /*! + * @brief What polling may cost, when there is a lot of it to do. + * + * A pass reads the status of every event it is waiting on, so watching many + * tasks makes a pass expensive, and passing at the intervals above would + * then take a whole core. There is nothing to gain by paying that: a pass + * that takes a long time because there is a lot to look at pushes the next + * pass out, keeping polling to `max_poll_percent` of the thread, so that a + * backlog costs a little release latency rather than a core. + */ + static constexpr int max_poll_percent = 5; + static constexpr std::chrono::microseconds max_backlog_poll_interval{ + 250000}; + + /*! + * @brief Creates the pool belonging to the shared object that defines this. + * + * Defined in `_sycl_queue.pyx` module. + */ + friend KeepAlivePool &local_keep_alive_pool(); + + KeepAlivePool() { std::thread(&KeepAlivePool::run, this).detach(); } + + /*! + * @brief Whether every event in `item.depends` is known to have completed. + */ + static bool is_complete(const Item &item) + { + static constexpr auto complete = + sycl::info::event_command_status::complete; + + try { + for (const auto &e : item.depends) { + if (e.get_info() != + complete) + { + return false; + } + } + } catch (...) { + return false; + } + + return true; + } + + /*! + * @brief Runs the tasks of the completed items and drops them from `items`. + */ + static std::chrono::steady_clock::duration sweep(std::vector &items) + { + std::chrono::steady_clock::duration ran_for{}; + const auto started = std::chrono::steady_clock::now(); + + std::size_t n_waiting = 0; + for (std::size_t i = 0; i < items.size(); ++i) { + if (!is_complete(items[i])) { + // keep watching, packed to the front + if (n_waiting != i) { + items[n_waiting] = std::move(items[i]); + } + ++n_waiting; + continue; + } + + const auto task_started = std::chrono::steady_clock::now(); + try { + items[i].task(); + } catch (...) { + // a throwing task must not take down the thread or later tasks + // will be lost. `task` is caller-provided, so anything at all + // may come out of here. + } + ran_for += std::chrono::steady_clock::now() - task_started; } + // drops what has run, and what was moved to the front + items.resize(n_waiting); + + return std::chrono::steady_clock::now() - started - ran_for; } void run() { + // only this thread touches these, so polling takes no lock + std::vector watched; + std::vector arrived; + + auto poll_interval = min_poll_interval; + auto next_pass = std::chrono::steady_clock::now(); + for (;;) { - std::pair, std::function> item; { - std::unique_lock lock(queue_mutex_); - condition_.wait(lock, [this] { return !tasks_.empty(); }); - item = std::move(tasks_.front()); - tasks_.pop(); + std::unique_lock lock(submitted_mutex_); + if (watched.empty()) { + // nothing to poll for, so wait to be given something + condition_.wait(lock, + [this] { return !submitted_.empty(); }); + } + else { + condition_.wait_until(lock, next_pass, [this] { + return !submitted_.empty(); + }); + } + // `arrived` has been emptied, so this leaves `submitted_` empty + arrived.swap(submitted_); } - try { - sycl::event::wait(item.first); - } catch (const std::exception &) { - // run the task anyway: an async error must not strand the - // task or else it may leak + // whatever has just arrived is looked at right away as a fast path + const std::size_t n_arrived = arrived.size(); + sweep(arrived); + // a task ran, so there is reason to look again soon + bool progressed = n_arrived != 0; + + if (std::chrono::steady_clock::now() >= next_pass) { + const std::size_t n_watched = watched.size(); + const auto polled_for = sweep(watched); + progressed = progressed || watched.size() != n_watched; + + poll_interval = + progressed ? min_poll_interval + : std::min(poll_interval * 2, max_poll_interval); + const auto affordable = + std::chrono::duration_cast( + polled_for * (100 - max_poll_percent) / + max_poll_percent); + poll_interval = + std::max(poll_interval, + std::min(affordable, max_backlog_poll_interval)); + + next_pass = std::chrono::steady_clock::now() + poll_interval; } - try { - item.second(); - } catch (const std::exception &) { - // a throwing task must not take down the worker or later - // tasks will be lost + // waiting items are moved to the watched list for the next pass + for (auto &item : arrived) { + watched.push_back(std::move(item)); } + arrived.clear(); } } - std::queue, std::function>> - tasks_; - std::mutex queue_mutex_; + std::vector submitted_; + std::mutex submitted_mutex_; std::condition_variable condition_; + + std::vector> retired_; + std::mutex retired_mutex_; }; +/*! + * @brief Name of the kernel submitted by `submit_keep_alive_marker`. + */ +class keep_alive_marker; + +/*! + * @brief An event that gates the release of objects used by work on `q`. + * + * Submits an empty kernel that waits for `depends`, so that one event stands + * for every dependency of the release. + */ +inline sycl::event +submit_keep_alive_marker(sycl::queue &q, const std::vector &deps) +{ + return q.single_task(deps, []() {}); +} + } // namespace detail } // namespace dpctl diff --git a/dpctl/apis/include/dpctl4pybind11.hpp b/dpctl/apis/include/dpctl4pybind11.hpp index a87dc264a9..4f39e169d8 100644 --- a/dpctl/apis/include/dpctl4pybind11.hpp +++ b/dpctl/apis/include/dpctl4pybind11.hpp @@ -55,10 +55,11 @@ namespace detail */ inline bool interpreter_is_live() { + const bool initialized = Py_IsInitialized(); #if PY_VERSION_HEX < 0x30d0000 - return !_Py_IsFinalizing(); + return initialized && !_Py_IsFinalizing(); #else - return !Py_IsFinalizing(); + return initialized && !Py_IsFinalizing(); #endif } @@ -305,6 +306,12 @@ class dpctl_capi /*! * @brief The `KeepAlivePool` singleton, owned by `dpctl._sycl_queue`. + * + * The supported way of reaching the pool, which cannot be created by anything + * other than `dpctl` itself. Use it to release anything that must outlive + * offloaded work, as `dpctl::utils::keep_args_alive` does for Python objects. + * + * Throws `std::runtime_error` if the pool could not be obtained. */ inline KeepAlivePool &get_keep_alive_pool() { @@ -822,19 +829,72 @@ struct ManagedMemory } }; +/*! + * @brief Drops references taken for a release task that was never submitted. + * + * The references collected for a release task are dropped by that task, so + * anything that throws on the way to submitting it would leak them. They are + * dropped here instead, unless `handed_off` reports that the task took them + * over. Expects the caller to hold the GIL, as taking the references does. + */ +class held_references +{ +public: + held_references(std::shared_ptr *handles, + const std::size_t &n_held) + : m_handles(handles), m_n_held(n_held) + { + } + + held_references(const held_references &) = delete; + held_references &operator=(const held_references &) = delete; + + /*! + * @brief Report that the submitted task is responsible for the references. + */ + void handed_off() { m_handed_off = true; } + + ~held_references() + { + if (m_handed_off) { + return; + } + + for (std::size_t i = 0; i < m_n_held; ++i) { + m_handles[i]->dec_ref(); + } + } + +private: + std::shared_ptr *m_handles; + const std::size_t &m_n_held; + bool m_handed_off = false; +}; + } // end of namespace detail +/*! + * @brief Keeps `py_objs` alive until the work gated by `depends` completes. + * + * Returns an event for an empty kernel submitted to `q` after `depends`, which + * is what the release waits for. Waiting on it says that `py_objs` are no + * longer in use, not that they were released: once the event completes, USM + * allocations owned in C++ are freed on `dpctl`'s keep-alive thread, and the + * references to everything else are dropped by whichever thread next enters + * `dpctl` from Python. + */ template sycl::event keep_args_alive(sycl::queue &q, const py::object (&py_objs)[num], const std::vector &depends = {}) { - // q is only retained for API compatibility - (void)q; - std::size_t n_objects_held = 0; std::array, num> shp_arr{}; + // the task submitted below drops the references taken here, so they must + // be dropped by this guard if it is never submitted + detail::held_references held(shp_arr.data(), n_objects_held); + std::size_t n_usm_owners_held = 0; std::array, num> shp_usm{}; @@ -853,28 +913,40 @@ sycl::event keep_args_alive(sycl::queue &q, } } - if (n_usm_owners_held > 0 || n_objects_held > 0) { - dpctl::detail::get_keep_alive_pool().submit( - depends, [n_usm_owners_held, shp_usm = std::move(shp_usm), - n_objects_held, shp_arr = std::move(shp_arr)]() mutable { - for (std::size_t i = 0; i < n_usm_owners_held; ++i) { - shp_usm[i].reset(); - } + const sycl::event marker = + dpctl::detail::submit_keep_alive_marker(q, depends); + + auto &pool = dpctl::detail::get_keep_alive_pool(); - // if the main thread has not finalized the interpreter yet - if (n_objects_held > 0 && - dpctl::detail::interpreter_is_live()) { - py::gil_scoped_acquire acquire; + // the caller holds the GIL, so this is an opportunity to drop the + // references of the releases that have come due + pool.drain_retired(); + // captured by copy rather than moved from, so that the guard can still + // find the references should `submit` throw + pool.submit({marker}, [n_usm_owners_held, shp_usm, n_objects_held, + shp_arr]() mutable { + // the USM allocations are owned in C++ and need no interpreter, so + // they are released here + for (std::size_t i = 0; i < n_usm_owners_held; ++i) { + shp_usm[i].reset(); + } + + // the references are handed to a thread that holds the GIL rather + // than dropped here, as the thread running this must not touch + // Python + if (n_objects_held > 0) { + dpctl::detail::get_keep_alive_pool().retire( + [n_objects_held, shp_arr]() { for (std::size_t i = 0; i < n_objects_held; ++i) { shp_arr[i]->dec_ref(); } - } - }); - } + }); + } + }); + held.handed_off(); - // return dummy event for API compatibility - return sycl::event{}; + return marker; } /*! @brief Check if all allocation queues are the same as the diff --git a/dpctl/memory/_memory.pyx b/dpctl/memory/_memory.pyx index 6b2050bb9e..86328947ec 100644 --- a/dpctl/memory/_memory.pyx +++ b/dpctl/memory/_memory.pyx @@ -66,7 +66,7 @@ from dpctl._backend cimport ( # noqa: E211 from .._sycl_context cimport SyclContext from .._sycl_device cimport SyclDevice -from .._sycl_queue cimport SyclQueue +from .._sycl_queue cimport SyclQueue, drain_retired from .._sycl_queue_manager cimport get_device_cached_queue import collections @@ -164,6 +164,45 @@ def _to_memory(unsigned char[::1] b, str usm_kind): return res +cdef DPCTLSyclUSMRef _usm_alloc(Py_ssize_t alignment, Py_ssize_t nbytes, + bytes ptr_type, DPCTLSyclQueueRef QRef): + """ + Allocates `nbytes` of USM of `ptr_type`, returning NULL if it could not + be done. `ptr_type` must be one of b"shared", b"host" or b"device". + """ + cdef DPCTLSyclUSMRef p = NULL + + if (ptr_type == b"shared"): + if alignment > 0: + with nogil: + p = DPCTLaligned_alloc_shared( + alignment, nbytes, QRef + ) + else: + with nogil: + p = DPCTLmalloc_shared(nbytes, QRef) + elif (ptr_type == b"host"): + if alignment > 0: + with nogil: + p = DPCTLaligned_alloc_host( + alignment, nbytes, QRef + ) + else: + with nogil: + p = DPCTLmalloc_host(nbytes, QRef) + else: + if (alignment > 0): + with nogil: + p = DPCTLaligned_alloc_device( + alignment, nbytes, QRef + ) + else: + with nogil: + p = DPCTLmalloc_device(nbytes, QRef) + + return p + + cdef class _Memory: """ Internal class implementing methods common to MemoryUSMShared, MemoryUSMDevice, MemoryUSMHost @@ -183,43 +222,23 @@ cdef class _Memory: self._cinit_empty() if (nbytes > 0): - if queue is None: - queue = get_device_cached_queue(dpctl.SyclDevice()) - - QRef = queue.get_queue_ref() - if (ptr_type == b"shared"): - if alignment > 0: - with nogil: - p = DPCTLaligned_alloc_shared( - alignment, nbytes, QRef - ) - else: - with nogil: - p = DPCTLmalloc_shared(nbytes, QRef) - elif (ptr_type == b"host"): - if alignment > 0: - with nogil: - p = DPCTLaligned_alloc_host( - alignment, nbytes, QRef - ) - else: - with nogil: - p = DPCTLmalloc_host(nbytes, QRef) - elif (ptr_type == b"device"): - if (alignment > 0): - with nogil: - p = DPCTLaligned_alloc_device( - alignment, nbytes, QRef - ) - else: - with nogil: - p = DPCTLmalloc_device(nbytes, QRef) - else: + if ptr_type not in (b"shared", b"host", b"device"): raise RuntimeError( f"Pointer type '{ptr_type.decode('UTF-8')}' is not " "recognized" ) + if queue is None: + queue = get_device_cached_queue(dpctl.SyclDevice()) + + QRef = queue.get_queue_ref() + p = _usm_alloc(alignment, nbytes, ptr_type, QRef) + + if not p: + # drain already retired references to possibly free up memory + if drain_retired(): + p = _usm_alloc(alignment, nbytes, ptr_type, QRef) + if (p): self._memory_ptr = p self._opaque_ptr = OpaqueSmartPtr_Make(p, QRef) diff --git a/dpctl/tests/test_sycl_queue.py b/dpctl/tests/test_sycl_queue.py index 08e7e6882d..5d06c3077a 100644 --- a/dpctl/tests/test_sycl_queue.py +++ b/dpctl/tests/test_sycl_queue.py @@ -18,11 +18,14 @@ import ctypes import sys +import threading +import time import pytest import dpctl import dpctl.memory +from dpctl._sycl_queue import _drain_retired_references from .helper import create_invalid_capsule @@ -426,3 +429,67 @@ def test_submit_keep_args_alive_deprecated(): with pytest.warns(DeprecationWarning): ht_ev = q._submit_keep_args_alive((usm,), []) ht_ev.wait() + + +def test_submit_keep_args_alive_event_gates_on_depends(): + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + + n_bytes = 4 * 1024 * 1024 + host_buf = bytearray(n_bytes) + usm = dpctl.memory.MemoryUSMDevice(n_bytes, queue=q) + + copy_ev = q.copy_async(usm, host_buf, n_bytes) + with pytest.warns(DeprecationWarning): + ht_ev = q._submit_keep_args_alive((usm,), [copy_ev]) + + # the returned event is submitted after the gating events, so it cannot + # complete before they do + ht_ev.wait() + assert copy_ev.execution_status == dpctl.event_status_type.complete + + +def _scheduled_release(): + """Schedule the release of an object that records who released it. + + Returns the list the releasing thread's id is appended to, once the + reference `dpctl.keep_args_alive` took has been dropped. + """ + released = [] + + class Sentinel: + def __del__(self): + released.append(threading.get_ident()) + + args = (Sentinel(),) + dpctl.keep_args_alive(args, []) + del args + + # the events are complete, so the thread watching them has had time to + # find the release due and hand it over + time.sleep(0.1) + + # that thread must not drop the reference itself, as it cannot take the + # GIL safely + assert not released + + return released + + +def test_keep_args_alive_releases_on_a_python_thread(): + released = _scheduled_release() + + # the reference is dropped by a thread running Python, which is this one + _drain_retired_references() + assert released == [threading.get_ident()] + + +def test_keep_args_alive_releases_on_the_next_call(): + released = _scheduled_release() + + # entering dpctl from Python is enough, so a program that keeps offloading + # never accumulates references + dpctl.keep_args_alive((), []) + assert released == [threading.get_ident()] diff --git a/dpctl/tests/test_utils.py b/dpctl/tests/test_utils.py index 314336b97b..404e31b3c9 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -16,9 +16,12 @@ """Defines unit test cases for utility functions.""" +import time + import pytest import dpctl +import dpctl.memory import dpctl.utils @@ -75,8 +78,12 @@ def test_order_manager(): _mngr = _som[q] assert isinstance(_mngr.num_submitted_events, int) assert isinstance(_mngr.submitted_events, list) + assert isinstance(_mngr.num_cleanup_events, int) + assert isinstance(_mngr.cleanup_events, list) _mngr.add_event(dpctl.SyclEvent()) _mngr.add_event([dpctl.SyclEvent(), dpctl.SyclEvent()]) + _mngr.add_cleanup_event(dpctl.SyclEvent()) + _mngr.add_cleanup_event([dpctl.SyclEvent(), dpctl.SyclEvent()]) _mngr.wait() cpy = _mngr.__copy__() _som.clear() @@ -91,6 +98,57 @@ def test_order_manager(): assert _passed +def test_order_manager_waits_for_cleanup_events(): + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + + n_bytes = 4 * 1024 * 1024 + host_buf = bytearray(n_bytes) + usm = dpctl.memory.MemoryUSMDevice(n_bytes, queue=q) + + copy_ev = q.copy_async(usm, host_buf, n_bytes) + with pytest.warns(DeprecationWarning): + cleanup_ev = q._submit_keep_args_alive((usm,), [copy_ev]) + # only the cleanup event is recorded, so waiting on the manager can only + # wait for the copy through it + _mngr.add_cleanup_event(cleanup_ev) + _mngr.wait() + assert copy_ev.execution_status == dpctl.event_status_type.complete + assert _mngr.num_cleanup_events == 0 + _som.clear() + + +def test_order_manager_wait_drops_references(): + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + + released = [] + + class Sentinel: + def __del__(self): + released.append(True) + + args = (Sentinel(),) + dpctl.keep_args_alive(args, []) + del args + time.sleep(0.1) + assert not released + + # waiting is a point where the references held for completed tasks can be + # dropped, so it drops them + _mngr.wait() + assert released == [True] + _som.clear() + + def test_order_manager_deprecated_host_task_api(): try: q = dpctl.SyclQueue() diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index aef60970d4..b911479815 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -5,7 +5,7 @@ from collections import defaultdict from .._sycl_event import SyclEvent -from .._sycl_queue import SyclQueue +from .._sycl_queue import SyclQueue, _drain_retired_references from ._seq_order_keeper import _OrderManager @@ -13,6 +13,13 @@ class _SequentialOrderManager: """ Class to orchestrate default sequential order of the tasks offloaded from Python. + + Record offloaded tasks with :meth:`add_event` and use + :attr:`submitted_events` as the dependencies of the tasks that follow + them. Record events that gate the release of objects used by a task, + such as those returned by :func:`dpctl.SyclQueue._submit_keep_args_alive`, + with :meth:`add_cleanup_event`: they are waited on, but never become + dependencies of later tasks. """ def __init__(self): @@ -23,15 +30,14 @@ def __del__(self): return _local = self._state SyclEvent.wait_for(_local.get_submitted_events()) - # TODO: remove once deprecated add_event_pair is removed - SyclEvent.wait_for(_local.get_host_task_events()) + SyclEvent.wait_for(_local.get_cleanup_events()) def add_event_pair(self, host_task_ev, comp_ev): warnings.warn( "add_event_pair is deprecated and will be removed in a future " - "release. dpctl no longer submits host tasks, so there is no " - "separate host task event to track. Use add_event(comp_ev) " - "instead.", + "release. dpctl no longer submits host tasks. Use " + "add_event(comp_ev), and add_cleanup_event for an event that " + "gates the release of objects used by a task.", DeprecationWarning, stacklevel=2, ) @@ -57,16 +63,31 @@ def add_event(self, comp_ev): for ev in comp_ev: _local.add_to_submitted_events(ev) + def add_cleanup_event(self, cleanup_ev): + _local = self._state + if isinstance(cleanup_ev, SyclEvent): + _local.add_to_cleanup_events(cleanup_ev) + else: + if not isinstance(cleanup_ev, (list, tuple)): + cleanup_ev = (cleanup_ev,) + for ev in cleanup_ev: + _local.add_to_cleanup_events(ev) + @property def num_host_task_events(self): warnings.warn( "num_host_task_events is deprecated and will be removed in a " - "future release. dpctl no longer submits host tasks.", + "future release. dpctl no longer submits host tasks. Use " + "num_cleanup_events instead.", DeprecationWarning, stacklevel=2, ) + return self.num_cleanup_events + + @property + def num_cleanup_events(self): _local = self._state - return _local.get_num_host_task_events() + return _local.get_num_cleanup_events() @property def num_submitted_events(self): @@ -77,12 +98,17 @@ def num_submitted_events(self): def host_task_events(self): warnings.warn( "host_task_events is deprecated and will be removed in a future " - "release. dpctl no longer submits host tasks.", + "release. dpctl no longer submits host tasks. Use cleanup_events " + "instead.", DeprecationWarning, stacklevel=2, ) + return self.cleanup_events + + @property + def cleanup_events(self): _local = self._state - return _local.get_host_task_events() + return _local.get_cleanup_events() @property def submitted_events(self): @@ -91,7 +117,11 @@ def submitted_events(self): def wait(self): _local = self._state - return _local.wait() + res = _local.wait() + # the events are complete, so the references held for the tasks they + # gated are dropped here rather than left for the next call into dpctl + _drain_retired_references() + return res def __copy__(self): res = _SequentialOrderManager.__new__(_SequentialOrderManager) diff --git a/dpctl/utils/src/order_keeper.cpp b/dpctl/utils/src/order_keeper.cpp index 7f0074f91a..0eddc82ecc 100644 --- a/dpctl/utils/src/order_keeper.cpp +++ b/dpctl/utils/src/order_keeper.cpp @@ -14,15 +14,13 @@ PYBIND11_MODULE(_seq_order_keeper, m, py::mod_gil_not_used()) .def(py::init()) .def("get_num_submitted_events", &SequentialOrder::get_num_submitted_events) - .def("get_num_host_task_events", - &SequentialOrder::get_num_host_task_events) + .def("get_num_cleanup_events", &SequentialOrder::get_num_cleanup_events) .def("get_submitted_events", &SequentialOrder::get_submitted_events) - .def("get_host_task_events", &SequentialOrder::get_host_task_events) + .def("get_cleanup_events", &SequentialOrder::get_cleanup_events) .def("add_to_both_events", &SequentialOrder::add_to_both_events) .def("add_vector_to_both_events", &SequentialOrder::add_vector_to_both_events) - .def("add_to_host_task_events", - &SequentialOrder::add_to_host_task_events) + .def("add_to_cleanup_events", &SequentialOrder::add_to_cleanup_events) .def("add_to_submitted_events", &SequentialOrder::add_to_submitted_events) .def("wait", &SequentialOrder::wait, diff --git a/dpctl/utils/src/sequential_order_keeper.hpp b/dpctl/utils/src/sequential_order_keeper.hpp index 9330d24ed4..01c3019c1d 100644 --- a/dpctl/utils/src/sequential_order_keeper.hpp +++ b/dpctl/utils/src/sequential_order_keeper.hpp @@ -24,16 +24,20 @@ class SequentialOrder { private: mutable std::mutex mu_events; - std::vector host_task_events; + // events that gate the release of objects used by offloaded tasks, such as + // those returned by `dpctl::utils::keep_args_alive`. They are waited on, + // but never used as dependencies of later tasks. + std::vector cleanup_events; + // events for the offloaded tasks themselves, used as the dependencies of + // the tasks that follow them std::vector submitted_events; // only called with mu_events held void prune_complete_nolock() { - const auto &ht_it = - std::remove_if(host_task_events.begin(), host_task_events.end(), - is_event_complete); - host_task_events.erase(ht_it, host_task_events.end()); + const auto &cl_it = std::remove_if( + cleanup_events.begin(), cleanup_events.end(), is_event_complete); + cleanup_events.erase(cl_it, cleanup_events.end()); const auto &sub_it = std::remove_if(submitted_events.begin(), submitted_events.end(), @@ -42,25 +46,25 @@ class SequentialOrder } public: - SequentialOrder() : host_task_events{}, submitted_events{} {} - SequentialOrder(std::size_t n) : host_task_events{}, submitted_events{} + SequentialOrder() : cleanup_events{}, submitted_events{} {} + SequentialOrder(std::size_t n) : cleanup_events{}, submitted_events{} { - host_task_events.reserve(n); + cleanup_events.reserve(n); submitted_events.reserve(n); } SequentialOrder(const SequentialOrder &other) { std::lock_guard lock(other.mu_events); - host_task_events = other.host_task_events; + cleanup_events = other.cleanup_events; submitted_events = other.submitted_events; prune_complete_nolock(); } SequentialOrder(SequentialOrder &&other) - : host_task_events{}, submitted_events{} + : cleanup_events{}, submitted_events{} { std::lock_guard lock(other.mu_events); - host_task_events = std::move(other.host_task_events); + cleanup_events = std::move(other.cleanup_events); submitted_events = std::move(other.submitted_events); prune_complete_nolock(); } @@ -69,7 +73,7 @@ class SequentialOrder { if (this != &other) { std::scoped_lock lock(mu_events, other.mu_events); - host_task_events = other.host_task_events; + cleanup_events = other.cleanup_events; submitted_events = other.submitted_events; prune_complete_nolock(); } @@ -80,7 +84,7 @@ class SequentialOrder { if (this != &other) { std::scoped_lock lock(mu_events, other.mu_events); - host_task_events = std::move(other.host_task_events); + cleanup_events = std::move(other.cleanup_events); submitted_events = std::move(other.submitted_events); prune_complete_nolock(); } @@ -95,17 +99,17 @@ class SequentialOrder // returns a copy to avoid returning a reference that // could be modified after the lock is released - std::vector get_host_task_events() + std::vector get_cleanup_events() { std::lock_guard lock(mu_events); prune_complete_nolock(); - return host_task_events; + return cleanup_events; } - std::size_t get_num_host_task_events() const + std::size_t get_num_cleanup_events() const { std::lock_guard lock(mu_events); - return host_task_events.size(); + return cleanup_events.size(); } // returns a copy to avoid returning a reference that @@ -117,25 +121,25 @@ class SequentialOrder return submitted_events; } - void add_to_both_events(const sycl::event &ht_ev, + void add_to_both_events(const sycl::event &cleanup_ev, const sycl::event &comp_ev) { std::lock_guard lock(mu_events); prune_complete_nolock(); - if (!is_event_complete(ht_ev)) - host_task_events.push_back(ht_ev); + if (!is_event_complete(cleanup_ev)) + cleanup_events.push_back(cleanup_ev); if (!is_event_complete(comp_ev)) submitted_events.push_back(comp_ev); } - void add_vector_to_both_events(const std::vector &ht_evs, + void add_vector_to_both_events(const std::vector &cleanup_evs, const std::vector &comp_evs) { std::lock_guard lock(mu_events); prune_complete_nolock(); - for (const auto &e : ht_evs) { + for (const auto &e : cleanup_evs) { if (!is_event_complete(e)) - host_task_events.push_back(e); + cleanup_events.push_back(e); } for (const auto &e : comp_evs) { if (!is_event_complete(e)) @@ -143,12 +147,12 @@ class SequentialOrder } } - void add_to_host_task_events(const sycl::event &ht_ev) + void add_to_cleanup_events(const sycl::event &cleanup_ev) { std::lock_guard lock(mu_events); prune_complete_nolock(); - if (!is_event_complete(ht_ev)) { - host_task_events.push_back(ht_ev); + if (!is_event_complete(cleanup_ev)) { + cleanup_events.push_back(cleanup_ev); } } @@ -162,14 +166,14 @@ class SequentialOrder } template - void add_list_to_host_task_events(const sycl::event (&ht_events)[num]) + void add_list_to_cleanup_events(const sycl::event (&cleanup_evs)[num]) { std::lock_guard lock(mu_events); prune_complete_nolock(); for (std::size_t i = 0; i < num; ++i) { - const auto &e = ht_events[i]; + const auto &e = cleanup_evs[i]; if (!is_event_complete(e)) - host_task_events.push_back(e); + cleanup_events.push_back(e); } } @@ -190,14 +194,14 @@ class SequentialOrder // snapshot events outside of mutex to avoid // calling wait inside mutex std::vector sub_copy; - std::vector ht_copy; + std::vector cl_copy; { std::lock_guard lock(mu_events); sub_copy = submitted_events; - ht_copy = host_task_events; + cl_copy = cleanup_events; } sycl::event::wait(sub_copy); - sycl::event::wait(ht_copy); + sycl::event::wait(cl_copy); { std::lock_guard lock(mu_events); prune_complete_nolock(); diff --git a/scripts/_build_helper.py b/scripts/_build_helper.py index 086e6382ec..dfbc86f46d 100644 --- a/scripts/_build_helper.py +++ b/scripts/_build_helper.py @@ -93,7 +93,6 @@ def make_cmake_args( glog: bool = False, verbose: bool = False, other_opts: str = None, - keep_alive_pool_size: int = None, ): args = [ f"-DCMAKE_C_COMPILER:PATH={c_compiler}" if c_compiler else "", @@ -103,10 +102,6 @@ def make_cmake_args( f"-DDPCTL_ENABLE_GLOG:BOOL={'ON' if glog else 'OFF'}", ] - if keep_alive_pool_size is not None: - args.append( - f"-DDPCTL_KEEP_ALIVE_POOL_SIZE:STRING={keep_alive_pool_size}" - ) if verbose: args.append("-DCMAKE_VERBOSE_MAKEFILE:BOOL=ON") if other_opts: diff --git a/scripts/build_locally.py b/scripts/build_locally.py index a399912ce6..6918a429e5 100644 --- a/scripts/build_locally.py +++ b/scripts/build_locally.py @@ -97,13 +97,6 @@ def parse_args(): help="Disable Level Zero backend", ) - p.add_argument( - "--keep-alive-pool-size", - type=int, - default=None, - help="Number of threads used to keep objects alive during offload.", - ) - p.add_argument( "--cmake-opts", type=str, @@ -156,11 +149,6 @@ def main(): # Level Zero state (on unless explicitly disabled) level_zero_enabled = False if args.no_level_zero else True - if args.keep_alive_pool_size is not None and args.keep_alive_pool_size < 1: - err( - "--keep-alive-pool-size must be a positive integer", "build_locally" - ) - cmake_args = make_cmake_args( c_compiler=c_compiler, cxx_compiler=cxx_compiler, @@ -168,7 +156,6 @@ def main(): glog=args.glog, verbose=args.verbose, other_opts=args.cmake_opts, - keep_alive_pool_size=args.keep_alive_pool_size, ) # handle architecture conflicts From 9a6f1560615c4533ce696cb9eb9d4f158d32ea4c Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 14 Sep 2026 14:28:51 -0700 Subject: [PATCH 6/7] KeepAlivePool changed to KeepAliveWatcher No longer uses a thread pool, so fixes the misnomer Also clean up docstrings and comments --- CHANGELOG.md | 2 + dpctl/CMakeLists.txt | 2 +- dpctl/_async_dec_ref.hpp | 76 +++++++++--------- dpctl/_sycl_queue.pyx | 6 +- ..._alive_pool.hpp => keep_alive_watcher.hpp} | 79 +++++++++---------- dpctl/apis/include/dpctl4pybind11.hpp | 35 ++++---- 6 files changed, 99 insertions(+), 101 deletions(-) rename dpctl/apis/include/detail/{keep_alive_pool.hpp => keep_alive_watcher.hpp} (73%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 04baf37c84..2789ee8a53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed * Bump minimum NumPy version to 1.26 [gh-2192](https://github.com/IntelPython/dpctl/pull/2192) * Implemented a background thread that polls events to manage object lifetime during offload rather than use `host_task` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* The references taken by `dpctl.keep_args_alive`, `dpctl::utils::keep_args_alive` and `dpctl.SyclQueue._submit_keep_args_alive` are now dropped by a thread that is running Python rather than by the background thread, which no longer calls into the interpreter at all. They are dropped the next time `dpctl` is entered from Python, including on a wait on the order manager, and are leaked rather than dropped if that never happens again [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) * The event returned by `dpctl::utils::keep_args_alive` and `dpctl.SyclQueue._submit_keep_args_alive` is now for an empty kernel that gates the deferred release rather than for a `host_task` that performs it, so its completion means that the objects are no longer in use rather than that they were released [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) * Rewrote USM Python examples into a single example [gh-2292](https://github.com/IntelPython/dpctl/pull/2292) * Registered `DPCTL_PARTITION_AFFINITY_DOMAIN_UNKNOWN` enumerator when `DPCTLDevice_GetPartitionAffinityDomains` receives an unrecognized value from the SYCL runtime [gh-2324](https://github.com/IntelPython/dpctl/pull/2324) @@ -38,6 +39,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed * Fixed incorrect paths in `GetLevelZeroHeaders.cmake` [gh-2366](https://github.com/IntelPython/dpctl/pull/2366) +* A USM allocation that fails now drops the references held for offloaded tasks that have completed, and is attempted once more if that released anything, so that memory only waiting to be given up is not reported as unavailable [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) ### Maintenance * Updated pybind11 version used by `dpctl` and examples [gh-2357](https://github.com/IntelPython/dpctl/pull/2357) diff --git a/dpctl/CMakeLists.txt b/dpctl/CMakeLists.txt index 5672445fdf..001f1b7a6e 100644 --- a/dpctl/CMakeLists.txt +++ b/dpctl/CMakeLists.txt @@ -200,7 +200,7 @@ set(_cy_file ${CMAKE_CURRENT_SOURCE_DIR}/_sycl_queue.pyx) get_filename_component(_trgt ${_cy_file} NAME_WLE) build_dpctl_ext(${_trgt} ${_cy_file} "dpctl" SYCL) # _sycl_queue includes _async_dec_ref.hpp, which includes -# detail/keep_alive_pool.hpp from the public include directory +# detail/keep_alive_watcher.hpp from the public include directory target_include_directories(${_trgt} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_SOURCE_DIR}/apis/include diff --git a/dpctl/_async_dec_ref.hpp b/dpctl/_async_dec_ref.hpp index d12b5a63fe..ee014d8e03 100644 --- a/dpctl/_async_dec_ref.hpp +++ b/dpctl/_async_dec_ref.hpp @@ -41,7 +41,7 @@ #include "Python.h" -#include "detail/keep_alive_pool.hpp" +#include "detail/keep_alive_watcher.hpp" #include "syclinterface/dpctl_data_types.h" #include "syclinterface/dpctl_sycl_type_casters.hpp" @@ -54,25 +54,25 @@ namespace { /*! - * @brief The pool, once there is one, for those who must not create it. + * @brief The watcher, once there is one, for those who must not create it. */ -std::atomic created_pool{nullptr}; +std::atomic created_watcher{nullptr}; } // namespace /*! - * @brief The pool of the `dpctl._sycl_queue` module, which everyone shares. + * @brief The watcher of the `dpctl._sycl_queue` module, which everyone shares. * - * `KeepAlivePool` befriends this, making it the only creator of a pool. + * `KeepAliveWatcher` befriends this, making it the only creator of a watcher. */ -KeepAlivePool &local_keep_alive_pool() +KeepAliveWatcher &local_keep_alive_watcher() { // deliberately leaked: the polling thread is detached and holds a bare - // `this`, so the pool must outlive it - static KeepAlivePool *instance = []() { - KeepAlivePool *pool = new KeepAlivePool(); - created_pool.store(pool, std::memory_order_release); - return pool; + // `this`, so the watcher must outlive it + static KeepAliveWatcher *instance = []() { + KeepAliveWatcher *watcher = new KeepAliveWatcher(); + created_watcher.store(watcher, std::memory_order_release); + return watcher; }(); return *instance; } @@ -81,14 +81,14 @@ KeepAlivePool &local_keep_alive_pool() } // namespace dpctl /*! - * @brief Address of the `KeepAlivePool`. + * @brief Address of the `KeepAliveWatcher`. * - * Returns nullptr if the pool could not be created. + * Returns nullptr if the watcher could not be created. */ -void *keep_alive_pool_ptr() +void *keep_alive_watcher_ptr() { try { - return static_cast(&dpctl::detail::local_keep_alive_pool()); + return static_cast(&dpctl::detail::local_keep_alive_watcher()); } catch (...) { // nothing may escape into the calling Cython code, which is not // prepared to handle a C++ exception @@ -99,22 +99,23 @@ void *keep_alive_pool_ptr() /*! * @brief Drop the references of the DECREFs that have come due. * - * Returns whether there were any, so that a caller draining in the hope of - * reclaiming something knows whether anything was given up. + * Does nothing until there is a watcher, and never creates one. * * Expects the caller to hold the GIL. + * + * @return Whether there were any. */ bool drain_retired_references() { - // a pool being created right now is left to the next drain - dpctl::detail::KeepAlivePool *pool = - dpctl::detail::created_pool.load(std::memory_order_acquire); - if (!pool) { + // a watcher being created right now is left to the next drain + dpctl::detail::KeepAliveWatcher *watcher = + dpctl::detail::created_watcher.load(std::memory_order_acquire); + if (!watcher) { return false; } try { - return pool->drain_retired(); + return watcher->drain_retired(); } catch (...) { // nothing may escape into the calling Cython code, which is not // prepared to handle a C++ exception @@ -148,22 +149,21 @@ std::vector unwrap_events(DPCTLSyclEventRef *depERefs, void submit_dec_ref(std::vector obj_vec, std::vector depends) { - auto &pool = dpctl::detail::local_keep_alive_pool(); - - // the caller holds the GIL, so this is an opportunity to drop the - // references of the DECREFs scheduled before this one - pool.drain_retired(); - - pool.submit(std::move(depends), [obj_vec = std::move(obj_vec)]() mutable { - // handed to a thread that holds the GIL rather than dropped here, as - // the thread running this must not touch Python - dpctl::detail::local_keep_alive_pool().retire( - [obj_vec = std::move(obj_vec)]() { - for (PyObject *obj : obj_vec) { - Py_DECREF(obj); - } - }); - }); + auto &watcher = dpctl::detail::local_keep_alive_watcher(); + + // the caller holds the GIL, so this is an opportunity to drain references + watcher.drain_retired(); + + watcher.submit(std::move(depends), + [obj_vec = std::move(obj_vec)]() mutable { + // handed to a thread that holds the GIL + dpctl::detail::local_keep_alive_watcher().retire( + [obj_vec = std::move(obj_vec)]() { + for (PyObject *obj : obj_vec) { + Py_DECREF(obj); + } + }); + }); } } // namespace diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index 256fda803f..5a1735d446 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -117,7 +117,7 @@ cdef extern from "_async_dec_ref.hpp": DPCTLSyclQueueRef, PyObject **, size_t, DPCTLSyclEventRef *, size_t, int * ) nogil - void *keep_alive_pool_ptr() nogil + void *keep_alive_watcher_ptr() nogil bint drain_retired_references() @@ -2286,8 +2286,8 @@ cdef api SyclQueue SyclQueue_Make(DPCTLSyclQueueRef QRef): return SyclQueue._create(copied_QRef) -cdef api void *KeepAlivePool_Get() noexcept nogil: - return keep_alive_pool_ptr() +cdef api void *KeepAliveWatcher_Get() noexcept nogil: + return keep_alive_watcher_ptr() cdef bint drain_retired(): diff --git a/dpctl/apis/include/detail/keep_alive_pool.hpp b/dpctl/apis/include/detail/keep_alive_watcher.hpp similarity index 73% rename from dpctl/apis/include/detail/keep_alive_pool.hpp rename to dpctl/apis/include/detail/keep_alive_watcher.hpp index 07a2526c29..d9039f77b6 100644 --- a/dpctl/apis/include/detail/keep_alive_pool.hpp +++ b/dpctl/apis/include/detail/keep_alive_watcher.hpp @@ -1,4 +1,4 @@ -//===--- keep_alive_pool.hpp - keeps owners alive during offload ----------===// +//===--- keep_alive_watcher.hpp - keeps owners alive during offload -------===// // // Data Parallel Control (dpctl) // @@ -48,29 +48,21 @@ namespace detail * @brief A thread that polls SYCL events and then runs a callable. * * A single thread polls the events of every task submitted to it, so that - * tasks run as soon as their own events complete, in whatever order that - * happens, and no thread ever blocks on offloaded work. Blocking would be - * costly as well as ordering: `sycl::event::wait` busy-waits on some backends, - * so a thread parked on a long-running task would burn a core to do nothing. + * tasks run as soon as their own events complete and no thread blocks on + * offloaded work. * - * The thread never runs Python code, so that it can never be waited for by a - * thread holding the GIL and can never be caught asking for the GIL as the - * interpreter finalizes. Work that needs the GIL goes to `retire` instead, and - * runs on a thread that already holds it. - * - * `dpctl` owns the single instance of the pool that every user of `dpctl` - * shares. Obtain it with `dpctl::detail::get_keep_alive_pool()`, declared in - * `dpctl4pybind11.hpp`, which is the only supported way of reaching it. + * The thread never runs Python code, and work that needs the GIL goes to + * `retire` instead, and runs on a thread that already holds it. */ -class KeepAlivePool +class KeepAliveWatcher { public: /*! * @brief Run `task` once every event in `depends` has completed. * - * It runs on the polling thread, which has to be kept moving, so `task` - * must not block on offloaded work, and must not touch Python. Pass work - * that needs the GIL to `retire` rather than doing it here. + * `task` must not block on offloaded work and must not touch Python. + * + * Pass work that needs the GIL to `retire` rather than doing it here. */ void submit(std::vector depends, std::function task) { @@ -93,15 +85,17 @@ class KeepAlivePool /*! * @brief Run everything retired so far, on the calling thread. * - * Returns a bool representing whether there was anything to run. + * Expects the caller to hold the GIL, as retired work generally needs it. + * + * @return Whether there was anything to run. */ bool drain_retired() { std::vector> to_run; { std::lock_guard lock(retired_mutex_); - // swapped out rather than run under the lock, as dropping the last - // reference to an object can run code that retires more work + // swapped out rather than run under the lock, as a release can + // retire more work to_run.swap(retired_); } @@ -111,17 +105,16 @@ class KeepAlivePool try { release(); } catch (...) { - // the rest must still run. `release` is caller-provided, so - // anything at all may come out of here. + // the rest must still run } } return ran_any; } - KeepAlivePool(const KeepAlivePool &) = delete; - KeepAlivePool &operator=(const KeepAlivePool &) = delete; - ~KeepAlivePool() = delete; + KeepAliveWatcher(const KeepAliveWatcher &) = delete; + KeepAliveWatcher &operator=(const KeepAliveWatcher &) = delete; + ~KeepAliveWatcher() = delete; private: struct Item @@ -139,28 +132,27 @@ class KeepAlivePool /*! * @brief What polling may cost, when there is a lot of it to do. * - * A pass reads the status of every event it is waiting on, so watching many - * tasks makes a pass expensive, and passing at the intervals above would - * then take a whole core. There is nothing to gain by paying that: a pass - * that takes a long time because there is a lot to look at pushes the next - * pass out, keeping polling to `max_poll_percent` of the thread, so that a - * backlog costs a little release latency rather than a core. + * An expensive pass pushes the next one out to keep polling to + * `max_poll_percent` of the thread, by no more than + * `max_backlog_poll_interval`. */ static constexpr int max_poll_percent = 5; static constexpr std::chrono::microseconds max_backlog_poll_interval{ 250000}; /*! - * @brief Creates the pool belonging to the shared object that defines this. + * @brief Creates the watcher, and is the only thing that can. * - * Defined in `_sycl_queue.pyx` module. + * Defined in the `_sycl_queue.pyx` module. */ - friend KeepAlivePool &local_keep_alive_pool(); + friend KeepAliveWatcher &local_keep_alive_watcher(); - KeepAlivePool() { std::thread(&KeepAlivePool::run, this).detach(); } + KeepAliveWatcher() { std::thread(&KeepAliveWatcher::run, this).detach(); } /*! - * @brief Whether every event in `item.depends` is known to have completed. + * @brief Whether every event in `item.depends` has completed. + * + * An event whose status cannot be read is reported as not complete. */ static bool is_complete(const Item &item) { @@ -184,6 +176,8 @@ class KeepAlivePool /*! * @brief Runs the tasks of the completed items and drops them from `items`. + * + * @return How long reading the statuses took, not counting the tasks. */ static std::chrono::steady_clock::duration sweep(std::vector &items) { @@ -205,13 +199,11 @@ class KeepAlivePool try { items[i].task(); } catch (...) { - // a throwing task must not take down the thread or later tasks - // will be lost. `task` is caller-provided, so anything at all - // may come out of here. + // a throwing task must not take down the thread } ran_for += std::chrono::steady_clock::now() - task_started; } - // drops what has run, and what was moved to the front + // drops what has run items.resize(n_waiting); return std::chrono::steady_clock::now() - started - ran_for; @@ -292,8 +284,11 @@ class keep_alive_marker; /*! * @brief An event that gates the release of objects used by work on `q`. * - * Submits an empty kernel that waits for `depends`, so that one event stands - * for every dependency of the release. + * Submits an empty kernel that waits for `deps`, so that a single event stands + * for every dependency of the release and, on an in-order queue, for the work + * already submitted to the queue as well. + * + * @return An event that completes once the objects have stopped being used. */ inline sycl::event submit_keep_alive_marker(sycl::queue &q, const std::vector &deps) diff --git a/dpctl/apis/include/dpctl4pybind11.hpp b/dpctl/apis/include/dpctl4pybind11.hpp index 4f39e169d8..22cfc98b5e 100644 --- a/dpctl/apis/include/dpctl4pybind11.hpp +++ b/dpctl/apis/include/dpctl4pybind11.hpp @@ -25,7 +25,7 @@ #pragma once -#include "detail/keep_alive_pool.hpp" +#include "detail/keep_alive_watcher.hpp" #include "dpctl_capi.h" #include @@ -305,27 +305,28 @@ class dpctl_capi }; // struct dpctl_capi /*! - * @brief The `KeepAlivePool` singleton, owned by `dpctl._sycl_queue`. + * @brief The `KeepAliveWatcher` singleton, owned by `dpctl._sycl_queue`. * - * The supported way of reaching the pool, which cannot be created by anything - * other than `dpctl` itself. Use it to release anything that must outlive - * offloaded work, as `dpctl::utils::keep_args_alive` does for Python objects. + * The supported way of reaching the watcher, which cannot be created by + * anything other than `dpctl` itself. Use it to release anything that must + * outlive offloaded work, as `dpctl::utils::keep_args_alive` does for Python + * objects. * - * Throws `std::runtime_error` if the pool could not be obtained. + * Throws `std::runtime_error` if the watcher could not be obtained. */ -inline KeepAlivePool &get_keep_alive_pool() +inline KeepAliveWatcher &get_keep_alive_watcher() { - static KeepAlivePool *pool = []() -> KeepAlivePool * { + static KeepAliveWatcher *watcher = []() -> KeepAliveWatcher * { // get dpctl_capi to prevent nullptr return static_cast(dpctl_capi::get()); - return static_cast(KeepAlivePool_Get()); + return static_cast(KeepAliveWatcher_Get()); }(); - if (!pool) { - throw std::runtime_error("Could not create dpctl's keep-alive pool"); + if (!watcher) { + throw std::runtime_error("Could not create dpctl's keep-alive watcher"); } - return *pool; + return *watcher; } } // namespace detail @@ -916,16 +917,16 @@ sycl::event keep_args_alive(sycl::queue &q, const sycl::event marker = dpctl::detail::submit_keep_alive_marker(q, depends); - auto &pool = dpctl::detail::get_keep_alive_pool(); + auto &watcher = dpctl::detail::get_keep_alive_watcher(); // the caller holds the GIL, so this is an opportunity to drop the // references of the releases that have come due - pool.drain_retired(); + watcher.drain_retired(); // captured by copy rather than moved from, so that the guard can still // find the references should `submit` throw - pool.submit({marker}, [n_usm_owners_held, shp_usm, n_objects_held, - shp_arr]() mutable { + watcher.submit({marker}, [n_usm_owners_held, shp_usm, n_objects_held, + shp_arr]() mutable { // the USM allocations are owned in C++ and need no interpreter, so // they are released here for (std::size_t i = 0; i < n_usm_owners_held; ++i) { @@ -936,7 +937,7 @@ sycl::event keep_args_alive(sycl::queue &q, // than dropped here, as the thread running this must not touch // Python if (n_objects_held > 0) { - dpctl::detail::get_keep_alive_pool().retire( + dpctl::detail::get_keep_alive_watcher().retire( [n_objects_held, shp_arr]() { for (std::size_t i = 0; i < n_objects_held; ++i) { shp_arr[i]->dec_ref(); From 84bf7e30eaf11826843b0a2a12f49d08c922f30f Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 14 Sep 2026 14:35:51 -0700 Subject: [PATCH 7/7] fix pre-commit --- pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index b1d88ddcfc..16d68e5800 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,7 +15,7 @@ requires = [ ] [project] -authors = [{name = "Intel Corporation"}] +authors = [{ name = "Intel Corporation" }] classifiers = [ "Development Status :: 4 - Beta", "Intended Audience :: Science/Research", @@ -57,7 +57,7 @@ keywords = [ ] license = "Apache-2.0" name = "dpctl" -readme = {file = "README.md", content-type = "text/markdown"} +readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.10" [project.optional-dependencies]