Skip to content
Merged
Show file tree
Hide file tree
Changes from 7 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
1 change: 1 addition & 0 deletions projects/rccl/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https:
* `RCCL_CE_ALLREDUCE` now defaults to `-1` (auto/enabled) instead of `0` (disabled). CE AllReduce is therefore on by default for gfx1250 communicators that meet all other eligibility criteria.
* CE-2-Shot AllReduce size cap is now resolved at runtime via `rcclCeAr2ShotMax()` (env var wins, then arch table `ceNonRegMax[AR]`, then the 256 MiB `NCCL_CE_AR_TMPBUF_DEFAULT_BYTES` fallback) rather than from a compile-time constant.
* `RCCL_FORCE_CE_ALLREDUCE=1` no longer overrides the CE AllReduce 2-shot selector cap. It bypasses the `CTA_POLICY_ZERO` check only; the selector cap is enforced regardless.
* With `NCCL_CTA_POLICY=2` (`NCCL_CTA_POLICY_ZERO`), AllReduce on registered symmetric windows now takes the registered CE path whenever it is available, even when the symmetric kernel is eligible, matching AllGather and upstream NCCL. A non-zero arch-table `ceRegMax[AllReduce]` no longer caps it under this policy; a zero entry and an explicit `RCCL_CE_AR_REG_MAX_MSG_BYTES` still apply.
* `tuning_model_11` registered for gfx1250 single-node; gfx1250 added to the AINIC tuning index map.
* Extended `ncclFunc_t` enum in rccl-tests `common.h` to match the librccl internal ABI: renumbered existing values and added `ncclFuncAlltoAll` (8), `ncclFuncScatter` (9), `ncclFuncGather` (10), `ncclFuncAllToAllPivot` (11), `ncclFuncAlltoAllGda` (12), `ncclFuncAlltoAllvGda` (13), `ncclFuncAllGatherV` (14), `ncclFuncPutSignal` (15), `ncclFuncSignal` (16), `ncclFuncWaitSignal` (17), `ncclFuncAlltoAllv` (18); `ncclNumFuncs` updated to 19.
* AlltoAll index variables in `alltoall_dda.h` and `alltoall_dda_fabric.h` widened from `int` to `size_t` to prevent overflow on large message sizes.
Expand Down
4 changes: 3 additions & 1 deletion projects/rccl/src/ce_coll.cc
Original file line number Diff line number Diff line change
Expand Up @@ -466,8 +466,10 @@ bool ncclCeImplemented(ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType
case ncclFuncAlltoAllv:
case ncclFuncScatter:
case ncclFuncGather:
case ncclFuncAllReduce:
return true;
case ncclFuncAllReduce:
// The CE local reduce kernels have no fp8 instantiation (ce_reduce.cc returns ncclInvalidArgument).
return ty != ncclFloat8e4m3 && ty != ncclFloat8e5m2;
default:
return false;
}
Expand Down
2 changes: 1 addition & 1 deletion projects/rccl/src/include/rccl_common.h
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,7 @@ NCCL_API(ncclResult_t, rcclGetCollImplInfo, struct ncclComm* comm, ncclFunc_t co
int* protocol, int* maxChannels);
// Single source of truth for AllReduce implementation selection. Runs the exact
// priority chain (GIN-SDMA -> symmetric -> CE 2-shot -> DDA LL/LL128/VMM/IPC -> CE registered
// -> kernel) and returns the decision.
// -> kernel) and returns the decision. Under CTAPolicy=ZERO an available CE registered path beats symmetric.
// query=false : live dispatch path (ncclAllReduce_impl). ceCapturing is probed
// from `stream`; the CE graph latch is ticked; graphCapturingHint
// is ignored.
Expand Down
20 changes: 12 additions & 8 deletions projects/rccl/src/rccl_wrap.cc
Original file line number Diff line number Diff line change
Expand Up @@ -1479,27 +1479,31 @@ ncclResult_t rcclSelectAllReduce(struct ncclComm* comm, const void* sendbuff, vo

// (5) Enqueue-bound backends: CE registered (Branch B) vs symmetric vs kernel.
// Reproduce taskAppend()'s AllReduce CE decision exactly so both agree.
// develop's taskAppend appends CE for AllReduce iff !hasSysmemSegment && ceAvailable
// && ((CTAPolicy & ZERO) || force): ceAvailable is the conjunction of the four
// sub-conditions below; split out so the disqualification log can name the blocker.
// Under ZERO (ceRegZero) CE wins over an eligible symk at any size; ceRegForce (force without ZERO) yields to symk and ceRegMax.
// ceAvailable is the conjunction of the sub-conditions below; split out so the disqualification log can name the blocker.
const bool ceBufferOk = !ceCapturing && ncclCeAvailable(comm, ncclFuncAllReduce, (int)op, datatype, winRegType, sendWin, recvWin);
const bool ceAllReduceOpSupported = (op == ncclSum || op == ncclProd || op == ncclMin || op == ncclMax);
const bool ceCountDivisible = (count % (size_t)comm->nRanks == 0);
const bool ceEnabledByArch = rcclCeAllReduceEnabledDef(ceArArchDefault);
const bool ceAvailable = ceArGraphAllowed && ceBufferOk && ceAllReduceOpSupported && ceCountDivisible && ceEnabledByArch;
// Tuning cap only: registered CE has no staging allocation, so this does not
// size a buffer. kThreshUnlimited (or null table) = no upper bound;
// env var 0 returns 0, making ceRegInWindow false (disables registered CE).
// size a buffer. kThreshUnlimited = no upper bound; 0 (null table, untuned
// collective, or env var 0) disables registered CE, under ZERO too.
// Independent of the 2-shot selector (ceNonRegMax/env, 0 = off) and ceARTmpBuf sizing.
const size_t ceArRegMax = rcclCeRegMaxTab(archTable, ncclFuncAllReduce);
const bool ceRegInWindow = ceArRegMax == kThreshUnlimited || msgBytes <= ceArRegMax;
if (!symEligible && ceRegInWindow && ceAvailable && !hasSysmemSegment &&
((comm->config.CTAPolicy & NCCL_CTA_POLICY_ZERO) || force)) {
// CTAPolicy=ZERO takes CE whenever it is available, like upstream taskAppend and rcclSelectAllGather Branch #3.
// Under ZERO symk does not veto CE and a positive table cap is ignored; a cap of 0 and an explicit env cap still apply.
const bool ceZeroPolicy = (comm->config.CTAPolicy & NCCL_CTA_POLICY_ZERO) != 0;
const bool ceRegEnvCapOk = rcclParamCeArRegMaxMsgBytes() < 0 || ceRegInWindow;
const bool ceRegZero = ceZeroPolicy && ceArRegMax != 0 && ceRegEnvCapOk;
const bool ceRegForce = force && !symEligible && ceRegInWindow;
Comment thread
pvallem marked this conversation as resolved.
Resolved
if (ceAvailable && !hasSysmemSegment && (ceRegZero || ceRegForce)) {
Comment thread
pvallem marked this conversation as resolved.
Resolved
decision->algo = RCCL_CE_REGISTERED;
decision->nMaxChannels = ncclCeLocalReduceBlocks(datatype, count / comm->nRanks);
return ncclSuccess;
}
if (!query && !symEligible) INFO(NCCL_TUNING,
if (!query && (ceZeroPolicy || !symEligible)) INFO(NCCL_TUNING,
"AR CE-registered disqualified: ceAvailable=%d(graphAllowed=%d bufOk=%d opOk=%d countDiv=%d archEnabled=%d)"
" ceRegInWindow=%d hasSysmem=%d CTAPolicy=%d force=%d",
(int)ceAvailable, (int)ceArGraphAllowed, (int)ceBufferOk, (int)ceAllReduceOpSupported,
Expand Down
13 changes: 13 additions & 0 deletions projects/rccl/test/CeAllReduceEligibilityTests.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,19 @@ TEST_F(CeAllReduceEligibilityTest, CeImplementedReturnsTrueForAllReduceOnSupport
EXPECT_TRUE(ncclCeImplemented(ncclFuncAllReduce, ncclDevSum, ncclFloat32));
}

TEST_F(CeAllReduceEligibilityTest, CeImplementedRejectsFp8AllReduceOnly)
{
if(!isCeRuntimeDriverSupported())
GTEST_SKIP() << "CE driver not in supported range";

EXPECT_FALSE(ncclCeImplemented(ncclFuncAllReduce, ncclDevSum, ncclFloat8e4m3));
EXPECT_FALSE(ncclCeImplemented(ncclFuncAllReduce, ncclDevSum, ncclFloat8e5m2));
EXPECT_FALSE(ncclCeAvailable(mockComm_.get(), ncclFuncAllReduce, ncclDevSum, ncclFloat8e4m3,
ncclSymSendRegRecvReg, nullptr, nullptr));
EXPECT_TRUE(ncclCeImplemented(ncclFuncAllReduce, ncclDevSum, ncclBfloat16));
EXPECT_TRUE(ncclCeImplemented(ncclFuncAllGather, ncclDevSum, ncclFloat8e4m3));
}

TEST_F(CeAllReduceEligibilityTest, CeAvailable_EligibleWithSymmetricSingleNode)
{
if(!isCeRuntimeDriverSupported())
Expand Down
132 changes: 118 additions & 14 deletions projects/rccl/test/ce/CeMPITests.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@

#include <gtest/gtest.h>
#include <hip/hip_runtime.h>
#include <algorithm>
#include <string>

#ifdef MPI_TESTS_ENABLED
Expand All @@ -41,6 +42,13 @@ constexpr size_t kMediumCount = 65536; // elements per rank for larger test
// partial tail chunk whenever nRanks is not a power of two. Costs a transient
// host buffer of the same size per rank in fill/verify.
constexpr size_t kPipelinedTotalBytes = 512ull * 1024 * 1024;
// Non-Sum AllReduce size inside the gfx942/gfx950 CE 2-shot cap (256 MiB), so the registered to 2-shot handoff can run.
constexpr size_t kCeOnlyTotalBytes = 128ull * 1024 * 1024;
constexpr size_t kCeOnlyRankStride = 517; // per-rank offset so every rank wins some elements of Max/Min
constexpr size_t kCeOnlyValueRange = 4096; // keeps fill values exact in float32
constexpr int kCeOnlyIters = 3; // first call is CE registered, later ones CE 2-shot once ceARTmpBuf exists
// A shard spans 8 default staging slots (128 MiB / nRanks vs 16 MiB / nRanks), so the CE pipeline must run.
constexpr size_t kCeOnlyMinChunksPerShard = 2;
constexpr int kMinRanks2 = 2;
constexpr int kMinRanks4 = 4;
constexpr int kMinRanks8 = 8;
Expand Down Expand Up @@ -247,6 +255,11 @@ class CeMPITest : public MPITestBase
}
}

static bool isCeAllReduceAlgo(int algo)
{
return algo == static_cast<int>(RCCL_CE_REGISTERED) || algo == static_cast<int>(RCCL_CE_2SHOT);
}

bool isCeAllReduceExpected() const
{
// CE AllReduce is single-node only (ceAllReduceFits / nNodes>1 gate).
Expand All @@ -272,8 +285,9 @@ class CeMPITest : public MPITestBase
<< context
<< ": CE AllReduce log marker found but CE AR was not expected";
if(!ceLogShowsAllReducePath(log))
TEST_INFO("%s: assertion passed — non-CE-AR path (CE AR prerequisites not met)",
context);
{
TEST_INFO("%s: assertion passed: non-CE-AR path (CE AR prerequisites not met)", context);
}
}
}

Expand Down Expand Up @@ -714,7 +728,9 @@ class CeMPI_AllReduce : public CeMPITest
void assertCEAllReduceChunking(size_t minChunksPerShard, const char* context)
{
if(!isCeAllReduceExpected() || minChunksPerShard == 0)
{
return;
}

const size_t chunksPerShard = ceLogChunksPerShard(readAllLogs());
EXPECT_GE(chunksPerShard, minChunksPerShard)
Expand Down Expand Up @@ -744,6 +760,17 @@ class CeMPI_AllReduce : public CeMPITest
ASSERT_EQ(ncclSuccess, allocSymBuf(bytes, recvSym));

fillRankScalar(sendSym.ptr, alignedCount, rank);
int predictedAlgo = -1, proto = 0, nChannels = 0;
ASSERT_EQ(ncclSuccess,
rcclGetCollImplInfo(getActiveCommunicator(), ncclFuncAllReduce, alignedCount, ncclFloat32, op,
sendSym.ptr, recvSym.ptr, /*graphCapturing=*/0, &predictedAlgo, &proto,
&nChannels));
if(isCeAllReduceExpected())
{
// Under CTAPolicy=ZERO an available CE registered path beats symk, so the reporter must name a CE variant.
ASSERT_TRUE(isCeAllReduceAlgo(predictedAlgo)) << testId << ": predicted algo " << predictedAlgo
<< " is not a CE AllReduce variant";
}

ASSERT_EQ(ncclSuccess,
ncclAllReduce(sendSym.ptr, recvSym.ptr, alignedCount, ncclFloat32, op,
Expand All @@ -761,6 +788,82 @@ class CeMPI_AllReduce : public CeMPITest
assertCEAllReducePathTaken(testId);
assertCEAllReduceChunking(minChunksPerShard, testId);
}

// Max/Min on symmetric windows: symk does not request non-Sum, so after the first call CE 2-shot can take over from registered.
void runCeOnlyAllReduce(ncclRedOp_t op, const char* testId)
{
if(!validateTestPrerequisites(kMinRanks2))
{
GTEST_SKIP() << "Need >= " << kMinRanks2 << " MPI ranks";
}
int device = 0;
hipDeviceProp_t prop{};
ASSERT_EQ(hipSuccess, hipGetDevice(&device));
ASSERT_EQ(hipSuccess, hipGetDeviceProperties(&prop, device));
const std::string arch(prop.gcnArchName);
const bool archOk = arch.find("gfx942") != std::string::npos || arch.find("gfx950") != std::string::npos;
const bool ceOk = isCeAllReduceDispatchConfigured() && !isMultiNodeTest();
const std::string skipReason = mpiCoordinatedSkipReason(
!archOk || !ceOk, !ceOk ? "Needs single-node CE AllReduce dispatch (NCCL_CTA_POLICY=2 NCCL_CUMEM_ENABLE=1 "
"NCCL_LOCAL_REGISTER=0 RCCL_CE_ALLREDUCE=1)"
: "CE AllReduce caps for this case are tuned for gfx942/gfx950 only");
if(!skipReason.empty())
{
GTEST_SKIP() << skipReason;
}

ASSERT_EQ(ncclSuccess, createTestCommunicator());
ncclComm_t comm = getActiveCommunicator();
int rank{}, nRanks{};
ncclCommUserRank(comm, &rank);
ncclCommCount(comm, &nRanks);

const size_t count = ceAllReduceAlignedCount(kCeOnlyTotalBytes / sizeof(float), nRanks);
const size_t bytes = count * sizeof(float);
SymBuf sendSym, recvSym;
ASSERT_EQ(ncclSuccess, allocSymBuf(bytes, sendSym));
ASSERT_EQ(ncclSuccess, allocSymBuf(bytes, recvSym));

const auto value = [](int r, size_t i) {
return static_cast<float>((i + static_cast<size_t>(r) * kCeOnlyRankStride) % kCeOnlyValueRange);
};
ASSERT_EQ(hipSuccess, initializeBufferWithPattern<float>(sendSym.ptr, count,
[rank, value](size_t i) { return value(rank, i); }));
const auto expected = [nRanks, op, value](size_t i) {
float acc = value(0, i);
for(int r = 1; r < nRanks; ++r)
{
acc = (op == ncclMax) ? std::max(acc, value(r, i)) : std::min(acc, value(r, i));
}
return acc;
};

for(int iter = 0; iter < kCeOnlyIters; ++iter)
{
// Fresh comm: no CE staging before the first call, so registered CE runs first and allocates it for 2-shot.
const int wantAlgo = (iter == 0) ? static_cast<int>(RCCL_CE_REGISTERED) : static_cast<int>(RCCL_CE_2SHOT);
int algo = -1, proto = 0, nChannels = 0;
ASSERT_EQ(ncclSuccess, rcclGetCollImplInfo(comm, ncclFuncAllReduce, count, ncclFloat32, op, sendSym.ptr,
recvSym.ptr, /*graphCapturing=*/0, &algo, &proto, &nChannels));
ASSERT_EQ(wantAlgo, algo) << testId << ": iteration " << iter << " predicted the wrong AllReduce backend";

// NaN-fill the output so a skipped or partial reduction cannot match the expected values.
ASSERT_EQ(hipSuccess, hipMemset(recvSym.ptr, 0xFF, bytes));
ASSERT_EQ(ncclSuccess, ncclAllReduce(sendSym.ptr, recvSym.ptr, count, ncclFloat32, op, comm,
getActiveStream()));
ASSERT_EQ(hipSuccess, hipStreamSynchronize(getActiveStream()));

size_t badIndex = 0;
float want = 0.0f, got = 0.0f;
ASSERT_TRUE(verifyBufferData<float>(recvSym.ptr, count, expected, /*num_samples=*/0, /*tolerance=*/0.0,
&badIndex, &want, &got))
<< testId << ": rank " << rank << " iteration " << iter << " mismatch at element " << badIndex
<< ": expected " << want << ", got " << got;
}

assertCEAllReducePathTaken(testId);
assertCEAllReduceChunking(kCeOnlyMinChunksPerShard, testId);
}
};

// CE-MPI-AR-01: 2 ranks, small buffers.
Expand Down Expand Up @@ -792,25 +895,26 @@ TEST_F(CeMPI_AllReduce, LargeMessage)
runAllReduce(kMinRanks4, largeCount, ncclSum, "CeMPI_AllReduce/LargeMessage");
}

// CE-MPI-AR-06: Multi-chunk pipeline (chunksPerShard >= 2).
//
// ncclCeAllReduce() pipelines only when a shard does not fit one staging slot,
// i.e. once the message passes NCCL_CE_AR_MAX_MSG_BYTES. That cap does not gate
// this path: it only sets ceAllReduceFits in taskAppend(), which gates the
// unregistered "force" branch. With symmetric windows registered — as allocSymBuf
// does here — ceAvailable alone selects CE, at any size.
//
// This exercises the persistent reduce kernel's double-buffered slot recycling,
// the cross-rank signal doorbells and the Phase 3 drain loop, none of which run
// on the single-shot path. At a non-power-of-two rank count it additionally hits
// the partial tail chunk and the slotChunkBytes rounding from ce75b9a1.
// CE-MPI-AR-06: Multi-chunk pipeline (chunksPerShard >= 2); under CTAPolicy=ZERO registered CE takes a 512 MiB Sum at any size.
TEST_F(CeMPI_AllReduce, PipelinedMultiChunk)
{
const size_t pipelinedCount = kPipelinedTotalBytes / sizeof(float);
runAllReduce(kMinRanks2, pipelinedCount, ncclSum, "CeMPI_AllReduce/PipelinedMultiChunk",
/*minChunksPerShard=*/2);
}

// CE-MPI-AR-07: Max on symmetric windows covers CE registered then CE 2-shot on one comm.
TEST_F(CeMPI_AllReduce, MaxOp128MiB)
Comment thread
pvallem marked this conversation as resolved.
Resolved
{
runCeOnlyAllReduce(ncclMax, "CeMPI_AllReduce/MaxOp128MiB");
}

// CE-MPI-AR-08: Min on symmetric windows, same selection as CE-MPI-AR-07.
TEST_F(CeMPI_AllReduce, MinOp128MiB)
{
runCeOnlyAllReduce(ncclMin, "CeMPI_AllReduce/MinOp128MiB");
}

// ===========================================================================
// CeMPI_Fallback – CE not taken for AllReduce when RCCL_CE_ALLREDUCE is off
// ===========================================================================
Expand Down
Loading
Loading