AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

reverting change to RcclReplayer (#1657)

Browse Source 
[ROCm/rccl commit: 45e1c3f3e2]
This commit is contained in:
Tim
2025-04-23 15:36:46 -04:00
committed by GitHub
parent fb1fdef8e2
commit 38f91fa2c8
2 changed files with 53 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
projects/rccl/tools/RcclReplayer/rcclReplayer.cpp
+27 -26
View File
@@ -180,36 +180,31 @@ void dataToCsv(GroupCall const& gc, std::ofstream &datafile, double runTime)
void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls& cc) void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls& cc)
{ {
bool verbose = isFirstRank && (getenv("VERBOSE") != NULL);
cc.globalRankComms.clear(); cc.globalRankComms.clear();
cc.globalRankComms.resize(cc.numGlobalRanks); cc.globalRankComms.resize(cc.numGlobalRanks);
cc.groupCalls.clear(); cc.groupCalls.clear();
std::ifstream fp(logFilename, std::ios::binary); FILE* fp = fopen(logFilename, "r");
if (!fp) { if (!fp) {
printf("[ERROR] Unable to open file %s\n", logFilename); printf("[ERROR] Unable to open file %s\n", logFilename);
exit(-1); exit(-1);
} }
constexpr size_t size = sizeof(LineItem) + 1; char line[2048];
char line[size]; //size of collectivecall struct LineItem li;
int lineNum = 0; int lineNum = 0;
while (fp.getline(line, size)) while (fgets(line, 2048, fp)) {
{ ++lineNum;
LineItem li = *((LineItem*)line);
if (li.coll == 10 || li.coll == 11) //Ignore invalid lines and collectives
{ if (!ParseLineItem(line, li) || li.nRanks != cc.numGlobalRanks) continue;
continue;
}
// if need hostname will parse together with num of line in output
// Ignore invalid lines and collectives
if (li.nRanks != cc.numGlobalRanks) continue;
// Figure out commIdx for this globalrank // Figure out commIdx for this globalrank
int commIdx = -1; int commIdx = -1;
for (auto i = 0; i < cc.globalRankComms[li.globalRank].size(); i++) { for (auto i = 0; i < cc.globalRankComms[li.globalRank].size(); i++) {
if (cc.globalRankComms[li.globalRank][i] != li.comm) { if (!strcmp(cc.globalRankComms[li.globalRank][i].c_str(), li.comm)) {
commIdx = i; commIdx = i;
break; break;
} }
@@ -220,8 +215,8 @@ void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls
} }
TaskInfo taskInfo; TaskInfo taskInfo;
taskInfo.funcType = (ncclFunc_t) li.coll; taskInfo.funcType = GetFuncType(li.opName);
taskInfo.inPlace = li.sendbuff == li.recvbuff; taskInfo.inPlace = !strcmp(li.sendbuff, li.recvbuff);
taskInfo.count = li.count; taskInfo.count = li.count;
taskInfo.datatype = (ncclDataType_t) li.datatype; taskInfo.datatype = (ncclDataType_t) li.datatype;
taskInfo.op = (ncclRedOp_t) li.op; taskInfo.op = (ncclRedOp_t) li.op;
@@ -234,7 +229,7 @@ void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls
if (gc.opCount != li.opCount) continue; if (gc.opCount != li.opCount) continue;
if (gc.rankData.count(li.globalRank)) { if (gc.rankData.count(li.globalRank)) {
RankData& rd = gc.rankData[li.globalRank]; RankData& rd = gc.rankData[li.globalRank];
if (rd.commIdx != commIdx || rd.tasks.size() != li.nTasks) if (rd.commIdx != commIdx || rd.tasks.size() != li.task)
continue; continue;
rd.tasks.push_back(taskInfo); rd.tasks.push_back(taskInfo);
@@ -242,7 +237,7 @@ void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls
break; break;
} }
// Rank has no tasks - make sure this is task 0 // Rank has no tasks - make sure this is task 0
else if (li.nTasks == 0) { else if (li.task == 0) {
gc.rankData[li.globalRank].lineNum = lineNum; gc.rankData[li.globalRank].lineNum = lineNum;
gc.rankData[li.globalRank].commIdx = commIdx; gc.rankData[li.globalRank].commIdx = commIdx;
gc.rankData[li.globalRank].tasks.push_back(taskInfo); gc.rankData[li.globalRank].tasks.push_back(taskInfo);
@@ -253,7 +248,7 @@ void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls
// If no collectives were found, create new one // If no collectives were found, create new one
if (!found) { if (!found) {
if (li.nTasks != 0) { if (li.task != 0) {
if (isFirstRank) printf("[WARN] Was unable to find corresponding collective for line %d\n", lineNum); if (isFirstRank) printf("[WARN] Was unable to find corresponding collective for line %d\n", lineNum);
} }
@@ -265,7 +260,7 @@ void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls
cc.groupCalls.push_back(gc); cc.groupCalls.push_back(gc);
} }
} }
fp.close(); fclose(fp);
// Validate group calls // Validate group calls
// - For non Send/Recv, check that all ranks participate with same parameters count // - For non Send/Recv, check that all ranks participate with same parameters count
@@ -380,6 +375,18 @@ void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls
} }
} }
bool ParseLineItem(char const* line, LineItem& li)
{
return sscanf(line,
"%[^:]:%d:%d [%d] NCCL INFO %[^:]: opCount %x sendbuff %s "
"recvbuff %s count %lu datatype %d op %d root %d comm %s "
"[nranks=%d] stream %p task %d globalrank %d",
li.hostname, &li.pid, &li.tid, &li.cudaDev, li.opName,
&li.opCount, li.sendbuff, li.recvbuff,
&li.count, &li.datatype, &li.op, &li.root, li.comm,
&li.nRanks, &li.stream, &li.task, &li.globalRank) == 17;
}
double ReplayRccl(CollectiveCalls& cc, int groupIdx, int& numInvalid) double ReplayRccl(CollectiveCalls& cc, int groupIdx, int& numInvalid)
{ {
int numLocalRanks = cc.localRankComms.size(); int numLocalRanks = cc.localRankComms.size();
@@ -559,12 +566,6 @@ void ExecuteCollective(TaskInfo& task, ncclComm_t const& comm, hipStream_t strea
case ncclCollRecv: case ncclCollRecv:
NCCL_CALL(ncclRecv(task.outputGpu.ptr, task.count, task.datatype, task.root, comm, stream)); NCCL_CALL(ncclRecv(task.outputGpu.ptr, task.count, task.datatype, task.root, comm, stream));
break; break;
case ncclStartGroup:
NCCL_CALL(ncclGroupStart());
break;
case ncclEndGroup:
NCCL_CALL(ncclGroupEnd());
break;
default: default:
printf("Error: unsupported collective\n"); printf("Error: unsupported collective\n");
exit(1); exit(1);
projects/rccl/tools/RcclReplayer/rcclReplayer.hpp
+24 -26
View File
@@ -1,6 +1,5 @@
#pragma once #pragma once
#include <map> #include <map>
#include <chrono>
#include <cstring> #include <cstring>
#include <rccl/rccl.h> #include <rccl/rccl.h>
@@ -37,25 +36,23 @@
struct LineItem struct LineItem
{ {
int pid; char hostname[MPI_MAX_PROCESSOR_NAME];
int tid; int pid;
int cudaDev; int tid;
//int graph; int cudaDev;
int groupDepth; char opName[32];
int opCount;
int coll; char sendbuff[32];
uint64_t opCount; char recvbuff[32];
void* sendbuff; size_t count;
void* recvbuff; int datatype;
size_t count; int op;
int datatype; int root;
int op; char comm[32];
int root; int nRanks;
void* comm; void* stream;
int nRanks; int task;
void* stream; int globalRank;
int nTasks;
int globalRank;
}; };
// Enumeration of all collective functions currently supported // Enumeration of all collective functions currently supported
@@ -72,9 +69,7 @@ typedef enum
ncclCollAllToAllv, ncclCollAllToAllv,
ncclCollSend, ncclCollSend,
ncclCollRecv, ncclCollRecv,
ncclNumFuncs, ncclNumFuncs
ncclStartGroup = 10;
ncclEndGroup = 11;
} ncclFunc_t; } ncclFunc_t;
char const ncclFuncNames[ncclNumFuncs][32] = char const ncclFuncNames[ncclNumFuncs][32] =
@@ -149,8 +144,8 @@ struct RankData
struct GroupCall struct GroupCall
{ {
bool isValid; bool isValid;
uint64_t opCount; int opCount;
std::map<int, RankData> rankData; std::map<int, RankData> rankData;
}; };
@@ -158,7 +153,7 @@ struct CollectiveCalls
{ {
int numGlobalRanks; int numGlobalRanks;
int numGpusPerMpiRank; int numGpusPerMpiRank;
std::vector<std::vector<void*>> globalRankComms; // Set of comms used by each global rank std::vector<std::vector<std::string>> globalRankComms; // Set of comms used by each global rank
std::vector<GroupCall> groupCalls; // List of group calls for each global rank std::vector<GroupCall> groupCalls; // List of group calls for each global rank
int localGpuOffset; // First local GPU device idx for this MPI process int localGpuOffset; // First local GPU device idx for this MPI process
@@ -389,6 +384,9 @@ bool IsRootUsed(ncclFunc_t funcType) {
funcType == ncclCollGather || funcType == ncclCollScatter); funcType == ncclCollGather || funcType == ncclCollScatter);
} }
// parse the logs and assign them into lineItem
bool ParseLineItem(char const* line, LineItem& li);
// this covers grouping the logs based on opCount and task number, // this covers grouping the logs based on opCount and task number,
// validatation of the groupCalls for both non-send/recv collectives and send/recv // validatation of the groupCalls for both non-send/recv collectives and send/recv
void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls& collectiveCalls); void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls& collectiveCalls);