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