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