infra: stabilize MPI tests and prevent hangs

- Sync gtest random_seed + filter across ranks
- Abort on worker failure; wrap RunAllTests with MPI_Abort on exceptions
- Make PPC_TEST_TMPDIR per MPI rank
- Pass env to mpiexec on Windows (-env), keep -x on *nix

Author: allnes

modules/runners/src/runners.cpp

@@ -6,16 +6,63 @@
 #include <chrono>
 #include <cstdint>
 #include <cstdlib>
+#include <exception>
 #include <format>
 #include <iostream>
 #include <memory>
 #include <random>
 #include <stdexcept>
 #include <string>
+#include <string_view>
 
 #include "oneapi/tbb/global_control.h"
 #include "util/include/util.hpp"
 
+namespace {
+[[maybe_unused]] void SyncGTestSeed() {
+  unsigned int seed = 0;
+  int rank = -1;
+  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+  if (rank == 0) {
+    try {
+      seed = std::random_device{}();
+    } catch (...) {
+      seed = 0;
+    }
+    if (seed == 0) {
+      const auto now = static_cast<std::uint64_t>(std::chrono::steady_clock::now().time_since_epoch().count());
+      seed = static_cast<unsigned int>(((now & 0x7fffffffULL) | 1ULL));
+    }
+  }
+  MPI_Bcast(&seed, 1, MPI_UNSIGNED, 0, MPI_COMM_WORLD);
+  ::testing::GTEST_FLAG(random_seed) = static_cast<int>(seed);
+}
+
+[[maybe_unused]] void SyncGTestFilter() {
+  int rank = -1;
+  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+  std::string filter = (rank == 0) ? ::testing::GTEST_FLAG(filter) : std::string{};
+  int len = static_cast<int>(filter.size());
+  MPI_Bcast(&len, 1, MPI_INT, 0, MPI_COMM_WORLD);
+  if (rank != 0) {
+    filter.resize(static_cast<std::size_t>(len));
+  }
+  if (len > 0) {
+    MPI_Bcast(filter.data(), len, MPI_CHAR, 0, MPI_COMM_WORLD);
+  }
+  ::testing::GTEST_FLAG(filter) = filter;
+}
+
+[[maybe_unused]] bool HasFlag(int argc, char **argv, std::string_view flag) {
+  for (int i = 1; i < argc; ++i) {
+    if (argv[i] != nullptr && std::string_view(argv[i]) == flag) {
+      return true;
+    }
+  }
+  return false;
+}
+}  // namespace
+
 namespace ppc::runners {
 
 void UnreadMessagesDetector::OnTestEnd(const ::testing::TestInfo & /*test_info*/) {
@@ -51,6 +98,8 @@ void WorkerTestFailurePrinter::OnTestEnd(const ::testing::TestInfo &test_info) {
   }
   PrintProcessRank();
   base_->OnTestEnd(test_info);
+  // Abort the whole MPI job on any test failure to avoid other ranks hanging on barriers.
+  MPI_Abort(MPI_COMM_WORLD, EXIT_FAILURE);
 }
 
 void WorkerTestFailurePrinter::OnTestPartResult(const ::testing::TestPartResult &test_part_result) {
@@ -76,6 +125,63 @@ int RunAllTests() {
   }
   return status;
 }
+
+void SyncGTestSeed() {
+  unsigned int seed = 0;
+  int rank = -1;
+  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+  if (rank == 0) {
+    try {
+      seed = std::random_device{}();
+    } catch (...) {
+      seed = 0;
+    }
+    if (seed == 0) {
+      const auto now = static_cast<std::uint64_t>(std::chrono::steady_clock::now().time_since_epoch().count());
+      seed = static_cast<unsigned int>(((now & 0x7fffffffULL) | 1ULL));
+    }
+  }
+  MPI_Bcast(&seed, 1, MPI_UNSIGNED, 0, MPI_COMM_WORLD);
+  ::testing::GTEST_FLAG(random_seed) = static_cast<int>(seed);
+}
+
+void SyncGTestFilter() {
+  int rank = -1;
+  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+  std::string filter = (rank == 0) ? ::testing::GTEST_FLAG(filter) : std::string{};
+  int len = static_cast<int>(filter.size());
+  MPI_Bcast(&len, 1, MPI_INT, 0, MPI_COMM_WORLD);
+  if (rank != 0) {
+    filter.resize(static_cast<std::size_t>(len));
+  }
+  if (len > 0) {
+    MPI_Bcast(filter.data(), len, MPI_CHAR, 0, MPI_COMM_WORLD);
+  }
+  ::testing::GTEST_FLAG(filter) = filter;
+}
+
+bool HasFlag(int argc, char **argv, std::string_view flag) {
+  for (int i = 1; i < argc; ++i) {
+    if (argv[i] != nullptr && std::string_view(argv[i]) == flag) {
+      return true;
+    }
+  }
+  return false;
+}
+
+int RunAllTestsSafely() {
+  try {
+    return RunAllTests();
+  } catch (const std::exception &e) {
+    std::cerr << std::format("[  ERROR  ] Exception after tests: {}", e.what()) << '\n';
+    MPI_Abort(MPI_COMM_WORLD, EXIT_FAILURE);
+    return EXIT_FAILURE;
+  } catch (...) {
+    std::cerr << "[  ERROR  ] Unknown exception after tests" << '\n';
+    MPI_Abort(MPI_COMM_WORLD, EXIT_FAILURE);
+    return EXIT_FAILURE;
+  }
+}
 }  // namespace
 
 int Init(int argc, char **argv) {
@@ -91,36 +197,21 @@ int Init(int argc, char **argv) {
 
   ::testing::InitGoogleTest(&argc, argv);
 
-  // Ensure consistent GoogleTest shuffle order across all MPI ranks.
-  unsigned int seed = 0;
-  int rank_for_seed = -1;
-  MPI_Comm_rank(MPI_COMM_WORLD, &rank_for_seed);
-
-  if (rank_for_seed == 0) {
-    try {
-      seed = std::random_device{}();
-    } catch (...) {
-      seed = 0;
-    }
-    if (seed == 0) {
-      const auto now = static_cast<std::uint64_t>(std::chrono::steady_clock::now().time_since_epoch().count());
-      seed = static_cast<unsigned int>(((now & 0x7fffffffULL) | 1ULL));
-    }
-  }
-
-  MPI_Bcast(&seed, 1, MPI_UNSIGNED, 0, MPI_COMM_WORLD);
-  ::testing::GTEST_FLAG(random_seed) = static_cast<int>(seed);
+  // Synchronize GoogleTest internals across ranks to avoid divergence
+  SyncGTestSeed();
+  SyncGTestFilter();
 
   auto &listeners = ::testing::UnitTest::GetInstance()->listeners();
   int rank = -1;
   MPI_Comm_rank(MPI_COMM_WORLD, &rank);
-  if (rank != 0 && (argc < 2 || argv[1] != std::string("--print-workers"))) {
+  const bool print_workers = HasFlag(argc, argv, "--print-workers");
+  if (rank != 0 && !print_workers) {
     auto *listener = listeners.Release(listeners.default_result_printer());
     listeners.Append(new WorkerTestFailurePrinter(std::shared_ptr<::testing::TestEventListener>(listener)));
   }
   listeners.Append(new UnreadMessagesDetector());
 
-  auto status = RunAllTests();
+  const int status = RunAllTestsSafely();
 
   const int finalize_res = MPI_Finalize();
   if (finalize_res != MPI_SUCCESS) {

modules/util/include/util.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <algorithm>
+#include <array>
 #include <atomic>
 #include <cctype>
 #include <cstdint>
@@ -26,6 +27,7 @@
 #include <gtest/gtest.h>
 
 #include <libenvpp/detail/environment.hpp>
+#include <libenvpp/detail/get.hpp>
 #include <nlohmann/json.hpp>
 
 /// @brief JSON namespace used for settings and config parsing.
@@ -123,7 +125,19 @@ class ScopedPerTestEnv {
  private:
   static std::string CreateTmpDir(const std::string &token) {
     namespace fs = std::filesystem;
-    const fs::path tmp = fs::temp_directory_path() / (std::string("ppc_test_") + token);
+    auto make_rank_suffix = []() -> std::string {
+      // Derive rank from common MPI env vars without including MPI headers
+      constexpr std::array<std::string_view, 5> kRankVars = {"OMPI_COMM_WORLD_RANK", "PMI_RANK", "PMIX_RANK",
+                                                             "SLURM_PROCID", "MSMPI_RANK"};
+      for (auto name : kRankVars) {
+        if (auto r = env::get<int>(name); r.has_value() && r.value() >= 0) {
+          return std::string("_rank_") + std::to_string(r.value());
+        }
+      }
+      return std::string{};
+    };
+    const std::string rank_suffix = IsUnderMpirun() ? make_rank_suffix() : std::string{};
+    const fs::path tmp = fs::temp_directory_path() / (std::string("ppc_test_") + token + rank_suffix);
     std::error_code ec;
     fs::create_directories(tmp, ec);
     (void)ec;

scripts/run_tests.py

@@ -53,6 +53,7 @@ def __init__(self, verbose=False):
             self.mpi_exec = "mpiexec"
         else:
             self.mpi_exec = "mpirun"
+        self.platform = platform.system()
 
     @staticmethod
     def __get_project_path():
@@ -133,10 +134,34 @@ def run_processes(self, additional_mpi_args):
             raise EnvironmentError(
                 "Required environment variable 'PPC_NUM_PROC' is not set."
             )
-
-        mpi_running = (
-            [self.mpi_exec] + shlex.split(additional_mpi_args) + ["-np", ppc_num_proc]
-        )
+        if self.platform == "Windows":
+            mpi_running = (
+                [self.mpi_exec]
+                + shlex.split(additional_mpi_args)
+                + [
+                    "-env",
+                    "PPC_NUM_THREADS",
+                    self.__ppc_env["PPC_NUM_THREADS"],
+                    "-env",
+                    "OMP_NUM_THREADS",
+                    self.__ppc_env["OMP_NUM_THREADS"],
+                    "-n",
+                    ppc_num_proc,
+                ]
+            )
+        else:
+            mpi_running = (
+                [self.mpi_exec]
+                + shlex.split(additional_mpi_args)
+                + [
+                    "-x",
+                    "PPC_NUM_THREADS",
+                    "-x",
+                    "OMP_NUM_THREADS",
+                    "-np",
+                    ppc_num_proc,
+                ]
+            )
         if not self.__ppc_env.get("PPC_ASAN_RUN"):
             for task_type in ["all", "mpi"]:
                 self.__run_exec(
@@ -147,7 +172,28 @@ def run_processes(self, additional_mpi_args):
 
     def run_performance(self):
         if not self.__ppc_env.get("PPC_ASAN_RUN"):
-            mpi_running = [self.mpi_exec, "-np", self.__ppc_num_proc]
+            if self.platform == "Windows":
+                mpi_running = [
+                    self.mpi_exec,
+                    "-env",
+                    "PPC_NUM_THREADS",
+                    self.__ppc_env["PPC_NUM_THREADS"],
+                    "-env",
+                    "OMP_NUM_THREADS",
+                    self.__ppc_env["OMP_NUM_THREADS"],
+                    "-n",
+                    self.__ppc_num_proc,
+                ]
+            else:
+                mpi_running = [
+                    self.mpi_exec,
+                    "-x",
+                    "PPC_NUM_THREADS",
+                    "-x",
+                    "OMP_NUM_THREADS",
+                    "-np",
+                    self.__ppc_num_proc,
+                ]
             for task_type in ["all", "mpi"]:
                 self.__run_exec(
                     mpi_running