Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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
3 changes: 2 additions & 1 deletion projects/rccl/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https:
* Fixed Copy Engine `ncclAlltoAllv` hanging on intra-node communicators of more than 8 ranks with uneven send sizes. Each rank decided from its own send sizes whether to add intra-batch barriers, so ranks could issue different barrier counts. The decision now uses the full send-size matrix, which every rank shares.
* Fixed `NCCL_P2P_DISABLE` and `NCCL_P2P_LEVEL` being ignored on Intel, ARM and Zhaoxin hosts, where the CPU-based default P2P level overwrote the user setting. On such hosts `NCCL_P2P_DISABLE=1` now falls back to SHM as documented, and `NCCL_P2P_LEVEL` is honored.
* Fixed LL-protocol latency on gfx950 when RCCL is built against a ROCm older than 7.0.2 without the 128-bit global load/store builtins (for example ROCm 7.0.1, or `-DDWORDX4_INTRINSICS=OFF`). Every store to the user output buffer was followed by a release fence and an L2 writeback, which made 4 KB to 64 KB `ncclBroadcast` about 30% slower on MI355X. Builds against ROCm 7.0.2 or later were not affected and are unchanged.
* Fixed a 15 to 30% `ncclScatter` bandwidth loss from 16 MB upward on multi-node AINIC jobs, caused by the move to two network channels per peer (for example on 4 nodes of 8 GPUs). Scatter now uses and connects the most channels per peer whose flows still fit the P2P channel pool. Point-to-point operations and the other collectives keep their channel count.

