diff --git a/projects/rccl/src/algorithms/dda/ipc/ipc_init.cu b/projects/rccl/src/algorithms/dda/ipc/ipc_init.cu index 7e580ad07f5c..4f983bc7dd41 100644 --- a/projects/rccl/src/algorithms/dda/ipc/ipc_init.cu +++ b/projects/rccl/src/algorithms/dda/ipc/ipc_init.cu @@ -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 @@ -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; } @@ -197,12 +205,56 @@ 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 and CE-scratch copies run on the user stream, which the + // hostStream/deviceStream syncs in commDestroySync do not cover. doneEvent + // is recorded after every kernel and CE launch, so waiting on it keeps + // in-flight work off unmapped peers. + cudaError_t syncErr = cudaEventSynchronize(comm->doneEvent); + 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(comm->ddaIpcBarrierState); + delete comm->ddaIpcBarrierState; comm->ddaIpcBarrierState = nullptr; } CUDACHECKIGNORE(cudaFree(comm->ddaPeerPtrsDev)); diff --git a/projects/rccl/src/algorithms/dda/ipc/ipc_init.h b/projects/rccl/src/algorithms/dda/ipc/ipc_init.h index e04d33eb023e..0b3ce0673472 100755 --- a/projects/rccl/src/algorithms/dda/ipc/ipc_init.h +++ b/projects/rccl/src/algorithms/dda/ipc/ipc_init.h @@ -14,4 +14,5 @@ struct ncclComm; ncclResult_t ncclDdaIpcCommInit(struct ncclComm* comm); +ncclResult_t ncclDdaIpcCommClosePeerMappings(struct ncclComm* comm); ncclResult_t ncclDdaIpcCommFini(struct ncclComm* comm); diff --git a/projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.cc b/projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.cc index 354e8d4614e3..e6fbc74e6697 100644 --- a/projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.cc +++ b/projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.cc @@ -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(i)] == nullptr) { continue; } - CUDACHECKIGNORE(cudaIpcCloseMemHandle(memPtrs_[static_cast(i)])); + cudaError_t err = cudaIpcCloseMemHandle(memPtrs_[static_cast(i)]); + if (err != cudaSuccess) { + WARN("ncclIpcMemHandler::closePeerMappings: cudaIpcCloseMemHandle(rank %d) failed (%s)", i, + cudaGetErrorString(err)); + result = ncclUnhandledCudaError; + } + memPtrs_[static_cast(i)] = nullptr; } + exchanged_ = false; + return result; } ncclResult_t ncclIpcMemHandler::addSelfDeviceMemPtr(void* deviceMemPtr) { diff --git a/projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.h b/projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.h index 3687c4b2af09..6c5c92bf4fb9 100644 --- a/projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.h +++ b/projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.h @@ -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: diff --git a/projects/rccl/src/ce_coll.cc b/projects/rccl/src/ce_coll.cc index a2cf0c89cd28..c8251ffbf959 100644 --- a/projects/rccl/src/ce_coll.cc +++ b/projects/rccl/src/ce_coll.cc @@ -854,6 +854,8 @@ ncclResult_t ncclCeLaunchBatchOps(struct ncclComm* comm, struct ncclCeBatchOpsPa int64_t chunkSizeParam = ncclParamCeChunkSize(); size_t ceChunkSize = chunkSizeParam > 0 ? (size_t)chunkSizeParam : 0; void* ceBatchHandle = NULL; + // Copy streams forked from stream; fail rejoins them so queued copies stay ordered before stream. + int forkedStreams = 0; #ifdef ENABLE_FAULT_INJECTION NCCLCHECK(ceFaultCheck(comm, CE_FAULT_LAUNCH_OP, "ncclCeLaunchBatchOps")); @@ -964,6 +966,7 @@ ncclResult_t ncclCeLaunchBatchOps(struct ncclComm* comm, struct ncclCeBatchOpsPa for (int s = 0; s < activeStreams; s++) { CUDACHECKGOTO(cudaEventRecord(comm->ceColl.copyEvents[s], stream), ret, fail); CUDACHECKGOTO(cudaStreamWaitEvent(comm->ceColl.copyStreams[s], comm->ceColl.copyEvents[s], 0), ret, fail); + forkedStreams = s + 1; } // Distribute copies round-robin across streams @@ -1001,6 +1004,10 @@ ncclResult_t ncclCeLaunchBatchOps(struct ncclComm* comm, struct ncclCeBatchOpsPa ncclProfilerStopCeBatchEvent(comm, ceBatchHandle, stream); return ret; fail: + for (int s = 0; s < forkedStreams; s++) { + CUDACHECKIGNORE(cudaEventRecord(comm->ceColl.copyEvents[s], comm->ceColl.copyStreams[s])); + CUDACHECKIGNORE(cudaStreamWaitEvent(stream, comm->ceColl.copyEvents[s], 0)); + } goto exit; } @@ -3894,6 +3901,18 @@ ncclResult_t ncclLaunchCeColl(struct ncclComm* comm, struct ncclKernelPlan* plan } exit: + // As ncclLaunchKernel does: CE copies may write peer DDA scratch, so teardown's + // doneEvent wait and the next launch on another stream must cover them too. Also + // on failure, since copies queued before the error may still be running. + if (comm->doneEvent != nullptr) { + cudaError_t recordErr = cudaEventRecord(comm->doneEvent, stream); + if (recordErr == cudaSuccess) { + comm->lastStreamTag = ncclStreamTag(stream); + } else if (ret == ncclSuccess) { + WARN("ncclLaunchCeColl: doneEvent record failed (%s)", cudaGetErrorString(recordErr)); + ret = ncclUnhandledCudaError; + } + } // Stop CE collective profiling - always attempt if started, even on error ncclProfilerStopCeCollEvent(comm, args, stream); return ret; diff --git a/projects/rccl/src/enqueue/enqueue.cc b/projects/rccl/src/enqueue/enqueue.cc index 1b3601080361..fcff85735880 100644 --- a/projects/rccl/src/enqueue/enqueue.cc +++ b/projects/rccl/src/enqueue/enqueue.cc @@ -2253,10 +2253,6 @@ static void persistentDestructor(void* plans_) { NCCL_PARAM(GraphStreamOrdering, "GRAPH_STREAM_ORDERING", NCCL_CONFIG_UNDEF_INT); -// Biased so the default stream (nullptr) tags as 1, leaving 0 free as the "no launch yet" -// sentinel. See ncclComm::lastStreamTag. -static inline uintptr_t ncclStreamTag(hipStream_t s) { return (uintptr_t)s + 1; } - namespace { enum ncclImplicitOrder { ncclImplicitOrderNone, diff --git a/projects/rccl/src/include/enqueue.h b/projects/rccl/src/include/enqueue.h index 7676e3306c6f..5961dfd90af1 100644 --- a/projects/rccl/src/include/enqueue.h +++ b/projects/rccl/src/include/enqueue.h @@ -49,6 +49,10 @@ ncclResult_t ncclLaunchKernelAfter_NoCuda(struct ncclComm* comm, struct ncclKern ncclResult_t ncclLaunchFinish(struct ncclComm* comm); ncclResult_t ncclValidateCollConfigLaunchCompletionEvents(struct ncclComm* comm); +// Biased so the default stream (nullptr) tags as 1, leaving 0 free as the "no launch yet" +// sentinel. See ncclComm::lastStreamTag. +static inline uintptr_t ncclStreamTag(cudaStream_t s) { return (uintptr_t)s + 1; } + // Addon backends launch onto the user's stream themselves instead of going through doLaunches, which is what // makes comm->cudaDev current and installs the cross-stream dependency. Bracket such a launch with these. // diff --git a/projects/rccl/src/init.cc b/projects/rccl/src/init.cc index c922352353af..3f8446e33b2d 100644 --- a/projects/rccl/src/init.cc +++ b/projects/rccl/src/init.cc @@ -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); if (COMPILER_ATOMIC_LOAD(comm->abortFlag, std::memory_order_acquire) == 0) { int* hostRanks; int hostRank = 0; @@ -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++) { diff --git a/projects/rccl/test/host/CMakeLists.txt b/projects/rccl/test/host/CMakeLists.txt index 3e7526ec640a..97c52272bb63 100644 --- a/projects/rccl/test/host/CMakeLists.txt +++ b/projects/rccl/test/host/CMakeLists.txt @@ -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 @@ -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 # --------------------------------------------------------------------------- diff --git a/projects/rccl/test/host/ce-coll-test.cc b/projects/rccl/test/host/ce-coll-test.cc index 5f6b28fa27b3..6e2b4b2c10ff 100644 --- a/projects/rccl/test/host/ce-coll-test.cc +++ b/projects/rccl/test/host/ce-coll-test.cc @@ -9,6 +9,7 @@ #include +#include #include #include #include @@ -149,6 +150,168 @@ TEST_F(CeCollMicrotest, LaunchAllReduceWithStaging_ReachesTheCollectiveWithoutRe EXPECT_TRUE(LogHas(log, "CE AllReduce: no valid chunk layout")) << log; } +// A single-rank CE AllGather that reaches the copy path, logging every stream operation. +class CeCollDoneEventMicrotest : public CeCollMicrotest { + protected: + static constexpr uintptr_t kCopyStreamBase = 0x5300; + static constexpr uintptr_t kCopyEventBase = 0x5400; + hipStream_t const launchStream_ = reinterpret_cast(0x5100); + hipEvent_t const doneEvent_ = reinterpret_cast(0x5200); + int lsaRankList_[1] = {0}; + uint32_t syncFlags_[2] = {}; + ncclDevrWindow syncWin_{}; + float send_ = 1.0f; + float recv_ = 0.0f; + + struct StreamOp { + enum Kind { kCopy, kSync, kRecord, kWait } kind; + hipStream_t stream; + hipEvent_t event; + }; + std::vector ops_; + int copies_ = 0; + int syncs_ = 0; + int failCopy_ = 0; // 1-based; 0 never fails + bool failSyncAfterCopy_ = false; + bool failDoneRecord_ = false; + + ScopedHook copyHook_{ + g_hipMemcpyAsync, [this](void*, const void*, size_t, hipMemcpyKind, hipStream_t stream) { + ops_.push_back({StreamOp::kCopy, stream, nullptr}); + return ++copies_ == failCopy_ ? hipErrorInvalidValue : hipSuccess; + }}; + ScopedHook syncHook_{ + g_hipStreamBatchMemOp, [this](hipStream_t stream, unsigned int, hipStreamBatchMemOpParams*, unsigned int) { + ops_.push_back({StreamOp::kSync, stream, nullptr}); + ++syncs_; + return failSyncAfterCopy_ && copies_ > 0 ? hipErrorInvalidValue : hipSuccess; + }}; + ScopedHook recordHook_{ + g_hipEventRecord, [this](hipEvent_t event, hipStream_t stream) { + ops_.push_back({StreamOp::kRecord, stream, event}); + return failDoneRecord_ && event == doneEvent_ ? hipErrorInvalidValue : hipSuccess; + }}; + ScopedHook waitHook_{ + g_hipStreamWaitEvent, [this](hipStream_t stream, hipEvent_t event, unsigned int) { + ops_.push_back({StreamOp::kWait, stream, event}); + return hipSuccess; + }}; + + void SetUp() override { + CeCollMicrotest::SetUp(); + streams_.stream = launchStream_; + comm_->planner.capturingGraph.graphId = ULLONG_MAX; // not capturing + comm_->doneEvent = doneEvent_; + comm_->nRanks = 1; + comm_->devrState.lsaSize = 1; + comm_->devrState.lsaSelf = 0; + comm_->devrState.lsaRankList = lsaRankList_; + syncWin_.userPtr = syncFlags_; + comm_->ceColl.baseUCSymReadyPtr = reinterpret_cast(&syncFlags_[0]); + comm_->ceColl.baseUCSymComplPtr = reinterpret_cast(&syncFlags_[1]); + comm_->ceColl.ceSyncWin = &syncWin_; + comm_->ceColl.intraBatchSyncFreq = CE_COLL_INTRA_BATCH_SYNC_FREQ; + comm_->ceColl.intraBatchSyncMsgThreshold = CE_COLL_INTRA_BATCH_SYNC_MSG_THRESHOLD; + args_.func = ncclFuncAllGather; + args_.nElts = 1; + args_.sendBuff = reinterpret_cast(&send_); + args_.recvBuff = reinterpret_cast(&recv_); + } + + int DoneRecords() const { + return static_cast(std::count_if(ops_.begin(), ops_.end(), [this](const StreamOp& op) { + return op.kind == StreamOp::kRecord && op.event == doneEvent_; + })); + } + // doneEvent must be the last operation on the launch stream, so a wait on it covers every copy. + void ExpectDoneEventRecordedLast() const { + ASSERT_FALSE(ops_.empty()); + EXPECT_EQ(StreamOp::kRecord, ops_.back().kind); + EXPECT_EQ(doneEvent_, ops_.back().event); + EXPECT_EQ(launchStream_, ops_.back().stream); + EXPECT_EQ(1, DoneRecords()); + } +}; + +TEST_F(CeCollDoneEventMicrotest, LaunchSucceeds_RecordsDoneEventAndStreamTagLast) { + std::string log; + + ASSERT_EQ(ncclSuccess, Launch(&log)) << log; + + EXPECT_GT(copies_, 0) << "the launch must reach the CE copy path"; + ExpectDoneEventRecordedLast(); + EXPECT_EQ(ncclStreamTag(launchStream_), comm_->lastStreamTag); +} + +TEST_F(CeCollDoneEventMicrotest, LaunchFailsAfterQueuingCopies_StillRecordsDoneEventLast) { + failSyncAfterCopy_ = true; // the completion barrier, after the copies are queued + std::string log; + + EXPECT_NE(ncclSuccess, Launch(&log)); + + EXPECT_GT(copies_, 0) << "the failure must come after the CE copies"; + ExpectDoneEventRecordedLast(); + EXPECT_EQ(ncclStreamTag(launchStream_), comm_->lastStreamTag); +} + +TEST_F(CeCollDoneEventMicrotest, LaunchRejectedBeforeAnyCopy_StillRecordsDoneEvent) { + args_.func = ncclFuncBroadcast; // rejected by the dispatch default arm + std::string log; + + EXPECT_EQ(ncclInvalidUsage, Launch(&log)); + + EXPECT_EQ(0, copies_); + ExpectDoneEventRecordedLast(); +} + +TEST_F(CeCollDoneEventMicrotest, DoneEventRecordFails_FailsTheLaunchAndKeepsTheStreamTag) { + failDoneRecord_ = true; + std::string log; + + EXPECT_EQ(ncclUnhandledCudaError, Launch(&log)); + + EXPECT_TRUE(LogHas(log, "doneEvent record failed")) << log; + EXPECT_EQ(0u, comm_->lastStreamTag); +} + +TEST_F(CeCollDoneEventMicrotest, MultiStreamCopyFails_RejoinsForkedCopyStreamsToTheLaunchStream) { + constexpr int kStreams = 2; + comm_->ceColl.nCopyStreams = kStreams; + for (int s = 0; s < kStreams; ++s) { + comm_->ceColl.copyStreams[s] = reinterpret_cast(kCopyStreamBase + s); + comm_->ceColl.copyEvents[s] = reinterpret_cast(kCopyEventBase + s); + } + uint8_t src[3] = {}, dst[3] = {}; + void* srcs[3] = {&src[0], &src[1], &src[2]}; + void* dsts[3] = {&dst[0], &dst[1], &dst[2]}; + size_t sizes[3] = {1, 1, 1}; + ncclCeBatchOpsParams params = {}; + params.srcs = srcs; + params.dsts = dsts; + params.sizes = sizes; + params.numOps = 3; + failCopy_ = 3; // copies 1 and 2 are already queued on both copy streams + + EXPECT_NE(ncclSuccess, ncclCeLaunchBatchOps(comm_.get(), ¶ms, launchStream_, &args_)); + + ASSERT_EQ(3, copies_) << "the multi-stream path must be taken"; + auto failed = std::find_if(ops_.rbegin(), ops_.rend(), [](const StreamOp& op) { + return op.kind == StreamOp::kCopy; + }).base(); + for (int s = 0; s < kStreams; ++s) { + hipStream_t const copyStream = comm_->ceColl.copyStreams[s]; + hipEvent_t const copyEvent = comm_->ceColl.copyEvents[s]; + auto rec = std::find_if(failed, ops_.end(), [&](const StreamOp& op) { + return op.kind == StreamOp::kRecord && op.stream == copyStream && op.event == copyEvent; + }); + ASSERT_NE(ops_.end(), rec) << "copy stream " << s << " is not joined after the failure"; + auto wait = std::find_if(rec, ops_.end(), [&](const StreamOp& op) { + return op.kind == StreamOp::kWait && op.stream == launchStream_ && op.event == copyEvent; + }); + EXPECT_NE(ops_.end(), wait) << "launch stream does not wait on copy stream " << s; + } +} + // The eager 2-shot path calls ncclCeAllReduce directly, bypassing ncclLaunchCeColl. TEST_F(CeCollMicrotest, CeAllReduceWithoutStaging_FailsLoudlyWithoutRegistering) { std::string log; diff --git a/projects/rccl/test/host/dda-ipc-lifecycle-test.cc b/projects/rccl/test/host/dda-ipc-lifecycle-test.cc new file mode 100644 index 000000000000..0ea37815dce5 --- /dev/null +++ b/projects/rccl/test/host/dda-ipc-lifecycle-test.cc @@ -0,0 +1,508 @@ +/************************************************************************* + * Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. + * + * Host-only regression tests for ordered DDA IPC teardown. + * + * See LICENSE.txt for license information. + ************************************************************************/ + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "ScopedHook.h" +#include "fakes/bootstrap_stubs.h" +#include "fakes/hip_fakes.h" + +#include "bootstrap.h" +#include "comm.h" + +#include DDA_IPC_MEM_HANDLER_CC_PATH +#include DDA_IPC_GPU_BARRIER_CU_PATH +#include DDA_IPC_INIT_CU_PATH + +// Owned by src/init.cc, which this binary does not compile. +struct allocationTracker allocTracker[MAX_ALLOC_TRACK_NGPU] = {}; + +namespace { + +constexpr int kRanks = nccl_dda_detail::kDdaNranks; +constexpr auto kRendezvousTimeout = std::chrono::seconds(3); + +int64_t g_ddaEnable = 1; + +class SimBootstrap { +public: + struct Handle { + SimBootstrap* sim; + int rank; + }; + + SimBootstrap() { + for (int rank = 0; rank < kRanks; ++rank) { + handles_[rank] = Handle{this, rank}; + } + } + + void* handle(int rank) { return &handles_[rank]; } + + ncclResult_t allGather(int rank, void* allData, int size) { + std::unique_lock lock(mu_); + GatherRound& round = gatherRounds_[nextGather_[rank]++]; + if (round.size < 0) { + round.size = size; + round.data.assign(static_cast(size) * kRanks, 0); + } else if (round.size != size) { + round.mismatch = true; + ++desyncs; + } + if (!round.mismatch) { + std::memcpy(round.data.data() + static_cast(rank) * size, + static_cast(allData) + static_cast(rank) * size, size); + } + ++round.arrived; + cv_.notify_all(); + if (!cv_.wait_for(lock, kRendezvousTimeout, [&] { return round.arrived == kRanks; })) { + ++timeouts; + return ncclRemoteError; + } + if (round.mismatch) { + return ncclInternalError; + } + std::memcpy(allData, round.data.data(), static_cast(size) * kRanks); + return ncclSuccess; + } + + ncclResult_t barrier(int rank, int nranks, int tag) { + std::unique_lock lock(mu_); + const int roundIndex = nextBarrier_[rank]++; + BarrierRound& round = barrierRounds_[roundIndex]; + if (round.nranks < 0) { + round.nranks = nranks; + round.tag = tag; + } else if (round.nranks != nranks || round.tag != tag) { + round.mismatch = true; + ++desyncs; + } + ++round.arrived; + ++barrierCalls; + cv_.notify_all(); + if (!cv_.wait_for(lock, kRendezvousTimeout, [&] { return round.arrived == kRanks; })) { + ++timeouts; + return ncclRemoteError; + } + if (round.mismatch) { + return ncclInternalError; + } + return ncclSuccess; + } + + int timeouts = 0; + int desyncs = 0; + int barrierCalls = 0; + +private: + struct GatherRound { + int size = -1; + int arrived = 0; + bool mismatch = false; + std::vector data; + }; + + struct BarrierRound { + int nranks = -1; + int tag = -1; + int arrived = 0; + bool mismatch = false; + }; + + std::mutex mu_; + std::condition_variable cv_; + std::map gatherRounds_; + std::map barrierRounds_; + int nextGather_[kRanks] = {}; + int nextBarrier_[kRanks] = {}; + Handle handles_[kRanks]; +}; + +class DdaIpcLifecycleMicrotest : public ::testing::Test { +protected: + void SetUp() override { + g_ddaEnable = 1; + for (int rank = 0; rank < kRanks; ++rank) { + peerInfo_[rank].rank = rank; + peerInfo_[rank].cudaDev = rank; + comms_[rank] = std::make_unique(); + ncclComm* comm = comms_[rank].get(); + comm->rank = rank; + comm->nRanks = kRanks; + comm->nNodes = 1; + comm->archName = archName_; + comm->peerInfo = peerInfo_; + comm->bootstrap = bootstrap_.handle(rank); + comm->abortFlag = &abortFlags_[rank]; + } + } + + void TearDown() override { + bool resourcesRemain = false; + for (const auto& comm : comms_) { + resourcesRemain |= comm->ddaIpcMemHandler != nullptr || comm->ddaIpcBarrierState != nullptr || + comm->ddaScratch != nullptr; + } + if (resourcesRemain) { + finalizeDdaIpcSerially(); + } + } + + void* allocate(size_t requested) { + const size_t bytes = requested == DDA_IPC_BUFFER_SIZE ? 64 : std::max(requested, 64); + void* ptr = std::malloc(bytes); + std::lock_guard lock(memoryMu_); + liveAllocations_.insert(ptr); + return ptr; + } + + void initializeDdaIpcForAllRanks() { + std::vector threads; + for (int rank = 0; rank < kRanks; ++rank) { + threads.emplace_back([this, rank] { initResults_[rank] = ncclDdaIpcCommInit(comms_[rank].get()); }); + } + for (auto& thread : threads) { + thread.join(); + } + } + + void closeDdaIpcPeerMappingsForAllRanks() { + std::vector threads; + for (int rank = 0; rank < kRanks; ++rank) { + threads.emplace_back( + [this, rank] { closeResults_[rank] = ncclDdaIpcCommClosePeerMappings(comms_[rank].get()); }); + } + for (auto& thread : threads) { + thread.join(); + } + } + + void finalizeDdaIpcSerially() { + for (int rank = 0; rank < kRanks; ++rank) { + finiResults_[rank] = ncclDdaIpcCommFini(comms_[rank].get()); + } + } + + int openMappingCount() { + std::lock_guard lock(memoryMu_); + int count = 0; + for (const auto& entry : openMappings_) { + count += entry.second; + } + return count; + } + + char archName_[8] = "gfx942"; + uint32_t abortFlags_[kRanks] = {}; + ncclPeerInfo peerInfo_[kRanks] = {}; + std::unique_ptr comms_[kRanks]; + ncclResult_t initResults_[kRanks] = {}; + ncclResult_t closeResults_[kRanks] = {}; + ncclResult_t finiResults_[kRanks] = {}; + SimBootstrap bootstrap_; + + std::mutex memoryMu_; + std::set liveAllocations_; + std::map openMappings_; + int earlyFrees_ = 0; + + ScopedHook bootstrapAllGatherHook_{ + g_bootstrapAllGather, [](void* state, void* allData, int size) { + auto* handle = static_cast(state); + return handle->sim->allGather(handle->rank, allData, size); + }}; + ScopedHook bootstrapBarrierHook_{ + g_bootstrapBarrier, [](void* state, int rank, int nranks, int tag) { + auto* handle = static_cast(state); + EXPECT_EQ(handle->rank, rank); + return handle->sim->barrier(rank, nranks, tag); + }}; + ScopedHook p2pHook_{g_hipDeviceCanAccessPeer, [](int* canAccess, int, int) { + *canAccess = 1; + return hipSuccess; + }}; + ScopedHook extMallocHook_{ + g_hipExtMallocWithFlags, [this](void** ptr, size_t size, unsigned) { + *ptr = allocate(size); + return *ptr == nullptr ? hipErrorOutOfMemory : hipSuccess; + }}; + ScopedHook mallocHook_{g_hipMalloc, [this](void** ptr, size_t size) { + *ptr = allocate(size); + return *ptr == nullptr ? hipErrorOutOfMemory : hipSuccess; + }}; + ScopedHook freeHook_{g_hipFree, [this](void* ptr) { + if (ptr == nullptr) { + return hipSuccess; + } + { + std::lock_guard lock(memoryMu_); + if (openMappings_[ptr] != 0) { + ++earlyFrees_; + } + EXPECT_EQ(1u, liveAllocations_.erase(ptr)); + } + std::free(ptr); + return hipSuccess; + }}; + ScopedHook memsetHook_{ + g_hipMemset, [](void*, int, size_t) { return hipSuccess; }}; + ScopedHook memcpyHook_{ + g_hipMemcpy, [](void* dst, const void* src, size_t bytes, hipMemcpyKind) { + std::memcpy(dst, src, bytes); + return hipSuccess; + }}; + ScopedHook exportHook_{ + g_hipIpcGetMemHandle, [](hipIpcMemHandle_t* handle, void* ptr) { + std::memset(handle, 0, sizeof(*handle)); + std::memcpy(handle, &ptr, sizeof(ptr)); + return hipSuccess; + }}; + ScopedHook openHook_{ + g_hipIpcOpenMemHandle, [this](void** ptr, hipIpcMemHandle_t handle, unsigned) { + std::memcpy(ptr, &handle, sizeof(*ptr)); + std::lock_guard lock(memoryMu_); + ++openMappings_[*ptr]; + return hipSuccess; + }}; + ScopedHook closeHook_{g_hipIpcCloseMemHandle, [this](void* ptr) { + std::lock_guard lock(memoryMu_); + const bool failThisClose = failCloseCalls_.count(++closeCalls_) != 0; + auto it = openMappings_.find(ptr); + EXPECT_NE(it, openMappings_.end()); + if (it != openMappings_.end()) { + EXPECT_GT(it->second, 0); + --it->second; + } + return failThisClose ? hipErrorInvalidValue : hipSuccess; + }}; + + int closeCalls_ = 0; + std::set failCloseCalls_; +}; + +TEST_F(DdaIpcLifecycleMicrotest, NormalDestroyClosesPeerMappingsBeforeSerialCommFree) { + initializeDdaIpcForAllRanks(); + for (int rank = 0; rank < kRanks; ++rank) { + ASSERT_EQ(ncclSuccess, initResults_[rank]) << "rank " << rank; + ASSERT_NE(nullptr, comms_[rank]->ddaIpcMemHandler) << "rank " << rank; + ASSERT_NE(nullptr, comms_[rank]->ddaIpcBarrierState) << "rank " << rank; + } + EXPECT_EQ(2 * kRanks * (kRanks - 1), openMappingCount()); + + // commDestroySync runs concurrently per local rank and closes mappings. + closeDdaIpcPeerMappingsForAllRanks(); + EXPECT_EQ(0, openMappingCount()); + for (int rank = 0; rank < kRanks; ++rank) { + EXPECT_EQ(ncclSuccess, closeResults_[rank]) << "rank " << rank; + } + + // commReclaim subsequently calls commFree serially for every local rank. + // This must neither block nor free an allocation that a peer still maps. + finalizeDdaIpcSerially(); + EXPECT_EQ(0, bootstrap_.timeouts); + EXPECT_EQ(0, bootstrap_.desyncs); + EXPECT_EQ(0, bootstrap_.barrierCalls); + EXPECT_EQ(0, earlyFrees_); + EXPECT_TRUE(liveAllocations_.empty()); + for (int rank = 0; rank < kRanks; ++rank) { + EXPECT_EQ(ncclSuccess, finiResults_[rank]) << "rank " << rank; + } +} + +TEST_F(DdaIpcLifecycleMicrotest, DisabledDdaKeepsScratchForCeScratchCollectives) { + // CE-scratch AllGather/AlltoAll select on comm->ddaScratch independently of + // RCCL_DDA_ENABLE, so disabling DDA dispatch must not skip IPC scratch setup. + g_ddaEnable = 0; + + initializeDdaIpcForAllRanks(); + for (int rank = 0; rank < kRanks; ++rank) { + ASSERT_EQ(ncclSuccess, initResults_[rank]) << "rank " << rank; + EXPECT_NE(nullptr, comms_[rank]->ddaScratch) << "rank " << rank; + EXPECT_GT(comms_[rank]->ddaScratchBytes, 0u) << "rank " << rank; + EXPECT_NE(nullptr, comms_[rank]->ddaPeerPtrsHost) << "rank " << rank; + EXPECT_NE(nullptr, comms_[rank]->ddaIpcMemHandler) << "rank " << rank; + } + closeDdaIpcPeerMappingsForAllRanks(); + finalizeDdaIpcSerially(); + + EXPECT_EQ(0, bootstrap_.timeouts); + EXPECT_EQ(0, bootstrap_.desyncs); + EXPECT_EQ(0, bootstrap_.barrierCalls); + EXPECT_EQ(0, earlyFrees_); + EXPECT_EQ(0, openMappingCount()); + EXPECT_TRUE(liveAllocations_.empty()); + for (int rank = 0; rank < kRanks; ++rank) { + EXPECT_EQ(ncclSuccess, initResults_[rank]) << "rank " << rank; + EXPECT_EQ(ncclSuccess, finiResults_[rank]) << "rank " << rank; + } +} + +TEST_F(DdaIpcLifecycleMicrotest, FinalizationWithoutIpcInitializationIsNonCollectiveNoOp) { + finalizeDdaIpcSerially(); + + EXPECT_EQ(0, bootstrap_.barrierCalls); + EXPECT_TRUE(liveAllocations_.empty()); + for (int rank = 0; rank < kRanks; ++rank) { + EXPECT_EQ(ncclSuccess, finiResults_[rank]) << "rank " << rank; + } +} + +TEST_F(DdaIpcLifecycleMicrotest, ClosePeerMappingsWaitsForInFlightDdaKernelsBeforeUnmapping) { + initializeDdaIpcForAllRanks(); + hipEvent_t doneEvents[kRanks]; + for (int rank = 0; rank < kRanks; ++rank) { + doneEvents[rank] = reinterpret_cast(static_cast(0x1000 + rank)); + comms_[rank]->doneEvent = doneEvents[rank]; + } + + std::vector syncedEvents; + std::vector closesAtSync; + ScopedHook syncHook{g_hipEventSynchronize, [&](hipEvent_t event) { + std::lock_guard lock(memoryMu_); + syncedEvents.push_back(event); + closesAtSync.push_back(closeCalls_); + // A failed wait must not stop the unmap and free path. + return syncedEvents.size() == 1 ? hipErrorLaunchFailure : hipSuccess; + }}; + + for (int rank = 0; rank < kRanks; ++rank) { + const int closesBefore = closeCalls_; + EXPECT_EQ(ncclSuccess, ncclDdaIpcCommClosePeerMappings(comms_[rank].get())) << "rank " << rank; + ASSERT_EQ(static_cast(rank + 1), syncedEvents.size()) << "rank " << rank; + EXPECT_EQ(doneEvents[rank], syncedEvents[rank]) << "rank " << rank; + EXPECT_EQ(closesBefore, closesAtSync[rank]) << "rank " << rank << " unmapped before waiting"; + EXPECT_GT(closeCalls_, closesBefore) << "rank " << rank; + } + EXPECT_EQ(0, openMappingCount()); + + finalizeDdaIpcSerially(); + EXPECT_EQ(0, earlyFrees_); + EXPECT_TRUE(liveAllocations_.empty()); +} + +TEST_F(DdaIpcLifecycleMicrotest, FiniWithoutPriorCloseStaysLocalAndClosesPeerMappings) { + initializeDdaIpcForAllRanks(); + abortFlags_[0] = 1; + for (int rank = 0; rank < kRanks; ++rank) { + comms_[rank]->doneEvent = reinterpret_cast(static_cast(0x1000 + rank)); + } + std::vector closesAtSync; + ScopedHook syncHook{g_hipEventSynchronize, [&](hipEvent_t) { + std::lock_guard lock(memoryMu_); + closesAtSync.push_back(closeCalls_); + return hipSuccess; + }}; + + // Direct commFree paths (e.g. init failure) have no earlier close stage. + // Fini must stay local, wait for in-flight launches before its handler + // destructors unmap, and still release every mapping and allocation. + finalizeDdaIpcSerially(); + + const int closesPerRank = 2 * (kRanks - 1); + ASSERT_EQ(static_cast(kRanks), closesAtSync.size()); + for (int rank = 0; rank < kRanks; ++rank) { + EXPECT_EQ(rank * closesPerRank, closesAtSync[rank]) << "rank " << rank << " unmapped before waiting"; + } + EXPECT_EQ(0, bootstrap_.timeouts); + EXPECT_EQ(0, bootstrap_.desyncs); + EXPECT_EQ(0, bootstrap_.barrierCalls); + EXPECT_EQ(0, openMappingCount()); + EXPECT_TRUE(liveAllocations_.empty()); + for (int rank = 0; rank < kRanks; ++rank) { + EXPECT_EQ(ncclSuccess, finiResults_[rank]) << "rank " << rank; + } +} + +TEST_F(DdaIpcLifecycleMicrotest, PeerCloseFailureIsBestEffortAndSerialCleanupCompletes) { + initializeDdaIpcForAllRanks(); + // Close rank by rank so the injected failures land deterministically: rank 0 + // closes its barrier imports (calls 1..kRanks-1) before its scratch imports. + failCloseCalls_ = {1, kRanks}; + + for (int rank = 0; rank < kRanks; ++rank) { + closeResults_[rank] = ncclDdaIpcCommClosePeerMappings(comms_[rank].get()); + } + finalizeDdaIpcSerially(); + + EXPECT_EQ(0, bootstrap_.timeouts); + EXPECT_EQ(0, bootstrap_.desyncs); + EXPECT_EQ(0, bootstrap_.barrierCalls); + EXPECT_EQ(2 * kRanks * (kRanks - 1), closeCalls_); + EXPECT_EQ(0, openMappingCount()); + EXPECT_TRUE(liveAllocations_.empty()); + for (int rank = 0; rank < kRanks; ++rank) { + EXPECT_EQ(ncclSuccess, closeResults_[rank]) << "rank " << rank; + EXPECT_EQ(ncclSuccess, finiResults_[rank]) << "rank " << rank; + } +} + +TEST(DdaIpcMemHandlerMicrotest, PartialOpenFailureClosesMappingsOpenedBeforeTheFailure) { + constexpr int nranks = 4; + constexpr int selfRank = 0; + void* const self = reinterpret_cast(0x1000); + int openCalls = 0; + int closeCalls = 0; + + ScopedHook allGather(g_bootstrapAllGather, [](void*, void* allData, int size) { + EXPECT_EQ(static_cast(sizeof(hipIpcMemHandle_t)), size); + auto* handles = static_cast(allData); + for (int rank = 0; rank < nranks; ++rank) { + void* ptr = reinterpret_cast(static_cast(0x1000 + rank * 0x100)); + std::memset(&handles[rank], 0, sizeof(handles[rank])); + std::memcpy(&handles[rank], &ptr, sizeof(ptr)); + } + return ncclSuccess; + }); + ScopedHook exportHandle(g_hipIpcGetMemHandle, [](hipIpcMemHandle_t* handle, void* ptr) { + std::memset(handle, 0, sizeof(*handle)); + std::memcpy(handle, &ptr, sizeof(ptr)); + return hipSuccess; + }); + ScopedHook openHandle(g_hipIpcOpenMemHandle, [&openCalls](void** ptr, hipIpcMemHandle_t handle, unsigned) { + ++openCalls; + if (openCalls == 2) { + *ptr = nullptr; + return hipErrorInvalidValue; + } + std::memcpy(ptr, &handle, sizeof(*ptr)); + return hipSuccess; + }); + ScopedHook closeHandle(g_hipIpcCloseMemHandle, [&closeCalls](void*) { + ++closeCalls; + return hipSuccess; + }); + + ncclIpcMemHandler handler(nullptr, selfRank, nranks); + ASSERT_EQ(ncclSuccess, handler.addSelfDeviceMemPtr(self)); + EXPECT_EQ(ncclUnhandledCudaError, handler.exchangeMemPtrs()); + EXPECT_EQ(2, openCalls); + EXPECT_EQ(0, closeCalls); + EXPECT_EQ(ncclSuccess, handler.closePeerMappings()); + EXPECT_EQ(1, closeCalls); + EXPECT_EQ(ncclSuccess, handler.closePeerMappings()); + EXPECT_EQ(1, closeCalls); +} + +} // namespace + +int64_t rcclParamDdaEnable() { + return g_ddaEnable; +} diff --git a/projects/rccl/test/host/fakes/bootstrap_stubs.cc b/projects/rccl/test/host/fakes/bootstrap_stubs.cc index 9d1fc0c31c38..c49b48102df7 100644 --- a/projects/rccl/test/host/fakes/bootstrap_stubs.cc +++ b/projects/rccl/test/host/fakes/bootstrap_stubs.cc @@ -27,6 +27,12 @@ ncclResult_t bootstrapAllGather(void* commState, void* allData, int size) { return g_bootstrapAllGather(commState, allData, size); } ASSERT_HOOK_MATCHES_PROD(g_bootstrapAllGather, bootstrapAllGather); +std::function g_bootstrapBarrier = + [](void*, int, int, int) { return ncclInternalError; }; +ncclResult_t bootstrapBarrier(void* commState, int rank, int nranks, int tag) { + return g_bootstrapBarrier(commState, rank, nranks, tag); +} +ASSERT_HOOK_MATCHES_PROD(g_bootstrapBarrier, bootstrapBarrier); // Only init.cc:2958 reaches this, to hand the rocSHMEM unique id to the other // ranks. Succeeds without touching bcastData; no test reaches it yet. ncclResult_t bootstrapBroadcast(void*, int, int, int, void*, int) { return ncclSuccess; } @@ -74,7 +80,13 @@ ncclResult_t bootstrapInit(int nHandles, void* handle, struct ncclComm* comm, st return g_bootstrapInit(nHandles, handle, comm, parent); } -ncclResult_t bootstrapIntraNodeBarrier(void*, int*, int, int, int) { return ncclSuccess; } +static ncclResult_t DefaultBootstrapIntraNodeBarrier(void*, int*, int, int, int) { return ncclSuccess; } +std::function g_bootstrapIntraNodeBarrier = + DefaultBootstrapIntraNodeBarrier; +ncclResult_t bootstrapIntraNodeBarrier(void* commState, int* ranks, int rank, int nranks, int tag) { + return g_bootstrapIntraNodeBarrier(commState, ranks, rank, nranks, tag); +} +ASSERT_HOOK_MATCHES_PROD(g_bootstrapIntraNodeBarrier, bootstrapIntraNodeBarrier); static ncclResult_t DefaultBootstrapSplit(uint64_t, struct ncclComm*, struct ncclComm*, int, int, int*) { ::abort(); } std::function g_bootstrapSplit = @@ -90,6 +102,8 @@ void ResetBootstrapStubs() { g_bootstrapCreateRoot = DefaultBootstrapCreateRoot; g_bootstrapNetInitFail = false; g_bootstrapAllGather = [](void*, void*, int) { return ncclInternalError; }; + g_bootstrapBarrier = [](void*, int, int, int) { return ncclInternalError; }; + g_bootstrapIntraNodeBarrier = DefaultBootstrapIntraNodeBarrier; g_bootstrapGetUniqueIdResult = ncclSuccess; g_bootstrapGetUniqueIdCalls = 0; g_bcastGrowHandleResult = ncclSuccess; diff --git a/projects/rccl/test/host/fakes/bootstrap_stubs.h b/projects/rccl/test/host/fakes/bootstrap_stubs.h index a6fc329151c2..47aa43cd2502 100644 --- a/projects/rccl/test/host/fakes/bootstrap_stubs.h +++ b/projects/rccl/test/host/fakes/bootstrap_stubs.h @@ -32,6 +32,9 @@ extern bool g_bootstrapNetInitFail; // A std::function, not a result code: tests must write the allgathered (color, key) table into allData. extern std::function g_bootstrapAllGather; +extern std::function g_bootstrapBarrier; +extern std::function + g_bootstrapIntraNodeBarrier; extern ncclResult_t g_bootstrapGetUniqueIdResult; extern int g_bootstrapGetUniqueIdCalls; diff --git a/projects/rccl/test/host/fakes/hip_fakes.cc b/projects/rccl/test/host/fakes/hip_fakes.cc index 16c9bbeb4f9d..ede38eda3ccc 100644 --- a/projects/rccl/test/host/fakes/hip_fakes.cc +++ b/projects/rccl/test/host/fakes/hip_fakes.cc @@ -602,6 +602,8 @@ static hipError_t DefaultHipStreamWaitEvent(hipStream_t, hipEvent_t, unsigned in std::function g_hipEventRecord = DefaultHipEventRecord; std::function g_hipStreamWaitEvent = DefaultHipStreamWaitEvent; +static hipError_t DefaultHipEventSynchronize(hipEvent_t) { return hipErrorInvalidValue; } +std::function g_hipEventSynchronize = DefaultHipEventSynchronize; static hipError_t DefaultHipStreamBatchMemOp(hipStream_t, unsigned int, hipStreamBatchMemOpParams*, unsigned int) @@ -709,6 +711,7 @@ void ResetHipFakes() g_hipLaunchKernel = DefaultHipLaunchKernel; g_hipEventRecord = DefaultHipEventRecord; g_hipStreamWaitEvent = DefaultHipStreamWaitEvent; + g_hipEventSynchronize = DefaultHipEventSynchronize; g_hipEventQueryRequiresRecord = false; g_recordedEvents.clear(); g_hipStreamBatchMemOp = DefaultHipStreamBatchMemOp; @@ -1071,7 +1074,7 @@ hipError_t hipStreamCreate(hipStream_t*) { return hipErrorInvalidValue; } hipError_t hipPointerGetAttributes(hipPointerAttribute_t*, const void*) { return hipErrorInvalidValue; } hipError_t hipHostGetDevicePointer(void**, void*, unsigned int) { return hipErrorInvalidValue; } hipError_t hipIpcGetEventHandle(hipIpcEventHandle_t*, hipEvent_t) { return hipErrorInvalidValue; } -hipError_t hipEventSynchronize(hipEvent_t) { return hipErrorInvalidValue; } +hipError_t hipEventSynchronize(hipEvent_t event) { return g_hipEventSynchronize(event); } // --- init.cc deep-path HIP stubs (commAlloc/devCommSetup) --------------- hipError_t hipRuntimeGetVersion(int* version) { return g_hipRuntimeGetVersion(version); } diff --git a/projects/rccl/test/host/fakes/hip_fakes.h b/projects/rccl/test/host/fakes/hip_fakes.h index 5b0be9d18827..e2177996ed75 100644 --- a/projects/rccl/test/host/fakes/hip_fakes.h +++ b/projects/rccl/test/host/fakes/hip_fakes.h @@ -244,6 +244,7 @@ extern std::function extern std::function g_hipStreamWaitEvent; +extern std::function g_hipEventSynchronize; extern std::function diff --git a/projects/rccl/test/host/fakes/nccl_stubs.cc b/projects/rccl/test/host/fakes/nccl_stubs.cc index f8a8ef0f7f95..55a3df1231d1 100644 --- a/projects/rccl/test/host/fakes/nccl_stubs.cc +++ b/projects/rccl/test/host/fakes/nccl_stubs.cc @@ -71,6 +71,12 @@ void ncclCudaContextDrop(struct ncclCudaContext* cxt) { ::abort(); } // ncclCudaContextTrack lives in strongstream_stubs.cc (v2.31 three-argument ABI). ncclResult_t ncclDdaFabricCommFini(struct ncclComm* comm) { return ncclSuccess; } ncclResult_t ncclDdaFabricCommInit(struct ncclComm* comm) { ::abort(); } +static ncclResult_t DefaultNcclDdaIpcCommClosePeerMappings(struct ncclComm*) { return ncclSuccess; } +std::function g_ncclDdaIpcCommClosePeerMappings = + DefaultNcclDdaIpcCommClosePeerMappings; +ncclResult_t ncclDdaIpcCommClosePeerMappings(struct ncclComm* comm) { + return g_ncclDdaIpcCommClosePeerMappings(comm); +} ncclResult_t ncclDdaIpcCommFini(struct ncclComm* comm) { return ncclSuccess; } ncclResult_t ncclDdaIpcCommInit(struct ncclComm* comm) { ::abort(); } bool ncclDdaUseFabricPath(struct ncclComm* comm) { return false; } @@ -280,6 +286,7 @@ void ResetNcclStubs() { g_ncclCommDestroy = DefaultNcclCommDestroy; g_ncclCommSplit = DefaultNcclCommSplit; g_collTraceDestroy = DefaultCollTraceDestroy; + g_ncclDdaIpcCommClosePeerMappings = DefaultNcclDdaIpcCommClosePeerMappings; g_ncclProfilerThreadDestroy = DefaultNcclProfilerThreadDestroy; g_ncclProfilerPluginFinalize = DefaultNcclProfilerPluginFinalize; g_ncclTunerPluginUnload = DefaultNcclTunerPluginUnload; diff --git a/projects/rccl/test/host/fakes/nccl_stubs.h b/projects/rccl/test/host/fakes/nccl_stubs.h index 3ad42a507e9e..64831ca93759 100644 --- a/projects/rccl/test/host/fakes/nccl_stubs.h +++ b/projects/rccl/test/host/fakes/nccl_stubs.h @@ -34,6 +34,7 @@ extern std::function g_collTraceDestroy; +extern std::function g_ncclDdaIpcCommClosePeerMappings; extern std::function g_ncclProfilerThreadDestroy; extern std::function g_ncclProfilerPluginFinalize; diff --git a/projects/rccl/test/host/init-test.cc b/projects/rccl/test/host/init-test.cc index 408234685007..4a788c692afe 100644 --- a/projects/rccl/test/host/init-test.cc +++ b/projects/rccl/test/host/init-test.cc @@ -6152,6 +6152,40 @@ TEST_F(InitMicrotest, CommDestroySync_InitStateSuccess_SetsDeviceStopsProxyAndDe EXPECT_EQ(1, proxyStop.calls); } +TEST_F(InitMicrotest, CommDestroySync_ClosesDdaIpcPeerMappingsBeforeProxyStop) { + Teardown_LifecycleComm c; + Teardown_AllowStreamCaptureExchange(); + std::vector callOrder; + ScopedHook collTrace(g_collTraceDestroy, [](ncclComm*) { return ncclSuccess; }); + ScopedHook closeMappings(g_ncclDdaIpcCommClosePeerMappings, [&](ncclComm*) { + callOrder.push_back("closeMappings"); + return ncclSuccess; + }); + ScopedHook barrier(g_bootstrapIntraNodeBarrier, [&](void*, int*, int, int, int) { + callOrder.push_back("intraNodeBarrier"); + return ncclSuccess; + }); + ScopedHook proxyStop(g_ncclProxyStop, [&](ncclComm*) { + callOrder.push_back("proxyStop"); + return ncclSuccess; + }); + + EXPECT_EQ(ncclSuccess, Teardown_RunDestroySync(c.get())); + EXPECT_EQ((std::vector{"closeMappings", "intraNodeBarrier", "proxyStop"}), callOrder); +} + +TEST_F(InitMicrotest, CommDestroySync_AbortedCommStillClosesDdaIpcPeerMappingsLocally) { + Teardown_LifecycleComm c; + *c.get()->abortFlag = 1; + Teardown_AllowStreamCaptureExchange(); + ScopedHook collTrace(g_collTraceDestroy, [](ncclComm*) { return ncclSuccess; }); + ScopedHook proxyStop(g_ncclProxyStop, [](ncclComm*) { return ncclSuccess; }); + ScopedHook closeMappings(g_ncclDdaIpcCommClosePeerMappings, [](ncclComm*) { return ncclSuccess; }); + + EXPECT_EQ(ncclSuccess, Teardown_RunDestroySync(c.get())); + EXPECT_EQ(1, closeMappings.calls); +} + TEST_F(InitMicrotest, CommDestroySync_SetDeviceFails_ReturnsErrorWithoutDestroyingCollTrace) { Teardown_LifecycleComm c; ScopedHook setDevice(g_hipSetDevice, [](int) { return hipErrorInvalidDevice; }); @@ -6276,6 +6310,9 @@ constexpr uint64_t Teardown_kTunerMagic = 0xC0FFEE01u; struct Teardown_TunerRecord { uint64_t magic = Teardown_kTunerMagic; int finalizeCalls = 0; + std::atomic* ddaCloseCalls = nullptr; + int expectedDdaCloseCalls = 0; + bool allDdaMappingsClosedBeforeCleanup = false; }; // Returning an error unless the magic survives is what makes "passed the wrong context" a failing test. @@ -6283,6 +6320,10 @@ ncclResult_t Teardown_TunerFinalize(void* context) { auto* rec = static_cast(context); if (rec == nullptr || rec->magic != Teardown_kTunerMagic) return ncclInternalError; rec->finalizeCalls++; + if (rec->ddaCloseCalls != nullptr) { + rec->allDdaMappingsClosedBeforeCleanup = + rec->ddaCloseCalls->load(std::memory_order_acquire) == rec->expectedDdaCloseCalls; + } return ncclSuccess; } @@ -6369,6 +6410,59 @@ TEST_F(InitMicrotest, CommReclaim_LastIntraRank_SyncsAndCleansEveryCommInTheChai } } +TEST_F(InitMicrotest, CommReclaim_ClosesAllDdaMappingsBeforeSerialCommCleanup) { + Teardown_AllowStreamCaptureExchange(); + constexpr int kChainLength = 3; + Teardown_ChainMember members[kChainLength]; + ASSERT_NO_FATAL_FAILURE(BuildCommChain(members, kChainLength)); + members[0].comm->finalizeRankCnt = kChainLength - 1; + + std::atomic ddaCloseCalls{0}; + for (auto& member : members) { + member.rec.ddaCloseCalls = &ddaCloseCalls; + member.rec.expectedDdaCloseCalls = kChainLength; + } + ScopedHook closeMappings(g_ncclDdaIpcCommClosePeerMappings, [&](ncclComm*) { + ddaCloseCalls.fetch_add(1, std::memory_order_release); + return ncclSuccess; + }); + ScopedHook proxyStop(g_ncclProxyStop, [](ncclComm*) { return ncclSuccess; }); + + EXPECT_EQ(ncclSuccess, Teardown_RunReclaim(members[kChainLength - 1].comm)); + EXPECT_EQ(kChainLength, closeMappings.calls); + for (int i = 0; i < kChainLength; ++i) { + EXPECT_TRUE(members[i].rec.allDdaMappingsClosedBeforeCleanup) << "chain member " << i; + } +} + +TEST_F(InitMicrotest, CommReclaim_RevokedChain_ClosesAllDdaMappingsBeforeSerialCommCleanup) { + constexpr int kChainLength = 3; + Teardown_ChainMember members[kChainLength]; + ASSERT_NO_FATAL_FAILURE(BuildCommChain(members, kChainLength)); + members[0].comm->finalizeRankCnt = kChainLength - 1; + + std::atomic ddaCloseCalls{0}; + for (auto& member : members) { + member.comm->finalizeCalled = true; + member.comm->revokedFlag = 1; + member.rec.ddaCloseCalls = &ddaCloseCalls; + member.rec.expectedDdaCloseCalls = kChainLength; + } + ScopedHook closeMappings(g_ncclDdaIpcCommClosePeerMappings, [&](ncclComm*) { + ddaCloseCalls.fetch_add(1, std::memory_order_release); + return ncclSuccess; + }); + ScopedHook proxyStop(g_ncclProxyStop, [](ncclComm*) { return ncclSuccess; }); + + EXPECT_EQ(ncclSuccess, Teardown_RunReclaim(members[kChainLength - 1].comm)); + EXPECT_EQ(kChainLength, closeMappings.calls); + EXPECT_EQ(0, proxyStop.calls); + for (int i = 0; i < kChainLength; ++i) { + EXPECT_TRUE(members[i].rec.allDdaMappingsClosedBeforeCleanup) << "chain member " << i; + EXPECT_EQ(1, members[i].rec.finalizeCalls) << "chain member " << i; + } +} + TEST_F(InitMicrotest, CommReclaim_DestroySyncFails_WarnsAndStillCleansTheChain) { const int kChainLength = 2; Teardown_ChainMember members[kChainLength]; diff --git a/projects/rccl/test/test_categories_micro_dda_ipc_lifecycle.yaml b/projects/rccl/test/test_categories_micro_dda_ipc_lifecycle.yaml new file mode 100644 index 000000000000..44529198df2f --- /dev/null +++ b/projects/rccl/test/test_categories_micro_dda_ipc_lifecycle.yaml @@ -0,0 +1,44 @@ +# CTest categories for host-only DDA IPC lifecycle microtests. +test_categories: + quick: µ_dda_ipc_lifecycle_quick + description: "Smoke - host-only DDA IPC lifecycle tests." + test_patterns: + - "DdaIpcLifecycleMicrotest.*" + - "DdaIpcMemHandlerMicrotest.*" + exclude_windows: + - "*" + labels: + - "rccl" + - "quick" + - "smoke" + - "precheck" + - "microtest" + standard: + <<: *micro_dda_ipc_lifecycle_quick + labels: + - "rccl" + - "standard" + - "regression" + - "microtest" + comprehensive: + <<: *micro_dda_ipc_lifecycle_quick + labels: + - "rccl" + - "comprehensive" + - "comprehensive_regression" + - "microtest" + full: + <<: *micro_dda_ipc_lifecycle_quick + labels: + - "rccl" + - "full" + - "full_regression" + - "microtest" + +execution_settings: + timeout_multiplier: 1 + category_timeouts: + quick: 600 + standard: 600 + comprehensive: 600 + full: 600