Use a no-op order manager for in-order queues

CHANGELOG.md

@@ -11,11 +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[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)

docs/doc_sources/api_reference/dpctl/utils.rst

@@ -31,6 +31,15 @@
     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, 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``
         are deprecated. Tasks are no longer paired with a host task event, so

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(

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

dpctl/tests/test_sycl_queue.py

@@ -169,6 +169,36 @@ 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")
+    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)

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,168 @@ 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():
+    q = _in_order_queue()
+    _som = dpctl.utils.SequentialOrderManager
+    _mngr = _som[q]
+    assert _som[q] is _mngr
+    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()])
+    # complete events are not kept
+    assert _mngr.num_cleanup_events == 0
+    _mngr.wait()
+    cpy = _mngr.__copy__()
+    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]
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        _mngr.add_event_pair(dpctl.SyclEvent(), dpctl.SyclEvent())
+        _mngr.add_event_pair([dpctl.SyclEvent()], dpctl.SyclEvent())
+        assert _mngr.host_task_events == []
+        assert _mngr.num_host_task_events == 0
+    _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)
+    # work which is not added to the order manager
+    q.memset_async(buf, 1, 1024)
+    if use_last_event:
+        assert _mngr.num_submitted_events == 1
+    else:
+        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_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]
+    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()
+
+
+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