Fix minor issues found during testing

Author: Marzieh Berenjkoub

src/ce_coll.cc

@@ -110,7 +110,7 @@ bool ncclCeImplemented(ncclFunc_t coll, int/*ncclDevRedOp_t*/ red, ncclDataType_
   return false;
 }
 
-ncclResult_t ncclPrepMCSync(struct ncclComm* comm, bool isComplete, CUstreamBatchMemOpParams* batchParams, size_t* opIdx, cudaStream_t stream) {
+ncclResult_t ncclPrepMCSync(struct ncclComm* comm, bool isComplete, hipStreamBatchMemOpParams* batchParams, size_t* opIdx, cudaStream_t stream) {
   ncclResult_t ret = ncclSuccess;
 
   uint32_t* readyPtrs    = (uint32_t*)comm->ceColl.baseUCSymReadyPtr;
@@ -142,7 +142,7 @@ ncclResult_t ncclPrepMCSync(struct ncclComm* comm, bool isComplete, CUstreamBatc
   for (int r = 0; r < comm->nRanks; ++r) {
     if (r == comm->rank) continue;
     batchParams[*opIdx] = {};
-    batchParams[*opIdx].waitValue.operation = CU_STREAM_MEM_OP_WAIT_VALUE_32;
+    //batchParams[*opIdx].waitValue.operation = HIP_STREAM_MEM_OP_WAIT_VALUE_32;
     batchParams[*opIdx].waitValue.address = (CUdeviceptr)(isComplete ? (void*)&completePtrs[r] : (void*)&readyPtrs[r]);
     batchParams[*opIdx].waitValue.value = waitValue;
     batchParams[*opIdx].waitValue.flags = CU_STREAM_WAIT_VALUE_EQ;
@@ -156,7 +156,7 @@ ncclResult_t ncclPrepMCSync(struct ncclComm* comm, bool isComplete, CUstreamBatc
 }
 
 ncclResult_t ncclPrepUCSync(struct ncclComm* comm, bool isComplete,
-                               CUstreamBatchMemOpParams* batchParams,
+                               hipStreamBatchMemOpParams* batchParams,
                                size_t* opIdx) {
   ncclResult_t ret = ncclSuccess;
 
@@ -175,7 +175,7 @@ ncclResult_t ncclPrepUCSync(struct ncclComm* comm, bool isComplete,
     size_t offset = (uint8_t*)dstPtr - (uint8_t*)comm->ceColl.ceSyncWin->userPtr;
     NCCLCHECKGOTO(ncclDevrGetLsaRankPtr(comm, comm->ceColl.ceSyncWin, offset, r, &peerDstPtr), ret, fail);
     batchParams[*opIdx] = {};
-    batchParams[*opIdx].writeValue.operation = CU_STREAM_MEM_OP_WRITE_VALUE_32;
+    batchParams[*opIdx].writeValue.operation = hipStreamMemOpWriteValue32;
     batchParams[*opIdx].writeValue.address  = (CUdeviceptr)peerDstPtr;
     batchParams[*opIdx].writeValue.value = waitValue;
     //batchParams[*opIdx].writeValue.flags = CU_STREAM_WRITE_VALUE_DEFAULT;
@@ -186,7 +186,7 @@ ncclResult_t ncclPrepUCSync(struct ncclComm* comm, bool isComplete,
   for (int r = 0; r < comm->nRanks; ++r) {
     if (r == comm->rank) continue;
     batchParams[*opIdx] = {};
-    batchParams[*opIdx].waitValue.operation = CU_STREAM_MEM_OP_WAIT_VALUE_32;
+    //batchParams[*opIdx].waitValue.operation = HIP_STREAM_MEM_OP_WAIT_VALUE_32;
     batchParams[*opIdx].waitValue.address  = (CUdeviceptr)(isComplete ? (void*)&completePtrs[r] : (void*)&readyPtrs[r]);
     batchParams[*opIdx].waitValue.value = waitValue;
     batchParams[*opIdx].waitValue.flags = CU_STREAM_WAIT_VALUE_EQ;
@@ -212,7 +212,7 @@ ncclResult_t ncclMemOpSync(struct ncclComm* comm, cudaStream_t stream) {
   size_t opIdx = 0;
 
   // Prepare batch memory operations for synchronization
-  CUstreamBatchMemOpParams* batchParams = nullptr;
+  hipStreamBatchMemOpParams* batchParams = nullptr;
   NCCLCHECKGOTO(ncclCalloc(&batchParams, batchSize), ret, fail);
 
   if (comm->nvlsSupport) {
@@ -225,7 +225,7 @@ ncclResult_t ncclMemOpSync(struct ncclComm* comm, cudaStream_t stream) {
   if (ncclCudaGraphValid(comm->planner.capturingGraph)) {
     for (int i = 0; i < comm->nRanks; i++) {
       batchParams[opIdx] = {};
-      batchParams[opIdx].writeValue.operation = CU_STREAM_MEM_OP_WRITE_VALUE_32;
+      batchParams[opIdx].writeValue.operation = hipStreamMemOpWriteValue32;
       batchParams[opIdx].writeValue.address = (CUdeviceptr)(comm->ceColl.useCompletePtr ? (void*)&completePtrs[i] : (void*)&readyPtrs[i]);
       batchParams[opIdx].writeValue.value = 0;
       //batchParams[opIdx].writeValue.flags = CU_STREAM_WRITE_VALUE_DEFAULT;
@@ -234,7 +234,7 @@ ncclResult_t ncclMemOpSync(struct ncclComm* comm, cudaStream_t stream) {
   }
 
   // Execute all memory operations in a single batch
-  CUDACHECKGOTO(cuStreamBatchMemOp(stream, opIdx, batchParams, 0), ret, fail);
+  CUDACHECKGOTO(hipStreamBatchMemOp(stream, opIdx, batchParams, 0), ret, fail);
 
   // Toggle the flag for next call
   comm->ceColl.useCompletePtr = !comm->ceColl.useCompletePtr;

src/dev_runtime.cc

@@ -70,7 +70,7 @@ ncclResult_t ncclDevrInitOnce(struct ncclComm* comm) {
   CUmemAllocationProp memProp = {};
   memProp.type = CU_MEM_ALLOCATION_TYPE_PINNED;
   memProp.location.type = CU_MEM_LOCATION_TYPE_DEVICE;
-  memProp.requestedHandleTypes = ncclCuMemHandleType;
+  memProp.requestedHandleType = ncclCuMemHandleType;
   memProp.location.id = comm->cudaDev;
   CUCHECKGOTO(cuMemGetAllocationGranularity(&devr->granularity, &memProp, CU_MEM_ALLOC_GRANULARITY_RECOMMENDED), ret, fail_lsaRankList);
 
@@ -745,7 +745,7 @@ ncclResult_t ncclDevrCommCreateInternal(
     CUmemAllocationProp memProp = {};
     memProp.type = CU_MEM_ALLOCATION_TYPE_PINNED;
     memProp.location.type = CU_MEM_LOCATION_TYPE_DEVICE;
-    memProp.requestedHandleTypes = ncclCuMemHandleType;
+    memProp.requestedHandleType = ncclCuMemHandleType;
     memProp.location.id = comm->cudaDev;
 
     CUCHECKGOTO(cuMemCreate(&memHandle, bufSizeTotal, &memProp, 0), ret, fail);

src/enqueue.cc

@@ -2682,7 +2682,6 @@ static ncclResult_t p2pTaskAppend(
   info->coll = coll;
   // Set capturing graph. Called here so that profiler can emit a group API event with this information
   NCCLCHECK(ncclPlannerSetCapturingGraph(comm, info));
-  printf("p2pTaskAppend isGraphCaptured: %d\n", ncclCudaGraphValid(planner->capturingGraph));
   bool isGraphCaptured = ncclCudaGraphValid(planner->capturingGraph);
   NCCLCHECK(ncclProfilerStartGroupApiEvent(info, isGraphCaptured));
   NCCLCHECK(ncclProfilerRecordGroupApiEventState(ncclProfilerGroupStartApiStop));
@@ -2759,7 +2758,6 @@ static ncclResult_t collTaskAppend(
   // Set capturing graph. Called here so that profiler can emit a group API event with this information
   NCCLCHECK(ncclPlannerSetCapturingGraph(comm, info));
   bool isGraphCaptured = ncclCudaGraphValid(planner->capturingGraph);
-  printf("collTaskAppend isGraphCaptured: %d\n", isGraphCaptured);
   NCCLCHECK(ncclProfilerStartGroupApiEvent(info, isGraphCaptured));
   NCCLCHECK(ncclProfilerRecordGroupApiEventState(ncclProfilerGroupStartApiStop));
   NCCLCHECK(ncclProfilerStartCollApiEvent(info, isGraphCaptured));
@@ -2790,7 +2788,6 @@ static ncclResult_t collTaskAppend(
   ncclTaskCollSorterInsert(&planner->collSorter, t, t->trafficBytes);
 
   ncclProfilerStopCollApiEvent();
-  printf("ceCollTaskAppend enddddddddd: %d\n", ncclCudaGraphValid(planner->capturingGraph));
   return ncclSuccess;
 }
 

src/misc/shmutils.cc

@@ -119,7 +119,6 @@ ncclResult_t ncclShmOpen(char* shmPath, size_t shmPathSize, size_t shmSize, void
 #if defined(HIP_HOST_UNCACHED_MEMORY)
     CUDACHECKGOTO(cudaHostRegister((void*)hptr, realShmSize, cudaHostRegisterPortable | cudaHostRegisterMapped | hipExtHostRegisterUncached), ret, fail);
 #else
-    cudaStreamCaptureMode mode = cudaStreamCaptureModeRelaxed;
     CUDACHECKGOTO(cudaThreadExchangeStreamCaptureMode(&mode), ret, fail);
     CUDACHECKGOTO(cudaHostRegister((void*)hptr, realShmSize, cudaHostRegisterPortable | cudaHostRegisterMapped), ret, fail);
 #endif

src/misc/utils.cc

@@ -316,3 +316,12 @@ ncclResult_t ncclBitsToString(uint32_t bits, uint32_t mask, const char* (*toStr)
 
   return ncclSuccess;
 }
+
+void get_aligned_ptr_and_size(const void *ptr, const size_t bufsize, void **aligned_ptr, size_t *aligned_size) {
+  if (!aligned_ptr || !aligned_size) return;
+  const size_t page_size = get_sc_page_size();
+  uintptr_t aligned_ptr_local = (uintptr_t)ptr & ~(page_size - 1);
+  size_t local_offset = (size_t)((uintptr_t)ptr - aligned_ptr_local);
+  *aligned_size = (bufsize + local_offset + page_size - 1) & ~(page_size - 1);
+  *aligned_ptr = (void *)aligned_ptr_local;
+}