Address review comments, add tests, benchmarks, and cleanups

Signed-off-by: Abhishek Bansal <abhibansal593@gmail.com>

Author: abhishek593

libs/full/collectives/include/hpx/collectives/barrier.hpp

@@ -102,9 +102,7 @@ namespace hpx { namespace distributed {
 #include <hpx/modules/components_base.hpp>
 #include <hpx/modules/errors.hpp>
 #include <hpx/modules/futures.hpp>
-#include <hpx/modules/memory.hpp>
 
-#include <array>
 #include <atomic>
 #include <cstddef>
 #include <cstdint>
@@ -263,14 +261,19 @@ namespace hpx::collectives {
                 HPX_GET_EXCEPTION(hpx::error::bad_parameter,
                     "hpx::collectives::barrier (hierarchical)",
                     "hierarchical barrier requires an explicit generation "
-                    "number for the 2k/2k+1 internal mapping"));
+                    "number for the 2k-1/2k internal mapping"));
         }
 
         if (this_site.is_default())
         {
             this_site = agas::get_locality_id();
         }
 
+        if (communicators.size() == 0)
+        {
+            return hpx::make_ready_future();
+        }
+
         generation_arg const reduce_gen(2 * generation - 1);
         generation_arg const broadcast_gen(2 * generation);
 
