Implement custom scan group function for generic binary operator

Using

```
import dpctl.tensor as dpt
import dpctl

x = dpt.ones(2048000, dtype="f4")

q_prof = dpctl.SyclQueue(x.sycl_context, x.sycl_device, property="enable_profiling")
xx = x.to_device(q_prof)
mm = dpt.cumulative_logsumexp(xx)

timer = dpctl.SyclTimer(device_timer="order_manager", time_scale=1e9)
with timer(q_prof):
    for _ in range(250):
        dpt.cumulative_logsumexp(xx, out=mm)

print(f"dpctl.__version__ = {dpctl.__version__}")
print(f"Device: {x.sycl_device}")
print(f"host_dt={timer.dt.host_dt/250}, device_dt={timer.dt.device_dt/250}")
```

Testing on Iris Xe from WSL.

This branch:

```
$ python ~/cumlogsumexp.py
dpctl.__version__ = 0.19.0dev0+351.gffd26092a0.dirty
Device: <dpctl.SyclDevice [backend_type.level_zero, device_type.gpu,  Intel(R) Graphics [0x9a49]] at 0x7f37a8f995f0>
host_dt=1059589.7079911083, device_dt=1154782.72
```

vs. main branch:

```
$ python cumlogsumexp.py
dpctl.__version__ = 0.19.0dev0+307.g04a8228748
Device: <dpctl.SyclDevice [backend_type.level_zero, device_type.gpu,  Intel(R) Graphics [0x9a49]] at 0x7ff6147d3cf0>
host_dt=2721938.803792, device_dt=10048323.168
```

So this is about 8x speed-up.

Author: oleksandr-pavlyk

dpctl/tensor/libtensor/include/kernels/accumulators.hpp

@@ -280,7 +280,8 @@ inclusive_scan_base_step_blocked(sycl::queue &exec_q,
             }
             else {
                 wg_iscan_val = su_ns::custom_inclusive_scan_over_group(
-                    it.get_group(), slm_iscan_tmp, local_iscan.back(), scan_op);
+                    it.get_group(), it.get_sub_group(), slm_iscan_tmp,
+                    local_iscan.back(), identity, scan_op);
                 // ensure all finished reading from SLM, to avoid race condition
                 // with subsequent writes into SLM
                 it.barrier(sycl::access::fence_space::local_space);
@@ -454,7 +455,8 @@ inclusive_scan_base_step_striped(sycl::queue &exec_q,
             }
             else {
                 wg_iscan_val = su_ns::custom_inclusive_scan_over_group(
-                    it.get_group(), slm_iscan_tmp, local_iscan.back(), scan_op);
+                    it.get_group(), sg, slm_iscan_tmp, local_iscan.back(),
+                    identity, scan_op);
                 // ensure all finished reading from SLM, to avoid race condition
                 // with subsequent writes into SLM
                 it.barrier(sycl::access::fence_space::local_space);

dpctl/tensor/libtensor/include/utils/sycl_utils.hpp

@@ -212,29 +212,85 @@ T custom_reduce_over_group(const GroupT &wg,
     return sycl::group_broadcast(wg, red_val_over_wg, 0);
 }
 
-template <typename T, typename GroupT, typename LocAccT, typename OpT>
-T custom_inclusive_scan_over_group(const GroupT &wg,
-                                   LocAccT local_mem_acc,
-                                   const T local_val,
-                                   const OpT &op)
+template <typename GroupT,
+          typename SubGroupT,
+          typename LocAccT,
+          typename T,
+          typename OpT>
+T custom_inclusive_scan_over_group(GroupT &&wg,
+                                   SubGroupT &&sg,
+                                   LocAccT &&local_mem_acc,
+                                   const T &local_val,
+                                   const T &identity,
+                                   OpT &&op)
 {
     const std::uint32_t local_id = wg.get_local_id(0);
     const std::uint32_t wgs = wg.get_local_range(0);
-    local_mem_acc[local_id] = local_val;
 
+    const std::uint32_t lane_id = sg.get_local_id()[0];
+    const std::uint32_t sgSize = sg.get_local_range()[0];
+
+    T scan_val = local_val;
+    for (std::uint32_t step = 1; step < sgSize; step *= 2) {
+        const bool advanced_lane = (lane_id >= step);
+        const std::uint32_t src_lane_id =
+            (advanced_lane ? lane_id - step : lane_id);
+        const T modifier = sycl::select_from_group(sg, scan_val, src_lane_id);
+        if (advanced_lane) {
+            scan_val = op(scan_val, modifier);
+        }
+    }
+
+    local_mem_acc[local_id] = scan_val;
     sycl::group_barrier(wg, sycl::memory_scope::work_group);
 
-    if (wg.leader()) {
-        T scan_val = local_mem_acc[0];
-        for (std::uint32_t i = 1; i < wgs; ++i) {
-            scan_val = op(local_mem_acc[i], scan_val);
-            local_mem_acc[i] = scan_val;
+    const std::uint32_t max_sgSize = sg.get_max_local_range()[0];
+    const std::uint32_t sgr_id = sg.get_group_id()[0];
+
+    // now scan
+    const std::uint32_t n_aggregates = 1 + ((wgs - 1) / max_sgSize);
+    const bool large_wg = (n_aggregates > max_sgSize);
+    if (large_wg) {
+        if (wg.leader()) {
+            T _scan_val = identity;
+            for (std::uint32_t i = 1; i <= n_aggregates - max_sgSize; ++i) {
+                _scan_val = op(local_mem_acc[i * max_sgSize - 1], _scan_val);
+                local_mem_acc[i * max_sgSize - 1] = _scan_val;
+            }
+        }
+        sycl::group_barrier(wg, sycl::memory_scope::work_group);
+    }
+
+    if (sgr_id == 0 && lane_id < n_aggregates) {
+        const std::uint32_t offset =
+            (large_wg) ? n_aggregates - max_sgSize : 0u;
+        T __scan_val = (offset + lane_id > 0)
+                           ? local_mem_acc[(offset + lane_id) * max_sgSize - 1]
+                           : identity;
+        for (std::uint32_t step = 1; step < sgSize; step *= 2) {
+            const bool advanced_lane = (lane_id >= step);
+            const std::uint32_t src_lane_id =
+                (advanced_lane ? lane_id - step : lane_id);
+            const T modifier =
+                sycl::select_from_group(sg, __scan_val, src_lane_id);
+            if (advanced_lane) {
+                __scan_val = op(__scan_val, modifier);
+            }
         }
+        sycl::group_barrier(sg);
+        local_mem_acc[(offset + lane_id) * max_sgSize - 1] = __scan_val;
     }
+    sycl::group_barrier(wg, sycl::memory_scope::work_group);
 
-    // ensure all work-items see the same SLM that leader updated
+    if (sgr_id > 0) {
+        const T modifier = local_mem_acc[sgr_id * max_sgSize - 1];
+        scan_val = op(scan_val, modifier);
+    }
+
+    // ensure all work-items finished reading from SLM
     sycl::group_barrier(wg, sycl::memory_scope::work_group);
-    return local_mem_acc[local_id];
+
+    return scan_val;
 }
 
 // Reduction functors