fix: hybrid execution crash (#17)

* fix(dist): crash in distributed contraction code when executing in hybrid mode due to non-parallelized section

* tests(dist): add hybryd endtoend test

* style(tests): add empty lines after namespace

kaminpar-dist/coarsening/contraction/global_cluster_contraction.cc

@@ -311,11 +311,13 @@ find_nonlocal_nodes(const Graph &graph, const StaticArray<GlobalNodeID> &lnode_t
   std::unordered_map<GlobalNodeID, NodeWeight> nonlocal_nodes;
   std::atomic<std::size_t> size = 0;
 
-  graph.pfor_all_nodes([&](const NodeID lnode) {
+  for (NodeID lnode : graph.all_nodes()) {
+    // graph.pfor_all_nodes([&](const NodeID lnode) {
     const GlobalNodeID gcluster = lnode_to_gcluster[lnode];
 
     if (graph.is_owned_global_node(gcluster)) {
-      return;
+      // return;
+      continue;
     }
 
     const NodeWeight weight = graph.is_owned_node(lnode) ? graph.node_weight(lnode) : 0;
@@ -333,7 +335,8 @@ find_nonlocal_nodes(const Graph &graph, const StaticArray<GlobalNodeID> &lnode_t
       size.fetch_add(1, std::memory_order_relaxed);
       nonlocal_nodes[gcluster + 1] = weight;
     }
-  });
+    //});
+  }
 
   RECORD("nonlocal_nodes") StaticArray<GlobalNode> dense_nonlocal_nodes(size);
   std::size_t i = 0;

tests/endtoend/dist_endtoend_test.cc

@@ -17,7 +17,9 @@
 #include "kaminpar-common/math.h"
 
 namespace kaminpar::dist {
+
 namespace data {
+
 static std::vector<GlobalEdgeID> global_xadj = {
 #include "data.graph.xadj"
 };
@@ -54,6 +56,7 @@ std::vector<GlobalEdgeID> create_xadj() {
 
   return xadj;
 }
+
 } // namespace data
 
 TEST(DistEndToEndTest, partitions_empty_unweighted_graph) {
@@ -144,17 +147,14 @@ TEST(DistEndToEndTest, partitions_unweighted_walshaw_data_graph) {
   EXPECT_EQ(reported_cut, actual_cut / 2);
 }
 
-// Disabled: can fail since we offset the PRNG seed by the thread ID, and not all calls are made by
-// the same threads across multiple runs
-/*
-TEST(DistEndToEndTest, partitions_unweighted_walshaw_data_graph_multiple_times_with_same_seed) {
-  const PEID size = mpi::get_comm_size(MPI_COMM_WORLD);
+TEST(
+    DistEndToEndTest, partitions_unweighted_walshaw_data_graph_multiple_times_with_different_seeds
+) {
   const PEID rank = mpi::get_comm_rank(MPI_COMM_WORLD);
 
   auto vtxdist = data::create_vtxdist();
   auto xadj = data::create_xadj();
 
-  const GlobalNodeID global_n = data::global_xadj.size() - 1;
   const NodeID n = xadj.size() - 1;
 
   GlobalNodeID *vtxdist_ptr = vtxdist.data();
@@ -166,23 +166,20 @@ TEST(DistEndToEndTest, partitions_unweighted_walshaw_data_graph_multiple_times_w
   dKaMinPar dist(MPI_COMM_WORLD, 1, create_default_context()); // 1 thread: deterministic
   dist.set_output_level(OutputLevel::QUIET);
   dist.import_graph(vtxdist_ptr, xadj_ptr, adjncy_ptr, nullptr, nullptr);
-  const EdgeWeight reported_cut = dist.compute_partition(16, seed0_partition.data());
+  dist.compute_partition(16, seed0_partition.data());
 
-  for (const int seed : {0, 0, 0}) {
+  for (const int seed : {1, 2, 3}) {
     std::vector<BlockID> partition(n);
     dKaMinPar::reseed(seed);
     dKaMinPar dist(MPI_COMM_WORLD, 1, create_default_context()); // 1 thread: deterministic
     dist.set_output_level(OutputLevel::QUIET);
     dist.import_graph(vtxdist_ptr, xadj_ptr, adjncy_ptr, nullptr, nullptr);
     dist.compute_partition(16, partition.data());
-    EXPECT_EQ(partition, seed0_partition);
+    EXPECT_NE(partition, seed0_partition);
   }
 }
-*/
 
-TEST(
-    DistEndToEndTest, partitions_unweighted_walshaw_data_graph_multiple_times_with_different_seeds
-) {
+TEST(DistEndToEndTest, partitions_unweighted_walshaw_data_graph_with_three_threads_per_mpi) {
   const PEID rank = mpi::get_comm_rank(MPI_COMM_WORLD);
 
   auto vtxdist = data::create_vtxdist();
@@ -196,18 +193,10 @@ TEST(
 
   std::vector<BlockID> seed0_partition(n);
   dKaMinPar::reseed(0);
-  dKaMinPar dist(MPI_COMM_WORLD, 1, create_default_context()); // 1 thread: deterministic
+  dKaMinPar dist(MPI_COMM_WORLD, 3, create_default_context());
   dist.set_output_level(OutputLevel::QUIET);
   dist.import_graph(vtxdist_ptr, xadj_ptr, adjncy_ptr, nullptr, nullptr);
-
-  for (const int seed : {1, 2, 3}) {
-    std::vector<BlockID> partition(n);
-    dKaMinPar::reseed(seed);
-    dKaMinPar dist(MPI_COMM_WORLD, 1, create_default_context()); // 1 thread: deterministic
-    dist.set_output_level(OutputLevel::QUIET);
-    dist.import_graph(vtxdist_ptr, xadj_ptr, adjncy_ptr, nullptr, nullptr);
-    dist.compute_partition(16, partition.data());
-    EXPECT_NE(partition, seed0_partition);
-  }
+  dist.compute_partition(16, seed0_partition.data());
 }
+
 } // namespace kaminpar::dist