diff --git a/CHANGELOG.md b/CHANGELOG.md index 0aa6be7eab..afb4bcef46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,16 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [dev] - XXX. XX, XXXX ### Added +* Added `dpctl.SyclQueue.keep_args_alive` method, which keeps Python objects alive until the work on that queue using them completes [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* Added `add_event` and `add_cleanup_event` methods 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) +* Added `SyclQueue_DeferRelease` C-API function, which defers a release callback until work submitted to a queue completes, and through which `dpctl::utils::keep_args_alive` keeps its arguments alive [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) ### Changed +* Object lifetime during offload is now managed by a list of releases deferred until the events that gate them have completed, rather than by a `host_task`. `dpctl` keeps one thread for looking over the list and running the releases that have come due, started by the first release deferred and again in the child of a `fork`, which takes the GIL to do so and lets a release call into the interpreter and free USM allocations [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* The event returned by `dpctl.SyclQueue.keep_args_alive`, `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 have been released [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) ### Deprecated +* Deprecated `dpctl.SyclQueue._submit_keep_args_alive` in favor of `dpctl.SyclQueue.keep_args_alive`, which it now calls and which behaves identically [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`; they do not yet emit a `DeprecationWarning` [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) ### Removed * Removed the deprecated `dpctl.program` submodule, use `dpctl.compiler` instead [gh-2407](https://github.com/IntelPython/dpctl/pull/2407) ### Fixed * Removed stale `select_host_device` and `has_host_device` names from `dpctl.__all__`, which made `from dpctl import *` fail with `AttributeError` [gh-2410](https://github.com/IntelPython/dpctl/pull/2410) +* 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 * Removed the `-fno-sycl-id-queries-fit-in-int` compile option, which none of the `dpctl` extensions need since `dpctl.tensor` moved to `dpnp` [gh-2413](https://github.com/IntelPython/dpctl/pull/2413) diff --git a/docs/doc_sources/api_reference/dpctl/utils.rst b/docs/doc_sources/api_reference/dpctl/utils.rst index 093f298ff6..5706213530 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -15,3 +15,25 @@ 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 + :meth:`dpctl.SyclQueue.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 + :meth:`dpctl.SyclQueue.keep_args_alive` took for tasks that have since + completed, which ``dpctl`` otherwise does on a thread of its own. This is + best-effort: references that thread is already dropping are left to it, so + they may still be held when ``wait`` returns. + + .. deprecated:: 0.24.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 48ac3885f9..9c9d4f9cad 100644 --- a/dpctl/CMakeLists.txt +++ b/dpctl/CMakeLists.txt @@ -198,7 +198,7 @@ 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 +# _sycl_queue includes _async_dec_ref.hpp target_include_directories(${_trgt} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}) target_link_libraries(DpctlCAPI INTERFACE ${_trgt}_headers) diff --git a/dpctl/_async_dec_ref.hpp b/dpctl/_async_dec_ref.hpp new file mode 100644 index 0000000000..79d37fb102 --- /dev/null +++ b/dpctl/_async_dec_ref.hpp @@ -0,0 +1,207 @@ +//===--- _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 the functions that defer releasing what offloaded work +/// uses until that work has completed: decrementing the reference counts of an +/// array of Python objects, or calling a release function another module +/// passes in. +/// +/// N.B.: The reference counts are dropped by the thread `dpctl` keeps for +/// running deferred releases, or by a thread that drains them. +/// +//===----------------------------------------------------------------------===// + +#pragma once +#include +#include +#include +#include +#include +#include + +#include "Python.h" + +#include "_deferred_release_watcher.hpp" +#include "syclinterface/dpctl_data_types.h" +#include "syclinterface/dpctl_sycl_type_casters.hpp" + +namespace dpctl +{ +namespace detail +{ + +/*! + * @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 after `deps`. On an in-order queue the submission is + * also ordered after everything already submitted there, so the event covers + * uses of the objects that the caller never tracked. + */ +inline sycl::event +submit_keep_alive_marker(sycl::queue &q, const std::vector &deps) +{ + return q.single_task(deps, []() {}); +} + +} // namespace detail +} // namespace dpctl + +namespace +{ + +/*! + * @brief Copy the events behind `nDepERefs` 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 The queue behind `QRef`, which must not be null. + */ +sycl::queue &unwrap_queue(DPCTLSyclQueueRef QRef) +{ + sycl::queue *q = dpctl::syclinterface::unwrap(QRef); + if (!q) { + throw std::invalid_argument("A queue is required"); + } + return *q; +} + +/*! + * @brief Schedule DECREFs of `obj_vec` for once `depends` have completed. + */ +void submit_dec_ref(std::vector obj_vec, + std::vector depends) +{ + dpctl::detail::local_deferred_releases().defer( + std::move(depends), [obj_vec = std::move(obj_vec)]() { + for (PyObject *obj : obj_vec) { + Py_DECREF(obj); + } + }); +} + +} // namespace + +/*! + * @brief Schedule DECREFs of `obj_array` for once `obj_array` is out of use. + * + * An empty kernel is submitted to `QRef` after `depERefs`, and the DECREFs wait + * for it. On an in-order queue submission also orders it after the work already + * submitted there, so the objects are covered regardless of providing dependent + * events. `QRef` should be the queue running the work that uses the objects. + * An event for the kernel is returned. + * + * Sets `*status` to 0 on success and 1 if scheduling threw, and returns nullptr + * on failure. A failed call has scheduled nothing, so the caller still owns the + * references it took. + */ +inline DPCTLSyclEventRef async_dec_ref(DPCTLSyclQueueRef QRef, + PyObject **obj_array, + size_t obj_array_size, + DPCTLSyclEventRef *depERefs, + size_t nDepERefs, + int *status) +{ + using dpctl::syclinterface::wrap; + + try { + sycl::queue &q = unwrap_queue(QRef); + + std::vector depends = unwrap_events(depERefs, nDepERefs); + std::vector obj_vec(obj_array, obj_array + obj_array_size); + + static constexpr int result_ok = 0; + + const sycl::event marker = + dpctl::detail::submit_keep_alive_marker(q, depends); + + // 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::move(obj_vec), {marker}); + + *status = result_ok; + + return wrap(e_ptr.release()); + } catch (...) { + // no exception may escape into the calling Cython code + static constexpr int result_exception = 1; + *status = result_exception; + return nullptr; + } +} + +/*! + * @brief Call `release(data)` once the work using what it releases is done. + * + * Submits an empty kernel to `QRef` after `depERefs`, as `async_dec_ref` does, + * and assigns its event to the `sycl::event` behind `markerERef`. `release` is + * called with the GIL held, on `dpctl`'s thread or on one that drains. + * + * @return 0 on success, and 1 if nothing was deferred, in which case `release` + * is never called. + */ +inline int defer_release(DPCTLSyclQueueRef QRef, + DPCTLSyclEventRef *depERefs, + size_t nDepERefs, + void (*release)(void *), + void *data, + DPCTLSyclEventRef markerERef) +{ + using dpctl::syclinterface::unwrap; + + try { + sycl::queue &q = unwrap_queue(QRef); + sycl::event *marker = unwrap(markerERef); + if (!release || !marker) { + throw std::invalid_argument("A release and an event are required"); + } + + *marker = dpctl::detail::submit_keep_alive_marker( + q, unwrap_events(depERefs, nDepERefs)); + + dpctl::detail::local_deferred_releases().defer( + {*marker}, [release, data]() { release(data); }); + + return 0; + } catch (...) { + // no exception may escape into the calling Cython code + return 1; + } +} diff --git a/dpctl/_deferred_release_watcher.hpp b/dpctl/_deferred_release_watcher.hpp new file mode 100644 index 0000000000..a32195d85c --- /dev/null +++ b/dpctl/_deferred_release_watcher.hpp @@ -0,0 +1,620 @@ +//===--- _deferred_release_watcher.hpp - Runs deferred releases -----------===// +// +// 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 the list of deferred releases that `dpctl` owns and the +/// thread that runs them. +/// +//===----------------------------------------------------------------------===// + +#pragma once + +#ifdef _WIN32 +#ifndef NOMINMAX +#define NOMINMAX +#endif +#include +#endif + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "Python.h" + +#include "_deferred_releases.hpp" + +namespace dpctl +{ +namespace detail +{ + +inline void start_watcher_if_enabled(); + +/*! + * @brief The deferred releases submitted. + * + * Other modules reach it through `SyclQueue_DeferRelease`. The first release + * deferred starts `dpctl`'s thread. + */ +inline DeferredReleases &local_deferred_releases() +{ + // leaked: releasing what the list holds runs Python code, which a static + // destructor could not do once the interpreter has shut down + static DeferredReleases *instance = + new DeferredReleases(&start_watcher_if_enabled); + return *instance; +} + +/*! + * @brief Runs the deferred releases, on a thread of its own. + * + * Looks over the list for releases that have come due and takes the GIL to run + * them. + * + * The thread is started by the first release deferred, and ended by the + * `atexit` hook `dpctl` registers at import, or by the destructor in a process + * that never finalizes the interpreter. In either case, it stops running + * releases once shutdown has begun, leaving whatever is still held for the + * process to reclaim. + */ +class DeferredReleaseWatcher +{ +public: + /*! + * @brief Start the thread, and start it again whenever it is needed. + * + * Waits up to `stop_timeout` for a thread that a timed-out `stop` let go + * of, so expects the caller not to hold the GIL. + */ + void start() + { + { + std::lock_guard lock(mutex_); + + enabled_ = true; + if (thread_.joinable() || stopping_ || + finished().load(std::memory_order_acquire)) + { + start_nolock(); + return; + } + } + + wait_for_thread(stop_timeout); + + std::lock_guard lock(mutex_); + if (enabled_) { + start_nolock(); + } + } + + /*! + * @brief Start the thread if it is not running and has not been stopped. + */ + void start_if_enabled() + { + if (!finished().load(std::memory_order_acquire)) { + return; + } + + std::lock_guard lock(mutex_); + + if (enabled_) { + start_nolock(); + } + } + + /*! + * @brief End the thread, and keep it from being started until `start`. + * + * Waits up to `stop_timeout` for the release in progress, if any, and + * detaches the thread if it has not finished by then. The releases not + * yet run are left on the list. + * + * Expects the caller not to hold the GIL, which the thread may be waiting + * for. + */ + void stop() + { + std::thread thread; + { + std::lock_guard lock(mutex_); + + enabled_ = false; + if (!thread_.joinable()) { + return; + } + stopping_ = true; + thread = std::move(thread_); + } + + // not under `mutex_`, which a thread holding the GIL may be waiting on + end_without_hanging(thread, /* may_join */ true, stop_timeout); + + std::lock_guard lock(mutex_); + stopping_ = false; + // a `start` made while stopping + if (enabled_) { + start_nolock(); + } + } + + /*! + * @brief Make the thread startable in the child of a `fork`. + * + * The thread is not copied into the child, so its handle is dropped + * without being joined or detached, and the lock it may have held is + * replaced. The list is reset as well, and the thread is started by the + * next release deferred. + */ + void reset_after_fork() + { + new (&thread_) std::thread(); + new (&mutex_) std::mutex(); + stopping_ = false; + finished().store(true, std::memory_order_release); + + local_deferred_releases().reset_after_fork(); + } + + /*! + * @brief End the thread as the module's static objects are destroyed. + * + * Only ends the thread in a process that never finalizes the interpreter, + * as the `atexit` hook ends it while the interpreter is still up. + * + * The thread is waited for rather than joined on Windows, where a thread + * cannot exit without the loader lock that unloading a module holds. + */ + ~DeferredReleaseWatcher() + { + std::lock_guard lock(mutex_); + + enabled_ = false; + unloading().store(true, std::memory_order_release); + end_without_hanging(thread_, joining_at_unload_is_safe, abandon_after); + } + + DeferredReleaseWatcher(const DeferredReleaseWatcher &) = delete; + DeferredReleaseWatcher &operator=(const DeferredReleaseWatcher &) = delete; + +private: + friend DeferredReleaseWatcher &deferred_release_watcher(); + + DeferredReleaseWatcher() = default; + + using clock = DeferredReleases::clock; + + /*! + * @brief Whether the interpreter can still be called into. + */ + static bool interpreter_is_live() + { + const bool initialized = Py_IsInitialized(); +#if PY_VERSION_HEX < 0x30d0000 + return initialized && !_Py_IsFinalizing(); +#else + return initialized && !Py_IsFinalizing(); +#endif + } + + /*! + * @brief How long the thread sleeps between passes, at the least. + */ + static constexpr clock::duration min_poll_interval = + std::chrono::milliseconds(5); + + /*! + * @brief How many times a pass over the list its own duration buys. + * + * The sleep is stretched with the cost of a pass, which grows with the + * number of items, so the thread costs about a `1 / poll_share` of a core + * however many there are. + */ + static constexpr clock::duration::rep poll_share = 20; + + /*! + * @brief How long `stop` waits for the release in progress. + */ + static constexpr clock::duration stop_timeout = std::chrono::seconds(1); + + /*! + * @brief How long an exit waits for the thread before detaching it. + */ + static constexpr clock::duration abandon_after = + std::chrono::milliseconds(200); + + /*! + * @brief The thread's name in Python's `threading`. + */ + static constexpr const char *thread_name = "dpctl-deferred-releases"; + + /*! + * @brief Whether a thread may be joined as this module is unloaded. + * + * False on Windows, where a thread cannot exit without the loader lock that + * unloading holds. + */ +#ifdef _WIN32 + static constexpr bool joining_at_unload_is_safe = false; +#else + static constexpr bool joining_at_unload_is_safe = true; +#endif + + /*! + * @brief Start the thread, unless one is running or still finishing. + * + * A thread that ended on its own is joined and replaced, unless the + * interpreter is shutting down. Expects `mutex_` to be held. + */ + void start_nolock() + { + if (stopping_ || !finished().load(std::memory_order_acquire)) { + return; + } + + if (thread_.joinable()) { + // a new thread would end at once + if (!interpreter_is_live()) { + return; + } + thread_.join(); + } + + local_deferred_releases().resume_waiting(); + finished().store(false, std::memory_order_release); + + try { + thread_ = std::thread([]() { watch(); }); + } catch (...) { + // without the thread the releases are left to the drains, so + // failing to start one is not fatal + finished().store(true, std::memory_order_release); + } + } + + /*! + * @brief Stop `thread` and let go of it, waiting only `timeout` for it. + * + * A thread that does not report itself finished in time is detached, so + * that an exit cannot hang on a release that blocks, or on a thread the + * interpreter parked for asking for the GIL. Expects the caller not to + * hold the GIL. + */ + static void end_without_hanging(std::thread &thread, + bool may_join, + clock::duration timeout) + { + if (!thread.joinable()) { + return; + } + + local_deferred_releases().stop_waiting(); + + // waited for whether or not it may be joined: the report says the + // thread is done with this object, which is what makes detaching safe + const bool done = has_exited(thread) || wait_for_thread(timeout); + + if (may_join && done) { + thread.join(); + } + else { + thread.detach(); + } + } + + /*! + * @brief Whether no thread is running the loop. + * + * In static storage, so that a detached thread can report itself without + * reaching a destroyed object. + */ + static std::atomic &finished() + { + static std::atomic flag{true}; + return flag; + } + + /*! + * @brief Whether the module's static objects are being destroyed. + * + * The interpreter may then be up but not running, so the thread must not + * wait for the GIL. + */ + static std::atomic &unloading() + { + static std::atomic flag{false}; + return flag; + } + + /*! + * @brief Name the calling thread in Python's `threading`. + * + * Expects the caller to hold the GIL. Failing to is not reported. + */ + static void name_python_thread() + { + PyObject *threading = PyImport_ImportModule("threading"); + PyObject *current = + threading + ? PyObject_CallMethod(threading, "current_thread", nullptr) + : nullptr; + if (current) { + PyObject *name = PyUnicode_FromString(thread_name); + if (name) { + static_cast( + PyObject_SetAttrString(current, "name", name)); + Py_DECREF(name); + } + } + Py_XDECREF(current); + Py_XDECREF(threading); + PyErr_Clear(); + } + + /*! + * @brief Run the releases that come due, until the list is stopped, the + * interpreter begins shutting down, or the list throws. + * + * A thread ended by an exception is replaced by the next release deferred. + */ + static void watch() + { + std::unique_ptr>> due; + + try { + due = std::make_unique>>(); + run_due_until_stopped(*due); + } +#if defined(__GLIBCXX__) + catch (abi::__forced_unwind &) + { + // the interpreter ending this thread, which must not be stopped + static_cast(due.release()); + finished().store(true, std::memory_order_release); + throw; + } +#endif + catch (...) + { + // an exception escaping the thread would terminate the process. + // What was collected is leaked rather than destroyed, as the + // destructors of releases not run may need the GIL + static_cast(due.release()); + } + + finished().store(true, std::memory_order_release); + } + + /*! + * @brief The loop run by `watch`, collecting releases onto `due`. + * + * The Python thread state made for the first releases run is kept until + * the loop ends, so that the thread is one Python thread throughout. + */ + static void run_due_until_stopped(std::vector> &due) + { + DeferredReleases &releases = local_deferred_releases(); + clock::duration poll_interval = min_poll_interval; + + bool keeps_thread_state = false; + PyGILState_STATE kept = PyGILState_UNLOCKED; + + while (releases.wait_for_due(poll_interval)) { + const clock::time_point started = clock::now(); + const bool any = releases.collect_due(due); + const clock::duration looked_for = clock::now() - started; + + poll_interval = looked_for * poll_share; + if (poll_interval < min_poll_interval) { + poll_interval = min_poll_interval; + } + + if (!any) { + continue; + } + + // what was collected is not run, like everything else still held, + // and the thread state is left to the interpreter's shutdown + if (!interpreter_is_live()) { + releases.put_back(due); + return; + } + + const PyGILState_STATE gstate = PyGILState_Ensure(); + // shutdown may have begun while the GIL was waited for + const bool live = interpreter_is_live(); + try { + if (live) { + if (!keeps_thread_state) { + kept = PyGILState_Ensure(); + keeps_thread_state = true; + name_python_thread(); + } + releases.run_collected(due, /* until_stopped */ true); + } + } +#if defined(__GLIBCXX__) + catch (abi::__forced_unwind &) + { + // the GIL was dropped by the interpreter ending this thread + throw; + } +#endif + catch (...) + { + // `kept` first, so that the release deleting the thread state + // is the one dropping the GIL + if (keeps_thread_state) { + PyGILState_Release(kept); + } + PyGILState_Release(gstate); + throw; + } + PyGILState_Release(gstate); + + if (!live) { + releases.put_back(due); + return; + } + } + + // stopped by `stop`, which waits for this without the GIL + if (keeps_thread_state && + !unloading().load(std::memory_order_acquire) && + interpreter_is_live()) + { + const PyGILState_STATE gstate = PyGILState_Ensure(); + PyGILState_Release(kept); + PyGILState_Release(gstate); + } + } + + /*! + * @brief Whether `thread` no longer runs, without having reported so. + * + * Only known on Windows, where an exiting process kills its other threads + * before destroying the module's static objects. + */ + static bool has_exited(std::thread &thread) + { +#ifdef _WIN32 + return WaitForSingleObject(thread.native_handle(), 0) == WAIT_OBJECT_0; +#else + static_cast(thread); + return false; +#endif + } + + /*! + * @brief Whether the thread left its loop within `timeout`. + */ + static bool wait_for_thread(clock::duration timeout) + { + const clock::time_point deadline = clock::now() + timeout; + + while (!finished().load(std::memory_order_acquire)) { + if (clock::now() >= deadline) { + return false; + } + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + + return true; + } + + std::thread thread_; + std::mutex mutex_; + bool enabled_ = true; + bool stopping_ = false; +}; + +/*! + * @brief The thread running the deferred releases. + * + * Destroyed with the module's other static objects, which is what ends the + * thread when a process exits without having stopped it. + */ +inline DeferredReleaseWatcher &deferred_release_watcher() +{ + static DeferredReleaseWatcher instance; + return instance; +} + +/*! + * @brief Start `dpctl`'s thread if it is needed and has not been stopped. + */ +inline void start_watcher_if_enabled() +{ + deferred_release_watcher().start_if_enabled(); +} + +} // namespace detail +} // namespace dpctl + +/*! + * @brief Run the releases that have come due, dropping the references held. + * + * Expects the caller to hold the GIL, as a release runs arbitrary Python code. + * + * @return Whether there were any. + */ +inline bool drain_deferred_releases() +{ + try { + return dpctl::detail::local_deferred_releases().drain(); + } +#if defined(__GLIBCXX__) + catch (abi::__forced_unwind &) + { + // the interpreter ending this thread, which must not be stopped + throw; + } +#endif + catch (...) + { + // no exception may escape into the calling Cython code + return false; + } +} + +/*! + * @brief Start running deferred releases on a thread of `dpctl`'s. + */ +inline void start_deferred_release_watcher() +{ + try { + dpctl::detail::deferred_release_watcher().start(); + } catch (...) { + // no exception may escape into the calling Cython code + } +} + +/*! + * @brief Stop running deferred releases, waiting a bounded time for the thread. + * + * Expects the caller not to hold the GIL. + */ +inline void stop_deferred_release_watcher() +{ + try { + dpctl::detail::deferred_release_watcher().stop(); + } catch (...) { + // no exception may escape into the calling Cython code + } +} + +/*! + * @brief Make deferred releases work again in the child of a `fork`. + */ +inline void reset_deferred_releases_after_fork() +{ + try { + dpctl::detail::deferred_release_watcher().reset_after_fork(); + } catch (...) { + // no exception may escape into the calling Cython code + } +} diff --git a/dpctl/_deferred_releases.hpp b/dpctl/_deferred_releases.hpp new file mode 100644 index 0000000000..fd8b03e926 --- /dev/null +++ b/dpctl/_deferred_releases.hpp @@ -0,0 +1,403 @@ +//===--- _deferred_releases.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 +/// This file implements a list of releases of the objects used by offloaded +/// tasks, each deferred until the task using them has completed, and run by a +/// thread that `dpctl` owns. +/// +//===----------------------------------------------------------------------===// + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#if defined(__GLIBCXX__) +#include +#endif + +namespace dpctl +{ +namespace detail +{ + +/*! + * @brief Releases of objects used by offloaded tasks, held until they complete. + * + * A release runs on the thread `dpctl` keeps for looking over the list, or on a + * thread that calls `drain`. The events an item is held for say when the + * objects stopped being used, not when they were released, so nothing should + * wait for a release to happen. + * + * An item is released only once its events report complete, so nothing is freed + * while a task may still be using it. Whether to keep releasing as the + * interpreter shuts down is up to the caller: nothing here checks. + * + * Neither the list nor the items left on it are ever destroyed, so the events + * and allocations they hold are not released during static teardown, when the + * SYCL runtime may be gone. What is still held when a process ends is left for + * it to reclaim. + * + * Deferring runs no release, so a caller may hold whatever locks it likes. + * Whoever runs one has to be able to run Python code. + * + * A release drops references, but what that reclaims is up to the interpreter. + * On a free-threaded build a reference dropped by a thread that does not own + * the object need not run its destructor at once, so an allocation the object + * holds can be reclaimed later than the reference drop. + */ +class DeferredReleases +{ +public: + using clock = std::chrono::steady_clock; + + /*! + * @brief Defer `release` until every event in `depends` has completed. + * + * `release` must own what it releases, so move USM `shared_ptr` copies or + * `PyObject *` references into it. + */ + void defer(std::vector depends, std::function release) + { + push(Item{std::move(depends), std::move(release)}); + + if (on_defer_) { + try { + on_defer_(); + } catch (...) { + // the release is already deferred, so a failure is not reported + } + } + } + + /*! + * @brief Move the releases that have come due onto `due`. + * + * Takes them off the list, so the caller is responsible for running them. + * The events are queried without holding the list's lock, so a long list + * does not hold up `defer`. + * + * @return Whether anything was added to `due`. + */ + bool collect_due(std::vector> &due) + { + auto items = std::make_unique>(); + { + std::lock_guard lock(mutex_); + items->swap(deferred_); + } + + bool any = false; + bool any_left = false; + try { + const auto due_begin = std::partition( + items->begin(), items->end(), + [](const Item &item) { return !is_complete(item); }); + any = due_begin != items->end(); + + due.reserve(due.size() + (items->end() - due_begin)); + for (auto it = due_begin; it != items->end(); ++it) { + due.push_back(std::move(it->release)); + } + items->erase(due_begin, items->end()); + + std::lock_guard lock(mutex_); + // add what was deferred during the scan + items->insert(items->end(), + std::make_move_iterator(deferred_.begin()), + std::make_move_iterator(deferred_.end())); + deferred_.swap(*items); + any_left = !deferred_.empty(); + } catch (...) { + // leaked rather than destroyed, as a task may still be using what + // the items hold + static_cast(items.release()); + throw; + } + + // a waiter may have seen the list empty while it was taken off + if (any_left) { + deferred_or_stopped_.notify_all(); + } + + return any; + } + + /*! + * @brief Run and drop the releases collected in `due`. + * + * Expects the caller to have a Python thread state, as a release runs + * Python code, which is free to defer more releases and to call back in + * here. A release that does call back in runs nothing itself, so releases + * do not nest on one thread. + * + * If `until_stopped`, the releases not yet run once `stop_waiting` has + * been called are put back on the list instead. + */ + void run_collected(std::vector> &due, + bool until_stopped = false) + { + bool &running = running_here(); + const bool was_running = running; + running = true; + + for (std::size_t i = 0; i < due.size(); ++i) { + if (until_stopped && stop_waiting_.load(std::memory_order_acquire)) + { + put_back(due, i); + break; + } + + try { + due[i](); + } +#if defined(__GLIBCXX__) + catch (abi::__forced_unwind &) + { + // the interpreter ending this thread, which must not be stopped + throw; + } +#endif + catch (...) + { + // the rest must still run + } + } + due.clear(); + + running = was_running; + } + + /*! + * @brief Put the releases in `due` from `first` on back on the list. + * + * They are held for no events, so they run on the next pass. + */ + void put_back(std::vector> &due, + std::size_t first = 0) + { + for (std::size_t i = first; i < due.size(); ++i) { + push(Item{{}, std::move(due[i])}); + } + due.clear(); + } + + /*! + * @brief Run every release that has come due, on the calling thread. + * + * Expects the caller to have a Python thread state, and does nothing if it + * is itself called from a release. + * + * Makes one pass, leaving what a release defers, what a drain called from + * inside one skipped, and what another pass has taken off the list at the + * time, to the thread looking over the list. + * + * @return Whether there was anything to run. + */ + bool drain() + { + if (running_here()) { + return false; + } + + std::vector> due; + if (!collect_due(due)) { + return false; + } + + run_collected(due); + + return true; + } + + /*! + * @brief Wait for something that may have come due. + * + * Blocks while the list is empty, as nothing can come due until something + * is deferred, and otherwise for at most `poll_interval`: an item comes due + * on its own, with nothing to announce it. + * + * @return Whether waiting is still in order, and false once `stop_waiting` + * has been called. + */ + bool wait_for_due(clock::duration poll_interval) + { + std::unique_lock lock(mutex_); + + if (deferred_.empty()) { + deferred_or_stopped_.wait(lock, [this]() { + return stop_waiting_.load(std::memory_order_relaxed) || + !deferred_.empty(); + }); + } + else { + deferred_or_stopped_.wait_for(lock, poll_interval, [this]() { + return stop_waiting_.load(std::memory_order_relaxed); + }); + } + + return !stop_waiting_.load(std::memory_order_relaxed); + } + + /*! + * @brief End every wait for something to come due, and refuse further ones. + */ + void stop_waiting() + { + { + std::lock_guard lock(mutex_); + stop_waiting_.store(true, std::memory_order_release); + } + + deferred_or_stopped_.notify_all(); + } + + /*! + * @brief Allow waiting again, after a `stop_waiting`. + */ + void resume_waiting() + { + std::lock_guard lock(mutex_); + stop_waiting_.store(false, std::memory_order_release); + } + + /*! + * @brief Make the list usable in the child of a `fork`. + * + * Replaces the lock and condition variable, which a thread that does not + * exist in the child may have held, and leaks the parent's items, whose + * events belong to the parent. + */ + void reset_after_fork() + { + new (&mutex_) std::mutex(); + new (&deferred_or_stopped_) std::condition_variable(); + static_cast(new std::vector(std::move(deferred_))); + deferred_.clear(); + stop_waiting_.store(false, std::memory_order_release); + } + + DeferredReleases(const DeferredReleases &) = delete; + DeferredReleases &operator=(const DeferredReleases &) = delete; + ~DeferredReleases() = delete; + +private: + struct Item + { + std::vector depends; + std::function release; + }; + + /*! + * @brief The only place a list is created. + * + * Defined by `dpctl._sycl_queue`, which other modules defer to through + * `SyclQueue_DeferRelease`. + */ + friend DeferredReleases &local_deferred_releases(); + + /*! + * @brief Create a list that calls `on_defer` after every `defer`. + */ + explicit DeferredReleases(void (*on_defer)()) : on_defer_(on_defer) {} + + /*! + * @brief Add `item` to the list, waking a waiter if it was empty. + */ + void push(Item item) + { + bool was_empty = false; + { + std::lock_guard lock(mutex_); + was_empty = deferred_.empty(); + deferred_.push_back(std::move(item)); + } + + // only an empty list is waited on without a timeout + if (was_empty) { + deferred_or_stopped_.notify_all(); + } + } + + /*! + * @brief Whether a release is running on this thread. + */ + static bool &running_here() + { + static thread_local bool running = false; + return running; + } + + /*! + * @brief Whether every event in `item.depends` has completed. + * + * An event whose status cannot be read is reported as not complete, as it + * is no evidence that the task is done with what the release would free. + * The item is looked at again by every later pass, and is leaked if the + * status never becomes readable. + */ + 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; + } + + std::vector deferred_; + std::atomic stop_waiting_{false}; + std::mutex mutex_; + std::condition_variable deferred_or_stopped_; + + /*! + * @brief Called after every `defer`. + * + * `dpctl` uses it to start its thread on the first release deferred. + */ + void (*on_defer_)() = nullptr; +}; + +} // namespace detail +} // namespace dpctl 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..46465b610d 100644 --- a/dpctl/_sycl_queue.pxd +++ b/dpctl/_sycl_queue.pxd @@ -78,6 +78,7 @@ cdef public api class SyclQueue (_SyclQueue) [ cpdef SyclContext get_sycl_context(self) cpdef SyclDevice get_sycl_device(self) cdef DPCTLSyclQueueRef get_queue_ref(self) + cpdef SyclEvent keep_args_alive(self, object args, object depends=*) cpdef SyclEvent _submit_keep_args_alive( self, object args, @@ -138,3 +139,5 @@ cdef public api class RawKernelArg(_RawKernelArg) [ object PyRawKernelArgObject, type PyRawKernelArgType ]: pass + +cdef bint drain_releases() diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index 3fbf9c7bf6..6dd7324691 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -101,17 +101,32 @@ from cpython.ref cimport Py_INCREF, PyObject from libc.stdint cimport uint8_t, uint16_t, uint32_t, uint64_t from libc.stdlib cimport free, malloc +import atexit import collections.abc import logging +import os import struct import sys +import warnings -cdef extern from "_host_task_util.hpp": +cdef extern from "_async_dec_ref.hpp": + # defers the decrements without running any, so touches no Python object DPCTLSyclEventRef async_dec_ref( DPCTLSyclQueueRef, PyObject **, size_t, DPCTLSyclEventRef *, size_t, int * ) nogil + int defer_release( + DPCTLSyclQueueRef, DPCTLSyclEventRef *, size_t, + void (*)(void *) noexcept nogil, void *, DPCTLSyclEventRef + ) nogil + # runs the releases that have come due, and so arbitrary Python code: + # requires the GIL + bint drain_deferred_releases() + void start_deferred_release_watcher() nogil + # waits for the thread, which may be waiting for the GIL + void stop_deferred_release_watcher() nogil + void reset_deferred_releases_after_fork() nogil __all__ = [ @@ -1337,81 +1352,149 @@ cdef class SyclQueue(_SyclQueue): """ return self._queue_ref - cpdef SyclEvent _submit_keep_args_alive( - self, - object args, - list dEvents - ): - """ SyclQueue._submit_keep_args_alive(args, events) + cpdef SyclEvent keep_args_alive(self, object args, object depends=None): + """ SyclQueue.keep_args_alive(args, depends=None) - Keeps objects in ``args`` alive until tasks associated with events - complete. + Keeps objects in ``args`` alive until the work on this queue that uses + them completes. Args: - args(object): - Python object to keep alive. - Typically a tuple with arguments to offloaded tasks - events(Tuple[dpctl.SyclEvent]): - Gating events. - The list or tuple of events associated with tasks - working on Python objects collected in ``args``. + args (object): + Python object to keep alive, typically a tuple of the arguments + passed to an offloaded task. + depends (Iterable[dpctl.SyclEvent], optional): + Gating events. On an out-of-order queue these decide when the + objects stop being used, so every task using them must be + covered. Default: ``None``. + Returns: - dpctl.SyclEvent - The event associated with the submission of host task. + dpctl.SyclEvent: + An event that completes once ``args`` have stopped being used. + Its completion does not mean that ``args`` were released, only + that releasing them has come due. - Increments reference count of ``args`` and schedules asynchronous - ``host_task`` to decrement the count once dependent events are - complete. + Raises: + TypeError: + If ``depends`` is not an iterable of :class:`dpctl.SyclEvent`. + RuntimeError: + If the release could not be scheduled. The reference taken on + ``args`` is then leaked deliberately, as the work using them + may still be running. + + Increments the reference count of ``args`` and defers the matching + decrement until an empty kernel, submitted to this queue after + ``depends``, has completed. On an in-order queue it is ordered after + everything submitted before it, so ``depends`` may be omitted. + + It is suggested to call this on the queue running the work that uses + ``args``: the kernel is ordered only against work submitted to the queue + it goes to, so on any other queue ``depends`` alone gates the release. + + :Example: + .. code-block:: python + + import dpctl + + q = dpctl.SyclQueue() + e = q.submit_async(kernel, [x_usm], [n]) + q.keep_args_alive((x_usm,), [e]) + + # an in-order queue needs no event + q_in_order = dpctl.SyclQueue(property="in_order") + q_in_order.submit_async(kernel, [y_usm], [n]) + q_in_order.keep_args_alive((y_usm,)) .. 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. + On a free-threaded build, the thread dropping the reference does not + own the object, so its ``__del__`` may run later than the decrement. """ - cdef size_t nDE = len(dEvents) + cdef size_t nDE = 0 cdef DPCTLSyclEventRef *depEvents = NULL + cdef DPCTLSyclEventRef ERef = NULL + cdef DPCTLSyclQueueRef QRef = self.get_queue_ref() cdef PyObject *args_raw = NULL - cdef DPCTLSyclEventRef htERef = NULL cdef int status = -1 + cdef tuple dep_events = () + + if depends is not None: + # a copy, so that its length is that of what it iterates over + dep_events = tuple(depends) + nDE = len(dep_events) - # Create the array of dependent events if any if nDE > 0: depEvents = ( malloc(nDE*sizeof(DPCTLSyclEventRef)) ) if not depEvents: raise MemoryError() - else: - for idx, de in enumerate(dEvents): - if isinstance(de, SyclEvent): - depEvents[idx] = (de).get_event_ref() - else: - free(depEvents) - raise TypeError( - "A sequence of dpctl.SyclEvent is expected" - ) + for idx, de in enumerate(dep_events): + if isinstance(de, SyclEvent): + depEvents[idx] = (de).get_event_ref() + else: + free(depEvents) + raise TypeError( + "An iterable of dpctl.SyclEvent is expected" + ) # increment reference counts to list of arguments Py_INCREF(args) - - # schedule decrement args_raw = args - htERef = async_dec_ref( - self.get_queue_ref(), - &args_raw, 1, - depEvents, nDE, &status - ) + # schedule decrement + with nogil: + ERef = async_dec_ref( + QRef, &args_raw, 1, depEvents, nDE, &status + ) free(depEvents) - if (status != 0): - with nogil: - DPCTLEvent_Wait(htERef) - DPCTLEvent_Delete(htERef) - raise RuntimeError("Could not submit keep_args_alive host_task") + if status != 0: + # nothing will drop the reference taken above, and it is leaked + # deliberately: the work in depends, and whatever this queue has + # already been given, may still be using args + raise RuntimeError("Could not schedule keep_args_alive") + + return SyclEvent._create(ERef) + + cpdef SyclEvent _submit_keep_args_alive( + self, + object args, + list dEvents + ): + """ SyclQueue._submit_keep_args_alive(args, events) - return SyclEvent._create(htERef) + Keeps objects in ``args`` alive until tasks associated with events + complete. + + Args: + args(object): + Python object to keep alive. + Typically a tuple with arguments to offloaded tasks + events(Tuple[dpctl.SyclEvent]): + Gating events. + The list or tuple of events associated with tasks + working on Python objects collected in ``args``. + Returns: + dpctl.SyclEvent + An event for an empty kernel submitted to this queue after + ``events``. It says when ``args`` stop being used, not when they + were released: see :meth:`dpctl.SyclQueue.keep_args_alive` for + when the reference taken is dropped. + + .. deprecated:: 0.24.0 + The method is deprecated, use + :meth:`dpctl.SyclQueue.keep_args_alive` instead, which this now + calls and which behaves identically. + """ + warnings.warn( + "dpctl.SyclQueue._submit_keep_args_alive is deprecated and will " + "be removed in a future release. Use " + "dpctl.SyclQueue.keep_args_alive instead.", + DeprecationWarning, + # a cpdef method has no frame, so the caller's is the first + stacklevel=1, + ) + + return self.keep_args_alive(args, dEvents) cpdef SyclEvent submit_async( self, @@ -1451,7 +1534,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`. + :meth:`dpctl.SyclQueue.keep_args_alive`. Raises: ValueError: @@ -2292,6 +2375,83 @@ cdef api SyclQueue SyclQueue_Make(DPCTLSyclQueueRef QRef): cdef DPCTLSyclQueueRef copied_QRef = DPCTLQueue_Copy(QRef) return SyclQueue._create(copied_QRef) + +cdef api int SyclQueue_DeferRelease( + DPCTLSyclQueueRef QRef, + DPCTLSyclEventRef *depERefs, + size_t nDepERefs, + void (*release)(void *) noexcept nogil, + void *data, + DPCTLSyclEventRef markerERef +) noexcept nogil: + """ + C-API function to call ``release(data)`` once an empty kernel, submitted + to ``QRef`` after ``depERefs``, has completed. The kernel's event is + assigned to the ``sycl::event`` behind ``markerERef``. + + ``release`` is called with the GIL held. Returns 0 on success, and 1 if + nothing was deferred, in which case ``release`` is never called. + """ + return defer_release( + QRef, depERefs, nDepERefs, release, data, markerERef + ) + + +cdef bint drain_releases(): + """Run the releases deferred for offloaded tasks that have completed. + + Returns whether there were any. + + Expects the caller to hold the GIL, as a release runs arbitrary Python code. + """ + return drain_deferred_releases() + + +def _drain_deferred_releases(): + """_drain_deferred_releases() + + Release the objects held for offloaded tasks that have completed, dropping + the references taken for them. + """ + drain_releases() + + +def _start_deferred_release_watcher(): + """_start_deferred_release_watcher() + + Run deferred releases on dpctl's own thread, starting it if it is stopped. + """ + with nogil: + start_deferred_release_watcher() + + +def _stop_deferred_release_watcher(): + """_stop_deferred_release_watcher() + + Stop dpctl's thread, waiting up to a second for the release in progress, + and keep it stopped until it is started again. What is still held stays + held. + """ + with nogil: + stop_deferred_release_watcher() + + +def _reset_deferred_releases_after_fork(): + """_reset_deferred_releases_after_fork() + + Make deferred releases work in the child of a fork, which has no copy of + dpctl's thread. + """ + reset_deferred_releases_after_fork() + + +# runs while the interpreter is still up so the thread can be waited for +atexit.register(_stop_deferred_release_watcher) + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_reset_deferred_releases_after_fork) + + cdef class _WorkGroupMemory: def __dealloc__(self): if(self._mem_ref): 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/dpctl4pybind11.hpp b/dpctl/apis/include/dpctl4pybind11.hpp index ecfc85596f..fc8dd9ec8e 100644 --- a/dpctl/apis/include/dpctl4pybind11.hpp +++ b/dpctl/apis/include/dpctl4pybind11.hpp @@ -46,6 +46,24 @@ namespace dpctl namespace detail { +/*! + * @brief Whether the interpreter can still be called into. + * + * A destructor that may run at shutdown, such as the one releasing + * `dpctl_capi`'s objects, must check this before touching Python: asking for + * the GIL while finalization runs may not return, and once finalization is over + * the call is not safe at all. + */ +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: @@ -76,6 +94,12 @@ class dpctl_capi DPCTLSyclQueueRef (*SyclQueue_GetQueueRef_)(PySyclQueueObject *); PySyclQueueObject *(*SyclQueue_Make_)(DPCTLSyclQueueRef); + int (*SyclQueue_DeferRelease_)(DPCTLSyclQueueRef, + DPCTLSyclEventRef *, + size_t, + void (*)(void *), + void *, + DPCTLSyclEventRef); // memory DPCTLSyclUSMRef (*Memory_GetUsmPointer_)(Py_MemoryObject *); @@ -173,15 +197,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; } } @@ -203,10 +219,11 @@ class dpctl_capi SyclContext_GetContextRef_(nullptr), SyclContext_Make_(nullptr), SyclEvent_GetEventRef_(nullptr), SyclEvent_Make_(nullptr), SyclQueue_GetQueueRef_(nullptr), SyclQueue_Make_(nullptr), - Memory_GetUsmPointer_(nullptr), Memory_GetOpaquePointer_(nullptr), - Memory_GetContextRef_(nullptr), Memory_GetQueueRef_(nullptr), - Memory_GetNumBytes_(nullptr), Memory_Make_(nullptr), - SyclKernel_GetKernelRef_(nullptr), SyclKernel_Make_(nullptr), + SyclQueue_DeferRelease_(nullptr), Memory_GetUsmPointer_(nullptr), + Memory_GetOpaquePointer_(nullptr), Memory_GetContextRef_(nullptr), + Memory_GetQueueRef_(nullptr), Memory_GetNumBytes_(nullptr), + Memory_Make_(nullptr), SyclKernel_GetKernelRef_(nullptr), + SyclKernel_Make_(nullptr), SyclKernelBundle_GetKernelBundleRef_(nullptr), SyclKernelBundle_Make_(nullptr), default_sycl_queue_{}, default_usm_memory_{}, as_usm_memory_{} @@ -250,6 +267,7 @@ class dpctl_capi // SyclQueue API this->SyclQueue_GetQueueRef_ = SyclQueue_GetQueueRef; this->SyclQueue_Make_ = SyclQueue_Make; + this->SyclQueue_DeferRelease_ = SyclQueue_DeferRelease; // dpctl.memory API this->Memory_GetUsmPointer_ = Memory_GetUsmPointer; @@ -294,6 +312,7 @@ class dpctl_capi dpctl_capi &operator=(dpctl_capi &&) = default; }; // struct dpctl_capi + } // namespace detail } // namespace dpctl @@ -795,74 +814,104 @@ struct ManagedMemory } }; +/*! + * @brief Call `release(data)` once the work covered by `depends` completes. + * + * Submits an empty kernel to `q` after `depends`, and returns its event, which + * is what the release waits for. `release` is called with the GIL held. + * + * Throws if nothing was deferred, in which case `release` is never called. + */ +inline sycl::event defer_release(sycl::queue &q, + const std::vector &depends, + void (*release)(void *), + void *data) +{ + std::vector dep_refs; + dep_refs.reserve(depends.size()); + for (const sycl::event &e : depends) { + dep_refs.push_back( + reinterpret_cast(const_cast(&e))); + } + + sycl::event marker; + const auto &api = ::dpctl::detail::dpctl_capi::get(); + if (api.SyclQueue_DeferRelease_( + reinterpret_cast(&q), dep_refs.data(), + dep_refs.size(), release, data, + reinterpret_cast(&marker)) != 0) + { + throw std::runtime_error("Could not defer the release"); + } + + return marker; +} + +/*! + * @brief What `keep_args_alive` holds for the work using its objects. + */ +template struct held_args +{ + std::size_t n_usm_owners = 0; + std::array, num> usm_owners{}; + std::size_t n_objects = 0; + std::array objects{}; + + /*! + * @brief Drop what `self` holds, and delete it. + */ + static void release(void *self) + { + std::unique_ptr held(static_cast(self)); + + for (std::size_t i = 0; i < held->n_usm_owners; ++i) { + held->usm_owners[i].reset(); + } + for (std::size_t i = 0; i < held->n_objects; ++i) { + held->objects[i].dec_ref(); + } + } +}; + } // end of namespace detail +/*! + * @brief Keeps `py_objs` alive until the work covered by `depends` completes. + * + * Returns an event for an empty kernel submitted to `q` after `depends`, which + * is what the release waits for. On an in-order queue the kernel is also + * ordered after the work already submitted there. + * + * `q` is intended to be the queue running the work that uses `py_objs`. + * + * Expects the caller to hold the GIL. Throws if the release could not be + * scheduled, having released nothing. + */ template sycl::event keep_args_alive(sycl::queue &q, const py::object (&py_objs)[num], const std::vector &depends = {}) { - std::size_t n_objects_held = 0; - std::array, num> shp_arr{}; + using held_t = detail::held_args; - std::size_t n_usm_owners_held = 0; - std::array, num> shp_usm{}; + // leaked if anything below throws, as the work in `depends` may still be + // using what it holds + held_t *held = new held_t{}; for (std::size_t i = 0; i < num; ++i) { const auto &py_obj_i = py_objs[i]; if (detail::ManagedMemory::is_usm_managed_by_shared_ptr(py_obj_i)) { - const auto &shp = + held->usm_owners[held->n_usm_owners] = detail::ManagedMemory::extract_shared_ptr(py_obj_i); - shp_usm[n_usm_owners_held] = shp; - ++n_usm_owners_held; + ++held->n_usm_owners; } else { - shp_arr[n_objects_held] = std::make_shared(py_obj_i); - shp_arr[n_objects_held]->inc_ref(); - ++n_objects_held; + held->objects[held->n_objects] = py_obj_i.inc_ref(); + ++held->n_objects; } } - 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_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; - - for (std::size_t i = 0; i < n_objects_held; ++i) { - shp_arr[i]->dec_ref(); - } - }); - }); - } - - return host_task_ev; + return detail::defer_release(q, depends, &held_t::release, held); } /*! @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..c81629d79d 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_releases 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,25 @@ 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: + # the objects held for tasks that have completed may account + # for enough memory for a second attempt to succeed, and + # dpctl's thread may not have got to them yet + if drain_releases(): + 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 2b7896e868..a2d4be191a 100644 --- a/dpctl/tests/test_sycl_compiler.py +++ b/dpctl/tests/test_sycl_compiler.py @@ -318,10 +318,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]) + q.keep_args_alive([x_usm], [e3]) e3.wait() - ht_e.wait() assert np.all(y == 43) @@ -358,10 +357,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]) + q.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 e173035ae6..9c57186d81 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]) + q.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..0abe97b8bf 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,50 @@ 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): + q.keep_args_alive((usm,), [None]) + + +def test_keep_args_alive_depends_length_differs_from_iteration(): + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + + class Events: + def __init__(self, events): + self._events = events + + def __len__(self): + return 1 + + def __iter__(self): + return iter(self._events) + + usm = dpctl.memory.MemoryUSMDevice(4096, queue=q) + events = [q.submit_barrier() for _ in range(4)] + q.keep_args_alive((usm,), Events(events)).wait() + q.keep_args_alive((usm,), (e for e in events)).wait() + + +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) as record: + ht_ev = q._submit_keep_args_alive((usm,), []) + ht_ev.wait() + # attributed to the caller + assert record[0].filename == __file__ diff --git a/dpctl/tests/test_sycl_timer.py b/dpctl/tests/test_sycl_timer.py index 9a99b1f997..6ce8c66f2e 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) + q.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) + q.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) + q.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) + q.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..7374e6c809 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -16,10 +16,72 @@ """Defines unit test cases for utility functions.""" +import subprocess +import sys +import sysconfig +import textwrap +import threading +import time +import warnings + import pytest import dpctl +import dpctl.memory import dpctl.utils +from dpctl._sycl_queue import ( + _drain_deferred_releases, + _start_deferred_release_watcher, + _stop_deferred_release_watcher, +) + +_free_threaded = bool(sysconfig.get_config_var("Py_GIL_DISABLED")) + + +@pytest.fixture +def stopped_release_watcher(): + """Leaves the deferred releases to the test, rather than to dpctl's thread. + + A test that expects an object to still be held, or to be released on the + thread it runs on, has to be the only one running them. + """ + _stop_deferred_release_watcher() + yield + _start_deferred_release_watcher() + + +def _queue(**kwargs): + try: + return dpctl.SyclQueue(**kwargs) + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + + +def _submit_long_work(q, usm): + """Copy into `usm` repeatedly, returning the event of the last copy.""" + # the first keep_args_alive on a device builds its kernel, which can take + # longer than the copies + q.keep_args_alive(()).wait() + host_buf = bytearray(usm.nbytes) + ev = None + for _ in range(8): + ev = q.copy_async(usm, host_buf, usm.nbytes, [ev] if ev else None) + q.keep_args_alive((host_buf,), [ev]) + return ev + + +def _complete(ev): + return ev.execution_status == dpctl.event_status_type.complete + + +class _Sentinel: + """Appends to `released` when it is destroyed.""" + + def __init__(self, released): + self._released = released + + def __del__(self): + self._released.append(True) @pytest.mark.filterwarnings("ignore:.*:RuntimeWarning") @@ -73,13 +135,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 +155,333 @@ def test_order_manager(): _passed = True finally: assert _passed + + +def test_order_manager_waits_for_cleanup_events(): + q = _queue() + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + + usm = dpctl.memory.MemoryUSMDevice(32 * 1024 * 1024, queue=q) + copy_ev = _submit_long_work(q, usm) + cleanup_ev = q.keep_args_alive((usm,), [copy_ev]) + # only the cleanup event is recorded, so waiting on the manager can only + # wait for the copies through it + _mngr.add_cleanup_event(cleanup_ev) + if _mngr.num_cleanup_events == 0: + pytest.skip("The copies completed before the event was recorded") + _mngr.wait() + assert _complete(copy_ev) + assert _mngr.num_cleanup_events == 0 + _som.clear() + + +def test_order_manager_wait_drops_references(stopped_release_watcher): + q = _queue() + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + + released = [] + args = (_Sentinel(released),) + ev = q.keep_args_alive(args) + del args + ev.wait() + assert not released, "dpctl's thread is stopped, so nothing has run it" + + # 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_wait_with_release_watcher_running(): + q = _queue() + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + + n = 200 + released = [] + for _ in range(n): + args = (_Sentinel(released),) + _mngr.add_cleanup_event(q.keep_args_alive(args)) + del args + # races the thread for the same releases + _mngr.wait() + + # what the thread took up is released by it rather than by `wait` + deadline = time.monotonic() + 10 + while len(released) < n and time.monotonic() < deadline: + time.sleep(0.01) + assert len(released) == n + _som.clear() + + +def test_keep_args_alive_on_in_order_queue_waits_for_earlier_work( + stopped_release_watcher, +): + q = _queue(property="in_order") + + usm = dpctl.memory.MemoryUSMDevice(32 * 1024 * 1024, queue=q) + copy_ev = _submit_long_work(q, usm) + + released = [] + args = (_Sentinel(released),) + # no events: the queue orders the release after the copies + q.keep_args_alive(args) + del args + + _drain_deferred_releases() + if not _complete(copy_ev): + assert not released + q.wait() + _drain_deferred_releases() + assert released == [True] + + +def test_failed_allocation_drains_releases(stopped_release_watcher): + q = _queue() + + released = [] + args = (_Sentinel(released),) + q.keep_args_alive(args).wait() + del args + assert not released + + with pytest.raises(dpctl.memory.USMAllocationError): + dpctl.memory.MemoryUSMDevice(1 << 62, queue=q) + assert released == [True] + + +def test_order_manager_host_task_api(): + q = _queue() + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + + # deprecated in the documentation only, as dpnp still calls them + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert isinstance(_mngr.num_host_task_events, int) + 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.wait() + _som.clear() + + +@pytest.mark.skipif( + _free_threaded, + reason="The thread owning the object runs its __del__", +) +def test_release_watcher_is_one_named_python_thread(): + q = _queue() + + ran_on = [] + + class Records: + def __del__(self): + ran_on.append(threading.current_thread()) + + def watcher_threads(): + main = threading.main_thread() + return [t for t in ran_on if t is not main] + + # the caller drops the last reference instead if the release runs before + # the call returns, so a release is repeated until two ran on the thread + for _ in range(50): + if len(watcher_threads()) >= 2: + break + n = len(ran_on) + q.keep_args_alive((Records(),)).wait() + deadline = time.monotonic() + 5 + while len(ran_on) == n and time.monotonic() < deadline: + time.sleep(0.01) + + seen = watcher_threads() + if len(seen) < 2: + pytest.skip("The releases did not run on dpctl's thread") + assert seen[0] is seen[1] + assert seen[0].name == "dpctl-deferred-releases" + + +def _run_python(script, timeout): + """Run `script` in a fresh interpreter and return what it printed. + + Skips the test if the script prints "skip". + """ + proc = subprocess.run( + [sys.executable, "-c", textwrap.dedent(script)], + capture_output=True, + text=True, + timeout=timeout, + ) + assert proc.returncode == 0, proc.stderr + out = proc.stdout.strip() + if out.startswith("skip"): + pytest.skip(out) + return out + + +# a release that blocks until `unblock` is set, if dpctl's thread runs it +_BLOCKING_RELEASE = """ +import sysconfig +import threading +import time + +if sysconfig.get_config_var("Py_GIL_DISABLED"): + # the thread owning the object runs its __del__, not dpctl's thread + print("skip: free-threaded build") + raise SystemExit + +import dpctl +from dpctl._sycl_queue import ( + _start_deferred_release_watcher, + _stop_deferred_release_watcher, +) + +entered = threading.Event() +unblock = threading.Event() +main_thread = threading.get_ident() + + +class Blocks: + def __del__(self): + # on a free-threaded build the owning thread runs this instead + if threading.get_ident() == main_thread: + return + entered.set() + unblock.wait(30) + + +try: + q = dpctl.SyclQueue() +except dpctl.SyclQueueCreationError: + print("skip: no default queue") + raise SystemExit + +args = (Blocks(),) +q.keep_args_alive(args).wait() +del args +if not entered.wait(5): + print("skip: the release did not run on dpctl's thread") + raise SystemExit +""" + + +def test_exit_does_not_wait_out_a_blocking_release(): + # the release blocks for 30 s unless the exit gives up on it + _run_python(_BLOCKING_RELEASE, timeout=20) + + +def test_stop_does_not_wait_out_a_blocking_release(): + out = _run_python( + _BLOCKING_RELEASE + """ +started = time.monotonic() +_stop_deferred_release_watcher() +print(time.monotonic() - started) +unblock.set() +""", + timeout=20, + ) + assert float(out) < 5 + + +def test_start_during_stop_does_not_deadlock(): + out = _run_python( + _BLOCKING_RELEASE + """ +stopper = threading.Thread(target=_stop_deferred_release_watcher) +stopper.start() +time.sleep(0.1) +starter = threading.Thread(target=_start_deferred_release_watcher) +starter.start() +time.sleep(0.1) +unblock.set() +stopper.join(10) +starter.join(10) +print("hung" if stopper.is_alive() or starter.is_alive() else "ok") +""", + timeout=30, + ) + assert out == "ok" + + +def test_start_after_timed_out_stop_runs_releases(): + # the stop gives up on the blocking release after a second + out = _run_python( + _BLOCKING_RELEASE + """ +_stop_deferred_release_watcher() +starter = threading.Thread(target=_start_deferred_release_watcher) +starter.start() + +released = [] + + +class Sentinel: + def __del__(self): + released.append(True) + + +# deferred while the thread the stop let go of is still running +args = (Sentinel(),) +q.keep_args_alive(args).wait() +del args +unblock.set() +starter.join(10) + +deadline = time.monotonic() + 5 +while not released and time.monotonic() < deadline: + time.sleep(0.01) +print("released" if released else "held") +""", + timeout=30, + ) + assert out == "released" + + +@pytest.mark.skipif( + not sys.platform.startswith("linux"), reason="Requires fork" +) +def test_keep_args_alive_releases_in_forked_child(): + out = _run_python( + """ +import multiprocessing +import time + +import dpctl +from dpctl._sycl_queue import _start_deferred_release_watcher + + +def work(_): + released = [] + + class Sentinel: + def __del__(self): + released.append(True) + + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + return None + args = (Sentinel(),) + q.keep_args_alive(args).wait() + del args + deadline = time.monotonic() + 5 + while not released and time.monotonic() < deadline: + time.sleep(0.01) + return bool(released) + + +if __name__ == "__main__": + # the child inherits the state of a running thread, but not the thread + _start_deferred_release_watcher() + with multiprocessing.get_context("fork").Pool(2) as pool: + results = pool.map(work, range(4)) + if None in results: + print("skip: no default queue in the child") + else: + print(sum(results)) +""", + timeout=120, + ) + assert out == "4" diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index 7c66d6bcc8..8820ff81ad 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -4,7 +4,7 @@ from collections import defaultdict from .._sycl_event import SyclEvent -from .._sycl_queue import SyclQueue +from .._sycl_queue import SyclQueue, _drain_deferred_releases from ._seq_order_keeper import _OrderManager @@ -12,6 +12,12 @@ 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 with + :meth:`add_cleanup_event`: they are waited on, but never become + dependencies of later tasks. """ def __init__(self): @@ -22,7 +28,7 @@ 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): _local = self._state @@ -37,10 +43,34 @@ 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): + 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 +79,12 @@ def num_submitted_events(self): @property def host_task_events(self): + 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 +93,13 @@ def submitted_events(self): def wait(self): _local = self._state - return _local.wait() + res = _local.wait() + # the events are complete, so what was held for those tasks is released + # here, unless dpctl's thread has already taken it to release; a release + # runs Python code, which finalization rules out + if not sys.is_finalizing(): + _drain_deferred_releases() + 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..0208a883e1 100644 --- a/examples/python/using_order_manager.py +++ b/examples/python/using_order_manager.py @@ -30,16 +30,19 @@ 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 - garbage-collected while the device is still reading/writing. + ``q.keep_args_alive`` prevents the buffer and the target from being + garbage-collected while the device is still reading/writing. The event it + returns gates their release, and is tracked as a cleanup event so that + ``om.wait()`` waits for it without later tasks depending on it. """ n = usm_buf.nbytes data = np.full(n, fill_byte, dtype=np.uint8) 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) + cleanup_ev = q.keep_args_alive((usm_buf, data), [comp_ev]) + om.add_event(comp_ev) + om.add_cleanup_event(cleanup_ev) return comp_ev @@ -101,8 +104,9 @@ 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) + cleanup_ev = q.keep_args_alive((usm_chunk, usm_data), [comp_ev]) + child_om.add_event(comp_ev) + child_om.add_cleanup_event(cleanup_ev) child_om.wait() return usm_chunk @@ -120,8 +124,9 @@ 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) + cleanup_ev = q.keep_args_alive((part, child_buf), [comp_ev]) + main_om.add_event(comp_ev) + main_om.add_cleanup_event(cleanup_ev) result_parts.append(part) main_om.wait() @@ -142,6 +147,8 @@ def explicit_event_passing(): Each child thread performs an async fill and gives its tracked events. The main thread adds those events to its own order manager so that a subsequent ``memcpy_async`` depends on the child work completing first. + The child's cleanup events are handed over as such, so the main thread + waits for them without making later tasks depend on them. """ nbytes = 1024 q = dpctl.SyclQueue() @@ -154,7 +161,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, child_om.cleanup_events with concurrent.futures.ThreadPoolExecutor( max_workers=n_threads @@ -162,15 +169,16 @@ 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: + collected_cleanup_events = [] + for buf, comp_events, cleanup_events in futures_results: child_buffers.append(buf) - collected_ht_events.extend(ht_events) collected_comp_events.extend(comp_events) + collected_cleanup_events.extend(cleanup_events) main_om = SequentialOrderManager[q] - main_om.add_event_pair(collected_ht_events, collected_comp_events) + main_om.add_event(collected_comp_events) + main_om.add_cleanup_event(collected_cleanup_events) results = [] for buf in child_buffers: @@ -178,8 +186,9 @@ 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) + cleanup_ev = q.keep_args_alive((out, buf), [comp_ev]) + main_om.add_event(comp_ev) + main_om.add_cleanup_event(cleanup_ev) results.append(out) main_om.wait()