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

replayer update v0 (#1733)

Browse Source 
* First version of new replayer, with comments on future TODOs

* plus minor fixes for UT

* Updated format of recorder, especially in binary department, according to replayer's need
This commit is contained in:
Tim
2025-06-13 15:05:34 -04:00
committed by GitHub
parent 4486d091b8
commit ba97c9c18b
8 changed files with 768 additions and 1129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/collectives.cc
+8
View File
@@ -170,12 +170,14 @@ ncclResult_t ncclAllToAll_impl(const void* sendbuff, void* recvbuff, size_t coun
int nRanks;
NCCLCHECK(ncclCommCount(comm, &nRanks));
if (count == 0) return ncclSuccess;
if (!mscclIsCaller()) Recorder::instance().skip(true);
NCCLCHECK(ncclGroupStart());
for (int r=0; r<nRanks; r++) {
NCCLCHECK(ncclSend(((char*)sendbuff)+r*rankOffset, count, datatype, r, comm, stream));
NCCLCHECK(ncclRecv(((char*)recvbuff)+r*rankOffset, count, datatype, r, comm, stream));
}
NCCLCHECK(ncclGroupEnd());
if (!mscclIsCaller()) Recorder::instance().skip(false);
return ncclSuccess;
}
}
@@ -205,6 +207,7 @@ ncclResult_t ncclAllToAllv_impl(const void *sendbuff, const size_t sendcounts[],
int nRanks;
NCCLCHECK(ncclCommCount(comm, &nRanks));
if (!mscclIsCaller()) Recorder::instance().skip(true);
NCCLCHECK(ncclGroupStart());
for (int r=0; r<nRanks; r++) {
NCCLCHECK(ncclSend(
@@ -223,6 +226,7 @@ ncclResult_t ncclAllToAllv_impl(const void *sendbuff, const size_t sendcounts[],
stream));
}
NCCLCHECK(ncclGroupEnd());
if (!mscclIsCaller()) Recorder::instance().skip(false);
return ncclSuccess;
}
@@ -285,6 +289,7 @@ ncclResult_t ncclGather_impl(const void* sendbuff, void* recvbuff, size_t sendco
if (sendcount == 0) return ncclSuccess;
int rank;
NCCLCHECK(ncclCommUserRank(comm, &rank));
if (!mscclIsCaller()) Recorder::instance().skip(true);
NCCLCHECK(ncclGroupStart());
if (rank == root) {
for (int r=0; r<nRanks; r++)
@@ -292,6 +297,7 @@ ncclResult_t ncclGather_impl(const void* sendbuff, void* recvbuff, size_t sendco
}
NCCLCHECK(ncclSend(sendbuff, sendcount, datatype, root, comm, stream));
NCCLCHECK(ncclGroupEnd());
if (!mscclIsCaller()) Recorder::instance().skip(false);
return ncclSuccess;
}
@@ -374,6 +380,7 @@ ncclResult_t ncclScatter_impl(const void* sendbuff, void* recvbuff, size_t recvc
if (recvcount == 0) return ncclSuccess;
int rank;
NCCLCHECK(ncclCommUserRank(comm, &rank));
if (!mscclIsCaller()) Recorder::instance().skip(true);
NCCLCHECK(ncclGroupStart());
if (rank == root) {
for (int r=0; r<nRanks; r++)
@@ -381,6 +388,7 @@ ncclResult_t ncclScatter_impl(const void* sendbuff, void* recvbuff, size_t recvc
}
NCCLCHECK(ncclRecv(recvbuff, recvcount, datatype, root, comm, stream));
NCCLCHECK(ncclGroupEnd());
if (!mscclIsCaller()) Recorder::instance().skip(false);
return ncclSuccess;
}
src/group.cc
+12 -3
View File
@@ -98,7 +98,10 @@ ncclResult_t ncclAsyncJobComplete(struct ncclAsyncJob* job) {
NCCL_API(ncclResult_t, ncclGroupStart);
ncclResult_t ncclGroupStart_impl() {
NCCLCHECK(Recorder::instance().record(rrGroupStart, ncclGroupDepth));
if (!mscclIsCaller())
{
NCCLCHECK(Recorder::instance().record(rrGroupStart, ncclGroupDepth));
}
ncclResult_t ret = ncclSuccess;
NVTX3_FUNC_RANGE_IN(nccl_domain);
@@ -117,7 +120,10 @@ ncclResult_t ncclGroupStartInternal() {
NCCL_API(ncclResult_t, ncclGroupEnd);
ncclResult_t ncclGroupEnd_impl() {
NCCLCHECK(Recorder::instance().record(rrGroupEnd, ncclGroupDepth));
if (!mscclIsCaller())
{
NCCLCHECK(Recorder::instance().record(rrGroupEnd, ncclGroupDepth));
}
ncclResult_t ret = ncclSuccess;
NVTX3_FUNC_RANGE_IN(nccl_domain);
NCCLCHECKGOTO(ncclGroupEndInternal(), ret, exit);
@@ -128,7 +134,10 @@ exit:
NCCL_API(ncclResult_t, ncclGroupSimulateEnd, ncclSimInfo_t* simInfo);
ncclResult_t ncclGroupSimulateEnd(ncclSimInfo_t* simInfo) {
Recorder::instance().record(ncclGroupDepth, simInfo);
if (!mscclIsCaller())
{
Recorder::instance().record(ncclGroupDepth, simInfo);
}
ncclResult_t ret = ncclSuccess;
NVTX3_FUNC_RANGE_IN(nccl_domain);
NCCLCHECKGOTO(ncclGroupEndInternal(simInfo), ret, exit);
src/include/recorder.h
+11 -5
View File
@@ -90,17 +90,21 @@ struct rcclApiCall {
int graphCaptured = -1;
// explicit data from header
rcclCall_t type; // in adjacent to op Name ^
rcclCall_t type;
uint64_t opCount = 0;
const void* sendbuff = NULL;
void* recvbuff = NULL;
void* sendPtrBase = NULL;
void* recvPtrBase = NULL;
size_t sendPtrExtent = 0;
size_t recvPtrExtent = 0;
size_t count = 0;
ncclDataType_t datatype;
ncclRedOp_t op;
int root = -1;
int nRanks = -1;
ncclComm_t comm;
hipStream_t stream;
ncclComm_t comm = NULL;
hipStream_t stream = NULL;
int nTasks = -1;
int globalRank = -1;
uint64_t commId = 0;
@@ -112,14 +116,15 @@ struct rcclApiCall {
class Recorder {
private:
std::ofstream outputFile; //1 per process
std::ofstream outputFile; // 1 per process
int output_json = 0; // 0 is to binary, 1 to json
std::string filename;
int logLevel = -1;
//std::string hostname;
int pid = -1;
int numCall = 0; //for debugging only
int numCall = 0; // reserved for future record format/debug
bool skipped = false; // number of sendrecv calls to skip for gather/scatter/a2a(v)
static __thread int rcclReplayThreadIdx;
static int depth; // for indentation purpose, will need thread safty later
@@ -136,6 +141,7 @@ class Recorder {
public:
static Recorder& instance();
void skip(bool b);
void record(const char* name); // non-replayable calls
ncclResult_t record(rcclApiCall& call);
ncclResult_t record(rcclCall_t type, const ncclInfo& info); // collective
src/misc/recorder.cc
+53 -25
View File
@@ -43,14 +43,21 @@ rcclApiCall::rcclApiCall(rcclCall_t type, const ncclInfo& info)://name(rcclCallS
nRanks(info.comm->nRanks),
stream(info.stream),
nTasks(info.comm->planner.nTasksP2p + info.comm->planner.nTasksColl),
globalRank(info.comm->localRankToRank[info.comm->localRank]){}
globalRank(info.comm->localRankToRank[info.comm->localRank])
{
hipMemGetAddressRange(&recvPtrBase, &recvPtrExtent, const_cast<void*>(info.recvbuff)); // should always exist for collectives
if (info.sendbuff) // ncclSend/Recv
{
hipMemGetAddressRange(&sendPtrBase, &sendPtrExtent, const_cast<void*>(info.sendbuff));
}
}
rcclApiCall::rcclApiCall(rcclCall_t type) : type(type){}
std::string siminfo_fmt = "[size : %zu, magic : %u, version : %u, estimated time : %f, timestamp : %f]";
std::string config_fmt = ", ncclConfig : [size : %zu, magic : %u, version : %u, blocking : %d, cgaClusterSize : %d, minCTA : %d, maxCTA : %d, netname : %s, splitshare : %d]";
std::string ctxt_fmt = "time : %lf, thread : %d, device : %d, captured : %d, graphID : %llu ]]"; // implicit context info
std::string ubr_fmt = "%s : [comm : %p, buff : %p, returned handle : %p, size : %zu, context : [";
std::string ubr_fmt = "%s : [comm : %p, buff : [addr : %p, base : %p, size : %zu], returned handle : %p, count : %zu, context : [";
std::string getId_fmt = "%s : [uniqueID : %llu, context : [";
std::string ubDereg_fmt = "%s : [comm : %p, handle : %p, context : [";
std::string rank_fmt = "%s : [size : %d, uniqueID : %llu, rank : %d, context : [";
@@ -62,7 +69,7 @@ std::string alloc_fmt = "%s : [returned ptr : %p, size : %zu, context : [";
std::string free_fmt = "%s : [ptr : %p, context : [";
std::string redop_fmt = "%s : [scalar : %p, datatype : %d, op : %d, residence : %d, comm : %p, context : [";
std::string redopdestroy_fmt = "%s : [op : %d, comm : %p, context : [";
std::string coll_fmt = "%s : [opCount : %lx, sendbuff : %p, recvbuff : %p, count : %zu, datatype : %d, op : %d, root : %d, comm : %p, nranks : %d, stream : %p, task : %d, globalrank : %d, context : [";
std::string coll_fmt = "%s : [opCount : %lx, sendbuff : [addr : %p, base : %p, size : %zu], recvbuff : [addr : %p, base : %p, size : %zu], count : %zu, datatype : %d, op : %d, root : %d, comm : %p, nranks : %d, stream : %p, task : %d, globalrank : %d, context : [";
Recorder::Recorder()
{
@@ -74,8 +81,8 @@ Recorder::Recorder()
}
logLevel = getenv("RCCL_LOG_LEVEL") ? std::stoi(getenv("RCCL_LOG_LEVEL")) : 1;
char hostname[1024];
getHostName(hostname, 1024, '.');
char hostname[256];
gethostname(hostname, 256);
pid = getpid();
output_json = 0;
@@ -94,13 +101,13 @@ Recorder::Recorder()
output_name = std::string(filename);
}
outputFile.open(output_name + std::to_string(pid) + output_extension,
outputFile.open(output_name + "." + std::to_string(pid) + "." + std::string(hostname) + output_extension,
output_json ? std::ofstream::out : std::ofstream::binary);
if (output_json)
{
outputFile << "{" << std::endl;
indent(2, outputFile);
outputFile << "hostname : " << hostname << ", version : 0,";
outputFile << "version : 1,";
}
}
@@ -110,6 +117,14 @@ Recorder& Recorder::instance()
return _instance;
}
void Recorder::skip(bool b)
{
if (filename.size())
{
skipped = b;
}
}
void Recorder::captureGpuContext(rcclApiCall& call) const
{
call.timestamp = duration_cast<duration<double>>(high_resolution_clock::now().time_since_epoch()).count() * 1000;
@@ -171,7 +186,7 @@ void Recorder::write(const rcclApiCall &call)
case rrCommRegister:
{
len = snprintf(buffer, 4096, ubr_fmt.c_str(),
rcclCallStr[call.type], call.comm, call.sendbuff, call.recvbuff, call.count);
rcclCallStr[call.type], call.comm, call.sendbuff, call.sendPtrBase, call.sendPtrExtent, call.recvbuff, call.count);
break;
}
case rrCommDeregister:
@@ -240,7 +255,8 @@ void Recorder::write(const rcclApiCall &call)
}
default: // collectives
len = snprintf(buffer, 4096, coll_fmt.c_str(),
rcclCallStr[call.type], call.opCount, call.sendbuff, call.recvbuff, call.count, call.datatype,
rcclCallStr[call.type], call.opCount, call.sendbuff, call.sendPtrBase, call.sendPtrExtent,
call.recvbuff, call.recvPtrBase, call.recvPtrExtent, call.count, call.datatype,
call.op, call.root, call.comm, call.nRanks, call.stream, call.nTasks, call.globalRank);
}
@@ -249,7 +265,6 @@ void Recorder::write(const rcclApiCall &call)
outputFile.write(buffer, len);
} else {
outputFile.write((char*)&call, sizeof(rcclApiCall));
outputFile << std::endl;
}
outputFile.flush();
return ;
@@ -314,7 +329,6 @@ ncclResult_t Recorder::record(rcclApiCall& call)
#if ROCM_VERSION >= 60100
hipStreamCaptureStatus status;
hipGraph_t graphCaptured;
unsigned long long graphID = 0;
CUDACHECK(hipStreamGetCaptureInfo_v2(call.stream, &status, &(call.graphID), &graphCaptured)); // shouldnt we need dependency?
if (status == hipStreamCaptureStatusActive) // when graph launched this should be disabled
@@ -338,7 +352,7 @@ ncclResult_t Recorder::record(rcclApiCall& call)
ncclResult_t Recorder::record(rcclCall_t type, const ncclInfo& info)
{
if (!filename.size())
if (!filename.size() || skipped)
{
return ncclSuccess;
}
@@ -366,9 +380,9 @@ ncclResult_t Recorder::record(rcclCall_t type, const void* sendbuff, void* recvb
ncclResult_t ret = record(call);
if (type == rrAllToAllv)
{
int size = call.nRanks - 1;
if (output_json)
{
int size = call.nRanks - 1;
outputFile << ", sendcounts : [";
for (int i = 0; i < size; i++) outputFile << sendcounts[i] << ", ";
outputFile << sendcounts[size] << "], sdispls : [";
@@ -378,9 +392,13 @@ ncclResult_t Recorder::record(rcclCall_t type, const void* sendbuff, void* recvb
outputFile << recvcounts[size] << "], rdispls : [";
for (int i = 0; i < size; i++) outputFile << rdispls[i] << ", ";
outputFile << rdispls[size] << "]";
outputFile.flush();
} else {
outputFile.write((char*)sendcounts, sizeof(size_t) * (size + 1));
outputFile.write((char*)sdispls, sizeof(size_t) * (size + 1));
outputFile.write((char*)recvcounts, sizeof(size_t) * (size + 1));
outputFile.write((char*)rdispls, sizeof(size_t) * (size + 1));
}
// else export to binary
outputFile.flush();
}
return ret;
}
@@ -406,7 +424,7 @@ ncclResult_t Recorder::record(rcclCall_t type, ncclRedOp_t op, ncclComm_t comm,
ncclResult_t Recorder::record(rcclCall_t type, int groupDepth)
{
if (!filename.size())
if (!filename.size() || skipped)
{
return ncclSuccess;
}
@@ -467,6 +485,7 @@ ncclResult_t Recorder::record(rcclCall_t type, ncclComm_t comm, void* handle, vo
call.recvbuff = handle;
if (type == rrCommRegister)
{
CUDACHECK(hipMemGetAddressRange(&call.sendPtrBase, &call.sendPtrExtent, userBuffer));
call.sendbuff = userBuffer;
call.count = size;
}
@@ -503,6 +522,7 @@ void Recorder::record(int groupDepth, ncclSimInfo_t *siminfo)
siminfo->size, siminfo->magic, siminfo->version, siminfo->estimatedTime, call.timestamp);
outputFile.write(buffer, len);
} // no tid for groupCall
// TODO: else flush siminfo in binary
outputFile.flush();
}
@@ -529,6 +549,7 @@ void Recorder::record(rcclCall_t type, int size, int rank, ncclUniqueId* commId,
outputFile.write(buffer, len);
outputFile.flush();
}
// TODO: else flush ncclConfig in binary
}
void Recorder::record(ncclComm_t* comms, int ndev, const int* devlist)
@@ -540,15 +561,20 @@ void Recorder::record(ncclComm_t* comms, int ndev, const int* devlist)
rcclApiCall call(rrCommInitAll);
call.root = ndev;
// call.sendbuff = comms; TODO: might log this too
call.sendbuff = devlist;
record(call);
if (output_json && devlist)
if (devlist)
{
outputFile << ", devlist : [";
for (int i = 0; i < call.root - 1; i++)
outputFile << devlist[i] << ", ";
outputFile << devlist[call.root - 1] << "]";
if (output_json)
{
outputFile << ", devlist : [";
for (int i = 0; i < call.root - 1; i++)
outputFile << devlist[i] << ", ";
outputFile << devlist[call.root - 1] << "]";
} else {
outputFile.write((char*)devlist, sizeof(int) * ndev);
}
outputFile.flush();
}
}
@@ -578,6 +604,7 @@ static rcclCall_t getFuncType(std::string func)
void parseJsonEntry(const char* entry, std::vector<rcclApiCall>& calls)
{
// TODO: parse comma too
rcclApiCall call;
std::string str(entry);
size_t begin = str.find_first_not_of(' ');
@@ -588,7 +615,7 @@ void parseJsonEntry(const char* entry, std::vector<rcclApiCall>& calls)
case rrCommRegister:
{
assert(sscanf(str.c_str() + end + 3, (ubr_fmt.substr(5) + ctxt_fmt).c_str(),
&call.comm, &call.sendbuff, &call.recvbuff, &call.count) == 4);
&call.comm, &call.sendbuff, &call.sendPtrBase, &call.sendPtrExtent, &call.recvbuff, &call.count) == 6);
break;
}
case rrCommDeregister:
@@ -658,9 +685,10 @@ void parseJsonEntry(const char* entry, std::vector<rcclApiCall>& calls)
}
default:
assert(sscanf(str.c_str() + end + 3, (coll_fmt.substr(5) + ctxt_fmt).c_str(),
&call.opCount, &call.sendbuff, &call.recvbuff, &call.count, &call.datatype, &call.op, &call.root,
&call.opCount, &call.sendbuff, &call.sendPtrBase, &call.sendPtrExtent, &call.recvbuff, &call.recvPtrBase, &call.recvPtrExtent,
&call.count, &call.datatype, &call.op, &call.root,
&call.comm, &call.nRanks, &call.stream, &call.nTasks, &call.globalRank, &call.timestamp, &call.tid,
&call.hipDev, &call.graphCaptured, &call.graphID) == 17);
&call.hipDev, &call.graphCaptured, &call.graphID) == 21);
}
calls.push_back(call);
}