From c20e9f99379c6b5dc0e81fcca38ef99b4fd1276b Mon Sep 17 00:00:00 2001 From: Alexey Kukanov Date: Wed, 16 Sep 2026 15:12:20 +0200 Subject: [PATCH 1/8] move block carry routines out of scan submitter --- .../parallel_backend_sycl_reduce_then_scan.h | 57 +++++++++---------- 1 file changed, 27 insertions(+), 30 deletions(-) diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h index caad6729025..5c863da6e4c 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h @@ -1617,11 +1617,8 @@ __scan_through_elements_helper_impl(const sycl::nd_item<1>& __ndi, _GenInput __g std::uint32_t __elements_to_process = static_cast(__subgroup_n - (__iters - 1) * __sub_group_size); __sub_group_scan_partial<__is_inclusive>(__ndi, __scan_input_transform(__v), __binary_op, __sub_group_carry, __elements_to_process, __comm_tag); - if constexpr (!std::is_same_v<_WriteOp, oneapi::dpl::__internal::__ignore_call_op>) - { - if (__offset < __n) - __write_op(__offset, __v); - } + if (__offset < __n) + __write_op(__offset, __v); } // Detecting TempData type alias in the specified structure @@ -1758,6 +1755,24 @@ struct __comm_slm_handler<__subgroup_only_tag, _InitValueType> } }; +// Helper functions to communicate between processing blocks via temporary storage. +// Each block writes a carry-out partial sum which serves as the carry-in for the next block. +// To prevent data race within the block, carry-in and carry-out values flip between odd & even blocks. + +template +_ValueType +__get_block_carry_in(const std::size_t __block_num, _ValueType* __tmp_ptr) +{ + return __tmp_ptr[__block_num % 2]; +} + +template +void +__set_block_carry_out(const std::size_t __block_num, _ValueType* __tmp_ptr, const _ValueType __block_carry_out) +{ + __tmp_ptr[1 - (__block_num % 2)] = __block_carry_out; +} + template class __reduce_then_scan_partition_kernel; @@ -1933,22 +1948,6 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is { using _InitValueType = typename _InitType::__value_type; - template - _InitValueType - __get_block_carry_in(const std::size_t __block_num, _TmpAcc __tmp_acc, - const std::size_t __num_sub_groups_global) const - { - return __tmp_acc[__num_sub_groups_global + (__block_num % 2)]; - } - - template - void - __set_block_carry_out(const std::size_t __block_num, _TmpAcc __tmp_acc, const _ValueType __block_carry_out, - const std::size_t __num_sub_groups_global) const - { - __tmp_acc[__num_sub_groups_global + 1 - (__block_num % 2)] = __block_carry_out; - } - template sycl::event operator()(sycl::queue& __q, const sycl::nd_range<1> __nd_range, _InRng&& __in_rng, _OutRng&& __out_rng, @@ -2157,23 +2156,21 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is } else { + _InitValueType __carry_in = + __get_block_carry_in(__block_num, __tmp_ptr + __max_num_sub_groups_global); if (__sub_group_id > 0) { _InitValueType __value = __sub_group_partials[std::min(__sub_group_id - 1, __active_subgroups - 1)]; - __sub_group_carry.__setup(__reduce_op( - __get_block_carry_in(__block_num, __tmp_ptr, __max_num_sub_groups_global), __value)); + __sub_group_carry.__setup(__reduce_op(__carry_in, __value)); } else if (__group_id > 0) { - __sub_group_carry.__setup( - __reduce_op(__get_block_carry_in(__block_num, __tmp_ptr, __max_num_sub_groups_global), - __sub_group_partials[__active_subgroups])); + __sub_group_carry.__setup(__reduce_op(__carry_in, __sub_group_partials[__active_subgroups])); } else { - __sub_group_carry.__setup( - __get_block_carry_in(__block_num, __tmp_ptr, __max_num_sub_groups_global)); + __sub_group_carry.__setup(__carry_in); } } @@ -2249,8 +2246,8 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is else { // capture the last carry out for the next block - __set_block_carry_out(__block_num, __tmp_ptr, __sub_group_carry.__get_cref(), - __max_num_sub_groups_global); + __set_block_carry_out(__block_num, __tmp_ptr + __max_num_sub_groups_global, + __sub_group_carry.__get_cref()); } } }); From 20280d5866469ae603aa29d2cdd5c55c42280863 Mon Sep 17 00:00:00 2001 From: Alexey Kukanov Date: Wed, 16 Sep 2026 18:19:18 +0200 Subject: [PATCH 2/8] lower tag dispatching to __scan_through_elements_helper_impl --- .../parallel_backend_sycl_reduce_then_scan.h | 144 +++++++++--------- 1 file changed, 71 insertions(+), 73 deletions(-) diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h index 5c863da6e4c..26869ff540e 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h @@ -1587,38 +1587,42 @@ __scan_through_elements_helper_impl(const sycl::nd_item<1>& __ndi, _GenInput __g { using _GenInputType = std::invoke_result_t<_GenInput, _InRng, std::size_t>; - const std::uint8_t __sub_group_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); - - // For partial thread, we need to handle the partial subgroup at the end of the range - const std::uint32_t __subgroup_n = static_cast( - std::min(__n - __subgroup_start_id, __iters_per_item * __sub_group_size)); - std::uint32_t __iters = oneapi::dpl::__internal::__dpl_ceiling_div(__subgroup_n, __sub_group_size); + // Hoist the sub-group-ops vs SLM-fallback decision to here. The element-scan body below is instantiated + // once per available communication path; the branch is taken a single time per call to this helper. + __dispatch_comm_tag(__comm_tag, [&](auto __comm_tag_concrete) { + const std::uint8_t __sub_group_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); + // For partial thread, we need to handle the partial subgroup at the end of the range + const std::uint32_t __subgroup_n = static_cast( + std::min(__n - __subgroup_start_id, __iters_per_item * __sub_group_size)); + std::uint32_t __iters = oneapi::dpl::__internal::__dpl_ceiling_div(__subgroup_n, __sub_group_size); - if (__iters > 1) - { - // peel first iteration out as workaround for issue set_union.pass and reduce_by_segment.pass - // with some compilers and environments - _GenInputType __v = __gen_input(__in_rng, __start_id); - __sub_group_scan<__is_inclusive>(__ndi, __scan_input_transform(__v), __binary_op, __sub_group_carry, - __comm_tag); - __write_op(__start_id, __v); - - for (std::uint32_t __j = 1; __j + 1 < __iters; __j++) + if (__iters > 1) { - __v = __gen_input(__in_rng, __start_id + __j * __sub_group_size); + // peel first iteration out as workaround for issue set_union.pass and reduce_by_segment.pass + // with some compilers and environments + _GenInputType __v = __gen_input(__in_rng, __start_id); __sub_group_scan<__is_inclusive>(__ndi, __scan_input_transform(__v), __binary_op, __sub_group_carry, - __comm_tag); - __write_op(__start_id + __j * __sub_group_size, __v); + __comm_tag_concrete); + __write_op(__start_id, __v); + + for (std::uint32_t __j = 1; __j + 1 < __iters; __j++) + { + __v = __gen_input(__in_rng, __start_id + __j * __sub_group_size); + __sub_group_scan<__is_inclusive>(__ndi, __scan_input_transform(__v), __binary_op, __sub_group_carry, + __comm_tag_concrete); + __write_op(__start_id + __j * __sub_group_size, __v); + } } - } - std::size_t __offset = __start_id + (__iters - 1) * __sub_group_size; - std::size_t __local_id = std::min(__offset, __n - 1); - _GenInputType __v = __gen_input(__in_rng, __local_id); - std::uint32_t __elements_to_process = static_cast(__subgroup_n - (__iters - 1) * __sub_group_size); - __sub_group_scan_partial<__is_inclusive>(__ndi, __scan_input_transform(__v), __binary_op, __sub_group_carry, - __elements_to_process, __comm_tag); - if (__offset < __n) - __write_op(__offset, __v); + std::size_t __offset = __start_id + (__iters - 1) * __sub_group_size; + std::size_t __local_id = std::min(__offset, __n - 1); + _GenInputType __v = __gen_input(__in_rng, __local_id); + std::uint32_t __elements_to_process = + static_cast(__subgroup_n - (__iters - 1) * __sub_group_size); + __sub_group_scan_partial<__is_inclusive>(__ndi, __scan_input_transform(__v), __binary_op, __sub_group_carry, + __elements_to_process, __comm_tag_concrete); + if (__offset < __n) + __write_op(__offset, __v); + }); } // Detecting TempData type alias in the specified structure @@ -1661,55 +1665,49 @@ __scan_through_elements_helper(const sycl::nd_item<1>& __ndi, _GenInput __gen_in return __gen_input(__rng, __id); }; - // Hoist the sub-group-ops vs SLM-fallback decision to here. The element-scan body below is instantiated - // once per available communication path; the branch is taken a single time per call to this helper. - __dispatch_comm_tag(__comm_tag, [&](auto __comm_tag_concrete) { - if constexpr (std::is_same_v<_WriteOp, oneapi::dpl::__internal::__ignore_call_op>) - { - __scan_through_elements_helper_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __binary_op, - oneapi::dpl::__internal::__ignore_call_op{}, __sub_group_carry, __in_rng, __start_id, __n, - __iters_per_item, __subgroup_start_id, __comm_tag_concrete); - } - else + if constexpr (std::is_same_v<_WriteOp, oneapi::dpl::__internal::__ignore_call_op>) + { + __scan_through_elements_helper_impl<__is_inclusive>( + __ndi, __gen_input_impl, __scan_input_transform, __binary_op, oneapi::dpl::__internal::__ignore_call_op{}, + __sub_group_carry, __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + } + else + { + if constexpr (_Bounded) { - if constexpr (_Bounded) + const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); + // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: + // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned + // element is a diagonal written through __write_multiple_to_id. The estimate must account for + // this many writes per scanned element, otherwise the unchecked write path could be selected for + // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). + const std::size_t __max_write_offset = + std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; + if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) { - const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); - // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: - // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned - // element is a diagonal written through __write_multiple_to_id. The estimate must account for - // this many writes per scanned element, otherwise the unchecked write path could be selected for - // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). - const std::size_t __max_write_offset = std::size_t{__is_unique_pattern_v} + - __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; - if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) - { - auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); - else - __write_op(__out_rng, __id, __v, __on_oob_reached); - }; - __scan_through_elements_helper_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __binary_op, __bounded_write_op, - __sub_group_carry, __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, - __comm_tag_concrete); - return; - } + auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); + else + __write_op(__out_rng, __id, __v, __on_oob_reached); + }; + __scan_through_elements_helper_impl<__is_inclusive>( + __ndi, __gen_input_impl, __scan_input_transform, __binary_op, __bounded_write_op, __sub_group_carry, + __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + return; } - - auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data); - else - __write_op(__out_rng, __id, __v); - }; - __scan_through_elements_helper_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __binary_op, __unbounded_write_op, __sub_group_carry, - __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag_concrete); } - }); + + auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data); + else + __write_op(__out_rng, __id, __v); + }; + __scan_through_elements_helper_impl<__is_inclusive>( + __ndi, __gen_input_impl, __scan_input_transform, __binary_op, __unbounded_write_op, __sub_group_carry, + __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + } } template From 1dc67185af10575c140108351ba818b7ebba483b Mon Sep 17 00:00:00 2001 From: Alexey Kukanov Date: Wed, 16 Sep 2026 19:25:10 +0200 Subject: [PATCH 3/8] simplify scan-through-elements for the reduce stage --- .../parallel_backend_sycl_reduce_then_scan.h | 128 ++++++++---------- 1 file changed, 57 insertions(+), 71 deletions(-) diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h index 26869ff540e..5ac395ced19 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h @@ -820,8 +820,6 @@ template struct __gen_set_balanced_path { - using TempData = __noop_temp_data; - // Locates and returns the "intersection" of a diagonal on the balanced path, based on merge path coordinates. // It returns coordinates in each set of the intersection with a boolean representing if the diagonal is "starred", // meaning that the balanced path "intersection" point does not lie directly on the diagonal, but one step forward in @@ -939,9 +937,9 @@ struct __gen_set_balanced_path } // Entry point for reduce then scan reduce input - template + template _RetType - operator()(const _InRng& __in_rng, _IndexT __id, TempData& __temp_data, _FinalPosSaver __final_pos_saver) const + operator()(const _InRng& __in_rng, _IndexT __id) const { // Get source tuple auto&& __tuple = __in_rng.base(); @@ -988,13 +986,14 @@ struct __gen_set_balanced_path __star = __local_star; } + __noop_temp_data __temp_data{}; // lvalue is required to call __set_op_count const __temp_data_array_idx_t __eles_to_process = static_cast<__temp_data_array_idx_t>( std::min(_IndexT{__diagonal_spacing} - (__star ? _IndexT{1} : _IndexT{0}), oneapi::dpl::__ranges::__size(__rng1) + oneapi::dpl::__ranges::__size(__rng2) - _IndexT{__id * __diagonal_spacing - 1})); return _RetType{__set_op_count(__rng1, __rng2, __rng1_balanced_pos, __rng2_balanced_pos, __eles_to_process, - __temp_data, __comp, __proj1, __proj2, __final_pos_saver)}; + __temp_data, __comp, __proj1, __proj2, __internal::__no_callback_tag{})}; } _SetOpCount __set_op_count; __temp_data_array_idx_t __diagonal_spacing; @@ -1577,13 +1576,11 @@ __sub_group_scan_partial(const sycl::nd_item<1>& __ndi, _ValueType& __value, _Bi template void -__scan_through_elements_helper_impl(const sycl::nd_item<1>& __ndi, _GenInput __gen_input, - _ScanInputTransform __scan_input_transform, _BinaryOp __binary_op, - _WriteOp __write_op, - oneapi::dpl::__internal::__opt_lazy_ctor_storage<_ValueType>& __sub_group_carry, - const _InRng& __in_rng, std::size_t __start_id, std::size_t __n, - std::uint32_t __iters_per_item, std::size_t __subgroup_start_id, - _CommTag __comm_tag) +__scan_through_elements_impl(const sycl::nd_item<1>& __ndi, _GenInput __gen_input, + _ScanInputTransform __scan_input_transform, _BinaryOp __binary_op, _WriteOp __write_op, + oneapi::dpl::__internal::__opt_lazy_ctor_storage<_ValueType>& __sub_group_carry, + const _InRng& __in_rng, std::size_t __start_id, std::size_t __n, + std::uint32_t __iters_per_item, std::size_t __subgroup_start_id, _CommTag __comm_tag) { using _GenInputType = std::invoke_result_t<_GenInput, _InRng, std::size_t>; @@ -1640,17 +1637,17 @@ struct __temp_data_required<_T, std::void_t> using type = typename _T::TempData; }; -template + typename _OutRng, typename _CommTag, typename _OnOOBReached, typename _FinalPosSaver> void -__scan_through_elements_helper(const sycl::nd_item<1>& __ndi, _GenInput __gen_input, - _ScanInputTransform __scan_input_transform, _BinaryOp __binary_op, _WriteOp __write_op, - oneapi::dpl::__internal::__opt_lazy_ctor_storage<_ValueType>& __sub_group_carry, - const _InRng& __in_rng, _OutRng& __out_rng, std::size_t __start_id, std::size_t __n, - std::uint32_t __iters_per_item, std::size_t __subgroup_start_id, _CommTag __comm_tag, - _OnOOBReached __on_oob_reached = {}, _FinalPosSaver __final_pos_saver = {}) +__scan_through_elements(const sycl::nd_item<1>& __ndi, _GenInput __gen_input, + _ScanInputTransform __scan_input_transform, _BinaryOp __binary_op, _WriteOp __write_op, + oneapi::dpl::__internal::__opt_lazy_ctor_storage<_ValueType>& __sub_group_carry, + const _InRng& __in_rng, _OutRng& __out_rng, std::size_t __start_id, std::size_t __n, + std::uint32_t __iters_per_item, std::size_t __subgroup_start_id, _CommTag __comm_tag, + _OnOOBReached __on_oob_reached, _FinalPosSaver __final_pos_saver) { using __temp_data_required_t = __temp_data_required<_GenInput>; constexpr bool __is_temp_data_required = __temp_data_required_t::value; @@ -1665,49 +1662,40 @@ __scan_through_elements_helper(const sycl::nd_item<1>& __ndi, _GenInput __gen_in return __gen_input(__rng, __id); }; - if constexpr (std::is_same_v<_WriteOp, oneapi::dpl::__internal::__ignore_call_op>) - { - __scan_through_elements_helper_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __binary_op, oneapi::dpl::__internal::__ignore_call_op{}, - __sub_group_carry, __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); - } - else + if constexpr (__is_bounded) { - if constexpr (_Bounded) + const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); + // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: + // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned + // element is a diagonal written through __write_multiple_to_id. The estimate must account for + // this many writes per scanned element, otherwise the unchecked write path could be selected for + // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). + const std::size_t __max_write_offset = + std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; + if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) { - const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); - // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: - // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned - // element is a diagonal written through __write_multiple_to_id. The estimate must account for - // this many writes per scanned element, otherwise the unchecked write path could be selected for - // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). - const std::size_t __max_write_offset = - std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; - if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) - { - auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); - else - __write_op(__out_rng, __id, __v, __on_oob_reached); - }; - __scan_through_elements_helper_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __binary_op, __bounded_write_op, __sub_group_carry, - __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); - return; - } + auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); + else + __write_op(__out_rng, __id, __v, __on_oob_reached); + }; + __scan_through_elements_impl<__is_inclusive>(__ndi, __gen_input_impl, __scan_input_transform, __binary_op, + __bounded_write_op, __sub_group_carry, __in_rng, __start_id, + __n, __iters_per_item, __subgroup_start_id, __comm_tag); + return; } - - auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data); - else - __write_op(__out_rng, __id, __v); - }; - __scan_through_elements_helper_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __binary_op, __unbounded_write_op, __sub_group_carry, - __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); } + + auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data); + else + __write_op(__out_rng, __id, __v); + }; + __scan_through_elements_impl<__is_inclusive>(__ndi, __gen_input_impl, __scan_input_transform, __binary_op, + __unbounded_write_op, __sub_group_carry, __in_rng, __start_id, + __n, __iters_per_item, __subgroup_start_id, __comm_tag); } template @@ -1817,7 +1805,7 @@ struct __parallel_reduce_then_scan_reduce_submitter<__is_inclusive, __is_unique_ _InitValueType* __temp_ptr = __temp_acc.__data(); // The sub-group-ops vs SLM-fallback decision is dispatched at each sub-group-scan region - // (see __scan_through_elements_helper and the carry-computation block below). + // (see __scan_through_elements and the carry-computation block below). const _ScanOpsTag __comm_scan_tag = __comm_handler.__get_tag_with_workspace(__comm_acc_or_placeholder); std::size_t __group_id = __ndi.get_group(0); std::uint32_t __sub_group_id = __sub_group.get_group_linear_id(); @@ -1841,12 +1829,11 @@ struct __parallel_reduce_then_scan_reduce_submitter<__is_inclusive, __is_unique_ if (__sub_group_id < __active_subgroups) { oneapi::dpl::__internal::__opt_lazy_ctor_storage<_InitValueType> __sub_group_carry; - // adjust for lane-id // compute sub-group local prefix on T0..63, K samples/T, send to accumulator kernel - __scan_through_elements_helper( + __scan_through_elements_impl<__is_inclusive>( __ndi, __gen_reduce_input, oneapi::dpl::identity{}, __reduce_op, - oneapi::dpl::__internal::__ignore_call_op{}, __sub_group_carry, __in_rng, /*unused*/ __in_rng, - __start_id, __n, __inputs_per_item, __subgroup_start_id, __comm_scan_tag); + oneapi::dpl::__internal::__ignore_call_op{}, __sub_group_carry, __in_rng, __start_id, __n, + __inputs_per_item, __subgroup_start_id, __comm_scan_tag); if (__sub_group_local_id == 0) __sub_group_partials[__sub_group_id] = __sub_group_carry.__get_cref(); } @@ -2182,8 +2169,8 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is __group_start_id + (std::size_t{__get_sub_group_base(__ndi)} * __inputs_per_item); std::size_t __start_id = __subgroup_start_id + __sub_group_local_id; - auto __call_scan_through_elements_helper = [&](auto __on_oob_reached, auto __final_pos_saver) { - __scan_through_elements_helper<_Bounded, __is_inclusive, __is_unique_pattern_v>( + auto __call_scan_through_elements = [&](auto __on_oob_reached, auto __final_pos_saver) { + __scan_through_elements<__is_inclusive, _Bounded, __is_unique_pattern_v>( __ndi, __gen_scan_input, __scan_input_transform, __reduce_op, __write_op, __sub_group_carry, __in_rng, __out_rng, __start_id, __n, __inputs_per_item, __subgroup_start_id, __comm_scan_tag, __on_oob_reached, __final_pos_saver); @@ -2206,7 +2193,7 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is if constexpr (__internal::__has_final_pos<__stop_pos_handler_type>) { using __final_pos_t = typename __stop_pos_handler_type::__final_pos_t; - __call_scan_through_elements_helper(__on_oob_reached, [&](__final_pos_t __final_pos) { + __call_scan_through_elements(__on_oob_reached, [&](__final_pos_t __final_pos) { // Exactly one work-item reaches the edge crossing, so no synchronization is needed // to store the shared final position. __stop_pos_acc_data.__final_pos = __final_pos; @@ -2221,15 +2208,14 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is } else { - __call_scan_through_elements_helper(__on_oob_reached, __internal::__no_callback_tag{}); + __call_scan_through_elements(__on_oob_reached, __internal::__no_callback_tag{}); if (__oob_detected) __stop_pos_acc_data = __oob_position; } } else { - __call_scan_through_elements_helper(__internal::__no_callback_tag{}, - __internal::__no_callback_tag{}); + __call_scan_through_elements(__internal::__no_callback_tag{}, __internal::__no_callback_tag{}); } } // If within the last active group and sub-group of the block, use the 0th work-item of the sub-group From bd4d7da2f66adf1dbbff7668302db228175dfd06 Mon Sep 17 00:00:00 2001 From: Alexey Kukanov Date: Fri, 18 Sep 2026 15:51:39 +0200 Subject: [PATCH 4/8] integrate __scan_through_elements into the scan submitter --- .../parallel_backend_sycl_reduce_then_scan.h | 157 +++++++++--------- 1 file changed, 77 insertions(+), 80 deletions(-) diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h index 5ac395ced19..78bfba0c509 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h @@ -1328,6 +1328,23 @@ struct __scan_by_seg_op _BinaryOp __binary_op; }; +// *** Traits of the input generators *** + +// Detecting TempData type alias in the specified structure +template +struct __temp_data_required +{ + static constexpr bool value = false; + using type = __noop_temp_data; +}; + +template +struct __temp_data_required<_T, std::void_t> +{ + static constexpr bool value = true; + using type = typename _T::TempData; +}; + // *** Main reduce then scan infrastructure *** // Sub-group communication wrappers with SLM fallback. @@ -1622,82 +1639,6 @@ __scan_through_elements_impl(const sycl::nd_item<1>& __ndi, _GenInput __gen_inpu }); } -// Detecting TempData type alias in the specified structure -template -struct __temp_data_required -{ - static constexpr bool value = false; - using type = __noop_temp_data; -}; - -template -struct __temp_data_required<_T, std::void_t> -{ - static constexpr bool value = true; - using type = typename _T::TempData; -}; - -// Group scan for the scan stage -template -void -__scan_through_elements(const sycl::nd_item<1>& __ndi, _GenInput __gen_input, - _ScanInputTransform __scan_input_transform, _BinaryOp __binary_op, _WriteOp __write_op, - oneapi::dpl::__internal::__opt_lazy_ctor_storage<_ValueType>& __sub_group_carry, - const _InRng& __in_rng, _OutRng& __out_rng, std::size_t __start_id, std::size_t __n, - std::uint32_t __iters_per_item, std::size_t __subgroup_start_id, _CommTag __comm_tag, - _OnOOBReached __on_oob_reached, _FinalPosSaver __final_pos_saver) -{ - using __temp_data_required_t = __temp_data_required<_GenInput>; - constexpr bool __is_temp_data_required = __temp_data_required_t::value; - - using _TempData = typename __temp_data_required_t::type; - _TempData __temp_data{}; - - auto __gen_input_impl = [&](const _InRng& __rng, std::size_t __id) { - if constexpr (__is_temp_data_required) - return __gen_input(__rng, __id, __temp_data, __final_pos_saver); - else - return __gen_input(__rng, __id); - }; - - if constexpr (__is_bounded) - { - const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); - // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: - // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned - // element is a diagonal written through __write_multiple_to_id. The estimate must account for - // this many writes per scanned element, otherwise the unchecked write path could be selected for - // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). - const std::size_t __max_write_offset = - std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; - if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) - { - auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); - else - __write_op(__out_rng, __id, __v, __on_oob_reached); - }; - __scan_through_elements_impl<__is_inclusive>(__ndi, __gen_input_impl, __scan_input_transform, __binary_op, - __bounded_write_op, __sub_group_carry, __in_rng, __start_id, - __n, __iters_per_item, __subgroup_start_id, __comm_tag); - return; - } - } - - auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data); - else - __write_op(__out_rng, __id, __v); - }; - __scan_through_elements_impl<__is_inclusive>(__ndi, __gen_input_impl, __scan_input_transform, __binary_op, - __unbounded_write_op, __sub_group_carry, __in_rng, __start_id, - __n, __iters_per_item, __subgroup_start_id, __comm_tag); -} - template struct __comm_slm_handler { @@ -1933,6 +1874,63 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is { using _InitValueType = typename _InitType::__value_type; + template + void + __scan_through_elements(const sycl::nd_item<1>& __ndi, + oneapi::dpl::__internal::__opt_lazy_ctor_storage<_InitValueType>& __sub_group_carry, + const _InRng& __in_rng, _OutRng& __out_rng, std::size_t __start_id, + std::uint32_t __iters_per_item, std::size_t __subgroup_start_id, _CommTag __comm_tag, + _OnOOBReached __on_oob_reached, _FinalPosSaver __final_pos_saver) const + { + using __temp_data_required_t = __temp_data_required<_GenScanInput>; + constexpr bool __is_temp_data_required = __temp_data_required_t::value; + + using _TempData = typename __temp_data_required_t::type; + _TempData __temp_data{}; + + auto __gen_input_impl = [&](const _InRng& __rng, std::size_t __id) { + if constexpr (__is_temp_data_required) + return __gen_scan_input(__rng, __id, __temp_data, __final_pos_saver); + else + return __gen_scan_input(__rng, __id); + }; + + if constexpr (_Bounded) + { + const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); + // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: + // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned + // element is a diagonal written through __write_multiple_to_id. The estimate must account for + // this many writes per scanned element, otherwise the unchecked write path could be selected for + // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). + const std::size_t __max_write_offset = + std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; + if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) + { + auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); + else + __write_op(__out_rng, __id, __v, __on_oob_reached); + }; + __scan_through_elements_impl<__is_inclusive>( + __ndi, __gen_input_impl, __scan_input_transform, __reduce_op, __bounded_write_op, __sub_group_carry, + __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + return; + } + } + + auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data); + else + __write_op(__out_rng, __id, __v); + }; + __scan_through_elements_impl<__is_inclusive>( + __ndi, __gen_input_impl, __scan_input_transform, __reduce_op, __unbounded_write_op, __sub_group_carry, + __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + } + template sycl::event operator()(sycl::queue& __q, const sycl::nd_range<1> __nd_range, _InRng&& __in_rng, _OutRng&& __out_rng, @@ -2170,10 +2168,9 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is std::size_t __start_id = __subgroup_start_id + __sub_group_local_id; auto __call_scan_through_elements = [&](auto __on_oob_reached, auto __final_pos_saver) { - __scan_through_elements<__is_inclusive, _Bounded, __is_unique_pattern_v>( - __ndi, __gen_scan_input, __scan_input_transform, __reduce_op, __write_op, __sub_group_carry, - __in_rng, __out_rng, __start_id, __n, __inputs_per_item, __subgroup_start_id, - __comm_scan_tag, __on_oob_reached, __final_pos_saver); + __scan_through_elements(__ndi, __sub_group_carry, __in_rng, __out_rng, __start_id, + __inputs_per_item, __subgroup_start_id, __comm_scan_tag, + __on_oob_reached, __final_pos_saver); }; if constexpr (_Bounded) From 6f44a2671cf1dfac4be0720e7c0f558d10ccc114 Mon Sep 17 00:00:00 2001 From: Alexey Kukanov Date: Fri, 18 Sep 2026 18:39:38 +0200 Subject: [PATCH 5/8] Move more code into __scan_through_elements --- .../parallel_backend_sycl_reduce_then_scan.h | 168 +++++++++--------- .../pstl/hetero/dpcpp/utils_storage_sycl.h | 2 + 2 files changed, 85 insertions(+), 85 deletions(-) diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h index 78bfba0c509..3e309601bf1 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h @@ -1874,61 +1874,104 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is { using _InitValueType = typename _InitType::__value_type; - template + template void __scan_through_elements(const sycl::nd_item<1>& __ndi, oneapi::dpl::__internal::__opt_lazy_ctor_storage<_InitValueType>& __sub_group_carry, const _InRng& __in_rng, _OutRng& __out_rng, std::size_t __start_id, std::uint32_t __iters_per_item, std::size_t __subgroup_start_id, _CommTag __comm_tag, - _OnOOBReached __on_oob_reached, _FinalPosSaver __final_pos_saver) const + _StopPosAcc __stop_pos_acc) const { using __temp_data_required_t = __temp_data_required<_GenScanInput>; + using _TempData = typename __temp_data_required_t::type; constexpr bool __is_temp_data_required = __temp_data_required_t::value; - using _TempData = typename __temp_data_required_t::type; - _TempData __temp_data{}; + auto __call_scan_through_elements = [&](auto __on_oob_reached, auto __final_pos_saver) { + _TempData __temp_data{}; - auto __gen_input_impl = [&](const _InRng& __rng, std::size_t __id) { - if constexpr (__is_temp_data_required) - return __gen_scan_input(__rng, __id, __temp_data, __final_pos_saver); - else - return __gen_scan_input(__rng, __id); + auto __gen_input_impl = [&](const _InRng& __rng, std::size_t __id) { + if constexpr (__is_temp_data_required) + return __gen_scan_input(__rng, __id, __temp_data, __final_pos_saver); + else + return __gen_scan_input(__rng, __id); + }; + + if constexpr (_Bounded) + { + const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); + // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: + // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned + // element is a diagonal written through __write_multiple_to_id. The estimate must account for + // this many writes per scanned element, otherwise the unchecked write path could be selected for + // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). + const std::size_t __max_write_offset = + std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; + if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) + { + auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); + else + __write_op(__out_rng, __id, __v, __on_oob_reached); + }; + __scan_through_elements_impl<__is_inclusive>( + __ndi, __gen_input_impl, __scan_input_transform, __reduce_op, __bounded_write_op, __sub_group_carry, + __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + return; + } + } + + auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data); + else + __write_op(__out_rng, __id, __v); + }; + __scan_through_elements_impl<__is_inclusive>( + __ndi, __gen_input_impl, __scan_input_transform, __reduce_op, __unbounded_write_op, __sub_group_carry, + __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); }; if constexpr (_Bounded) { - const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); - // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: - // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned - // element is a diagonal written through __write_multiple_to_id. The estimate must account for - // this many writes per scanned element, otherwise the unchecked write path could be selected for - // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). - const std::size_t __max_write_offset = - std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; - if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) - { - auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); - else - __write_op(__out_rng, __id, __v, __on_oob_reached); - }; - __scan_through_elements_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __reduce_op, __bounded_write_op, __sub_group_carry, - __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); - return; - } - } + std::size_t __start_id_on_oob = __start_id; + typename _WriteOp::__position_type __oob_position{}; + bool __oob_detected = false; + auto& __stop_pos_acc_data = __stop_pos_acc.__data()[0]; + + auto __on_oob_reached = [&](std::size_t __start_id, typename _WriteOp::__position_type __pos) { + __start_id_on_oob = __start_id; + __oob_position = __pos; + __oob_detected = true; + }; - auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data); + { + using __final_pos_t = typename _StopPosAcc::type::__final_pos_t; + __call_scan_through_elements(__on_oob_reached, [&](__final_pos_t __final_pos) { + // Exactly one work-item reaches the edge crossing, so no synchronization is needed + // to store the shared final position. + __stop_pos_acc_data.__final_pos = __final_pos; + }); + if (__oob_detected) + { + // Exactly one work-item reaches the OOB position, so no synchronization is needed + // to update __stop_pos_acc. + __stop_pos_acc_data.__oob_pos = __internal::__finalize_oob_pos<__final_pos_t>( + __in_rng, __oob_position, __start_id_on_oob, __gen_scan_input); + } + } else - __write_op(__out_rng, __id, __v); - }; - __scan_through_elements_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __reduce_op, __unbounded_write_op, __sub_group_carry, - __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + { + __call_scan_through_elements(__on_oob_reached, __internal::__no_callback_tag{}); + if (__oob_detected) + __stop_pos_acc_data = __oob_position; + } + } + else + { + __call_scan_through_elements(__internal::__no_callback_tag{}, __internal::__no_callback_tag{}); + } } template @@ -2167,53 +2210,8 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is __group_start_id + (std::size_t{__get_sub_group_base(__ndi)} * __inputs_per_item); std::size_t __start_id = __subgroup_start_id + __sub_group_local_id; - auto __call_scan_through_elements = [&](auto __on_oob_reached, auto __final_pos_saver) { - __scan_through_elements(__ndi, __sub_group_carry, __in_rng, __out_rng, __start_id, - __inputs_per_item, __subgroup_start_id, __comm_scan_tag, - __on_oob_reached, __final_pos_saver); - }; - - if constexpr (_Bounded) - { - std::size_t __start_id_on_oob = __start_id; - typename _WriteOp::__position_type __oob_position{}; - bool __oob_detected = false; - auto& __stop_pos_acc_data = __stop_pos_acc.__data()[0]; - - auto __on_oob_reached = [&](std::size_t __start_id, typename _WriteOp::__position_type __pos) { - __start_id_on_oob = __start_id; - __oob_position = __pos; - __oob_detected = true; - }; - - using __stop_pos_handler_type = typename _StopPosStorage::type; - if constexpr (__internal::__has_final_pos<__stop_pos_handler_type>) - { - using __final_pos_t = typename __stop_pos_handler_type::__final_pos_t; - __call_scan_through_elements(__on_oob_reached, [&](__final_pos_t __final_pos) { - // Exactly one work-item reaches the edge crossing, so no synchronization is needed - // to store the shared final position. - __stop_pos_acc_data.__final_pos = __final_pos; - }); - if (__oob_detected) - { - // Exactly one work-item reaches the OOB position, so no synchronization is needed - // to update __stop_pos_acc. - __stop_pos_acc_data.__oob_pos = __internal::__finalize_oob_pos<__final_pos_t>( - __in_rng, __oob_position, __start_id_on_oob, __gen_scan_input); - } - } - else - { - __call_scan_through_elements(__on_oob_reached, __internal::__no_callback_tag{}); - if (__oob_detected) - __stop_pos_acc_data = __oob_position; - } - } - else - { - __call_scan_through_elements(__internal::__no_callback_tag{}, __internal::__no_callback_tag{}); - } + __scan_through_elements(__ndi, __sub_group_carry, __in_rng, __out_rng, __start_id, + __inputs_per_item, __subgroup_start_id, __comm_scan_tag, __stop_pos_acc); } // If within the last active group and sub-group of the block, use the 0th work-item of the sub-group // to write out the last carry out for either the return value or the next block diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/utils_storage_sycl.h b/include/oneapi/dpl/pstl/hetero/dpcpp/utils_storage_sycl.h index 9d463492831..52245d4ee65 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/utils_storage_sycl.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/utils_storage_sycl.h @@ -271,6 +271,8 @@ struct __combi_accessor } public: + using type = _T; + __combi_accessor(sycl::handler& __cgh, sycl::buffer<_T, 1>& __sycl_buf, _T* __usm_buf, const sycl::property_list& __prop_list) : __ptr(__usm_buf), __acc(__make_accessor(__usm_buf != nullptr, __sycl_buf, __cgh, __prop_list)) From 9605e58a6142d7f276d5aa9443938651f441cda8 Mon Sep 17 00:00:00 2001 From: Alexey Kukanov Date: Fri, 18 Sep 2026 20:20:56 +0200 Subject: [PATCH 6/8] flatten __scan_through_elements --- .../parallel_backend_sycl_reduce_then_scan.h | 137 ++++++++---------- ..._backend_sycl_reduce_then_scan_pos_tools.h | 7 - .../pstl/hetero/dpcpp/utils_storage_sycl.h | 2 +- 3 files changed, 65 insertions(+), 81 deletions(-) diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h index 3e309601bf1..ffaad7cc610 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan.h @@ -1884,94 +1884,85 @@ struct __parallel_reduce_then_scan_scan_submitter<_Bounded, __is_inclusive, __is { using __temp_data_required_t = __temp_data_required<_GenScanInput>; using _TempData = typename __temp_data_required_t::type; - constexpr bool __is_temp_data_required = __temp_data_required_t::value; - - auto __call_scan_through_elements = [&](auto __on_oob_reached, auto __final_pos_saver) { - _TempData __temp_data{}; - auto __gen_input_impl = [&](const _InRng& __rng, std::size_t __id) { - if constexpr (__is_temp_data_required) - return __gen_scan_input(__rng, __id, __temp_data, __final_pos_saver); - else - return __gen_scan_input(__rng, __id); - }; + constexpr bool __is_temp_data_required = __temp_data_required_t::value; + _TempData __temp_data{}; - if constexpr (_Bounded) + auto __gen_input = [&](const _InRng& __rng, std::size_t __id) { + if constexpr (_Bounded && __is_temp_data_required) { - const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); - // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: - // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned - // element is a diagonal written through __write_multiple_to_id. The estimate must account for - // this many writes per scanned element, otherwise the unchecked write path could be selected for - // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). - const std::size_t __max_write_offset = - std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; - if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) - { - auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); - else - __write_op(__out_rng, __id, __v, __on_oob_reached); - }; - __scan_through_elements_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __reduce_op, __bounded_write_op, __sub_group_carry, - __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); - return; - } + using __final_pos_t = typename _StopPosAcc::type::__final_pos_t; + auto __final_pos_saver = [&](__final_pos_t __final_pos) { + // Exactly one work-item reaches the edge crossing, so no synchronization is needed + // to store the shared final position. + __stop_pos_acc.__data()[0].__final_pos = __final_pos; + }; + return __gen_scan_input(__rng, __id, __temp_data, __final_pos_saver); } - - auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { - if constexpr (__is_temp_data_required) - __write_op(__out_rng, __id, __v, __temp_data); - else - __write_op(__out_rng, __id, __v); - }; - __scan_through_elements_impl<__is_inclusive>( - __ndi, __gen_input_impl, __scan_input_transform, __reduce_op, __unbounded_write_op, __sub_group_carry, - __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + else if constexpr (__is_temp_data_required) + return __gen_scan_input(__rng, __id, __temp_data, __internal::__no_callback_tag{}); + else + return __gen_scan_input(__rng, __id); }; if constexpr (_Bounded) { - std::size_t __start_id_on_oob = __start_id; - typename _WriteOp::__position_type __oob_position{}; - bool __oob_detected = false; - auto& __stop_pos_acc_data = __stop_pos_acc.__data()[0]; - - auto __on_oob_reached = [&](std::size_t __start_id, typename _WriteOp::__position_type __pos) { - __start_id_on_oob = __start_id; - __oob_position = __pos; - __oob_detected = true; - }; - - if constexpr (__is_temp_data_required) + // A single scanned element may emit up to _TempData::__max_outputs_per_input output elements: + // one for copy_if/unique, but up to __diagonal_spacing for set operations, where each scanned + // element is a diagonal written through __write_multiple_to_id. The estimate must account for + // this many writes per scanned element, otherwise the unchecked write path could be selected for + // set operations and overrun __out_rng (corrupting memory and skipping OOB position detection). + const std::uint8_t __sg_size = __get_reduce_then_scan_actual_sub_group_size(__ndi.get_sub_group()); + const std::size_t __max_write_offset = + std::size_t{__is_unique_pattern_v} + __iters_per_item * __sg_size * _TempData::__max_outputs_per_input; + + if (__write_op.__oob_write_possible(__max_write_offset, __subgroup_start_id, __sub_group_carry)) { - using __final_pos_t = typename _StopPosAcc::type::__final_pos_t; - __call_scan_through_elements(__on_oob_reached, [&](__final_pos_t __final_pos) { - // Exactly one work-item reaches the edge crossing, so no synchronization is needed - // to store the shared final position. - __stop_pos_acc_data.__final_pos = __final_pos; - }); + bool __oob_detected = false; + std::size_t __start_id_on_oob = __start_id; + typename _WriteOp::__position_type __oob_position{}; + + auto __on_oob_reached = [&](std::size_t __sid, typename _WriteOp::__position_type __pos) { + __oob_detected = true; + __start_id_on_oob = __sid; + __oob_position = __pos; + }; + auto __bounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data, __on_oob_reached); + else + __write_op(__out_rng, __id, __v, __on_oob_reached); + }; + __scan_through_elements_impl<__is_inclusive>( + __ndi, __gen_input, __scan_input_transform, __reduce_op, __bounded_write_op, __sub_group_carry, + __in_rng, __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); + if (__oob_detected) { + auto& __stop_pos_acc_data = __stop_pos_acc.__data()[0]; // Exactly one work-item reaches the OOB position, so no synchronization is needed - // to update __stop_pos_acc. - __stop_pos_acc_data.__oob_pos = __internal::__finalize_oob_pos<__final_pos_t>( - __in_rng, __oob_position, __start_id_on_oob, __gen_scan_input); + // to update __stop_pos_acc_data. + if constexpr (__is_temp_data_required) + { + using __final_pos_t = typename _StopPosAcc::type::__final_pos_t; + __stop_pos_acc_data.__oob_pos = __internal::__finalize_oob_pos<__final_pos_t>( + __in_rng, __oob_position, __start_id_on_oob, __gen_scan_input); + } + else + __stop_pos_acc_data = __oob_position; } + return; } - else - { - __call_scan_through_elements(__on_oob_reached, __internal::__no_callback_tag{}); - if (__oob_detected) - __stop_pos_acc_data = __oob_position; - } - } - else - { - __call_scan_through_elements(__internal::__no_callback_tag{}, __internal::__no_callback_tag{}); } + auto __unbounded_write_op = [&](std::size_t __id, const auto& __v) { + if constexpr (__is_temp_data_required) + __write_op(__out_rng, __id, __v, __temp_data); + else + __write_op(__out_rng, __id, __v); + }; + __scan_through_elements_impl<__is_inclusive>( + __ndi, __gen_input, __scan_input_transform, __reduce_op, __unbounded_write_op, __sub_group_carry, __in_rng, + __start_id, __n, __iters_per_item, __subgroup_start_id, __comm_tag); } template diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan_pos_tools.h b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan_pos_tools.h index e59f7b51301..32da847b9bb 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan_pos_tools.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl_reduce_then_scan_pos_tools.h @@ -83,13 +83,6 @@ using _SetOpFinalAndOOBPosType = _SetOpFinalAndOOBPosTypeImpl -inline constexpr bool __has_final_pos = false; - -template -inline constexpr bool __has_final_pos<_SetOpFinalAndOOBPosTypeImpl<_Range1, _Range2>> = true; - // Temporary data stand-in which discards the stored values and instead captures // the source position of the element at a specific index during a reduce then scan operation. template diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/utils_storage_sycl.h b/include/oneapi/dpl/pstl/hetero/dpcpp/utils_storage_sycl.h index 52245d4ee65..df8eab9ea15 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/utils_storage_sycl.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/utils_storage_sycl.h @@ -272,7 +272,7 @@ struct __combi_accessor public: using type = _T; - + __combi_accessor(sycl::handler& __cgh, sycl::buffer<_T, 1>& __sycl_buf, _T* __usm_buf, const sycl::property_list& __prop_list) : __ptr(__usm_buf), __acc(__make_accessor(__usm_buf != nullptr, __sycl_buf, __cgh, __prop_list)) From c6582339ac74a95ab2efac62b6b5b3ca11cd6f00 Mon Sep 17 00:00:00 2001 From: Alexey Kukanov Date: Fri, 11 Sep 2026 20:24:25 +0200 Subject: [PATCH 7/8] improve file structure --- .../pstl/hetero/dpcpp/parallel_backend_sycl.h | 295 +++++++++--------- 1 file changed, 152 insertions(+), 143 deletions(-) diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl.h b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl.h index d226d647d97..0ca6c446d67 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/parallel_backend_sycl.h @@ -100,7 +100,7 @@ __parallel_copy_impl(sycl::queue& __q, _Index __count, _Range1&& __rng1, _Range2 } //------------------------------------------------------------------------ -// parallel_transform_scan single group - async pattern +// parallel_transform_scan - async pattern //------------------------------------------------------------------------ template - static std::pair, std::make_unsigned_t<_Size>> - __local_memory_needed(_Size __n) - { - // Next power of 2 greater than or equal to __n - std::make_unsigned_t<_Size> __n_uniform = - oneapi::dpl::__internal::__dpl_bit_ceil(static_cast>(__n)); - // The kernel needs memory for: N predicate evaluations, N output offsets, and the input stop position - return {__n_uniform * 2 + 1, __n_uniform}; - } - - template - static bool - __enough_local_memory(sycl::queue __q, _Size __n) - { - // Pessimistically expect only half of local memory to account for possible memory use by the compiled code - std::size_t __available_size = __q.get_device().template get_info() / 2; - return __available_size >= __local_memory_needed(__n).first * sizeof(_ValueType); - } -}; - -template -struct __parallel_copy_if_single_group_functor; - -template -struct __parallel_copy_if_single_group_functor<__internal::__optional_kernel_name<_ScanKernelName...>> - : __parallel_copy_if_single_group_base -{ - template - std::array<_Size, 2> - operator()(sycl::queue& __q, _InRng&& __in_rng, _OutRng&& __out_rng, _Size __n, _Size __n_out, _IndexPred __pred, - _Assign __assign, std::size_t __max_wg_size) - { - assert(__max_wg_size <= std::numeric_limits::max()); - // This type is used as a workaround for when an internal tuple is assigned to std::tuple, such as - // with zip_iterator - using __tuple_type = typename oneapi::dpl::__internal::__get_tuple_type< - std::decay_t, std::decay_t>::__type; - - __result_storage<_Size> __result{__q, 2}; - - __q.submit([&](sycl::handler& __hdl) { - oneapi::dpl::__ranges::__require_access(__hdl, __in_rng, __out_rng); - - std::make_unsigned_t<_Size> __lsize, __n_uniform; - // Since __n_uniform is captured into a lambda, structured binding cannot be used here till C++20 - std::tie(__lsize, __n_uniform) = __local_memory_needed(__n); - auto __lacc = __dpl_sycl::__local_accessor<_ValueType>(sycl::range<1>(__lsize), __hdl); - auto __res_acc = __get_accessor(sycl::write_only, __result, __hdl, __dpl_sycl::__no_init{}); - const auto __wg_size = static_cast(std::min(__n_uniform, __max_wg_size)); - - __hdl.parallel_for<_ScanKernelName...>(sycl::nd_range<1>(__wg_size, __wg_size), - [=](sycl::nd_item<1> __self_item) { - sycl::group __group = __self_item.get_group(); - // This kernel is only launched for sizes less than 2^16 - const std::uint16_t __item_id = __self_item.get_local_linear_id(); - _ValueType* __lacc_ptr = __dpl_sycl::__get_accessor_ptr(__lacc); - for (std::uint16_t __idx = __item_id; __idx < __n; __idx += __wg_size) - { - __lacc[__idx] = __pred(__in_rng, __idx); - } - if (__item_id == 0) - { - // Store the input size as the expected stop position - __lacc[2 * __n_uniform] = __n; - } - - __scan_work_group<_ValueType, /* _Inclusive */ false>( - __group, __lacc_ptr, __lacc_ptr + __n, __lacc_ptr + __n_uniform, sycl::plus<_ValueType>{}); - - for (std::uint16_t __idx = __item_id; __idx < __n; __idx += __wg_size) - { - if (__lacc[__idx]) { - _ValueType __out_idx = __lacc[__idx + __n_uniform]; - if (__out_idx < __n_out) - __assign(static_cast<__tuple_type>(__in_rng[__idx]), __out_rng[__out_idx]); - if (__out_idx == __n_out) - __lacc[2 * __n_uniform] = __idx; // the actual stop position in the input - } - } - sycl::group_barrier(__group); - - if (__item_id == 0) - { - _Size* __res_ptr = __res_acc.__data(); - _ValueType __stop_in = __lacc[2 * __n_uniform]; - __res_ptr[1] = __stop_in; - // Add predicate of last element to account for the scan's exclusivity - __res_ptr[0] = (__stop_in == __n) ? __lacc[__n_uniform + __n - 1] + __lacc[__n - 1] : __n_out; - } - }); - }).wait_and_throw(); - - std::array<_Size, 2> __ret; - __result.__copy_result(__ret.data(), __ret.size()); - return __ret; - } -}; - template sycl::event @@ -448,6 +345,113 @@ __parallel_transform_scan(oneapi::dpl::__internal::__device_backend_tag, _Execut return __future(std::move(__event), std::move(__holder).__extract()); } +//------------------------------------------------------------------------ +// Filtering patterns: copy_if, unique_copy, etc.; also partition_copy +//------------------------------------------------------------------------ + +struct __parallel_copy_if_single_group_base +{ + using _ValueType = std::uint16_t; + + template + static std::pair, std::make_unsigned_t<_Size>> + __local_memory_needed(_Size __n) + { + // Next power of 2 greater than or equal to __n + std::make_unsigned_t<_Size> __n_uniform = + oneapi::dpl::__internal::__dpl_bit_ceil(static_cast>(__n)); + // The kernel needs memory for: N predicate evaluations, N output offsets, and the input stop position + return {__n_uniform * 2 + 1, __n_uniform}; + } + + template + static bool + __enough_local_memory(sycl::queue __q, _Size __n) + { + // Pessimistically expect only half of local memory to account for possible memory use by the compiled code + std::size_t __available_size = __q.get_device().template get_info() / 2; + return __available_size >= __local_memory_needed(__n).first * sizeof(_ValueType); + } +}; + +template +struct __parallel_copy_if_single_group_functor; + +template +struct __parallel_copy_if_single_group_functor<__internal::__optional_kernel_name<_ScanKernelName...>> + : __parallel_copy_if_single_group_base +{ + template + std::array<_Size, 2> + operator()(sycl::queue& __q, _InRng&& __in_rng, _OutRng&& __out_rng, _Size __n, _Size __n_out, _IndexPred __pred, + _Assign __assign, std::size_t __max_wg_size) + { + assert(__max_wg_size <= std::numeric_limits::max()); + // This type is used as a workaround for when an internal tuple is assigned to std::tuple, such as + // with zip_iterator + using __tuple_type = typename oneapi::dpl::__internal::__get_tuple_type< + std::decay_t, std::decay_t>::__type; + + __result_storage<_Size> __result{__q, 2}; + + __q.submit([&](sycl::handler& __hdl) { + oneapi::dpl::__ranges::__require_access(__hdl, __in_rng, __out_rng); + + std::make_unsigned_t<_Size> __lsize, __n_uniform; + // Since __n_uniform is captured into a lambda, structured binding cannot be used here till C++20 + std::tie(__lsize, __n_uniform) = __local_memory_needed(__n); + auto __lacc = __dpl_sycl::__local_accessor<_ValueType>(sycl::range<1>(__lsize), __hdl); + auto __res_acc = __get_accessor(sycl::write_only, __result, __hdl, __dpl_sycl::__no_init{}); + const auto __wg_size = static_cast(std::min(__n_uniform, __max_wg_size)); + + __hdl.parallel_for<_ScanKernelName...>(sycl::nd_range<1>(__wg_size, __wg_size), + [=](sycl::nd_item<1> __self_item) { + sycl::group __group = __self_item.get_group(); + // This kernel is only launched for sizes less than 2^16 + const std::uint16_t __item_id = __self_item.get_local_linear_id(); + _ValueType* __lacc_ptr = __dpl_sycl::__get_accessor_ptr(__lacc); + for (std::uint16_t __idx = __item_id; __idx < __n; __idx += __wg_size) + { + __lacc[__idx] = __pred(__in_rng, __idx); + } + if (__item_id == 0) + { + // Store the input size as the expected stop position + __lacc[2 * __n_uniform] = __n; + } + + __scan_work_group<_ValueType, /* _Inclusive */ false>( + __group, __lacc_ptr, __lacc_ptr + __n, __lacc_ptr + __n_uniform, sycl::plus<_ValueType>{}); + + for (std::uint16_t __idx = __item_id; __idx < __n; __idx += __wg_size) + { + if (__lacc[__idx]) { + _ValueType __out_idx = __lacc[__idx + __n_uniform]; + if (__out_idx < __n_out) + __assign(static_cast<__tuple_type>(__in_rng[__idx]), __out_rng[__out_idx]); + if (__out_idx == __n_out) + __lacc[2 * __n_uniform] = __idx; // the actual stop position in the input + } + } + sycl::group_barrier(__group); + + if (__item_id == 0) + { + _Size* __res_ptr = __res_acc.__data(); + _ValueType __stop_in = __lacc[2 * __n_uniform]; + __res_ptr[1] = __stop_in; + // Add predicate of last element to account for the scan's exclusivity + __res_ptr[0] = (__stop_in == __n) ? __lacc[__n_uniform + __n - 1] + __lacc[__n - 1] : __n_out; + } + }); + }).wait_and_throw(); + + std::array<_Size, 2> __ret; + __result.__copy_result(__ret.data(), __ret.size()); + return __ret; + } +}; + template std::array<_Size, 2> @@ -522,45 +526,6 @@ __parallel_unique_copy(oneapi::dpl::__internal::__device_backend_tag, _Execution return __ret; } -template -auto /*__future*/ -__parallel_reduce_by_segment_reduce_then_scan(sycl::queue& __q, _Range1&& __keys, _Range2&& __values, - _Range3&& __out_keys, _Range4&& __out_values, - _BinaryPredicate __binary_pred, _BinaryOperator __binary_op) -{ - // Flags new segments and passes input value through a 2-tuple - using _GenReduceInput = __gen_red_by_seg_reduce_input<_BinaryPredicate>; - // Operation that computes output indices and output reduction values per segment - using _ReduceOp = __red_by_seg_op<_BinaryOperator>; - // Returns 4-component tuple which contains flags, keys, value, and a flag to write output - using _GenScanInput = __gen_red_by_seg_scan_input<_BinaryPredicate>; - // Returns the first component from scan input which is scanned over - using _ScanInputTransform = __get_zeroth_element; - // Writes current segment's output reduction and the next segment's output key - using _WriteOp = __write_red_by_seg<_BinaryPredicate>; - using _KeyType = oneapi::dpl::__internal::__value_t<_Range1>; - using _ValueType = oneapi::dpl::__internal::__value_t<_Range2>; - using _ResultType = oneapi::dpl::__internal::tuple; - - std::size_t __n = oneapi::dpl::__ranges::__size(__keys); - // __gen_red_by_seg_scan_input requires that __n > 1 - assert(__n > 1); - __transform_scan_storage_holder_simple<_ResultType> __holder(__q); - // Each work-item iteration reads one key and one value from the zipped input. The comparison against the previous - // key is not counted separately, as that key is read by the adjacent index's iteration. - constexpr std::uint32_t __bytes_per_work_item_iter = sizeof(_KeyType) + sizeof(_ValueType); - - sycl::event __event = __parallel_transform_reduce_then_scan< - /*_Bounded*/ false, __bytes_per_work_item_iter, _CustomName>( - __q, __n, oneapi::dpl::__ranges::make_zip_view(std::forward<_Range1>(__keys), std::forward<_Range2>(__values)), - oneapi::dpl::__ranges::make_zip_view(std::forward<_Range3>(__out_keys), std::forward<_Range4>(__out_values)), - _GenReduceInput{__binary_pred}, _ReduceOp{__binary_op}, _GenScanInput{__binary_pred, __n}, - _ScanInputTransform{}, _WriteOp{__binary_pred, __n}, oneapi::dpl::unseq_backend::__no_init_value<_ResultType>{}, - __holder, /*Inclusive*/ std::true_type{}, /*_IsUniquePattern=*/std::false_type{}); - return __future(std::move(__event), std::move(__holder).__extract()); -} - template std::array @@ -655,6 +620,10 @@ __parallel_copy_if(oneapi::dpl::__internal::__device_backend_tag, _ExecutionPoli return __ret; } +//------------------------------------------------------------------------ +// Set operations +//------------------------------------------------------------------------ + // balanced path template @@ -1497,6 +1466,46 @@ __parallel_partial_sort(oneapi::dpl::__internal::__device_backend_tag, _Executio // inability to create event dependency chains across separate parallel pattern calls. If we ever add support for // cross parallel pattern dependencies, then we can implement this as an async pattern. //------------------------------------------------------------------------ + +template +auto /*__future*/ +__parallel_reduce_by_segment_reduce_then_scan(sycl::queue& __q, _Range1&& __keys, _Range2&& __values, + _Range3&& __out_keys, _Range4&& __out_values, + _BinaryPredicate __binary_pred, _BinaryOperator __binary_op) +{ + // Flags new segments and passes input value through a 2-tuple + using _GenReduceInput = __gen_red_by_seg_reduce_input<_BinaryPredicate>; + // Operation that computes output indices and output reduction values per segment + using _ReduceOp = __red_by_seg_op<_BinaryOperator>; + // Returns 4-component tuple which contains flags, keys, value, and a flag to write output + using _GenScanInput = __gen_red_by_seg_scan_input<_BinaryPredicate>; + // Returns the first component from scan input which is scanned over + using _ScanInputTransform = __get_zeroth_element; + // Writes current segment's output reduction and the next segment's output key + using _WriteOp = __write_red_by_seg<_BinaryPredicate>; + using _KeyType = oneapi::dpl::__internal::__value_t<_Range1>; + using _ValueType = oneapi::dpl::__internal::__value_t<_Range2>; + using _ResultType = oneapi::dpl::__internal::tuple; + + std::size_t __n = oneapi::dpl::__ranges::__size(__keys); + // __gen_red_by_seg_scan_input requires that __n > 1 + assert(__n > 1); + __transform_scan_storage_holder_simple<_ResultType> __holder(__q); + // Each work-item iteration reads one key and one value from the zipped input. The comparison against the previous + // key is not counted separately, as that key is read by the adjacent index's iteration. + constexpr std::uint32_t __bytes_per_work_item_iter = sizeof(_KeyType) + sizeof(_ValueType); + + sycl::event __event = __parallel_transform_reduce_then_scan< + /*_Bounded*/ false, __bytes_per_work_item_iter, _CustomName>( + __q, __n, oneapi::dpl::__ranges::make_zip_view(std::forward<_Range1>(__keys), std::forward<_Range2>(__values)), + oneapi::dpl::__ranges::make_zip_view(std::forward<_Range3>(__out_keys), std::forward<_Range4>(__out_values)), + _GenReduceInput{__binary_pred}, _ReduceOp{__binary_op}, _GenScanInput{__binary_pred, __n}, + _ScanInputTransform{}, _WriteOp{__binary_pred, __n}, oneapi::dpl::unseq_backend::__no_init_value<_ResultType>{}, + __holder, /*Inclusive*/ std::true_type{}, /*_IsUniquePattern=*/std::false_type{}); + return __future(std::move(__event), std::move(__holder).__extract()); +} + template struct __reduce1_wrapper; From bd47d73b9c8a81fb6ba4325f7daf924c34d6ba55 Mon Sep 17 00:00:00 2001 From: Alexey Kukanov Date: Fri, 11 Sep 2026 21:38:47 +0200 Subject: [PATCH 8/8] simplify _require_access_args --- .../dpl/pstl/hetero/dpcpp/utils_ranges_sycl.h | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/include/oneapi/dpl/pstl/hetero/dpcpp/utils_ranges_sycl.h b/include/oneapi/dpl/pstl/hetero/dpcpp/utils_ranges_sycl.h index 1db4c73f303..97ec8ddad6f 100644 --- a/include/oneapi/dpl/pstl/hetero/dpcpp/utils_ranges_sycl.h +++ b/include/oneapi/dpl/pstl/hetero/dpcpp/utils_ranges_sycl.h @@ -285,15 +285,14 @@ template void __require_access(sycl::handler& __cgh, _Range&& __rng, _Ranges&&... __rest); -template struct _require_access_args { - _Cgh __cgh; + sycl::handler& __cgh; template void operator()(Args&&... args) { - __require_access(__cgh, ::std::forward(args)...); + __require_access(__cgh, std::forward(args)...); } }; @@ -301,9 +300,8 @@ template void __require_access_zip(sycl::handler& __cgh, _dpl_ranges_zip::zip_view<_Ranges...>& __zip) { - const ::std::size_t __num_ranges = sizeof...(_Ranges); - oneapi::dpl::__ranges::invoke(__zip.base(), _require_access_args{__cgh}, - ::std::make_index_sequence<__num_ranges>()); + const std::size_t __num_ranges = sizeof...(_Ranges); + oneapi::dpl::__ranges::invoke(__zip.base(), _require_access_args{__cgh}, std::make_index_sequence<__num_ranges>()); } //__require_access utility @@ -331,9 +329,8 @@ template void __require_access_range(sycl::handler& __cgh, oneapi::dpl::__internal::tuple<_Ranges...>& __tuple) { - const ::std::size_t __num_ranges = sizeof...(_Ranges); - oneapi::dpl::__ranges::invoke(__tuple, _require_access_args{__cgh}, - ::std::make_index_sequence<__num_ranges>()); + const std::size_t __num_ranges = sizeof...(_Ranges); + oneapi::dpl::__ranges::invoke(__tuple, _require_access_args{__cgh}, std::make_index_sequence<__num_ranges>()); } template