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

CHANGELOG.md

@@ -8,8 +8,12 @@ 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
 
+* `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

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.

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,
@@ -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

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)

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