Skip to content
Merged
Show file tree
Hide file tree
Changes from 12 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 @@ -47,6 +47,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https:

### Removed
* `NCCL_TOPO_SCATTER_START_NET`, which selected how the first network device was scattered across GPUs. Rail and plane assignment replaces it.
* `RCCL_DDA_LL128_RS_THREADS` removed (was default `1024`): reduce-scatter LL128 now launches a fixed 512-thread block, one warp per slice, so the thread count is no longer a tuning knob.
Comment thread
pvallem marked this conversation as resolved.
Resolved

### Resolved issues
* Fixed profiler overhead when no profiler plugin is loaded or only kernel-channel events are enabled.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,7 @@ namespace dda::common {
// slices, so a slot always begins on a slice boundary and the 16B wire accesses
// stay aligned however nRanks divides the bank.
constexpr size_t ddaLL128AgSlotWords(size_t bankSize, int nRanks) {
return ddaLLSlotPkts(bankSize, sizeof(uint64_t) * (size_t)nRanks,
(size_t)kDdaLL128WireWordsPerSlice);
return ddaLL128SlotWords(bankSize, nRanks, 1);
}

// LL128 all-gather. 2D grid: grid.x == nRanks - 1 places one column per remote
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,10 @@ bool ncclAllGatherDdaIpcEligible(ncclComm* comm, const void* sendbuff, void* rec
return false;
}

if (!dda::common::ddaUserBuffers16ByteAligned(sendbuff, recvbuff)) {
return false;
}

return true;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,7 @@ namespace dda::common {
// slices, so a slot always begins on a slice boundary and the 16B wire accesses
// stay aligned however nRanks divides the bank
constexpr size_t ddaLL128ArSlotWords(size_t bankSize, int nRanks) {
return ddaLLSlotPkts(bankSize, sizeof(uint64_t) * (size_t)nRanks,
(size_t)kDdaLL128WireWordsPerSlice);
return ddaLL128SlotWords(bankSize, nRanks, 1);
}

// LL128 flat all-reduce kernel. One warp owns one slice at a time; the grid is
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,7 @@ namespace dda::common {
// same reason as the one-shot: an unrounded stride can come out an odd number of
// 8B words and misalign every slot but rank 0's.
constexpr size_t ddaLL128ArTwoShotSlotWords(size_t bankSize, int nRanks) {
return ddaLLSlotPkts(bankSize, sizeof(uint64_t) * (size_t)nRanks * 2,
(size_t)kDdaLL128WireWordsPerSlice);
return ddaLL128SlotWords(bankSize, nRanks, 2);
}

// Fixed-width peer staging for phase 1: unlike the one-shot, each peer receives a
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -92,8 +92,6 @@ static ncclResult_t ncclAllReduceDdaIpcTyped(const void* sendbuff, void* recvbuf

bool ncclAllReduceDdaIpcEligible(ncclComm* comm, const void* sendbuff, void* recvbuff, size_t count,
ncclDataType_t datatype, ncclRedOp_t op) {
(void)sendbuff;
(void)recvbuff;
if (comm == nullptr) {
return false;
}
Expand Down Expand Up @@ -133,6 +131,9 @@ bool ncclAllReduceDdaIpcEligible(ncclComm* comm, const void* sendbuff, void* rec
// 16-byte alignment: the DDA kernels do 16-byte vectorized loads.
return false;
}
if (!dda::common::ddaUserBuffers16ByteAligned(sendbuff, recvbuff)) {
return false;
}
if (bytes > kDdaFlatTreeThresholdBytes) {
if (count % comm->nRanks || ((count / comm->nRanks) * ncclTypeSize(datatype)) % 16) {
// Two-shot/tree path: each rank reduces count/nRanks elements, so that
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,10 @@ bool ncclAllToAllDdaIpcEligible(ncclComm* comm, const void* sendbuff, void* recv
return false;
}

if (!dda::common::ddaUserBuffers16ByteAligned(sendbuff, recvbuff)) {
return false;
}

return true;
}

Expand Down
28 changes: 28 additions & 0 deletions projects/rccl/src/algorithms/dda/device/CollCommon.h
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,34 @@ inline bool ddaAlltoAllSingleBlockGrid(size_t count, int typeSize) {
return count < elementsPerThread * kThreadsPerBlock;
}

// copyFromSrcToDest loads user buffers as uint4, so both pointers have to be
// 16-byte aligned. Shared by the IPC eligibility checks.
inline bool ddaUserBuffers16ByteAligned(const void* sendbuff, const void* recvbuff) {
return (reinterpret_cast<uintptr_t>(sendbuff) % 16) == 0 &&
(reinterpret_cast<uintptr_t>(recvbuff) % 16) == 0;
}

// Reduce-scatter IPC copies a shard inside the kernel at or below this size.
// Above it the host pre-copies the whole input into scratch. The cutoff is
// per shard, matching the kernel, so a multi-rank total cannot stage twice.
constexpr size_t kDdaRsIpcKernelCopyMaxBytes = 1ULL * 1024 * 1024;

enum class DdaRsIpcStaging {
FusedInKernel, // one block: fuse the sendbuff copy into the kernel
PerShardInKernel, // multi-block, shard small enough for the in-kernel copy
HostPreCopy, // multi-block, host cudaMemcpyAsync before launch
};

inline DdaRsIpcStaging ddaReduceScatterIpcStaging(size_t recvcount, int typeSize) {
if (ddaAlltoAllSingleBlockGrid(recvcount, typeSize)) {
return DdaRsIpcStaging::FusedInKernel;
}
if (recvcount * static_cast<size_t>(typeSize) <= kDdaRsIpcKernelCopyMaxBytes) {
return DdaRsIpcStaging::PerShardInKernel;
}
return DdaRsIpcStaging::HostPreCopy;
}

constexpr uint32_t calcBlockCount(size_t numThreads, size_t threadsPerBlock, size_t maxBlocks) {
const auto uNumThreads = static_cast<uint64_t>(numThreads);
const auto uThreadsPerBlock = static_cast<uint64_t>(threadsPerBlock);
Expand Down
9 changes: 9 additions & 0 deletions projects/rccl/src/algorithms/dda/device/CollCommon_ll128.h
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,15 @@ constexpr int kDdaLL128DataBytesPerSlice =
static_assert(kDdaLL128WireBytesPerSlice % (int)kDdaLL128LineBytes == 0,
"a slice must be a whole number of lines");

// Per-rank slot in 8B wire words. nStagingAreas is 1 for one-shot publish/reduce
// and 2 for two-shot (publish plus write-back). Shared by all-reduce, all-gather
// and reduce-scatter so the four tiers carve the bank the same way.
constexpr size_t ddaLL128SlotWords(size_t bankSize, int nRanks, int nStagingAreas) {
return ddaLLSlotPkts(bankSize,
sizeof(uint64_t) * (size_t)nRanks * (size_t)nStagingAreas,
(size_t)kDdaLL128WireWordsPerSlice);
}

// The last lane of each line's lane group owns that line's flag word: lanes
// 7,15,23,31 for a 128-byte line
__device__ __forceinline__ bool ddaLL128IsFlagLane(int wid) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,9 @@
* Copyright (c) 2026, Advanced Micro Devices, Inc. All rights reserved.
*
* Host launcher + eligibility for the LL128-protocol DDA fabric reduce-scatter.
* Reduce-scatter analogue of dda_all_reduce_fabric_ll128.cu; reuses the codepath-
* agnostic ddaReduceScatterFabricLL128 kernel from reduce_scatter_dda_fabric_ll128.h.
* Reduce-scatter analogue of the LL128 one-shot launcher in
* dda_all_reduce_fabric_ll.cu; reuses the codepath-agnostic
* ddaReduceScatterFabricLL128 kernel from reduce_scatter_dda_fabric_ll128.h.
* See LICENSE.txt for license information.
************************************************************************/

Expand All @@ -15,93 +16,94 @@
#include "debug.h"
#include "algorithms/dda/fabric/fabric_gpu_barrier.h" // dda::common::kDdaMaxNranks
#include "param.h"
#include "rccl_common.h"

#include <cuda_runtime.h>
#include <hip/hip_ext.h>

#include <algorithm>
#include <cstddef>
#include <cstdint>

// Runtime-adjustable LL128 ReduceScatter block size (threads/block). Must be a
// multiple of 16 (lanes per 128B line) in [16, 1024]; invalid values fall back
// to the tuned default. Env: RCCL_DDA_LL128_RS_THREADS.
RCCL_PARAM(DdaLL128RsThreads, "DDA_LL128_RS_THREADS", 1024);
#include <utility>

namespace {

using dda::common::kDdaLL128DataElems;
using dda::common::kDdaLL128Lanes;
using dda::common::kDdaLL128RsMaxBytes;
using dda::common::kDdaLL128RsSlotStrideLines;
using dda::common::LLLine128;

// LL128 scratch: 2 banks * nRanks slots * kDdaLL128RsSlotStrideLines * 128B.
static inline size_t ddaLL128RsScratchSize(int nRanks) {
return (size_t)2 * (size_t)nRanks * kDdaLL128RsSlotStrideLines * sizeof(LLLine128);
}

// Validated block size from the runtime flag; falls back to `dflt` if the
// configured value is not a multiple of 16 in [16, 1024].
static inline unsigned ddaLL128RsThreads(unsigned dflt) {
const int64_t v = rcclParamDdaLL128RsThreads();
if (v >= 16 && v <= 1024 && (v % 16) == 0) {
return (unsigned)v;
}
return dflt;
}

template <typename T>
static ncclResult_t ncclReduceScatterDdaFabricLL128Typed(const void* sendbuff, void* recvbuff, size_t recvcount,
ncclComm* comm, cudaStream_t stream) {
const int nRanks = comm->nRanks;
const size_t bytes = recvcount * sizeof(T); // per-rank shard bytes
const size_t nWords = bytes >> 3;
const size_t numLines = (nWords + (size_t)kDdaLL128DataElems - 1) / (size_t)kDdaLL128DataElems;

// 1D grid over line-groups; each block has threads/16 groups.
const unsigned threads = ddaLL128RsThreads(1024); // multiple of 16 (lanes/line)
const size_t groups = threads / (unsigned)kDdaLL128Lanes;
using dda::common::ddaBankSize;
using dda::common::ddaLL128RsSlotWords;
using dda::common::ddaLL128Slices;
using dda::common::kDdaLL128Warp;
using dda::common::kDdaLL128WireWordsPerSlice;

// One warp carries one slice, so the grid is sized in slices rather than in
// lines: blocks = ceil(slices / warpsPerBlock), capped by the grid budget. A
// shard with more slices than the capped grid has warps just means each warp
// takes several passes through the slice loop.
static inline std::pair<dim3, dim3> ddaReduceScatterFabricLL128Geom(ncclComm* comm, size_t shardBytes) {
const size_t slices = ddaLL128Slices(shardBytes);
const unsigned threads = 512;
const size_t warps = threads / (unsigned)kDdaLL128Warp;
int nBlocksMax = comm->ddaFabricMaxBlocks;
if (nBlocksMax < 1) {
nBlocksMax = 1;
}
unsigned blocks = (unsigned)std::min<size_t>((numLines + groups - 1) / groups, (size_t)nBlocksMax);
unsigned blocks = (unsigned)std::min<size_t>((slices + warps - 1) / warps, (size_t)nBlocksMax);
Comment thread
nusislam marked this conversation as resolved.
Resolved
if (blocks == 0) {
blocks = 1;
}
// flatBlockId (blockIdx.x) must stay within the device epoch array.
if ((int)blocks > comm->ddaLLEpochLen) {
blocks = (unsigned)comm->ddaLLEpochLen;
if (blocks == 0) blocks = 1;
if (blocks == 0) {
blocks = 1;
}
}
dim3 block(threads);
dim3 grid(blocks);
return std::make_pair(dim3(blocks), dim3(threads));
}

template <typename T>
static ncclResult_t ncclReduceScatterDdaFabricLL128Typed(const void* sendbuff, void* recvbuff, size_t recvcount,
ncclComm* comm, cudaStream_t stream) {
const int nRanks = comm->nRanks;
const size_t shardBytes = recvcount * sizeof(T);
const size_t slices = ddaLL128Slices(shardBytes);
const size_t bankSize = ddaBankSize(comm->ddaScratchBytes);

auto gridBlock = ddaReduceScatterFabricLL128Geom(comm, shardBytes);
const dim3 grid = gridBlock.first;
const dim3 block = gridBlock.second;

T** peers = reinterpret_cast<T**>(comm->ddaPeerPtrsDev);
// Same epoch counter every other LL/LL128 tier uses: they share a scratch
// layout, so one monotonic flag is what keeps either from accepting a line
// the other left.
uint32_t* epochDev = comm->ddaLLEpochDev;
const int epochLen = comm->ddaLLEpochLen;

INFO(NCCL_COLL, "DDA fabric ReduceScatter LL128: nRanks=%d shardBytes=%zu numLines=%zu grid=%u block=%u", nRanks,
bytes, numLines, grid.x, block.x);
INFO(NCCL_COLL,
"DDA fabric ReduceScatter LL128: nRanks=%d shardBytes=%zu slices=%zu grid=%u block=%u "
"(warp-per-slice, bankSize=%zu)",
nRanks, shardBytes, slices, grid.x, block.x, bankSize);

const hipEvent_t stopEvent = rcclTakeAddonStopEvent(comm);
// NRANKS_CT 4/8: unrolled reduce loop; 0: runtime fallback.
// NRANKS_CT 4/8: unrolled peer loops; 0: runtime fallback.
switch (nRanks) {
case 4:
hipExtLaunchKernelGGL((dda::common::ddaReduceScatterFabricLL128<T, 4>), grid, block, 0, stream,
/*startEvent=*/nullptr, stopEvent, /*flags=*/0, peers, static_cast<T*>(recvbuff),
static_cast<const T*>(sendbuff), recvcount, comm->rank, nRanks, epochDev, epochLen);
static_cast<const T*>(sendbuff), shardBytes, comm->rank, nRanks, epochDev, epochLen, slices,
bankSize);
break;
case 8:
hipExtLaunchKernelGGL((dda::common::ddaReduceScatterFabricLL128<T, 8>), grid, block, 0, stream,
/*startEvent=*/nullptr, stopEvent, /*flags=*/0, peers, static_cast<T*>(recvbuff),
static_cast<const T*>(sendbuff), recvcount, comm->rank, nRanks, epochDev, epochLen);
static_cast<const T*>(sendbuff), shardBytes, comm->rank, nRanks, epochDev, epochLen, slices,
bankSize);
break;
default:
hipExtLaunchKernelGGL((dda::common::ddaReduceScatterFabricLL128<T, 0>), grid, block, 0, stream,
/*startEvent=*/nullptr, stopEvent, /*flags=*/0, peers, static_cast<T*>(recvbuff),
static_cast<const T*>(sendbuff), recvcount, comm->rank, nRanks, epochDev, epochLen);
static_cast<const T*>(sendbuff), shardBytes, comm->rank, nRanks, epochDev, epochLen, slices,
bankSize);
break;
}

Expand All @@ -114,8 +116,6 @@ static ncclResult_t ncclReduceScatterDdaFabricLL128Typed(const void* sendbuff, v

bool ncclReduceScatterDdaFabricLL128Eligible(ncclComm* comm, const void* sendbuff, void* recvbuff, size_t recvcount,
Comment thread
pvallem marked this conversation as resolved.
Resolved
ncclDataType_t datatype, ncclRedOp_t op) {
(void)sendbuff;
(void)recvbuff;
if (!rcclParamDdaLL()) {
return false;
}
Expand All @@ -125,6 +125,9 @@ bool ncclReduceScatterDdaFabricLL128Eligible(ncclComm* comm, const void* sendbuf
if (comm->ddaFabricMemHandler == nullptr || comm->ddaScratch == nullptr || comm->ddaPeerPtrsDev == nullptr) {
return false;
}
if (comm->ddaLLEpochDev == nullptr || comm->ddaLLEpochLen < 1) {
return false;
}
if (comm->nRanks < 2 || comm->nRanks > dda::common::kDdaMaxNranks) {
return false;
}
Expand All @@ -138,16 +141,27 @@ bool ncclReduceScatterDdaFabricLL128Eligible(ncclComm* comm, const void* sendbuf
return false;
}

const size_t bytes = recvcount * ncclTypeSize(datatype); // per-rank shard
// Payload is staged as 8-byte words, so the shard must be a whole number of
// words.
if (bytes % 8 != 0) {
const size_t shardBytes = recvcount * ncclTypeSize(datatype);

// The LL128 line format packs 16B-aligned data with no chunk straddling a
// line, so a partial 16B chunk has nowhere to go, and the 16B wire accesses
// need both user buffers aligned to match.
if (shardBytes % 16 != 0) {
return false;
}
if (!dda::common::ddaUserBuffers16ByteAligned(sendbuff, recvbuff)) {
return false;
}
if (bytes > kDdaLL128RsMaxBytes) {

// Wrap compares the full message (nRanks shards) against the arch table; keep
// that same total here so direct Eligible() callers agree with the selector.
if (shardBytes * (size_t)comm->nRanks > rcclDdaLL128Threshold(comm, ncclFuncReduceScatter)) {
return false;
Comment thread
nusislam marked this conversation as resolved.
Resolved
}
if (ddaLL128RsScratchSize(comm->nRanks) > comm->ddaScratchBytes) {

// The shard has to fit the slices one rank's slot holds.
const size_t slotWords = ddaLL128RsSlotWords(ddaBankSize(comm->ddaScratchBytes), comm->nRanks);
if (ddaLL128Slices(shardBytes) * (size_t)kDdaLL128WireWordsPerSlice > slotWords) {
return false;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -55,11 +55,20 @@ static ncclResult_t ncclReduceScatterDdaIpcTyped(const void* sendbuff, void* rec
void* peerPtrsDev = comm->ddaPeerPtrsDev;
T** d_ipcbuffs = reinterpret_cast<T**>(peerPtrsDev);

CUDACHECK(cudaMemcpyAsync(comm->ddaScratch, sendbuff, totalCount * sizeof(T), cudaMemcpyDeviceToDevice, stream));
const hipEvent_t stopEvent = rcclTakeAddonStopEvent(comm);
hipExtLaunchKernelGGL((dda::common::ddaReduceScatterIpc<T, kDdaNranks, false>), grid, block, 0, stream,
/*startEvent=*/nullptr, stopEvent, /*flags=*/0, d_ipcbuffs, static_cast<T*>(recvbuff),
recvcount, static_cast<const T*>(sendbuff), comm->rank, barrierHost);
const dda::common::DdaRsIpcStaging staging = dda::common::ddaReduceScatterIpcStaging(recvcount, sizeof(T));
if (staging == dda::common::DdaRsIpcStaging::HostPreCopy) {
CUDACHECK(cudaMemcpyAsync(comm->ddaScratch, sendbuff, totalCount * sizeof(T), cudaMemcpyDeviceToDevice, stream));
}
if (staging == dda::common::DdaRsIpcStaging::FusedInKernel) {
hipExtLaunchKernelGGL((dda::common::ddaReduceScatterIpc<T, kDdaNranks, false, true>), grid, block, 0, stream,
/*startEvent=*/nullptr, stopEvent, /*flags=*/0, d_ipcbuffs, static_cast<T*>(recvbuff),
recvcount, static_cast<const T*>(sendbuff), comm->rank, barrierHost);
} else {
hipExtLaunchKernelGGL((dda::common::ddaReduceScatterIpc<T, kDdaNranks, false, false>), grid, block, 0, stream,
/*startEvent=*/nullptr, stopEvent, /*flags=*/0, d_ipcbuffs, static_cast<T*>(recvbuff),
recvcount, static_cast<const T*>(sendbuff), comm->rank, barrierHost);
}
CUDACHECK(cudaGetLastError());

return ncclSuccess;
Expand Down Expand Up @@ -108,6 +117,10 @@ bool ncclReduceScatterDdaIpcEligible(ncclComm* comm, const void* sendbuff, void*
return false;
}

if (!dda::common::ddaUserBuffers16ByteAligned(sendbuff, recvbuff)) {
return false;
}

return true;
}

Expand Down
Loading
Loading