diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d2230894a..2789ee8a53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,9 +15,18 @@ 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 [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, 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 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) @@ -30,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/CMakeLists.txt b/CMakeLists.txt index 0d27c9d6c3..a9f139e598 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -48,7 +48,6 @@ option( size of shared object with offloading sections" OFF ) - 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/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..cd0bea5866 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -15,3 +15,23 @@ 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`. + + 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, and cleanup is + tracked by ``add_cleanup_event``, ``cleanup_events`` and + ``num_cleanup_events``. diff --git a/dpctl/CMakeLists.txt b/dpctl/CMakeLists.txt index a4378a03d0..001f1b7a6e 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_watcher.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/__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 new file mode 100644 index 0000000000..ee014d8e03 --- /dev/null +++ b/dpctl/_async_dec_ref.hpp @@ -0,0 +1,246 @@ +//===--- _async_dec_ref.hpp - Implements async DECREF ---------------------===// +// +// 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 +/// 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 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 "Python.h" + +#include "detail/keep_alive_watcher.hpp" +#include "syclinterface/dpctl_data_types.h" +#include "syclinterface/dpctl_sycl_type_casters.hpp" + +namespace dpctl +{ +namespace detail +{ + +namespace +{ + +/*! + * @brief The watcher, once there is one, for those who must not create it. + */ +std::atomic created_watcher{nullptr}; + +} // namespace + +/*! + * @brief The watcher of the `dpctl._sycl_queue` module, which everyone shares. + * + * `KeepAliveWatcher` befriends this, making it the only creator of a watcher. + */ +KeepAliveWatcher &local_keep_alive_watcher() +{ + // deliberately leaked: the polling thread is detached and holds a bare + // `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; +} + +} // namespace detail +} // namespace dpctl + +/*! + * @brief Address of the `KeepAliveWatcher`. + * + * Returns nullptr if the watcher could not be created. + */ +void *keep_alive_watcher_ptr() +{ + try { + 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 + return nullptr; + } +} + +/*! + * @brief Drop the references of the DECREFs that have come due. + * + * 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 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 watcher->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 &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 + +/*! + * @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) +{ + try { + 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 (...) { + // 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 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 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::unwrap; + using dpctl::syclinterface::wrap; + + 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)); + + submit_dec_ref( + std::vector(obj_array, obj_array + obj_array_size), + {marker}); + + 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; + } +} diff --git a/dpctl/_host_task_util.hpp b/dpctl/_host_task_util.hpp deleted file mode 100644 index 6898893bdd..0000000000 --- a/dpctl/_host_task_util.hpp +++ /dev/null @@ -1,96 +0,0 @@ -//===--- _host_tasl_util.hpp - Implements async DECREF =// -// -// Data Parallel Control (dpctl) -// -// Copyright 2022 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 -/// 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. -/// -/// N.B.: The host task attempts to acquire GIL, so queue wait, event wait and -/// other synchronization mechanisms should be called after releasing the GIL to -/// avoid deadlocks. -/// -//===----------------------------------------------------------------------===// - -#pragma once -#include -#include -#include - -#include "Python.h" - -#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) -{ - using dpctl::syclinterface::unwrap; - using dpctl::syclinterface::wrap; - - sycl::queue *q = unwrap(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]() { - 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) { - PyGILState_STATE gstate; - gstate = PyGILState_Ensure(); - for (size_t i = 0; i < obj_array_size; ++i) { - Py_DECREF(obj_vec[i]); - } - PyGILState_Release(gstate); - } - }); - }); - - static constexpr int result_ok = 0; - - *status = result_ok; - auto e_ptr = new sycl::event(ht_ev); - 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; - - *status = result_other_abnormal; - return nullptr; -} 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 47bd7f2f67..5a1735d446 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,24 @@ import collections.abc import logging import struct import sys +import warnings -cdef extern from "_host_task_util.hpp": - DPCTLSyclEventRef async_dec_ref( +cdef extern from "_async_dec_ref.hpp": + 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 + void *keep_alive_watcher_ptr() nogil + bint drain_retired_references() __all__ = [ + "keep_args_alive", "SyclQueue", "SyclKernelInvalidRangeError", "SyclKernelSubmitError", @@ -1347,6 +1355,9 @@ cdef class SyclQueue(_SyclQueue): Keeps objects in ``args`` alive until tasks associated with events complete. + Deprecated. 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. @@ -1357,18 +1368,33 @@ cdef class SyclQueue(_SyclQueue): working on Python objects collected in ``args``. Returns: dpctl.SyclEvent - The event associated with the submission of host task. - - Increments reference count of ``args`` and schedules asynchronous - ``host_task`` to decrement the count once dependent events are - complete. + 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 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 ``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`` - 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 " + "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 @@ -1398,7 +1424,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 @@ -1406,10 +1432,8 @@ cdef class SyclQueue(_SyclQueue): free(depEvents) if (status != 0): - with nogil: - DPCTLEvent_Wait(htERef) - DPCTLEvent_Delete(htERef) - raise RuntimeError("Could not submit keep_args_alive host_task") + Py_DECREF(args) + raise RuntimeError("Could not schedule keep_args_alive") return SyclEvent._create(htERef) @@ -1451,7 +1475,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 @@ -2261,6 +2285,27 @@ cdef api SyclQueue SyclQueue_Make(DPCTLSyclQueueRef QRef): cdef DPCTLSyclQueueRef copied_QRef = DPCTLQueue_Copy(QRef) return SyclQueue._create(copied_QRef) + +cdef api void *KeepAliveWatcher_Get() noexcept nogil: + return keep_alive_watcher_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): @@ -2454,3 +2499,77 @@ 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 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 + + import dpctl + + q = dpctl.SyclQueue() + e = q.submit_async(kernel, [x_usm], [n]) + dpctl.keep_args_alive((x_usm,), [e]) + + .. note:: + 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 + 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/apis/include/detail/keep_alive_watcher.hpp b/dpctl/apis/include/detail/keep_alive_watcher.hpp new file mode 100644 index 0000000000..d9039f77b6 --- /dev/null +++ b/dpctl/apis/include/detail/keep_alive_watcher.hpp @@ -0,0 +1,300 @@ +//===--- keep_alive_watcher.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 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 + +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 and no thread blocks on + * offloaded work. + * + * 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 KeepAliveWatcher +{ +public: + /*! + * @brief Run `task` once every event in `depends` has completed. + * + * `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) + { + { + std::lock_guard lock(submitted_mutex_); + submitted_.push_back(Item{std::move(depends), std::move(task)}); + } + condition_.notify_one(); + } + + /*! + * @brief Hand `release` to whichever thread drains next. + */ + void retire(std::function release) + { + std::lock_guard lock(retired_mutex_); + retired_.push_back(std::move(release)); + } + + /*! + * @brief Run everything retired so far, on the calling thread. + * + * 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 a release can + // retire more work + to_run.swap(retired_); + } + + const bool ran_any = !to_run.empty(); + + for (auto &release : to_run) { + try { + release(); + } catch (...) { + // the rest must still run + } + } + + return ran_any; + } + + KeepAliveWatcher(const KeepAliveWatcher &) = delete; + KeepAliveWatcher &operator=(const KeepAliveWatcher &) = delete; + ~KeepAliveWatcher() = delete; + +private: + struct Item + { + 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. + * + * 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 watcher, and is the only thing that can. + * + * Defined in the `_sycl_queue.pyx` module. + */ + friend KeepAliveWatcher &local_keep_alive_watcher(); + + KeepAliveWatcher() { std::thread(&KeepAliveWatcher::run, this).detach(); } + + /*! + * @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) + { + 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`. + * + * @return How long reading the statuses took, not counting the tasks. + */ + 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 + } + ran_for += std::chrono::steady_clock::now() - task_started; + } + // drops what has run + 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::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_); + } + + // 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; + } + + // 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::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 `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) +{ + return q.single_task(deps, []() {}); +} + +} // namespace detail +} // namespace dpctl diff --git a/dpctl/apis/include/dpctl4pybind11.hpp b/dpctl/apis/include/dpctl4pybind11.hpp index ecfc85596f..22cfc98b5e 100644 --- a/dpctl/apis/include/dpctl4pybind11.hpp +++ b/dpctl/apis/include/dpctl4pybind11.hpp @@ -25,6 +25,7 @@ #pragma once +#include "detail/keep_alive_watcher.hpp" #include "dpctl_capi.h" #include @@ -46,6 +47,22 @@ 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() +{ + const bool initialized = Py_IsInitialized(); +#if PY_VERSION_HEX < 0x30d0000 + return initialized && !_Py_IsFinalizing(); +#else + return initialized && !Py_IsFinalizing(); +#endif +} + class dpctl_capi { public: @@ -173,15 +190,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; } } @@ -294,6 +303,32 @@ class dpctl_capi dpctl_capi &operator=(dpctl_capi &&) = default; }; // struct dpctl_capi + +/*! + * @brief The `KeepAliveWatcher` singleton, owned by `dpctl._sycl_queue`. + * + * 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 watcher could not be obtained. + */ +inline KeepAliveWatcher &get_keep_alive_watcher() +{ + static KeepAliveWatcher *watcher = []() -> KeepAliveWatcher * { + // get dpctl_capi to prevent nullptr return + static_cast(dpctl_capi::get()); + + return static_cast(KeepAliveWatcher_Get()); + }(); + + if (!watcher) { + throw std::runtime_error("Could not create dpctl's keep-alive watcher"); + } + return *watcher; +} + } // namespace detail } // namespace dpctl @@ -795,8 +830,60 @@ 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], @@ -805,6 +892,10 @@ sycl::event keep_args_alive(sycl::queue &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{}; @@ -823,46 +914,40 @@ sycl::event keep_args_alive(sycl::queue &q, } } - bool use_depends = true; - sycl::event host_task_ev; + const sycl::event marker = + dpctl::detail::submit_keep_alive_marker(q, depends); - 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_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; + auto &watcher = dpctl::detail::get_keep_alive_watcher(); - for (std::size_t i = 0; i < n_objects_held; ++i) { - shp_arr[i]->dec_ref(); - } - }); - }); - } + // the caller holds the GIL, so this is an opportunity to drop the + // references of the releases that have come due + watcher.drain_retired(); + + // captured by copy rather than moved from, so that the guard can still + // find the references should `submit` throw + 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) { + 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_watcher().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 host_task_ev; + 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_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..5d06c3077a 100644 --- a/dpctl/tests/test_sycl_queue.py +++ b/dpctl/tests/test_sycl_queue.py @@ -18,10 +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 @@ -402,3 +406,90 @@ 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() + + +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_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..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 @@ -73,13 +76,14 @@ 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()]) + 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() @@ -92,3 +96,76 @@ def test_order_manager(): _passed = True finally: 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() + 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..b911479815 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -1,10 +1,11 @@ import sys import threading +import warnings import weakref 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 @@ -12,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): @@ -22,9 +30,17 @@ def __del__(self): return _local = self._state SyclEvent.wait_for(_local.get_submitted_events()) - 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. 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, + ) _local = self._state if isinstance(host_task_ev, SyclEvent) and isinstance( comp_ev, SyclEvent @@ -37,10 +53,41 @@ 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) + + 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. 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): @@ -49,8 +96,19 @@ 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. 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): @@ -59,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/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() 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]