From ba97c9c18bc67b5cff3cdd94021df4556aa32f4a Mon Sep 17 00:00:00 2001 From: Tim <43156029+AtlantaPepsi@users.noreply.github.com> Date: Fri, 13 Jun 2025 15:05:34 -0400 Subject: [PATCH] replayer update v0 (#1733) * 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 --- src/collectives.cc | 8 + src/group.cc | 15 +- src/include/recorder.h | 16 +- src/misc/recorder.cc | 78 +- test/_RecorderTests.cpp | 10 +- tools/RcclReplayer/Makefile | 2 +- tools/RcclReplayer/rcclReplayer.cpp | 1339 +++++++++++++-------------- tools/RcclReplayer/rcclReplayer.hpp | 429 +-------- 8 files changed, 768 insertions(+), 1129 deletions(-) diff --git a/src/collectives.cc b/src/collectives.cc index 37860cc25d..09f5eeca37 100644 --- a/src/collectives.cc +++ b/src/collectives.cc @@ -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; rnRanks), 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(info.recvbuff)); // should always exist for collectives + if (info.sendbuff) // ncclSend/Recv + { + hipMemGetAddressRange(&sendPtrBase, &sendPtrExtent, const_cast(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>(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& 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& 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& 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); } diff --git a/test/_RecorderTests.cpp b/test/_RecorderTests.cpp index c80d5b4f72..16fec58761 100644 --- a/test/_RecorderTests.cpp +++ b/test/_RecorderTests.cpp @@ -40,16 +40,18 @@ namespace RcclUnitTesting std::vector calls; char entry[4096]; + gethostname(entry, 256); //parse the outfile - std::string filename = "test" + std::to_string(pid) + ".json"; - std::ifstream fp("/tmp/test" + std::to_string(pid) + ".json"); + std::string filename = "/tmp/test." + std::to_string(pid) + "." + std::string(entry) + ".json"; + std::ifstream fp(filename); fp.getline(entry, 4096); fp.getline(entry, 4096); fp.getline(entry, 4096); parseJsonEntry(entry, calls); - int result = memcmp((char*)&calls[0]+4, (char*)&call+4, sizeof(rccl::rcclApiCall)-4); + // compare only the fields after the pid field + int result = memcmp(&(calls[0].pid)+1, &(call.pid)+1, sizeof(rccl::rcclApiCall)-sizeof(call.pid)); fp.close(); // care that recorder is not designed to anticipate fp closing before destructor - remove(filename.c_str()); + //remove(filename.c_str()); TODO: after implement fp closing mechanism via envvar, remove file after test case unsetenv("RCCL_REPLAY_FILE"); assert(!result); } diff --git a/tools/RcclReplayer/Makefile b/tools/RcclReplayer/Makefile index d2f12d3783..45197ecdf5 100644 --- a/tools/RcclReplayer/Makefile +++ b/tools/RcclReplayer/Makefile @@ -6,7 +6,7 @@ INCLUDES = -I$(MPI_DIR)/include -I$(RCCL_DIR)/include -I$(RCCL_DIR)/hipify/src/i LDFLAGS = -L$(MPI_DIR)/lib -L$(RCCL_DIR) -lmpi -lrccl main: rcclReplayer.cpp - $(ROCM_DIR)/bin/hipcc rcclReplayer.cpp -O1 -g -o rcclReplayer $(INCLUDES) $(LDFLAGS) + $(ROCM_DIR)/bin/hipcc -std=c++20 rcclReplayer.cpp -O0 -g -o rcclReplayer $(INCLUDES) $(LDFLAGS) clean: rm -f ./rcclReplayer diff --git a/tools/RcclReplayer/rcclReplayer.cpp b/tools/RcclReplayer/rcclReplayer.cpp index 413a505593..f93cd86d8d 100644 --- a/tools/RcclReplayer/rcclReplayer.cpp +++ b/tools/RcclReplayer/rcclReplayer.cpp @@ -4,14 +4,565 @@ #include #include #include +#include #include #include #include +#include #include "rcclReplayer.hpp" +#include +#include + +using namespace rccl; + +static int json_format = 0; // binary by default + +// move to inside class or kept as static var +static constexpr size_t rcclCallSize = sizeof(rcclApiCall); +static char line[rcclCallSize]; // size of collectivecall struct +static int lineNum = 0; +static ncclUniqueId uniqueId; + +// assuming shared file system or similar +// should this be replayer or in main +static int ParseLogFormat(const char* logFormat, std::string& filename, std::string& extension) +{ + int json_format = 0; + size_t dot; + if ((dot = std::string(logFormat).find(".")) != std::string::npos) + { + filename = std::string(logFormat).substr(0, dot); + extension = std::string(logFormat).substr(dot); + if (extension.compare(".json") == 0) + { + json_format = 1; + } + } else { + filename = std::string(logFormat); + } + return json_format; + // TODO: modularize and reuse this snippet from recorder +} + +Replayer::Replayer(const std::string& logname, int json_format, int rank, int size) : myRank(rank), + numGlobalRanks(size) +{ + log.open(logname, json_format ? std::ifstream::in : std::ifstream::binary); +} + +void Replayer::parse() +{ + while (log.read(line, rcclCallSize)) // istream::get fail here when running into newline + { + rcclApiCall call = *((rcclApiCall*) line); + + if (call.sendPtrBase) + { + if (!dMemMap.contains(call.sendPtrBase)) + { + dMemMap[call.sendPtrBase].size = call.sendPtrExtent; + } + dMemMap[call.sendPtrBase].lastLineUsed = lineNum; + } + if (call.recvPtrBase) + { + if (!dMemMap.contains(call.recvPtrBase)) + { + dMemMap[call.recvPtrBase].size = call.recvPtrExtent; + } + dMemMap[call.recvPtrBase].lastLineUsed = lineNum; + } + if (call.stream) + { + streams[call.stream].second = lineNum; + } + + switch (call.type) { + case rrGroupStart: + case rrGroupEnd: + case rrGroupSimulatedEnd: // TODO + case rrCommInitRank: + /// case rrCommInitRankConfig: <-- these all should depend on CommInitDev + case rrCommSplit: // <-- not covered for now dealt with in replay time + case rrCommFinalize: + case rrCommDestroy: + case rrCommAbort: + case rrCommRegister: + case rrCommDeregister: // I think commDeregister is not affected by handle in both way? + case rrMemFree: + case rrRedOpCreatePreMulSum: + case rrRedOpDestroy: + case rrOtherCall: + { + break; // no op + } + // Communicator + case rrGetUniqueId: + { + idRankMap[call.commId]; + break; + } + + case rrCommInitDev: // which should capture all comm - uniqueID relations + { + Ids.push_back(call.commId); + // for debugging might want a reverse map + break; + } + case rrCommInitAll: + { + if (call.sendbuff) + { + log.ignore(call.root * sizeof(int)); + } + break; + } + + // Memory allocation + //integrate these later + case rrMemAlloc: + { + // Replayer will not free this without explicit ncclMemFree + dMemMap[call.recvbuff].size = call.count; + break; + } + + case rrAllToAllv: + { + log.ignore(4 * call.nRanks * sizeof(size_t)); // will allocate s/rdispls/count each time + } + default: // collectives + { + /* if capturing: + * if first time (start.empty) + * init stream + * push this line for replayer later + * increment depth + * else + * use internal counter to separate diff graph launch + */ + if (call.graphCaptured == 1) + { + if (!graphLife.contains(call.graphID)) + { + graphLife[call.graphID].starts.insert(lineNum); + graphLife[call.graphID].stream = call.stream; + } + graphLife[call.graphID].depth++; + graphLife[call.graphID].counter++; + graphLife[call.graphID].end = lineNum; // in case the graph never gets launched + } else if (call.graphID) { + if (graphLife[call.graphID].counter == graphLife[call.graphID].depth) + { + graphLife[call.graphID].starts.insert(lineNum); + } + graphLife[call.graphID].counter--; + if (graphLife[call.graphID].counter == 0) + { + graphLife[call.graphID].end = lineNum; // we currently sync graph after its last launch + // for convenience of graph destroy, may later + // need a comm->graphs map so that CommReclaim dont hang + graphLife[call.graphID].counter = graphLife[call.graphID].depth; + } + } + } + } + lineNum++; + } + + // exchange communicator info + std::vector comm_count(numGlobalRanks); + comm_count[myRank] = Ids.size(); + MPI_Allgather(MPI_IN_PLACE, 0, MPI_DATATYPE_NULL, comm_count.data(), 1, MPI_INT, MPI_COMM_WORLD); + + std::vector displs(comm_count.size() + 1, 0); + std::inclusive_scan(comm_count.begin(), comm_count.end(), displs.begin() + 1); + int aggragatedCommCount = std::reduce(comm_count.begin(), comm_count.end()); + /* + * rank1, comm_count[1]xID r2, comm_count[2] r3 ... r4 ... + * AllRankCommIds [------------------------+-----------------+-------+---------+....] + */ + std::vector AllRankCommIds(aggragatedCommCount); + MPI_Allgatherv(Ids.data(), Ids.size(), MPI_UINT64_T, + AllRankCommIds.data(), comm_count.data(), displs.data(), MPI_UINT64_T, MPI_COMM_WORLD); + + int k = 0; + for (int i = 0; i < numGlobalRanks; i++) + { + if (i == myRank) + { + k += Ids.size(); + continue; + } + for (int j = 0; j < comm_count[i]; j++) + { + if (idRankMap.contains(AllRankCommIds[k])) + { + idRankMap[AllRankCommIds[k]].push_back(i); + } + k++; + } + } + + lineNum = 0; + log.clear(); + log.seekg(0, std::ios_base::beg); + // TODO: print out resources here allocated if requested +} + +void Replayer::replay() +{ + while (log.read(line, rcclCallSize)) + { + rcclApiCall call = *((rcclApiCall*) line); + printf("[INFO ] Rank %d - Line %d : %s\n", myRank, lineNum, rcclCallStr[call.type]); + HIP_CALL(hipSetDevice(call.hipDev)); + void *sbuffer = NULL, *rbuffer = NULL; + + if (call.type < rrGroupStart) + { + if ((call.sendPtrBase && !dMemMap.contains(call.sendPtrBase)) || (call.recvPtrBase && !dMemMap.contains(call.recvPtrBase))) + { + printf("[ERROR ] Rank %d - Line %d : Unknown buffer in collectives\n", myRank, lineNum); + exit(1); + } + + if (call.sendPtrBase) + { + if (!dMemMap[call.sendPtrBase].base) + { + HIP_CALL(hipMalloc(&dMemMap[call.sendPtrBase].base, dMemMap[call.sendPtrBase].size)); + } + std::ptrdiff_t diff = (char*)call.sendbuff - (char*)call.sendPtrBase; + sbuffer = (char*)dMemMap[call.sendPtrBase].base + diff; + } + if (call.recvPtrBase) + { + if (!dMemMap[call.recvPtrBase].base) + { + HIP_CALL(hipMalloc(&dMemMap[call.recvPtrBase].base, dMemMap[call.recvPtrBase].size)); + } + std::ptrdiff_t diff = (char*)call.recvbuff - (char*)call.recvPtrBase; + rbuffer = (char*)dMemMap[call.recvPtrBase].base + diff; + } + + //stream + if (call.stream && !streams[call.stream].first) + { + HIP_CALL(hipStreamCreate(&streams[call.stream].first)); + } + + //graph + /* + * if capturing + * if firstime (line in start) + * stream capture begin + * else if stream differ from initial capturing stream + * //create dependency + * if depth reached // after call execution switch + * conclude graph + * else (launching) + */ + if (call.graphCaptured == 1) + { + graphLife[call.graphID].counter--; + if (graphLife[call.graphID].starts.contains(lineNum)) + { + HIP_CALL(hipStreamBeginCapture(streams[call.stream].first, hipStreamCaptureModeGlobal)); + printf("[INFO ] Rank %d - Line %d : starting capture graph %llu\n", myRank, lineNum, call.graphID); + } else if (graphLife[call.graphID].stream != call.stream) { + printf("[WARNING ] \x1b[31mRank %d - Line %d : multi-stream graph may not replay original dependency accurately\x1b[0m\n", myRank, lineNum); + hipEvent_t event; + HIP_CALL(hipEventCreate(&event)); + graphLife[call.graphID].events.push_back(event); + HIP_CALL(hipEventRecord(event, streams[graphLife[call.graphID].stream].first)); + HIP_CALL(hipStreamWaitEvent(streams[call.stream].first, event)); + } + } else if (call.graphID) { + if (graphLife[call.graphID].starts.contains(lineNum)) + { + printf("[INFO ] Rank %d - Line %d : launching graph %llu\n", myRank, lineNum, call.graphID); + HIP_CALL(hipGraphLaunch(graphLife[call.graphID].graphExec, streams[call.stream].first)); + } + printf("[INFO ] Rank %d - Line %d : being played by previous graph %llu\n", myRank, lineNum, call.graphID); + goto cleanup; + } + } + + switch (call.type) { + case rrGroupSimulatedEnd: // TODO: cannot test atm + /// case rrCommInitRankConfig: <-- these all should depend on CommInitDev + case rrRedOpCreatePreMulSum: + case rrRedOpDestroy: + case rrOtherCall: + { + printf("[ERROR ] Rank %d - Line %d : Unexpected call: %s\n", myRank, lineNum, rcclCallStr[call.type]); + exit(1); + } + + // To be integrated later + case rrCommFinalize: + { + NCCL_CALL(ncclCommFinalize(commMap[call.comm])); + break; + } + case rrCommDestroy: + { + NCCL_CALL(ncclCommDestroy(commMap[call.comm])); + break; + } + case rrCommAbort: + { + NCCL_CALL(ncclCommAbort(commMap[call.comm])); + break; + } + + case rrGroupStart: + { + NCCL_CALL(ncclGroupStart()); + break; + } + case rrGroupEnd: + { + NCCL_CALL(ncclGroupEnd()); + break; + } + + case rrGetUniqueId: + { + NCCL_CALL(ncclGetUniqueId(&uniqueId)); + idMap[call.commId] = uniqueId; + break; + } + case rrCommInitRank: + { + lastCall = rrCommInitRank; + break; + } + /// case rrCommInitRankConfig: + case rrCommInitDev: + { + if (lastCall == rrCommInitAll) // no other calls between ncclCommInitAll and ncclCommInitRankDev + { // nor ncclCommInitRankDev not proceeded by ncclCommInitAll/Rank() + goto cleanup; + } + // set device + // TODO: double check this, since some version of NCCL theres a reset to original device + HIP_CALL(hipSetDevice(call.root)); + + if (!idMap.contains(call.commId)) + { + MPI_Recv(&uniqueId, sizeof(ncclUniqueId), MPI_BYTE, MPI_ANY_SOURCE, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE); + } else { + for (int rank : idRankMap[call.commId]) + { + MPI_Send(&idMap[call.commId], sizeof(ncclUniqueId), MPI_BYTE, rank, 0, MPI_COMM_WORLD); + } + uniqueId = idMap[call.commId]; // <- double check in case of bug/communicator init hang + } + ncclComm_t comm; + NCCL_CALL(ncclCommInitRank(&comm, call.nRanks, uniqueId, call.globalRank)); + commMap[call.comm] = comm; + break; + } + case rrCommInitAll: + { + int ndev = call.root; + int *devlist = NULL; + if (call.sendbuff) + { + std::vector devices(ndev); + log.read((char*)devices.data(), ndev * sizeof(int)); + devlist = devices.data(); + } + ncclComm_t comm; + NCCL_CALL(ncclCommInitAll(&comm, ndev, devlist)); + commMap[call.comm] = comm; + break; + } + case rrCommSplit: + { + int color = call.nRanks; + int key = call.globalRank; + // TODO: parse config later + ncclComm_t newcomm; + ncclComm_t comm = (ncclComm_t) call.commId; + NCCL_CALL(ncclCommSplit(commMap[comm], color, key, &newcomm, NULL)); + commMap[call.comm/*original newcomm to commSplit call*/] = newcomm; + break; + } + + + case rrCommRegister: + { + if (!dMemMap.contains(call.sendPtrBase) || !commMap.contains(call.comm)) + { + printf("[ERROR ] Rank %d - Line %d : Unknown buffer for CommRegister\n", myRank, lineNum); + exit(1); + } + if (!dMemMap[call.sendPtrBase].base) + { + HIP_CALL(hipMalloc(&dMemMap[call.sendPtrBase].base, dMemMap[call.sendPtrBase].size)); + } + sbuffer = (char*)dMemMap[call.sendPtrBase].base + (std::ptrdiff_t)((char*)call.sendbuff - (char*)call.sendPtrBase); + NCCL_CALL(ncclCommRegister(commMap[call.comm], sbuffer, dMemMap[call.sendPtrBase].size, &handleMap[call.recvbuff])); + break; + } + case rrCommDeregister: + { + NCCL_CALL(ncclCommDeregister(commMap[call.comm], handleMap[call.recvbuff])); + break; + } + case rrMemAlloc: + { + NCCL_CALL(ncclMemAlloc(&dMemMap[call.recvbuff].base, call.count)); + break ; + } + case rrMemFree: + { + NCCL_CALL(ncclMemFree(dMemMap[call.recvbuff].base)); + break; + } + + // TODO: further simplify switch base on common parameters + // no op or root + case rrAllToAll: + { + NCCL_CALL(ncclAllToAll(sbuffer, rbuffer, call.count, call.datatype, commMap[call.comm], streams[call.stream].first)); + break; + } + case rrAllGather: + { + NCCL_CALL(ncclAllGather(sbuffer, rbuffer, call.count, call.datatype, commMap[call.comm], streams[call.stream].first)); + break; + } + // op root + case rrReduce: + { + NCCL_CALL(ncclReduce(sbuffer, rbuffer, call.count, call.datatype, call.op, call.root, commMap[call.comm], streams[call.stream].first)); + break; + } + // root + case rrBroadcast: + { + NCCL_CALL(ncclBroadcast(sbuffer, rbuffer, call.count, call.datatype, call.root, commMap[call.comm], streams[call.stream].first)); + break; + } + case rrScatter: + { + NCCL_CALL(ncclScatter(sbuffer, rbuffer, call.count, call.datatype, call.root, commMap[call.comm], streams[call.stream].first)); + break; + } + case rrGather: + { + NCCL_CALL(ncclGather(sbuffer, rbuffer, call.count, call.datatype, call.root, commMap[call.comm], streams[call.stream].first)); + break; + } + // root - + case rrBcast: + { + NCCL_CALL(ncclBcast(rbuffer, call.count, call.datatype, call.root, commMap[call.comm], streams[call.stream].first)); + break; + } + case rrSend: + { + NCCL_CALL(ncclSend(rbuffer, call.count, call.datatype, call.root, commMap[call.comm], streams[call.stream].first)); + break; + } + case rrRecv: + { + NCCL_CALL(ncclRecv(rbuffer, call.count, call.datatype, call.root, commMap[call.comm], streams[call.stream].first)); + break; + } + // op + case rrReduceScatter: + { + NCCL_CALL(ncclReduceScatter(sbuffer, rbuffer, call.count, call.datatype, call.op, commMap[call.comm], streams[call.stream].first)); + break; + } + case rrAllReduce: + { + NCCL_CALL(ncclAllReduce(sbuffer, rbuffer, call.count, call.datatype, call.op, commMap[call.comm], streams[call.stream].first)); + break; + } + // a2av + case rrAllToAllv: + { + // timer pause here + // assuming blocking for now + int size = call.nRanks; + std::vector sendcounts(size), sdispls(size), recvcounts(size), rdispls(size); + log.read((char*)sendcounts.data(), size * sizeof(size_t)); + log.read((char*)sdispls.data(), size * sizeof(size_t)); + log.read((char*)recvcounts.data(), size * sizeof(size_t)); + log.read((char*)rdispls.data(), size * sizeof(size_t)); + + NCCL_CALL(ncclAllToAllv(sbuffer, sendcounts.data(), sdispls.data(), rbuffer, recvcounts.data(), rdispls.data(), + call.datatype, commMap[call.comm], streams[call.stream].first)); + HIP_CALL(hipStreamSynchronize(streams[call.stream].first)); // TODO: remove + break; + } + } //switch + printf("[INFO ] Rank %d - Line %d : %s called\n", myRank, lineNum, rcclCallStr[call.type]); + lastCall = call.type; + + if (call.graphCaptured == 1) + { + // TODO: This requires further testing + if (graphLife[call.graphID].stream != call.stream) + { + hipEvent_t event; + HIP_CALL(hipEventCreate(&event)); + graphLife[call.graphID].events.push_back(event); + HIP_CALL(hipEventRecord(event, streams[call.stream].first)); + HIP_CALL(hipStreamWaitEvent(streams[graphLife[call.graphID].stream].first, event)); + } + if (graphLife[call.graphID].counter == 0) + { + hipGraphNode_t temp; + char errbuff[3000]; + HIP_CALL(hipStreamEndCapture(streams[graphLife[call.graphID].stream].first, &graphLife[call.graphID].graph)); + // TODO: confirm with clr behavior of graphInstantiate in face of failure + HIP_CALL(hipGraphInstantiate(&graphLife[call.graphID].graphExec, graphLife[call.graphID].graph, &temp, errbuff, 3000)); + for (hipEvent_t e : graphLife[call.graphID].events) + { + HIP_CALL(hipEventDestroy(e)); + } + } + } + +cleanup: + printf("[INFO ] Rank %d - Line %d : cleaning up\n", myRank, lineNum); + + // Free resources if possible + if (call.sendPtrBase && lineNum == dMemMap[call.sendPtrBase].lastLineUsed) { + // TODO: free contains a sync, may need a second thought + // also this may proceed commDeregister in case of UBR thus susceptible to change in implementation + HIP_CALL(hipFree(dMemMap[call.sendPtrBase].base)); + dMemMap[call.sendPtrBase].base = NULL; // in case of in place ops + } + if (call.recvPtrBase && lineNum == dMemMap[call.recvPtrBase].lastLineUsed && dMemMap[call.recvPtrBase].base) { + HIP_CALL(hipFree(dMemMap[call.recvPtrBase].base)); + } + if (call.graphID && lineNum == graphLife[call.graphID].end) { + HIP_CALL(hipStreamSynchronize(streams[call.stream].first)); + HIP_CALL(hipGraphExecDestroy(graphLife[call.graphID].graphExec)); + HIP_CALL(hipGraphDestroy(graphLife[call.graphID].graph)); + } + if (call.stream && lineNum == streams[call.stream].second) + { + HIP_CALL(hipStreamSynchronize(streams[call.stream].first)); // ? + HIP_CALL(hipStreamDestroy(streams[call.stream].first)); + } + lineNum++; // change for a2av + } +} + int main(int argc, char **argv) { + unsetenv("RCCL_REPLAY_FILE"); MPI_Init(&argc, &argv); if (argc <= 1) { printf("Usage: %s logfile [numGpusPerMpiRank = 1]\n", argv[0]); @@ -26,735 +577,97 @@ int main(int argc, char **argv) // Parse command line arguments char* logFilename = argv[1]; int numGpusPerMpiRank = (argc > 2 ? atoi(argv[2]) : 1); - int parseOnly = (argc > 3 ? atoi(argv[3]) : 0); - - CollectiveCalls collCalls; - collCalls.firstGlobalRank = mpiRank * numGpusPerMpiRank; - collCalls.numGlobalRanks = numMpiRanks * numGpusPerMpiRank; + /// int parseOnly = (argc > 3 ? atoi(argv[3]) : 0); + assert(numGpusPerMpiRank == 1); // Figure out starting GPU index to use based on hostname - int nameLen; - char name[MPI_MAX_PROCESSOR_NAME]; - std::vector allnames(numMpiRanks * MPI_MAX_PROCESSOR_NAME, 0); - MPI_Get_processor_name(name, &nameLen); - MPI_Allgather(name, MPI_MAX_PROCESSOR_NAME, MPI_CHAR, - allnames.data(), MPI_MAX_PROCESSOR_NAME, MPI_CHAR, MPI_COMM_WORLD); + int nameLen, pid; + char hostname[MPI_MAX_PROCESSOR_NAME]; + MPI_Get_processor_name(hostname, &nameLen); - // Offset local gpu device index based on number of previous ranks on the same host - collCalls.localGpuOffset = 0; - for (int rank = 0; rank < mpiRank; rank++) { - if (!strcmp(name, allnames.data() + (rank * MPI_MAX_PROCESSOR_NAME))) - collCalls.localGpuOffset += numGpusPerMpiRank; - } - if (mpiRank == 0) - printf("RCCL Replayer: %d x %d = %d total ranks\n", numMpiRanks, numGpusPerMpiRank, collCalls.numGlobalRanks); - printf("Rank %d [%s] LocalGpuOffset: %d GlobalRankFirst %d GlobalRankLast %d\n", - mpiRank, name, collCalls.localGpuOffset, collCalls.firstGlobalRank, collCalls.firstGlobalRank + numGpusPerMpiRank - 1); + std::string output_file, output_extension; + int json_format = ParseLogFormat(logFilename, output_file, output_extension); + assert(json_format == 0); - // Parse collectives from logfile - if (parseOnly) collCalls.numGlobalRanks = parseOnly; - ParseCollectives(logFilename, mpiRank == 0, collCalls); - if (collCalls.groupCalls.size() == 0) { - MPI_Finalize(); - return 0; - } - if (parseOnly) return 0; + // Only root handles file-rank assignment to avoid file handle pressure + if (mpiRank != 0) + { + MPI_Gather(hostname, MPI_MAX_PROCESSOR_NAME, MPI_CHAR, + NULL, 0, MPI_DATATYPE_NULL, 0, MPI_COMM_WORLD); - // Setup all communicators - if (mpiRank == 0) printf("Preparing %d communicator(s) per rank\n", collCalls.numCommsPerRank); - collCalls.localRankComms.resize(numGpusPerMpiRank, std::vector(collCalls.numCommsPerRank)); - collCalls.localRankStreams.resize(numGpusPerMpiRank, std::vector(collCalls.numCommsPerRank)); + MPI_Scatter(NULL, 0, MPI_DATATYPE_NULL, + hostname, MPI_MAX_PROCESSOR_NAME, MPI_CHAR, 0, MPI_COMM_WORLD); + MPI_Scatter(NULL, 0, MPI_DATATYPE_NULL, + &pid, 1, MPI_INT, 0, MPI_COMM_WORLD); + } else { + std::vector allhosts(numMpiRanks * MPI_MAX_PROCESSOR_NAME, 0); + std::vector pids(numMpiRanks * sizeof(int), 0); - for (int commIdx = 0; commIdx < collCalls.numCommsPerRank; commIdx++) { - // Create a unique ID and broadcast it to all ranks - ncclUniqueId uniqueId; - if (mpiRank == 0) ncclGetUniqueId(&uniqueId); - MPI_Bcast(&uniqueId, sizeof(ncclUniqueId), MPI_BYTE, 0, MPI_COMM_WORLD); + MPI_Gather(hostname, MPI_MAX_PROCESSOR_NAME, MPI_CHAR, + allhosts.data(), MPI_MAX_PROCESSOR_NAME, MPI_CHAR, 0, MPI_COMM_WORLD); - // Initialize comms and strams - NCCL_CALL(ncclGroupStart()); - for (int i = 0; i < numGpusPerMpiRank; i++) { - HIP_CALL(hipSetDevice(collCalls.localGpuOffset + i)); - NCCL_CALL(ncclCommInitRank(&collCalls.localRankComms[i][commIdx], collCalls.numGlobalRanks, uniqueId, collCalls.firstGlobalRank + i)); - HIP_CALL(hipStreamCreate(&collCalls.localRankStreams[i][commIdx])); + // All hostnames in the recorded program + std::unordered_set hostnames; + for (int i = 0; i < numMpiRanks; i++) + { + hostnames.insert(std::string(allhosts.data() + i * MPI_MAX_PROCESSOR_NAME)); // assuming null terminator included } - NCCL_CALL(ncclGroupEnd()); - } - printf("Rank %d Done setting up communicators\n", mpiRank); - int numSkippedCalls = 0; - int numInvalid = 0; - double runTime; - std::ofstream datafile; - datafile.open("replayer_data.csv"); - if (!datafile.is_open()) { - printf("[ERROR] Unable to open file replayer_data.csv\n"); - exit(-1); - } - datafile << "callNumber, functionName, inPlace, count(numElements), datatype, op, root, time(msec), groupCallBusBandwidth(GB/s)\n"; - auto start = std::chrono::high_resolution_clock::now(); - for (size_t i = 0; i < collCalls.groupCalls.size(); i++) { - MPI_Barrier(MPI_COMM_WORLD); - if (collCalls.groupCalls[i].isValid) { - if (mpiRank == 0) - { - printf("Running Collective Call %lu of %lu\n", i+1, collCalls.groupCalls.size()); - PrintGroupCall(collCalls.groupCalls[i]); - } - double runTime = ReplayRccl(collCalls, i, numInvalid); - if (mpiRank == 0) { - dataToCsv(collCalls.groupCalls[i], datafile, runTime); - } - } else { - if (mpiRank == 0) { - printf("[ERROR] in group call: (skipping...)\n"); - for (auto const& rd : collCalls.groupCalls[i].rankData) { - printf(" - Rank %02d: comm %d in line %d\n", rd.first, rd.second.commIdx, rd.second.lineNum); - for (int task = 0; task < rd.second.tasks.size(); task++) { - TaskInfo ti = rd.second.tasks[task]; - printf(" - Task %02d: %32s inPlace=%d count=%lu datatype=%d op=%d root=%d\n", - task, ncclFuncNames[ti.funcType], ti.inPlace, ti.count, ti.datatype, ti.op, ti.root); - } + // Register all hostnames and pid from recorder logs + std::unordered_map> logHosts; + int file_pid, a = 0/*counter*/; + DIR *d; + struct dirent *dir; + if (d = opendir(".")) { + while ((dir = readdir(d)) != NULL) { + // MPI_MAX_PROCESSOR_NAME = 256 + if (sscanf(dir->d_name, (output_file + ".%d.%256[^.]" + output_extension).c_str(), &file_pid, hostname) == 2) + { + logHosts[std::string(hostname)].push_back(file_pid); + a++; } } - numSkippedCalls++; + closedir(d); + } + // Double check number of nodes and number of processes match for recorder and replayer + assert(logHosts.size() == hostnames.size()); + assert(a == numMpiRanks); + // Assign mapping of replayer hostname to recorder hostname + std::unordered_map hostAssignment; + auto it = logHosts.begin(); + for (const auto &host : hostnames) + { + hostAssignment[host] = (*it).first; + it++; + } + for (int i = 0; i < numMpiRanks; i++) + { + std::string host(allhosts.data() + i * MPI_MAX_PROCESSOR_NAME); + strcpy(allhosts.data() + i * MPI_MAX_PROCESSOR_NAME, hostAssignment[host].c_str()); + pids[i] = logHosts[hostAssignment[host]].back(); + logHosts[hostAssignment[host]].pop_back(); } - } - auto end = std::chrono::high_resolution_clock::now(); - std::chrono::duration duration = end - start; - datafile.close(); - // Destroy all communicators - for (int commIdx = 0; commIdx < collCalls.numCommsPerRank; commIdx++) { - for (int i = 0; i < numGpusPerMpiRank; i++) { - NCCL_CALL(ncclCommDestroy(collCalls.localRankComms[i][commIdx])); - HIP_CALL(hipStreamDestroy(collCalls.localRankStreams[i][commIdx])); - } + // Distribute the target log for each rank (pid and hostname) + MPI_Scatter(allhosts.data(), MPI_MAX_PROCESSOR_NAME, MPI_CHAR, + hostname, MPI_MAX_PROCESSOR_NAME, MPI_CHAR, 0, MPI_COMM_WORLD); + MPI_Scatter(pids.data(), 1, MPI_INT, + &pid, 1, MPI_INT, 0, MPI_COMM_WORLD); } - if (mpiRank == 0) printf("Executed group calls: %zu\n", collCalls.groupCalls.size() - numSkippedCalls); - if (mpiRank == 0) printf("Skipped group calls: %d\n", numSkippedCalls); + // Initialize Replayer + std::string logfile = output_file + "." + std::to_string(pid) + "." + + std::string(hostname) + output_extension; /// perhaps another func for assemble logname + std::cout << mpiRank << " : " << logfile<, std::vector> arrivalCounter; - - gc.isValid = true; - - for (auto rd : gc.rankData) { - for (int task = 0; task < rd.second.tasks.size(); task++) { - TaskInfo ti = rd.second.tasks[task]; - - std::string funcName = (ti.funcType == ncclCollSend || ti.funcType == ncclCollRecv) ? "Send/Recv" : ncclFuncNames[ti.funcType]; - std::tuple key(funcName, ti.count, ti.datatype, ti.op); - - auto& rankVector = arrivalCounter[key]; - if (rankVector.size() < cc.numGlobalRanks) - rankVector.resize(cc.numGlobalRanks); - - // rankVector in arrivalCount represents the rank information - // Count the number of tasks that are going to be executed by each rank. This is to validate the group call later on. - // Nom-Send/Recv rank counts (rankVector elements) should be equal at the end, and for Send/Recv, all the elements of rankVector should be equal to 0 - if (ti.funcType == ncclCollRecv) { - rankVector[ti.root]--; - } else { - rankVector[rd.first]++; - } - } - } - - // Iterate through the map variable and report/validate the results - for (const auto& e : arrivalCounter) { - int maxVal; - std::string funcName = std::get<0>(e.first); - size_t count = std::get<1>(e.first); - int const datatype = std::get<2>(e.first); - int const op = std::get<3>(e.first); - - bool isp2p = (funcName == "Send/Recv"); - if (!isp2p) maxVal = *std::max_element(e.second.begin(), e.second.end()); - - // Validate all the ranks have required amount of collective call (task) - for (int i = 0; i < e.second.size(); i++) { - if (e.second[i] != (isp2p ? 0 : maxVal)) { - std::string warning = (isp2p ? (e.second[i] > 0 ? "[WARN] Missing Recv" : "[WARN] Missing Send") : "[WARN] Missing " + std::string(funcName)) - + " count=" + std::to_string(count) + " datatype=" + std::to_string(datatype) + " op=" + std::to_string(op) + " at rank [" + std::to_string(i) + "]"; - if(isFirstRank) printf("%s\n", warning.c_str()); - - gc.isValid = false; - } - } - } - } - - // Check number of comms per rank - cc.numCommsPerRank = cc.globalRankComms[0].size(); - for (int i = 1; i < cc.numGlobalRanks; i++) { - if (cc.numCommsPerRank != cc.globalRankComms[i].size()) { - printf("[ERROR] Replayer currently only supports identical number of communicators across all ranks\n"); - printf("[ERROR] Rank %d has %lu communicators (expecting %d)\n", i, cc.globalRankComms[i].size(), cc.numCommsPerRank); - exit(1); - } - } - - // Detect and replace scatter patterns - for (auto& gc : cc.groupCalls) { - if (!gc.isValid) continue; - int scatterRoot = -1; - bool isScatter = true; - for (auto& [rank, rankData] : gc.rankData) { - int sendCount = 0, recvCount = 0; - for (const auto& task : rankData.tasks) { - if (task.funcType == ncclCollSend) - sendCount++; - else if (task.funcType == ncclCollRecv) - recvCount++; - } - if (sendCount == cc.numGlobalRanks && recvCount == 1) { - if (scatterRoot == -1) { - // Root is the first rank that matches the condition - scatterRoot = rank; - } else { - isScatter = false; - break; - } - } else if (recvCount != 1 || sendCount != 0) { - // Non-root ranks must only recv and not send - isScatter = false; - break; - } - } - - // Replace send/recv calls with scatter call for the group call - if (isScatter) { - TaskInfo scatterTask; - scatterTask.funcType = ncclCollScatter; - scatterTask.count = gc.rankData[scatterRoot].tasks[0].count; - scatterTask.datatype = gc.rankData[scatterRoot].tasks[0].datatype; - scatterTask.root = scatterRoot; - - for (auto& [rank, rankData] : gc.rankData) { - rankData.tasks.clear(); - rankData.tasks.push_back(scatterTask); - } - - if (isFirstRank) - printf("[INFO] Scatter pattern detected and replaced with scatter collective\n"); - } - } -} - -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) -{ - int numLocalRanks = cc.localRankComms.size(); - - for (int localIdx = 0; localIdx < numLocalRanks; localIdx++) { - int globalRank = cc.firstGlobalRank + localIdx; - if (cc.groupCalls[groupIdx].rankData.count(globalRank) == 0) continue; - HIP_CALL(hipSetDevice(cc.localGpuOffset + localIdx)); - - RankData& rankData = cc.groupCalls[groupIdx].rankData.at(globalRank); - int numTasks = rankData.tasks.size(); - - for (int taskId = 0; taskId < numTasks; taskId++) { - TaskInfo& task = rankData.tasks[taskId]; - - // Each task has a size based on the type of collective (funcType) - std::pair numBytes = GetSize(task, cc.numGlobalRanks); - - if (task.inPlace) { - numBytes.first = std::max(numBytes.first, numBytes.second); - numBytes.second = numBytes.first; - } - - // Allocate memory - AllocateMem(task.inputGpu, numBytes.first, true); - AllocateMem(task.outputCpu, numBytes.second); - AllocateMem(task.expected, numBytes.second); - - if (!task.inPlace) { - AllocateMem(task.outputGpu, numBytes.second, true); - } else { - task.outputGpu = task.inputGpu; - } - - // Prepare input/output for each task based on collective type - PrepareDataFunc(task, globalRank, cc.numGlobalRanks); - - HIP_CALL(hipDeviceSynchronize()); - } - } - - // Execute the collective call (task) - std::chrono::time_point start = std::chrono::high_resolution_clock::now(); - NCCL_CALL(ncclGroupStart()); - for (int localIdx = 0; localIdx < numLocalRanks; localIdx++) { - int globalRank = cc.firstGlobalRank + localIdx; - if (cc.groupCalls[groupIdx].rankData.count(globalRank) == 0) continue; - - RankData& rankData = cc.groupCalls[groupIdx].rankData.at(globalRank); - int numTasks = rankData.tasks.size(); - int commIdx = rankData.commIdx; - for (int taskId = 0; taskId < numTasks; taskId++) { - TaskInfo& task = rankData.tasks[taskId]; - ExecuteCollective(task, cc.localRankComms[localIdx][commIdx], cc.localRankStreams[localIdx][commIdx]); - } - } - NCCL_CALL(ncclGroupEnd()); - - // Synchronize devices and free memory - for (int localIdx = 0; localIdx < numLocalRanks; localIdx++) { - int globalRank = cc.firstGlobalRank + localIdx; - if (cc.groupCalls[groupIdx].rankData.count(globalRank) == 0) continue; - - RankData const& rankData = cc.groupCalls[groupIdx].rankData.at(globalRank); - int commIdx = rankData.commIdx; - HIP_CALL(hipStreamSynchronize(cc.localRankStreams[localIdx][commIdx])); - } - - std::chrono::time_point end = std::chrono::high_resolution_clock::now(); - std::chrono::duration duration = (end - start); - double runTime = duration.count(); - runTime *= 1000; //convering into milliseconds - - // Data validation - bool isValid = true; - for (int localIdx = 0; localIdx < numLocalRanks; localIdx++) { - int globalRank = cc.firstGlobalRank + localIdx; - RankData& rankData = cc.groupCalls[groupIdx].rankData.at(globalRank); - int numTasks = rankData.tasks.size(); - for (int taskId = 0; taskId < numTasks; taskId++) { - TaskInfo& task = rankData.tasks[taskId]; - - // Only need Recv to validate - if (task.funcType == ncclCollSend) break; - // Ignore non-root ranks - if (IsRootUsed(task.funcType) && task.root != globalRank) break; - - std::pair numBytes = GetSize(task, cc.numGlobalRanks); - if (task.inPlace) { - numBytes.first = std::max(numBytes.first, numBytes.second); - numBytes.second = numBytes.first; - } - HIP_CALL(hipMemcpy(task.outputCpu.ptr, task.outputGpu.ptr, numBytes.second, hipMemcpyDeviceToHost)); - if (!IsEqual(task.outputCpu, task.expected, task.datatype, task.count, globalRank)) { - isValid = false; - break; // Check other ranks - } - } - } - - if (!isValid) numInvalid++; - - // Free memory - for (int localIdx = 0; localIdx < numLocalRanks; localIdx++) { - int globalRank = cc.firstGlobalRank + localIdx; - RankData& rankData = cc.groupCalls[groupIdx].rankData.at(globalRank); - int numTasks = rankData.tasks.size(); - for (int taskId = 0; taskId < numTasks; taskId++) { - TaskInfo& task = rankData.tasks[taskId]; - FreeMem(task.inputGpu, true); - if (!task.inPlace) FreeMem(task.outputGpu, true); - FreeMem(task.outputCpu); - FreeMem(task.expected); - } - } - return runTime; -} - -// GetSize will return a pair of bytes where first element in pair represents bytesSent and the second bytesRecv -std::pair GetSize(TaskInfo taskInfo, int numGlobalRanks) { - size_t sendNumBytes, recvNumBytes; - - switch (taskInfo.funcType) { - case ncclCollBroadcast: case ncclCollReduce: case ncclCollAllReduce: - sendNumBytes = taskInfo.count * DataTypeToBytes(taskInfo.datatype); - recvNumBytes = sendNumBytes; - break; - case ncclCollAllGather: case ncclCollGather: - sendNumBytes = taskInfo.count * DataTypeToBytes(taskInfo.datatype); - recvNumBytes = numGlobalRanks * sendNumBytes; - break; - case ncclCollReduceScatter: case ncclCollScatter: - recvNumBytes = taskInfo.count * DataTypeToBytes(taskInfo.datatype); - sendNumBytes = numGlobalRanks * recvNumBytes; - break; - case ncclCollAllToAll: - sendNumBytes = numGlobalRanks * taskInfo.count * DataTypeToBytes(taskInfo.datatype); - recvNumBytes = sendNumBytes; - break; - default: - sendNumBytes = taskInfo.count * DataTypeToBytes(taskInfo.datatype); - recvNumBytes = sendNumBytes; - } - return std::make_pair(sendNumBytes, recvNumBytes); -} - -void ExecuteCollective(TaskInfo& task, ncclComm_t const& comm, hipStream_t stream) -{ - switch (task.funcType) { - case ncclCollAllGather: - NCCL_CALL(ncclAllGather(task.inputGpu.ptr, task.outputGpu.ptr, task.count, task.datatype, comm, stream)); - break; - case ncclCollAllReduce: - NCCL_CALL(ncclAllReduce(task.inputGpu.ptr, task.outputGpu.ptr, task.count, task.datatype, task.op, comm, stream)); - break; - case ncclCollBroadcast: - NCCL_CALL(ncclBroadcast(task.inputGpu.ptr, task.outputGpu.ptr, task.count, task.datatype, task.root, comm, stream)); - break; - case ncclCollReduce: - NCCL_CALL(ncclReduce(task.inputGpu.ptr, task.outputGpu.ptr, task.count, task.datatype, task.op, task.root, comm, stream)); - break; - case ncclCollReduceScatter: - NCCL_CALL(ncclReduceScatter(task.inputGpu.ptr, task.outputGpu.ptr, task.count, task.datatype, task.op, comm, stream)); - break; - case ncclCollGather: - NCCL_CALL(ncclGather(task.inputGpu.ptr, task.outputGpu.ptr, task.count, task.datatype, task.root, comm, stream)); - break; - case ncclCollScatter: - NCCL_CALL(ncclScatter(task.inputGpu.ptr, task.outputGpu.ptr, task.count, task.datatype, task.root, comm, stream)); - break; - case ncclCollAllToAll: - NCCL_CALL(ncclAllToAll(task.inputGpu.ptr, task.outputGpu.ptr, task.count, task.datatype, comm, stream)); - break; - case ncclCollSend: - NCCL_CALL(ncclSend(task.inputGpu.ptr, task.count, task.datatype, task.root, comm, stream)); - break; - case ncclCollRecv: - NCCL_CALL(ncclRecv(task.outputGpu.ptr, task.count, task.datatype, task.root, comm, stream)); - break; - default: - printf("Error: unsupported collective\n"); - exit(1); - } -} - -void AllocateMem(PtrUnion& ptrUnion, size_t const numBytes, bool isGpu) { - if (numBytes) { - if (isGpu) { - HIP_CALL(hipMalloc(&ptrUnion.ptr, numBytes)); - HIP_CALL(hipMemset(ptrUnion.ptr, 0, numBytes)); - HIP_CALL(hipStreamSynchronize(NULL)); - } else { - ptrUnion.ptr = calloc(numBytes, 1); - memset(ptrUnion.ptr, 0, numBytes); - if (!ptrUnion.ptr) { - printf("Unable to allocate memory (%lu bytes)\n", numBytes); - } - } - } -} - -void FreeMem(PtrUnion& ptrUnion, bool isGpu) { - if (ptrUnion.ptr != nullptr) { - if (isGpu) - HIP_CALL(hipFree(ptrUnion.ptr)); - else - free(ptrUnion.ptr); - ptrUnion.ptr = nullptr; - } -} - -void FillPattern(PtrUnion& ptrUnion, ncclDataType_t const dataType, size_t const numElements, int globalRank, bool isGpu) { - PtrUnion temp; - size_t const numBytes = numElements * DataTypeToBytes(dataType); - - if (isGpu) - AllocateMem(temp, numBytes); - else - temp.ptr = ptrUnion.ptr; - - for (int i = 0; i < numElements; i++) { - int valueI = (globalRank + i) % 256; - double valueF = 1.0L/((double)valueI+1.0L); - SetPtr(temp, dataType, i, valueI, valueF); - } - - if (isGpu) { - HIP_CALL(hipMemcpy(ptrUnion.ptr, temp.ptr, numBytes, hipMemcpyHostToDevice)); - FreeMem(temp); - } -} - -void PrepareDataFunc(TaskInfo& taskInfo, int globalRank, int totalRanks) -{ - switch (taskInfo.funcType) - { - case ncclCollBroadcast: PrepData_Broadcast(taskInfo, globalRank); break; - case ncclCollReduce: PrepData_Reduce(taskInfo, globalRank, totalRanks, false); break; - case ncclCollAllGather: PrepData_Gather(taskInfo, globalRank, totalRanks, true); break; - case ncclCollReduceScatter: PrepData_ReduceScatter(taskInfo, globalRank, totalRanks); break; - case ncclCollAllReduce: PrepData_Reduce(taskInfo, globalRank, totalRanks, true); break; - case ncclCollGather: PrepData_Gather(taskInfo, globalRank, totalRanks, false); break; - case ncclCollScatter: PrepData_Scatter(taskInfo, globalRank, totalRanks); break; - case ncclCollAllToAll: PrepData_AlltoAll(taskInfo, globalRank, totalRanks); break; - case ncclCollSend: PrepData_Send(taskInfo, globalRank); break; - case ncclCollRecv: PrepData_Recv(taskInfo, globalRank); break; - default: - printf("Error: unsupported collective\n"); - exit(1); - } -} - -void PrepData_Broadcast(TaskInfo& taskInfo, int globalRank) { - // Only root needs input pattern - if (globalRank == taskInfo.root) - FillPattern(taskInfo.inputGpu, taskInfo.datatype, taskInfo.count, taskInfo.root, true); - - // Otherwise all other ranks expected output is the same as input of root - FillPattern(taskInfo.expected, taskInfo.datatype, taskInfo.count, taskInfo.root); -} - -void PrepData_Reduce(TaskInfo& taskInfo, int globalRank, int totalRanks, bool isAllReduce) { - size_t const numBytes = taskInfo.count * DataTypeToBytes(taskInfo.datatype); - - // If average or custom reduction operator is used, perform a summation instead - ncclRedOp_t const tempOp = (taskInfo.op >= ncclAvg ? ncclSum : taskInfo.op); - - for (int rank = 0; rank < totalRanks; ++rank) { - FillPattern(taskInfo.outputCpu, taskInfo.datatype, taskInfo.count, rank); - if (rank == globalRank) - HIP_CALL(hipMemcpy(taskInfo.inputGpu.ptr, taskInfo.outputCpu.ptr, numBytes, hipMemcpyHostToDevice)); - if (isAllReduce || taskInfo.root == globalRank) { - if (rank == 0) - memcpy(taskInfo.expected.ptr, taskInfo.outputCpu.ptr, numBytes); - else - Reduce(taskInfo.expected, taskInfo.outputCpu, taskInfo.count, taskInfo.datatype, tempOp); - } - } - - if (taskInfo.op == ncclAvg && (isAllReduce || taskInfo.root == globalRank)) - DivideByInt(taskInfo.expected, taskInfo.datatype, taskInfo.count, totalRanks); -} - -void PrepData_ReduceScatter(TaskInfo& taskInfo, int globalRank, int totalRanks) { - int const numInputElements = taskInfo.count * totalRanks; - int const numOutputElements = taskInfo.count; - std::pair numBytes = GetSize(taskInfo, totalRanks); - - PtrUnion tempInputCpu; - PtrUnion tempResultCpu; - AllocateMem(tempInputCpu, numBytes.first); - AllocateMem(tempResultCpu, numBytes.first); - - // If average or custom reduction operator is used, perform a summation instead - ncclRedOp_t const tempOp = (taskInfo.op >= ncclAvg ? ncclSum : taskInfo.op); - - for (int rank = 0; rank < totalRanks; ++rank) { - FillPattern(tempInputCpu, taskInfo.datatype, numInputElements, rank); - if (rank == globalRank) - HIP_CALL(hipMemcpy(taskInfo.inputGpu.ptr, tempInputCpu.ptr, numBytes.first, hipMemcpyHostToDevice)); - if (rank == 0) - memcpy(tempResultCpu.ptr, tempInputCpu.ptr, numBytes.first); - else - Reduce(tempResultCpu, tempInputCpu, numInputElements, taskInfo.datatype, tempOp); - } - - if (taskInfo.op == ncclAvg) - DivideByInt(tempResultCpu, taskInfo.datatype, numInputElements, totalRanks); - - memcpy(taskInfo.expected.I1, tempResultCpu.I1 + globalRank * numBytes.second, numBytes.second); - FreeMem(tempInputCpu); - FreeMem(tempResultCpu); -} - -void PrepData_Gather(TaskInfo& taskInfo, int globalRank, int totalRanks, bool isAllGather) { - int numInputElements = taskInfo.count; - int numOutputElements = totalRanks * taskInfo.count; - std::pair numBytes = GetSize(taskInfo, totalRanks); - - for (int rank = 0; rank < totalRanks; ++rank) { - FillPattern(taskInfo.outputCpu, taskInfo.datatype, numInputElements, rank); - if (rank == globalRank) - HIP_CALL(hipMemcpy(taskInfo.inputGpu.ptr, taskInfo.outputCpu.ptr, numBytes.first, hipMemcpyHostToDevice)); - if (isAllGather || taskInfo.root == globalRank) - memcpy(taskInfo.expected.I1 + (rank * numBytes.first), taskInfo.outputCpu.ptr, numBytes.first); - } -} - -void PrepData_Scatter(TaskInfo& taskInfo, int globalRank, int totalRanks) { - int const numInputElements = taskInfo.count * totalRanks; - int const numOutputElements = taskInfo.count; - std::pair numBytes = GetSize(taskInfo, totalRanks); - - PtrUnion tempInput; - AllocateMem(tempInput, numBytes.first); - - FillPattern(tempInput, taskInfo.datatype, numInputElements, taskInfo.root); - - if (globalRank == taskInfo.root) - HIP_CALL(hipMemcpy(taskInfo.inputGpu.ptr, tempInput.ptr, numBytes.first, hipMemcpyHostToDevice)); - - memcpy(taskInfo.expected.U1, tempInput.U1 + globalRank * numBytes.second, numBytes.second); - - FreeMem(tempInput); -} - -void PrepData_AlltoAll(TaskInfo& taskInfo, int globalRank, int totalRanks) { - int const numInputElements = taskInfo.count * totalRanks; - int const numOutputElements = numInputElements; - std::pair numBytes = GetSize(taskInfo, totalRanks); - size_t const numBytesPerRank = numBytes.first / totalRanks; - - for (int rank = 0; rank < totalRanks; ++rank) { - FillPattern(taskInfo.outputCpu, taskInfo.datatype, numInputElements, rank); - - if (rank == globalRank) - HIP_CALL(hipMemcpy(taskInfo.inputGpu.ptr, taskInfo.outputCpu.ptr, numBytes.first, hipMemcpyHostToDevice)); - - memcpy(taskInfo.expected.U1 + numBytesPerRank * rank, taskInfo.outputCpu.U1 + numBytesPerRank * globalRank, numBytesPerRank); - } -} - -void PrepData_Send(TaskInfo& taskInfo, int globalRank) { - FillPattern(taskInfo.inputGpu, taskInfo.datatype, taskInfo.count, globalRank, true); -} - -void PrepData_Recv(TaskInfo& taskInfo, int globalRank) { - FillPattern(taskInfo.expected, taskInfo.datatype, taskInfo.count, globalRank); -} +} \ No newline at end of file diff --git a/tools/RcclReplayer/rcclReplayer.hpp b/tools/RcclReplayer/rcclReplayer.hpp index 46834c7a45..86e9e8aec7 100644 --- a/tools/RcclReplayer/rcclReplayer.hpp +++ b/tools/RcclReplayer/rcclReplayer.hpp @@ -2,6 +2,8 @@ #pragma once #include +#include +#include #include #include @@ -9,6 +11,9 @@ #include "hip/hip_fp16.h" #include "rccl_float8.h" +#include "info.h" +#include "recorder.h" + // NOTE: Parsing is based on this line logging collective information in enqueue.cc // INFO(NCCL_COLL,"%s: opCount %lx sendbuff %p recvbuff %p count %zi datatype %d op %d \ root %d comm %p [nranks=%d] stream %p task %d globalrank %d", @@ -36,396 +41,64 @@ } \ } while(0) -struct LineItem +struct DeviceMemAllocation { - char hostname[MPI_MAX_PROCESSOR_NAME]; - int pid; - int tid; - int cudaDev; - char opName[32]; - int opCount; - char sendbuff[32]; - char recvbuff[32]; - size_t count; - int datatype; - int op; - int root; - char comm[32]; - int nRanks; - void* stream; - int task; - int globalRank; + void* base = NULL; + size_t size = 0; + int lastLineUsed = -1; }; -// Enumeration of all collective functions currently supported -typedef enum +struct DeviceGraphInfo { - ncclCollBroadcast = 0, - ncclCollReduce, - ncclCollAllGather, - ncclCollReduceScatter, - ncclCollAllReduce, - ncclCollGather, - ncclCollScatter, - ncclCollAllToAll, - ncclCollAllToAllv, - ncclCollSend, - ncclCollRecv, - ncclNumFuncs -} ncclFunc_t; + int depth = 0; + std::unordered_set + starts; + int end = -1; + hipStream_t stream = NULL; + hipGraph_t graph = NULL; + hipGraphExec_t graphExec = NULL; -char const ncclFuncNames[ncclNumFuncs][32] = -{ - "Broadcast", - "Reduce", - "AllGather", - "ReduceScatter", - "AllReduce", - "Gather", - "Scatter", - "AllToAll", - "AllToAllv", - "Send", - "Recv" + std::vector + events; + int counter = 0; }; -char const mscclFuncNames[ncclNumFuncs][32] = +class Replayer { - "mscclFuncBroadcast", - "mscclFuncReduce", - "mscclFuncAllGather", - "mscclFuncReduceScatter", - "mscclFuncAllReduce", - "mscclFuncGather", - "mscclFuncScatter", - "mscclFuncAllToAll", - "mscclFuncAllToAllv", - "mscclFuncSend", - "mscclFuncRecv" -}; + private: + // rank specific info + int myRank; + int numGlobalRanks; + /// int numGpusPerMpiRank; + /// int localGpuOffset; // First local GPU device idx for this MPI process + /// int firstGlobalRank; // First global rank for this MPI process + std::ifstream log; -union PtrUnion -{ - void* ptr; - int8_t* I1; // ncclInt8 - uint8_t* U1; // ncclUint8 - int32_t* I4; // ncclInt32 - uint32_t* U4; // ncclUint32 - int64_t* I8; // ncclInt64 - uint64_t* U8; // ncclUint64 - __half* F2; // ncclFloat16 - rccl_float8* F1; // ncclFloat8e4m3 - float* F4; // ncclFloat32 - double* F8; // ncclFloat64 - rccl_bfloat8* B1; // ncclFloat8e5m2 - hip_bfloat16* B2; // ncclBfloat16 + // Contextual info parsed from first pass, to assist replay later + // Communicator + std::vector Ids; // all communicators (uniqueIDs) created from commInit, assuming called once only + std::unordered_map> idRankMap; // all ranks in the communicator created by an ID on this rank - constexpr PtrUnion() : ptr(nullptr) {} -}; -struct TaskInfo -{ - ncclFunc_t funcType; - bool inPlace; - size_t count; - ncclDataType_t datatype; - ncclRedOp_t op; - int root; - PtrUnion inputGpu; - PtrUnion outputCpu; - PtrUnion outputGpu; - PtrUnion expected; -}; + // Memory allocation and lifespan + std::unordered_map dMemMap; -struct RankData -{ - int lineNum; - int commIdx; - std::vector tasks; -}; + // Resources for replayer, mostly maps from pointer in log to resources in replay time + std::unordered_map idMap; // replayer uniqueID mapped to logged ones, for ID creator rank only + std::unordered_map commMap; // replayer communicator mapped to the logged ones -struct GroupCall -{ - bool isValid; - int opCount; - std::map rankData; -}; + std::unordered_map> + streams; // replayer streams mapped to the logged ones // use using later? + std::unordered_map handleMap; // UBR handle + std::unordered_map + graphLife; // when does a graph (graphID) end and how many node it contains -struct CollectiveCalls -{ - int numGlobalRanks; - int numGpusPerMpiRank; - std::vector> globalRankComms; // Set of comms used by each global rank - std::vector groupCalls; // List of group calls for each global rank + // auxiliary variables for replayer + ncclUniqueId uniqueID; + rccl::rcclCall_t lastCall; - int localGpuOffset; // First local GPU device idx for this MPI process - int firstGlobalRank; // First global rank for this MPI process - int numCommsPerRank; // Number of communicators per rank - std::vector> localRankComms; // comms per local rank - std::vector> localRankStreams; // streams per local rank -}; - -std::string DataTypeToName(ncclDataType_t const dataType) -{ - switch (dataType) { - case ncclInt8: return "Int8"; - case ncclUint8: return "Uint8"; - case ncclInt32: return "Int32"; - case ncclUint32: return "Uint32"; - case ncclInt64: return "Int64"; - case ncclUint64: return "Uint64"; - case ncclFloat16: return "Float16"; - case ncclFloat32: return "Float32"; - case ncclFloat64: return "Float64"; - case ncclBfloat16: return "Bfloat16"; - case ncclFloat8e4m3: return "Fp8E4M3"; - case ncclFloat8e5m2: return "Fp8E5M2"; - default: - printf("Unsupported datatype (%d)\n", dataType); - exit(0); - } -} - -size_t DataTypeToBytes(ncclDataType_t const dataType) -{ - switch (dataType) { - case ncclInt8: return 1; - case ncclUint8: return 1; - case ncclInt32: return 4; - case ncclUint32: return 4; - case ncclInt64: return 8; - case ncclUint64: return 8; - case ncclFloat16: return 2; - case ncclFloat32: return 4; - case ncclFloat64: return 8; - case ncclBfloat16: return 2; - case ncclFloat8e4m3: return 1; - case ncclFloat8e5m2: return 1; - default: - printf("Unsupported datatype (%s)\n", DataTypeToName(dataType).c_str()); - exit(0); - } -} - -std::string RedOpToName(ncclRedOp_t const op) -{ - switch (op) { - case ncclSum: return "Sum"; - case ncclProd: return "Product"; - case ncclMax: return "Max"; - case ncclMin: return "Min"; - case ncclAvg: return "Average"; - case ncclNumOps: return "Number of built-in reduction ops"; - case ncclMaxRedOp: return "Largest value for ncclRedOp_t"; - default: - printf("Unsupported redOp (%d)\n", op); - exit(0); - } -} - -ncclFunc_t GetFuncType(char* func) -{ - for (int i = 0; i < ncclNumFuncs; i++) - if (!strcmp(func, ncclFuncNames[i]) || !strcmp(func, mscclFuncNames[i])) return (ncclFunc_t)i; - printf("[ERROR] Unrecognized func %s\n", func); - exit(1); -} - -// Set data for ptrUnion (Used during fillPattern) -void SetPtr(PtrUnion& ptrUnion, ncclDataType_t const dataType, int const idx, int valueI, double valueF) { - switch (dataType) - { - case ncclInt8: ptrUnion.I1[idx] = valueI; break; - case ncclUint8: ptrUnion.U1[idx] = valueI; break; - case ncclInt32: ptrUnion.I4[idx] = valueI; break; - case ncclUint32: ptrUnion.U4[idx] = valueI; break; - case ncclInt64: ptrUnion.I8[idx] = valueI; break; - case ncclUint64: ptrUnion.U8[idx] = valueI; break; - case ncclFloat8e4m3: ptrUnion.F1[idx] = rccl_float8(valueF); break; - case ncclFloat16: ptrUnion.F2[idx] = __float2half(static_cast(valueF)); break; - case ncclFloat32: ptrUnion.F4[idx] = valueF; break; - case ncclFloat64: ptrUnion.F8[idx] = valueF; break; - case ncclFloat8e5m2: ptrUnion.B1[idx] = rccl_bfloat8(valueF); break; - case ncclBfloat16: ptrUnion.B2[idx] = hip_bfloat16(static_cast(valueF)); break; - default: - printf("Unsupported datatype (%s)\n", DataTypeToName(dataType).c_str()); - exit(0); - } -} - -// Check if each element in actual equals to expected -bool IsEqual(PtrUnion const& actual, PtrUnion const& expected, ncclDataType_t const dataType, size_t const numElements, int const globalRank) { - bool isMatch = true; - size_t idx = 0; - for (idx = 0; idx < numElements; ++idx) - { - switch (dataType) - { - case ncclInt8: isMatch = (actual.I1[idx] == expected.I1[idx]); break; - case ncclUint8: isMatch = (actual.U1[idx] == expected.U1[idx]); break; - case ncclInt32: isMatch = (actual.I4[idx] == expected.I4[idx]); break; - case ncclUint32: isMatch = (actual.U4[idx] == expected.U4[idx]); break; - case ncclInt64: isMatch = (actual.I8[idx] == expected.I8[idx]); break; - case ncclUint64: isMatch = (actual.U8[idx] == expected.U8[idx]); break; - case ncclFloat8e4m3: isMatch = (fabs(float(actual.F1[idx]) - float(expected.F1[idx])) < 9e-2); break; - case ncclFloat16: isMatch = (fabs(__half2float(actual.F2[idx]) - __half2float(expected.F2[idx])) < 9e-2); break; - case ncclFloat32: isMatch = (fabs(actual.F4[idx] - expected.F4[idx]) < 1e-5); break; - case ncclFloat64: isMatch = (fabs(actual.F8[idx] - expected.F8[idx]) < 1e-12); break; - case ncclFloat8e5m2: isMatch = (fabs(float(actual.B1[idx]) - float(expected.B1[idx])) < 9e-2); break; - case ncclBfloat16: isMatch = (fabs((float)actual.B2[idx] - (float)expected.B2[idx]) < 9e-2); break; - default: - printf("Unsupported datatype (%s)\n", DataTypeToName(dataType).c_str()); - isMatch = false; - } - if (!isMatch) { - switch (dataType) - { - case ncclInt8: - printf("[Error Rank = %d] Expected output: %d. Actual output: %d at index %lu\n", globalRank, expected.I1[idx], actual.I1[idx], idx); break; - case ncclUint8: - printf("[Error Rank = %d] Expected output: %u. Actual output: %u at index %lu\n", globalRank, expected.U1[idx], actual.U1[idx], idx); break; - case ncclInt32: - printf("[Error Rank = %d] Expected output: %d. Actual output: %d at index %lu\n", globalRank, expected.I4[idx], actual.I4[idx], idx); break; - case ncclUint32: - printf("[Error Rank = %d] Expected output: %u. Actual output: %u at index %lu\n", globalRank, expected.U4[idx], actual.U4[idx], idx); break; - case ncclInt64: - printf("[Error Rank = %d] Expected output: %ld. Actual output: %ld at index %lu\n", globalRank, expected.I8[idx], actual.I8[idx], idx); break; - case ncclUint64: - printf("[Error Rank = %d] Expected output: %lu. Actual output: %lu at index %lu\n", globalRank, expected.U8[idx], actual.U8[idx], idx); break; - case ncclFloat8e4m3: - printf("[Error Rank = %d] Expected output: %f. Actual output: %f at index %lu\n", globalRank, (float)expected.F1[idx], (float)actual.F1[idx], idx); break; - case ncclFloat16: - printf("[Error Rank = %d] Expected output: %f. Actual output: %f at index %lu\n", globalRank, __half2float(expected.F2[idx]), __half2float(actual.F2[idx]), idx); break; - case ncclFloat32: - printf("[Error Rank = %d] Expected output: %f. Actual output: %f at index %lu\n", globalRank, expected.F4[idx], actual.F4[idx], idx); break; - case ncclFloat64: - printf("[Error Rank = %d] Expected output: %lf. Actual output: %lf at index %lu\n", globalRank, expected.F8[idx], actual.F8[idx], idx); break; - case ncclFloat8e5m2: - printf("[Error Rank = %d] Expected output: %f. Actual output: %f at index %lu\n", globalRank, (float)expected.B1[idx], (float)actual.B1[idx], idx); break; - case ncclBfloat16: - printf("[Error Rank = %d] Expected output: %f. Actual output: %f at index %lu\n", globalRank, (float)expected.B2[idx], (float)actual.B2[idx], idx); break; - default: - break; - } - return isMatch; - } - } - - return isMatch; -} - -// Performs the various basic reduction operations -template -T ReduceOp(ncclRedOp_t const op, T const A, T const B) -{ - switch (op) - { - case ncclSum: return A + B; - case ncclProd: return A * B; - case ncclMax: return std::max(A, B); - case ncclMin: return std::min(A, B); - default: - printf("Unsupported reduction operator (%s)\n", RedOpToName(op).c_str()); - exit(0); - } -} - -// Perform various reduction ops to ptrUnion -void Reduce(PtrUnion& ptrUnion, PtrUnion const& otherPtrUnion, size_t const numElements, ncclDataType_t const dataType, ncclRedOp_t const op) { - for (size_t idx = 0; idx < numElements; ++idx) - { - switch (dataType) - { - case ncclInt8: ptrUnion.I1[idx] = ReduceOp(op, ptrUnion.I1[idx], otherPtrUnion.I1[idx]); break; - case ncclUint8: ptrUnion.U1[idx] = ReduceOp(op, ptrUnion.U1[idx], otherPtrUnion.U1[idx]); break; - case ncclInt32: ptrUnion.I4[idx] = ReduceOp(op, ptrUnion.I4[idx], otherPtrUnion.I4[idx]); break; - case ncclUint32: ptrUnion.U4[idx] = ReduceOp(op, ptrUnion.U4[idx], otherPtrUnion.U4[idx]); break; - case ncclInt64: ptrUnion.I8[idx] = ReduceOp(op, ptrUnion.I8[idx], otherPtrUnion.I8[idx]); break; - case ncclUint64: ptrUnion.U8[idx] = ReduceOp(op, ptrUnion.U8[idx], otherPtrUnion.U8[idx]); break; - case ncclFloat8e4m3: ptrUnion.F1[idx] = rccl_float8(ReduceOp(op, float(ptrUnion.F1[idx]), float(otherPtrUnion.F1[idx]))); break; - case ncclFloat16: ptrUnion.F2[idx] = __float2half(ReduceOp(op, __half2float(ptrUnion.F2[idx]), __half2float(otherPtrUnion.F2[idx]))); break; - case ncclFloat32: ptrUnion.F4[idx] = ReduceOp(op, ptrUnion.F4[idx], otherPtrUnion.F4[idx]); break; - case ncclFloat64: ptrUnion.F8[idx] = ReduceOp(op, ptrUnion.F8[idx], otherPtrUnion.F8[idx]); break; - case ncclFloat8e5m2: ptrUnion.B1[idx] = rccl_bfloat8(ReduceOp(op, float(ptrUnion.B1[idx]), float(otherPtrUnion.B1[idx]))); break; - case ncclBfloat16: ptrUnion.B2[idx] = ReduceOp(op, ptrUnion.B2[idx], otherPtrUnion.B2[idx]); break; - default: - printf("Unsupported datatype (%s)\n", DataTypeToName(dataType).c_str()); - exit(0); - } - } -} - -// Divide each element in ptrUnion by divisor -void DivideByInt(PtrUnion& ptrUnion, ncclDataType_t const dataType, size_t const numElements, int const divisor) { - for (size_t idx = 0; idx < numElements; ++idx) - { - switch (dataType) - { - case ncclInt8: ptrUnion.I1[idx] /= divisor; break; - case ncclUint8: ptrUnion.U1[idx] /= divisor; break; - case ncclInt32: ptrUnion.I4[idx] /= divisor; break; - case ncclUint32: ptrUnion.U4[idx] /= divisor; break; - case ncclInt64: ptrUnion.I8[idx] /= divisor; break; - case ncclUint64: ptrUnion.U8[idx] /= divisor; break; - case ncclFloat8e4m3: ptrUnion.F1[idx] = (rccl_float8((float)(ptrUnion.F1[idx]) / divisor)); break; - case ncclFloat16: ptrUnion.F2[idx] = __float2half(__half2float(ptrUnion.F2[idx])/divisor); break; - case ncclFloat32: ptrUnion.F4[idx] /= divisor; break; - case ncclFloat64: ptrUnion.F8[idx] /= divisor; break; - case ncclFloat8e5m2: ptrUnion.B1[idx] = (rccl_bfloat8((float)(ptrUnion.B1[idx]) / divisor)); break; - case ncclBfloat16: ptrUnion.B2[idx] = (hip_bfloat16((float)(ptrUnion.B2[idx]) / divisor)); break; - default: - printf("Unsupported datatype (%s)\n", DataTypeToName(dataType).c_str()); - exit(0); - } - } -} - -// Check if a collective uses a root -bool IsRootUsed(ncclFunc_t funcType) { - return (funcType == ncclCollBroadcast || funcType == ncclCollReduce || - 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, -// validatation of the groupCalls for both non-send/recv collectives and send/recv -void ParseCollectives(char const* logFilename, bool isFirstRank, CollectiveCalls& collectiveCalls); - -// allocates send/recv buff, sets the device based on which rank the task belongs to, -// syncronize devices after executing all the tasks and free device memory. -double ReplayRccl(CollectiveCalls& collCall, int groupIdx, int& numInvalid); - -// Print information about a group call -void PrintGroupCall(GroupCall const& gc); - -// Records performance data of each group call in a csv file named replayer_data.csv -void dataToCsv(GroupCall const& gc, std::ofstream &datafile, double runTime); - -// size differ for each collective call and getSize gives a specific size in bytes depending on type of task, -// global rank, element count and data type -std::pair GetSize(TaskInfo taskInfo, int numGlobalRanks); - -// executes the collective call (task) -void ExecuteCollective(TaskInfo& task, ncclComm_t const& comm, hipStream_t stream); - -// Allocate CPU/GPU memory for ptrUnion -void AllocateMem(PtrUnion& ptrUnion, size_t const numBytes, bool isGpu = false); - -// Free CPU/GPU memory for ptrUnion -void FreeMem(PtrUnion& ptrUnion, bool isGpu = false); - -// Fill buffers based on pattern using globalRank -void FillPattern(PtrUnion& ptrUnion, ncclDataType_t const dataType, size_t const numElements, int globalRank, bool isGpu = false); - -// PrepareData functions are responsible for setting up input / expected for the given taskInfo -void PrepareDataFunc(TaskInfo& taskInfo, int globalRank, int totalRanks); -void PrepData_Broadcast(TaskInfo& taskInfo, int globalRank); -void PrepData_Reduce(TaskInfo& taskInfo, int globalRank, int totalRanks, bool isAllReduce); -void PrepData_ReduceScatter(TaskInfo& taskInfo, int globalRank, int totalRanks); -void PrepData_Gather(TaskInfo& taskInfo, int globalRank, int totalRanks, bool isAllGather); -void PrepData_Scatter(TaskInfo& taskInfo, int globalRank, int totalRanks); -void PrepData_AlltoAll(TaskInfo& taskInfo, int globalRank, int totalRanks); -void PrepData_Send(TaskInfo& taskInfo, int globalRank); -void PrepData_Recv(TaskInfo& taskInfo, int globalRank); + public: + Replayer(const std::string& logname, int json_format, int rank, int size); + void parse(); + void replay(); +}; \ No newline at end of file