Introduce support for executing the IPC conduit without MPI (#153)

* relax MPI dependency from code This commit (series) removes the strict dependency on MPI in code base. rocSHMEM will still be compiled with MPI, but the goal is to make the code work even if MPI_Init_thread has not been invoked, at least for certain, well-defined scenarios. Hence, the goal is not remove any mentioning of MPI from rocSHMEM, but to ensure correct execution of the ipc conduit even if the library has been initialized using other means. Details: - add non-MPI version of remote_heap and WindowInfo classes - host interfaces work on WindowInfoMPI, they will not work with the non-MPI code path. Since it is unclear whether we plan to support the host interfaces at all, this is probably not a major limitation. * update symmetric_heap structures and backend * first cut on initialization and enabling non-MPI initialization of the IPCBackend * add non-MPI hostInterface methods at the moment, only barrier_all and sync_all are explicitely supported. * add non-mpi version of ipc_policy and a number of smaller fixes required in other files. A small init/finalize test already passes now with the branch. * add non-mpi team_split_strided code * minor fixes for non-MPI use-case * disable symmetric-heap-window-ionfo test disable this test for now just to make the compilation pass. Will have to rework it. * make no-mpi great again after rebasing on top of the MPI singleton changes. * enable running functional tests with uuid init to run the functional tests using rocshmem_init_attr and the uuid mechanism requires a) a PMIx installation on the system b) setting the environment variable ROCSHMEM_TEST_UUID=1 * fix multi-team creation bug fix a bug occuring when creating many teams, which was the result of incorrectly applying two indices in our own implementation of Allreduce. * make unit tests pass again * reverse offload was impacted by code change fix the RO conduit to cope wioth the non-MPI path introduced for the IPC conduit. * update to cmake logic to find pmix * Update src/memory/window_info.hpp Co-authored-by: Yiltan <ytemucin@amd.com> * Update CMakeLists.txt Co-authored-by: Yiltan <ytemucin@amd.com> * document ROCSHMEM_UNIQUEID_NO_MPI * rename env. variable to UNIQUEID_WITH_MPI * update host.cpp to use USE_HDP_FLUSH macro instead of the deprecated USE_COHERENT_HEAP. * add note for running example with RO conduit add a note clarifying that running init_attr_test from the example directory requires setting an additional environment variable with the RO conduit. * Find PMIx in more cases, only apply pmix build options to the test that needs it, if OMPI_COMM_WORLD_LOCA_RANK is not setenv, abort --------- Co-authored-by: Yiltan <ytemucin@amd.com> Co-authored-by: Aurelien Bouteiller <abouteil@amd.com> [ROCm/rocshmem commit: 6ea5edc951]
2025-06-21 13:23:11 -05:00
parent 08d8324f74
commit e167f50803
32 changed files with 1028 additions and 228 deletions
@@ -38,7 +38,12 @@ namespace rocshmem {
 __host__ HostContextWindowInfo::HostContextWindowInfo(MPI_Comm comm_world,
                                                      SymmetricHeap* heap) {
  window_info_ =
-      new WindowInfo(comm_world, heap->get_local_heap_base(), heap->get_size());
+      new WindowInfoMPI(comm_world, heap->get_local_heap_base(), heap->get_size());
+}
+
+__host__ HostContextWindowInfo::HostContextWindowInfo(SymmetricHeap* heap) {
+  window_info_ =
+      new WindowInfo(heap->get_local_heap_base(), heap->get_size());
 }

 __host__ HostContextWindowInfo::~HostContextWindowInfo() {
@@ -146,6 +151,41 @@ __host__ void HostInterface::create_hdp_window() {
 }
 #endif  // USE_HDP_FLUSH

+__host__ HostInterface::HostInterface(HdpPolicy* hdp_policy,
+                                      TcpBootstrap *bootstr,
+                                      SymmetricHeap* heap) {
+  host_bootstrap_ = bootstr;
+  my_pe_ = bootstr->getRank();
+  num_pes_ = bootstr->getNranks();
+
+  /*
+   * Not sure we need this.
+   */
+  hdp_policy_ = hdp_policy;
+
+  /*
+   * Allocate and initialize pool of windows for contexts
+   */
+  char* value{nullptr};
+  if ((value = getenv("ROCSHMEM_MAX_NUM_HOST_CONTEXTS"))) {
+    max_num_ctxs_ = atoi(value);
+  }
+
+  size_t pool_size = max_num_ctxs_ * sizeof(HostContextWindowInfo*);
+  host_window_context_pool_ =
+      reinterpret_cast<HostContextWindowInfo**>(malloc(pool_size));
+
+  for (int ctx_i = 0; ctx_i < max_num_ctxs_; ctx_i++) {
+    host_window_context_pool_[ctx_i] =
+        new HostContextWindowInfo(heap);
+  }
+
+#if defined USE_HDP_FLUSH &&  not defined USE_SINGLE_NODE
+  printf("Non-mpi use-cases only supported with coherent heap at the moment. Aborting.\n");
+  abort();
+#endif
+}
+
 __host__ HostInterface::~HostInterface() {
 #if defined USE_HDP_FLUSH
  MPI_Win_unlock_all(hdp_win);
@@ -154,41 +194,61 @@ __host__ HostInterface::~HostInterface() {
 #endif  // USE_HDP_FLUSH

  /* Detroy the pool of contexts */
-  for (int ctx_i = 0; ctx_i < max_num_ctxs_; ctx_i++) {
-    delete host_window_context_pool_[ctx_i];
+
+  if (host_window_context_pool_ != nullptr) {
+    for (int ctx_i = 0; ctx_i < max_num_ctxs_; ctx_i++) {
+      delete host_window_context_pool_[ctx_i];
+    }
+    free(host_window_context_pool_);
  }

-  free(host_window_context_pool_);
-
-  MPI_Comm_free(&host_comm_world_);
+  if (host_comm_world_ != MPI_COMM_NULL) {
+    MPI_Comm_free(&host_comm_world_);
+  }
 }

 __host__ void HostInterface::putmem_nbi(void* dest, const void* source,
                                        size_t nelems, int pe,
                                        WindowInfo* window_info) {
-  initiate_put(dest, source, nelems, pe, window_info);
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
+  initiate_put(dest, source, nelems, pe, window_info_mpi);
 }

 __host__ void HostInterface::getmem_nbi(void* dest, const void* source,
                                        size_t nelems, int pe,
                                        WindowInfo* window_info) {
-  initiate_get(dest, source, nelems, pe, window_info);
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
+  initiate_get(dest, source, nelems, pe, window_info_mpi);
 }

 __host__ void HostInterface::putmem(void* dest, const void* source,
                                    size_t nelems, int pe,
                                    WindowInfo* window_info) {
-  initiate_put(dest, source, nelems, pe, window_info);
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
+  initiate_put(dest, source, nelems, pe, window_info_mpi);

-  MPI_Win_flush_local(pe, window_info->get_win());
+  MPI_Win_flush_local(pe, window_info_mpi->get_win());
 }

 __host__ void HostInterface::getmem(void* dest, const void* source,
                                    size_t nelems, int pe,
                                    WindowInfo* window_info) {
-  initiate_get(dest, source, nelems, pe, window_info);
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
+  initiate_get(dest, source, nelems, pe, window_info_mpi);

-  MPI_Win_flush_local(pe, window_info->get_win());
+  MPI_Win_flush_local(pe, window_info_mpi->get_win());

  /*
   * Flush local HDP to ensure that the NIC's write
@@ -198,7 +258,11 @@ __host__ void HostInterface::getmem(void* dest, const void* source,
 }

 __host__ void HostInterface::fence(WindowInfo* window_info) {
-  complete_all(window_info->get_win());
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
+  complete_all(window_info_mpi->get_win());

  /*
   * Flush my HDP and the HDPs of remote GPUs.
@@ -216,7 +280,11 @@ __host__ void HostInterface::fence(WindowInfo* window_info) {
 }

 __host__ void HostInterface::quiet(WindowInfo* window_info) {
-  complete_all(window_info->get_win());
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
+  complete_all(window_info_mpi->get_win());

  /* Same explanation as in fence */
  hdp_policy_->hdp_flush();
@@ -226,34 +294,53 @@ __host__ void HostInterface::quiet(WindowInfo* window_info) {
 }

 __host__ void HostInterface::sync_all(WindowInfo* window_info) {
-  MPI_Win_sync(window_info->get_win());
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    MPI_Win_sync(window_info_mpi->get_win());

-  hdp_policy_->hdp_flush();
-  /*
-   * No need to flush remote
-   * HDPs here since all PEs are
-   * participating.
-   */
+    hdp_policy_->hdp_flush();
+    /*
+     * No need to flush remote
+     * HDPs here since all PEs are
+     * participating.
+     */

-  MPI_Barrier(host_comm_world_);
+    MPI_Barrier(host_comm_world_);
+  } else {
+    hdp_policy_->hdp_flush();
+    host_bootstrap_->barrier();
+  }

  return;
 }

 __host__ void HostInterface::barrier_all(WindowInfo* window_info) {
-  complete_all(window_info->get_win());
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (window_info_mpi) {
+    complete_all(window_info_mpi->get_win());

-  /*
-   * Flush my HDP cache so remote NICs will
-   * see the latest values in device memory
-   */
-  hdp_policy_->hdp_flush();
+    /*
+     * Flush my HDP cache so remote NICs will
+     * see the latest values in device memory
+     */
+    hdp_policy_->hdp_flush();

-  MPI_Barrier(host_comm_world_);
+    MPI_Barrier(host_comm_world_);
+  } else {
+    // Probably not required
+    hdp_policy_->hdp_flush();
+    host_bootstrap_->barrier();
+  }
+
+  return;
 }

 __host__ void HostInterface::barrier_for_sync() {
-  MPI_Barrier(host_comm_world_);
+  if (host_comm_world_ != MPI_COMM_NULL) {
+    MPI_Barrier(host_comm_world_);
+  } else {
+    host_bootstrap_->barrier();
+  }
 }

 }  // namespace rocshmem
@@ -42,6 +42,7 @@
 #include "../hdp_policy.hpp"
 #include "../memory/symmetric_heap.hpp"
 #include "../memory/window_info.hpp"
+#include "../bootstrap/bootstrap.hpp"

 namespace rocshmem {

@@ -59,6 +60,7 @@ class HostContextWindowInfo {
   * @param[in] team_info information about participating PEs
   */
  HostContextWindowInfo(MPI_Comm comm_world, SymmetricHeap* heap);
+  HostContextWindowInfo(SymmetricHeap* heap);

  /**
   * @brief Destructor
@@ -104,11 +106,14 @@ class HostContextWindowInfo {
 class HostInterface {
 public:
  /**
-   * @brief Primary constructor
+   * @brief Primary constructors
   */
  __host__ HostInterface(HdpPolicy* hdp_policy, MPI_Comm rocshmem_comm,
                         SymmetricHeap* heap);

+  __host__ HostInterface(HdpPolicy* hdp_policy, TcpBootstrap *bootstrap,
+                         SymmetricHeap* heap);
+
  /**
   * @brief Destructor
   */
@@ -278,10 +283,10 @@ class HostInterface {
  }

  __host__ void initiate_put(void* dest, const void* source, size_t nelems,
-                             int pe, WindowInfo* window_info);
+                             int pe, WindowInfoMPI* window_info);

  __host__ void initiate_get(void* dest, const void* source, size_t nelems,
-                             int pe, WindowInfo* window_info);
+                             int pe, WindowInfoMPI* window_info);

  __host__ void complete_all(MPI_Win win);

@@ -321,7 +326,12 @@ class HostInterface {
  /**
   * @brief Global MPI communicator for those host API
   */
-  MPI_Comm host_comm_world_{};
+  MPI_Comm host_comm_world_{MPI_COMM_NULL};
+
+  /**
+   * @brief Bootstrap object used in the non-mpi workloads
+   */
+  TcpBootstrap *host_bootstrap_{nullptr};

  /**
   * @brief Duplicate of this processing element's id within global rank
@@ -55,7 +55,7 @@ __host__ inline void HostInterface::complete_all(MPI_Win win) {

 __host__ inline void HostInterface::initiate_put(void* dest, const void* source,
                                                 size_t nelems, int pe,
-                                                 WindowInfo* window_info) {
+                                                 WindowInfoMPI* window_info) {
  MPI_Win win{window_info->get_win()};
  void* win_start{window_info->get_start()};
  void* win_end{window_info->get_end()};
@@ -79,7 +79,7 @@ __host__ inline void HostInterface::initiate_put(void* dest, const void* source,

 __host__ inline void HostInterface::initiate_get(void* dest, const void* source,
                                                 size_t nelems, int pe,
-                                                 WindowInfo* window_info) {
+                                                 WindowInfoMPI* window_info) {
  MPI_Win win{window_info->get_win()};
  void* win_start{window_info->get_start()};
  void* win_end{window_info->get_end()};
@@ -58,6 +58,10 @@ __host__ void HostInterface::put_nbi(T* dest, const T* source, size_t nelems,

 template <typename T>
 __host__ T HostInterface::g(const T* source, int pe, WindowInfo* window_info) {
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
  DPRINTF("Function: host_g\n");

  T ret{};
@@ -70,7 +74,7 @@ __host__ T HostInterface::g(const T* source, int pe, WindowInfo* window_info) {
   */
  getmem_nbi(&ret, source, sizeof(T), pe, window_info);

-  MPI_Win_flush_local(pe, window_info->get_win());
+  MPI_Win_flush_local(pe, window_info_mpi->get_win());

  return ret;
 }
@@ -289,6 +293,11 @@ __host__ void HostInterface::amo_cas(void* dst, T value, T cond, int pe,
 template <typename T>
 __host__ T HostInterface::amo_fetch_add(void* dst, T value, int pe,
                                        WindowInfo* window_info) {
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
+
  /* Calculate offset of remote dest from base address of window */
  MPI_Aint offset{
      compute_offset(dst, window_info->get_start(), window_info->get_end())};
@@ -301,7 +310,7 @@ __host__ T HostInterface::amo_fetch_add(void* dst, T value, int pe,

  /* Offload remote fetch and op operation to MPI */
  T ret{};
-  MPI_Win win{window_info->get_win()};
+  MPI_Win win{window_info_mpi->get_win()};
  MPI_Datatype mpi_type{get_mpi_type<T>()};
  MPI_Fetch_and_op(&value, &ret, mpi_type, pe, offset, MPI_SUM, win);

@@ -313,6 +322,11 @@ __host__ T HostInterface::amo_fetch_add(void* dst, T value, int pe,
 template <typename T>
 __host__ T HostInterface::amo_fetch_cas(void* dst, T value, T cond, int pe,
                                        WindowInfo* window_info) {
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
+
  /* Calculate offset of remote dest from base address of window */
  MPI_Aint offset{
      compute_offset(dst, window_info->get_start(), window_info->get_end())};
@@ -325,7 +339,7 @@ __host__ T HostInterface::amo_fetch_cas(void* dst, T value, T cond, int pe,

  /* Offload remote compare and swap operation to MPI */
  T ret{};
-  MPI_Win win{window_info->get_win()};
+  MPI_Win win{window_info_mpi->get_win()};
  MPI_Datatype mpi_type{get_mpi_type<T>()};
  MPI_Compare_and_swap(&value, &cond, &ret, mpi_type, pe, offset, win);

@@ -452,6 +466,10 @@ __host__ inline int HostInterface::test_and_compare(MPI_Aint offset,
 template <typename T>
 __host__ void HostInterface::wait_until(T *ivars, int cmp, T val,
                                        WindowInfo* window_info) {
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
  DPRINTF("Function: host_wait_until\n");

  /*
@@ -461,7 +479,7 @@ __host__ void HostInterface::wait_until(T *ivars, int cmp, T val,
      compute_offset(ivars, window_info->get_start(), window_info->get_end())};

  MPI_Datatype mpi_type{get_mpi_type<T>()};
-  MPI_Win win{window_info->get_win()};
+  MPI_Win win{window_info_mpi->get_win()};

  /*
   * Continuously read the ivars atomically until it satisfies the condition
@@ -631,6 +649,10 @@ __host__ size_t HostInterface::wait_until_some_vector(T* ivars, size_t nelems,
 template <typename T>
 __host__ int HostInterface::test(T* ivars, int cmp, T val,
                                 WindowInfo* window_info) {
+  WindowInfoMPI* window_info_mpi = dynamic_cast<WindowInfoMPI*>(window_info);
+  if (!window_info_mpi) {
+    abort();
+  }
  DPRINTF("Function: host_test\n");

  /*
@@ -641,7 +663,7 @@ __host__ int HostInterface::test(T* ivars, int cmp, T val,

  MPI_Datatype mpi_type{get_mpi_type<T>()};

-  return test_and_compare(offset, mpi_type, cmp, val, window_info->get_win());
+  return test_and_compare(offset, mpi_type, cmp, val, window_info_mpi->get_win());
 }

 }  // namespace rocshmem