Add: Taskflow backend

Author: ashvardanian

CMakeLists.txt

@@ -81,7 +81,7 @@ find_package(OpenMP QUIET)
 find_package(OpenCL QUIET)
 if (NOT APPLE)
     find_package(BLAS REQUIRED)
-endif()
+endif ()
 
 set(FETCHCONTENT_QUIET OFF)
 include(FetchContent)
@@ -93,6 +93,13 @@ FetchContent_Declare(
 )
 FetchContent_MakeAvailable(fmt)
 
+FetchContent_Declare(
+    taskflow
+    GIT_REPOSITORY https://github.com/taskflow/taskflow.git
+    GIT_TAG v3.10.0
+)
+FetchContent_MakeAvailable(taskflow)
+
 FetchContent_Declare(
     fork_union
     GIT_REPOSITORY https://github.com/ashvardanian/fork_union.git
@@ -164,13 +171,13 @@ set(CMAKE_CUDA_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -O2")
 set(CMAKE_GCC_FLAGS "${CMAKE_GCC_FLAGS} -march=native -fopenmp")
 
 add_executable(reduce_bench reduce_bench.cpp)
-target_link_libraries(reduce_bench PRIVATE benchmark::benchmark fmt::fmt fork_union Threads::Threads)
+target_link_libraries(reduce_bench PRIVATE benchmark::benchmark fmt::fmt fork_union Taskflow Threads::Threads)
 
 if (APPLE)
     target_link_libraries(reduce_bench PRIVATE "-framework Accelerate")
-else()
+else ()
     target_compile_definitions(reduce_bench PRIVATE BLAS::BLAS)
-endif()
+endif ()
 
 if (USE_INTEL_TBB)
     target_link_libraries(reduce_bench PRIVATE TBB::tbb)

reduce_bench.cpp

@@ -337,13 +337,15 @@ int main(int argc, char **argv) {
     register_("std::accumulate/f32", stl_accumulate_gt<float> {}, dataset);
     register_("std::accumulate/f64", stl_accumulate_gt<double> {}, dataset);
     register_("serial/f32/av::fork_union", fork_union_gt<unrolled_gt<float>> {}, dataset);
+    register_("serial/f32/taskflow", taskflow_gt<unrolled_gt<float>> {}, dataset);
     register_("serial/f64/av::fork_union", fork_union_gt<unrolled_gt<double>> {}, dataset);
+    register_("serial/f64/taskflow", taskflow_gt<unrolled_gt<double>> {}, dataset);
 #if defined(_OPENMP)
     register_("serial/f32/openmp", openmp_t {}, dataset);
 #endif // defined(_OPENMP)
 
-    //! BLAS struggles with zero-strided arguments!
-    //! register_("blas/f32", blas_dot_t {}, dataset);
+    // ! BLAS struggles with zero-strided arguments!
+    // ! register_("blas/f32", blas_dot_t {}, dataset);
 
 #if defined(__cpp_lib_execution)
     register_("std::reduce<par>/f32", stl_par_reduce_gt<float> {}, dataset);
@@ -380,6 +382,7 @@ int main(int argc, char **argv) {
 #if defined(__ARM_NEON)
     register_("neon/f32", neon_f32_t {}, dataset);
     register_("neon/f32/av::fork_union", fork_union_gt<neon_f32_t> {}, dataset);
+    register_("neon/f32/taskflow", taskflow_gt<neon_f32_t> {}, dataset);
     register_("neon/f32/std::threads", threads_gt<neon_f32_t> {}, dataset);
     register_("neon/f32/openmp", openmp_gt<neon_f32_t> {}, dataset);
 #endif
@@ -388,6 +391,7 @@ int main(int argc, char **argv) {
 #if defined(__ARM_FEATURE_SVE)
     register_("sve/f32", sve_f32_t {}, dataset);
     register_("sve/f32/av::fork_union", fork_union_gt<sve_f32_t> {}, dataset);
+    register_("sve/f32/taskflow", taskflow_gt<sve_f32_t> {}, dataset);
     register_("sve/f32/std::threads", threads_gt<sve_f32_t> {}, dataset);
     register_("sve/f32/openmp", openmp_gt<sve_f32_t> {}, dataset);
 #endif // defined(__ARM_FEATURE_SVE__)

reduce_cpu.hpp

@@ -27,6 +27,7 @@
 #endif
 
 #include <fork_union.hpp>
+#include <taskflow/taskflow.hpp>
 
 namespace ashvardanian {
 
@@ -719,6 +720,44 @@ class fork_union_gt {
     }
 };
 
+template <typename serial_at = stl_accumulate_gt<float>>
+class taskflow_gt {
+    float const *const begin_ = nullptr;
+    float const *const end_ = nullptr;
+    std::size_t const cores_ = 0;
+
+    tf::Executor executor_;
+    tf::Taskflow taskflow_;
+
+    struct alignas(128) thread_result_t {
+        double partial_sum = 0.0;
+    };
+    std::vector<thread_result_t> sums_;
+
+  public:
+    taskflow_gt() = default;
+    taskflow_gt(float const *b, float const *e)
+        : begin_ {b}, end_ {e}, cores_ {total_cores()}, executor_ {static_cast<unsigned>(cores_)}, sums_(cores_) {}
+
+    double operator()() {
+        auto const input_size = static_cast<std::size_t>(end_ - begin_);
+        auto const chunk_size = scalars_per_core(input_size, cores_);
+
+        taskflow_.clear();
+        for (std::size_t tid = 0; tid < cores_; ++tid) {
+            taskflow_.emplace([&, tid] {
+                std::size_t const start = std::min(tid * chunk_size, input_size);
+                std::size_t const stop = std::min(start + chunk_size, input_size);
+                sums_[tid].partial_sum = serial_at {begin_ + start, begin_ + stop}();
+            });
+        }
+
+        executor_.run(taskflow_).wait();
+        return std::accumulate(sums_.begin(), sums_.end(), 0.0,
+                               [](double acc, thread_result_t const &x) noexcept { return acc + x.partial_sum; });
+    }
+};
+
 #pragma endregion - Multicore
 
 } // namespace ashvardanian