Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 : #include "profiling_reporter.h"
11 : #include "dlprof_function.h"
12 : #include "communicator_impl.h"
13 : #include "comm_engine_utils.h"
14 :
15 : namespace Hccl {
16 : constexpr size_t TASK_INFO_BATCH_RESERVE_SIZE = 128;
17 : std::array<ProfilingReporter::lastPosesMap, MAX_MODULE_DEVICE_NUM> ProfilingReporter::allLastPoses_{};
18 260 : ProfilingReporter::ProfilingReporter(MirrorTaskManager *mirrorTaskMgr, ProfilingHandler* profilingHandler)
19 260 : : mirrorTaskMgr_(mirrorTaskMgr), profilingHandler_(profilingHandler)
20 : {
21 260 : taskInfoBatch_.reserve(TASK_INFO_BATCH_RESERVE_SIZE);
22 260 : }
23 :
24 517 : ProfilingReporter::~ProfilingReporter()
25 : {
26 517 : }
27 :
28 260 : HcclResult ProfilingReporter::Init()
29 : {
30 260 : if (initializedFlag_) {
31 0 : return HCCL_SUCCESS;
32 : }
33 260 : if (mirrorTaskMgr_ == nullptr || profilingHandler_ == nullptr) {
34 0 : HCCL_ERROR("[ProfilingReporter][Init] mirrorTaskMgr or profilingHandler is nullptr.");
35 0 : return HCCL_E_PTR;
36 : }
37 260 : mirrorTaskMgr_->RegFullyCallBack([this]() { ReportCallBackAllTasks(); });
38 260 : deviceLogicId_ = HrtGetDevice();
39 260 : if (deviceLogicId_ >= static_cast<s32>(MAX_MODULE_DEVICE_NUM) || deviceLogicId_ < 0) {
40 0 : HCCL_ERROR("[ProfilingReporter][Init] deviceLogicId_[%d] out of range", deviceLogicId_);
41 0 : return HCCL_E_INTERNAL;
42 : }
43 260 : initializedFlag_ = true;
44 260 : return HCCL_SUCCESS;
45 : }
46 :
47 0 : void ProfilingReporter::SetCurrDfxOpInfo(std::shared_ptr<DfxOpInfo> dfxOpInfo) const
48 : {
49 0 : HCCL_INFO("[ProfilingReporter][SetCurrDfxOpInfo] L1State[%d] L0State[%d]", profilingHandler_->GetHcclL1State(), profilingHandler_->GetHcclL0State());
50 0 : if (profilingHandler_->GetHcclL1State() || profilingHandler_->GetHcclL0State()) { //这两个值只有profiling使用 如果没开就不进行hash
51 0 : auto it = CMD_OP_TYPE_INFO_MAP.find(static_cast<HcclCMDType>(dfxOpInfo->op_.oldOpType));
52 0 : if (it == CMD_OP_TYPE_INFO_MAP.end()) {
53 0 : HCCL_WARNING("%s dfxOpInfo.opType[%u] is not supported.", __func__, dfxOpInfo->op_.oldOpType);
54 : } else {
55 0 : dfxOpInfo->op_.opType = it->second.first; // A3转A5
56 0 : dfxOpInfo->tag_ = it->second.second; // A5转字符串 延后
57 : }
58 :
59 0 : HCCL_INFO("[ProfilingReporter][SetCurrDfxOpInfo] dfxOpInfo->op_.oldOpType[%u] dfxOpInfo.opType[%u] tag_[%s]", dfxOpInfo->op_.oldOpType, dfxOpInfo->op_.opType, dfxOpInfo->tag_.c_str());
60 0 : dfxOpInfo->op_.reduceOp = Hccl::HcclReduceOpToReduceOp(static_cast<HcclReduceOp>(dfxOpInfo->op_.oldReduceOp));
61 0 : dfxOpInfo->op_.dataType = Hccl::HcclDataTypeToDataType(static_cast<HcclDataType>(dfxOpInfo->op_.oldDataType));
62 : }
63 0 : mirrorTaskMgr_->SetCurrDfxOpInfo(dfxOpInfo);
64 0 : }
65 :
66 4 : void ProfilingReporter::ReportOp(uint64_t beginTime, bool cachedReq, bool opbased) const
67 : {
68 4 : std::shared_ptr<DfxOpInfo> opInfo = mirrorTaskMgr_->GetCurrDfxOpInfo();
69 4 : if (opInfo == nullptr) {
70 3 : HCCL_WARNING("[ProfilingReporter::ReportOp] opInfo is nullptr, skip ReportOp!");
71 1 : return;
72 : }
73 3 : uint64_t endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
74 3 : OpType opType = opInfo->op_.opType;
75 3 : bool isAiCpu = false;
76 : // 新老流程判断
77 3 : if (opInfo->isIndop_ == true) {
78 0 : if (opInfo->engine == COMM_ENGINE_AICPU_TS || opInfo->engine == COMM_ENGINE_AICPU) {
79 0 : HCCL_INFO("[ProfilingReporter][ReportOp] ReportOp Aicpu,opInfo->engine:[%s]", GetEnumToString(GetCommEngineStatusStrMap(), opInfo->engine).c_str());
80 0 : isAiCpu = true;
81 : }
82 : } else {
83 3 : CommunicatorImpl *commImp = static_cast<CommunicatorImpl *>(opInfo->comm_);
84 3 : if (commImp == nullptr) {
85 3 : HCCL_WARNING("[ProfilingReporter::ReportOp] commImp is nullptr, skip ReportOp!");
86 1 : return;
87 : }
88 2 : isAiCpu = commImp->GetOpAiCpuTSFeatureFlag();
89 : }
90 : // 上报op信息
91 2 : opInfo->endTime_ = endTime;
92 2 : profilingHandler_->ReportHcclOp(*opInfo, cachedReq);
93 :
94 : // 单算子模式涉及HOST API信息上报 注意这个地方
95 2 : if (opbased) {
96 2 : profilingHandler_->ReportHostApi(opType, beginTime, endTime, !opbased, isAiCpu);
97 : }
98 4 : }
99 :
100 2 : void ProfilingReporter::ReportAllTasksLog() const
101 : {
102 2 : if (LIKELY(HcclCheckLogLevel(HCCL_LOG_INFO) == 0)) {
103 0 : return;
104 : }
105 2 : auto& curLastPoses = allLastPoses_[deviceLogicId_];
106 4 : for (auto it = mirrorTaskMgr_->Begin(); it != mirrorTaskMgr_->End(); ++it) {
107 2 : u32 streamId = it->first;
108 2 : Queue<std::unique_ptr<TaskInfo>> *currQueue = it->second.queue;
109 2 : if (currQueue == nullptr) {
110 2 : continue;
111 : }
112 2 : if (**(currQueue->Begin()) == nullptr) {
113 0 : continue;
114 : }
115 2 : if (curLastPoses.find(streamId) == curLastPoses.end() && currQueue->Begin() != nullptr) {
116 2 : TaskInfo *task = (*currQueue->Begin())->get();
117 6 : HCCL_INFO("[ProfilingReporter] ReportAllTasksLog, %s", task->Describe().c_str());
118 : }
119 2 : if (curLastPoses.find(streamId) == curLastPoses.end()) {
120 2 : continue;
121 : }
122 0 : bool pastLastPos = false;
123 0 : auto logIter = currQueue->Begin();
124 0 : for (; *logIter != *currQueue->End(); ++(*logIter)) {
125 0 : if (!pastLastPos && *logIter == *curLastPoses[streamId]) {
126 0 : pastLastPos = true;
127 0 : continue;
128 : }
129 0 : if (pastLastPos) {
130 0 : TaskInfo *task = (*logIter)->get();
131 0 : HCCL_INFO("[ProfilingReporter] ReportAllTasksLog, %s", task->Describe().c_str());
132 : }
133 : }
134 0 : }
135 : }
136 :
137 0 : void ProfilingReporter::ReportCallBackAllTasks(bool cachedReq)
138 : {
139 0 : ReportAllTasks(cachedReq);
140 0 : }
141 :
142 2 : void ProfilingReporter::ReportAllTasks(bool cachedReq)
143 : {
144 2 : std::lock_guard<std::mutex> lock(mirrorTaskMgr_->GetTaskMutex());
145 2 : ReportAllTasksLog();
146 2 : auto& curLastPoses = allLastPoses_[deviceLogicId_];
147 2 : taskInfoBatch_.clear();
148 4 : for (auto it = mirrorTaskMgr_->Begin(); it != mirrorTaskMgr_->End(); ++it) {
149 2 : u32 streamId = it->first;
150 2 : Queue<std::unique_ptr<TaskInfo>> *currQueue = it->second.queue;
151 2 : if (currQueue == nullptr || currQueue->Begin() == nullptr || currQueue->Tail() == nullptr) {
152 0 : HCCL_WARNING("[ProfilingReporter][ReportAllTasks] currQueue is nullptr, continue to next task.");
153 0 : continue;
154 0 : }
155 2 : if (*(*(currQueue->Begin())) == nullptr) {
156 0 : HCCL_WARNING("[ProfilingReporter][ReportAllTasks] (*(*(currQueue->Begin())) is nullptr, continue to next task.");
157 0 : continue;
158 0 : }
159 2 : if (curLastPoses.find(streamId) == curLastPoses.end() && currQueue->Begin() != nullptr) {
160 2 : TaskInfo *task = (*currQueue->Begin())->get();
161 2 : profilingHandler_->ReportHcclTaskApi(task->taskParam_.taskType, task->taskParam_.beginTime,
162 2 : task->taskParam_.endTime, task->isMaster_, cachedReq, true);
163 2 : taskInfoBatch_.emplace_back(task);
164 2 : curLastPoses[streamId] = currQueue->Begin();
165 : }
166 :
167 2 : auto endPos = currQueue->Tail();
168 2 : auto iter = curLastPoses[streamId];
169 2 : ++(*(iter));
170 2 : for (; (*(iter)) != (*(currQueue->End())); ++(*(iter))) {
171 0 : TaskInfo *task = (*iter)->get();
172 0 : profilingHandler_->ReportHcclTaskApi(task->taskParam_.taskType, task->taskParam_.beginTime,
173 0 : task->taskParam_.endTime, task->isMaster_, cachedReq, true);
174 0 : taskInfoBatch_.emplace_back(task);
175 : }
176 2 : curLastPoses[streamId] = endPos;
177 2 : }
178 2 : if (!taskInfoBatch_.empty()) {
179 1 : profilingHandler_->ReportHcclTaskDetailsBatch(taskInfoBatch_, cachedReq);
180 : }
181 2 : }
182 :
183 : /* 中途打开profiling开关 */
184 4 : void ProfilingReporter::UpdateProfStat(void)
185 : {
186 4 : if (enableHcclL1_ == true) {
187 0 : return;
188 : }
189 : // 读取L1开关状态，更新reporter中的开关；
190 4 : bool newEnableHcclL1 = profilingHandler_->GetHcclL1State();
191 4 : if (enableHcclL1_ != newEnableHcclL1) {
192 1 : enableHcclL1_ = newEnableHcclL1;
193 1 : auto& curLastPoses = allLastPoses_[deviceLogicId_];
194 3 : for (auto it = mirrorTaskMgr_->Begin(); it != mirrorTaskMgr_->End(); ++it) {
195 2 : u32 streamId = it->first;
196 2 : if (it->second.queue == nullptr) {
197 0 : continue;
198 : }
199 2 : curLastPoses[streamId] = it->second.queue->Tail();
200 : }
201 : }
202 : }
203 :
204 2 : void ProfilingReporter::CallReportMc2CommInfo(const Stream &kfcStream, const Stream &stream,
205 : const std::vector<Stream *> &aicpuStreams,
206 : const std::string &id, RankId myRank, u32 rankSize, RankId rankInParentComm) const
207 : {
208 2 : profilingHandler_->ReportHcclMC2CommInfo(kfcStream, stream, aicpuStreams, id, myRank, rankSize, rankInParentComm);
209 2 : }
210 :
211 0 : void ProfilingReporter::CallReportMc2CommInfo(const u32 kfcStreamId,
212 : const std::vector<u32> &aicpuStreamsId, const std::string &id,
213 : RankId myRank, u32 rankSize, RankId rankInParentComm) const
214 : {
215 0 : profilingHandler_->ReportHcclMC2CommInfo(kfcStreamId, aicpuStreamsId, id,
216 : myRank, rankSize, rankInParentComm);
217 0 : }
218 :
219 : } // namespace Hccl
|
