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 @@ -35,6 +35,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
@@ -1,9 +1,10 @@
/*************************************************************************
* 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 @@

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 @@
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 @@
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