### Known issues
* The FP8 ReduceSum and ReduceCopy device APIs are not exported in the LLVM bitcode library.
Expand Down Expand Up @@ -101,7 +102,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https:
* `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.
* 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.
* DDA fabric and IPC AlltoAll launchers refactored to derive launch geometry from a single inline helper (`ddaAllToAllFabricGeom`, `ddaAllToAllFabricLLGeom`, `ddaAllToAllFabricLL128Geom`, `ddaAllToAllIpcGeom`) shared between the dispatch path and the new block-count query functions.
* `rcclGetCollImplInfo` now covers ReduceScatter and AlltoAll (previously both fell back to `rcclGetAlgoInfo()`); rccl-tests reports accurate algo/proto for all four collective types.
Expand Down
40 changes: 33 additions & 7 deletions projects/rccl/src/enqueue/enqueue.cc
Original file line number Diff line number Diff line change
Expand Up @@ -1326,19 +1326,36 @@ static bool rcclP2pBatchEligible(struct ncclComm* comm, ssize_t sendBytes, ssize
rcclParamP2pBatchThreshold());
}

// Number of the p2pnChannelsPerPeer parts a task of collAPI uses and connects. Multi-node AINIC scatter
// uses and connects only the most parts whose flows from the root still fit the pool; connecting more
// slows it down. A user-set nChannelsPerNetPeer is kept, as in ncclTopoComputeP2pChannels. Depend only
// on values both ends of a pair share, or they disagree on the parts.
int rcclP2pTaskParts(struct ncclComm* comm, ncclFunc_t collAPI) {
if (collAPI == ncclFuncScatter && comm->nNodes > 1 && comm->config.nChannelsPerNetPeer == NCCL_CONFIG_UNDEF_INT &&
rcclUseAinic()) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this only impact AINIC?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, we enabled this only for AINIC, as the regression was observed exclusively on AINIC nodes. We want to avoid introducing unnecessary changes to other hardware configurations.

int partsThatFit = comm->p2pnChannels / (comm->nRanks - 1);
return std::min(comm->p2pnChannelsPerPeer, partsThatFit >= 1 ? pow2Down(partsThatFit) : 1);
}
return comm->p2pnChannelsPerPeer;
Comment thread
i-kosarev marked this conversation as resolved.
Resolved
}

static int rcclP2pPolicyChannels(struct ncclComm* comm, struct ncclTaskP2p* task) {
struct Policy {
const char* arch;
ncclFunc_t collAPI;
int nChannels;
};
// This workaround targets direct P2P/IPC traffic only. When P2P is disabled,
if (task == nullptr) return -1;
// The Scatter part limit is about network flows, so it applies with P2P disabled too.
int taskParts = rcclP2pTaskParts(comm, task->collAPI);
if (taskParts < comm->p2pnChannelsPerPeer) return taskParts;
// The policies below target direct P2P/IPC traffic only. When P2P is disabled,
// preserve the normal channel selection for the fallback transport.
if (task == nullptr || ncclParamP2pDisable()) return -1;
if (ncclParamP2pDisable()) return -1;
constexpr Policy policies[] = {
// Two-channel full-mesh traffic has high variance on gfx110x. Keep the
// workaround specific to AllToAll so Gather, Scatter, and SendRecv retain
// their higher-throughput multi-channel paths.
// workaround specific to AllToAll so Gather, SendRecv and Scatter (outside the
// AINIC part limit above) retain their higher-throughput multi-channel paths.
{"gfx110", ncclFuncAlltoAll, 1},
};
for (const auto& policy : policies) {
Expand Down Expand Up @@ -1371,6 +1388,10 @@ static ncclResult_t addP2pToPlan(struct ncclComm* comm, struct ncclKernelPlan* p
// Keep policy direction-local so a different task paired in the same planner
// round cannot change the channel count selected by this task's peer.
int kChannels[2] = {rcclP2pPolicyChannels(comm, p2pTasks[0]), rcclP2pPolicyChannels(comm, p2pTasks[1])};
int taskParts[2];
for (int dir = 0; dir < 2; dir++)
taskParts[dir] =
p2pTasks[dir] ? std::min(nChannelsMax, rcclP2pTaskParts(comm, p2pTasks[dir]->collAPI)) : nChannelsMax;
bool batchP2P = rcclP2pBatchEligible(comm, sendBytes, recvBytes);
// Keep work-fusion size-gated (batchP2P) but select the channel map from the
// communicator flag. Keying the map to eligibility collapsed large AllToAll
Expand Down Expand Up @@ -1431,6 +1452,7 @@ static ncclResult_t addP2pToPlan(struct ncclComm* comm, struct ncclKernelPlan* p
ncclP2pChannelForPart(comm->p2pnChannels, base, part, nChannelsMax, comm->nNodes, comm->p2pChannelShiftSize);
struct ncclChannelPeer** channelPeers = comm->channels[channelId].peers;
for (int dir = 0; dir <= 1; dir++) {
if (part >= taskParts[dir]) continue;
int peerRank = dir ? sendRank : recvRank;
struct ncclConnector* conn =
dir ? &channelPeers[peerRank]->send[connIndex[dir]] : &channelPeers[peerRank]->recv[connIndex[dir]];
Expand Down Expand Up @@ -1545,7 +1567,7 @@ static ncclResult_t addP2pToPlan(struct ncclComm* comm, struct ncclKernelPlan* p
if (bytes[dir] > 0 && proxySameProcess[dir] && protocol[dir] == NCCL_PROTO_SIMPLE && (!pxnUsed)) {
int regFlag = 0;
NCCLCHECKGOTO(ncclCalloc(&handles[dir], nChannelsMax), ret, cleanup);
for (int part = 0; part < nChannelsMax; part++) {
for (int part = 0; part < taskParts[dir]; part++) {
int channelId = ncclP2pChannelForPart(comm->p2pnChannels, base, part, nChannelsMax, comm->nNodes,
comm->p2pChannelShiftSize);
struct ncclChannelPeer** channelPeers = comm->channels[channelId].peers;
Expand Down Expand Up @@ -3854,14 +3876,18 @@ static ncclResult_t p2pTaskAppend(struct ncclComm* comm, struct ncclInfo* info,
// Mark channels that need pre-connect
if (comm->rank != peer) {
if (!(isSendNotRecv ? planner->peers[peer].sendSeen : planner->peers[peer].recvSeen)) {
int nParts = rcclP2pTaskParts(comm, collAPI);
// planner->peers[peer].send/recvSeen is private to each comm, so we need to set it anyway.
(isSendNotRecv ? planner->peers[peer].sendSeen : planner->peers[peer].recvSeen) = true;
// Leave it clear when only some parts are connected, so a later task to this peer that uses
// all of them still connects the rest.
if (nParts == comm->p2pnChannelsPerPeer)
(isSendNotRecv ? planner->peers[peer].sendSeen : planner->peers[peer].recvSeen) = true;
int round = 0;
while (peer != (isSendNotRecv ? comm->p2pSchedule[round].sendRank : comm->p2pSchedule[round].recvRank)) {
round += 1;
}
uint8_t base = ncclP2pChannelBaseForRound(comm, round, rcclEffectiveP2pBatchEnable(comm));
for (int c = 0; c < comm->p2pnChannelsPerPeer; c++) {
for (int c = 0; c < nParts; c++) {
int channelId = ncclP2pChannelForPart(comm->p2pnChannels, base, c, comm->p2pnChannelsPerPeer, comm->nNodes,
comm->p2pChannelShiftSize);
if (isSendNotRecv) {
Expand Down
13 changes: 8 additions & 5 deletions projects/rccl/src/enqueue/task_prep/task_posttuning.cc
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ static ncclResult_t postTuneP2pChannelBase(struct ncclComm* comm, int peer, bool
return ncclSuccess;
}

static ncclResult_t postTuneP2pRecordPreconnect(struct ncclComm* comm, int peer, bool isSendNotRecv,
static ncclResult_t postTuneP2pRecordPreconnect(struct ncclComm* comm, int peer, bool isSendNotRecv, ncclFunc_t collAPI,
bool* needPreconnect) {
Comment thread
i-kosarev marked this conversation as resolved.
Resolved
struct ncclKernelPlanner* planner = &comm->planner;
uint8_t base;
Expand All @@ -162,9 +162,12 @@ static ncclResult_t postTuneP2pRecordPreconnect(struct ncclComm* comm, int peer,
NCCLCHECK(postTuneP2pChannelBase(comm, peer, isSendNotRecv, &base));

// Mark channels that need pre-connect. planner->peers[peer].send/recvSeen is
// private to each comm, so we need to set it anyway.
(isSendNotRecv ? planner->peers[peer].sendSeen : planner->peers[peer].recvSeen) = true;
for (int c = 0; c < comm->p2pnChannelsPerPeer; c++) {
// private to each comm, so we need to set it anyway. Leave it clear when only
// some parts are connected, so a later task to this peer still connects the rest.
int nParts = rcclP2pTaskParts(comm, collAPI);
Comment thread
i-kosarev marked this conversation as resolved.
Resolved
if (nParts == comm->p2pnChannelsPerPeer)
(isSendNotRecv ? planner->peers[peer].sendSeen : planner->peers[peer].recvSeen) = true;
for (int c = 0; c < nParts; c++) {
int channelId = ncclP2pChannelForPart(comm->p2pnChannels, base, c, comm->p2pnChannelsPerPeer, comm->nNodes,
comm->p2pChannelShiftSize);

Expand Down Expand Up @@ -901,7 +904,7 @@ static ncclResult_t postTuneP2pTasks(
struct ncclRawTaskSendRecv* raw = &tInfo->raw->sendRecv;
bool isSendNotRecv = raw->func == ncclFuncSend;

NCCLCHECK(postTuneP2pRecordPreconnect(comm, raw->peer, isSendNotRecv, &needPreconnect));
NCCLCHECK(postTuneP2pRecordPreconnect(comm, raw->peer, isSendNotRecv, raw->collAPI, &needPreconnect));
}

if (needPreconnect) {
Expand Down
1 change: 1 addition & 0 deletions projects/rccl/src/include/channel.h
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

bool rcclUseAinic();
int rcclEffectiveP2pBatchEnable(struct ncclComm* comm);
int rcclP2pTaskParts(struct ncclComm* comm, ncclFunc_t collAPI);

RCCL_PARAM_DECLARE(
PxnOptQpUsage); // RCCL_PXN_OPT_QP_USAGE: uses batch stride of comm->maxLocalRanks instead of 1 to reduce QP usage when p2p-batching is disabled
Expand Down
62 changes: 62 additions & 0 deletions projects/rccl/test/host/enqueue-test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -339,6 +339,68 @@ TEST_F(EnqueueMicrotest, ShmemDynamicSize_DeviceLinker_IsAlwaysZero) {
}
#endif

// ===========================================================================
// rcclP2pTaskParts (enqueue.cc:1333) and rcclP2pPolicyChannels (enqueue.cc:1342)
//
// Multi-node AINIC Scatter takes the most power-of-two parts whose nRanks - 1
// flows fit p2pnChannels, at most p2pnChannelsPerPeer; everything else takes
// p2pnChannelsPerPeer. Both ends of a pair must get the same answer, so it may
// depend only on comm-wide fields.
// ===========================================================================

namespace {
std::unique_ptr<ncclComm> P2pPartsComm(int nNodes, int nRanks, int p2pnChannels, int perPeer) {
std::unique_ptr<ncclComm> comm{new ncclComm{}};
comm->nNodes = nNodes;
comm->nRanks = nRanks;
comm->p2pnChannels = p2pnChannels;
comm->p2pnChannelsPerPeer = perPeer;
comm->config.nChannelsPerNetPeer = NCCL_CONFIG_UNDEF_INT;
return comm;
}
} // namespace

TEST_F(EnqueueMicrotest, P2pTaskParts_AinicScatterTakesThePartsWhoseFlowsFitThePool) {
g_rcclUseAinicValue = true;
struct Case {
int nNodes, nRanks, p2pnChannels, perPeer, scatterParts;
};
// 8 GPUs per node with the MI355X pools, the pool boundary (16 peers x 2 parts fill 32 channels,
// 17 peers do not), power-of-two rounding (64 / 19 = 3) and the one-part floor (64 / 127 = 0).
for (Case k : {Case{2, 16, 32, 2, 2}, Case{3, 24, 64, 4, 2}, Case{4, 32, 32, 2, 1}, Case{8, 64, 64, 2, 1},
Case{2, 17, 32, 2, 2}, Case{2, 18, 32, 2, 1}, Case{3, 20, 64, 8, 2}, Case{16, 128, 64, 2, 1}}) {
auto comm = P2pPartsComm(k.nNodes, k.nRanks, k.p2pnChannels, k.perPeer);
EXPECT_EQ(k.scatterParts, rcclP2pTaskParts(comm.get(), ncclFuncScatter))
<< k.nRanks << " ranks, " << k.p2pnChannels << " channels";
for (ncclFunc_t f : {ncclFuncSendRecv, ncclFuncSend, ncclFuncRecv, ncclFuncAlltoAll, ncclFuncGather,
ncclFuncAlltoAllv}) {
EXPECT_EQ(k.perPeer, rcclP2pTaskParts(comm.get(), f)) << "collAPI=" << int(f);
}
}
}

TEST_F(EnqueueMicrotest, P2pTaskParts_ScatterKeepsPerPeerOutsideMultiNodeAinicOrWithAUserCount) {
auto comm = P2pPartsComm(/*nNodes=*/4, /*nRanks=*/32, /*p2pnChannels=*/32, /*perPeer=*/2);
g_rcclUseAinicValue = false;
EXPECT_EQ(2, rcclP2pTaskParts(comm.get(), ncclFuncScatter)) << "not AINIC";
g_rcclUseAinicValue = true;
comm->config.nChannelsPerNetPeer = 2;
EXPECT_EQ(2, rcclP2pTaskParts(comm.get(), ncclFuncScatter)) << "NCCL_NCHANNELS_PER_NET_PEER set";
comm->config.nChannelsPerNetPeer = NCCL_CONFIG_UNDEF_INT;
comm->nNodes = 1;
EXPECT_EQ(2, rcclP2pTaskParts(comm.get(), ncclFuncScatter)) << "single node";
}

TEST_F(EnqueueMicrotest, P2pPolicyChannels_ScatterBelowPerPeerReturnsItsParts) {
// Returns before the arch policy table, so comm->topo is never read.
g_rcclUseAinicValue = true;
auto comm = P2pPartsComm(/*nNodes=*/4, /*nRanks=*/32, /*p2pnChannels=*/32, /*perPeer=*/2);
ncclTaskP2p task{};
task.collAPI = ncclFuncScatter;
EXPECT_EQ(1, rcclP2pPolicyChannels(comm.get(), &task));
EXPECT_EQ(-1, rcclP2pPolicyChannels(comm.get(), nullptr));
}

// ===========================================================================
// calcP2pChannelCount (enqueue.cc:1550)
//
Expand Down
11 changes: 10 additions & 1 deletion projects/rccl/test/host/fakes/enqueue_fakes.cc
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

#include "signature-drift.h"

#include "channel.h" // rcclEffectiveP2pBatchEnable
#include "channel.h" // rcclEffectiveP2pBatchEnable, rcclP2pTaskParts
#include "enqueue_fakes.h"
#include "nccl_fakes.h" // g_loadParam, for the NCCL_PARAM body this file stands in for
#include "sym_kernels_fakes.h" // g_symkAvailable and the sym_kernels.cc seams below's canonical home
Expand All @@ -31,6 +31,7 @@ ASSERT_HOOK_MATCHES_PROD(g_addProxyOpIfNeeded, ncclAddProxyOpIfNeeded);
ASSERT_HOOK_MATCHES_PROD(g_getCollNetSupport, ncclGetCollNetSupport);
ASSERT_HOOK_MATCHES_PROD(g_getRegBuff, ncclGetRegBuff);
ASSERT_HOOK_MATCHES_PROD(g_rcclEffectiveP2pBatchEnable, rcclEffectiveP2pBatchEnable);
ASSERT_HOOK_MATCHES_PROD(g_rcclP2pTaskParts, rcclP2pTaskParts);
#undef ASSERT_HOOK_MATCHES_PROD

// Generous default: a deny-everything default would make even a single small task look over budget.
Expand Down Expand Up @@ -75,6 +76,10 @@ std::function<ncclResult_t(struct ncclComm*, struct ncclKernelPlan*, struct nccl
static int DefaultEffectiveP2pBatchEnable(struct ncclComm*) { return 0; }
std::function<int(struct ncclComm*)> g_rcclEffectiveP2pBatchEnable = DefaultEffectiveP2pBatchEnable;

// enqueue.cc's answer for every task outside multi-node AINIC Scatter.
static int DefaultP2pTaskParts(struct ncclComm* comm, ncclFunc_t) { return comm->p2pnChannelsPerPeer; }
std::function<int(struct ncclComm*, ncclFunc_t)> g_rcclP2pTaskParts = DefaultP2pTaskParts;

// enqueue.cc:437-441 splits the default on ROCm version; mirrored rather than pinned to one arm.
#if ROCM_VERSION >= 71200
int64_t ncclParamGraphRegister() { return g_loadParam("GRAPH_REGISTER", 1); }
Expand Down Expand Up @@ -105,11 +110,15 @@ void ResetEnqueueFakes() {
g_getCollNetSupport = DefaultGetCollNetSupport;
g_getRegBuff = DefaultGetRegBuff;
g_rcclEffectiveP2pBatchEnable = DefaultEffectiveP2pBatchEnable;
g_rcclP2pTaskParts = DefaultP2pTaskParts;
}

int rcclEffectiveP2pBatchEnable(struct ncclComm* comm) {
return g_rcclEffectiveP2pBatchEnable(comm);
}
int rcclP2pTaskParts(struct ncclComm* comm, ncclFunc_t collAPI) {
return g_rcclP2pTaskParts(comm, collAPI);
}

// src/enqueue/enqueue.cc
ncclResult_t ncclGetAlgoInfo(struct ncclComm* comm, struct ncclTaskColl* task, int collNetSupport, int nvlsSupport,
Expand Down
3 changes: 3 additions & 0 deletions projects/rccl/test/host/fakes/enqueue_fakes.h
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,9 @@ extern std::function<ncclResult_t(struct ncclComm*, struct ncclTaskColl*, int*)>
// enqueue.cc's rcclEffectiveP2pBatchEnable: default 0 is batching off, the single-node answer.
extern std::function<int(struct ncclComm*)> g_rcclEffectiveP2pBatchEnable;

// enqueue.cc's rcclP2pTaskParts: default p2pnChannelsPerPeer; install the multi-node AINIC Scatter limit per test.
extern std::function<int(struct ncclComm*, ncclFunc_t)> g_rcclP2pTaskParts;

void ResetEnqueueFakes();

#endif // RCCL_TEST_HOST_FAKES_ENQUEUE_FAKES_H_
Loading
Loading