Add collective latency profiler (#1785)
* [LatencyProfiler] Initial commit
* [LatencyProfiler] Add unit tests
* [LatencyProfiler] add more
* [LatencyProfiler] Pass unit tests
* [LatencyProfiler] Add hooks to integrate with meta internal tools
* [LatencyProfiler] Restore install.sh
* [LatencyProfiler] Resolved comments 1. add proper license 2. use proper namespace
* [LatencyProfiler] Add header
[ROCm/rccl commit: 874cd657ef]
This commit is contained in:
ycui1984
GitHub
@@ -0,0 +1,228 @@
|
||||
/*
|
||||
* Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
*
|
||||
* This source code is licensed under the MIT license found in the
|
||||
* LICENSE file in the root directory of this source tree.
|
||||
*/
|
||||
#include "latency_profiler/CollTrace.h"
|
||||
#include "bootstrap.h"
|
||||
#include "checks.h"
|
||||
#include "comm.h"
|
||||
#include "param.h"
|
||||
|
||||
NCCL_PARAM(ColltraceRecordMax, "COLLTRACE_RECORD_MAX", 100);
|
||||
NCCL_PARAM(ColltraceMaxDumpSize, "COLLTRACE_MAX_DUMP_SIZE", 20);
|
||||
NCCL_PARAM(ColltraceDumpIntervalSec, "COLLTRACE_DUMP_INTERVAL_SEC", 300);
|
||||
|
||||
constexpr int RANKS_PER_HOST = 8;
|
||||
|
||||
namespace latency_profiler {
|
||||
|
||||
namespace {
|
||||
CudaEventPtr getCudaEventPtr() {
|
||||
cudaEvent_t newEvent = nullptr;
|
||||
CUDACHECKIGNORE(cudaEventCreate(&newEvent));
|
||||
CudaEventPtr item(newEvent);
|
||||
return item;
|
||||
}
|
||||
} // namespace
|
||||
|
||||
CollTrace::CollTrace(ncclComm* comm)
|
||||
: comm_(comm),
|
||||
commHash_(std::to_string(comm->commHash)),
|
||||
rank_(comm->rank) {
|
||||
profilingWorkerThread_ =
|
||||
std::thread{[this]() { return collTraceThreadFn(comm_->cudaDev); }};
|
||||
}
|
||||
|
||||
CollTrace::~CollTrace() {
|
||||
try {
|
||||
INFO(
|
||||
NCCL_INIT,
|
||||
"COLLTRACE: commHash %s rank %d - Destroy START",
|
||||
commHash_.c_str(),
|
||||
rank_);
|
||||
eventQueue_.push(std::unique_ptr<CollTraceEvent>(
|
||||
new CollTraceEvent(CollTraceEvent::EventType::TERMINATE)));
|
||||
if (profilingWorkerThread_.joinable()) {
|
||||
profilingWorkerThread_.join();
|
||||
}
|
||||
|
||||
if (rank_ == 0) {
|
||||
reportIfNeeded(false);
|
||||
}
|
||||
|
||||
INFO(
|
||||
NCCL_INIT,
|
||||
"COLLTRACE: commHash %s rank %d - Destroy COMPLETE",
|
||||
commHash_.c_str(),
|
||||
rank_);
|
||||
} catch (const std::exception& e) {
|
||||
WARN(
|
||||
"COLLTRACE: commHash %s rank %d - Destroy FAILED: %s",
|
||||
commHash_.c_str(),
|
||||
rank_,
|
||||
e.what());
|
||||
}
|
||||
}
|
||||
|
||||
void* CollTrace::collTraceThreadFn(int cudaDev) {
|
||||
INFO(NCCL_INIT, "CollTrace thread started for cudaDev %d", cudaDev);
|
||||
auto err = cudaSetDevice(cudaDev);
|
||||
if (err != cudaSuccess) {
|
||||
WARN("Cuda failure '%s'", cudaGetErrorString(err));
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
lastReportTime_ = std::chrono::steady_clock::now();
|
||||
|
||||
INFO(
|
||||
NCCL_INIT,
|
||||
"COLLTRACE: commHash %s rank %d - worker thread STARTED",
|
||||
commHash_.c_str(),
|
||||
rank_);
|
||||
while (true) {
|
||||
curEvent_ = eventQueue_.waitPop();
|
||||
if (curEvent_->eventType == CollTraceEvent::EventType::TERMINATE) {
|
||||
break;
|
||||
}
|
||||
curEvent_->start->waitEventFinish();
|
||||
auto ncclRes = curEvent_->stop->waitEventFinish();
|
||||
float latency = -1;
|
||||
|
||||
if (ncclRes == ncclSuccess) {
|
||||
auto latencyMaybe =
|
||||
curEvent_->stop->getElapsedTimeSinceEvent(curEvent_->start.get());
|
||||
// latencyMaybe could be nullopt when cudaEventElapsedTime failed
|
||||
// this could happen when events are not recorded or stream is not valid
|
||||
if (latencyMaybe == nullptr) {
|
||||
WARN(
|
||||
"CollTrace: getElapsedTimeSinceEvent failed, aborting worker thread");
|
||||
return nullptr;
|
||||
}
|
||||
latency = *latencyMaybe;
|
||||
}
|
||||
|
||||
recordCurCollResult(cudaDev, latency);
|
||||
curEvent_.reset();
|
||||
}
|
||||
|
||||
INFO(
|
||||
NCCL_INIT,
|
||||
"COLLTRACE: commHash %s rank %d - worker thread TERMINATE",
|
||||
commHash_.c_str(),
|
||||
rank_);
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
void CollTrace::enqueueEvent(std::unique_ptr<CollTraceEvent> event) {
|
||||
event->coll.collId = curCollId_.fetch_add(1);
|
||||
eventQueue_.push(std::move(event));
|
||||
}
|
||||
|
||||
std::unique_ptr<CollTraceEvent> CollTrace::createEvent(
|
||||
CollTraceEvent::EventType type) {
|
||||
auto eventInfo = std::make_unique<CollTraceEvent>(type);
|
||||
eventInfo->start = std::make_unique<CudaWaitEvent>(getCudaEventPtr());
|
||||
eventInfo->stop = std::make_unique<CudaWaitEvent>(getCudaEventPtr());
|
||||
|
||||
if (!eventInfo->start || !eventInfo->stop) {
|
||||
std::unique_ptr<CollTraceEvent> nullCollTraceEvent(nullptr);
|
||||
return nullCollTraceEvent;
|
||||
}
|
||||
return eventInfo;
|
||||
}
|
||||
|
||||
bool shouldAggregateRingBuffer(int collId) {
|
||||
const int NCCL_COLLTRACE_RECORD_MAX = ncclParamColltraceRecordMax();
|
||||
return ((collId + 1) % NCCL_COLLTRACE_RECORD_MAX == 0);
|
||||
}
|
||||
|
||||
void CollTrace::reportIfNeeded(bool checkInterval = true) {
|
||||
auto now = std::chrono::steady_clock::now();
|
||||
auto secs_passed =
|
||||
std::chrono::duration_cast<std::chrono::seconds>(now - lastReportTime_)
|
||||
.count();
|
||||
if (checkInterval) {
|
||||
if (secs_passed < ncclParamColltraceDumpIntervalSec() &&
|
||||
stats_.size() < ncclParamColltraceMaxDumpSize()) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
INFO(
|
||||
NCCL_COLL,
|
||||
"CollTrace: %ld seconds passed since last report, stats size = %zu, checkInterval = %d",
|
||||
secs_passed,
|
||||
stats_.size(),
|
||||
checkInterval);
|
||||
|
||||
// reportToScuba is a placeholder for oss environment.
|
||||
// meta production reports to scuba instead of file, which enables
|
||||
// filering, aggregation and visualization.
|
||||
#ifdef ENABLE_SCUBA_LOGGING
|
||||
reportToScuba(stats_, commHash_);
|
||||
#else
|
||||
reportToFile(stats_, commHash_);
|
||||
#endif
|
||||
lastReportTime_ = std::chrono::steady_clock::now();
|
||||
stats_.clear();
|
||||
}
|
||||
|
||||
void CollTrace::recordCurCollResult(int rank, float latency) {
|
||||
const int NCCL_COLLTRACE_RECORD_MAX = ncclParamColltraceRecordMax();
|
||||
|
||||
auto result = std::make_unique<CollTraceInfo>(curEvent_->coll);
|
||||
auto collId = result->collId;
|
||||
result->latencyMs = latency;
|
||||
|
||||
pastColls_.push_back(std::move(result));
|
||||
if (pastColls_.size() > NCCL_COLLTRACE_RECORD_MAX) {
|
||||
pastColls_.pop_front();
|
||||
}
|
||||
|
||||
if (shouldAggregateRingBuffer(collId) && pastColls_.size() > 0) {
|
||||
std::vector<float> latencyAllGather;
|
||||
latencyAllGather.resize(RANKS_PER_HOST * NCCL_COLLTRACE_RECORD_MAX, 0);
|
||||
int start = (comm_->localRank) * NCCL_COLLTRACE_RECORD_MAX;
|
||||
for (int i = start; i < start + NCCL_COLLTRACE_RECORD_MAX; i++) {
|
||||
latencyAllGather[i] = pastColls_[i - start]->latencyMs;
|
||||
}
|
||||
auto before = std::chrono::high_resolution_clock::now();
|
||||
auto ncclResult = bootstrapIntraNodeAllGather(
|
||||
comm_->bootstrap,
|
||||
comm_->localRankToRank,
|
||||
comm_->localRank,
|
||||
comm_->localRanks,
|
||||
latencyAllGather.data(),
|
||||
NCCL_COLLTRACE_RECORD_MAX * sizeof(float));
|
||||
auto after = std::chrono::high_resolution_clock::now();
|
||||
auto interval_us =
|
||||
std::chrono::duration_cast<std::chrono::microseconds>(after - before)
|
||||
.count();
|
||||
if (ncclResult != ncclSuccess) {
|
||||
WARN("CollTrace: All gather exchange latency data failed");
|
||||
return;
|
||||
}
|
||||
|
||||
if (rank == 0) {
|
||||
INFO(NCCL_COLL, "latency metrics all gather takes %ld us", interval_us);
|
||||
try {
|
||||
auto stats = aggregateResults(
|
||||
pastColls_,
|
||||
latencyAllGather,
|
||||
RANKS_PER_HOST,
|
||||
NCCL_COLLTRACE_RECORD_MAX);
|
||||
stats_.push_back(stats);
|
||||
if (stats_.size() > ncclParamColltraceMaxDumpSize()) {
|
||||
stats_.pop_front();
|
||||
}
|
||||
reportIfNeeded();
|
||||
} catch (const std::exception& e) {
|
||||
WARN("Aggregating error: %s", e.what());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace latency_profiler
|
||||
@@ -0,0 +1,42 @@
|
||||
/*
|
||||
* Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
*
|
||||
* This source code is licensed under the MIT license found in the
|
||||
* LICENSE file in the root directory of this source tree.
|
||||
*/
|
||||
#include <thread>
|
||||
#include "latency_profiler/CollTraceEvent.h"
|
||||
#include "param.h"
|
||||
|
||||
NCCL_PARAM(ColltraceCheckIntervalMs, "COLLTRACE_CHECK_INTERVAL_MS", 10);
|
||||
|
||||
namespace latency_profiler {
|
||||
|
||||
ncclResult_t CudaWaitEvent::waitEventFinish() {
|
||||
// async polling case, query cuda whether event is ready every
|
||||
// NCCL_COLLTRACE_CHECK_INTERVAL_MS ms
|
||||
auto res = cudaEventQuery(event_.get());
|
||||
while (res != cudaSuccess) {
|
||||
if (res != cudaErrorNotReady) {
|
||||
CUDACHECK(res);
|
||||
}
|
||||
std::this_thread::sleep_for(
|
||||
std::chrono::milliseconds(ncclParamColltraceCheckIntervalMs()));
|
||||
res = cudaEventQuery(event_.get());
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
std::shared_ptr<float> CudaWaitEvent::getElapsedTimeSinceEvent(
|
||||
CudaWaitEvent* start) {
|
||||
float elapsedTime;
|
||||
auto res =
|
||||
cudaEventElapsedTime(&elapsedTime, start->event_.get(), event_.get());
|
||||
if (res != cudaSuccess) {
|
||||
WARN("get elapsed time failed error: %s", cudaGetErrorString(res));
|
||||
return nullptr;
|
||||
}
|
||||
return std::make_shared<float>(elapsedTime);
|
||||
}
|
||||
|
||||
} // namespace latency_profiler
|
||||
@@ -0,0 +1,140 @@
|
||||
/*
|
||||
* Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
*
|
||||
* This source code is licensed under the MIT license found in the
|
||||
* LICENSE file in the root directory of this source tree.
|
||||
*/
|
||||
#include "latency_profiler/CollTraceFunc.h"
|
||||
|
||||
namespace latency_profiler {
|
||||
|
||||
namespace {
|
||||
bool enableCollTrace() {
|
||||
const char* colltraceEnable = ncclGetEnv("RCCL_LATENCY_PROFILER");
|
||||
if (colltraceEnable != NULL) {
|
||||
INFO(
|
||||
NCCL_INIT,
|
||||
"RCCL_LATENCY_PROFILER set by environment to %s.",
|
||||
colltraceEnable);
|
||||
if (strcmp(colltraceEnable, "1") == 0) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
} // namespace
|
||||
|
||||
ncclResult_t collTraceInit(ncclComm* comm) {
|
||||
if (!enableCollTrace()) {
|
||||
return ncclSuccess;
|
||||
}
|
||||
comm->ctrace = std::make_unique<CollTrace>(comm);
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t collTraceDestroy(ncclComm* comm) {
|
||||
if (comm->ctrace == nullptr) {
|
||||
return ncclSuccess;
|
||||
}
|
||||
comm->ctrace.reset();
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t collTraceRecordStartEvent(
|
||||
ncclComm* comm,
|
||||
cudaStream_t launchStream,
|
||||
CollTraceEvent* event) {
|
||||
if (comm->ctrace && event) {
|
||||
CUDACHECK(
|
||||
cudaEventRecord(event->start.get()->getCudaEvent(), launchStream));
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t collTraceRecordEndEvent(
|
||||
ncclComm* comm,
|
||||
ncclKernelPlan* plan,
|
||||
cudaStream_t launchStream,
|
||||
std::unique_ptr<CollTraceEvent> event) {
|
||||
if (comm->ctrace && event) {
|
||||
CUDACHECK(cudaEventRecord(event->stop.get()->getCudaEvent(), launchStream));
|
||||
comm->ctrace->enqueueEvent(std::move(event));
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
CollTraceInfo parseCollInfoFromCollTask(const ncclTaskColl& collTask) {
|
||||
return CollTraceInfo{
|
||||
.opName = std::string{ncclFuncToString(collTask.func)},
|
||||
.dataType = std::string{ncclDatatypeToString(collTask.datatype)},
|
||||
.count = (int64_t)collTask.count,
|
||||
};
|
||||
}
|
||||
|
||||
std::shared_ptr<CollTraceInfo> parseCollInfoFromNcclKernelPlan(
|
||||
ncclKernelPlan& plan,
|
||||
cudaStream_t stream) {
|
||||
if (plan.comm == nullptr || plan.comm->ctrace == nullptr) {
|
||||
return nullptr;
|
||||
}
|
||||
auto collTaskHead = ncclIntruQueueHead(&plan.collTaskQueue);
|
||||
if (collTaskHead == nullptr) {
|
||||
WARN("CollTrace: no coll task in this plan, this plan is empty");
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
CollTraceInfo collInfo = parseCollInfoFromCollTask(*collTaskHead);
|
||||
return std::make_shared<CollTraceInfo>(collInfo);
|
||||
}
|
||||
|
||||
std::unique_ptr<CollTraceEvent> collTraceAquireEventCommon(
|
||||
ncclComm* comm,
|
||||
CollTraceEvent::EventType type,
|
||||
cudaStream_t stream) {
|
||||
if (!comm->ctrace) {
|
||||
return nullptr;
|
||||
}
|
||||
struct ncclCudaGraph graph;
|
||||
auto res = ncclCudaGetCapturingGraph(&graph, stream);
|
||||
if (res != ncclSuccess) {
|
||||
WARN("Internal error: ncclCudaGetCapturingGraph failed by %d", res);
|
||||
return nullptr;
|
||||
}
|
||||
if (graph.graph != nullptr) {
|
||||
// We are in a cuda graph, this is currently unsupported
|
||||
WARN(
|
||||
"COLLTRACE: does not support cuda graph. Collectives from comm %lx will be skipped",
|
||||
comm->commHash);
|
||||
return nullptr;
|
||||
}
|
||||
auto event = comm->ctrace->createEvent(type);
|
||||
if (!event) {
|
||||
throw CollTraceError("Event init failed");
|
||||
return nullptr; /*Event init failed*/
|
||||
}
|
||||
return event;
|
||||
}
|
||||
|
||||
std::unique_ptr<CollTraceEvent> collTraceAquireEventBaseline(
|
||||
ncclKernelPlan* plan,
|
||||
cudaStream_t stream) {
|
||||
auto collPtr = parseCollInfoFromNcclKernelPlan(*plan, stream);
|
||||
if (collPtr == nullptr) {
|
||||
return nullptr;
|
||||
}
|
||||
auto comm = plan->comm;
|
||||
if (!comm->ctrace) {
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
auto event =
|
||||
collTraceAquireEventCommon(comm, CollTraceEvent::EventType::COMM, stream);
|
||||
if (event == nullptr) {
|
||||
WARN("COLLTRACE: failed to aquire event");
|
||||
return nullptr;
|
||||
}
|
||||
event->coll = *collPtr;
|
||||
return event;
|
||||
}
|
||||
|
||||
} // namespace latency_profiler
|
||||
@@ -0,0 +1,87 @@
|
||||
/*
|
||||
* Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
*
|
||||
* This source code is licensed under the MIT license found in the
|
||||
* LICENSE file in the root directory of this source tree.
|
||||
*/
|
||||
#include "latency_profiler/CollTraceUtils.h"
|
||||
#include "nccl_common.h"
|
||||
#include "debug.h"
|
||||
|
||||
namespace latency_profiler {
|
||||
|
||||
float getSizeMb(const std::string& dataType, int count) {
|
||||
if (dataType == "ncclInt8" || dataType == "ncclFp8E4M3" ||
|
||||
dataType == "ncclFp8E5M2") {
|
||||
return count / 1024.0 / 1024.0;
|
||||
}
|
||||
|
||||
if (dataType == "ncclFloat16" || dataType == "ncclBfloat16") {
|
||||
return 2 * count / 1024.0 / 1024.0;
|
||||
}
|
||||
|
||||
if (dataType == "ncclInt32" || dataType == "ncclUint32" ||
|
||||
dataType == "ncclFloat32") {
|
||||
return 4 * count / 1024.0 / 1024.0;
|
||||
}
|
||||
|
||||
if (dataType == "ncclInt64" || dataType == "ncclUint64" ||
|
||||
dataType == "ncclFloat64") {
|
||||
return 8 * count / 1024.0 / 1024.0;
|
||||
}
|
||||
|
||||
throw std::runtime_error("CollTrace: unsupported data type " + dataType);
|
||||
}
|
||||
|
||||
void reportToFile(
|
||||
const std::deque<std::vector<CollStats>>& stats,
|
||||
const std::string& commHash) {
|
||||
for (const auto& oneDumpStats : stats) {
|
||||
for (const auto& elem : oneDumpStats) {
|
||||
auto size_mb = getSizeMb(elem.dataType, elem.count);
|
||||
INFO(NCCL_COLL, "coll_id %ld, percent %d, min_latency_us %f, max_latency_us %f, op_name %s, data_type %s, count %ld, message_size_MB %f, comm_hash %s", elem.collId, elem.percent, elem.minLatencyUs, elem.maxLatencyUs, elem.opName.c_str(), elem.dataType.c_str(), elem.count, size_mb, commHash.c_str());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
std::vector<CollStats> aggregateResults(
|
||||
const std::deque<std::unique_ptr<CollTraceInfo>>& info,
|
||||
const std::vector<float>& latencyAllGather,
|
||||
int RANKS_PER_HOST,
|
||||
int NCCL_COLLTRACE_RECORD_MAX) {
|
||||
std::vector<std::pair<float, float>> latencyMetrics;
|
||||
for (auto rank = 0; rank < RANKS_PER_HOST; rank++) {
|
||||
for (auto i = 0; i < NCCL_COLLTRACE_RECORD_MAX; i++) {
|
||||
auto val = latencyAllGather.at(rank * NCCL_COLLTRACE_RECORD_MAX + i);
|
||||
if (val == 0) {
|
||||
throw std::runtime_error(
|
||||
"CollTrace: latency value cannot be zero, CPU all gather failed");
|
||||
}
|
||||
if (rank == 0) {
|
||||
latencyMetrics.emplace_back(val, val);
|
||||
} else {
|
||||
latencyMetrics.at(i).first =
|
||||
std::min<float>(latencyMetrics.at(i).first, val);
|
||||
latencyMetrics.at(i).second =
|
||||
std::max<float>(latencyMetrics.at(i).second, val);
|
||||
}
|
||||
}
|
||||
}
|
||||
std::vector<CollStats> results;
|
||||
for (int i = 0; i < info.size(); i++) {
|
||||
int percent = 100 *
|
||||
(latencyMetrics.at(i).second - latencyMetrics.at(i).first) /
|
||||
latencyMetrics.at(i).first;
|
||||
results.emplace_back(CollStats(
|
||||
(int)info[i]->collId,
|
||||
percent,
|
||||
latencyMetrics.at(i).first * 1000,
|
||||
latencyMetrics.at(i).second * 1000,
|
||||
info[i]->opName,
|
||||
info[i]->dataType,
|
||||
info[i]->count));
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
} // namespace latency_profiler
|
||||
Reference in New Issue
Block a user