From a87d9cfa4e43f128f34dd067933a5f90df6550d4 Mon Sep 17 00:00:00 2001 From: Vladislav Perevezentsev Date: Wed, 11 Nov 2026 11:13:26 -0700 Subject: [PATCH 1/3] Support weak references to SyclQueue Add a __weakref__ slot to SyclQueue. It follows the vtable pointer, so extensions built against an earlier dpctl find the existing fields and the vtable where they expect them; only the object grows, which they report as a size change at import. SyclQueue gets an empty __dealloc__, so that Cython gives it a deallocator of its own, which clears the weak references. Cython does not count __weakref__ as a field that needs one and would reuse the one of _SyclQueue, leaving weak references to freed memory. --- CHANGELOG.md | 1 + dpctl/_sycl_queue.pxd | 5 +++++ dpctl/_sycl_queue.pyx | 8 ++++++++ dpctl/tests/test_sycl_queue.py | 18 ++++++++++++++++++ 4 files changed, 32 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a9eb1d2f1f..1e55a65dbe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * 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) * Added `dpctl.SyclPlatform.has_aspect_*` properties and the `DPCTLPlatform_HasAspect` C-API function for querying aspects supported by all devices of a platform [gh-2412](https://github.com/IntelPython/dpctl/pull/2412) +* Added support for weak references to `dpctl.SyclQueue` [gh-2426](https://github.com/IntelPython/dpctl/pull/2426) ### 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) diff --git a/dpctl/_sycl_queue.pxd b/dpctl/_sycl_queue.pxd index 8ffa1f33f4..09e70880c2 100644 --- a/dpctl/_sycl_queue.pxd +++ b/dpctl/_sycl_queue.pxd @@ -50,6 +50,11 @@ cdef public api class SyclQueue (_SyclQueue) [ ]: """ Python wrapper class for a sycl::queue. """ + # lets dpctl.utils.SequentialOrderManager hold queues by weak reference; + # here rather than in _SyclQueue, so that it follows the vtable pointer and + # leaves the offsets that extensions built against earlier dpctl use as + # they were + cdef object __weakref__ cdef int _init_queue_default(self, int) cdef int _init_queue_from__SyclQueue(self, _SyclQueue) cdef int _init_queue_from_DPCTLSyclDeviceRef(self, DPCTLSyclDeviceRef, int) diff --git a/dpctl/_sycl_queue.pyx b/dpctl/_sycl_queue.pyx index 07fcd0cf20..20f297fe65 100644 --- a/dpctl/_sycl_queue.pyx +++ b/dpctl/_sycl_queue.pyx @@ -984,6 +984,14 @@ cdef class SyclQueue(_SyclQueue): "be renamed" ) + def __dealloc__(self): + # Nothing to release beyond what _SyclQueue does, but defining it + # makes Cython give SyclQueue a deallocator of its own, which clears + # the weak references to the queue. Cython does not count __weakref__ + # as a field that needs one, and would use _SyclQueue's instead, + # leaving them pointing to freed memory. + pass + cdef int _init_queue_from__SyclQueue(self, _SyclQueue other): """ Copy data container _SyclQueue fields over. """ diff --git a/dpctl/tests/test_sycl_queue.py b/dpctl/tests/test_sycl_queue.py index fbc25be08f..536efd3d6f 100644 --- a/dpctl/tests/test_sycl_queue.py +++ b/dpctl/tests/test_sycl_queue.py @@ -18,6 +18,7 @@ import ctypes import sys +import weakref import pytest @@ -132,6 +133,23 @@ def test_hashing_of_queue(): assert queue_dict +@pytest.mark.parametrize( + "cls", [dpctl.SyclQueue, type("Q", (dpctl.SyclQueue,), {})] +) +def test_weak_reference_to_queue(cls): + try: + q = cls() + except dpctl.SyclQueueCreationError: + pytest.skip("Default-constructed queue could not be created") + gone = [] + r = weakref.ref(q, gone.append) + assert r() is q + del q + # the queue's deallocator has to clear its weak references + assert gone == [r] + assert r() is None + + def test_channeling_device_properties(capsys): try: q = dpctl.SyclQueue() From 1addd5295ef4139a084d277058d635d87766e582 Mon Sep 17 00:00:00 2001 From: Vladislav Perevezentsev Date: Wed, 11 Nov 2026 11:13:30 -0700 Subject: [PATCH 2/3] Keep a copy of the queue in its order manager An order manager now holds its sycl::queue in a SyclQueue object of its own, rather than the queue it was looked up by, so that the values of the order manager map do not keep its keys alive. _InOrderManager used to hold its key. The in-order test that counts waits on the queue now makes the manager itself, as the managers of the map wait on their copy. --- dpctl/tests/test_utils.py | 22 +++++++++++++++++++--- dpctl/utils/_order_manager.py | 17 ++++++++++------- 2 files changed, 29 insertions(+), 10 deletions(-) diff --git a/dpctl/tests/test_utils.py b/dpctl/tests/test_utils.py index b55fabedac..b5915dab99 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -659,14 +659,30 @@ def wait(self): def test_in_order_order_manager_waits_on_queue(): q = _in_order_queue(cls=_WaitCountingQueue) - _som = dpctl.utils.SequentialOrderManager - _mngr = _som[q] + # the map gives its managers a copy of the queue, so the manager is made + # here to see its waits + _mngr = _order_manager._InOrderManager(q) _mngr.wait() assert q.num_waits == 1 cpy = _mngr.__copy__() del cpy assert q.num_waits == 2 del _mngr + assert q.num_waits == 3 + + +def test_order_manager_clear_waits_on_in_order_queues(monkeypatch): + waited = [] + _wait = _order_manager._InOrderManager.wait + + def wait(self): + waited.append(self) + return _wait(self) + + monkeypatch.setattr(_order_manager._InOrderManager, "wait", wait) + q = _in_order_queue() + _som = dpctl.utils.SequentialOrderManager + _mngr = _som[q] # clear() runs at exit, and must wait on in-order queues as well _som.clear() - assert q.num_waits > 2 + assert any(m is _mngr for m in waited) diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index 32ce9ab71c..d58189aefa 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -59,8 +59,11 @@ class _SequentialOrderManager: dependencies of later tasks. """ - def __init__(self): + def __init__(self, q=None): self._state = _OrderManager(16) + # the queue whose tasks are recorded, a copy of the one it is looked up + # by, so that the map never references the queue that is its key + self._queue = q def __del__(self): if sys.is_finalizing(): @@ -143,6 +146,7 @@ def wait(self): def __copy__(self): res = _SequentialOrderManager.__new__(_SequentialOrderManager) res._state = _OrderManager(self._state) + res._queue = self._queue return res @@ -160,10 +164,6 @@ class _InOrderManager(_SequentialOrderManager): the queue. """ - def __init__(self, q: SyclQueue): - super().__init__() - self._queue = q - def __del__(self): if sys.is_finalizing(): return @@ -227,10 +227,13 @@ def __getitem__(self, q: SyclQueue) -> OrderManagerProtocol: _local = self._get_map() om = _local.get(q) if om is None: + # the same sycl::queue in a new object, so that the value does + # not keep its key alive + q_copy = SyclQueue(q._get_capsule()) if q.is_in_order: - om = _InOrderManager(q) + om = _InOrderManager(q_copy) else: - om = _SequentialOrderManager() + om = _SequentialOrderManager(q_copy) _local[q] = om return om From 10dac33d99f96f203bf33c8b26b880bf8d462649 Mon Sep 17 00:00:00 2001 From: Vladislav Perevezentsev Date: Wed, 11 Nov 2026 11:13:33 -0700 Subject: [PATCH 3/3] Hold queues by weak reference in SequentialOrderManager The order manager map kept every queue it was asked for alive until clear(), so a program creating queues in a loop ran out of device or system memory (up to ~15 MB per live queue on Level Zero v2). Queues are now map keys by weak reference. A queue that is gone is only recorded by its weak reference callback; the owning thread drops its order manager on a later lookup once the work recorded in it is complete, without waiting, as the queue may be dropped on any thread, including inside a host task that the order manager would wait on. Until then, and while anything else holds it, the order manager is handed to an equal queue, so that its work stays ordered. Whether an in-order manager is idle is decided from the events recorded in it only: Level Zero v2 reports the last event of an in-order queue as submitted until it is waited on. --- CHANGELOG.md | 1 + .../doc_sources/api_reference/dpctl/utils.rst | 8 + dpctl/tests/test_utils.py | 136 ++++++++++++++ dpctl/utils/_order_manager.py | 171 ++++++++++++++++-- 4 files changed, 300 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e55a65dbe..969bed4e16 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * Removed stale `select_host_device` and `has_host_device` names from `dpctl.__all__`, which made `from dpctl import *` fail with `AttributeError` [gh-2410](https://github.com/IntelPython/dpctl/pull/2410) * A USM allocation that fails now drops the references held for offloaded tasks that have completed, and is attempted once more if that released anything, so that memory only waiting to be given up is not reported as unavailable [gh-2359](https://github.com/IntelPython/dpctl/pull/2359) * `dpctl.SyclPlatform.get_devices` no longer prints Unified Runtime warnings for `device_type="custom"` on non-OpenCL platforms, where custom devices are not supported [gh-2306](https://github.com/IntelPython/dpctl/pull/2306) +* `dpctl.utils.SequentialOrderManager` no longer keeps every queue alive until `clear()`: it holds queues by weak reference and drops the order manager of a queue once the queue is gone and its work is complete, so a program that keeps creating queues no longer runs out of memory [gh-2426](https://github.com/IntelPython/dpctl/pull/2426) ### Maintenance * Removed the `-fno-sycl-id-queries-fit-in-int` compile option, which none of the `dpctl` extensions need since `dpctl.tensor` moved to `dpnp` [gh-2413](https://github.com/IntelPython/dpctl/pull/2413) diff --git a/docs/doc_sources/api_reference/dpctl/utils.rst b/docs/doc_sources/api_reference/dpctl/utils.rst index b75f8625a6..49f3ea53ab 100644 --- a/docs/doc_sources/api_reference/dpctl/utils.rst +++ b/docs/doc_sources/api_reference/dpctl/utils.rst @@ -16,6 +16,14 @@ Thread-local object mapping each :class:`dpctl.SyclQueue` to an order manager, used to ensure sequential ordering of offloaded tasks. + Queues are held by weak reference, so creating queues does not keep them + alive. Equal queues, which share a ``sycl::queue``, share an order manager, + which lives while the queue it was looked up by does. Once that queue is + gone, the order manager is dropped when the work recorded in it is + complete and nothing else holds it; until then, an equal queue is given + the same order manager, so that its tasks still follow that work. Nothing + waits when a queue is dropped. + Record submitted tasks with ``add_event`` and use ``submitted_events`` as the dependency list of subsequent submissions. To keep Python objects referenced by a task alive until it completes, use diff --git a/dpctl/tests/test_utils.py b/dpctl/tests/test_utils.py index b5915dab99..fe148708cd 100644 --- a/dpctl/tests/test_utils.py +++ b/dpctl/tests/test_utils.py @@ -23,6 +23,7 @@ import threading import time import warnings +import weakref import numpy as np import pytest @@ -686,3 +687,138 @@ def wait(self): # clear() runs at exit, and must wait on in-order queues as well _som.clear() assert any(m is _mngr for m in waited) + + +def _thread_map(): + return dpctl.utils.SequentialOrderManager._get_thread_map() + + +@pytest.mark.parametrize("in_order", [False, True], ids=["ooo", "in_order"]) +def test_order_manager_does_not_keep_queues_alive(in_order): + _som = dpctl.utils.SequentialOrderManager + _som.clear() + refs = [] + for _ in range(16): + q = _in_order_queue() if in_order else _queue() + _mngr = _som[q] + buf = dpctl.memory.MemoryUSMDevice(1024, queue=q) + _mngr.add_event(q.memset_async(buf, 1, buf.nbytes)) + # holds nothing: buf would keep q alive until dpctl's thread drops it + _mngr.add_cleanup_event(q.keep_args_alive(())) + _mngr.wait() + refs.append(weakref.ref(q)) + del q, _mngr, buf + # each lookup drops the managers of the queues gone before it + assert len(_thread_map().managers) <= 2 + assert all(r() is None for r in refs) + m = _thread_map() + assert len(m.dead) == 1 + q = _in_order_queue() if in_order else _queue() + _som[q] + assert len(m.managers) == 1 and len(m.owned) == 1 + assert not m.retired and not m.dead and not m.dropped + _som.clear() + + +def test_order_manager_hands_pending_work_to_equal_queue(): + _som = dpctl.utils.SequentialOrderManager + _som.clear() + q1 = _queue() + # the same sycl::queue in another object + q2 = dpctl.SyclQueue(q1._get_capsule()) + assert q1 == q2 and q1 is not q2 + _mngr = _som[q1] + assert _som[q2] is _mngr + # as if the work recorded through q1 were not complete when q1 is gone + _mngr._is_idle = lambda: False + _mngr_ref = weakref.ref(_mngr) + del q1, _mngr + _som[_queue()] + # the map holds it, as its work is not complete + assert _mngr_ref() is not None + assert _som[q2] is _mngr_ref() + _som.clear() + + +def test_order_manager_drops_idle_manager_of_queue_gone(): + _som = dpctl.utils.SequentialOrderManager + _som.clear() + q1 = _queue() + q2 = dpctl.SyclQueue(q1._get_capsule()) + _mngr = _som[q1] + _mngr.wait() + _mngr_ref = weakref.ref(_mngr) + del q1, _mngr + # nothing holds it once the map drops it + _som[q2] + assert _mngr_ref() is None + m = _thread_map() + assert len(m.owned) == 1 and not m.retired and not m.dropped + _som.clear() + + +def test_order_manager_keeps_manager_held_for_equal_queue(): + _som = dpctl.utils.SequentialOrderManager + _som.clear() + q1 = _queue() + q2 = dpctl.SyclQueue(q1._get_capsule()) + _som[q1] + # looked up by q2, but the map has it under q1 + _mngr = _som[q2] + _mngr.wait() + del q1 + # the map drops it, its work being complete, while it is still in use + _som[_queue()] + assert _mngr._dropped + # work recorded in it now has to be ordered before the work on q2 + assert _som[q2] is _mngr + assert not _mngr._dropped + _som.clear() + + +def test_order_manager_orders_work_of_queue_gone(): + _som = dpctl.utils.SequentialOrderManager + _som.clear() + q2 = _queue() + buf = dpctl.memory.MemoryUSMDevice(64 * 1024 * 1024, queue=q2) + host = np.empty(buf.nbytes, dtype=np.uint8) + for _ in range(5): + q2.memset(buf, 0, buf.nbytes) + _som.clear() + # the same sycl::queue in another object, which is the key + q1 = dpctl.SyclQueue(q2._get_capsule()) + _mngr = _som[q1] + _fill_in_steps(q1, buf, 8, _mngr) + del _mngr + q1_ref = weakref.ref(q1) + del q1 + assert q1_ref() is None + _mngr = _som[q2] + ev = q2.memset_async(buf, 9, buf.nbytes, _mngr.submitted_events) + _mngr.add_event(ev) + q2.memcpy_async(host, buf, buf.nbytes, [ev]).wait() + assert np.all(host == 9) + del _mngr + _som.clear() + + +def test_order_manager_queue_gone_on_other_thread(monkeypatch): + _som = dpctl.utils.SequentialOrderManager + _som.clear() + q = _queue() + _mngr = _som[q] + monkeypatch.setattr(_mngr, "_is_idle", lambda: False) + holder = [q] + del q + # the last reference is dropped on another thread, which neither waits + # nor changes this thread's map + t = threading.Thread(target=holder.clear) + t.start() + t.join(timeout=60) + assert not t.is_alive() + m = _thread_map() + assert len(m.dead) == 1 and len(m.managers) == 1 + _som[_queue()] + assert [om for om in m.retired.values()] == [_mngr] + _som.clear() + assert not m.retired diff --git a/dpctl/utils/_order_manager.py b/dpctl/utils/_order_manager.py index d58189aefa..b4573ccc08 100644 --- a/dpctl/utils/_order_manager.py +++ b/dpctl/utils/_order_manager.py @@ -64,6 +64,11 @@ def __init__(self, q=None): # the queue whose tasks are recorded, a copy of the one it is looked up # by, so that the map never references the queue that is its key self._queue = q + # weak reference to the queue the map has it under, None if that + # queue is gone + self._key = None + # set while the map does not hold it, once its work was complete + self._dropped = False def __del__(self): if sys.is_finalizing(): @@ -143,14 +148,32 @@ def wait(self): _drain_deferred_releases() return res + def _is_idle(self): + """ + Whether all the work recorded here is complete; never blocks. + + For an in-order queue, work submitted without being recorded needs no + order manager: the queue keeps its order, and the last event of the + queue is a dependency of the work submitted after it. That event is not + looked at, as Level Zero v2 reports it submitted until waited on. + """ + _local = self._state + # both prune the complete events first + return not ( + _local.get_submitted_events() or _local.get_cleanup_events() + ) + def __copy__(self): res = _SequentialOrderManager.__new__(_SequentialOrderManager) res._state = _OrderManager(self._state) res._queue = self._queue + res._key = None + res._dropped = False return res _use_last_event = SyclQueue.is_get_last_event_available() +_weak_ref = weakref.ref class _InOrderManager(_SequentialOrderManager): @@ -167,7 +190,10 @@ class _InOrderManager(_SequentialOrderManager): def __del__(self): if sys.is_finalizing(): return - self._queue.wait() + # the queue keeps its own order, so one the map dropped need not wait + # for work submitted to the queue elsewhere since + if not self._dropped: + self._queue.wait() super().__del__() @property @@ -191,9 +217,51 @@ def __copy__(self): res = _InOrderManager.__new__(_InOrderManager) res._state = _OrderManager(self._state) res._queue = self._queue + res._key = None + res._dropped = False return res +class _QueueDeath(weakref.ref): + """ + Weak reference to a queue used as a key of the map, which tells the map + that the queue is gone. ``key`` is the weak reference the map is keyed by. + """ + + __slots__ = ("key",) + + def __new__(cls, q, callback, key): + return super().__new__(cls, q, callback) + + def __init__(self, q, callback, key): + super().__init__(q, callback) + self.key = key + + +class _ThreadMap: + """The order managers of one thread.""" + + __slots__ = ("managers", "deaths", "dead", "owned", "retired", "dropped") + + def __init__(self): + # weakref.ref(queue) -> order manager, for the queues alive + self.managers = {} + # weakref.ref(queue) -> _QueueDeath of the queue + self.deaths = {} + # _QueueDeath of the keys that are gone, appended by the weak + # reference callback on whichever thread drops the queue + self.dead = [] + # copy of the queue -> order manager, for all the map holds: those of + # the queues alive and those of the queues gone with work not complete + self.owned = {} + # id -> order manager of a queue that is gone, with work not complete + self.retired = {} + # copy of the queue -> order manager of a queue that is gone, which + # the map dropped once its work was complete, for as long as anything + # else holds it + self.dropped = weakref.WeakValueDictionary() + + class SyclQueueToOrderManagerMap: """ Utility class used to ensure sequential ordering of offloaded tasks @@ -202,31 +270,75 @@ class SyclQueueToOrderManagerMap: Maintains a thread-local dictionary mapping SyclQueue instances to order managers: _InOrderManager for in-order queues, _SequentialOrderManager otherwise. + + Queues are held by weak reference, and a ``sycl::queue`` has one order + manager in a thread for as long as the map or anything else holds it: the + map does while the queue it was looked up by is alive and while its work + is not complete. A queue looked up after that queue is gone gets the same + order manager if it is still held, so that the work recorded in it stays + ordered. Nothing waits when a queue is gone, as that can be on any thread, + including one that a recorded task is waiting on. """ def __init__(self): # each thread gets its own dictionary of order managers self._tls = threading.local() - def _get_map(self): + def _get_thread_map(self): """ Factory method to get or create an order manager map for the current thread. """ try: - return self._tls.order_manager_map + return self._tls.thread_map except AttributeError: - m = {} - self._tls.order_manager_map = m + m = _ThreadMap() + self._tls.thread_map = m return m - 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)}") - _local = self._get_map() - om = _local.get(q) + @staticmethod + def _prune(m): + """ + Drop the order managers of the queues that are gone once their work + is complete. Runs on the thread that owns ``m``. + """ + dead = m.dead + while dead: + death = dead.pop() + key = death.key + if m.deaths.get(key) is not death: + # taken over by another queue + continue + del m.deaths[key] + om = m.managers.pop(key) + om._key = None + m.retired[id(om)] = om + for k, om in list(m.retired.items()): + if om._is_idle(): + del m.retired[k] + if m.owned.get(om._queue) is om: + del m.owned[om._queue] + om._dropped = True + m.dropped[om._queue] = om + + @staticmethod + def _add(m, q): + """Give ``q`` an order manager, as ``m`` has none under it.""" + # one the sycl::queue already has: its queue is gone, even if the map + # is yet to be told + om = m.owned.get(q) if om is None: + om = m.dropped.pop(q, None) + if om is not None: + om._dropped = False + m.owned[om._queue] = om + if om is not None: + old = om._key + if old is not None: + m.managers.pop(old, None) + m.deaths.pop(old, None) + m.retired.pop(id(om), None) + else: # the same sycl::queue in a new object, so that the value does # not keep its key alive q_copy = SyclQueue(q._get_capsule()) @@ -234,15 +346,42 @@ def __getitem__(self, q: SyclQueue) -> OrderManagerProtocol: om = _InOrderManager(q_copy) else: om = _SequentialOrderManager(q_copy) - _local[q] = om + m.owned[q_copy] = om + key = _weak_ref(q) + om._key = key + m.managers[key] = om + m.deaths[key] = _QueueDeath(q, m.dead.append, key) + return om + + 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)}") + try: + m = self._tls.thread_map + except AttributeError: + m = self._get_thread_map() + if m.dead or m.retired: + self._prune(m) + # the key the map holds, so the lookup finds it by identity + om = m.managers.get(_weak_ref(q)) + if om is None: + om = self._add(m, q) return om def clear(self): """Clear content of internal dictionary""" - _local = self._get_map() - for v in _local.values(): - v.wait() - _local.clear() + m = self._get_thread_map() + for om in list(m.owned.values()): + om.wait() + for om in list(m.dropped.values()): + om.wait() + m.managers.clear() + m.deaths.clear() + m.dead.clear() + m.owned.clear() + m.retired.clear() + m.dropped.clear() SequentialOrderManager = SyclQueueToOrderManagerMap()