diff --git a/projects/rccl/src/bootstrap.cc b/projects/rccl/src/bootstrap.cc index a51ce790cd3f..e539af1cc568 100644 --- a/projects/rccl/src/bootstrap.cc +++ b/projects/rccl/src/bootstrap.cc @@ -1609,10 +1609,12 @@ static ncclResult_t socketRingAllGather(struct ncclSocket* nextSock, struct nccl BOOTSTRAP_PROF_OPEN(tFirst); for (int step = 0; step < totalSteps; step++) { bool isFinalUnidirectional = (step == totalSteps - 1) && (nranks % 2 == 0); - int sendSliceRing0 = (rank - step + nranks) % nranks; - int recvSliceRing0 = (rank - step - 1 + nranks) % nranks; - int sendSliceRing1 = (rank + step) % nranks; - int recvSliceRing1 = (rank + step + 1) % nranks; + // Use size_t slice indices so that `slice * size` stays 64-bit. An int multiply overflows + // with a large payload (the ~100 MB XML at 64 ranks), giving a bad pointer / EFAULT "Bad address". + size_t sendSliceRing0 = (rank - step + nranks) % nranks; + size_t recvSliceRing0 = (rank - step - 1 + nranks) % nranks; + size_t sendSliceRing1 = (rank + step) % nranks; + size_t recvSliceRing1 = (rank + step + 1) % nranks; if (isFinalUnidirectional) { NCCLCHECKGOTO(socketSendRecv(nextSock, data + sendSliceRing0 * size, size, prevSock, data + recvSliceRing0 * size, size), res, exit); } else { @@ -1680,10 +1682,12 @@ static ncclResult_t netBiDirRingAllGather(ncclNet_t* net, BOOTSTRAP_PROF_OPEN(tFirst); for (int step = 0; step < totalSteps; step++) { bool isFinalUnidirectional = (step == totalSteps - 1) && (nranks % 2 == 0); - int sendSliceRing0 = (rank - step + nranks) % nranks; // Ring0 send to next - int recvSliceRing0 = (rank - step - 1 + nranks) % nranks; // Ring0 recv from prev - int sendSliceRing1 = (rank + step) % nranks; // Ring1 send to prev - int recvSliceRing1 = (rank + step + 1) % nranks; // Ring1 recv from next + // Use size_t slice indices so that `slice * size` is computed in 64-bit, for the + // same overflow reason described in socketRingAllGather. + size_t sendSliceRing0 = (rank - step + nranks) % nranks; // Ring0 send to next + size_t recvSliceRing0 = (rank - step - 1 + nranks) % nranks; // Ring0 recv from prev + size_t sendSliceRing1 = (rank + step) % nranks; // Ring1 send to prev + size_t recvSliceRing1 = (rank + step + 1) % nranks; // Ring1 recv from next if (isFinalUnidirectional) { // Final step on even N: only ring0 over the forward QP pair. res = netSendRecv(net, sendComm, data + sendSliceRing0 * size, size, sendH, diff --git a/projects/rccl/src/graph/connect.cc b/projects/rccl/src/graph/connect.cc index af05908bec64..fa047225fb22 100644 --- a/projects/rccl/src/graph/connect.cc +++ b/projects/rccl/src/graph/connect.cc @@ -1214,7 +1214,9 @@ ncclResult_t ncclTopoPostset(struct ncclComm* comm, int* firstRanks, int* treePa // If user didn't override, use requested channels; otherwise keep capped max. if (!userUpdatedMaxChannels) { maxNchannels = nc * comm->nChannels * channelMultiplier; - nc = singleNode? maxNchannels : std::min(maxNchannels, maxChannels); + // Cap the single-node count at MAXCHANNELS/2, the maximum WarpSpeed supports. Left unbounded it + // overruns the fixed comm->channels[MAXCHANNELS] array (for example nc reaches 1760 in CPX mode at 64 ranks). + nc = singleNode? std::min(maxNchannels, MAXCHANNELS/2) : std::min(maxNchannels, maxChannels); } else { nc = maxNchannels = std::min(adjustedMaxNchannels * channelMultiplier, MAXCHANNELS); } diff --git a/projects/rccl/test/WarpSpeedMPITests.cpp b/projects/rccl/test/WarpSpeedMPITests.cpp index 2d3667ce371b..67435a0dbc04 100644 --- a/projects/rccl/test/WarpSpeedMPITests.cpp +++ b/projects/rccl/test/WarpSpeedMPITests.cpp @@ -228,4 +228,53 @@ TEST_F(WarpSpeedMPITest, MixedThresholdRace) TEST_INFO("All %d iterations passed", kTrainIterations); } +// High-rank single-node comm init + AllGather; exercises the bootstrap topology +// AllGather and single-node WarpSpeed channel setup (a regression aborts init). +TEST_F(WarpSpeedMPITest, HighRankCommInitAndAllGather) +{ + // Single node, all available ranks (repro uses 64 on CPX+NPS2). + ASSERT_TRUE(validateTestPrerequisites(kMinProcessesForMPI, + kNoProcessLimit, + kNoPowerOfTwoRequired, + kRequireSingleNode, + kRequireSingleNode)) + << "Requires a single-node run"; + + // Init exercises the bootstrap topology AllGather and WarpSpeed channel setup. + ASSERT_EQ(ncclSuccess, createTestCommunicator()); + skipUnlessWarpSpeedAutoEnabled(); + HIP_TEST_CHECK_GTEST_FAIL(hipStreamCreate(&stream_)); + + const int nranks = MPIEnvironment::world_size; + const int rank = MPIEnvironment::world_rank; + const size_t sendCount = 1u << 20; // 1M floats (4 MB) per rank + const size_t recvCount = sendCount * static_cast(nranks); + + void* d_send = nullptr; + void* d_recv = nullptr; + HIP_TEST_CHECK_GTEST_FAIL(hipMalloc(&d_send, sendCount * sizeof(float))); + HIP_TEST_CHECK_GTEST_FAIL(hipMalloc(&d_recv, recvCount * sizeof(float))); + // Free on any exit path (including early ASSERT/FAIL returns) to avoid leaks. + SCOPE_EXIT(if(d_send) { (void)hipFree(d_send); }); + SCOPE_EXIT(if(d_recv) { (void)hipFree(d_recv); }); + + // Each rank fills its send buffer with its own rank value. + HIP_TEST_CHECK_GTEST_FAIL(initializeBufferWithPattern( + d_send, sendCount, [rank](size_t) { return static_cast(rank); })); + + ncclComm_t comm = getActiveCommunicator(); + ASSERT_MPI_EQ(ncclSuccess, + ncclAllGather(d_send, d_recv, sendCount, ncclFloat, comm, stream_)); + HIP_TEST_CHECK_GTEST_FAIL(hipStreamSynchronize(stream_)); + + // After the AllGather, rank i's chunk (indices [i*sendCount, (i+1)*sendCount)) + // must contain the value i. + const bool ok = verifyBufferData( + d_recv, recvCount, + [sendCount](size_t idx) { return static_cast(idx / sendCount); }); + EXPECT_TRUE(ok) << "AllGather result mismatch across " << nranks << " ranks"; + + TEST_INFO("HighRankCommInitAndAllGather passed on %d ranks (single node)", nranks); +} + #endif // ENABLE_WARP_SPEED