Implement rocshmem_ptr in IPC conduit (#197)

* Implement `rocshmem_ptr` in IPC conduit

* tests: add functional test for `rocshmem_ptr`
  - Add safety check for pointer access and condition check before printing results for `rocshmem_ptr` test
  - Use `rocshmem_put` to store `rocshmem_ptr` availability for data validation

tests/functional_tests/shmem_ptr_tester.cpp

@@ -31,18 +31,86 @@ using namespace rocshmem;
 /******************************************************************************
  * DEVICE TEST KERNEL
  *****************************************************************************/
-__global__ void ShmemPtrTest(char *r_buf, int *available) {
-  rocshmem_wg_init();
+__global__ void ShmemPtrTest(int loop, int skip, long long int *start_time,
+                             long long int *end_time, char *dest, int wf_size,
+                             ShmemContextType ctx_type, int *available) {
+  __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_ctx_create(ctx_type, &ctx);
+
+  /**
+   * 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];
+
+
+  /**
+   * Calculate start index for each thread within the grid
+   */
+  dest += get_flat_id();
+
+  char *local_addr = dest;
+  void *remote_addr = rocshmem_ptr((void *)local_addr, 1);
+  if (remote_addr != NULL) {
+    *available = 1;
+  }
+
+  if(*available) {
+    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();
+      }
 
-  if (hipThreadIdx_x == 0) {
-    char *local_addr = r_buf + 4;
-    void *remote_addr = rocshmem_ptr((void *)local_addr, 1);
-    if (remote_addr != NULL) {
-      *available = 1;
       ((char *)remote_addr)[0] = '1';
     }
   }
 
+  __syncthreads();
+  if(is_thread_zero_in_block()) {
+    rocshmem_ctx_quiet(ctx);
+  }
+
+  /**
+   * 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]);
+    }
+  }
+
+  // For data validation in remote PE
+  if( get_flat_id() == 0 ) {
+    int *store_avail = (int*)(dest + get_flat_grid_size());
+    *store_avail = *available;
+    rocshmem_ctx_int_put(ctx, store_avail, store_avail, 1, 1);
+  }
+
+  __syncthreads();
+
+  if (t_id == 0) {
+    start_time[wg_id] = wf_start_time[0];
+  }
+
+  rocshmem_wg_ctx_destroy(&ctx);
   rocshmem_wg_finalize();
 }
 
@@ -50,17 +118,26 @@ __global__ void ShmemPtrTest(char *r_buf, int *available) {
  * HOST TESTER CLASS METHODS
  *****************************************************************************/
 ShmemPtrTester::ShmemPtrTester(TesterArguments args) : Tester(args) {
+  size_t buff_size = args.wg_size * args.num_wgs + sizeof(int);
   CHECK_HIP(hipMalloc((void **)&_available, sizeof(int)));
-  r_buf = (char *)rocshmem_malloc(args.max_msg_size);
+  dest = (char *)rocshmem_malloc(buff_size);
+
+  if (dest == nullptr) {
+    std::cerr << "Error allocating memory from symmetric heap" << std::endl;
+    std::cerr << "dest: " << dest << std::endl;
+
+    rocshmem_global_exit(1);
+  }
 }
 
 ShmemPtrTester::~ShmemPtrTester() {
   CHECK_HIP(hipFree(_available));
-  rocshmem_free(r_buf);
+  rocshmem_free(dest);
 }
 
 void ShmemPtrTester::resetBuffers(size_t size) {
-  memset(r_buf, '0', args.max_msg_size);
+  size_t buff_size = args.wg_size * args.num_wgs + sizeof(int);
+  memset(dest, '0', buff_size);
   memset(_available, 0, sizeof(int));
 }
 
@@ -68,22 +145,32 @@ void ShmemPtrTester::launchKernel(dim3 gridSize, dim3 blockSize, int loop,
                                   size_t size) {
   size_t shared_bytes = 0;
 
-  hipLaunchKernelGGL(ShmemPtrTest, gridSize, blockSize, shared_bytes, stream,
-                     r_buf, _available);
+  hipLaunchKernelGGL(ShmemPtrTest, gridSize, blockSize, shared_bytes,
+                     stream, loop, args.skip, start_time, end_time,
+                     dest, wf_size, _type, _available);
 
-  num_msgs = 0;
-  num_timed_msgs = 0;
+  num_msgs = (loop + args.skip) * gridSize.x * blockSize.x;
+  num_timed_msgs = loop * gridSize.x * blockSize.x;
 }
 
 void ShmemPtrTester::verifyResults(size_t size) {
   if (args.myid == 0) {
     if (*_available == 0) {
-      fprintf(stderr, "SHMEM_PTR NOT AVAILBLE \n");
+      _print_results = false;
+      std::cout << "rocshmem ptr not available\n" << std::endl;
     }
-  } else {
-    if (r_buf[4] != '1') {
-      fprintf(stderr, "Data validation error \n");
-      fprintf(stderr, "Got %c, Expected %c\n", r_buf[4], '1');
+  }
+  else {
+    size_t buff_size = args.wg_size * args.num_wgs;
+    int *available = (int*)(dest + buff_size);
+    if(*available == 1) {
+      for (size_t i = 0; i < buff_size; i++) {
+        if (dest[i] != '1') {
+          std::cerr << "Data validation error at idx " << i << std::endl;
+          std::cerr << " Got " << dest[i] << ", Expected 1 " << std::endl;
+          exit(-1);
+        }
+      }
     }
   }
 }