* Update primitive tests for multi-workgroup support

* Update workgroup primitive tests for multi-workgroup support

* Update workfront primitive tests for multi-workgroup support

* Update team based primitive tests for multi-workgroup support

* Update RMA functional tests to capture timing after quiet call
   - Modified RMA functional tests to record the time after a `quiet` call in thread, wavefront, and workgroup RMA calls.

* Improve error handling and memory management
   - Replaced `cout` with `cerr` for improved error reporting.
   - Ensured all allocated memory is freed when `rocshmem_malloc` fails.

* Update start time in primitive tests and latency calculations
   - Modified primitive tests to capture the earliest start time.
   - Updated latency calculations in functional tests.

* Remove `GetSwarmTester`

* Update start time in team primitive tests

* Invoke quiet call from a single thread within a block on a rocshmem context
Этот коммит содержится в:
Avinash Kethineedi authored and GitHub committed 2025-03-18 14:39:57 -05:00
1 родитель b7f3839f27
Коммит aa3121a967
16 files changed
+463 -460

No files matched your search

+97 -38
Просмотреть файл
@@ -32,43 +32,83 @@ rocshmem_team_t team_primitive_world_dup;
* DEVICE TEST KERNEL
*****************************************************************************/
__global__ void TeamCtxPrimitiveTest(int loop, int skip, long long int *start_time,
long long int *end_time, char *s_buf,
char *r_buf, int size, TestType type,
ShmemContextType ctx_type,
long long int *end_time, char *source,
char *dest, int size, TestType type,
ShmemContextType ctx_type, int wf_size,
rocshmem_team_t team) {
__shared__ rocshmem_ctx_t ctx;
int wg_id = get_flat_grid_id();
int t_id = get_flat_block_id();
int wf_id = t_id / wf_size;
rocshmem_wg_init();
rocshmem_wg_team_create_ctx(team, ctx_type, &ctx);
if (hipThreadIdx_x == 0) {
/**
* Shared array to capture the start time for each wavefront
* Max threads per block = 1024, wavefront size = 64 (in most GPUs)
* Maximum array size required = 1024/64 = 16
*/
__shared__ long long int wf_start_time[16];
for (int i = 0; i < loop + skip; i++) {
if (i == skip) {
start_time[wg_id] = wall_clock64();
}
switch (type) {
case TeamCtxGetTestType:
rocshmem_ctx_getmem(ctx, r_buf, s_buf, size, 1);
break;
case TeamCtxGetNBITestType:
rocshmem_ctx_getmem_nbi(ctx, r_buf, s_buf, size, 1);
break;
case TeamCtxPutTestType:
rocshmem_ctx_putmem(ctx, r_buf, s_buf, size, 1);
break;
case TeamCtxPutNBITestType:
rocshmem_ctx_putmem_nbi(ctx, r_buf, s_buf, size, 1);
break;
default:
break;
/**
* Calculate start index for each thread within the grid
*/
uint64_t offset = size * get_flat_id();
source += offset;
dest += offset;
for (int i = 0; i < loop + skip; i++) {
if (i == skip) {
__syncthreads();
// Ensures all RMA calls from the skip loops are completed
if(is_thread_zero_in_block()) {
rocshmem_ctx_quiet(ctx);
}
__syncthreads();
// Capture the start time of each wavefront to identify the earliest one
wf_start_time[wf_id] = wall_clock64();
}
switch (type) {
case TeamCtxGetTestType:
rocshmem_ctx_getmem(ctx, dest, source, size, 1);
break;
case TeamCtxGetNBITestType:
rocshmem_ctx_getmem_nbi(ctx, dest, source, size, 1);
break;
case TeamCtxPutTestType:
rocshmem_ctx_putmem(ctx, dest, source, size, 1);
break;
case TeamCtxPutNBITestType:
rocshmem_ctx_putmem_nbi(ctx, dest, source, size, 1);
break;
default:
break;
}
}
__syncthreads();
if(is_thread_zero_in_block()) {
rocshmem_ctx_quiet(ctx);
}
end_time[wg_id] = wall_clock64();
/**
* End time of the last wavefront is recorded by overwriting
* the value previously set by earlier wavefronts.
*/
end_time[wg_id] = wall_clock64();
// Find the earliest start time
int num_wfs = (get_flat_block_size() - 1 ) / wf_size + 1;
for (int i = num_wfs / 2; i > 0; i >>= 1 ) {
if(t_id < i) {
wf_start_time[t_id] = min(wf_start_time[t_id], wf_start_time[t_id + i]);
}
}
__syncthreads();
if (t_id == 0) {
start_time[wg_id] = wf_start_time[0];
}
rocshmem_wg_ctx_destroy(&ctx);
@@ -80,18 +120,35 @@ __global__ void TeamCtxPrimitiveTest(int loop, int skip, long long int *start_ti
*****************************************************************************/
TeamCtxPrimitiveTester::TeamCtxPrimitiveTester(TesterArguments args)
: Tester(args) {
s_buf = (char *)rocshmem_malloc(args.max_msg_size * args.wg_size);
r_buf = (char *)rocshmem_malloc(args.max_msg_size * args.wg_size);
size_t buff_size = args.max_msg_size * args.wg_size * args.num_wgs;
source = (char *)rocshmem_malloc(buff_size);
dest = (char *)rocshmem_malloc(buff_size);
if (source == nullptr || dest == nullptr) {
std::cerr << "Error allocating memory from symmetric heap" << std::endl;
std::cerr << "source: " << source << ", dest: " << dest << std::endl;
if (source) {
rocshmem_free(source);
}
if (dest) {
rocshmem_free(dest);
}
rocshmem_global_exit(1);
}
for(size_t i = 0; i < buff_size; i++) {
source[i] = static_cast<char>('a' + i % 26);
}
}
TeamCtxPrimitiveTester::~TeamCtxPrimitiveTester() {
rocshmem_free(s_buf);
rocshmem_free(r_buf);
rocshmem_free(source);
rocshmem_free(dest);
}
void TeamCtxPrimitiveTester::resetBuffers(uint64_t size) {
memset(s_buf, '0', args.max_msg_size * args.wg_size);
memset(r_buf, '1', args.max_msg_size * args.wg_size);
size_t buff_size = size * args.wg_size * args.num_wgs;
memset(dest, '1', buff_size);
}
void TeamCtxPrimitiveTester::preLaunchKernel() {
@@ -107,12 +164,12 @@ void TeamCtxPrimitiveTester::launchKernel(dim3 gridSize, dim3 blockSize,
size_t shared_bytes = 0;
hipLaunchKernelGGL(TeamCtxPrimitiveTest, gridSize, blockSize, shared_bytes,
stream, loop, args.skip, start_time, end_time, s_buf,
r_buf, size, _type, _shmem_context,
stream, loop, args.skip, start_time, end_time, source,
dest, size, _type, _shmem_context, wf_size,
team_primitive_world_dup);
num_msgs = (loop + args.skip) * gridSize.x;
num_timed_msgs = loop * gridSize.x;
num_msgs = (loop + args.skip) * gridSize.x * blockSize.x;
num_timed_msgs = loop * gridSize.x * blockSize.x;
}
void TeamCtxPrimitiveTester::postLaunchKernel() {
@@ -124,10 +181,12 @@ void TeamCtxPrimitiveTester::verifyResults(uint64_t size) {
(_type == TeamCtxGetTestType || _type == TeamCtxGetNBITestType) ? 0 : 1;
if (args.myid == check_id) {
for (uint64_t i = 0; i < size; i++) {
if (r_buf[i] != '0') {
fprintf(stderr, "Data validation error at idx %lu\n", i);
fprintf(stderr, "Got %c, Expected %c\n", r_buf[i], '0');
size_t buff_size = size * args.wg_size * args.num_wgs;
for (uint64_t i = 0; i < buff_size; i++) {
if (dest[i] != source[i]) {
std::cerr << "Data validation error at idx " << i << std::endl;
std::cerr << " Got " << dest[i] << ", Expected "
<< source[i] << std::endl;
exit(-1);
}
}