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.

Author: ndgrigorian

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)

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``

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()

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()
 
 

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