Alpaka: Parallel Algorithm Unification (#1074)

* Add thrust/oneDPL abstraction, for Alpaka.

Hides the selection of a execution policy, as well as the specific
algorithm in use.

* Update algorithms to use new wrapper.

* Fixes for SYCL + CPU backend.

The memory resource isn't used for SYCL, and neither are used for CPU.

* Fix template deduction fail with older HIP compiler.

* Address PR comments.

* Convert missed thrust call.

---------

Co-authored-by: Stephen Nicholas Swatman <[email protected]>

Author: CrossR

device/alpaka/CMakeLists.txt

@@ -35,6 +35,8 @@ traccc_add_alpaka_library( traccc_alpaka alpaka TYPE SHARED
   "src/utils/get_queue.cpp"
   "include/traccc/alpaka/utils/vecmem_objects.hpp"
   "src/utils/vecmem_objects.cpp"
+  "src/utils/oneDPL.hpp"
+  "src/utils/parallel_algorithms.hpp"
   # Clusterization
   "include/traccc/alpaka/clusterization/clusterization_algorithm.hpp"
   "src/clusterization/clusterization_algorithm.cpp"

device/alpaka/src/clusterization/measurement_sorting_algorithm.cpp

@@ -9,10 +9,7 @@
 #include "traccc/alpaka/clusterization/measurement_sorting_algorithm.hpp"
 
 #include "../utils/get_queue.hpp"
-
-// Thrust include(s).
-#include <thrust/execution_policy.h>
-#include <thrust/sort.h>
+#include "../utils/parallel_algorithms.hpp"
 
 // System include(s).
 #include <memory_resource>
@@ -33,29 +30,14 @@ measurement_sorting_algorithm::operator()(
 
     // Get the number of measurements. This is necessary because the input
     // container may not be fixed sized. And we can't give invalid pointers /
-    // iterators to Thrust.
+    // iterators to Thrust / oneDPL.
     const measurement_collection_types::view::size_type n_measurements =
         m_copy.get().get_size(measurements_view);
 
     // Sort the measurements in place
-#if defined(ALPAKA_ACC_GPU_CUDA_ENABLED)
-    auto stream = ::alpaka::getNativeHandle(queue);
-    auto execPolicy =
-        thrust::cuda::par_nosync(std::pmr::polymorphic_allocator(&(m_mr.main)))
-            .on(stream);
-#elif defined(ALPAKA_ACC_GPU_HIP_ENABLED)
-    auto stream = ::alpaka::getNativeHandle(queue);
-    auto execPolicy =
-        thrust::hip_rocprim::par_nosync(
-            std::pmr::polymorphic_allocator<std::byte>(&(m_mr.main)))
-            .on(stream);
-#else
-    auto execPolicy = thrust::host;
-#endif
-
-    thrust::sort(execPolicy, measurements_view.ptr(),
-                 measurements_view.ptr() + n_measurements,
-                 measurement_sort_comp());
+    details::sort(queue, m_mr, measurements_view.ptr(),
+                  measurements_view.ptr() + n_measurements,
+                  measurement_sort_comp());
 
     // Return the view of the sorted measurements.
     return measurements_view;

device/alpaka/src/finding/combinatorial_kalman_filter.hpp

@@ -9,6 +9,7 @@
 
 // Local include(s).
 #include "../utils/barrier.hpp"
+#include "../utils/parallel_algorithms.hpp"
 #include "../utils/thread_id.hpp"
 #include "../utils/utils.hpp"
 
@@ -36,14 +37,6 @@
 // VecMem include(s).
 #include <vecmem/utils/copy.hpp>
 
-// Thrust include(s).
-#include <thrust/copy.h>
-#include <thrust/execution_policy.h>
-#include <thrust/fill.h>
-#include <thrust/scan.h>
-#include <thrust/sort.h>
-#include <thrust/unique.h>
-
 namespace traccc::alpaka::details {
 namespace kernels {
 
@@ -219,13 +212,6 @@ combinatorial_kalman_filter(
     // Create a logger.
     auto logger = [&log]() -> const Logger& { return log; };
 
-    /// Thrust policy to use.
-#if defined(ALPAKA_ACC_GPU_CUDA_ENABLED) || defined(ALPAKA_ACC_GPU_HIP_ENABLED)
-    auto thrustExecPolicy = thrust::device;
-#else
-    auto thrustExecPolicy = thrust::host;
-#endif
-
     // Number of threads per block to use.
     const Idx threadsPerBlock = getWarpSize<Acc>() * 2;
 
@@ -243,9 +229,9 @@ combinatorial_kalman_filter(
     measurement_collection_types::device uniques(uniques_buffer);
 
     measurement_collection_types::device::iterator uniques_end =
-        thrust::unique_copy(thrustExecPolicy, measurements.ptr(),
-                            measurements.ptr() + n_measurements,
-                            uniques.begin(), measurement_equal_comp());
+        details::unique_copy(queue, mr, measurements.ptr(),
+                             measurements.ptr() + n_measurements,
+                             uniques.begin(), measurement_equal_comp());
     const unsigned int n_modules =
         static_cast<unsigned int>(uniques_end - uniques.begin());
 
@@ -255,10 +241,10 @@ combinatorial_kalman_filter(
     copy.setup(upper_bounds_buffer)->wait();
     vecmem::device_vector<unsigned int> upper_bounds(upper_bounds_buffer);
 
-    thrust::upper_bound(thrustExecPolicy, measurements.ptr(),
-                        measurements.ptr() + n_measurements, uniques.begin(),
-                        uniques.begin() + n_modules, upper_bounds.begin(),
-                        measurement_sort_comp());
+    details::upper_bound(queue, mr, measurements.ptr(),
+                         measurements.ptr() + n_measurements, uniques.begin(),
+                         uniques.begin() + n_modules, upper_bounds.begin(),
+                         measurement_sort_comp());
 
     /*****************************************************************
      * Kernel1: Create barcode sequence
@@ -479,8 +465,8 @@ combinatorial_kalman_filter(
                 link_last_measurement_buffer);
             vecmem::device_vector<unsigned int> param_ids_device(
                 param_ids_buffer);
-            thrust::sort_by_key(thrustExecPolicy, keys_device.begin(),
-                                keys_device.end(), param_ids_device.begin());
+            details::sort_by_key(queue, mr, keys_device.begin(),
+                                 keys_device.end(), param_ids_device.begin());
 
             /*
              * Then, we run the actual duplicate removal kernel.
@@ -542,9 +528,9 @@ combinatorial_kalman_filter(
                     keys_buffer);
                 vecmem::device_vector<unsigned int> param_ids_device(
                     param_ids_buffer);
-                thrust::sort_by_key(thrustExecPolicy, keys_device.begin(),
-                                    keys_device.end(),
-                                    param_ids_device.begin());
+                details::sort_by_key(queue, mr, keys_device.begin(),
+                                     keys_device.end(),
+                                     param_ids_device.begin());
             }
 
             /*****************************************************************

device/alpaka/src/fitting/kalman_fitting.hpp

@@ -8,6 +8,7 @@
 #pragma once
 
 // Local include(s).
+#include "../utils/parallel_algorithms.hpp"
 #include "../utils/utils.hpp"
 
 // Project include(s).
@@ -26,10 +27,6 @@
 // VecMem include(s).
 #include <vecmem/utils/copy.hpp>
 
-// Thrust include(s).
-#include <thrust/execution_policy.h>
-#include <thrust/sort.h>
-
 namespace traccc::alpaka::details {
 namespace kernels {
 
@@ -122,13 +119,6 @@ track_state_container_types::buffer kalman_fitting(
     const fitting_config& config, const memory_resource& mr, vecmem::copy& copy,
     Queue& queue) {
 
-    /// Thrust policy to use.
-#if defined(ALPAKA_ACC_GPU_CUDA_ENABLED) || defined(ALPAKA_ACC_GPU_HIP_ENABLED)
-    auto thrustExecPolicy = thrust::device;
-#else
-    auto thrustExecPolicy = thrust::host;
-#endif
-
     // Number of threads per block to use.
     const Idx threadsPerBlock = getWarpSize<Acc>() * 2;
 
@@ -199,8 +189,8 @@ track_state_container_types::buffer kalman_fitting(
     // Sort the key to get the sorted parameter ids
     vecmem::device_vector<device::sort_key> keys_device(keys_buffer);
     vecmem::device_vector<unsigned int> param_ids_device(param_ids_buffer);
-    thrust::sort_by_key(thrustExecPolicy, keys_device.begin(),
-                        keys_device.end(), param_ids_device.begin());
+    details::sort_by_key(queue, mr, keys_device.begin(), keys_device.end(),
+                         param_ids_device.begin());
 
     // Run the fitting, using the sorted parameter IDs.
     track_state_container_types::view track_states_view = track_states_buffer;

device/alpaka/src/utils/oneDPL.hpp

@@ -0,0 +1,17 @@
+/** TRACCC library, part of the ACTS project (R&D line)
+ *
+ * (c) 2025 CERN for the benefit of the ACTS project
+ *
+ * Mozilla Public License Version 2.0
+ */
+
+#pragma once
+
+// Mark this as a "system header". To suppress all warnings from oneDPL.
+// This is needed because at the time of writing we cannot provide oneDPL with
+// "-isystem" to the oneAPI compiler.
+#pragma clang system_header
+
+// oneDPL include(s).
+#include <oneapi/dpl/algorithm>
+#include <oneapi/dpl/execution>

device/alpaka/src/utils/parallel_algorithms.hpp

@@ -0,0 +1,126 @@
+/**
+ * traccc library, part of the ACTS project (R&D line)
+ *
+ * (c) 2025 CERN for the benefit of the ACTS project
+ *
+ * Mozilla Public License Version 2.0
+ */
+
+#pragma once
+
+// Local include(s).
+#include "utils.hpp"
+
+// Project include(s).
+#include "traccc/utils/memory_resource.hpp"
+
+// Thrust include(s).
+#if !defined(ALPAKA_ACC_SYCL_ENABLED)
+#include <thrust/binary_search.h>
+#include <thrust/copy.h>
+#include <thrust/execution_policy.h>
+#include <thrust/fill.h>
+#include <thrust/scan.h>
+#include <thrust/sort.h>
+#include <thrust/unique.h>
+#endif
+
+// OneDPL include.
+//
+// This is left to a separate file to turn off warnings from oneDPL.
+#if defined(ALPAKA_ACC_SYCL_ENABLED)
+#include "oneDPL.hpp"
+#endif
+
+namespace traccc::alpaka::details {
+
+inline auto getExecutionPolicy([[maybe_unused]] Queue &q,
+                               [[maybe_unused]] const memory_resource &mr) {
+#if defined(ALPAKA_ACC_GPU_CUDA_ENABLED)
+    auto stream = ::alpaka::getNativeHandle(q);
+    return thrust::cuda::par_nosync(std::pmr::polymorphic_allocator(&(mr.main)))
+        .on(stream);
+#elif defined(ALPAKA_ACC_GPU_HIP_ENABLED)
+    auto stream = ::alpaka::getNativeHandle(q);
+    return thrust::hip_rocprim::par_nosync(
+               std::pmr::polymorphic_allocator<std::byte>(&(mr.main)))
+        .on(stream);
+#elif defined(ALPAKA_ACC_SYCL_ENABLED)
+    auto queue = ::alpaka::getNativeHandle(q);
+    return oneapi::dpl::execution::device_policy{queue};
+#else
+    return thrust::host;
+#endif
+}
+
+template <typename RandomAccessIterator, typename Compare>
+void sort(Queue &q, const memory_resource mr, RandomAccessIterator first,
+          RandomAccessIterator last, Compare comp) {
+    auto execPolicy = getExecutionPolicy(q, mr);
+
+#if defined(ALPAKA_ACC_SYCL_ENABLED)
+    oneapi::dpl::sort(execPolicy, first, last, comp);
+#else
+    thrust::sort(execPolicy, first, last, comp);
+#endif
+}
+
+template <typename RandomAccessIterator1, typename RandomAccessIterator2,
+          typename Compare>
+void sort_by_key(Queue &q, const memory_resource &mr,
+                 RandomAccessIterator1 keys_first,
+                 RandomAccessIterator1 keys_last,
+                 RandomAccessIterator2 values_first, Compare comp) {
+    auto execPolicy = getExecutionPolicy(q, mr);
+
+#if defined(ALPAKA_ACC_SYCL_ENABLED)
+    oneapi::dpl::sort_by_key(execPolicy, keys_first, keys_last, values_first,
+                             comp);
+#else
+    thrust::sort_by_key(execPolicy, keys_first, keys_last, values_first, comp);
+#endif
+}
+
+template <typename RandomAccessIterator1, typename RandomAccessIterator2>
+void sort_by_key(Queue &q, const memory_resource &mr,
+                 RandomAccessIterator1 keys_first,
+                 RandomAccessIterator1 keys_last,
+                 RandomAccessIterator2 values_first) {
+    auto execPolicy = getExecutionPolicy(q, mr);
+
+#if defined(ALPAKA_ACC_SYCL_ENABLED)
+    oneapi::dpl::sort_by_key(execPolicy, keys_first, keys_last, values_first);
+#else
+    thrust::sort_by_key(execPolicy, keys_first, keys_last, values_first);
+#endif
+}
+
+template <typename ForwardIt1, typename ForwardIt2, typename OutputIt,
+          typename Compare>
+void upper_bound(Queue &q, const memory_resource &mr, ForwardIt1 first1,
+                 ForwardIt1 last1, ForwardIt2 first2, ForwardIt2 last2,
+                 OutputIt d_first, Compare comp) {
+
+    auto execPolicy = getExecutionPolicy(q, mr);
+#if defined(ALPAKA_ACC_SYCL_ENABLED)
+    oneapi::dpl::upper_bound(execPolicy, first1, last1, first2, last2, d_first,
+                             comp);
+#else
+    thrust::upper_bound(execPolicy, first1, last1, first2, last2, d_first,
+                        comp);
+#endif
+}
+
+template <typename InputIt, typename OutputIt, typename Compare>
+OutputIt unique_copy(Queue &q, const memory_resource &mr, InputIt first,
+                     InputIt last, OutputIt d_first, Compare comp) {
+    auto execPolicy = getExecutionPolicy(q, mr);
+
+#if defined(ALPAKA_ACC_SYCL_ENABLED)
+    return oneapi::dpl::unique_copy(execPolicy, first, last, d_first, comp);
+#else
+    return thrust::unique_copy(execPolicy, first, last, d_first, comp);
+#endif
+}
+
+}  // namespace traccc::alpaka::details