AI/rocm-systems
2
0
Garpu 0
Kode Masalah Tarik Permintaan Actions Packages Projects Rilis Wiki Kegiatan

Merge pull request #561 from edgargabriel/multi-rank-devel

Telusuri Sumber 
Multi rank devel
This commit is contained in:
Edgar Gabriel
2022-06-10 11:19:20 -05:00
committed by GitHub
orang tua 5cb2aca3d9 a87d61db2b
melakukan 45e611dffd
29 mengubah file dengan 1322 tambahan dan 123 penghapusan
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+5
Melihat File
@@ -8,6 +8,11 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https:
- Packages for test and benchmark executables on all supported OSes using CPack.
- Adding custom signal handler - opt-in with RCCL_ENABLE_SIGNALHANDLER=1
- Additional details provided if Binary File Descriptor library (BFD) is pre-installed
- Adding experimental support for using multiple ranks per device
- Requires using a new interface to create communicator (ncclCommInitRankMulti), please
refer to the interface documentation for details.
- To avoid potential deadlocks, user might have to set an environment variables increasing
the number of hardware queues (e.g. export GPU_MAX_HW_QUEUES=16)
### Removed
- Removed experimental clique-based kernels
src/bootstrap.cc
+4 -1
Melihat File
@@ -208,20 +208,23 @@ struct bootstrapState {
int cudaDev;
int rank;
int nranks;
int virtualId;
volatile uint32_t *abortFlag;
};
ncclResult_t bootstrapInit(ncclUniqueId * id, struct ncclComm* comm) {
int rank = comm->rank;
int nranks = comm->nRanks;
int virtualId = comm->virtualId;
struct bootstrapState* state;
NCCLCHECK(ncclCalloc(&state, 1));
state->rank = rank;
state->nranks = nranks;
state->abortFlag = comm->abortFlag;
state->virtualId = virtualId;
comm->bootstrap = state;
TRACE(NCCL_INIT, "rank %d nranks %d", rank, nranks);
TRACE(NCCL_INIT, "rank %d nranks %d virtualId %d", rank, nranks, virtualId);
// [RCCL] Register custom signal handlers if requested
RegisterSignalHandlers();
src/graph/connect.cc
+4 -1
Melihat File
@@ -19,8 +19,11 @@ ncclResult_t ncclTopoPreset(struct ncclComm* comm,
struct ncclTopoGraph* treeGraph, struct ncclTopoGraph* ringGraph,
struct ncclTopoRanks* topoRanks) {
int rank = comm->rank;
int localRanks = comm->topo->nodes[GPU].count;
int nChannels = comm->nChannels;
int localRanks = 0;
for (int i=0; i<comm->topo->nodes[GPU].count; i++) {
localRanks += comm->topo->nodes[GPU].nodes[i].gpu.nRanksPerGpu;
}
for (int c=0; c<nChannels; c++) {
struct ncclChannel* channel = comm->channels+c;
src/graph/paths.cc
+26 -12
Melihat File
@@ -268,7 +268,7 @@ ncclResult_t ncclTopoCheckP2p(struct ncclTopoSystem* system, int64_t id1, int64_
struct ncclTopoNode* intermediateNode = path->list[0]->remNode;
if (intermediateNode->type == GPU) {
intermediateIndex = intermediateNode - system->nodes[GPU].nodes;
if (intermediateRank) *intermediateRank = intermediateNode->gpu.rank;
if (intermediateRank) *intermediateRank = intermediateNode->gpu.rank[0];
}
}
@@ -404,7 +404,7 @@ ncclResult_t ncclTopoCheckGdr(struct ncclTopoSystem* system, int64_t busId, int
if (distance == PATH_PXN) {
// In case of PXN, use the intermediate GPU distance instead
int proxyRank, g;
NCCLCHECK(ncclTopoGetIntermediateRank(system, gpu->gpu.rank, netDev, &proxyRank));
NCCLCHECK(ncclTopoGetIntermediateRank(system, gpu->gpu.rank[0], netDev, &proxyRank));
NCCLCHECK(ncclTopoRankToIndex(system, proxyRank, &g));
struct ncclTopoNode* proxyGpu = system->nodes[GPU].nodes+g;
distance = proxyGpu->paths[NET][n].type;
@@ -437,7 +437,7 @@ ncclResult_t ncclTopoGetIntermediateRank(struct ncclTopoSystem* system, int rank
WARN("Could not find intermediate GPU between GPU rank %d and NIC %d\n", rank, netDev);
return ncclInternalError;
}
*intermediateRank = node->gpu.rank;
*intermediateRank = node->gpu.rank[0];
} else {
*intermediateRank = rank;
}
@@ -520,10 +520,10 @@ ncclResult_t ncclTopoComputePaths(struct ncclTopoSystem* system, struct ncclPeer
if (peerInfos == NULL) continue;
// Remove GPUs we can't talk to because of containers.
struct ncclPeerInfo* dstInfo = peerInfos+system->nodes[GPU].nodes[g].gpu.rank;
struct ncclPeerInfo* dstInfo = peerInfos+system->nodes[GPU].nodes[g].gpu.rank[0];
for (int p=0; p<system->nodes[GPU].count; p++) {
if (p == g) continue;
struct ncclPeerInfo* srcInfo = peerInfos+system->nodes[GPU].nodes[p].gpu.rank;
struct ncclPeerInfo* srcInfo = peerInfos+system->nodes[GPU].nodes[p].gpu.rank[0];
int shm;
NCCLCHECK(ncclTransports[TRANSPORT_SHM].canConnect(&shm, system, NULL, srcInfo, dstInfo));
int p2p;
@@ -556,7 +556,8 @@ ncclResult_t ncclTopoComputePaths(struct ncclTopoSystem* system, struct ncclPeer
pxnGpu = p;
int netDev;
NCCLCHECK(ncclTopoGetLocalNet(system, peerNode->gpu.rank, &netDev));
NCCLCHECK(ncclTopoGetLocalNet(system, peerNode->gpu.rank[0], &netDev));
// To ensure proper balancing, use preferably a local GPU which advertised that NIC as its preferred one.
if (netDev == netNode->id) break;
}
@@ -599,7 +600,12 @@ ncclResult_t ncclTopoTrimSystem(struct ncclTopoSystem* system, struct ncclComm*
domains[g] = std::min(domains[g], domains[p]);
}
}
if (gpu->gpu.rank == comm->rank) myDomain = domains[g];
for (int j=0; j<gpu->gpu.nRanksPerGpu; j++ ) {
if (gpu->gpu.rank[j] == comm->rank) {
myDomain = domains[g];
break;
}
}
}
int ngpus = system->nodes[GPU].count;
@@ -650,7 +656,7 @@ ncclResult_t ncclTopoTrimSystem(struct ncclTopoSystem* system, struct ncclComm*
int gdr, ret = 1;
int net;
for (int g = 0; g < system->nodes[GPU].count; g++) {
NCCLCHECK(ncclTopoGetLocalNet(system, system->nodes[GPU].nodes[g].gpu.rank, &net));
NCCLCHECK(ncclTopoGetLocalNet(system, system->nodes[GPU].nodes[g].gpu.rank[0], &net));
NCCLCHECK(ncclTopoCheckGdr(system, system->nodes[GPU].nodes[g].id, net, 1, &gdr));
if (!gdr) {
ret = 0;
@@ -677,12 +683,16 @@ ncclResult_t ncclTopoTrimSystem(struct ncclTopoSystem* system, struct ncclComm*
}
}
}
if (rcclParamEnableIntranet()) {
remove = 0;
system->type |= RCCL_TOPO_FORCE_INTRA;
}
comm->localRanks = system->nodes[GPU].count;
if (system->nodes[GPU].count == comm->nRanks && remove) {
comm->localRanks = 0;
for (int n=0; n<system->nodes[GPU].count; n++ ) {
comm->localRanks += system->nodes[GPU].nodes[n].gpu.nRanksPerGpu;
}
if (comm->localRanks == comm->nRanks && remove) {
for (int n=system->nodes[NET].count-1; n>=0; n--)
NCCLCHECK(ncclTopoRemoveNode(system, NET, n));
}
@@ -777,10 +787,14 @@ ncclResult_t ncclTopoGetNvbGpus(struct ncclTopoSystem* system, int rank, int* nr
int nvbGpus = 0;
for (int g=0; g<ngpus; g++) {
struct ncclTopoNode* gpu = system->nodes[GPU].nodes+g;
if (gpu->gpu.rank != rank) continue;
int j=0;
for ( ; j<gpu->gpu.nRanksPerGpu; j++ ){
if (gpu->gpu.rank[j] == rank) break;
}
if ( j == gpu->gpu.nRanksPerGpu ) continue;
for (int p=0; p<ngpus; p++) {
if (gpu->paths[GPU][p].type == PATH_NVB) {
(*ranks)[nvbGpus++] = system->nodes[GPU].nodes[p].gpu.rank;
(*ranks)[nvbGpus++] = system->nodes[GPU].nodes[p].gpu.rank[j];
}
}
}
src/graph/rome_models.cc
+5 -5
Melihat File
@@ -691,7 +691,7 @@ ncclResult_t parseGraph(const char* str, struct ncclTopoSystem* system, struct n
if (g == system->nodes[GPU].nodes[j].gpu.dev)
break;
if (j < ngpus)
graph->intra[nChannels*ngpus+r] = system->nodes[GPU].nodes[j].gpu.rank;
graph->intra[nChannels*ngpus+r] = system->nodes[GPU].nodes[j].gpu.rank[0];
else
return ncclInternalError;
}
@@ -781,7 +781,7 @@ ncclResult_t parseGraphLight(const char* str, struct ncclTopoSystem* system, str
break;
if (j < ngpus)
{
graph->treeBase[r][x] = system->nodes[GPU].nodes[j].gpu.rank;
graph->treeBase[r][x] = system->nodes[GPU].nodes[j].gpu.rank[0];
y=r;
}
else
@@ -913,15 +913,15 @@ ncclResult_t parseChordalRing(struct ncclTopoSystem* system, struct ncclTopoGrap
// find the first unsed GPU that is closest to NIC
int f, m;
for (f = 0; f < ngpus; f++) {
int j = 0; for (j = 0; j < n; j++) if(used[j] == system->nodes[GPU].nodes[f].gpu.rank) break;
int j = 0; for (j = 0; j < n; j++) if(used[j] == system->nodes[GPU].nodes[f].gpu.rank[0]) break;
if(j >= n) break;
}
for (int i = 0; i < ngpus; i++) {
int j = 0; for (j = 0; j < n; j++) if(used[j] == system->nodes[GPU].nodes[i].gpu.rank) break;
int j = 0; for (j = 0; j < n; j++) if(used[j] == system->nodes[GPU].nodes[i].gpu.rank[0]) break;
if (j < n) continue;
if (paths[i].count < paths[f].count) f = i;
}
for (m = 0; m<ngpus; m++) if (graph->intra[n*ngpus+m] == system->nodes[GPU].nodes[f].gpu.rank) break;
for (m = 0; m<ngpus; m++) if (graph->intra[n*ngpus+m] == system->nodes[GPU].nodes[f].gpu.rank[0]) break;
used[n] = graph->intra[n*ngpus+m];
for (int i = 0; i < ngpus; i++) intra[i] = graph->intra[n*ngpus+((i+m)%ngpus)];
for (int i = 0; i < ngpus; i++) graph->intra[n*ngpus+i] = intra[i];
src/graph/search.cc
+74 -24
Melihat File
@@ -186,9 +186,11 @@ static int cmpIntraScores(struct ncclGpuScore* scores, int count) {
static ncclResult_t getGpuIndex(struct ncclTopoSystem* system, int rank, int* index) {
for (int g=0; g<system->nodes[GPU].count; g++) {
if (system->nodes[GPU].nodes[g].gpu.rank == rank) {
*index = g;
return ncclSuccess;
for (int j=0; j<system->nodes[GPU].nodes[g].gpu.nRanksPerGpu; j++) {
if (system->nodes[GPU].nodes[g].gpu.rank[j] == rank) {
*index = g;
return ncclSuccess;
}
}
}
WARN("Could not find gpu rank %d", rank);
@@ -270,9 +272,13 @@ ncclResult_t ncclTopoReplayGetGpu(struct ncclTopoSystem* system, struct ncclTopo
if (graph->nChannels == 0) return ncclInternalError;
int ngpus = system->nodes[GPU].count;
int nextRank = graph->intra[(graph->nChannels-1)*ngpus+step+1];
for (int i=0; i<ngpus; i++) if (system->nodes[GPU].nodes[i].gpu.rank == nextRank) {
*g = i;
return ncclSuccess;
for (int i=0; i<ngpus; i++) {
for (int j=0; j<system->nodes[GPU].nodes[i].gpu.nRanksPerGpu; j++ ) {
if (system->nodes[GPU].nodes[i].gpu.rank[j] == nextRank) {
*g = i;
return ncclSuccess;
}
}
}
if (*g == -1) return ncclInternalError;
return ncclSuccess;
@@ -302,18 +308,26 @@ static int ncclTopoCountXGMI(struct ncclTopoSystem* system, struct ncclTopoGraph
int n = graph->intra[ngpus*c+((i+1)%ngpus)];
struct ncclTopoNode *node;
int j;
for (j=0; j<ngpus; j++)
if (system->nodes[GPU].nodes[j].gpu.rank == g) break;
for (j=0; j<ngpus; j++) {
bool found=false;
for (int k=0; k<system->nodes[GPU].nodes[j].gpu.nRanksPerGpu; k++) {
if (system->nodes[GPU].nodes[j].gpu.rank[k] == g)
found = true;
}
if (found) break;
}
if (j<ngpus) {
node = system->nodes[GPU].nodes+j;
for (int k = 0; k<system->nodes[GPU].count; k++) {
if (node->paths[GPU][k].count == 1) {
struct ncclTopoLink* link = node->paths[GPU][k].list[0];
struct ncclTopoNode* remNode = link->remNode;
if (remNode->gpu.rank == n) {
if (link->type == LINK_NVL)
count ++;
}
for (int l=0; l<remNode->gpu.nRanksPerGpu; l++) {
if (remNode->gpu.rank[l] == n) {
if (link->type == LINK_NVL)
count ++;
}
}
}
}
}
@@ -412,7 +426,7 @@ ncclResult_t ncclTopoSearchRecGpu(struct ncclTopoSystem* system, struct ncclTopo
graph->nChannels--;
return ncclSuccess;
}
graph->intra[graph->nChannels*ngpus+step] = gpu->gpu.rank;
graph->intra[graph->nChannels*ngpus+step] = gpu->gpu.rank[0];
int g = gpu - system->nodes[GPU].nodes;
if (step == backToNet) {
// first get back to NIC
@@ -669,7 +683,7 @@ ncclResult_t ncclTopoGetChannelFromXml(struct ncclXmlNode *xmlChannel, int c, st
} else if (strcmp(sub->name, "gpu") == 0) {
int rank = -1;
for (int g=0; g<ngpus; g++) {
if (system->nodes[GPU].nodes[g].gpu.dev == dev) rank = system->nodes[GPU].nodes[g].gpu.rank;
if (system->nodes[GPU].nodes[g].gpu.dev == dev) rank = system->nodes[GPU].nodes[g].gpu.rank[0];
}
if (rank == -1) {
WARN("XML Import Channel : dev %d not found.", dev);
@@ -730,7 +744,9 @@ ncclResult_t ncclTopoGetXmlFromChannel(struct ncclTopoGraph* graph, int c, struc
NCCLCHECK(xmlAddNode(xml, xmlChannel, "gpu", &node));
int dev = -1;
for (int i=0; i<ngpus; i++) {
if (system->nodes[GPU].nodes[i].gpu.rank == intra[g]) dev = system->nodes[GPU].nodes[i].gpu.dev;
for ( int j=0; j<system->nodes[GPU].nodes[i].gpu.nRanksPerGpu; j++ ) {
if (system->nodes[GPU].nodes[i].gpu.rank[j] == intra[g]) dev = system->nodes[GPU].nodes[i].gpu.dev;
}
}
if (dev == -1) {
WARN("XML Export Channel : rank %d not found.", intra[g]);
@@ -789,6 +805,27 @@ float speedArrayInter[] = { 48.0, 30.0, 24.0, 22.0, 18.0, 15.0, 12.0, 10.0, 9.0,
RCCL_PARAM(ModelMatchingDisable, "MODEL_MATCHING_DISABLE", 0);
NCCL_PARAM(CrossNic, "CROSS_NIC", 2);
static void ncclExpandMultiRank(ncclTopoSystem* system, struct ncclTopoGraph* graph)
{
// Expand the intra array to the multi-ranks per node scenario
int ngpus = system->nodes[GPU].count;
int intraCpy[MAXCHANNELS*NCCL_TOPO_MAX_NODES];
TRACE(NCCL_GRAPH, "TopoCompute: expanding intra array for multi-rank per GPU scenarios nChannels %d", graph->nChannels);
memcpy(intraCpy, graph->intra, ngpus*sizeof(int)*graph->nChannels);
int tk=0;
for (int n=0; n<graph->nChannels; n++ ) {
for (int i=0; i<ngpus; i++) {
for (int j=0; j<ngpus; j++) {
if (intraCpy[n*ngpus+i] == system->nodes[GPU].nodes[j].gpu.rank[0] ) {
for (int k=0; k<system->nodes[GPU].nodes[j].gpu.nRanksPerGpu; k++) {
graph->intra[tk++] = system->nodes[GPU].nodes[j].gpu.rank[k];
}
}
}
}
}
}
ncclResult_t ncclTopoCompute(ncclTopoSystem* system, struct ncclTopoGraph* graph) {
int ngpus = system->nodes[GPU].count;
graph->crossNic = ncclParamCrossNic();
@@ -813,7 +850,10 @@ ncclResult_t ncclTopoCompute(ncclTopoSystem* system, struct ncclTopoGraph* graph
NCCLCHECK(ncclTopoGetGraphFromXml(xml->nodes, system, graph, &nChannels));
INFO(NCCL_GRAPH, "Search %d : %d channels loaded from XML graph", graph->id, nChannels);
free(xml);
if (graph->nChannels > 0) return ncclSuccess;
if (graph->nChannels > 0) {
ncclExpandMultiRank(system, graph);
return ncclSuccess;
}
}
str = getenv("NCCL_RINGS");
@@ -826,17 +866,29 @@ ncclResult_t ncclTopoCompute(ncclTopoSystem* system, struct ncclTopoGraph* graph
} else if (!rcclParamModelMatchingDisable() && !graph->collNet) {
// try to match 8P6L
NCCLCHECK(parseChordalRing(system, graph));
if (graph->nChannels) return ncclSuccess;
if (graph->nChannels) {
ncclExpandMultiRank(system, graph);
return ncclSuccess;
}
// try to match Rome 4P2H
NCCLCHECK(parseRome4P2H(system, graph));
if (graph->nChannels) return ncclSuccess;
if (graph->nChannels) {
ncclExpandMultiRank(system, graph);
return ncclSuccess;
}
// try to match 1H16P
NCCLCHECK(parse1H16P(system, graph));
if (graph->nChannels) return ncclSuccess;
if (graph->nChannels) {
ncclExpandMultiRank(system, graph);
return ncclSuccess;
}
// try to match 4H4P
NCCLCHECK(parse4H4P(system, graph));
}
if (graph->nChannels) return ncclSuccess;
if (graph->nChannels) {
ncclExpandMultiRank(system, graph);
return ncclSuccess;
}
if ((graph->pattern == NCCL_TOPO_PATTERN_RING) && (system->type & RCCL_TOPO_4P2H_ROME) && (ngpus == system->nRanks)) {
// limit single node max channels when searching ring graph on Rome
@@ -866,7 +918,6 @@ ncclResult_t ncclTopoCompute(ncclTopoSystem* system, struct ncclTopoGraph* graph
while (speedArray[speedIndex] > system->maxWidth && speedIndex < nspeeds-1) speedIndex++;
tmpGraph.speedIntra = tmpGraph.speedInter = speedArray[speedIndex];
int64_t globalTimeout = NCCL_SEARCH_GLOBAL_TIMEOUT;
search:
int time = tmpGraph.sameChannels ? NCCL_SEARCH_TIMEOUT_SAMECHANNELS :
tmpGraph.pattern == NCCL_TOPO_PATTERN_TREE ? NCCL_SEARCH_TIMEOUT_TREE : NCCL_SEARCH_TIMEOUT;
@@ -902,7 +953,6 @@ search:
if (time != -1) globalTimeout += time;
else globalTimeout = NCCL_SEARCH_GLOBAL_TIMEOUT;
if (globalTimeout < 0 && graph->nChannels) goto done;
int maxTypeIntra = system->nodes[NET].count > 0 ? tmpGraph.typeInter : PATH_SYS;
if (tmpGraph.typeIntra < maxTypeIntra && (graph->nChannels == 0 || tmpGraph.typeIntra < graph->typeIntra)) {
tmpGraph.typeIntra += 1;
@@ -967,13 +1017,12 @@ done:
if (graph->nChannels == 0 && graph->collNet == 0) {
WARN("Could not find a path for pattern %d, falling back to simple order", graph->pattern);
for (int i=0; i<ngpus; i++) graph->intra[i] = system->nodes[GPU].nodes[i].gpu.rank;
for (int i=0; i<ngpus; i++) graph->intra[i] = system->nodes[GPU].nodes[i].gpu.rank[0];
graph->inter[0] = graph->inter[1] = 0;
graph->speedIntra = graph->speedInter = 0.1;
graph->typeIntra = graph->typeInter = PATH_SYS;
graph->nChannels = 1;
}
if (graph->speedIntra >= 25.0) {
int dupChannels = std::min(graph->nChannels*2, graph->maxChannels);
memcpy(graph->intra+graph->nChannels*ngpus, graph->intra, (dupChannels-graph->nChannels)*ngpus*sizeof(int));
@@ -983,6 +1032,7 @@ done:
graph->speedInter /= DIVUP(dupChannels, graph->nChannels);
graph->nChannels = dupChannels;
}
ncclExpandMultiRank(system, graph);
return ncclSuccess;
}
src/graph/topo.cc
+45 -18
Melihat File
@@ -117,7 +117,10 @@ ncclResult_t ncclTopoCreateNode(struct ncclTopoSystem* system, struct ncclTopoNo
n->links[0].remNode = n;
n->links[0].width = LOC_WIDTH;
n->gpu.dev = NCCL_TOPO_UNDEF;
n->gpu.rank = NCCL_TOPO_UNDEF;
for (int i=0; i<RCCL_TOPO_MAX_RANKS_PER_GPU; i++) {
n->gpu.rank[i] = NCCL_TOPO_UNDEF;
}
n->gpu.nRanksPerGpu = NCCL_TOPO_UNDEF;
n->gpu.cudaCompCap = NCCL_TOPO_UNDEF;
} else if (type == CPU) {
n->cpu.arch = NCCL_TOPO_UNDEF;
@@ -253,7 +256,15 @@ ncclResult_t ncclTopoConnectCpus(struct ncclTopoSystem* system) {
static ncclResult_t ncclTopoPrintRec(struct ncclTopoNode* node, struct ncclTopoNode* prevNode, char* line, int offset) {
if (node->type == GPU) {
sprintf(line+offset, "%s/%lX (%d)", topoNodeTypeStr[node->type], node->id, node->gpu.rank);
sprintf(line+offset, "%s/%lX (%d", topoNodeTypeStr[node->type], node->id, node->gpu.rank[0]);
int nextOffset;
int nextRank = 1;
while ( nextRank < node->gpu.nRanksPerGpu ) {
nextOffset = strlen(line);
sprintf(line+nextOffset, "/%d", node->gpu.rank[nextRank++]);
}
nextOffset = strlen(line);
sprintf(line+nextOffset, ")");
} else if (node->type == CPU) {
sprintf(line+offset, "%s/%lX (%d/%d/%d)", topoNodeTypeStr[node->type], node->id, node->cpu.arch, node->cpu.vendor, node->cpu.model);
} else if (node->type == PCI) {
@@ -373,7 +384,17 @@ ncclResult_t ncclTopoAddGpu(struct ncclXmlNode* xmlGpu, struct ncclTopoSystem* s
rcclHipDeviceArch_t arch;
NCCLCHECK(xmlGetAttrInt(xmlGpu, "arch", &arch.value));
memcpy(&gpu->gpu.arch, &arch.arch, sizeof(hipDeviceArch_t));
NCCLCHECK(xmlGetAttrInt(xmlGpu, "rank", &gpu->gpu.rank));
//NCCLCHECK(xmlGetAttrInt(xmlGpu, "rank", &gpu->gpu.rank));
const char *rankStr;
NCCLCHECK(xmlGetAttrStr(xmlGpu, "rank", &rankStr));
char *tmpStr;
char *token = strtok_r ( (char *)rankStr, ",", &tmpStr);
gpu->gpu.nRanksPerGpu = 0;
while (token != NULL && gpu->gpu.nRanksPerGpu < RCCL_TOPO_MAX_RANKS_PER_GPU) {
gpu->gpu.rank[gpu->gpu.nRanksPerGpu++] = atoi(token);
token = strtok_r(NULL, ",", &tmpStr);
}
NCCLCHECK(xmlGetAttrInt(xmlGpu, "dev", &gpu->gpu.dev));
NCCLCHECK(xmlGetAttrInt(xmlGpu, "gdr", &gpu->gpu.gdrSupport));
// Do not go any further, nvlinks will be added in a second pass
@@ -385,6 +406,7 @@ struct kvDict kvDictPciGen[] = {
{ "2.5 GT/s", 15 }, { "5 GT/s", 30 }, { "8 GT/s", 60 }, { "16 GT/s", 120 }, { "32 GT/s", 240 }, /* Kernel 5.6 and earlier */
{ "2.5 GT/s PCIe", 15 }, { "5.0 GT/s PCIe", 30 }, { "8.0 GT/s PCIe", 60 }, { "16.0 GT/s PCIe", 120 }, { "32.0 GT/s PCIe", 240 }, { "64.0 GT/s PCIe", 480 },
{ NULL, 60 /* Default fallback */ } }; // x100 Mbps per lane
ncclResult_t ncclTopoAddPci(struct ncclXmlNode* xmlPci, struct ncclTopoSystem* system, struct ncclTopoNode* parent) {
const char* str;
@@ -694,7 +716,8 @@ ncclResult_t ncclTopoGetSystem(struct ncclComm* comm, struct ncclTopoSystem** sy
NCCLCHECK(ncclTopoFillGpu(xml, busId, &node));
if (node == NULL) continue;
NCCLCHECK(xmlSetAttrInt(node, "keep", 1));
NCCLCHECK(xmlSetAttrInt(node, "rank", r));
//NCCLCHECK(xmlSetAttrInt(node, "rank", r));
NCCLCHECK(xmlSetOrAppendAttrInt(node, "rank", r));
NCCLCHECK(xmlInitAttrInt(node, "gdr", comm->peerInfo[r].gdrSupport));
}
}
@@ -795,18 +818,20 @@ NCCL_PARAM(IgnoreCpuAffinity, "IGNORE_CPU_AFFINITY", 0);
ncclResult_t ncclTopoGetCpuAffinity(struct ncclTopoSystem* system, int rank, cpu_set_t* affinity) {
struct ncclTopoNode* cpu = NULL, *gpu = NULL;
for (int g=0; g<system->nodes[GPU].count; g++) {
if (system->nodes[GPU].nodes[g].gpu.rank == rank) {
gpu = system->nodes[GPU].nodes+g;
// Find closer CPU
int cpuIndex = -1, minHops = 0;
for (int c=0; c<system->nodes[CPU].count; c++) {
int nHops = system->nodes[GPU].nodes[g].paths[CPU][c].count;
if (cpuIndex == -1 || nHops < minHops) {
cpuIndex = c;
minHops = nHops;
}
for (int j=0; j<system->nodes[GPU].nodes[g].gpu.nRanksPerGpu; j++) {
if (system->nodes[GPU].nodes[g].gpu.rank[j] == rank) {
gpu = system->nodes[GPU].nodes+g;
// Find closer CPU
int cpuIndex = -1, minHops = 0;
for (int c=0; c<system->nodes[CPU].count; c++) {
int nHops = system->nodes[GPU].nodes[g].paths[CPU][c].count;
if (cpuIndex == -1 || nHops < minHops) {
cpuIndex = c;
minHops = nHops;
}
}
cpu = system->nodes[CPU].nodes+cpuIndex;
}
cpu = system->nodes[CPU].nodes+cpuIndex;
}
}
if (cpu == NULL) {
@@ -876,9 +901,11 @@ ncclResult_t ncclTopoGetCompCap(struct ncclTopoSystem* system, int* ccMin, int*
ncclResult_t ncclTopoGetLocalRank(struct ncclTopoSystem* system, int rank, int* localRank) {
for (int g=0; g<system->nodes[GPU].count; g++) {
if (system->nodes[GPU].nodes[g].gpu.rank == rank) {
*localRank = g;
return ncclSuccess;
for ( int j=0; j<system->nodes[GPU].nodes[g].gpu.nRanksPerGpu; j++ ){
if (system->nodes[GPU].nodes[g].gpu.rank[j] == rank) {
*localRank = g;
return ncclSuccess;
}
}
}
WARN("Could not find local GPU with rank %d\n", rank);
src/graph/topo.h
+8 -4
Melihat File
@@ -105,6 +105,7 @@ struct ncclTopoLinkList {
#define RCCL_TOPO_16P1H 8
#define RCCL_TOPO_FORCE_INTRA 16
#define RCCL_TOPO_MAX_RANKS_PER_GPU 8
struct ncclTopoNode {
int type;
int64_t id;
@@ -112,7 +113,8 @@ struct ncclTopoNode {
union {
struct {
int dev; // NVML dev number
int rank;
int rank[RCCL_TOPO_MAX_RANKS_PER_GPU];
int nRanksPerGpu;
int cudaCompCap;
int gdrSupport;
int gcn;
@@ -192,9 +194,11 @@ static ncclResult_t ncclTopoIdToIndex(struct ncclTopoSystem* system, int type, i
static ncclResult_t ncclTopoRankToIndex(struct ncclTopoSystem* system, int rank, int* index) {
*index = -1;
for (int i=0; i<system->nodes[GPU].count; i++) {
if (system->nodes[GPU].nodes[i].gpu.rank == rank) {
*index = i;
return ncclSuccess;
for (int j=0; j<system->nodes[GPU].nodes[i].gpu.nRanksPerGpu; j++ ) {
if (system->nodes[GPU].nodes[i].gpu.rank[j] == rank) {
*index = i;
return ncclSuccess;
}
}
}
return ncclInternalError;
src/graph/xml.h
+19
Melihat File
@@ -176,6 +176,25 @@ static ncclResult_t xmlSetAttrInt(struct ncclXmlNode* node, const char* attrName
return ncclSuccess;
}
static ncclResult_t xmlSetOrAppendAttrInt(struct ncclXmlNode* node, const char* attrName, const int value) {
int index;
NCCLCHECK(xmlGetAttrIndex(node, attrName, &index));
if (index == -1) {
index = node->nAttrs++;
strncpy(node->attrs[index].key, attrName, MAX_STR_LEN);
node->attrs[index].key[MAX_STR_LEN] = '\0';
snprintf(node->attrs[index].value, MAX_STR_LEN, "%d", value);
node->attrs[index].value[MAX_STR_LEN] = '\0';
return ncclSuccess;
}
char *tmp = strdup(node->attrs[index].value);
snprintf(node->attrs[index].value, MAX_STR_LEN, "%s,%d", tmp, value);
node->attrs[index].value[MAX_STR_LEN] = '\0';
free (tmp);
return ncclSuccess;
}
static ncclResult_t xmlSetAttrFloat(struct ncclXmlNode* node, const char* attrName, const float value) {
int index;
NCCLCHECK(xmlGetAttrIndex(node, attrName, &index));
src/group.cc
+5 -2
Melihat File
@@ -34,6 +34,7 @@ struct ncclInitArgs {
int ndev;
ncclUniqueId commId;
int myrank;
int virtualId;
};
struct ncclCollArgs {
ncclComm_t comm;
@@ -58,11 +59,12 @@ thread_local struct ncclAsyncArgs ncclGroupArgs[MAX_ASYNC_OPS];
void* ncclAsyncThreadMain(void* args_) {
struct ncclAsyncArgs* args = (struct ncclAsyncArgs*)args_;
NCCLCHECKTHREAD(args->init.func(args->init.newcomm, args->init.ndev, args->init.commId, args->init.myrank, args->init.cudaDev));
NCCLCHECKTHREAD(args->init.func(args->init.newcomm, args->init.ndev, args->init.commId, args->init.myrank,
args->init.cudaDev, args->init.virtualId));
return args;
}
ncclResult_t ncclAsyncInit(ncclInitFunc_t func, ncclComm_t* newcomm, int ndev, ncclUniqueId commId, int myrank, int cudaDev) {
ncclResult_t ncclAsyncInit(ncclInitFunc_t func, ncclComm_t* newcomm, int ndev, ncclUniqueId commId, int myrank, int cudaDev, int virtualId) {
if (ncclGroupIndex >= MAX_ASYNC_OPS) {
WARN("Too many async operations in progress, max is %d", MAX_ASYNC_OPS);
return ncclAsyncErrCheck(ncclInvalidUsage);
@@ -76,6 +78,7 @@ ncclResult_t ncclAsyncInit(ncclInitFunc_t func, ncclComm_t* newcomm, int ndev, n
args->init.ndev = ndev;
memcpy(&args->init.commId, &commId, sizeof(commId));
args->init.myrank = myrank;
args->init.virtualId = virtualId;
return ncclSuccess;
}
src/include/comm.h
+1
Melihat File
@@ -112,6 +112,7 @@ struct ncclComm {
int64_t busId; // my PCI bus ID in int format
cpu_set_t cpuAffinity; // CPU affinity of the GPU
int WarpSize;
int virtualId;
int node;
int nNodes;
src/include/group.h
+2 -2
Melihat File
@@ -14,9 +14,9 @@
bool ncclAsyncMode();
ncclResult_t ncclAsyncErrCheck(ncclResult_t ret);
typedef ncclResult_t(*ncclInitFunc_t)(ncclComm_t* newcomm, int ndev, ncclUniqueId commId, int myrank, int cudaDev);
typedef ncclResult_t(*ncclInitFunc_t)(ncclComm_t* newcomm, int ndev, ncclUniqueId commId, int myrank, int cudaDev, int virtualId);
ncclResult_t ncclAsyncInit(ncclInitFunc_t func, ncclComm_t* newcomm, int ndev, ncclUniqueId commId, int myrank, int cudaDev);
ncclResult_t ncclAsyncInit(ncclInitFunc_t func, ncclComm_t* newcomm, int ndev, ncclUniqueId commId, int myrank, int cudaDev, int virtualId);
typedef ncclResult_t(*ncclCollFunc_t)(const void* sendbuff, void* recvbuff, size_t count,
ncclDataType_t type, ncclRedOp_t op, int root, ncclComm_t comm, hipStream_t stream);
src/include/transport.h
+1
Melihat File
@@ -40,6 +40,7 @@ struct ncclPeerInfo {
int64_t busId;
struct ncclComm* comm;
int cudaCompCap;
int virtualId;
};
#define CONNECT_SIZE 128
src/init.cc
+42 -13
Melihat File
@@ -380,7 +380,7 @@ NCCL_PARAM(AggChannelSize, "AGG_CHANNEL_SIZE", -2);
NCCL_PARAM(DisableGraphHelper, "GRAPH_HELPER_DISABLE", 0);
NCCL_PARAM(GraphRegister, "GRAPH_REGISTER", 0);
static ncclResult_t commAlloc(ncclComm_t* comret, int ndev, int rank) {
static ncclResult_t commAlloc(ncclComm_t* comret, int ndev, int rank, int virtualId) {
if (ndev < 1) {
WARN("invalid device count (%d) requested", ndev);
return ncclInvalidArgument;
@@ -402,6 +402,7 @@ static ncclResult_t commAlloc(ncclComm_t* comret, int ndev, int rank) {
comm->rank = comm->hostDevComm.rank = rank;
comm->nRanks = comm->hostDevComm.nRanks = ndev;
comm->virtualId = virtualId;
hipGetDevice(&comm->cudaDev);
NCCLCHECK(getBusId(comm->cudaDev, &comm->busId));
TRACE(NCCL_INIT,"comm %p rank %d nranks %d cudaDev %d busId %lx", comm, rank, ndev, comm->cudaDev, comm->busId);
@@ -524,6 +525,7 @@ static void showVersion() {
static ncclResult_t fillInfo(struct ncclComm* comm, struct ncclPeerInfo* info, uint64_t commHash) {
info->rank = comm->rank;
info->virtualId = comm->virtualId;
CUDACHECK(hipGetDevice(&info->cudaDev));
info->hostHash=getHostHash()+commHash;
info->pidHash=getPidHash()+commHash;
@@ -702,13 +704,30 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, ncclUniqueId* comm
NCCLCHECK(fillInfo(comm, comm->peerInfo+rank, commHash));
NCCLCHECK(bootstrapAllGather(comm->bootstrap, comm->peerInfo, sizeof(struct ncclPeerInfo)));
for (int i = 0; i < nranks; i++) {
if ((i != rank) && (comm->peerInfo[i].hostHash == comm->peerInfo[rank].hostHash) && (comm->peerInfo[i].busId == comm->peerInfo[rank].busId)) {
WARN("Duplicate GPU detected : rank %d and rank %d both on CUDA device %lx", rank, i, comm->peerInfo[rank].busId);
return ncclInvalidUsage;
//If virtualId == -1 multiRank support has not been requested by user, using original interface
if (comm->virtualId == -1) {
for (int i = 0; i < nranks; i++) {
if ((i != rank) && (comm->peerInfo[i].hostHash == comm->peerInfo[rank].hostHash) && (comm->peerInfo[i].busId == comm->peerInfo[rank].busId)) {
WARN("Duplicate GPU detected : rank %d and rank %d both on CUDA device %lx", rank, i, comm->peerInfo[rank].busId);
return ncclInvalidUsage;
}
}
}
else {
//Multiple ranks can use the same device, but need to have different virtualId's.
for (int i = 0; i < nranks; i++) {
for (int j=0; j < nranks; j++) {
if (j==i) continue;
if((comm->peerInfo[i].hostHash == comm->peerInfo[j].hostHash) &&
(comm->peerInfo[i].busId == comm->peerInfo[j].busId) &&
(comm->peerInfo[i].virtualId == comm->peerInfo[j].virtualId)) {
WARN("Duplicate virtualId detected : rank %d and rank %d both on GPU device %lx virtualId %d",
i, j, comm->peerInfo[rank].busId, comm->peerInfo[i].virtualId);
return ncclInvalidUsage;
}
}
}
}
// AllGather1 - end
// Topo detection / System graph creation
@@ -1235,7 +1254,7 @@ affinity_restore:
NCCL_PARAM(SetStackSize, "SET_STACK_SIZE", 0);
ncclResult_t ncclCommInitRankSync(ncclComm_t* newcomm, int nranks, ncclUniqueId commId, int myrank, int cudaDev) {
ncclResult_t ncclCommInitRankSync(ncclComm_t* newcomm, int nranks, ncclUniqueId commId, int myrank, int cudaDev, int virtualId) {
ncclResult_t res;
CUDACHECK(hipSetDevice(cudaDev));
@@ -1246,7 +1265,7 @@ ncclResult_t ncclCommInitRankSync(ncclComm_t* newcomm, int nranks, ncclUniqueId
//CUDACHECKIGNORE(hipDeviceSetLimit(hipLimitStackSize, maxLocalSizeBytes));
}
*newcomm = NULL;
NCCLCHECKGOTO(commAlloc(newcomm, nranks, myrank), res, cleanup);
NCCLCHECKGOTO(commAlloc(newcomm, nranks, myrank, virtualId), res, cleanup);
NCCLCHECKGOTO(initTransportsRank(*newcomm, &commId), res, cleanup);
NCCLCHECKGOTO(devCommSetup(*newcomm), res, cleanup);
@@ -1259,7 +1278,7 @@ cleanup:
return res;
}
static ncclResult_t ncclCommInitRankDev(ncclComm_t* newcomm, int nranks, ncclUniqueId commId, int myrank, int cudaDev) {
static ncclResult_t ncclCommInitRankDev(ncclComm_t* newcomm, int nranks, ncclUniqueId commId, int myrank, int cudaDev, int virtualId) {
ncclResult_t res;
char* env = getenv("NCCL_COMM_ID");
if (env && myrank == 0) {
@@ -1282,9 +1301,9 @@ static ncclResult_t ncclCommInitRankDev(ncclComm_t* newcomm, int nranks, ncclUni
}
if (ncclAsyncMode()) {
NCCLCHECKGOTO(ncclAsyncInit(ncclCommInitRankSync, newcomm, nranks, commId, myrank, cudaDev), res, end);
NCCLCHECKGOTO(ncclAsyncInit(ncclCommInitRankSync, newcomm, nranks, commId, myrank, cudaDev, virtualId), res, end);
} else {
NCCLCHECKGOTO(ncclCommInitRankSync(newcomm, nranks, commId, myrank, cudaDev), res, end);
NCCLCHECKGOTO(ncclCommInitRankSync(newcomm, nranks, commId, myrank, cudaDev, virtualId), res, end);
}
end:
@@ -1297,10 +1316,20 @@ ncclResult_t ncclCommInitRank(ncclComm_t* newcomm, int nranks, ncclUniqueId comm
NVTX3_FUNC_RANGE_IN(nccl_domain);
int cudaDev;
CUDACHECK(hipGetDevice(&cudaDev));
NCCLCHECK(ncclCommInitRankDev(newcomm, nranks, commId, myrank, cudaDev));
NCCLCHECK(ncclCommInitRankDev(newcomm, nranks, commId, myrank, cudaDev, -1));
return ncclSuccess;
}
NCCL_API(ncclResult_t, ncclCommInitRankMulti, ncclComm_t* newcomm, int nranks, ncclUniqueId commId, int myrank, int virtualId);
ncclResult_t ncclCommInitRankMulti(ncclComm_t* newcomm, int nranks, ncclUniqueId commId, int myrank, int virtualId) {
NVTX3_FUNC_RANGE_IN(nccl_domain);
int cudaDev;
CUDACHECK(hipGetDevice(&cudaDev));
NCCLCHECK(ncclCommInitRankDev(newcomm, nranks, commId, myrank, cudaDev, virtualId));
return ncclSuccess;
}
NCCL_API(ncclResult_t, ncclCommInitAll, ncclComm_t* comms, int ndev, const int* devlist);
ncclResult_t ncclCommInitAll(ncclComm_t* comms, int ndev, const int* devlist) {
NVTX3_FUNC_RANGE_IN(nccl_domain);
@@ -1315,7 +1344,7 @@ ncclResult_t ncclCommInitAll(ncclComm_t* comms, int ndev, const int* devlist) {
NCCLCHECK(ncclGroupStart());
for (int i=0; i<ndev; i++) {
// Ignore return codes .. we need to call ncclGroupEnd to clean up anyway
ncclCommInitRankDev(comms+i, ndev, uniqueId, i, devlist ? devlist[i] : i);
ncclCommInitRankDev(comms+i, ndev, uniqueId, i, devlist ? devlist[i] : i, -1);
}
NCCLCHECK(ncclGroupEnd());
return ncclSuccess;
src/nccl.h.in
+23
Melihat File
@@ -22,6 +22,7 @@
#define RCCL_BFLOAT16 1
#define RCCL_GATHER_SCATTER 1
#define RCCL_ALLTOALLV 1
#define RCCL_MULTIRANKPERGPU 1
#ifdef __cplusplus
extern "C" {
@@ -87,6 +88,28 @@ ncclResult_t ncclCommInitRank(ncclComm_t* comm, int nranks, ncclUniqueId commId
ncclResult_t pncclCommInitRank(ncclComm_t* comm, int nranks, ncclUniqueId commId, int rank);
/// @endcond
/*! @brief Creates a new communicator (multi thread/process version) allowing multiple ranks per device.
@details
rank must be between 0 and nranks-1 and unique within a communicator clique.
Each rank is associated to a HIP device, which has to be set before calling
ncclCommInitRankMulti.
Since this version of the function allows multiple ranks to utilize the same
HIP device, a unique virtualId per device has to be provided by each calling
rank.
ncclCommInitRankMulti implicitly syncronizes with other ranks, so it must be
called by different threads/processes or use ncclGroupStart/ncclGroupEnd.
@param[in]
comm ncclComm_t*
communicator struct pointer
*/
ncclResult_t ncclCommInitRankMulti(ncclComm_t* comm, int nranks, ncclUniqueId commId, int rank, int virtualId);
/// @cond include_hidden
ncclResult_t pncclCommInitRankMulti(ncclComm_t* comm, int nranks, ncclUniqueId commId, int rank, int virtualId);
/// @endcond
/*! @brief Creates a clique of communicators (single process version).
*
* @details This is a convenience function to create a single-process communicator clique.
src/transport/p2p.cc
+4 -1
Melihat File
@@ -92,10 +92,13 @@ ncclResult_t p2pCanConnect(int* ret, struct ncclTopoSystem* topo, struct ncclTop
int p2p;
if (hipDeviceCanAccessPeer(&p2p, cudaDev1, cudaDev2) != hipSuccess) {
INFO(NCCL_INIT|NCCL_P2P,"peer query failed between dev %d(=%lx) and dev %d(=%lx)",
cudaDev1, info1->busId, cudaDev2, info2->busId);
cudaDev1, info1->busId, cudaDev2, info2->busId);
*ret = 0;
return ncclSuccess;
}
if (p2p == 0 && cudaDev1 == cudaDev2 && info1->busId == info2->busId) {
p2p = 1;
}
#if defined(__HIP_PLATFORM_HCC__) || defined(__HCC__) || defined(__HIPCC__)
#else
test/SendRecv_SinglePairs.cpp
+6 -4
Melihat File
@@ -19,13 +19,15 @@ namespace RcclUnitTesting
OptionalColArgs options;
bool isCorrect = true;
int totalRanks = testBed.ev.maxGpus;
int numGpus = testBed.ev.maxGpus;
for (int rpg=0; rpg < 2 && isCorrect; ++rpg)
for (int isMultiProcess = 0; isMultiProcess <= 1 && isCorrect; ++isMultiProcess)
{
if (!(testBed.ev.processMask & (1 << isMultiProcess))) continue;
int const numProcesses = isMultiProcess ? totalRanks : 1;
testBed.InitComms(TestBed::GetDeviceIdsList(numProcesses, totalRanks), 1);
int ranksPerGpu = rpg == 0 ? 1 : testBed.ev.maxRanksPerGpu;
int totalRanks = numGpus * ranksPerGpu;
int const numProcesses = isMultiProcess ? numGpus : 1;
testBed.InitComms(TestBed::GetDeviceIdsList(numProcesses, numGpus, ranksPerGpu), 1);
for (int dataIdx = 0; dataIdx < dataTypes.size() && isCorrect; ++dataIdx)
for (int numIdx = 0; numIdx < numElements.size() && isCorrect; ++numIdx)
test/common/EnvVars.cpp
+16 -14
Melihat File
@@ -32,12 +32,13 @@ namespace RcclUnitTesting
hsa_iterate_agents(CountGpus, &numDevicesAvailable);
hsa_shut_down();
showNames = GetEnvVar("UT_SHOW_NAMES" , 1);
minGpus = GetEnvVar("UT_MIN_GPUS" , 2);
maxGpus = GetEnvVar("UT_MAX_GPUS" , numDevicesAvailable);
processMask = GetEnvVar("UT_PROCESS_MASK", UT_SINGLE_PROCESS | UT_MULTI_PROCESS);
verbose = GetEnvVar("UT_VERBOSE" , 0);
printValues = GetEnvVar("UT_PRINT_VALUES", 0);
showNames = GetEnvVar("UT_SHOW_NAMES" , 1);
minGpus = GetEnvVar("UT_MIN_GPUS" , 2);
maxGpus = GetEnvVar("UT_MAX_GPUS" , numDevicesAvailable);
processMask = GetEnvVar("UT_PROCESS_MASK", UT_SINGLE_PROCESS | UT_MULTI_PROCESS);
verbose = GetEnvVar("UT_VERBOSE" , 0);
printValues = GetEnvVar("UT_PRINT_VALUES", 0);
maxRanksPerGpu = GetEnvVar("UT_MAX_RANKS_PER_GPU", 2);
// Limit number of supported reduction operators to just ncclSum if only allReduce is built
#ifdef BUILD_ALLREDUCE_ONLY
@@ -139,14 +140,15 @@ namespace RcclUnitTesting
{
std::vector<std::pair<std::string, std::string>> supported =
{
std::make_pair("UT_SHOW_NAMES" , "Show test case names"),
std::make_pair("UT_MIN_GPUS" , "Minimum number of GPUs to use"),
std::make_pair("UT_MAX_GPUS" , "Maximum number of GPUs to use"),
std::make_pair("UT_PROCESS_MASK", "Whether to run single/multi process"),
std::make_pair("UT_VERBOSE" , "Show verbose unit test output"),
std::make_pair("UT_REDOPS" , "List of reduction ops to test"),
std::make_pair("UT_DATATYPES" , "List of datatypes to test"),
std::make_pair("UT_PRINT_VALUES", "Print array values (# of values to print, < 0 for all)")
std::make_pair("UT_SHOW_NAMES" , "Show test case names"),
std::make_pair("UT_MIN_GPUS" , "Minimum number of GPUs to use"),
std::make_pair("UT_MAX_GPUS" , "Maximum number of GPUs to use"),
std::make_pair("UT_PROCESS_MASK" , "Whether to run single/multi process"),
std::make_pair("UT_VERBOSE" , "Show verbose unit test output"),
std::make_pair("UT_REDOPS" , "List of reduction ops to test"),
std::make_pair("UT_DATATYPES" , "List of datatypes to test"),
std::make_pair("UT_MAX_RANKS_PER_GPU", "Maximum number of ranks using the same GPU"),
std::make_pair("UT_PRINT_VALUES" , "Print array values (# of values to print, < 0 for all)")
};
printf("================================================================================\n");
test/common/EnvVars.hpp
+1
Melihat File
@@ -24,6 +24,7 @@ namespace RcclUnitTesting
int processMask; // Filter single/multi process [UT_PROCESS_MASK]
bool verbose; // Show verbose TestBed output for debug [UT_VERBOSE]
int printValues; // Print out input/output/expected arrays [UT_PRINT_VALUES]
int maxRanksPerGpu; // Number of ranks using the same GPU [UT_MAX_RANKS_PER_GPU]
// Constructor that parses and collects environment variables
EnvVars();
test/common/TestBed.cpp
+44 -13
Melihat File
@@ -104,6 +104,12 @@ namespace RcclUnitTesting
}
}
//Determine number of unique GPUs being used.
std::set<int> unique_devices;
for (auto a: this->rankToDeviceMap)
unique_devices.insert(a);
bool useMulti = unique_devices.size() < this->rankToDeviceMap.size() ? true : false;
// Tell first rank to get ncclUniqueId
int getIdCmd = TestBedChild::CHILD_GET_UNIQUE_ID;
PIPE_WRITE(0, getIdCmd);
@@ -133,6 +139,9 @@ namespace RcclUnitTesting
// Send the number of collectives to be run per group call
PIPE_WRITE(childId, numCollectivesInGroup);
// Send whether to use MultiRank interfaces or not.
PIPE_WRITE(childId, useMulti);
// Send the GPUs this child uses
int const numGpus = deviceIdsPerProcess[childId].size();
PIPE_WRITE(childId, numGpus);
@@ -357,11 +366,23 @@ namespace RcclUnitTesting
}
std::vector<std::vector<int>> TestBed::GetDeviceIdsList(int const numProcesses,
int const numGpus)
int const numGpus)
{
return GetDeviceIdsList(numProcesses, numGpus, 1);
}
std::vector<std::vector<int>> TestBed::GetDeviceIdsList(int const numProcesses,
int const numGpus,
int const ranksPerGpu)
{
std::vector<std::vector<int>> result(numProcesses);
for (int i = 0; i < numGpus; i++)
result[i % numProcesses].push_back(i);
int ntasks = numProcesses == 1 ? numGpus : 1;
int k=0;
for (int i = 0; i < numProcesses; i++)
for (int j = 0; j < ntasks * ranksPerGpu; j++) {
result[i].push_back(k%numGpus);
k++;
}
return result;
}
@@ -372,11 +393,17 @@ namespace RcclUnitTesting
ncclRedOp_t const redOp,
int const root,
bool const inPlace,
bool const managedMem)
bool const managedMem,
int const ranksPerProc)
{
std::stringstream ss;
ss << (isMultiProcess ? "MP" : "SP") << " ";
ss << totalRanks << " ranks ";
ss << totalRanks;
if (ranksPerProc > 1)
ss << "(" << ranksPerProc << ") ";
else
ss << " ";
ss << "ranks ";
ss << ncclFuncNames[funcType] << " ";
ss << "(" << (inPlace ? "IP" : "OP") << "," << (managedMem ? "MM" : "GM") << ") ";
ss << ncclDataTypeNames[dataType] << " ";
@@ -430,14 +457,16 @@ namespace RcclUnitTesting
bool isCorrect = true;
// Sweep over the number of ranks
for (int totalRanks = ev.minGpus; totalRanks <= ev.maxGpus && isCorrect; ++totalRanks)
for (int ranksPerGpu=1; ranksPerGpu <= ev.maxRanksPerGpu; ranksPerGpu++)
for (int numGpus = ev.minGpus; numGpus <= ev.maxGpus && isCorrect; ++numGpus)
for (int isMultiProcess = 0; isMultiProcess <= 1 && isCorrect; ++isMultiProcess)
{
if (!(ev.processMask & (1 << isMultiProcess))) continue;
// Test either single process all GPUs, or 1 process per GPU
int const numProcesses = isMultiProcess ? totalRanks : 1;
this->InitComms(TestBed::GetDeviceIdsList(numProcesses, totalRanks));
int const numChildren = isMultiProcess ? numGpus : 1;
int const numRanks = numGpus*ranksPerGpu;
this->InitComms(TestBed::GetDeviceIdsList(numChildren, numGpus, ranksPerGpu));
for (int ftIdx = 0; ftIdx < funcTypes.size() && isCorrect; ++ftIdx)
for (int dtIdx = 0; dtIdx < dataTypes.size() && isCorrect; ++dtIdx)
@@ -448,10 +477,11 @@ namespace RcclUnitTesting
{
if (ev.showNames)
{
std::string name = this->GetTestCaseName(totalRanks, isMultiProcess,
std::string name = this->GetTestCaseName(numGpus, isMultiProcess,
funcTypes[ftIdx], dataTypes[dtIdx],
redOps[rdIdx], roots[rtIdx],
inPlaceList[ipIdx], managedMemList[mmIdx]);
inPlaceList[ipIdx], managedMemList[mmIdx],
ranksPerGpu);
INFO("%s\n", name.c_str());
}
@@ -460,7 +490,7 @@ namespace RcclUnitTesting
int numInputElements, numOutputElements;
CollectiveArgs::GetNumElementsForFuncType(funcTypes[ftIdx],
sortedN[neIdx],
totalRanks,
numRanks,
&numInputElements,
&numOutputElements);
optionalArgs.redOp = redOps[rdIdx];
@@ -486,10 +516,11 @@ namespace RcclUnitTesting
this->ValidateResults(isCorrect);
if (!isCorrect)
{
std::string name = this->GetTestCaseName(totalRanks, isMultiProcess,
std::string name = this->GetTestCaseName(numGpus, isMultiProcess,
funcTypes[ftIdx], dataTypes[dtIdx],
redOps[rdIdx], roots[rtIdx],
inPlaceList[ipIdx], managedMemList[mmIdx]);
inPlaceList[ipIdx], managedMemList[mmIdx],
ranksPerGpu);
ERROR("Incorrect output for %s\n", name.c_str());
}
}
test/common/TestBed.hpp
+5 -1
Melihat File
@@ -94,6 +94,9 @@ namespace RcclUnitTesting
std::vector<ncclDataType_t> const& GetAllSupportedDataTypes();
// Helper function that splits up GPUs to the given number of processes
static std::vector<std::vector<int>> GetDeviceIdsList(int const numProcesses,
int const numGpus,
int const ranksPerGpu);
static std::vector<std::vector<int>> GetDeviceIdsList(int const numProcesses,
int const numGpus);
@@ -105,7 +108,8 @@ namespace RcclUnitTesting
ncclRedOp_t const redOp,
int const root,
bool const inPlace,
bool const managedMem);
bool const managedMem,
int const ranksPerProc=1);
// Run a simple sweep
void RunSimpleSweep(std::vector<ncclFunc_t> const& funcTypes,
test/common/TestBedChild.cpp
+18 -4
Melihat File
@@ -126,6 +126,8 @@ namespace RcclUnitTesting
PIPE_READ(this->totalRanks);
PIPE_READ(this->rankOffset);
PIPE_READ(this->numCollectivesInGroup);
bool useMultiRankPerGpu;
PIPE_READ(useMultiRankPerGpu);
// Read the GPUs this child uses and prepare storage for collective args / datasets
int numGpus;
@@ -166,11 +168,23 @@ namespace RcclUnitTesting
break;
}
if (ncclCommInitRank(&this->comms[localRank], this->totalRanks, id, globalRank) != ncclSuccess)
if (useMultiRankPerGpu)
{
ERROR("Rank %d on child %d unable to call ncclCommInitRank\n", globalRank, this->childId);
status = TEST_FAIL;
break;
if (ncclCommInitRankMulti(&this->comms[localRank], this->totalRanks, id, globalRank, globalRank) != ncclSuccess)
{
ERROR("Rank %d on child %d unable to call ncclCommInitRankMulti\n", globalRank, this->childId);
status = TEST_FAIL;
break;
}
}
else
{
if (ncclCommInitRank(&this->comms[localRank], this->totalRanks, id, globalRank) != ncclSuccess)
{
ERROR("Rank %d on child %d unable to call ncclCommInitRank\n", globalRank, this->childId);
status = TEST_FAIL;
break;
}
}
}
if (status == TEST_SUCCESS)
tools/MultiRank/Makefile
+24
Melihat File
@@ -0,0 +1,24 @@
# Copyright (c) 2020 Advanced Micro Devices, Inc. All rights reserved.
# Set to where RCCL is installed
RCCL_INSTALL=../../build/release
HIP_PATH?= $(wildcard /opt/rocm/hip)
ifeq (,$(HIP_PATH))
HIP_PATH=../../..
endif
HIPCC=$(HIP_PATH)/bin/hipcc
EXE=rccl-allreduce-multirank rccl-reducescatter-multirank
CXXFLAGS = -std=c++11 -O3 -I$(RCCL_INSTALL)/include/rccl/ -L$(RCCL_INSTALL) -lrccl
all: $(EXE)
rccl-allreduce-multirank: rccl-allreduce-multirank.cc $(shell find -regex ".*\.\hpp")
$(HIPCC) $(CXXFLAGS) $< -o $@
rccl-reducescatter-multirank: rccl-reducescatter-multirank.cc $(shell find -regex ".*\.\hpp")
$(HIPCC) $(CXXFLAGS) $< -o $@
clean:
rm -f *.o $(EXE)
tools/MultiRank/README.mpi
Executable
+13
Melihat File
@@ -0,0 +1,13 @@
#!/bin/bash
export MPI_INSTALL_PATH=
export RCCL_INSTALL_PATH=
export ROCM_INSTALL_PATH=
$MPI_INSTALL_PATH/bin/mpiCC -o rccl-allreduce-mpi-multirank rccl-allreduce-mpi-multirank.cc -I$ROCM_INSTALL_PATH/include -I$RCCL_INSTALL_PATH/include -D__HIP_PLATFORM_AMD__ -L$ROCM_INSTALL_PATH/lib -lamdhip64 -L$RCCL_INSTALL_PATH/lib -lrccl
$MPI_INSTALL_PATH/bin/mpiCC -o rccl-reducescatter-mpi-multirank rccl-reducescatter-mpi-multirank.cc -I$ROCM_INSTALL_PATH/include -I$RCCL_INSTALL_PATH/include -D__HIP_PLATFORM_AMD__ -L$ROCM_INSTALL_PATH/lib -lamdhip64 -L$RCCL_INSTALL_PATH/lib -lrccl
export GPU_MAX_HW_QUEUES=16
$MPI_INSTALL_PATH/bin/mpirun --mca pml ucx -np 4 ./rccl-allreduce-mpi-multirank 0 0 2
$MPI_INSTALL_PATH/bin/mpirun --mca pml ucx -np 4 ./rccl-reducescatter-mpi-multirank 0 0 2
tools/MultiRank/rccl-allreduce-mpi-multirank.cc
+222
Melihat File
@@ -0,0 +1,222 @@
/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
#include <stdio.h>
#include <unistd.h>
#include <stdint.h>
#include <stdlib.h>
#include "hip/hip_runtime.h"
#include "rccl.h"
#include "mpi.h"
#define MPICHECK(cmd) do { \
int e = cmd; \
if( e != MPI_SUCCESS ) { \
printf("Failed: MPI error %s:%d '%d'\n", \
__FILE__,__LINE__, e); \
exit(EXIT_FAILURE); \
} \
} while(0)
#define HIPCHECK(cmd) do { \
hipError_t e = cmd; \
if( e != hipSuccess ) { \
printf("Failed: HIP error %s:%d '%s'\n", \
__FILE__,__LINE__,hipGetErrorString(e)); \
exit(EXIT_FAILURE); \
} \
} while(0)
#define NCCLCHECK(cmd) do { \
ncclResult_t r = cmd; \
if (r!= ncclSuccess) { \
printf("Failed, NCCL error %s:%d '%s'\n", \
__FILE__,__LINE__,ncclGetErrorString(r)); \
exit(EXIT_FAILURE); \
} \
} while(0)
static void init_sendbuf (float *sendbuf, int count, int val)
{
for (int i = 0; i < count; i++) {
sendbuf[i] = (float)val+1;
}
}
static void init_zero (float *recvbuf, int count)
{
for (int i = 0; i < count; i++) {
recvbuf[i] = 0.0;
}
}
static bool check_recvbuf (float *recvbuf, int count, int ndevices)
{
bool result = true;
float expected=0.0;
for (int i=0; i<ndevices; i++){
expected += (float)i+1;
}
for (int i = 0; i < count; i++) {
if (recvbuf[i] != expected) {
result = false;
printf("Element %d is %f expected %f\n", i, recvbuf[i], expected);
break;
}
}
return result;
}
static void print_help()
{
printf("Usage: rccl-allreduce-mpi-multirank <distMode> <startDev> <numDevs> \n");
printf(" all arguments are optional, but have to be provided in this order\n");
printf(" distMode : 0 - 1 (default: 0 - block distribution of rank to devices)\n");
printf(" startDev : id of first Device to use (default: 0) \n");
printf(" numDevs : number of Devices to use (default: 2) \n");
}
static int distmode=0;
static int startdev=0;
static int numdevices=2;
static int maxdevices=0;
static void devicemode_init( int argc, char **argv)
{
char *modeexpl[4];
int myRank;
MPICHECK(MPI_Comm_rank (MPI_COMM_WORLD, &myRank));
modeexpl[0] = strdup("0: contiguous assignment of ranks to devices");
modeexpl[1] = strdup("1: round robin assignment of ranks to devices");
if (argc > 1 ) {
distmode = atoi(argv[1]);
}
if (argc > 2 ) {
startdev = atoi(argv[2]);
}
if ( argc > 3 ) {
numdevices = atoi(argv[3]);
}
if ( distmode > 1) {
if ( myRank == 0 ) {
printf("Unknown distribution mode %d. Known distribution modes are 0-1\n", distmode);
print_help();
}
MPI_Abort (MPI_COMM_WORLD, -1);
}
HIPCHECK(hipGetDeviceCount(&maxdevices));
if ( numdevices > maxdevices) {
if ( myRank == 0 ) {
printf("Requesting %d devices, %d devices available. Aborting.\n", numdevices, maxdevices);
print_help();
}
MPI_Abort (MPI_COMM_WORLD, -1);
}
if ( startdev > maxdevices-1) {
if ( myRank == 0 ) {
printf("Startdevice is %d, max. number of devices is %d. Valid values are 0 - %d\n", startdev, maxdevices, maxdevices-1);
print_help();
}
MPI_Abort (MPI_COMM_WORLD, -1);
}
if ( myRank == 0 ) {
printf("Using binding mode %s\n", modeexpl[distmode]);
printf("Starting devices is %d, %d devices used.\n\n", startdev, numdevices);
}
}
static bool report_binding=true;
static void device_set(int id, int nDev)
{
int dev=0;
if (distmode == 0 ) {
int tmp = (id*numdevices)/nDev;
dev = (startdev+tmp)%maxdevices;
}
else if (distmode == 1) {
dev = (startdev+id)%numdevices;
}
HIPCHECK(hipSetDevice(dev));
if (report_binding) {
printf("Rank %d using device %d\n", id, dev);
if ( id == nDev-1) {
report_binding=false;
}
}
}
int main(int argc, char* argv[])
{
int size = 32*1024*1024;
int myRank, nRanks, localRank = 0;
ncclUniqueId id;
ncclComm_t comm;
float *h_sendbuff, *h_recvbuff;
float *sendbuff, *recvbuff;
hipStream_t s;
//initializing MPI
MPICHECK(MPI_Init(&argc, &argv));
MPICHECK(MPI_Comm_rank(MPI_COMM_WORLD, &myRank));
MPICHECK(MPI_Comm_size(MPI_COMM_WORLD, &nRanks));
//get deviceId to be used for each rank, e.g. localRank%numberOfDevices
devicemode_init( argc, argv);
int nDev = numdevices;
//get NCCL unique ID at rank 0 and broadcast it to all others
if (myRank == 0) ncclGetUniqueId(&id);
MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));
//initializing RCCL
device_set(myRank, nRanks);
NCCLCHECK(ncclCommInitRankMulti(&comm, nRanks, id, myRank, myRank));
//allocate buffers
HIPCHECK(hipMalloc(&sendbuff, size * sizeof(float)));
h_sendbuff = (float*) malloc ( size *sizeof(float));
init_sendbuf(h_sendbuff, size, myRank);
HIPCHECK(hipMemcpy(sendbuff, h_sendbuff, size * sizeof(float), hipMemcpyDefault));
HIPCHECK(hipMalloc(&recvbuff, size * sizeof(float)));
h_recvbuff = (float*) malloc ( size *sizeof(float));
init_zero(h_recvbuff, size);
HIPCHECK(hipMemcpy(recvbuff, h_recvbuff, size * sizeof(float), hipMemcpyDefault));
HIPCHECK(hipStreamCreate(&s));
NCCLCHECK(ncclAllReduce((const void*)sendbuff, (void*)recvbuff, size, ncclFloat,
ncclSum, comm, s));
//completing NCCL operation by synchronizing on the HIP stream
HIPCHECK(hipStreamSynchronize(s));
//check result
HIPCHECK(hipMemcpy(h_recvbuff, recvbuff, size*sizeof(float), hipMemcpyDefault));
bool res = check_recvbuf(h_recvbuff, size, nRanks);
printf("[%d] Checking buffer result is %s\n", myRank, res == true ? "correct" : "wrong" );
//free buffers
HIPCHECK(hipFree(sendbuff));
free (h_sendbuff);
HIPCHECK(hipFree(recvbuff));
free (h_recvbuff);
//finalizing NCCL
ncclCommDestroy(comm);
HIPCHECK(hipStreamDestroy(s));
//finalizing MPI
printf("[MPI Rank %d] Success \n", myRank);
MPICHECK(MPI_Finalize());
return 0;
}
tools/MultiRank/rccl-allreduce-multirank.cc
+237
Melihat File
@@ -0,0 +1,237 @@
/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
#include <stdio.h>
#include "hip/hip_runtime.h"
#include "rccl.h"
#define HIPCHECK(cmd) do { \
hipError_t e = cmd; \
if( e != hipSuccess ) { \
printf("Failed: HIP error %s:%d '%s'\n", \
__FILE__,__LINE__,hipGetErrorString(e)); \
exit(EXIT_FAILURE); \
} \
} while(0)
#define NCCLCHECK(cmd) do { \
ncclResult_t r = cmd; \
if (r!= ncclSuccess) { \
printf("Failed, NCCL error %s:%d '%s'\n", \
__FILE__,__LINE__,ncclGetErrorString(r)); \
exit(EXIT_FAILURE); \
} \
} while(0)
static void init_sendbuf (float *sendbuf, int count, int val)
{
for (int i = 0; i < count; i++) {
sendbuf[i] = (float)val;
}
}
static void init_zero (float *recvbuf, int count)
{
for (int i = 0; i < count; i++) {
recvbuf[i] = 0.0;
}
}
static bool check_recvbuf (float *recvbuf, int count, int ndevices)
{
bool result = true;
float expected=0.0;
for (int i=0; i<ndevices; i++){
expected += (float)i+1;
}
for (int i = 0; i < count; i++) {
if (recvbuf[i] != expected) {
result = false;
printf("Element %d is %f expected %f\n", i, recvbuf[i], expected);
break;
}
}
return result;
}
static int distmode=0;
static int startdev=0;
static int numdevices=2;
static int maxdevices=0;
static int ranksperdev=1;
static void print_help()
{
printf("Usage: rccl-allreduce-multirank <distMode> <startDev> <numDevs> <ranksPerDev> \n");
printf(" all arguments are optional, but have to be provided in this order\n");
printf(" distMode : 0 - 3 (default: 0 - all ranks are on different devices)\n");
printf(" startDev : id of first Device to use (default: 0) \n");
printf(" numDevs : number of Devices to use (default: 2) \n");
printf(" ranksPerDev: number of Ranks per Device (default: 1) \n");
}
static void devicemode_init( int argc, char **argv)
{
char *modeexpl[4];
modeexpl[0] = strdup("0: all ranks are on different devices");
modeexpl[1] = strdup("1: all ranks are on same device");
modeexpl[2] = strdup("2: contiguous assignment of ranks to devices");
modeexpl[3] = strdup("3: round robin assignment of ranks to devices");
if (argc > 1 ) {
distmode = atoi(argv[1]);
}
if (argc > 2 ) {
startdev = atoi(argv[2]);
}
if ( argc > 3 ) {
numdevices = atoi(argv[3]);
}
if ( argc > 4 ) {
ranksperdev = atoi(argv[4]);
}
if ( distmode > 3) {
printf("Unknown distribution mode %d. Known distribution modes are 0-3\n", distmode);
print_help();
exit(-1);
}
HIPCHECK(hipGetDeviceCount(&maxdevices));
if ( numdevices > maxdevices) {
printf("Requesting %d devices, %d devices available. Aborting.\n", numdevices, maxdevices);
print_help();
exit(-1);
}
if ( startdev > maxdevices-1) {
printf("Startdevice is %d, max. number of devices is %d. Valid values are 0 - %d\n", startdev, maxdevices, maxdevices-1);
print_help();
exit(-1);
}
if (distmode == 1) numdevices = 1;
if (distmode == 0) ranksperdev = 1;
printf("Using binding mode %s\n", modeexpl[distmode]);
printf("Starting devices is %d, %d devices used, %d ranks per device.\n\n", startdev, numdevices, ranksperdev);
}
static bool report_binding=true;
static void device_set(int id, int nDev)
{
int dev=0;
if (distmode == 0 )
dev = (startdev+id)%numdevices;
else if (distmode == 1) {
dev = startdev;
}
else if (distmode == 2) {
int tmp = (id*numdevices)/nDev;
dev = (startdev+tmp)%maxdevices;
}
else if (distmode == 3) {
dev = (startdev+id)%numdevices;
}
HIPCHECK(hipSetDevice(dev));
if (report_binding) {
printf("Rank %d using device %d\n", id, dev);
if ( id == nDev-1) {
report_binding=false;
}
}
}
int main(int argc, char* argv[])
{
int nDev;
int size = 32*1024*1024;
devicemode_init( argc, argv);
nDev = numdevices * ranksperdev;
//allocating and initializing device buffers
float** h_sendbuff = (float**)malloc(nDev * sizeof(float*));
float** sendbuff = (float**)malloc(nDev * sizeof(float*));
float** h_recvbuff = (float**)malloc(nDev * sizeof(float*));
float** recvbuff = (float**)malloc(nDev * sizeof(float*));
hipStream_t* s = (hipStream_t*)malloc(sizeof(hipStream_t)*nDev);
ncclComm_t* comms = (ncclComm_t*)malloc(sizeof(ncclComm_t)*nDev);
for (int i = 0; i < nDev; ++i) {
device_set(i, nDev);
HIPCHECK(hipMalloc(sendbuff+i, size * sizeof(float)));
h_sendbuff[i] = (float*) malloc (size *sizeof(float));
init_sendbuf(h_sendbuff[i], size, i+1);
HIPCHECK(hipMemcpy(sendbuff[i], h_sendbuff[i], size * sizeof(float), hipMemcpyDefault));
HIPCHECK(hipMalloc(recvbuff+i, size*sizeof(float)));
h_recvbuff[i] = (float*) malloc (size *sizeof(float));
HIPCHECK(hipMemset(recvbuff[i], 0, size*sizeof(float)));
HIPCHECK(hipStreamCreate(s+i));
}
//initializing NCCL
ncclUniqueId id;
ncclGetUniqueId(&id);
NCCLCHECK(ncclGroupStart());
for (int i=0; i<nDev; i++ ) {
device_set(i, nDev);
NCCLCHECK(ncclCommInitRankMulti(&comms[i], nDev, id, i, i));
}
NCCLCHECK(ncclGroupEnd());
//calling NCCL communication API. Group API is required when using
//multiple devices per thread
NCCLCHECK(ncclGroupStart());
for (int i = 0; i < nDev; ++i) {
NCCLCHECK(ncclAllReduce((const void*)sendbuff[i], (void*)recvbuff[i], size, ncclFloat, ncclSum, comms[i], s[i]));
}
NCCLCHECK(ncclGroupEnd());
//synchronizing on HIP streams to wait for completion of NCCL operation
for (int i = 0; i < nDev; ++i) {
device_set(i, nDev);
HIPCHECK(hipStreamSynchronize(s[i]));
}
for (int i = 0; i < nDev; ++i) {
device_set(i, nDev);
HIPCHECK(hipMemcpy(h_recvbuff[i], recvbuff[i], size*sizeof(float), hipMemcpyDefault ));
bool res = check_recvbuf(h_recvbuff[i], size, nDev);
printf("Checking buffer %d result is %s\n",i, res == true ? "correct" : "wrong" );
}
//free buffers
for (int i = 0; i < nDev; ++i) {
device_set(i, nDev);
HIPCHECK(hipFree(sendbuff[i]));
free (h_sendbuff[i]);
HIPCHECK(hipFree(recvbuff[i]));
free (h_recvbuff[i]);
}
//finalizing NCCL
for(int i = 0; i < nDev; ++i) {
ncclCommDestroy(comms[i]);
HIPCHECK(hipStreamDestroy(s[i]));
}
free (h_sendbuff);
free (sendbuff);
free (h_recvbuff);
free (recvbuff);
free (s);
free (comms);
printf("Success \n");
return 0;
}
tools/MultiRank/rccl-reducescatter-mpi-multirank.cc
+226
Melihat File
@@ -0,0 +1,226 @@
/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
#include <stdio.h>
#include <unistd.h>
#include <stdint.h>
#include <stdlib.h>
#include "hip/hip_runtime.h"
#include "rccl.h"
#include "mpi.h"
#define MPICHECK(cmd) do { \
int e = cmd; \
if( e != MPI_SUCCESS ) { \
printf("Failed: MPI error %s:%d '%d'\n", \
__FILE__,__LINE__, e); \
exit(EXIT_FAILURE); \
} \
} while(0)
#define HIPCHECK(cmd) do { \
hipError_t e = cmd; \
if( e != hipSuccess ) { \
printf("Failed: HIP error %s:%d '%s'\n", \
__FILE__,__LINE__,hipGetErrorString(e)); \
exit(EXIT_FAILURE); \
} \
} while(0)
#define NCCLCHECK(cmd) do { \
ncclResult_t r = cmd; \
if (r!= ncclSuccess) { \
printf("Failed, NCCL error %s:%d '%s'\n", \
__FILE__,__LINE__,ncclGetErrorString(r)); \
exit(EXIT_FAILURE); \
} \
} while(0)
static void init_sendbuf (float *sendbuf, int count, int val)
{
for (int i = 0; i < count; i++) {
sendbuf[i] = (float)val+1;
}
}
static void init_zero (float *recvbuf, int count)
{
for (int i = 0; i < count; i++) {
recvbuf[i] = 0.0;
}
}
static bool check_recvbuf (float *recvbuf, int count, int ndevices)
{
bool result = true;
float expected=0.0;
for (int i=0; i<ndevices; i++){
expected += (float)i+1;
}
for (int i = 0; i < count; i++) {
if (recvbuf[i] != expected) {
result = false;
printf("Element %d is %f expected %f\n", i, recvbuf[i], expected);
break;
}
}
return result;
}
static void print_help()
{
printf("Usage: rccl-reducescatter-mpi-multirank <distMode> <startDev> <numDevs> \n");
printf(" all arguments are optional, but have to be provided in this order\n");
printf(" distMode : 0 - 1 (default: 0 - block distribution of rank to devices)\n");
printf(" startDev : id of first Device to use (default: 0) \n");
printf(" numDevs : number of Devices to use (default: 2) \n");
}
static int distmode=0;
static int startdev=0;
static int numdevices=2;
static int maxdevices=0;
static void devicemode_init( int argc, char **argv)
{
char *modeexpl[4];
int myRank;
MPICHECK(MPI_Comm_rank (MPI_COMM_WORLD, &myRank));
modeexpl[0] = strdup("0: contiguous assignment of ranks to devices");
modeexpl[1] = strdup("1: round robin assignment of ranks to devices");
if (argc > 1 ) {
distmode = atoi(argv[1]);
}
if (argc > 2 ) {
startdev = atoi(argv[2]);
}
if ( argc > 3 ) {
numdevices = atoi(argv[3]);
}
if ( distmode > 1) {
if ( myRank == 0 ) {
printf("Unknown distribution mode %d. Known distribution modes are 0-1\n", distmode);
print_help();
}
MPI_Abort (MPI_COMM_WORLD, -1);
}
HIPCHECK(hipGetDeviceCount(&maxdevices));
if ( numdevices > maxdevices) {
if ( myRank == 0 ) {
printf("Requesting %d devices, %d devices available. Aborting.\n", numdevices, maxdevices);
print_help();
}
MPI_Abort (MPI_COMM_WORLD, -1);
}
if ( startdev > maxdevices-1) {
if ( myRank == 0 ) {
printf("Startdevice is %d, max. number of devices is %d. Valid values are 0 - %d\n", startdev, maxdevices, maxdevices-1);
print_help();
}
MPI_Abort (MPI_COMM_WORLD, -1);
}
if ( myRank == 0 ) {
printf("Using binding mode %s\n", modeexpl[distmode]);
printf("Starting devices is %d, %d devices used.\n\n", startdev, numdevices);
}
}
static bool report_binding=true;
static void device_set(int id, int nDev)
{
int dev=0;
if (distmode == 0 ) {
int tmp = (id*numdevices)/nDev;
dev = (startdev+tmp)%maxdevices;
}
else if (distmode == 1) {
dev = (startdev+id)%numdevices;
}
HIPCHECK(hipSetDevice(dev));
if (report_binding) {
printf("Rank %d using device %d\n", id, dev);
if ( id == nDev-1) {
report_binding=false;
}
}
}
int main(int argc, char* argv[])
{
int myRank, nRanks, localRank = 0;
ncclUniqueId id;
ncclComm_t comm;
float *h_sendbuff, *h_recvbuff;
float *sendbuff, *recvbuff;
hipStream_t s;
//initializing MPI
MPICHECK(MPI_Init(&argc, &argv));
MPICHECK(MPI_Comm_rank(MPI_COMM_WORLD, &myRank));
MPICHECK(MPI_Comm_size(MPI_COMM_WORLD, &nRanks));
//get deviceId to be used for each rank, e.g. localRank%numberOfDevices
devicemode_init( argc, argv);
int nDev = numdevices;
int sendsize = 32*1024*1024;
int recvsize = sendsize / nRanks;
//get NCCL unique ID at rank 0 and broadcast it to all others
if (myRank == 0) ncclGetUniqueId(&id);
MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));
//initializing RCCL
device_set(myRank, nRanks);
NCCLCHECK(ncclCommInitRankMulti(&comm, nRanks, id, myRank, myRank));
//allocate buffers
HIPCHECK(hipMalloc(&sendbuff, sendsize * sizeof(float)));
h_sendbuff = (float*) malloc ( sendsize *sizeof(float));
init_sendbuf(h_sendbuff, sendsize, myRank);
HIPCHECK(hipMemcpy(sendbuff, h_sendbuff, sendsize * sizeof(float), hipMemcpyDefault));
HIPCHECK(hipMalloc(&recvbuff, recvsize * sizeof(float)));
h_recvbuff = (float*) malloc ( recvsize *sizeof(float));
init_zero(h_recvbuff, recvsize);
HIPCHECK(hipMemcpy(recvbuff, h_recvbuff, recvsize * sizeof(float), hipMemcpyDefault));
HIPCHECK(hipStreamCreate(&s));
NCCLCHECK(ncclReduceScatter((const void*)sendbuff, (void*)recvbuff, recvsize, ncclFloat,
ncclSum, comm, s));
//completing NCCL operation by synchronizing on the HIP stream
HIPCHECK(hipStreamSynchronize(s));
//check result
HIPCHECK(hipMemcpy(h_recvbuff, recvbuff, recvsize*sizeof(float), hipMemcpyDefault));
bool res = check_recvbuf(h_recvbuff, recvsize, nRanks);
printf("[%d] Checking buffer result is %s\n", myRank, res == true ? "correct" : "wrong" );
//free buffers
HIPCHECK(hipFree(sendbuff));
free (h_sendbuff);
HIPCHECK(hipFree(recvbuff));
free (h_recvbuff);
//finalizing NCCL
ncclCommDestroy(comm);
HIPCHECK(hipStreamDestroy(s));
//finalizing MPI
printf("[MPI Rank %d] Success \n", myRank);
MPICHECK(MPI_Finalize());
return 0;
}
tools/MultiRank/rccl-reducescatter-multirank.cc
+238
Melihat File
@@ -0,0 +1,238 @@
/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
#include <stdio.h>
#include "hip/hip_runtime.h"
#include "rccl.h"
#define HIPCHECK(cmd) do { \
hipError_t e = cmd; \
if( e != hipSuccess ) { \
printf("Failed: HIP error %s:%d '%s'\n", \
__FILE__,__LINE__,hipGetErrorString(e)); \
exit(EXIT_FAILURE); \
} \
} while(0)
#define NCCLCHECK(cmd) do { \
ncclResult_t r = cmd; \
if (r!= ncclSuccess) { \
printf("Failed, NCCL error %s:%d '%s'\n", \
__FILE__,__LINE__,ncclGetErrorString(r)); \
exit(EXIT_FAILURE); \
} \
} while(0)
static void init_sendbuf (float *sendbuf, int count, int val)
{
for (int i = 0; i < count; i++) {
sendbuf[i] = (float)val;
}
}
static void init_zero (float *recvbuf, int count)
{
for (int i = 0; i < count; i++) {
recvbuf[i] = 0.0;
}
}
static bool check_recvbuf (float *recvbuf, int count, int ndevices)
{
bool result = true;
float expected=0.0;
for (int i=0; i<ndevices; i++){
expected += (float)i+1;
}
for (int i = 0; i < count; i++) {
if (recvbuf[i] != expected) {
result = false;
printf("Element %d is %f expected %f\n", i, recvbuf[i], expected);
break;
}
}
return result;
}
static int distmode=0;
static int startdev=0;
static int numdevices=2;
static int maxdevices=0;
static int ranksperdev=1;
static void print_help()
{
printf("Usage: rccl-reducescatter-multirank <distMode> <startDev> <numDevs> <ranksPerDev> \n");
printf(" all arguments are optional, but have to be provided in this order\n");
printf(" distMode : 0 - 3 (default: 0 - all ranks are on different devices)\n");
printf(" startDev : id of first Device to use (default: 0) \n");
printf(" numDevs : number of Devices to use (default: 2) \n");
printf(" ranksPerDev: number of Ranks per Device (default: 1) \n");
}
static void devicemode_init( int argc, char **argv)
{
char *modeexpl[4];
modeexpl[0] = strdup("0: all ranks are on different devices");
modeexpl[1] = strdup("1: all ranks are on same device");
modeexpl[2] = strdup("2: contiguous assignment of ranks to devices");
modeexpl[3] = strdup("3: round robin assignment of ranks to devices");
if (argc > 1 ) {
distmode = atoi(argv[1]);
}
if (argc > 2 ) {
startdev = atoi(argv[2]);
}
if ( argc > 3 ) {
numdevices = atoi(argv[3]);
}
if ( argc > 4 ) {
ranksperdev = atoi(argv[4]);
}
if ( distmode > 3) {
printf("Unknown distribution mode %d. Known distribution modes are 0-3\n", distmode);
print_help();
exit(-1);
}
HIPCHECK(hipGetDeviceCount(&maxdevices));
if ( numdevices > maxdevices) {
printf("Requesting %d devices, %d devices available. Aborting.\n", numdevices, maxdevices);
print_help();
exit(-1);
}
if ( startdev > maxdevices-1) {
printf("Startdevice is %d, max. number of devices is %d. Valid values are 0 - %d\n", startdev, maxdevices, maxdevices-1);
print_help();
exit(-1);
}
if (distmode == 1) numdevices = 1;
if (distmode == 0) ranksperdev = 1;
printf("Using binding mode %s\n", modeexpl[distmode]);
printf("Starting devices is %d, %d devices used, %d ranks per device.\n\n", startdev, numdevices, ranksperdev);
}
static bool report_binding=true;
static void device_set(int id, int nDev)
{
int dev=0;
if (distmode == 0 )
dev = (startdev+id)%numdevices;
else if (distmode == 1) {
dev = startdev;
}
else if (distmode == 2) {
int tmp = (id*numdevices)/nDev;
dev = (startdev+tmp)%maxdevices;
}
else if (distmode == 3) {
dev = (startdev+id)%numdevices;
}
HIPCHECK(hipSetDevice(dev));
if (report_binding) {
printf("Rank %d using device %d\n", id, dev);
if ( id == nDev-1) {
report_binding=false;
}
}
}
int main(int argc, char* argv[])
{
int nDev;
devicemode_init( argc, argv);
nDev = numdevices * ranksperdev;
int sendsize = 32*1024*1024;
int recvsize = sendsize / nDev;
//allocating and initializing device buffers
float** h_sendbuff = (float**)malloc(nDev * sizeof(float*));
float** sendbuff = (float**)malloc(nDev * sizeof(float*));
float** h_recvbuff = (float**)malloc(nDev * sizeof(float*));
float** recvbuff = (float**)malloc(nDev * sizeof(float*));
hipStream_t* s = (hipStream_t*)malloc(sizeof(hipStream_t)*nDev);
ncclComm_t* comms = (ncclComm_t*)malloc(sizeof(ncclComm_t)*nDev);
for (int i = 0; i < nDev; ++i) {
device_set(i, nDev);
HIPCHECK(hipMalloc(sendbuff+i, sendsize * sizeof(float)));
h_sendbuff[i] = (float*) malloc (sendsize *sizeof(float));
init_sendbuf(h_sendbuff[i], sendsize, i+1);
HIPCHECK(hipMemcpy(sendbuff[i], h_sendbuff[i], sendsize * sizeof(float), hipMemcpyDefault));
HIPCHECK(hipMalloc(recvbuff+i, recvsize*sizeof(float)));
h_recvbuff[i] = (float*) malloc (recvsize *sizeof(float));
HIPCHECK(hipMemset(recvbuff[i], 0, recvsize*sizeof(float)));
HIPCHECK(hipStreamCreate(s+i));
}
//initializing NCCL
ncclUniqueId id;
ncclGetUniqueId(&id);
NCCLCHECK(ncclGroupStart());
for (int i=0; i<nDev; i++ ) {
device_set(i, nDev);
NCCLCHECK(ncclCommInitRankMulti(&comms[i], nDev, id, i, i));
}
NCCLCHECK(ncclGroupEnd());
//calling NCCL communication API. Group API is required when using
//multiple devices per thread
NCCLCHECK(ncclGroupStart());
for (int i = 0; i < nDev; ++i) {
NCCLCHECK(ncclReduceScatter((const void*)sendbuff[i], (void*)recvbuff[i], recvsize, ncclFloat, ncclSum, comms[i], s[i]));
}
NCCLCHECK(ncclGroupEnd());
//synchronizing on HIP streams to wait for completion of NCCL operation
for (int i = 0; i < nDev; ++i) {
device_set(i, nDev);
HIPCHECK(hipStreamSynchronize(s[i]));
}
for (int i = 0; i < nDev; ++i) {
device_set(i, nDev);
HIPCHECK(hipMemcpy(h_recvbuff[i], recvbuff[i], recvsize*sizeof(float), hipMemcpyDefault ));
bool res = check_recvbuf(h_recvbuff[i], recvsize, nDev);
printf("Checking buffer %d result is %s\n",i, res == true ? "correct" : "wrong" );
}
//free buffers
for (int i = 0; i < nDev; ++i) {
device_set(i, nDev);
HIPCHECK(hipFree(sendbuff[i]));
free (h_sendbuff[i]);
HIPCHECK(hipFree(recvbuff[i]));
free (h_recvbuff[i]);
}
//finalizing RCCL
for(int i = 0; i < nDev; ++i) {
ncclCommDestroy(comms[i]);
HIPCHECK(hipStreamDestroy(s[i]));
}
free (h_sendbuff);
free (sendbuff);
free (h_recvbuff);
free (recvbuff);
free (s);
free (comms);
printf("Success \n");
return 0;
}
tools/topo_expl/include/model.h
+4 -4
Melihat File
@@ -69,7 +69,7 @@ public:
int rankToCudaDev(int rank) {
for (int i=0; i<getNumGpus(); i++) {
if (rank == systems[0]->nodes[GPU].nodes[i].gpu.rank)
if (rank == systems[0]->nodes[GPU].nodes[i].gpu.rank[0])
return systems[0]->nodes[GPU].nodes[i].gpu.dev;
}
return -1;
@@ -77,7 +77,7 @@ public:
int64_t getGpuBusId(int rank) {
for (int i=0; i<getNumGpus(); i++) {
if (rank == systems[0]->nodes[GPU].nodes[i].gpu.rank)
if (rank == systems[0]->nodes[GPU].nodes[i].gpu.rank[0])
return systems[0]->nodes[GPU].nodes[i].id;
}
return -1;
@@ -93,7 +93,7 @@ public:
void setRanks() {
for (int r=0; r<getNumGpus(); r++)
for (int i=0; i<getNumGpus(); i++)
systems[r]->nodes[GPU].nodes[i].gpu.rank += firstRank;
systems[r]->nodes[GPU].nodes[i].gpu.rank[0] += firstRank;
}
int p2pCanConnect(int device1, int device2) { return 1; }
@@ -133,4 +133,4 @@ public:
NetworkModel() : nRanks(0) {}
};
#endif
#endif