From b3412858814b209accebc92032aeb3668eef4a6f Mon Sep 17 00:00:00 2001 From: Yiltan Date: Fri, 1 Aug 2025 09:23:17 -0400 Subject: [PATCH] [RO] Fix MPI Communicator Check (#203) * Fix MPI Communicator casting --- src/reverse_offload/context_ro_device.cpp | 58 +++++++++---------- src/reverse_offload/context_ro_device.hpp | 2 +- .../context_ro_tmpl_device.hpp | 22 +++---- src/reverse_offload/mpi_transport.cpp | 12 ++-- src/reverse_offload/queue_proxy.hpp | 2 +- 5 files changed, 48 insertions(+), 48 deletions(-) diff --git a/src/reverse_offload/context_ro_device.cpp b/src/reverse_offload/context_ro_device.cpp index 6c3160a9d3..4e1b789a13 100644 --- a/src/reverse_offload/context_ro_device.cpp +++ b/src/reverse_offload/context_ro_device.cpp @@ -78,7 +78,7 @@ __device__ void ROContext::putmem(void *dest, const void *source, size_t nelems, return; } build_queue_element(RO_NET_PUT, dest, const_cast(source), nelems, - pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -98,7 +98,7 @@ __device__ void ROContext::getmem(void *dest, const void *source, size_t nelems, return; } build_queue_element(RO_NET_GET, dest, const_cast(source), nelems, - pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -118,7 +118,7 @@ __device__ void ROContext::putmem_nbi(void *dest, const void *source, return; } build_queue_element(RO_NET_PUT_NBI, dest, const_cast(source), - nelems, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + nelems, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, false); } } @@ -137,27 +137,27 @@ __device__ void ROContext::getmem_nbi(void *dest, const void *source, return; } build_queue_element(RO_NET_GET_NBI, dest, const_cast(source), - nelems, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + nelems, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, false); } } __device__ void ROContext::fence() { build_queue_element(RO_NET_FENCE, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, (MPI_Comm)NULL, ro_net_win_id, block_handle, + nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } __device__ void ROContext::fence(int pe) { // TODO(khamidou): need to check if per pe has any special handling build_queue_element(RO_NET_FENCE, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, (MPI_Comm)NULL, ro_net_win_id, block_handle, + nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } __device__ void ROContext::quiet() { build_queue_element(RO_NET_QUIET, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, (MPI_Comm)NULL, ro_net_win_id, block_handle, + nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -175,14 +175,14 @@ __device__ void *ROContext::shmem_ptr(const void *dest, int pe) { __device__ void ROContext::barrier_all() { build_queue_element(RO_NET_BARRIER, nullptr, nullptr, 0, 0, 0, 0, 0, - nullptr, nullptr, (MPI_Comm)NULL, ro_net_win_id, + nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } __device__ void ROContext::barrier_all_wave() { if (is_thread_zero_in_wave()) { build_queue_element(RO_NET_BARRIER, nullptr, nullptr, 0, 0, 0, 0, 0, - nullptr, nullptr, (MPI_Comm)NULL, ro_net_win_id, + nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } } @@ -190,7 +190,7 @@ __device__ void ROContext::barrier_all_wave() { __device__ void ROContext::barrier_all_wg() { if (is_thread_zero_in_block()) { build_queue_element(RO_NET_BARRIER, nullptr, nullptr, 0, 0, 0, 0, 0, - nullptr, nullptr, (MPI_Comm)NULL, ro_net_win_id, + nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } __syncthreads(); @@ -199,7 +199,7 @@ __device__ void ROContext::barrier_all_wg() { __device__ void ROContext::barrier(rocshmem_team_t team) { ROTeam *team_obj = reinterpret_cast(team); build_queue_element(RO_NET_BARRIER, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, team_obj->mpi_comm, ro_net_win_id, block_handle, + nullptr, (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -207,7 +207,7 @@ __device__ void ROContext::barrier_wave(rocshmem_team_t team) { ROTeam *team_obj = reinterpret_cast(team); if (is_thread_zero_in_wave()) { build_queue_element(RO_NET_BARRIER, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, team_obj->mpi_comm, ro_net_win_id, block_handle, + nullptr, (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } } @@ -216,7 +216,7 @@ __device__ void ROContext::barrier_wg(rocshmem_team_t team) { ROTeam *team_obj = reinterpret_cast(team); if (is_thread_zero_in_block()) { build_queue_element(RO_NET_BARRIER, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, team_obj->mpi_comm, ro_net_win_id, block_handle, + nullptr, (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } __syncthreads(); @@ -224,14 +224,14 @@ __device__ void ROContext::barrier_wg(rocshmem_team_t team) { __device__ void ROContext::sync_all() { build_queue_element(RO_NET_SYNC, nullptr, nullptr, 0, 0, 0, 0, 0, - nullptr, nullptr, (MPI_Comm)NULL, ro_net_win_id, + nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } __device__ void ROContext::sync_all_wave() { if (is_thread_zero_in_wave()) { build_queue_element(RO_NET_SYNC, nullptr, nullptr, 0, 0, 0, 0, 0, - nullptr, nullptr, (MPI_Comm)NULL, ro_net_win_id, + nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } } @@ -239,7 +239,7 @@ __device__ void ROContext::sync_all_wave() { __device__ void ROContext::sync_all_wg() { if (is_thread_zero_in_block()) { build_queue_element(RO_NET_SYNC, nullptr, nullptr, 0, 0, 0, 0, 0, - nullptr, nullptr, (MPI_Comm)NULL, ro_net_win_id, + nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } __syncthreads(); @@ -248,7 +248,7 @@ __device__ void ROContext::sync_all_wg() { __device__ void ROContext::sync(rocshmem_team_t team) { ROTeam *team_obj = reinterpret_cast(team); build_queue_element(RO_NET_SYNC, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, team_obj->mpi_comm, ro_net_win_id, block_handle, + nullptr, (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -256,7 +256,7 @@ __device__ void ROContext::sync_wave(rocshmem_team_t team) { ROTeam *team_obj = reinterpret_cast(team); if (is_thread_zero_in_wave()) { build_queue_element(RO_NET_SYNC, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, team_obj->mpi_comm, ro_net_win_id, block_handle, + nullptr, (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } } @@ -265,7 +265,7 @@ __device__ void ROContext::sync_wg(rocshmem_team_t team) { ROTeam *team_obj = reinterpret_cast(team); if (is_thread_zero_in_block()) { build_queue_element(RO_NET_SYNC, nullptr, nullptr, 0, 0, 0, 0, 0, nullptr, - nullptr, team_obj->mpi_comm, ro_net_win_id, block_handle, + nullptr, (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } __syncthreads(); @@ -278,7 +278,7 @@ __device__ void ROContext::ctx_destroy() { auto *proxy{backend_proxy.get()}; build_queue_element(RO_NET_FINALIZE, nullptr, nullptr, 0, 0, 0, 0, 0, - nullptr, nullptr, (MPI_Comm)NULL, ro_net_win_id, + nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); int buffer_id = ro_net_win_id; @@ -302,7 +302,7 @@ __device__ void ROContext::putmem_wg(void *dest, const void *source, } else { if (is_thread_zero_in_block()) { build_queue_element(RO_NET_PUT, dest, const_cast(source), nelems, - pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -321,7 +321,7 @@ __device__ void ROContext::getmem_wg(void *dest, const void *source, } else { if (is_thread_zero_in_block()) { build_queue_element(RO_NET_GET, dest, const_cast(source), nelems, - pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -340,7 +340,7 @@ __device__ void ROContext::putmem_nbi_wg(void *dest, const void *source, } else { if (is_thread_zero_in_block()) { build_queue_element(RO_NET_PUT_NBI, dest, const_cast(source), - nelems, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + nelems, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, false); } } @@ -358,7 +358,7 @@ __device__ void ROContext::getmem_nbi_wg(void *dest, const void *source, } else { if (is_thread_zero_in_block()) { build_queue_element(RO_NET_GET_NBI, dest, const_cast(source), - nelems, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + nelems, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, false); } } @@ -376,7 +376,7 @@ __device__ void ROContext::putmem_wave(void *dest, const void *source, } else { if (is_thread_zero_in_wave()) { build_queue_element(RO_NET_PUT, dest, const_cast(source), nelems, - pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -395,7 +395,7 @@ __device__ void ROContext::getmem_wave(void *dest, const void *source, } else { if (is_thread_zero_in_wave()) { build_queue_element(RO_NET_GET, dest, const_cast(source), nelems, - pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } @@ -413,7 +413,7 @@ __device__ void ROContext::putmem_nbi_wave(void *dest, const void *source, } else { if (is_thread_zero_in_wave()) { build_queue_element(RO_NET_PUT_NBI, dest, const_cast(source), - nelems, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + nelems, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, false); } } @@ -431,7 +431,7 @@ __device__ void ROContext::getmem_nbi_wave(void *dest, const void *source, } else { if (is_thread_zero_in_wave()) { build_queue_element(RO_NET_GET_NBI, dest, const_cast(source), - nelems, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + nelems, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, false); } } @@ -665,7 +665,7 @@ __device__ uint64_t next_write_slot(BlockHandle *handle) { __device__ void build_queue_element( ro_net_cmds type, void *dst, void *src, size_t size, int pe, int logPE_stride, int PE_size, int PE_root, void *pWrk, long *pSync, - MPI_Comm team_comm, int ro_net_win_id, BlockHandle *handle, + intptr_t team_comm, int ro_net_win_id, BlockHandle *handle, bool blocking, volatile char *status, bool default_ctx, ROCSHMEM_OP op, ro_net_types datatype) { auto write_slot{next_write_slot(handle)}; diff --git a/src/reverse_offload/context_ro_device.hpp b/src/reverse_offload/context_ro_device.hpp index 8f013634e3..1b06f8c0c7 100644 --- a/src/reverse_offload/context_ro_device.hpp +++ b/src/reverse_offload/context_ro_device.hpp @@ -35,7 +35,7 @@ namespace rocshmem { __device__ void build_queue_element( ro_net_cmds type, void *dst, void *src, size_t size, int pe, int logPE_stride, int PE_size, int PE_root, void *pWrk, long *pSync, - MPI_Comm team_comm, int ro_net_win_id, BlockHandle *handle, + intptr_t team_comm, int ro_net_win_id, BlockHandle *handle, bool blocking, volatile char *status = nullptr, bool default_ctx = false, ROCSHMEM_OP op = ROCSHMEM_SUM, ro_net_types datatype = RO_NET_INT); diff --git a/src/reverse_offload/context_ro_tmpl_device.hpp b/src/reverse_offload/context_ro_tmpl_device.hpp index 16a1b6f857..e8748e73b1 100644 --- a/src/reverse_offload/context_ro_tmpl_device.hpp +++ b/src/reverse_offload/context_ro_tmpl_device.hpp @@ -121,7 +121,7 @@ __device__ int ROContext::reduce(rocshmem_team_t team, T *dest, ROTeam *team_obj{reinterpret_cast(team)}; build_queue_element(RO_NET_TEAM_REDUCE, dest, const_cast(source), - nreduce, 0, 0, 0, 0, nullptr, nullptr, team_obj->mpi_comm, + nreduce, 0, 0, 0, 0, nullptr, nullptr, (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, Op, GetROType::Type); @@ -152,7 +152,7 @@ __device__ void ROContext::p(T *dest, T value, int pe) { reinterpret_cast(&value), sizeof(T)); } else { build_queue_element(RO_NET_P, dest, &value, sizeof(T), pe, 0, 0, 0, nullptr, - nullptr, (MPI_Comm)NULL, ro_net_win_id, + nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx); } } @@ -196,7 +196,7 @@ __device__ T ROContext::amo_fetch_cas(void *dst, T value, T cond, int pe) { build_queue_element(RO_NET_AMO_FCAS, dst, reinterpret_cast(source), value, pe, 0, 0, 0, reinterpret_cast(static_cast(cond)), - nullptr, (MPI_Comm)NULL, ro_net_win_id, block_handle, + nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_SUM, GetROType::Type); __threadfence(); @@ -215,7 +215,7 @@ template __device__ T ROContext::amo_fetch_add(void *dst, T value, int pe) { auto source{get_atomic_ret_buf()}; build_queue_element(RO_NET_AMO_FOP, dst, reinterpret_cast(source), value, - pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_SUM, GetROType::Type); __threadfence(); @@ -234,7 +234,7 @@ template __device__ T ROContext::amo_swap(void *dst, T value, int pe) { auto source{get_atomic_ret_buf()}; build_queue_element(RO_NET_AMO_FOP, dst, reinterpret_cast(source), - value, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + value, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_REPLACE, GetROType::Type); __threadfence(); @@ -253,7 +253,7 @@ template __device__ T ROContext::amo_fetch_and(void *dst, T value, int pe) { auto source{get_atomic_ret_buf()}; build_queue_element(RO_NET_AMO_FOP, dst, reinterpret_cast(source), - value, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + value, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_AND, GetROType::Type); __threadfence(); @@ -272,7 +272,7 @@ template __device__ T ROContext::amo_fetch_or(void *dst, T value, int pe) { auto source{get_atomic_ret_buf()}; build_queue_element(RO_NET_AMO_FOP, dst, reinterpret_cast(source), - value, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + value, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_OR, GetROType::Type); __threadfence(); @@ -291,7 +291,7 @@ template __device__ T ROContext::amo_fetch_xor(void *dst, T value, int pe) { auto source{get_atomic_ret_buf()}; build_queue_element(RO_NET_AMO_FOP, dst, reinterpret_cast(source), - value, pe, 0, 0, 0, nullptr, nullptr, (MPI_Comm)NULL, + value, pe, 0, 0, 0, nullptr, nullptr, NULL, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_XOR, GetROType::Type); __threadfence(); @@ -318,7 +318,7 @@ __device__ void ROContext::broadcast(rocshmem_team_t team, T *dest, build_queue_element(RO_NET_TEAM_BROADCAST, dest, const_cast(source), nelems, 0, 0, 0, pe_root, nullptr, nullptr, - team_obj->mpi_comm, ro_net_win_id, block_handle, true, + (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_SUM, GetROType::Type); @@ -337,7 +337,7 @@ __device__ void ROContext::alltoall(rocshmem_team_t team, T *dest, build_queue_element(RO_NET_ALLTOALL, dest, const_cast(source), nelems, 0, 0, 0, 0, team_obj->ata_buffer, nullptr, - team_obj->mpi_comm, ro_net_win_id, block_handle, true, + (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_SUM, GetROType::Type); @@ -356,7 +356,7 @@ __device__ void ROContext::fcollect(rocshmem_team_t team, T *dest, build_queue_element(RO_NET_FCOLLECT, dest, const_cast(source), nelems, 0, 0, 0, 0, team_obj->ata_buffer, nullptr, - team_obj->mpi_comm, ro_net_win_id, block_handle, true, + (intptr_t)team_obj->mpi_comm, ro_net_win_id, block_handle, true, get_status_flag(), is_default_ctx, ROCSHMEM_SUM, GetROType::Type); diff --git a/src/reverse_offload/mpi_transport.cpp b/src/reverse_offload/mpi_transport.cpp index 1757137905..93f722fe3c 100644 --- a/src/reverse_offload/mpi_transport.cpp +++ b/src/reverse_offload/mpi_transport.cpp @@ -159,7 +159,7 @@ void MPITransport::submitRequestsToMPI() { case RO_NET_TEAM_REDUCE: team_reduction(next_element.dst, next_element.src, next_element.ol1.size, next_element.ro_net_win_id, queue_idx, - next_element.team_comm, + (MPI_Comm)next_element.team_comm, static_cast(next_element.op), static_cast(next_element.datatype), next_element.status, true); @@ -170,7 +170,7 @@ void MPITransport::submitRequestsToMPI() { case RO_NET_TEAM_BROADCAST: team_broadcast(next_element.dst, next_element.src, next_element.ol1.size, next_element.ro_net_win_id, queue_idx, - next_element.team_comm, next_element.PE_root, + (MPI_Comm)next_element.team_comm, next_element.PE_root, static_cast(next_element.datatype), next_element.status, true); DPRINTF( @@ -181,7 +181,7 @@ void MPITransport::submitRequestsToMPI() { break; case RO_NET_ALLTOALL: alltoall(next_element.dst, next_element.src, next_element.ol1.size, - next_element.ro_net_win_id, queue_idx, next_element.team_comm, + next_element.ro_net_win_id, queue_idx, (MPI_Comm)next_element.team_comm, next_element.ol2.pWrk, static_cast(next_element.datatype), next_element.status, true); @@ -191,7 +191,7 @@ void MPITransport::submitRequestsToMPI() { break; case RO_NET_FCOLLECT: fcollect(next_element.dst, next_element.src, next_element.ol1.size, - next_element.ro_net_win_id, queue_idx, next_element.team_comm, + next_element.ro_net_win_id, queue_idx, (MPI_Comm)next_element.team_comm, next_element.ol2.pWrk, static_cast(next_element.datatype), next_element.status, true); @@ -201,13 +201,13 @@ void MPITransport::submitRequestsToMPI() { break; case RO_NET_BARRIER: barrier(queue_idx, next_element.status, true, - next_element.team_comm == NULL ? ro_net_comm_world : next_element.team_comm, + next_element.team_comm == ((intptr_t) NULL) ? ro_net_comm_world : (MPI_Comm)next_element.team_comm, true); DPRINTF("Submitted Barrier_all\n"); break; case RO_NET_SYNC: barrier(queue_idx, next_element.status, true, - next_element.team_comm == NULL ? ro_net_comm_world : next_element.team_comm, + next_element.team_comm == ((intptr_t) NULL) ? ro_net_comm_world : (MPI_Comm)next_element.team_comm, false); DPRINTF("Submitted Sync\n"); break; diff --git a/src/reverse_offload/queue_proxy.hpp b/src/reverse_offload/queue_proxy.hpp index 5898b93d00..bfa60e690b 100644 --- a/src/reverse_offload/queue_proxy.hpp +++ b/src/reverse_offload/queue_proxy.hpp @@ -66,7 +66,7 @@ typedef struct queue_element { int datatype{-1}; int PE_root{-1}; volatile char *status{nullptr}; - MPI_Comm team_comm{}; + intptr_t team_comm{}; union { size_t size; unsigned long long atomic_value;