Merge remote-tracking branch 'origin/add-empty-task-submission' into elementwise-functions-tuning

Author: oleksandr-pavlyk

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Improved performance of `tensor.argsort` function for all types [gh-1859](https://github.com/IntelPython/dpctl/pull/1859)
 * Improved performance of `tensor.sort` and `tensor.argsort` for short arrays in the range [16, 64] elements [gh-1866](https://github.com/IntelPython/dpctl/pull/1866)
 * Implement radix sort algorithm to be used in `dpt.sort` and `dpt.argsort` [gh-1867](https://github.com/IntelPython/dpctl/pull/1867)
+* Extended `dpctl.SyclTimer` with `device_timer` keyword, implementing different methods of collecting device times [gh-1872](https://github.com/IntelPython/dpctl/pull/1872)
 
 ### Fixed
 * Fix for `tensor.result_type` when all inputs are Python built-in scalars [gh-1877](https://github.com/IntelPython/dpctl/pull/1877)

dpctl/_sycl_timer.py

@@ -44,6 +44,44 @@ def device_dt(self):
         return self._device_dt
 
 
+class BaseDeviceTimer:
+    __slots__ = ["queue"]
+
+    def __init__(self, sycl_queue):
+        if not isinstance(sycl_queue, SyclQueue):
+            raise TypeError(f"Expected type SyclQueue, got {type(sycl_queue)}")
+        self.queue = sycl_queue
+
+
+class QueueBarrierDeviceTimer(BaseDeviceTimer):
+    __slots__ = []
+
+    def __init__(self, sycl_queue):
+        super(QueueBarrierDeviceTimer, self).__init__(sycl_queue)
+
+    def get_event(self):
+        return self.queue.submit_barrier()
+
+
+class OrderManagerDeviceTimer(BaseDeviceTimer):
+    __slots__ = ["_order_manager", "_submit_empty_task_fn"]
+
+    def __init__(self, sycl_queue):
+        import dpctl.utils._seq_order_keeper as s_ok
+        from dpctl.utils import SequentialOrderManager as seq_om
+
+        super(OrderManagerDeviceTimer, self).__init__(sycl_queue)
+        self._order_manager = seq_om[self.queue]
+        self._submit_empty_task_fn = s_ok._submit_empty_task
+
+    def get_event(self):
+        ev = self._submit_empty_task_fn(
+            sycl_queue=self.queue, depends=self._order_manager.submitted_events
+        )
+        self._order_manager.add_event_pair(ev, ev)
+        return ev
+
+
 class SyclTimer:
     """
     Context to measure device time and host wall-time of execution
@@ -58,7 +96,7 @@ class SyclTimer:
             q = dpctl.SyclQueue(property="enable_profiling")
 
             # create the timer
-            milliseconds_sc = 1e-3
+            milliseconds_sc = 1e3
             timer = dpctl.SyclTimer(time_scale = milliseconds_sc)
 
             # use the timer
@@ -73,25 +111,36 @@ class SyclTimer:
             wall_dt, device_dt = timer.dt
 
     .. note::
-        The timer submits barriers to the queue at the entrance and the
+        The timer submits tasks to the queue at the entrance and the
         exit of the context and uses profiling information from events
         associated with these submissions to perform the timing. Thus
         :class:`dpctl.SyclTimer` requires the queue with ``"enable_profiling"``
         property. In order to be able to collect the profiling information,
         the ``dt`` property ensures that both submitted barriers complete their
         execution and thus effectively synchronizes the queue.
 
+        `device_timer` keyword argument controls the type of tasks submitted.
+        With `device_timer="queue_barrier"`, queue barrier tasks are used. With
+        `device_timer="order_manager"`, a single empty body task is inserted
+        instead relying on order manager (used by `dpctl.tensor` operations) to
+        order these tasks so that they fence operations performed within
+        timer's context.
+
     Args:
         host_timer (callable, optional):
             A callable such that host_timer() returns current
             host time in seconds.
             Default: :py:func:`timeit.default_timer`.
+        device_timer (Literal["queue_barrier", "order_manager"], optional):
+            Device timing method. Default: "queue_barrier".
         time_scale (Union[int, float], optional):
             Ratio of the unit of time of interest and one second.
             Default: ``1``.
     """
 
-    def __init__(self, host_timer=timeit.default_timer, time_scale=1):
+    def __init__(
+        self, host_timer=timeit.default_timer, device_timer=None, time_scale=1
+    ):
         """
         Create new instance of :class:`.SyclTimer`.
 
@@ -100,6 +149,8 @@ def __init__(self, host_timer=timeit.default_timer, time_scale=1):
                 A function that takes no arguments and returns a value
                 measuring time.
                 Default: :meth:`timeit.default_timer`.
+            device_timer (Literal["queue_barrier", "order_manager"], optional):
+                Device timing method. Default: "queue_barrier"
             time_scale (Union[int, float], optional):
                 Scaling factor applied to durations measured by
                 the host_timer. Default: ``1``.
@@ -109,11 +160,26 @@ def __init__(self, host_timer=timeit.default_timer, time_scale=1):
         self.queue = None
         self.host_times = []
         self.bracketing_events = []
+        self._context_data = list()
+        if device_timer is None:
+            device_timer = "queue_barrier"
+        if device_timer == "queue_barrier":
+            self._device_timer_class = QueueBarrierDeviceTimer
+        elif device_timer == "order_manager":
+            self._device_timer_class = OrderManagerDeviceTimer
+        else:
+            raise ValueError(
+                "Supported values for device_timer keyword are "
+                "'queue_barrier', 'order_manager', got "
+                f"'{device_timer}'"
+            )
+        self._device_timer = None
 
     def __call__(self, queue=None):
         if isinstance(queue, SyclQueue):
             if queue.has_enable_profiling:
                 self.queue = queue
+                self._device_timer = self._device_timer_class(queue)
             else:
                 raise ValueError(
                     "The given queue was not created with the "
@@ -127,17 +193,17 @@ def __call__(self, queue=None):
         return self
 
     def __enter__(self):
-        self._event_start = self.queue.submit_barrier()
-        self._host_start = self.timer()
+        _event_start = self._device_timer.get_event()
+        _host_start = self.timer()
+        self._context_data.append((_event_start, _host_start))
         return self
 
     def __exit__(self, *args):
-        self.host_times.append((self._host_start, self.timer()))
-        self.bracketing_events.append(
-            (self._event_start, self.queue.submit_barrier())
-        )
-        del self._event_start
-        del self._host_start
+        _event_end = self._device_timer.get_event()
+        _host_end = self.timer()
+        _event_start, _host_start = self._context_data.pop()
+        self.host_times.append((_host_start, _host_end))
+        self.bracketing_events.append((_event_start, _event_end))
 
     @property
     def dt(self):

dpctl/tests/test_sycl_timer.py

@@ -0,0 +1,108 @@
+#                      Data Parallel Control (dpctl)
+#
+# Copyright 2020-2024 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+
+import pytest
+
+import dpctl
+import dpctl.tensor as dpt
+
+
+@pytest.fixture
+def profiling_queue():
+    try:
+        q = dpctl.SyclQueue(property="enable_profiling")
+    except dpctl.SyclQueueCreationError:
+        pytest.skip(
+            "Could not created profiling queue " "for default-selected device"
+        )
+    return q
+
+
+@pytest.mark.parametrize(
+    "device_timer", [None, "queue_barrier", "order_manager"]
+)
+def test_sycl_timer_queue_barrier(profiling_queue, device_timer):
+    dev = dpt.Device.create_device(profiling_queue)
+
+    timer = dpctl.SyclTimer(
+        host_timer=time.perf_counter, device_timer=device_timer, time_scale=1e3
+    )
+
+    with timer(dev.sycl_queue):
+        x = dpt.linspace(0, 1, num=10**6, device=dev)
+        y = 3.0 - dpt.square(x - 0.5)
+        z = dpt.sort(y)
+        res1 = z[-1]
+        res2 = dpt.max(y)
+
+    host_dt, device_dt = timer.dt
+
+    assert dpt.all(res1 == res2)
+    assert host_dt > 0
+    assert device_dt > 0
+
+
+def test_sycl_timer_accumulation(profiling_queue):
+    q = profiling_queue
+
+    timer = dpctl.SyclTimer(
+        host_timer=time.perf_counter,
+        device_timer="order_manager",
+        time_scale=1e3,
+    )
+
+    # initial condition
+    x = dpt.linspace(0, 1, num=10**6, sycl_queue=q)
+
+    aitkens_data = [
+        x,
+    ]
+
+    # 16 iterations of Aitken's accelerated Newton's method
+    # x <- x - f(x)/f'(x) for f(x) = x - cos(x)
+    for _ in range(16):
+        # only time Newton step
+        with timer(q):
+            s = dpt.sin(x)
+            x = (dpt.cos(x) + x * s) / (1 + s)
+        aitkens_data.append(x)
+        aitkens_data = aitkens_data[-3:]
+        if len(aitkens_data) == 3:
+            # apply Aitkens acceleration
+            d1 = aitkens_data[-1] - aitkens_data[-2]
+            d2 = aitkens_data[-2] - aitkens_data[-3]
+            if not dpt.any(d1 == d2):
+                x = aitkens_data[-1] - dpt.square(d1) / (d1 - d2)
+
+    # Total time for 16 iterations
+    dev_dt = timer.dt.device_dt
+    assert dev_dt > 0
+
+    # check convergence
+    assert dpt.max(x) - dpt.min(x) < 1e-5
+
+
+def test_sycl_timer_validation():
+    with pytest.raises(ValueError):
+        dpctl.SyclTimer(device_timer="invalid")
+
+    timer = dpctl.SyclTimer()
+    mock_queue = Ellipsis
+
+    with pytest.raises(TypeError):
+        timer(mock_queue)

dpctl/utils/src/order_keeper.cpp

@@ -26,4 +26,17 @@ PYBIND11_MODULE(_seq_order_keeper, m)
              &SequentialOrder::add_to_submitted_events)
         .def("wait", &SequentialOrder::wait,
              py::call_guard<py::gil_scoped_release>());
+
+    auto submit_empty_task_fn =
+        [](sycl::queue &exec_q,
+           const std::vector<sycl::event> &depends) -> sycl::event {
+        return exec_q.submit([&](sycl::handler &cgh) {
+            cgh.depends_on(depends);
+            cgh.single_task([]() {
+                // empty body
+            });
+        });
+    };
+    m.def("_submit_empty_task", submit_empty_task_fn, py::arg("sycl_queue"),
+          py::arg("depends") = py::list());
 }