Files
rocm-systems/src/reverse_offload/context_ro_host.cpp
T
Anatolii Rozanov d0c8380650 Add host API for *_on_stream operations (#340)
* Add functional test for barrier_all_on_stream

* Add rocshmem_barrier_all_on_stream support for GDA and RO backends

Implements rocshmem_barrier_all_on_stream operation for
GPU Direct Access and Reverse Offload backends.

Previously, rocshmem_barrier_all_on_stream was only supported for IPC backend.

* Add functional test for rocshmem_broadcastmem_on_stream

* Add host-side rocshmem_broadcastmem_on_stream API

Implement stream-based broadcast collective operation

- Add rocshmem_broadcastmem_on_stream host API and kernel implementation
- Add functional test TeamBroadcastmemOnStreamTester with multi-stream
  support and correctness verification
- Use per-workgroup contexts to avoid contention across parallel streams

API:
rocshmem_broadcastmem_on_stream(team, dest, source, nelems, pe_root, stream)

* Add functional test for rocshmem_getmem_on_stream

* Add host-side rocshmem_getmem_on_stream API

Implement stream-based point-to-point RMA get operation

- Add rocshmem_getmem_on_stream host API and kernel implementation
- Support for asynchronous getmem operations on HIP streams
- Add backend support for GDA, RO, and IPC contexts
- Use work-group collective getmem for efficient memory transfer

API:
rocshmem_getmem_on_stream(dest, source, nelems, pe, stream)

(AI Assist)

* Add host-side rocshmem_putmem_on_stream API

- Add rocshmem_putmem_on_stream for asynchronous remote writes
- Support for concurrent RMA operations on HIP streams
- Add backend support for GDA, RO, and IPC contexts
- Use work-group device collective operation

API:
rocshmem_putmem_on_stream(dest, source, bytes, pe, stream)

(AI Assist)

* Add functional test for rocshmem_putmem_on_stream

* Add host-side rocshmem_putmem_signal_on_stream API

Enables asynchronous putmem operations with signaling on HIP streams.

The implementation includes:
- Kernel wrapper rocshmem_putmem_signal_kernel
- Host interface putmem_signal_on_stream method
- Context layer support across all backends (IPC, GDA, RO)
- Public API

Function signature:
void rocshmem_putmem_signal_on_stream(void *dest, const void *source,
                                      size_t bytes, uint64_t *sig_addr,
                                      uint64_t signal, int sig_op,
                                      int pe, hipStream_t stream);

* Add functional test for rocshmem_putmem_signal_on_stream

* Add host-side rocshmem_signal_wait_until_on_stream API

Enables asynchronous signal wait operations on HIP streams.

The implementation includes:
- Kernel wrapper rocshmem_signal_wait_until_kernel
- Host interface signal_wait_until_on_stream method
- Context layer support across all backends (IPC, GDA, RO)
- Native uint64_t support in wait_until API (generated from P2P_SYNC.py)

Function signature:
void rocshmem_signal_wait_until_on_stream(uint64_t *sig_addr, int cmp,
                                          uint64_t cmp_value,
                                          hipStream_t stream);

(AI Assist)

* Add functional test for rocshmem_signal_wait_until_on_stream

* Add documentation for stream API functions

This commit adds API documentation for the following host-side
stream functions:

- rocshmem_barrier_all_on_stream (collective routines)
- rocshmem_broadcastmem_on_stream (collective routines)
- rocshmem_getmem_on_stream (RMA operations)
- rocshmem_putmem_on_stream (RMA operations)
- rocshmem_putmem_signal_on_stream (signaling operations)
- rocshmem_signal_wait_until_on_stream (point-to-point sync)

The documentation includes function signatures, parameter descriptions,
and detailed explanations of asynchronous behavior and stream handling.

(AI Assist)

* Rename "bytes" -> "nelems"

* Add "_TEST_" to the variables used in tests

* Remove incorrect hipStreamDefault usage

hipStreamDefault is not a default stream. This is a flag.

If stream == nullptr, then just pass it to kernel. It will launch the kernel on the default stream
2025-12-09 08:55:46 -06:00

198 lines
7.5 KiB
C++

/******************************************************************************
* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
*
* SPDX-License-Identifier: MIT
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to
* deal in the Software without restriction, including without limitation the
* rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
* sell copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
* IN THE SOFTWARE.
*****************************************************************************/
#include "context_ro_host.hpp"
#include "rocshmem/rocshmem_config.h" // NOLINT(build/include_subdir)
#include "backend_type.hpp"
#include "context_incl.hpp"
#include "host/host.hpp"
#include "backend_ro.hpp"
namespace rocshmem {
__host__ ROHostContext::ROHostContext(Backend *backend, long options)
: Context(backend) {
ROBackend *b{static_cast<ROBackend *>(backend)};
host_interface = b->host_interface;
context_window_info = dynamic_cast<WindowInfoMPI*>(host_interface->acquire_window_context());
int *pes_with_ipc_avail = new int[backend->ipcImpl.shm_size];
char** ipc_bases = new char*[b->ipcImpl.shm_size];
if (backend->ipcImpl.pes_with_ipc_avail != nullptr) {
CHECK_HIP(hipMemcpy(pes_with_ipc_avail,
backend->ipcImpl.pes_with_ipc_avail,
backend->ipcImpl.shm_size * sizeof(int),
hipMemcpyDeviceToHost));
CHECK_HIP(hipMemcpy(ipc_bases,
backend->ipcImpl.ipc_bases,
backend->ipcImpl.shm_size * sizeof(char *),
hipMemcpyDeviceToHost));
}
ipcImpl_.pes_with_ipc_avail = pes_with_ipc_avail;
ipcImpl_.ipc_bases = ipc_bases;
ipcImpl_.shm_size = backend->ipcImpl.shm_size;
ipcImpl_.shm_rank = backend->ipcImpl.shm_rank;
}
__host__ ROHostContext::~ROHostContext() {
delete[] ipcImpl_.pes_with_ipc_avail;
delete[] ipcImpl_.ipc_bases;
// host_interface->release_window_context(context_window_info);
}
__host__ void ROHostContext::putmem_nbi(void *dest, const void *source,
size_t nelems, int pe) {
DPRINTF("Function: ro_net_host_putmem_nbi\n");
host_interface->putmem_nbi(dest, source, nelems, pe, context_window_info);
}
__host__ void ROHostContext::getmem_nbi(void *dest, const void *source,
size_t nelems, int pe) {
DPRINTF("Function: ro_net_host_getmem_nbi\n");
host_interface->getmem_nbi(dest, source, nelems, pe, context_window_info);
}
__host__ void ROHostContext::putmem(void *dest, const void *source,
size_t nelems, int pe) {
DPRINTF("Function: ro_net_host_putmem\n");
host_interface->putmem(dest, source, nelems, pe, context_window_info);
}
__host__ void ROHostContext::getmem(void *dest, const void *source,
size_t nelems, int pe) {
DPRINTF("Function: ro_net_host_getmem\n");
host_interface->getmem(dest, source, nelems, pe, context_window_info);
}
__host__ void ROHostContext::fence() {
DPRINTF("Function: ro_net_host_fence\n");
host_interface->fence(context_window_info);
}
__host__ void ROHostContext::quiet() {
DPRINTF("Function: ro_net_host_quiet\n");
host_interface->quiet(context_window_info);
}
__host__ void *ROHostContext::shmem_ptr(const void *dest, int pe) {
DPRINTF("Function: ro_net_host_shmem_ptr\n");
void *ret = nullptr;
int local_pe{-1};
if (ipcImpl_.isIpcAvailable(my_pe, pe, &local_pe)) {
void *dst = const_cast<void *>(dest);
uint64_t L_offset =
reinterpret_cast<char *>(dst) - ipcImpl_.ipc_bases[ipcImpl_.shm_rank];
ret = ipcImpl_.ipc_bases[local_pe] + L_offset;
}
return ret;
}
__host__ void ROHostContext::sync_all() {
DPRINTF("Function: ro_net_host_sync_all\n");
host_interface->sync_all(context_window_info);
}
__host__ void ROHostContext::barrier_all() {
DPRINTF("Function: ro_net_host_barrier_all\n");
host_interface->fence(context_window_info);
host_interface->barrier_for_sync();
}
__host__ void ROHostContext::barrier_all_on_stream(hipStream_t stream) {
DPRINTF("Function: ro_net_host_barrier_all_on_stream\n");
host_interface->barrier_all_on_stream(stream);
}
__host__ void ROHostContext::alltoallmem_on_stream(rocshmem_team_t team,
void *dest,
const void *source,
size_t size,
hipStream_t stream) {
DPRINTF("Function: ro_net_host_alltoallmem_on_stream\n");
host_interface->alltoallmem_on_stream(team, dest, source, size, stream);
}
__host__ void ROHostContext::broadcastmem_on_stream(rocshmem_team_t team,
void *dest,
const void *source,
size_t nelems, int pe_root,
hipStream_t stream) {
DPRINTF("Function: ro_net_host_broadcastmem_on_stream\n");
host_interface->broadcastmem_on_stream(team, dest, source, nelems, pe_root,
stream);
}
__host__ void ROHostContext::getmem_on_stream(void *dest, const void *source,
size_t nelems, int pe,
hipStream_t stream) {
DPRINTF("Function: ro_net_host_getmem_on_stream\n");
host_interface->getmem_on_stream(dest, source, nelems, pe, stream);
}
__host__ void ROHostContext::putmem_on_stream(void *dest, const void *source,
size_t nelems, int pe,
hipStream_t stream) {
DPRINTF("Function: ro_net_host_putmem_on_stream\n");
host_interface->putmem_on_stream(dest, source, nelems, pe, stream);
}
__host__ void ROHostContext::putmem_signal_on_stream(
void *dest, const void *source, size_t nelems, uint64_t *sig_addr,
uint64_t signal, int sig_op, int pe, hipStream_t stream) {
DPRINTF("Function: ro_net_host_putmem_signal_on_stream\n");
host_interface->putmem_signal_on_stream(dest, source, nelems, sig_addr,
signal, sig_op, pe, stream);
}
__host__ void ROHostContext::signal_wait_until_on_stream(uint64_t *sig_addr,
int cmp,
uint64_t cmp_value,
hipStream_t stream) {
DPRINTF("Function: ro_net_host_signal_wait_until_on_stream\n");
host_interface->signal_wait_until_on_stream(sig_addr, cmp, cmp_value, stream);
}
} // namespace rocshmem