From 0a0bb05b90582c18f5f8ebee5f61702a3a280d6b Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 2 Nov 2026 00:14:31 -0800 Subject: [PATCH 1/2] 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 2cc5c4bf8a..97eb4f3e4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +* 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 ### Deprecated 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 3fbf9c7bf6..a78705e433 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, @@ -2247,6 +2249,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 af0fe71dcd..2c9fa5d62b 100644 --- a/dpctl/tests/test_sycl_queue.py +++ b/dpctl/tests/test_sycl_queue.py @@ -168,6 +168,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 2d6d8ea3df8237488fea719b9dc2b9d0f66a7790 Mon Sep 17 00:00:00 2001 From: Nikita Grigorian Date: Mon, 2 Nov 2026 00:14:31 -0800 Subject: [PATCH 2/2] 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 | 112 ++++++++++++++++++ dpctl/utils/_order_manager.py | 110 ++++++++++++++++- 4 files changed, 227 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 97eb4f3e4c..3274797040 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 ### Changed +* `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 ### Removed diff --git a/docs/doc_sources/api_reference/dpctl/utils.rst b/docs/doc_sources/api_reference/dpctl/utils.rst index 093f298ff6..1947382956 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -15,3 +15,11 @@ Thread-local object mapping each :class:`dpctl.SyclQueue` to an order manager, used to ensure sequential ordering of offloaded tasks. + + 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. diff --git a/dpctl/tests/test_utils.py b/dpctl/tests/test_utils.py index 03298c3f22..59a9776fef 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -16,10 +16,13 @@ """Defines unit test cases for utility functions.""" +import numpy as np import pytest import dpctl +import dpctl.memory import dpctl.utils +from dpctl.utils import _order_manager @pytest.mark.filterwarnings("ignore:.*:RuntimeWarning") @@ -92,3 +95,112 @@ def test_order_manager(): _passed = True finally: assert _passed + + +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 7c66d6bcc8..a04c0834d6 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 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 @@ -67,13 +94,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): @@ -88,15 +180,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"""