From e6f590317d9c1e8f077866138b3d61e140aca449 Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 2 Nov 2026 00:14:31 -0800 Subject: [PATCH 1/4] add SyclQueue.get_last_event Wraps ext_oneapi_get_last_event() from sycl_ext_oneapi_in_order_queue_events, with DPCTLQueue_GetLastEvent and DPCTLQueue_GetLastEvent_Available in the C-API. Builds without the extension report it as unavailable. --- CHANGELOG.md | 2 + dpctl/_backend.pxd | 2 + dpctl/_sycl_queue.pyx | 44 +++++++++++ dpctl/tests/test_sycl_queue.py | 31 ++++++++ .../dpctl_sycl_queue_interface.h | 23 ++++++ .../source/dpctl_sycl_queue_interface.cpp | 41 +++++++++++ .../tests/test_sycl_queue_interface.cpp | 73 +++++++++++++++++++ 7 files changed, 216 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index f660fa239b..0eb677fe7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * 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) * Added `dpctl::utils::defer_release` to `dpctl4pybind11.hpp`, which calls a release callback once the given events have completed [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) +* Added `dpctl.SyclQueue.get_last_event` and `dpctl.SyclQueue.is_get_last_event_available`, and the `DPCTLQueue_GetLastEvent` and `DPCTLQueue_GetLastEvent_Available` C-API functions, wrapping the DPC++ `sycl_ext_oneapi_in_order_queue_events` extension [gh-2299](https://github.com/IntelPython/dpctl/pull/2299) + ### 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) * `import_dpctl()` now returns 0 on success and -1 with a Python exception set on failure, and `dpctl4pybind11.hpp` raises that exception rather than failing on the first C-API call [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) diff --git a/dpctl/_backend.pxd b/dpctl/_backend.pxd index 3708ea9130..90b25e8b07 100644 --- a/dpctl/_backend.pxd +++ b/dpctl/_backend.pxd @@ -756,6 +756,8 @@ cdef extern from "syclinterface/dpctl_sycl_queue_interface.h": size_t Count, int Advice) cdef bool DPCTLQueue_IsInOrder(const DPCTLSyclQueueRef QRef) + cdef bool DPCTLQueue_GetLastEvent_Available() + cdef DPCTLSyclEventRef DPCTLQueue_GetLastEvent(const DPCTLSyclQueueRef QRef) cdef DPCTLSyclEventRef DPCTLQueue_SubmitBarrier( const DPCTLSyclQueueRef QRef) cdef DPCTLSyclEventRef DPCTLQueue_SubmitBarrierForEvents( diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index 6dd7324691..07fcd0cf20 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -53,6 +53,8 @@ from ._backend cimport ( # noqa: E211 DPCTLQueue_GetBackend, DPCTLQueue_GetContext, DPCTLQueue_GetDevice, + DPCTLQueue_GetLastEvent, + DPCTLQueue_GetLastEvent_Available, DPCTLQueue_HasEnableProfiling, DPCTLQueue_Hash, DPCTLQueue_IsInOrder, @@ -2330,6 +2332,48 @@ cdef class SyclQueue(_SyclQueue): return SyclEvent._create(ERef) + @staticmethod + def is_get_last_event_available(): + """Returns ``True`` if dpctl was built with a compiler that supports + the DPC++ ``sycl_ext_oneapi_in_order_queue_events`` extension used by + :meth:`dpctl.SyclQueue.get_last_event`. + """ + return DPCTLQueue_GetLastEvent_Available() + + def get_last_event(self): + """ + get_last_event() + + Returns the event of the last command submitted to an in-order queue. + + Wraps ``sycl::queue::ext_oneapi_get_last_event()`` from the + experimental DPC++ ``sycl_ext_oneapi_in_order_queue_events`` + extension. + + Returns: + dpctl.SyclEvent | None: + Event of the last submitted command, or ``None`` if no + command was submitted to the queue. + + Raises: + RuntimeError: + If the extension is not available, see + :meth:`dpctl.SyclQueue.is_get_last_event_available`. + ValueError: + If the queue is not in-order. + """ + cdef DPCTLSyclEventRef ERef = NULL + if not DPCTLQueue_GetLastEvent_Available(): + raise RuntimeError( + "sycl_ext_oneapi_in_order_queue_events extension not available" + ) + if not DPCTLQueue_IsInOrder(self._queue_ref): + raise ValueError("get_last_event requires an in-order queue") + ERef = DPCTLQueue_GetLastEvent(self._queue_ref) + if ERef is NULL: + return None + return SyclEvent._create(ERef) + @property def name(self): """Returns the device name for the device diff --git a/dpctl/tests/test_sycl_queue.py b/dpctl/tests/test_sycl_queue.py index 0abe97b8bf..050818f8c2 100644 --- a/dpctl/tests/test_sycl_queue.py +++ b/dpctl/tests/test_sycl_queue.py @@ -169,6 +169,37 @@ def test_queue_submit_barrier(valid_filter): q.submit_barrier(range(3)) +def test_queue_is_get_last_event_available(): + assert isinstance(dpctl.SyclQueue.is_get_last_event_available(), bool) + + +def test_queue_get_last_event(): + if not dpctl.SyclQueue.is_get_last_event_available(): + pytest.skip("sycl_ext_oneapi_in_order_queue_events not available") + try: + q = dpctl.SyclQueue(property="in_order") + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + assert q.get_last_event() is None + buf = dpctl.memory.MemoryUSMDevice(1024, queue=q) + ev = q.memset_async(buf, 1, 1024) + last_ev = q.get_last_event() + assert isinstance(last_ev, dpctl.SyclEvent) + last_ev.wait() + assert ev.execution_status == dpctl.event_status_type.complete + + +def test_queue_get_last_event_out_of_order(): + if not dpctl.SyclQueue.is_get_last_event_available(): + pytest.skip("sycl_ext_oneapi_in_order_queue_events not available") + try: + q = dpctl.SyclQueue() + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + with pytest.raises(ValueError): + q.get_last_event() + + def test_queue__repr__(): try: q1 = dpctl.SyclQueue(property=0) diff --git a/libsyclinterface/include/syclinterface/dpctl_sycl_queue_interface.h b/libsyclinterface/include/syclinterface/dpctl_sycl_queue_interface.h index fb8b76a236..f04b2948da 100644 --- a/libsyclinterface/include/syclinterface/dpctl_sycl_queue_interface.h +++ b/libsyclinterface/include/syclinterface/dpctl_sycl_queue_interface.h @@ -414,6 +414,29 @@ DPCTLQueue_MemAdvise(__dpctl_keep DPCTLSyclQueueRef QRef, DPCTL_API bool DPCTLQueue_IsInOrder(__dpctl_keep const DPCTLSyclQueueRef QRef); +/*! + * @brief Check if ``DPCTLQueue_GetLastEvent`` is supported by this build, + * i.e. if the ``sycl_ext_oneapi_in_order_queue_events`` extension is + * available. + * + * @ingroup QueueInterface + */ +DPCTL_API +bool DPCTLQueue_GetLastEvent_Available(void); + +/*! + * @brief C-API wrapper for ``sycl::queue::ext_oneapi_get_last_event()``. + * + * @param QRef An opaque pointer to an in-order ``sycl::queue``. + * @return An opaque pointer to the event of the last command submitted to + * the queue, or ``nullptr`` if no command was submitted, the queue + * is out-of-order, or the extension is not available. + * @ingroup QueueInterface + */ +DPCTL_API +__dpctl_give DPCTLSyclEventRef +DPCTLQueue_GetLastEvent(__dpctl_keep const DPCTLSyclQueueRef QRef); + /*! * @brief C-API wrapper for * sycl::queue::has_property() that diff --git a/libsyclinterface/source/dpctl_sycl_queue_interface.cpp b/libsyclinterface/source/dpctl_sycl_queue_interface.cpp index a7d2a49c8b..fe9b87eed0 100644 --- a/libsyclinterface/source/dpctl_sycl_queue_interface.cpp +++ b/libsyclinterface/source/dpctl_sycl_queue_interface.cpp @@ -37,6 +37,7 @@ #include #include +#include #include #include #include /* SYCL headers */ @@ -844,6 +845,46 @@ bool DPCTLQueue_IsInOrder(__dpctl_keep const DPCTLSyclQueueRef QRef) return false; } +bool DPCTLQueue_GetLastEvent_Available(void) +{ +#ifdef SYCL_EXT_ONEAPI_IN_ORDER_QUEUE_EVENTS + return true; +#else + return false; +#endif +} + +__dpctl_give DPCTLSyclEventRef +DPCTLQueue_GetLastEvent(__dpctl_keep const DPCTLSyclQueueRef QRef) +{ +#ifdef SYCL_EXT_ONEAPI_IN_ORDER_QUEUE_EVENTS + auto Q = unwrap(QRef); + if (!Q) { + error_handler("QRef passed to GetLastEvent was NULL.", __FILE__, + __func__, __LINE__); + return nullptr; + } + if (!Q->is_in_order()) { + error_handler("Queue passed to GetLastEvent is not in-order.", __FILE__, + __func__, __LINE__); + return nullptr; + } + try { + std::optional ev = Q->ext_oneapi_get_last_event(); + if (!ev.has_value()) { + return nullptr; + } + return wrap(new event(std::move(*ev))); + } catch (std::exception const &e) { + error_handler(e, __FILE__, __func__, __LINE__); + return nullptr; + } +#else + (void)QRef; + return nullptr; +#endif +} + bool DPCTLQueue_HasEnableProfiling(__dpctl_keep const DPCTLSyclQueueRef QRef) { auto Q = unwrap(QRef); diff --git a/libsyclinterface/tests/test_sycl_queue_interface.cpp b/libsyclinterface/tests/test_sycl_queue_interface.cpp index f5d8959a93..b0ee399949 100644 --- a/libsyclinterface/tests/test_sycl_queue_interface.cpp +++ b/libsyclinterface/tests/test_sycl_queue_interface.cpp @@ -287,6 +287,79 @@ TEST(TestDPCTLSyclQueueInterface, CheckIsInOrderInvalid) EXPECT_FALSE(ioq); } +TEST(TestDPCTLSyclQueueInterface, CheckGetLastEventAvailable) +{ +#ifdef SYCL_EXT_ONEAPI_IN_ORDER_QUEUE_EVENTS + EXPECT_TRUE(DPCTLQueue_GetLastEvent_Available()); +#else + EXPECT_FALSE(DPCTLQueue_GetLastEvent_Available()); +#endif +} + +TEST(TestDPCTLSyclQueueInterface, CheckGetLastEvent) +{ + if (!DPCTLQueue_GetLastEvent_Available()) { + GTEST_SKIP() << "sycl_ext_oneapi_in_order_queue_events not available"; + } + DPCTLSyclDeviceSelectorRef DSRef = nullptr; + DPCTLSyclDeviceRef DRef = nullptr; + DPCTLSyclQueueRef QRef = nullptr; + DPCTLSyclEventRef ERef = nullptr; + DPCTLSyclEventRef LastERef = nullptr; + + EXPECT_NO_FATAL_FAILURE(DSRef = DPCTLDefaultSelector_Create()); + EXPECT_NO_FATAL_FAILURE(DRef = DPCTLDevice_CreateFromSelector(DSRef)); + EXPECT_NO_FATAL_FAILURE( + QRef = DPCTLQueue_CreateForDevice(DRef, nullptr, DPCTL_IN_ORDER)); + ASSERT_TRUE(QRef); + + // nothing submitted yet + EXPECT_NO_FATAL_FAILURE(LastERef = DPCTLQueue_GetLastEvent(QRef)); + EXPECT_FALSE(LastERef); + + constexpr size_t nbytes = 1024; + DPCTLSyclUSMRef ptr = nullptr; + EXPECT_NO_FATAL_FAILURE(ptr = DPCTLmalloc_device(nbytes, QRef)); + ASSERT_TRUE(ptr); + EXPECT_NO_FATAL_FAILURE(ERef = DPCTLQueue_Memset(QRef, ptr, 1, nbytes)); + ASSERT_TRUE(ERef); + + EXPECT_NO_FATAL_FAILURE(LastERef = DPCTLQueue_GetLastEvent(QRef)); + ASSERT_TRUE(LastERef); + EXPECT_NO_FATAL_FAILURE(DPCTLEvent_Wait(LastERef)); + EXPECT_TRUE(DPCTLEvent_GetCommandExecutionStatus(ERef) == DPCTL_COMPLETE); + + EXPECT_NO_FATAL_FAILURE(DPCTLEvent_Delete(LastERef)); + EXPECT_NO_FATAL_FAILURE(DPCTLEvent_Delete(ERef)); + EXPECT_NO_FATAL_FAILURE(DPCTLfree_with_queue(ptr, QRef)); + EXPECT_NO_FATAL_FAILURE(DPCTLQueue_Delete(QRef)); + EXPECT_NO_FATAL_FAILURE(DPCTLDevice_Delete(DRef)); + EXPECT_NO_FATAL_FAILURE(DPCTLDeviceSelector_Delete(DSRef)); +} + +TEST(TestDPCTLSyclQueueInterface, CheckGetLastEventInvalid) +{ + DPCTLSyclDeviceSelectorRef DSRef = nullptr; + DPCTLSyclDeviceRef DRef = nullptr; + DPCTLSyclQueueRef QRef = nullptr; + DPCTLSyclEventRef ERef = nullptr; + + EXPECT_NO_FATAL_FAILURE(ERef = DPCTLQueue_GetLastEvent(nullptr)); + EXPECT_FALSE(ERef); + + EXPECT_NO_FATAL_FAILURE(DSRef = DPCTLDefaultSelector_Create()); + EXPECT_NO_FATAL_FAILURE(DRef = DPCTLDevice_CreateFromSelector(DSRef)); + EXPECT_NO_FATAL_FAILURE(QRef = DPCTLQueue_CreateForDevice( + DRef, nullptr, DPCTL_DEFAULT_PROPERTY)); + ASSERT_TRUE(QRef); + EXPECT_NO_FATAL_FAILURE(ERef = DPCTLQueue_GetLastEvent(QRef)); + EXPECT_FALSE(ERef); + + EXPECT_NO_FATAL_FAILURE(DPCTLQueue_Delete(QRef)); + EXPECT_NO_FATAL_FAILURE(DPCTLDevice_Delete(DRef)); + EXPECT_NO_FATAL_FAILURE(DPCTLDeviceSelector_Delete(DSRef)); +} + TEST(TestDPCTLSyclQueueInterface, CheckHasEnableProfiling) { bool ioq = true; From 2b08f6eb0aac73c5e294a88125d4a6213ac7b786 Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 2 Nov 2026 00:14:31 -0800 Subject: [PATCH 2/4] use an in-order order manager for in-order queues _InOrderManager keeps only the most recently added events, and submitted_events is the queue's last event when get_last_event is available. It is cached like _SequentialOrderManager, so clear() and thread exit wait on the queue. --- CHANGELOG.md | 2 + .../doc_sources/api_reference/dpctl/utils.rst | 8 ++ dpctl/tests/test_utils.py | 111 ++++++++++++++++++ dpctl/utils/_order_manager.py | 110 ++++++++++++++++- 4 files changed, 226 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0eb677fe7f..2829f3af39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * `import_dpctl()` now returns 0 on success and -1 with a Python exception set on failure, and `dpctl4pybind11.hpp` raises that exception rather than failing on the first C-API call [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) +* `dpctl.utils.SequentialOrderManager` keeps only the most recently added events for in-order queues, and `submitted_events` returns the queue's last event when `sycl_ext_oneapi_in_order_queue_events` is available [gh-2299](https://github.com/IntelPython/dpctl/pull/2299) + ### 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) diff --git a/docs/doc_sources/api_reference/dpctl/utils.rst b/docs/doc_sources/api_reference/dpctl/utils.rst index 5706213530..dbb6641d73 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -31,6 +31,14 @@ best-effort: references that thread is already dropping are left to it, so they may still be held when ``wait`` returns. + For an in-order queue, the queue orders the tasks, and the order manager + keeps only the most recently added events. ``submitted_events`` is the + event of the last command submitted to the queue if + :meth:`dpctl.SyclQueue.is_get_last_event_available` is ``True``, and the + most recently added computation events otherwise. Only in the first case + does it cover tasks submitted to the queue without being added to the + order manager, e.g. by other threads. ``wait`` waits on the queue. + .. 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 diff --git a/dpctl/tests/test_utils.py b/dpctl/tests/test_utils.py index f1b2598862..988a4b0747 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -24,6 +24,7 @@ import time import warnings +import numpy as np import pytest import dpctl @@ -34,6 +35,7 @@ _start_deferred_release_watcher, _stop_deferred_release_watcher, ) +from dpctl.utils import _order_manager _free_threaded = bool(sysconfig.get_config_var("Py_GIL_DISABLED")) @@ -503,3 +505,112 @@ def __del__(self): timeout=120, ) assert out == "4" + + +def _in_order_queue(*args, cls=dpctl.SyclQueue): + try: + return cls(*args, property="in_order") + except dpctl.SyclQueueCreationError: + pytest.skip("Queue could not be created for default-selected device") + + +@pytest.fixture(params=[True, False], ids=["last_event", "added_events"]) +def use_last_event(request, monkeypatch): + if request.param and not dpctl.SyclQueue.is_get_last_event_available(): + pytest.skip("sycl_ext_oneapi_in_order_queue_events not available") + monkeypatch.setattr(_order_manager, "_use_last_event", request.param) + return request.param + + +def test_in_order_order_manager(use_last_event): + q = _in_order_queue() + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + assert _som[q] is _mngr + assert _mngr.num_host_task_events == 0 + assert _mngr.num_submitted_events == 0 + 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()) + ht_ev, comp_ev = dpctl.SyclEvent(), dpctl.SyclEvent() + _mngr.add_event_pair(ht_ev, [comp_ev]) + # only the most recently added events are kept + assert _mngr.host_task_events == [ht_ev] + assert _mngr.num_host_task_events == 1 + # nothing was submitted to the queue, so both modes return added events + assert _mngr.submitted_events == [comp_ev] + assert _mngr.num_submitted_events == 1 + _mngr.wait() + cpy = _mngr.__copy__() + assert cpy.host_task_events == [ht_ev] + assert cpy.submitted_events == [comp_ev] + _som.clear() + del cpy + + +def test_in_order_order_manager_submitted_events(use_last_event): + q = _in_order_queue() + _mngr = dpctl.utils.SequentialOrderManager[q] + buf = dpctl.memory.MemoryUSMDevice(1024, queue=q) + comp_ev = q.memset_async(buf, 1, 1024) + ht_ev = q._submit_keep_args_alive((buf,), [comp_ev]) + _mngr.add_event_pair(ht_ev, comp_ev) + if use_last_event: + # the keep-alive task is the last command submitted to the queue + assert len(_mngr.submitted_events) == 1 + assert _mngr.submitted_events[0] is not comp_ev + else: + assert _mngr.submitted_events == [comp_ev] + # work which is not added to the order manager + q.memset_async(buf, 2, 1024) + if use_last_event: + assert len(_mngr.submitted_events) == 1 + else: + assert _mngr.submitted_events == [comp_ev] + dpctl.utils.SequentialOrderManager.clear() + + +def test_in_order_order_manager_other_queue_depends(use_last_event): + q_in = _in_order_queue() + q_other = dpctl.SyclQueue(q_in.sycl_context, q_in.sycl_device) + _mngr = dpctl.utils.SequentialOrderManager[q_in] + nbytes, n_fills = 32 * 1024 * 1024, 8 + buf = dpctl.memory.MemoryUSMDevice(nbytes, queue=q_in) + host = np.empty(nbytes, dtype=np.uint8) + for _ in range(3): + q_in.memset(buf, 0, nbytes) + for v in range(1, n_fills + 1): + comp_ev = q_in.memset_async(buf, v, nbytes) + ht_ev = q_in._submit_keep_args_alive((buf,), [comp_ev]) + _mngr.add_event_pair(ht_ev, comp_ev) + # work on another queue must wait for work done on the in-order queue + deps = _mngr.submitted_events + assert deps + q_other.memcpy_async(host, buf, nbytes, dEvents=deps).wait() + assert np.all(host == n_fills) + dpctl.utils.SequentialOrderManager.clear() + + +class _WaitCountingQueue(dpctl.SyclQueue): + def __init__(self, *args, **kwargs): + self.num_waits = 0 + + def wait(self): + self.num_waits += 1 + super().wait() + + +def test_in_order_order_manager_waits_on_queue(): + q = _in_order_queue(cls=_WaitCountingQueue) + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + _mngr.wait() + assert q.num_waits == 1 + cpy = _mngr.__copy__() + del cpy + assert q.num_waits == 2 + del _mngr + # clear() runs at exit, and must wait on in-order queues as well + _som.clear() + assert q.num_waits > 2 diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index 8820ff81ad..62e7daaf6e 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -1,13 +1,40 @@ import sys import threading import weakref -from collections import defaultdict +from collections.abc import Sequence +from typing import Protocol from .._sycl_event import SyclEvent from .._sycl_queue import SyclQueue, _drain_deferred_releases from ._seq_order_keeper import _OrderManager +class OrderManagerProtocol(Protocol): + """Interface of order managers returned by ``SequentialOrderManager``.""" + + def add_event_pair( + self, + host_task_ev: SyclEvent | Sequence[SyclEvent], + comp_ev: SyclEvent | Sequence[SyclEvent], + ) -> None: ... + + @property + def num_host_task_events(self) -> int: ... + + @property + def num_submitted_events(self) -> int: ... + + @property + def host_task_events(self) -> list[SyclEvent]: ... + + @property + def submitted_events(self) -> list[SyclEvent]: ... + + def wait(self) -> None: ... + + def __copy__(self) -> "OrderManagerProtocol": ... + + class _SequentialOrderManager: """ Class to orchestrate default sequential order @@ -107,13 +134,78 @@ def __copy__(self): return res +def _as_event_list(ev): + if isinstance(ev, (list, tuple)): + return list(ev) + return [ev] + + +_use_last_event = SyclQueue.is_get_last_event_available() + + +class _InOrderManager: + """ + Order manager for an in-order queue. The queue orders its tasks, so only + the most recently added events are kept. + + ``submitted_events`` is the event of the last command submitted to the + queue if :meth:`dpctl.SyclQueue.get_last_event` is available, and the + most recently added computation events otherwise. ``wait`` waits on the + queue. + """ + + def __init__(self, q: SyclQueue): + self._queue = q + self._host_task_evs = [] + self._comp_evs = [] + + def __del__(self): + if sys.is_finalizing(): + return + self._queue.wait() + + def add_event_pair(self, host_task_ev, comp_ev): + self._host_task_evs = _as_event_list(host_task_ev) + self._comp_evs = _as_event_list(comp_ev) + + @property + def num_host_task_events(self): + return len(self._host_task_evs) + + @property + def num_submitted_events(self): + return len(self.submitted_events) + + @property + def host_task_events(self): + return list(self._host_task_evs) + + @property + def submitted_events(self): + if _use_last_event: + ev = self._queue.get_last_event() + if ev is not None: + return [ev] + return list(self._comp_evs) + + def wait(self): + self._queue.wait() + + def __copy__(self): + res = _InOrderManager(self._queue) + res._host_task_evs = list(self._host_task_evs) + res._comp_evs = list(self._comp_evs) + return res + + class SyclQueueToOrderManagerMap: """ Utility class used to ensure sequential ordering of offloaded tasks when passed to order manager. Maintains a thread-local dictionary mapping SyclQueue instances to - _SequentialOrderManager instances. + order managers: _InOrderManager for in-order queues, + _SequentialOrderManager otherwise. """ def __init__(self): @@ -128,15 +220,23 @@ def _get_map(self): try: return self._tls.order_manager_map except AttributeError: - m = defaultdict(_SequentialOrderManager) + m = {} self._tls.order_manager_map = m return m - def __getitem__(self, q: SyclQueue) -> _SequentialOrderManager: + def __getitem__(self, q: SyclQueue) -> OrderManagerProtocol: """Get order manager for given SyclQueue""" if not isinstance(q, SyclQueue): raise TypeError(f"Expected `dpctl.SyclQueue`, got {type(q)}") - return self._get_map()[q] + _local = self._get_map() + om = _local.get(q) + if om is None: + if q.is_in_order: + om = _InOrderManager(q) + else: + om = _SequentialOrderManager() + _local[q] = om + return om def clear(self): """Clear content of internal dictionary""" From 2fe078298f907cc5f990513d78af002b2da9021a Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Thu, 5 Nov 2026 09:41:25 -0800 Subject: [PATCH 3/4] port the in-order order manager to cleanup events _InOrderManager gains add_event, add_cleanup_event, cleanup_events and num_cleanup_events, with the host task names as aliases, and wait drains the deferred releases like _SequentialOrderManager.wait. --- dpctl/tests/test_utils.py | 64 ++++++++++++++++++++++++++++------- dpctl/utils/_order_manager.py | 39 ++++++++++++++++++--- 2 files changed, 85 insertions(+), 18 deletions(-) diff --git a/dpctl/tests/test_utils.py b/dpctl/tests/test_utils.py index 988a4b0747..4de04c153d 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -527,35 +527,72 @@ def test_in_order_order_manager(use_last_event): _som = dpctl.utils.SequentialOrderManager _mngr = _som[q] assert _som[q] is _mngr - assert _mngr.num_host_task_events == 0 + assert _mngr.num_cleanup_events == 0 assert _mngr.num_submitted_events == 0 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()) - ht_ev, comp_ev = dpctl.SyclEvent(), dpctl.SyclEvent() - _mngr.add_event_pair(ht_ev, [comp_ev]) + 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()]) + cl_ev, comp_ev = dpctl.SyclEvent(), dpctl.SyclEvent() + _mngr.add_event([comp_ev]) + _mngr.add_cleanup_event(cl_ev) # only the most recently added events are kept - assert _mngr.host_task_events == [ht_ev] - assert _mngr.num_host_task_events == 1 + assert _mngr.cleanup_events == [cl_ev] + assert _mngr.num_cleanup_events == 1 # nothing was submitted to the queue, so both modes return added events assert _mngr.submitted_events == [comp_ev] assert _mngr.num_submitted_events == 1 _mngr.wait() cpy = _mngr.__copy__() - assert cpy.host_task_events == [ht_ev] + assert cpy.cleanup_events == [cl_ev] assert cpy.submitted_events == [comp_ev] _som.clear() del cpy +def test_in_order_order_manager_host_task_api(): + q = _in_order_queue() + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] + ht_ev, comp_ev = dpctl.SyclEvent(), dpctl.SyclEvent() + with warnings.catch_warnings(): + warnings.simplefilter("error") + _mngr.add_event_pair(dpctl.SyclEvent(), dpctl.SyclEvent()) + _mngr.add_event_pair([dpctl.SyclEvent()], dpctl.SyclEvent()) + _mngr.add_event_pair(ht_ev, [comp_ev]) + assert _mngr.host_task_events == [ht_ev] + assert _mngr.num_host_task_events == 1 + assert _mngr.cleanup_events == [ht_ev] + _som.clear() + + +def test_in_order_order_manager_wait_drops_references(stopped_release_watcher): + q = _in_order_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" + + _mngr.wait() + assert released == [True] + _som.clear() + + def test_in_order_order_manager_submitted_events(use_last_event): q = _in_order_queue() _mngr = dpctl.utils.SequentialOrderManager[q] buf = dpctl.memory.MemoryUSMDevice(1024, queue=q) comp_ev = q.memset_async(buf, 1, 1024) - ht_ev = q._submit_keep_args_alive((buf,), [comp_ev]) - _mngr.add_event_pair(ht_ev, comp_ev) + ht_ev = q.keep_args_alive((buf,), [comp_ev]) + _mngr.add_event(comp_ev) + _mngr.add_cleanup_event(ht_ev) if use_last_event: # the keep-alive task is the last command submitted to the queue assert len(_mngr.submitted_events) == 1 @@ -582,8 +619,9 @@ def test_in_order_order_manager_other_queue_depends(use_last_event): q_in.memset(buf, 0, nbytes) for v in range(1, n_fills + 1): comp_ev = q_in.memset_async(buf, v, nbytes) - ht_ev = q_in._submit_keep_args_alive((buf,), [comp_ev]) - _mngr.add_event_pair(ht_ev, comp_ev) + ht_ev = q_in.keep_args_alive((buf,), [comp_ev]) + _mngr.add_event(comp_ev) + _mngr.add_cleanup_event(ht_ev) # work on another queue must wait for work done on the in-order queue deps = _mngr.submitted_events assert deps diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index 62e7daaf6e..eb8b1c45bb 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -18,15 +18,27 @@ def add_event_pair( comp_ev: SyclEvent | Sequence[SyclEvent], ) -> None: ... + def add_event(self, comp_ev: SyclEvent | Sequence[SyclEvent]) -> None: ... + + def add_cleanup_event( + self, cleanup_ev: SyclEvent | Sequence[SyclEvent] + ) -> None: ... + @property def num_host_task_events(self) -> int: ... + @property + def num_cleanup_events(self) -> int: ... + @property def num_submitted_events(self) -> int: ... @property def host_task_events(self) -> list[SyclEvent]: ... + @property + def cleanup_events(self) -> list[SyclEvent]: ... + @property def submitted_events(self) -> list[SyclEvent]: ... @@ -156,7 +168,7 @@ class _InOrderManager: def __init__(self, q: SyclQueue): self._queue = q - self._host_task_evs = [] + self._cleanup_evs = [] self._comp_evs = [] def __del__(self): @@ -165,12 +177,22 @@ def __del__(self): self._queue.wait() def add_event_pair(self, host_task_ev, comp_ev): - self._host_task_evs = _as_event_list(host_task_ev) + self._cleanup_evs = _as_event_list(host_task_ev) self._comp_evs = _as_event_list(comp_ev) + def add_event(self, comp_ev): + self._comp_evs = _as_event_list(comp_ev) + + def add_cleanup_event(self, cleanup_ev): + self._cleanup_evs = _as_event_list(cleanup_ev) + @property def num_host_task_events(self): - return len(self._host_task_evs) + return self.num_cleanup_events + + @property + def num_cleanup_events(self): + return len(self._cleanup_evs) @property def num_submitted_events(self): @@ -178,7 +200,11 @@ def num_submitted_events(self): @property def host_task_events(self): - return list(self._host_task_evs) + return self.cleanup_events + + @property + def cleanup_events(self): + return list(self._cleanup_evs) @property def submitted_events(self): @@ -190,10 +216,13 @@ def submitted_events(self): def wait(self): self._queue.wait() + # see _SequentialOrderManager.wait + if not sys.is_finalizing(): + _drain_deferred_releases() def __copy__(self): res = _InOrderManager(self._queue) - res._host_task_evs = list(self._host_task_evs) + res._cleanup_evs = list(self._cleanup_evs) res._comp_evs = list(self._comp_evs) return res From c2c6c2b0a112acd72000850aef41eea168ed042f Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Fri, 6 Nov 2026 08:56:13 -0800 Subject: [PATCH 4/4] keep incomplete events in the in-order order manager _InOrderManager now builds on _SequentialOrderManager, so events added from other queues are kept and non-events raise right away. submitted_events adds the queue's last event to them. --- CHANGELOG.md | 4 +- .../doc_sources/api_reference/dpctl/utils.rst | 15 +-- dpctl/tests/test_sycl_queue.py | 1 - dpctl/tests/test_utils.py | 112 ++++++++++-------- dpctl/utils/_order_manager.py | 65 +++------- 5 files changed, 90 insertions(+), 107 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2829f3af39..3967d4cc3b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,15 +11,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * 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) * Added `dpctl::utils::defer_release` to `dpctl4pybind11.hpp`, which calls a release callback once the given events have completed [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) - * Added `dpctl.SyclQueue.get_last_event` and `dpctl.SyclQueue.is_get_last_event_available`, and the `DPCTLQueue_GetLastEvent` and `DPCTLQueue_GetLastEvent_Available` C-API functions, wrapping the DPC++ `sycl_ext_oneapi_in_order_queue_events` extension [gh-2299](https://github.com/IntelPython/dpctl/pull/2299) ### 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) * `import_dpctl()` now returns 0 on success and -1 with a Python exception set on failure, and `dpctl4pybind11.hpp` raises that exception rather than failing on the first C-API call [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) - -* `dpctl.utils.SequentialOrderManager` keeps only the most recently added events for in-order queues, and `submitted_events` returns the queue's last event when `sycl_ext_oneapi_in_order_queue_events` is available [gh-2299](https://github.com/IntelPython/dpctl/pull/2299) +* `dpctl.utils.SequentialOrderManager[q].submitted_events` includes the queue's last event for in-order queues when `sycl_ext_oneapi_in_order_queue_events` is available, and `wait()` waits for all work on an in-order queue [gh-2299](https://github.com/IntelPython/dpctl/pull/2299) ### 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) diff --git a/docs/doc_sources/api_reference/dpctl/utils.rst b/docs/doc_sources/api_reference/dpctl/utils.rst index dbb6641d73..b75f8625a6 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -31,13 +31,14 @@ best-effort: references that thread is already dropping are left to it, so they may still be held when ``wait`` returns. - For an in-order queue, the queue orders the tasks, and the order manager - keeps only the most recently added events. ``submitted_events`` is the - event of the last command submitted to the queue if - :meth:`dpctl.SyclQueue.is_get_last_event_available` is ``True``, and the - most recently added computation events otherwise. Only in the first case - does it cover tasks submitted to the queue without being added to the - order manager, e.g. by other threads. ``wait`` waits on the queue. + For an in-order queue, if + :meth:`dpctl.SyclQueue.is_get_last_event_available` is ``True``, + ``submitted_events`` also has the event of the last command submitted to + the queue. That covers tasks submitted to the queue without being added + to the order manager, e.g. by other threads, and may be the event of an + empty kernel submitted by :meth:`dpctl.SyclQueue.keep_args_alive`. + ``wait``, ``clear`` and thread exit wait for all work on the queue, + including work submitted by other threads. .. deprecated:: 0.24.0 ``add_event_pair``, ``host_task_events`` and ``num_host_task_events`` diff --git a/dpctl/tests/test_sycl_queue.py b/dpctl/tests/test_sycl_queue.py index 050818f8c2..fbc25be08f 100644 --- a/dpctl/tests/test_sycl_queue.py +++ b/dpctl/tests/test_sycl_queue.py @@ -180,7 +180,6 @@ def test_queue_get_last_event(): q = dpctl.SyclQueue(property="in_order") except dpctl.SyclQueueCreationError: pytest.skip("Queue could not be created for default-selected device") - assert q.get_last_event() is None buf = dpctl.memory.MemoryUSMDevice(1024, queue=q) ev = q.memset_async(buf, 1, 1024) last_ev = q.get_last_event() diff --git a/dpctl/tests/test_utils.py b/dpctl/tests/test_utils.py index 4de04c153d..b55fabedac 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -522,49 +522,49 @@ def use_last_event(request, monkeypatch): return request.param -def test_in_order_order_manager(use_last_event): +def test_in_order_order_manager(): q = _in_order_queue() _som = dpctl.utils.SequentialOrderManager _mngr = _som[q] assert _som[q] is _mngr - assert _mngr.num_cleanup_events == 0 - assert _mngr.num_submitted_events == 0 + assert isinstance(_mngr, _order_manager._InOrderManager) assert isinstance(_mngr.submitted_events, list) 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()]) - cl_ev, comp_ev = dpctl.SyclEvent(), dpctl.SyclEvent() - _mngr.add_event([comp_ev]) - _mngr.add_cleanup_event(cl_ev) - # only the most recently added events are kept - assert _mngr.cleanup_events == [cl_ev] - assert _mngr.num_cleanup_events == 1 - # nothing was submitted to the queue, so both modes return added events - assert _mngr.submitted_events == [comp_ev] - assert _mngr.num_submitted_events == 1 + # complete events are not kept + assert _mngr.num_cleanup_events == 0 _mngr.wait() cpy = _mngr.__copy__() - assert cpy.cleanup_events == [cl_ev] - assert cpy.submitted_events == [comp_ev] + assert isinstance(cpy, _order_manager._InOrderManager) + assert cpy.num_cleanup_events == 0 _som.clear() del cpy +@pytest.mark.parametrize("in_order", [True, False]) +def test_order_manager_add_non_event(in_order): + q = _in_order_queue() if in_order else dpctl.SyclQueue() + _mngr = dpctl.utils.SequentialOrderManager[q] + with pytest.raises(TypeError): + _mngr.add_event(None) + with pytest.raises(TypeError): + _mngr.add_cleanup_event([None]) + dpctl.utils.SequentialOrderManager.clear() + + def test_in_order_order_manager_host_task_api(): q = _in_order_queue() _som = dpctl.utils.SequentialOrderManager _mngr = _som[q] - ht_ev, comp_ev = dpctl.SyclEvent(), dpctl.SyclEvent() with warnings.catch_warnings(): warnings.simplefilter("error") _mngr.add_event_pair(dpctl.SyclEvent(), dpctl.SyclEvent()) _mngr.add_event_pair([dpctl.SyclEvent()], dpctl.SyclEvent()) - _mngr.add_event_pair(ht_ev, [comp_ev]) - assert _mngr.host_task_events == [ht_ev] - assert _mngr.num_host_task_events == 1 - assert _mngr.cleanup_events == [ht_ev] + assert _mngr.host_task_events == [] + assert _mngr.num_host_task_events == 0 _som.clear() @@ -589,44 +589,62 @@ def test_in_order_order_manager_submitted_events(use_last_event): q = _in_order_queue() _mngr = dpctl.utils.SequentialOrderManager[q] buf = dpctl.memory.MemoryUSMDevice(1024, queue=q) - comp_ev = q.memset_async(buf, 1, 1024) - ht_ev = q.keep_args_alive((buf,), [comp_ev]) - _mngr.add_event(comp_ev) - _mngr.add_cleanup_event(ht_ev) - if use_last_event: - # the keep-alive task is the last command submitted to the queue - assert len(_mngr.submitted_events) == 1 - assert _mngr.submitted_events[0] is not comp_ev - else: - assert _mngr.submitted_events == [comp_ev] # work which is not added to the order manager - q.memset_async(buf, 2, 1024) + q.memset_async(buf, 1, 1024) if use_last_event: - assert len(_mngr.submitted_events) == 1 + assert _mngr.num_submitted_events == 1 else: - assert _mngr.submitted_events == [comp_ev] + assert _mngr.num_submitted_events == 0 + dpctl.utils.SequentialOrderManager.clear() + + +def _fill_in_steps(q, buf, n_fills, mngr=None): + nbytes = buf.nbytes + ev = None + for v in range(1, n_fills + 1): + ev = q.memset_async(buf, v, nbytes, dEvents=[ev] if ev else None) + if mngr is not None: + mngr.add_event(ev) + mngr.add_cleanup_event(q.keep_args_alive((buf,), [ev])) + return ev + + +def test_in_order_order_manager_other_queue_depends(): + if not dpctl.SyclQueue.is_get_last_event_available(): + pytest.skip("sycl_ext_oneapi_in_order_queue_events not available") + q_in = _in_order_queue() + q_other = dpctl.SyclQueue(q_in.sycl_context, q_in.sycl_device) + _mngr = dpctl.utils.SequentialOrderManager[q_in] + n_fills = 8 + buf = dpctl.memory.MemoryUSMDevice(64 * 1024 * 1024, queue=q_in) + host = np.empty(buf.nbytes, dtype=np.uint8) + for _ in range(5): + q_in.memset(buf, 0, buf.nbytes) + _fill_in_steps(q_in, buf, n_fills - 1, _mngr) + # the last fill is not added to the order manager + q_in.memset_async(buf, n_fills, buf.nbytes) + q_other.memcpy_async( + host, buf, buf.nbytes, dEvents=_mngr.submitted_events + ).wait() + assert np.all(host == n_fills) dpctl.utils.SequentialOrderManager.clear() -def test_in_order_order_manager_other_queue_depends(use_last_event): +def test_in_order_order_manager_other_queue_event(use_last_event): q_in = _in_order_queue() q_other = dpctl.SyclQueue(q_in.sycl_context, q_in.sycl_device) _mngr = dpctl.utils.SequentialOrderManager[q_in] - nbytes, n_fills = 32 * 1024 * 1024, 8 - buf = dpctl.memory.MemoryUSMDevice(nbytes, queue=q_in) - host = np.empty(nbytes, dtype=np.uint8) - for _ in range(3): - q_in.memset(buf, 0, nbytes) - for v in range(1, n_fills + 1): - comp_ev = q_in.memset_async(buf, v, nbytes) - ht_ev = q_in.keep_args_alive((buf,), [comp_ev]) - _mngr.add_event(comp_ev) - _mngr.add_cleanup_event(ht_ev) - # work on another queue must wait for work done on the in-order queue - deps = _mngr.submitted_events - assert deps - q_other.memcpy_async(host, buf, nbytes, dEvents=deps).wait() + n_fills = 8 + buf = dpctl.memory.MemoryUSMDevice(64 * 1024 * 1024, queue=q_in) + host = np.empty(buf.nbytes, dtype=np.uint8) + for _ in range(5): + q_in.memset(buf, 0, buf.nbytes) + _mngr.add_event(_fill_in_steps(q_other, buf, n_fills)) + q_in.memcpy_async( + host, buf, buf.nbytes, dEvents=_mngr.submitted_events + ).wait() assert np.all(host == n_fills) + q_other.wait() dpctl.utils.SequentialOrderManager.clear() diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index eb8b1c45bb..32ce9ab71c 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -146,84 +146,51 @@ def __copy__(self): return res -def _as_event_list(ev): - if isinstance(ev, (list, tuple)): - return list(ev) - return [ev] - - _use_last_event = SyclQueue.is_get_last_event_available() -class _InOrderManager: +class _InOrderManager(_SequentialOrderManager): """ - Order manager for an in-order queue. The queue orders its tasks, so only - the most recently added events are kept. + Order manager for an in-order queue. - ``submitted_events`` is the event of the last command submitted to the - queue if :meth:`dpctl.SyclQueue.get_last_event` is available, and the - most recently added computation events otherwise. ``wait`` waits on the - queue. + Holds the added events which are not complete, as + :class:`_SequentialOrderManager` does. ``submitted_events`` also has the + event of the last command submitted to the queue if + :meth:`dpctl.SyclQueue.get_last_event` is available. ``wait`` waits on + the queue. """ def __init__(self, q: SyclQueue): + super().__init__() self._queue = q - self._cleanup_evs = [] - self._comp_evs = [] def __del__(self): if sys.is_finalizing(): return self._queue.wait() - - def add_event_pair(self, host_task_ev, comp_ev): - self._cleanup_evs = _as_event_list(host_task_ev) - self._comp_evs = _as_event_list(comp_ev) - - def add_event(self, comp_ev): - self._comp_evs = _as_event_list(comp_ev) - - def add_cleanup_event(self, cleanup_ev): - self._cleanup_evs = _as_event_list(cleanup_ev) - - @property - def num_host_task_events(self): - return self.num_cleanup_events - - @property - def num_cleanup_events(self): - return len(self._cleanup_evs) + super().__del__() @property def num_submitted_events(self): return len(self.submitted_events) - @property - def host_task_events(self): - return self.cleanup_events - - @property - def cleanup_events(self): - return list(self._cleanup_evs) - @property def submitted_events(self): + evs = self._state.get_submitted_events() if _use_last_event: ev = self._queue.get_last_event() if ev is not None: - return [ev] - return list(self._comp_evs) + evs.append(ev) + return evs def wait(self): self._queue.wait() - # see _SequentialOrderManager.wait - if not sys.is_finalizing(): - _drain_deferred_releases() + return super().wait() def __copy__(self): - res = _InOrderManager(self._queue) - res._cleanup_evs = list(self._cleanup_evs) - res._comp_evs = list(self._comp_evs) + res = _InOrderManager.__new__(_InOrderManager) + res._state = _OrderManager(self._state) + res._queue = self._queue return res