Files
rocm-systems/tests/functional_tests/barrier_all_on_stream_tester.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

152 строки
5.3 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 "barrier_all_on_stream_tester.hpp"
#include <rocshmem/rocshmem.hpp>
#include <hip/hip_runtime.h>
#include <cstring>
#include <cassert>
#include <vector>
/******************************************************************************
* HOST TESTER CLASS METHODS
*****************************************************************************/
BarrierAllOnStreamTester::BarrierAllOnStreamTester(TesterArguments args)
: Tester(args) {
my_pe = rocshmem_my_pe();
n_pes = rocshmem_n_pes();
char *value{nullptr};
if ((value = getenv("ROCSHMEM_TEST_NUM_STREAMS"))) {
num_streams = atoi(value);
} else {
// Default to 1 stream
num_streams = 1;
}
// Check if we should test with nullptr (default stream)
use_default_stream = false;
if ((value = getenv("ROCSHMEM_TEST_USE_DEFAULT_STREAM"))) {
use_default_stream = (atoi(value) != 0);
if (use_default_stream) {
num_streams = 1; // Only test with one nullptr stream
}
}
streams.resize(num_streams);
start_events_timed.resize(num_streams);
stop_events_timed.resize(num_streams);
for (int i = 0; i < num_streams; i++) {
if (use_default_stream) {
streams[i] = nullptr; // Use default stream (0)
} else {
CHECK_HIP(hipStreamCreate(&streams[i]));
}
CHECK_HIP(hipEventCreate(&start_events_timed[i]));
CHECK_HIP(hipEventCreate(&stop_events_timed[i]));
}
}
BarrierAllOnStreamTester::~BarrierAllOnStreamTester() {
for (int i = 0; i < num_streams; i++) {
CHECK_HIP(hipEventDestroy(stop_events_timed[i]));
CHECK_HIP(hipEventDestroy(start_events_timed[i]));
// Don't destroy default stream (nullptr)
if (!use_default_stream) {
CHECK_HIP(hipStreamDestroy(streams[i]));
}
}
}
void BarrierAllOnStreamTester::preLaunchKernel() {
// No specific setup needed for barrier
}
void BarrierAllOnStreamTester::postLaunchKernel() {
// Synchronize all streams to ensure events are recorded
for (int i = 0; i < num_streams; i++) {
CHECK_HIP(hipStreamSynchronize(streams[i]));
}
// Get elapsed time for each stream from HIP events
for (int stream_id = 0; stream_id < num_streams && stream_id < num_timers;
stream_id++) {
float elapsed_time_ms = 0.0f;
CHECK_HIP(hipEventElapsedTime(&elapsed_time_ms,
start_events_timed[stream_id],
stop_events_timed[stream_id]));
// Convert milliseconds to GPU cycles
// wall_clk_rate is in kHz, so: cycles = ms * wall_clk_rate
long long int elapsed_cycles =
static_cast<long long int>(elapsed_time_ms *
static_cast<float>(wall_clk_rate));
start_time[stream_id] = 0;
end_time[stream_id] = elapsed_cycles;
}
// Fill remaining timers with zero if num_timers > num_streams
for (int i = num_streams; i < num_timers; i++) {
start_time[i] = 0;
end_time[i] = 0;
}
}
void BarrierAllOnStreamTester::resetBuffers(size_t size) {}
void BarrierAllOnStreamTester::launchKernel(dim3 gridSize, dim3 blockSize,
int loop, size_t size) {
// Execute warmup iterations (skip)
for (int i = 0; i < args.skip; i++) {
for (int stream_id = 0; stream_id < num_streams; stream_id++) {
rocshmem_barrier_all_on_stream(streams[stream_id]);
}
}
for (int i = 0; i < loop; i++) {
for (int stream_id = 0; stream_id < num_streams; stream_id++) {
// Record start event for this stream on first iteration
if (i == 0) {
CHECK_HIP(hipEventRecord(start_events_timed[stream_id],
streams[stream_id]));
}
rocshmem_barrier_all_on_stream(streams[stream_id]);
// Record stop event for this stream on last iteration
if (i == loop - 1) {
CHECK_HIP(hipEventRecord(stop_events_timed[stream_id],
streams[stream_id]));
}
}
}
num_msgs = (loop + args.skip) * num_streams;
num_timed_msgs = loop * num_streams;
}
void BarrierAllOnStreamTester::verifyResults(size_t size) {}