Skip to content
Merged
Show file tree
Hide file tree
Changes from 13 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 58 additions & 7 deletions projects/rccl/src/algorithms/dda/ipc/ipc_init.cu
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,21 @@ using nccl_dda_detail::kDdaNranks;
} \
} while (0)

ncclResult_t ncclDdaIpcCommInit(ncclComm* comm) {
namespace {

bool ddaIpcCommEligible(const ncclComm* comm) {
if (comm == nullptr) {
return ncclSuccess;
return false;
}
const bool ddaArchSupported =
comm->archName != nullptr && (IsArchMatch(comm->archName, "gfx942") || IsArchMatch(comm->archName, "gfx950"));
return comm->nRanks == kDdaNranks && comm->nNodes == 1 && comm->bootstrap != nullptr && !comm->directMode &&
!comm->MNNVL && ddaArchSupported;
}

} // namespace

ncclResult_t ncclDdaIpcCommInit(ncclComm* comm) {
// Skip DDA if:
// - nRanks is not exactly kDdaNranks (currently hardcoded to 8)
// - multi-node runs
Expand All @@ -44,10 +55,7 @@ ncclResult_t ncclDdaIpcCommInit(ncclComm* comm) {
// scratch/barrier here is not necessary. On gfx12xx (RDNA4) the
// uncached-memory IPC export fails (hipIpcGetMemHandle -> hipErrorInvalidValue),
// which aborts comm init entirely. Gate init to match dispatch.
const bool ddaArchSupported =
comm->archName != nullptr && (IsArchMatch(comm->archName, "gfx942") || IsArchMatch(comm->archName, "gfx950"));
if (comm->nRanks != kDdaNranks || comm->nNodes != 1 || comm->bootstrap == nullptr || comm->directMode ||
comm->MNNVL || !ddaArchSupported) {
if (!ddaIpcCommEligible(comm)) {
return ncclSuccess;
}

Expand Down Expand Up @@ -197,12 +205,55 @@ ncclResult_t ncclDdaIpcCommInit(ncclComm* comm) {
return ncclSuccess;
}

ncclResult_t ncclDdaIpcCommClosePeerMappings(ncclComm* comm) {
if (comm == nullptr) {
return ncclSuccess;
}

auto* barrierState = comm->ddaIpcBarrierState;
const bool hasBarrierMappings = barrierState != nullptr && barrierState->resources != nullptr &&
barrierState->resources->ipcMemHandler != nullptr;
if ((hasBarrierMappings || comm->ddaIpcMemHandler != nullptr) && comm->doneEvent != nullptr) {
// DDA kernels run on the user stream, which the hostStream/deviceStream
// syncs in commDestroySync do not cover. doneEvent is recorded after
// every launch, so waiting on it keeps in-flight kernels off unmapped peers.
cudaError_t syncErr = cudaEventSynchronize(comm->doneEvent);
Comment thread
PJAvinash marked this conversation as resolved.
Resolved
if (syncErr != cudaSuccess) {
WARN("ncclDdaIpcCommClosePeerMappings: doneEvent sync failed (%s); continuing cleanup",
cudaGetErrorString(syncErr));
}
}
if (hasBarrierMappings) {
ncclResult_t closeResult = barrierState->resources->ipcMemHandler->closePeerMappings();
if (closeResult != ncclSuccess) {
WARN("ncclDdaIpcCommClosePeerMappings: barrier peer mapping close failed (%d); continuing cleanup",
closeResult);
}
}
if (comm->ddaIpcMemHandler != nullptr) {
ncclResult_t closeResult = comm->ddaIpcMemHandler->closePeerMappings();
if (closeResult != ncclSuccess) {
WARN("ncclDdaIpcCommClosePeerMappings: scratch peer mapping close failed (%d); continuing cleanup",
closeResult);
}
}
return ncclSuccess;
}

ncclResult_t ncclDdaIpcCommFini(ncclComm* comm) {
if (comm == nullptr) {
return ncclSuccess;
}

// commFree must remain non-collective. Destroy and abort close mappings in
// commDestroySync and revoke closes them in commReclaim, before the serial
// commFree loop; this idempotent call is the local backstop for init-failure
// and other direct commFree paths.
ncclResult_t closeResult = ncclSuccess;
NCCLCHECKIGNORE(ncclDdaIpcCommClosePeerMappings(comm), closeResult);

if (comm->ddaIpcBarrierState != nullptr) {
delete static_cast<DdaIpcBarrierState*>(comm->ddaIpcBarrierState);
delete comm->ddaIpcBarrierState;
comm->ddaIpcBarrierState = nullptr;
}
CUDACHECKIGNORE(cudaFree(comm->ddaPeerPtrsDev));
Expand Down
1 change: 1 addition & 0 deletions projects/rccl/src/algorithms/dda/ipc/ipc_init.h
Original file line number Diff line number Diff line change
Expand Up @@ -14,4 +14,5 @@
struct ncclComm;

ncclResult_t ncclDdaIpcCommInit(struct ncclComm* comm);
ncclResult_t ncclDdaIpcCommClosePeerMappings(struct ncclComm* comm);
ncclResult_t ncclDdaIpcCommFini(struct ncclComm* comm);
Comment thread
PJAvinash marked this conversation as resolved.
Resolved
22 changes: 17 additions & 5 deletions projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.cc
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,27 @@ ncclIpcMemHandler::ncclIpcMemHandler(void* bootstrap, int rank, int nranks)
exchanged_(false) {}

ncclIpcMemHandler::~ncclIpcMemHandler() {
if (!exchanged_) {
return;
}
(void)closePeerMappings();
}

ncclResult_t ncclIpcMemHandler::closePeerMappings() {
// Close every mapping that was opened, including a partial exchange that
// returned before exchanged_ could be set.
ncclResult_t result = ncclSuccess;
for (int i = 0; i < nranks_; ++i) {
if (i == rank_) {
if (i == rank_ || memPtrs_[static_cast<size_t>(i)] == nullptr) {
continue;
}
CUDACHECKIGNORE(cudaIpcCloseMemHandle(memPtrs_[static_cast<size_t>(i)]));
cudaError_t err = cudaIpcCloseMemHandle(memPtrs_[static_cast<size_t>(i)]);
if (err != cudaSuccess) {
WARN("ncclIpcMemHandler::closePeerMappings: cudaIpcCloseMemHandle(rank %d) failed (%s)", i,
cudaGetErrorString(err));
result = ncclUnhandledCudaError;
}
memPtrs_[static_cast<size_t>(i)] = nullptr;
}
exchanged_ = false;
return result;
}

ncclResult_t ncclIpcMemHandler::addSelfDeviceMemPtr(void* deviceMemPtr) {
Expand Down
1 change: 1 addition & 0 deletions projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.h
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ class ncclIpcMemHandler {

ncclResult_t addSelfDeviceMemPtr(void* deviceMemPtr);
ncclResult_t exchangeMemPtrs();
ncclResult_t closePeerMappings();
ncclResult_t getPeerDeviceMemPtr(int peerRank, void** outPeerPtr) const;

private:
Expand Down
11 changes: 11 additions & 0 deletions projects/rccl/src/init.cc
Original file line number Diff line number Diff line change
Expand Up @@ -4433,6 +4433,12 @@ static ncclResult_t commDestroySync(struct ncclAsyncJob* job_) {
comm->commHash, comm->rank);
}
}
// Close imported DDA IPC mappings while commDestroySync is still running
// concurrently for every local rank. The existing intranode barrier below
// then guarantees all mappings are closed before serial commFree releases
// any rank's exported allocations. Revoked comms close in commReclaim.
ncclResult_t ddaCloseRet = ncclSuccess;
NCCLCHECKIGNORE(ncclDdaIpcCommClosePeerMappings(comm), ddaCloseRet);
Comment thread
PJAvinash marked this conversation as resolved.
Resolved
if (COMPILER_ATOMIC_LOAD(comm->abortFlag, std::memory_order_acquire) == 0) {
int* hostRanks;
int hostRank = 0;
Expand Down Expand Up @@ -4629,6 +4635,11 @@ static ncclResult_t commReclaim(struct ncclAsyncJob* job_) {
cb);
}
}
// Revoke skips commDestroySync, so close this rank's DDA IPC imports
// here, before the serial commCleanup loop below frees any local rank's
// exported allocations.
ncclResult_t ddaCloseRet = ncclSuccess;
NCCLCHECKIGNORE(ncclDdaIpcCommClosePeerMappings(curIntraComm), ddaCloseRet);
}
}
for (int i = 0; i < nIntraThreads; i++) {
Expand Down
25 changes: 25 additions & 0 deletions projects/rccl/test/host/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -1059,6 +1059,19 @@ function(rccl_define_micro_source_lists)
../common/ProcessIsolatedTestRunner.cpp
../common/main_altrsmi.cpp
PARENT_SCOPE)

# rccl-UnitTestsMicroDdaIpcLifecycle -- src/algorithms/dda/ipc/{ipc_init.cu,ipc_mem_handler.cc,ipc_gpu_barrier.cu},
# included by the test TU so teardown ordering can be tested with host fakes.
set(TEST_MICRO_DDA_IPC_LIFECYCLE_SOURCE_FILES
dda-ipc-lifecycle-test.cc
fakes/bootstrap_stubs.cc
fakes/hip_fakes.cc
fakes/nccl_fakes.cc
fakes/utils_fakes.cc
${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/device/device_buffer.cc
${RCCL_HOST_HIPIFY_DIR}/src/misc/archinfo.cc
../common/main_altrsmi.cpp
PARENT_SCOPE)
endfunction()

# Create one host-only micro-test binary and apply all the boilerplate that
Expand Down Expand Up @@ -3044,6 +3057,18 @@ rccl_add_micro_binary(rccl-UnitTestsMicroHostDispatch
EXTRA_DEFS ${_MICRO_MODE_DEFS}
CATEGORIES_YAML "${CMAKE_CURRENT_SOURCE_DIR}/../test_categories_micro.yaml")

# ---------------------------------------------------------------------------
# rccl-UnitTestsMicroDdaIpcLifecycle -- host-only DDA IPC init/teardown lifecycle
# ---------------------------------------------------------------------------
rccl_add_micro_binary(rccl-UnitTestsMicroDdaIpcLifecycle
SOURCES ${TEST_MICRO_DDA_IPC_LIFECYCLE_SOURCE_FILES}
EXTRA_DEFS
${_MICRO_MODE_DEFS}
DDA_IPC_INIT_CU_PATH="${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/ipc/ipc_init.cu.cpp"
DDA_IPC_MEM_HANDLER_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/ipc/ipc_mem_handler.cc"
DDA_IPC_GPU_BARRIER_CU_PATH="${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/ipc/ipc_gpu_barrier.cu.cpp"
CATEGORIES_YAML "${CMAKE_CURRENT_SOURCE_DIR}/../test_categories_micro_dda_ipc_lifecycle.yaml")

# ---------------------------------------------------------------------------
# rccl-UnitTestsHostAll
# ---------------------------------------------------------------------------
Expand Down
Loading
Loading