From 470ede8ef89868d01c157bfe5c5a2bb56e030252 Mon Sep 17 00:00:00 2001 From: Damon McDougall Date: Tue, 15 Jul 2025 14:48:08 -0500 Subject: [PATCH 1/5] SWDEV-538614: Add HIP IPC event sync test --- catch/perftests/event/CMakeLists.txt | 1 + catch/perftests/event/hipIpcEventSyncTest.cc | 280 +++++++++++++++++++ 2 files changed, 281 insertions(+) create mode 100644 catch/perftests/event/hipIpcEventSyncTest.cc diff --git a/catch/perftests/event/CMakeLists.txt b/catch/perftests/event/CMakeLists.txt index a6566f3a169b..d17c92f80855 100644 --- a/catch/perftests/event/CMakeLists.txt +++ b/catch/perftests/event/CMakeLists.txt @@ -18,6 +18,7 @@ set(TEST_SRC hipKernelLookUpPerf.cc hipEventOverFlowPerf.cc + hipIpcEventSyncTest.cc ) hip_add_exe_to_target(NAME perfEventTest TEST_SRC ${TEST_SRC} diff --git a/catch/perftests/event/hipIpcEventSyncTest.cc b/catch/perftests/event/hipIpcEventSyncTest.cc new file mode 100644 index 000000000000..9440ae189c8e --- /dev/null +++ b/catch/perftests/event/hipIpcEventSyncTest.cc @@ -0,0 +1,280 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +const char * sem_name = "/ipc_sem"; +const char * shm_name = "/ipc_shm"; + +#define HIP_CHECK_KERNEL(call) \ +{ \ + do { \ + call; \ + hipError_t err = hipGetLastError(); \ + if (err != hipSuccess) { \ + std::cerr << std::endl \ + << "error: '" << hipGetErrorString(err) << "'(" << err << ") at " << __FILE__ << ":" << __LINE__ << std::endl \ + << std::endl; \ + exit(1); \ + } \ + } while (0); \ +} + +__global__ void kernel1(double * a, std::size_t N) { + int tid = blockIdx.x * blockDim.x + threadIdx.x; + for (std::size_t i = tid; i < N; i += gridDim.x * blockDim.x) { + a[i] = a[i] * a[i]; + } +} + +void die(const char * name, int rank) { + perror(name); + if (rank == 0) { + exit(EXIT_FAILURE); // parent + } + else { + _exit(EXIT_FAILURE); // child + } +} + +void barrier(int size) { + // Semaphore to count the number of processes that arrived + const char * sem_count_name = "/ipc_count"; + sem_t * count = sem_open(sem_count_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, /* initial value */ 0); // 0644 + if (count == SEM_FAILED) die("sem_open", 0); + + // Need another semaphore for the actual barrier + const char * sem_barrier_name = "/ipc_barrier"; + sem_t * barrier = sem_open(sem_barrier_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, /* initial value */ 0); // 0644 + if (barrier == SEM_FAILED) die("sem_open", 0); + + // Each process increments count + sem_post(count); + + int count_value; + sem_getvalue(count, &count_value); + + if (count_value == size) { + // All processes completed the increment. The last one that gets here unlocks the barrier + sem_post(barrier); + } + + // Processes will wait until the last one sees `count` is `size` and when that happens, one process is + // unblocked, and that unblocked process will unblock the next one, and so on... + sem_wait(barrier); + sem_post(barrier); + + // Cleanup + sem_unlink(sem_count_name); + sem_close(count); + sem_unlink(sem_barrier_name); + sem_close(barrier); +} + +int execute(int rank) { + int size = 2; // There are only two processes in this test + int device_count; + HIPCHECK(hipGetDeviceCount(&device_count)); + + if (device_count < 2) { + if (rank == 0) { + printf("Expected 2 GPUs, but only %d visible\n", device_count); + } + return EXIT_FAILURE; + } + + HIPCHECK(hipSetDevice(rank)); + + if (rank == 0) printf("Allocating memory and launching warmup workload\n"); + + std::size_t N = 1000 * 1000 * 1000; + printf("rank %d allocating %zu bytes on the host\n", rank, N * sizeof(double)); + double * a = (double *)malloc(sizeof(double) * N); // 8 GB + double * da; + + printf("rank %d allocating %zu bytes on the device\n", rank, N * sizeof(double)); + HIPCHECK(hipMalloc(&da, sizeof(double) * N)); + printf("rank done %d allocating %zu bytes on the device\n", rank, N * sizeof(double)); + + // Fill a + for (std::size_t i = 0; i < N; i++) { + a[i] = 0.5 * i; + } + + // Copy to device + HIPCHECK(hipMemcpy(da, a, sizeof(double) * N, hipMemcpyHostToDevice)); + + hipStream_t stream; + HIPCHECK(hipStreamCreate(&stream)); + + // Warm up + int num_kernels = 10; + for (int i = 0; i < num_kernels; i++) { + // HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); + hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N); + } + + hipEvent_t end_warmup; + HIPCHECK(hipEventCreate(&end_warmup)); + + // Time the workload + if (rank == 0) { + num_kernels = 1000; // Make device zero very busy + } + + for (int i = 0; i < num_kernels; i++) { + //HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); + hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N); + } + HIPCHECK(hipEventRecord(end_warmup, stream)); + + // Wait for the warmup to finish + HIPCHECK(hipEventSynchronize(end_warmup)); + + // Let's flush everything + HIPCHECK(hipDeviceSynchronize()); + + hipEvent_t local_event; + hipEvent_t remote_event; + hipIpcEventHandle_t local_event_handle; + hipIpcEventHandle_t remote_event_handle; + HIPCHECK(hipEventCreateWithFlags(&local_event, hipEventDisableTiming | hipEventInterprocess)); + HIPCHECK(hipIpcGetEventHandle(&local_event_handle, local_event)); + + // Open a semaphore -- we don't care which process gets here first + sem_t * sem = sem_open(sem_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, /* initial value */ 0); // 0644 + if (sem == SEM_FAILED) die("sem_open", rank); + + // Now open the shm region for the ipc handle + int fd = shm_open(shm_name, O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH); // 0644 + if (fd < 0) die("shm_open", rank); + + if (ftruncate(fd, sizeof(hipIpcEventHandle_t)) != 0) { // What if this thing is an opaque pointer type? + die("ftruncate", rank); + } + + void * shm = mmap(NULL, sizeof(hipIpcEventHandle_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); + close(fd); + if (shm == MAP_FAILED) die("mmap", rank); + + // Share ipc event handle with rank 1 + // Only rank 0 writes + if (rank == 0) { + memcpy(shm, &local_event_handle, sizeof(hipIpcEventHandle_t)); + if (sem_post(sem)) die("sem_post", rank); + } + else { + if (sem_wait(sem)) die("sem_wait", rank); + memcpy(&remote_event_handle, shm, sizeof(hipIpcEventHandle_t)); + + // open handle + HIPCHECK(hipIpcOpenEventHandle(&remote_event, remote_event_handle)); + } + + // Now reproduce the problem + + // kernel in stream + if (rank == 0) printf("Launching workload\n"); + for (int i = 0; i < num_kernels; i++) { + //HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); + hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N); + } + + if (rank == 0) { + printf("Testing ipc event sync\n"); + HIPCHECK(hipEventRecord(local_event, stream)); + } + + barrier(size); + + // At this point, rank 1 knows that the remote ipc event has been recorded + + // wait on remote event + double rank_0_time, rank_1_time; + if (rank == 1) { + auto ipc_start = std::chrono::steady_clock::now(); + HIPCHECK(hipEventSynchronize(remote_event)); + auto ipc_end = std::chrono::steady_clock::now(); + const std::chrono::duration ipc_elapsed = ipc_end - ipc_start; + + double elapsed_time = ipc_elapsed.count(); + + // Copy time to shm so that rank 0 can later read it + memcpy(shm, &elapsed_time, sizeof(double)); + } + if (rank == 0) { + auto local_start = std::chrono::steady_clock::now(); + HIPCHECK(hipDeviceSynchronize()); + auto local_end = std::chrono::steady_clock::now(); + const std::chrono::duration local_elapsed = local_end - local_start; + rank_0_time = local_elapsed.count(); + } + + barrier(size); + + // Compute the difference in recorded times + // Re-use the shm region we created to communicate the child's computed time and compare + int ret_code = EXIT_FAILURE; + if (rank == 0) { + memcpy(&rank_1_time, shm, sizeof(double)); + printf("hipDeviceSynchronize time: %lf s\n", rank_0_time); + printf("IPC hipEventSynchronize time: %lf s\n", rank_1_time); + + double diff = rank_0_time - rank_1_time; + double err = diff / rank_0_time; + + // Pass if the error isn't "too large" + if (err < 0.3) { + printf("Test: PASSED\n"); + ret_code = EXIT_SUCCESS; + } + else { + printf("Test: FAILED\n"); + ret_code = EXIT_FAILURE; + } + } + else { + // Child process doesn't fail + ret_code = EXIT_SUCCESS; + } + + // Clean up + munmap(shm, sizeof(hipIpcEventHandle_t)); + shm_unlink(shm_name); + sem_unlink(sem_name); + sem_close(sem); + + // Clean up + // We don't need to destroy the remote event because we destroy the exported event. + HIPCHECK(hipEventDestroy(local_event)); + + HIPCHECK(hipEventDestroy(end_warmup)); + HIPCHECK(hipStreamDestroy(stream)); + HIPCHECK(hipFree(da)); + free(a); + + return ret_code; +} + +TEST_CASE("Unit_hipIpcEventSynchronize_Test") { + int ret; + pid_t pid = fork(); + switch (pid) { + case -1: + perror("fork"); + exit(EXIT_FAILURE); + case 0: + // We're the child process + ret = execute(1); + _exit(ret); + default: + // We're the parent process + ret = execute(0); + exit(ret); + } +} From 9cca1cc05a7375c9e76b538b4a294cea77ab6f6c Mon Sep 17 00:00:00 2001 From: Damon McDougall Date: Tue, 15 Jul 2025 14:56:19 -0500 Subject: [PATCH 2/5] SWDEV-538614: Fix memory leak --- catch/perftests/event/hipEventOverFlowPerf.cc | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/catch/perftests/event/hipEventOverFlowPerf.cc b/catch/perftests/event/hipEventOverFlowPerf.cc index 105a70cafdd2..3a05f98daef3 100644 --- a/catch/perftests/event/hipEventOverFlowPerf.cc +++ b/catch/perftests/event/hipEventOverFlowPerf.cc @@ -109,7 +109,7 @@ TEST_CASE("Unit_hipEventOverFlow_PerfTest") { HIP_CHECK_PERF(hipGetDeviceCount(&mgpu)); stream_pool.resize(mgpu); HIP_CHECK_PERF(hipSetDeviceFlags(hipDeviceScheduleSpin)); - std::vector memory_buffers[2]; + std::vector memory_buffers[mgpu]; for (int i = 0; i < mgpu; i++) { HIP_CHECK_PERF(hipSetDevice(i)); stream_pool[i].resize(12); @@ -160,6 +160,11 @@ TEST_CASE("Unit_hipEventOverFlow_PerfTest") { HIP_CHECK_PERF(hipDeviceSynchronize()); } } + for (int i = 0; i < mgpu; i++) { + HIP_CHECK_PERF(hipSetDevice(i)); + for (int j = 0; j < 128; j++) + HIP_CHECK_PERF(hipFree(memory_buffers[i][j])); + } } /** * End doxygen group PerformanceTest. From fdbbf5847bac56814e2c6b017bcb895e093b04c8 Mon Sep 17 00:00:00 2001 From: Damon McDougall Date: Wed, 16 Jul 2025 12:04:51 -0500 Subject: [PATCH 3/5] SWDEV-538614: Fix clang-format-check failure --- catch/perftests/event/hipEventOverFlowPerf.cc | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/catch/perftests/event/hipEventOverFlowPerf.cc b/catch/perftests/event/hipEventOverFlowPerf.cc index 3a05f98daef3..dfebaa83f220 100644 --- a/catch/perftests/event/hipEventOverFlowPerf.cc +++ b/catch/perftests/event/hipEventOverFlowPerf.cc @@ -109,7 +109,7 @@ TEST_CASE("Unit_hipEventOverFlow_PerfTest") { HIP_CHECK_PERF(hipGetDeviceCount(&mgpu)); stream_pool.resize(mgpu); HIP_CHECK_PERF(hipSetDeviceFlags(hipDeviceScheduleSpin)); - std::vector memory_buffers[mgpu]; + std::vector memory_buffers[mgpu]; for (int i = 0; i < mgpu; i++) { HIP_CHECK_PERF(hipSetDevice(i)); stream_pool[i].resize(12); @@ -162,8 +162,7 @@ TEST_CASE("Unit_hipEventOverFlow_PerfTest") { } for (int i = 0; i < mgpu; i++) { HIP_CHECK_PERF(hipSetDevice(i)); - for (int j = 0; j < 128; j++) - HIP_CHECK_PERF(hipFree(memory_buffers[i][j])); + for (int j = 0; j < 128; j++) HIP_CHECK_PERF(hipFree(memory_buffers[i][j])); } } /** From 6bd5454aaf656eec8f107fce2cb77a0eb0317db2 Mon Sep 17 00:00:00 2001 From: Damon McDougall Date: Wed, 16 Jul 2025 13:03:54 -0500 Subject: [PATCH 4/5] SWDEV-538614: Fix clang-format-check failure --- catch/perftests/event/hipIpcEventSyncTest.cc | 101 ++++++++++--------- 1 file changed, 51 insertions(+), 50 deletions(-) diff --git a/catch/perftests/event/hipIpcEventSyncTest.cc b/catch/perftests/event/hipIpcEventSyncTest.cc index 9440ae189c8e..acf12d3dd661 100644 --- a/catch/perftests/event/hipIpcEventSyncTest.cc +++ b/catch/perftests/event/hipIpcEventSyncTest.cc @@ -8,49 +8,51 @@ #include #include -const char * sem_name = "/ipc_sem"; -const char * shm_name = "/ipc_shm"; - -#define HIP_CHECK_KERNEL(call) \ -{ \ - do { \ - call; \ - hipError_t err = hipGetLastError(); \ - if (err != hipSuccess) { \ - std::cerr << std::endl \ - << "error: '" << hipGetErrorString(err) << "'(" << err << ") at " << __FILE__ << ":" << __LINE__ << std::endl \ - << std::endl; \ - exit(1); \ - } \ - } while (0); \ -} +const char* sem_name = "/ipc_sem"; +const char* shm_name = "/ipc_shm"; + +#define HIP_CHECK_KERNEL(call) \ + { \ + do { \ + call; \ + hipError_t err = hipGetLastError(); \ + if (err != hipSuccess) { \ + std::cerr << std::endl \ + << "error: '" << hipGetErrorString(err) << "'(" << err << ") at " << __FILE__ \ + << ":" << __LINE__ << std::endl \ + << std::endl; \ + exit(1); \ + } \ + } while (0); \ + } -__global__ void kernel1(double * a, std::size_t N) { +__global__ void kernel1(double* a, std::size_t N) { int tid = blockIdx.x * blockDim.x + threadIdx.x; for (std::size_t i = tid; i < N; i += gridDim.x * blockDim.x) { a[i] = a[i] * a[i]; } } -void die(const char * name, int rank) { +void die(const char* name, int rank) { perror(name); if (rank == 0) { exit(EXIT_FAILURE); // parent - } - else { + } else { _exit(EXIT_FAILURE); // child } } void barrier(int size) { // Semaphore to count the number of processes that arrived - const char * sem_count_name = "/ipc_count"; - sem_t * count = sem_open(sem_count_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, /* initial value */ 0); // 0644 + const char* sem_count_name = "/ipc_count"; + sem_t* count = sem_open(sem_count_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, + /* initial value */ 0); // 0644 if (count == SEM_FAILED) die("sem_open", 0); // Need another semaphore for the actual barrier - const char * sem_barrier_name = "/ipc_barrier"; - sem_t * barrier = sem_open(sem_barrier_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, /* initial value */ 0); // 0644 + const char* sem_barrier_name = "/ipc_barrier"; + sem_t* barrier = sem_open(sem_barrier_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, + /* initial value */ 0); // 0644 if (barrier == SEM_FAILED) die("sem_open", 0); // Each process increments count @@ -64,8 +66,8 @@ void barrier(int size) { sem_post(barrier); } - // Processes will wait until the last one sees `count` is `size` and when that happens, one process is - // unblocked, and that unblocked process will unblock the next one, and so on... + // Processes will wait until the last one sees `count` is `size` and when that happens, one + // process is unblocked, and that unblocked process will unblock the next one, and so on... sem_wait(barrier); sem_post(barrier); @@ -94,8 +96,8 @@ int execute(int rank) { std::size_t N = 1000 * 1000 * 1000; printf("rank %d allocating %zu bytes on the host\n", rank, N * sizeof(double)); - double * a = (double *)malloc(sizeof(double) * N); // 8 GB - double * da; + double* a = (double*)malloc(sizeof(double) * N); // 8 GB + double* da; printf("rank %d allocating %zu bytes on the device\n", rank, N * sizeof(double)); HIPCHECK(hipMalloc(&da, sizeof(double) * N)); @@ -128,7 +130,7 @@ int execute(int rank) { } for (int i = 0; i < num_kernels; i++) { - //HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); + // HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N); } HIPCHECK(hipEventRecord(end_warmup, stream)); @@ -147,18 +149,20 @@ int execute(int rank) { HIPCHECK(hipIpcGetEventHandle(&local_event_handle, local_event)); // Open a semaphore -- we don't care which process gets here first - sem_t * sem = sem_open(sem_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, /* initial value */ 0); // 0644 + sem_t* sem = sem_open(sem_name, O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH, + /* initial value */ 0); // 0644 if (sem == SEM_FAILED) die("sem_open", rank); // Now open the shm region for the ipc handle int fd = shm_open(shm_name, O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH); // 0644 if (fd < 0) die("shm_open", rank); - if (ftruncate(fd, sizeof(hipIpcEventHandle_t)) != 0) { // What if this thing is an opaque pointer type? + if (ftruncate(fd, sizeof(hipIpcEventHandle_t)) != + 0) { // What if this thing is an opaque pointer type? die("ftruncate", rank); } - void * shm = mmap(NULL, sizeof(hipIpcEventHandle_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); + void* shm = mmap(NULL, sizeof(hipIpcEventHandle_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); close(fd); if (shm == MAP_FAILED) die("mmap", rank); @@ -167,8 +171,7 @@ int execute(int rank) { if (rank == 0) { memcpy(shm, &local_event_handle, sizeof(hipIpcEventHandle_t)); if (sem_post(sem)) die("sem_post", rank); - } - else { + } else { if (sem_wait(sem)) die("sem_wait", rank); memcpy(&remote_event_handle, shm, sizeof(hipIpcEventHandle_t)); @@ -181,7 +184,7 @@ int execute(int rank) { // kernel in stream if (rank == 0) printf("Launching workload\n"); for (int i = 0; i < num_kernels; i++) { - //HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); + // HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N); } @@ -232,13 +235,11 @@ int execute(int rank) { if (err < 0.3) { printf("Test: PASSED\n"); ret_code = EXIT_SUCCESS; - } - else { + } else { printf("Test: FAILED\n"); ret_code = EXIT_FAILURE; } - } - else { + } else { // Child process doesn't fail ret_code = EXIT_SUCCESS; } @@ -265,16 +266,16 @@ TEST_CASE("Unit_hipIpcEventSynchronize_Test") { int ret; pid_t pid = fork(); switch (pid) { - case -1: - perror("fork"); - exit(EXIT_FAILURE); - case 0: - // We're the child process - ret = execute(1); - _exit(ret); - default: - // We're the parent process - ret = execute(0); - exit(ret); + case -1: + perror("fork"); + exit(EXIT_FAILURE); + case 0: + // We're the child process + ret = execute(1); + _exit(ret); + default: + // We're the parent process + ret = execute(0); + exit(ret); } } From b00d39c17d0a565d391e005268d9444995a3881d Mon Sep 17 00:00:00 2001 From: Damon McDougall Date: Wed, 16 Jul 2025 13:04:44 -0500 Subject: [PATCH 5/5] SWDEV-538614: Deleting commented out code --- catch/perftests/event/hipIpcEventSyncTest.cc | 3 --- 1 file changed, 3 deletions(-) diff --git a/catch/perftests/event/hipIpcEventSyncTest.cc b/catch/perftests/event/hipIpcEventSyncTest.cc index acf12d3dd661..3d1deffd26a9 100644 --- a/catch/perftests/event/hipIpcEventSyncTest.cc +++ b/catch/perftests/event/hipIpcEventSyncTest.cc @@ -117,7 +117,6 @@ int execute(int rank) { // Warm up int num_kernels = 10; for (int i = 0; i < num_kernels; i++) { - // HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N); } @@ -130,7 +129,6 @@ int execute(int rank) { } for (int i = 0; i < num_kernels; i++) { - // HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N); } HIPCHECK(hipEventRecord(end_warmup, stream)); @@ -184,7 +182,6 @@ int execute(int rank) { // kernel in stream if (rank == 0) printf("Launching workload\n"); for (int i = 0; i < num_kernels; i++) { - // HIP_CHECK_KERNEL(hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N)); hipLaunchKernelGGL(kernel1, 1024, 1024, 0, stream, da, N); }