@@ -287,10 +290,6 @@ namespace hpx::collectives {
 
         // Broadcast phase: walk sub-communicators from shallowest to deepest.
         // Returning the final future lets the caller chain on completion.
-        if (communicators.size() == 0)
-        {
-            return hpx::make_ready_future();
-        }
 
         for (std::size_t i = 0; i + 1 < communicators.size(); ++i)
         {
@@ -343,8 +342,8 @@ namespace hpx::distributed {
         void detach();
 
         // Get the instance of the global barrier
-        static std::array<barrier, 2>& get_global_barrier();
-        static std::array<barrier, 2> create_global_barrier();
+        static barrier& get_global_barrier();
+        static barrier create_global_barrier();
 
         static void synchronize();
 

libs/full/collectives/src/barrier.cpp

@@ -20,11 +20,11 @@
 #include <hpx/modules/type_support.hpp>
 
 #include <algorithm>
-#include <array>
 #include <atomic>
 #include <cstddef>
 #include <iterator>
 #include <string>
+#include <type_traits>
 #include <utility>
 #include <variant>
 #include <vector>
@@ -227,7 +227,7 @@ namespace hpx::distributed {
             {
                 wait(hpx::launch::async).get();
             }
-            catch (...)
+            catch (...)    // NOLINT(bugprone-empty-catch)
             {
             }
         }
@@ -242,28 +242,26 @@ namespace hpx::distributed {
         comm_ = std::monostate{};
     }
 
-    std::array<barrier, 2> barrier::create_global_barrier()
+    barrier barrier::create_global_barrier()
     {
         runtime& rt = get_runtime();
         util::runtime_configuration const& cfg = rt.get_config();
         auto const num = static_cast<std::size_t>(cfg.get_num_localities());
         auto const rank = static_cast<std::size_t>(hpx::get_locality_id());
-        barrier b1("/0/hpx/global_barrier0", num, rank, force_flat_tag::tag);
-        barrier b2("/0/hpx/global_barrier1", num, rank, force_flat_tag::tag);
-        return {{std::move(b1), std::move(b2)}};
+        return barrier("/0/hpx/global_barrier", num, rank, force_flat_tag::tag);
     }
 
-    std::array<barrier, 2>& barrier::get_global_barrier()
+    barrier& barrier::get_global_barrier()
     {
-        static std::array<barrier, 2> bs = {};
-        return bs;
+        static barrier b;
+        return b;
     }
 
     void barrier::synchronize()
     {
-        std::array<barrier, 2>& b = get_global_barrier();
-        HPX_ASSERT(!std::holds_alternative<std::monostate>(b[0].comm_));
-        b[0].wait();
+        barrier& b = get_global_barrier();
+        HPX_ASSERT(!std::holds_alternative<std::monostate>(b.comm_));
+        b.wait();
     }
 }    // namespace hpx::distributed
 

libs/full/collectives/tests/performance/benchmark_collectives.cpp

@@ -2,6 +2,7 @@
 //  Copyright (c) 2025 Alexander Strack
 //  Copyright (c) 2025 Hartmut Kaiser
 //  Copyright (c) 2026 Anshuman Agrawal
+//  Copyright (c) 2026 Abhishek Bansal
 //
 //  SPDX-License-Identifier: BSL-1.0
 //  Distributed under the Boost Software License, Version 1.0. (See accompanying
@@ -12,6 +13,7 @@
 #if !defined(HPX_COMPUTE_DEVICE_CODE)
 #include <hpx/collectives/all_gather.hpp>
 #include <hpx/collectives/all_reduce.hpp>
+#include <hpx/collectives/barrier.hpp>
 #include <hpx/hpx.hpp>
 #include <hpx/hpx_init.hpp>
 #include <hpx/modules/collectives.hpp>
@@ -31,6 +33,7 @@ constexpr char const* broadcast_direct_basename = "/test/broadcast_direct/";
 constexpr char const* gather_direct_basename = "/test/gather_direct/";
 constexpr char const* all_reduce_direct_basename = "/test/all_reduce_direct/";
 constexpr char const* all_gather_direct_basename = "/test/all_gather_direct/";
+constexpr char const* barrier_bench_basename = "/test/barrier_bench/";
 
 struct vector_adder
 {
@@ -514,6 +517,45 @@ void test_all_reduce_hierarchical(int arity, int lpn, std::size_t iterations,
     }
 }
 
+void test_barrier_hierarchical(int arity, int lpn, std::size_t iterations,
+    int test_size, std::string const& operation, int fallback_threshold)
+{
+    std::size_t const num_localities =
+        hpx::get_num_localities(hpx::launch::sync);
+    std::size_t const this_locality = hpx::get_locality_id();
+    HPX_TEST_LTE(static_cast<std::size_t>(2), num_localities);
+    auto communicators =
+        create_hierarchical_communicator(barrier_bench_basename,
+            num_sites_arg(num_localities), this_site_arg(this_locality),
+            arity_arg(arity), generation_arg(1), root_site_arg(0),
+            fallback_threshold < 0 ?
+                flat_fallback_threshold_arg() :
+                flat_fallback_threshold_arg(
+                    static_cast<std::size_t>(fallback_threshold)));
+    char const* const barrier_sync_name = "/test/barrier/hierarchical";
+    hpx::distributed::barrier sync(barrier_sync_name);
+    std::vector<double> result(iterations, 0.0);
+
+    for (std::size_t i = 0; i != iterations; ++i)
+    {
+        hpx::chrono::high_resolution_timer const timer;
+        hpx::collectives::barrier(
+            communicators, this_site_arg(this_locality), generation_arg(i + 1))
+            .get();
+        sync.wait();
+        result[i] = timer.elapsed();
+    }
+
+    if (this_locality == 0)
+    {
+        std::string const mod_name = fallback_threshold < 0 ?
+            std::string("hierarchical") :
+            "hierarchical_t" + std::to_string(fallback_threshold);
+        write_to_file(operation, mod_name, arity, num_localities, lpn,
+            test_size, iterations, result);
+    }
+}
+
 ////////////////////////////////////////////////////////////////////////////////////////
 // One shot collectives
 void test_one_shot_use_scatter(int lpn, std::size_t iterations, int test_size,
@@ -1116,6 +1158,36 @@ void test_multiple_use_with_generation_all_reduce(int lpn,
     }
 }
 
+void test_multiple_use_with_generation_barrier(int lpn, std::size_t iterations,
+    int test_size, std::string const& operation)
+{
+    std::size_t const num_localities =
+        hpx::get_num_localities(hpx::launch::sync);
+    std::size_t const this_locality = hpx::get_locality_id();
+    HPX_TEST_LTE(static_cast<std::size_t>(2), num_localities);
+    auto const barrier_client = create_communicator(barrier_bench_basename,
+        num_sites_arg(num_localities), this_site_arg(this_locality));
+    char const* const barrier_sync_name = "/test/barrier/generation";
+    hpx::distributed::barrier sync(barrier_sync_name);
+    std::vector<double> result(iterations, 0.0);
+
+    for (std::size_t i = 0; i != iterations; ++i)
+    {
+        hpx::chrono::high_resolution_timer const timer;
+        hpx::collectives::barrier(
+            barrier_client, this_site_arg(), generation_arg(i + 1))
+            .get();
+        sync.wait();
+        result[i] = timer.elapsed();
+    }
+
+    if (this_locality == 0)
+    {
+        write_to_file(operation, "multi_use", -1, num_localities, lpn,
+            test_size, iterations, result);
+    }
+}
+
 void test_all_gather_hierarchical(int arity, int lpn, std::size_t iterations,
     int test_size, std::string const& operation, int fallback_threshold)
 {
@@ -1367,6 +1439,19 @@ int hpx_main(hpx::program_options::variables_map& vm)
                     operation, fallback_threshold);
             }
         }
+        else if (operation == "barrier")
+        {
+            if (arity == -1)
+            {
+                test_multiple_use_with_generation_barrier(
+                    lpn, iterations, test_size, operation);
+            }
+            else
+            {
+                test_barrier_hierarchical(arity, lpn, iterations, test_size,
+                    operation, fallback_threshold);
+            }
+        }
     }
 
     return hpx::finalize();
@@ -1388,7 +1473,7 @@ int main(int argc, char* argv[])
             "Number of Iteration the collective is executed")
         ("operation", value<std::string>()->default_value("scatter"),
             "Collective Operation (scatter, reduce, broadcast, gather, "
-            "all_reduce, all_gather)")
+            "all_reduce, all_gather, barrier)")
         ("fallback_threshold", value<int>()->default_value(-1),
             "Flat fallback threshold for hierarchical mode. -1 uses library "
             "default (16). Set to 0 to force tree construction. Only meaningful "

libs/full/collectives/tests/unit/CMakeLists.txt

@@ -27,6 +27,7 @@ if(HPX_WITH_NETWORKING)
       ${tests}
       all_gather_hierarchical
       all_reduce_hierarchical
+      barrier_hierarchical
       broadcast_direct
       concurrent_collectives
       exclusive_scan_