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 <algorithm>
11 : #include "ins_to_sqe_rule.h"
12 : #include "null_ptr_exception.h"
13 : #include "ins_executor.h"
14 : #include "communicator_impl_lite_manager.h"
15 :
16 : namespace Hccl {
17 :
18 : constexpr u64 FOUR_BYTES = 4;
19 : constexpr u32 LAUNCH_PRINT_INTERVAL = 20;
20 :
21 0 : void InsExecutor::Execute(const InsQueue& insQueue)
22 : {
23 0 : StreamLiteMgr* streamLiteMgr = resMgrFetcher_->GetStreamLiteMgr();
24 0 : int slaveStreamIndex = 0;
25 0 : for (auto slaveIter = insQueue.IterSlaves(); slaveIter.HasNext(); ++slaveIter) {
26 0 : ExecuteSingleQue(*slaveIter, streamLiteMgr->GetSlave(slaveStreamIndex++));
27 0 : }
28 0 : CHECK_NULLPTR(streamLiteMgr->GetMaster(), "[Execute]master stream is nullptr!");
29 0 : ExecuteSingleQue(insQueue, streamLiteMgr->GetMaster(), true);
30 0 : }
31 :
32 8 : void InsExecutor::AddOpCounter(const StreamLite& stream, bool isHead) const
33 : {
34 8 : CHECK_NULLPTR(resMgrFetcher_, "[InsExecutor::AddOpCounter] resMgrFetcher_ is nullptr!");
35 8 : auto rtsq = stream.GetRtsq();
36 8 : if (rtsq == nullptr) {
37 0 : HCCL_ERROR("[InsExecutor::%s] stream.GetRtsq() is nullptr", __func__);
38 8 : return;
39 : }
40 8 : u64 counterSrcAddr = resMgrFetcher_->GetCounterAddr();
41 8 : if (counterSrcAddr == 0) {
42 24 : HCCL_ERROR("InsExecutor::%s counter addr is null.", __func__);
43 8 : return;
44 : }
45 0 : u64 dstAddr = isHead == true ? counterSrcAddr + FOUR_BYTES : counterSrcAddr + FOUR_BYTES * 2;
46 0 : u64 count = FOUR_BYTES;
47 0 : HCCL_INFO("%s AddOpCounter start", __func__);
48 0 : auto taskId = rtsq->GetTaskId();
49 0 : rtsq->SdmaReduce(counterSrcAddr, dstAddr, count, 0, ReduceIn(DataType::FP32, ReduceOp::SUM));
50 :
51 0 : TaskParam taskParam{};
52 0 : taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
53 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
54 0 : taskParam.taskPara.Reduce.src = reinterpret_cast<void*>(counterSrcAddr);
55 0 : taskParam.taskPara.Reduce.dst = reinterpret_cast<void*>(dstAddr);
56 0 : taskParam.taskPara.Reduce.size = count;
57 0 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
58 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::ONCHIP;
59 0 : taskParam.taskPara.Reduce.reduceOp = HcclReduceOp::HCCL_REDUCE_SUM;
60 0 : taskParam.taskPara.Reduce.dataType = HcclDataType::HCCL_DATA_TYPE_FP32;
61 0 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
62 0 : resMgrFetcher_->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
63 0 : }
64 :
65 5 : void InsExecutor::ExecuteV82(const InsQueue& insQueue, bool isMc2)
66 : {
67 : // InsQueue 非空已经在外部进行了校验
68 5 : if (resMgrFetcher_ == nullptr) {
69 0 : THROW<NullPtrException>(StringFormat("InsExecutor::%s resMgrFetcher is null, isMc2 %d.", __func__, isMc2));
70 : return;
71 : }
72 5 : StreamLiteMgr* streamLiteMgr = resMgrFetcher_->GetStreamLiteMgr();
73 5 : if (streamLiteMgr == nullptr) {
74 0 : THROW<NullPtrException>(StringFormat("InsExecutor::%s streamLiteMgr is null, isMc2 %d.", __func__, isMc2));
75 : return;
76 : }
77 : // 先下主流上的notify wait任务，包括和host同步和op计数任务
78 5 : StreamLite* masterStream = streamLiteMgr->GetMaster();
79 5 : if (masterStream == nullptr) {
80 2 : THROW<NullPtrException>(StringFormat("InsExecutor::%s masterStream is null, isMc2 %d.", __func__, isMc2));
81 : return;
82 : }
83 4 : ReportMainStreamTask(*masterStream, MainStreamTaskType::HEAD);
84 4 : auto deviceWaitNotifyId = resMgrFetcher_->GetHostDeviceSyncNotifyLiteMgr()->GetDeviceWaitNotify()->GetId();
85 12 : HCCL_INFO("InsExecutor::%s GetDeviceWaitNotify id %u", __func__, deviceWaitNotifyId);
86 4 : if (!isMc2) {
87 4 : auto taskId = masterStream->GetRtsq()->GetTaskId();
88 4 : masterStream->GetRtsq()->NotifyWait(deviceWaitNotifyId);
89 4 : TaskParam taskParam{};
90 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
91 4 : taskParam.beginTime = ProfGetCurCpuTimestamp();
92 4 : taskParam.taskPara.Notify.notifyID = deviceWaitNotifyId;
93 4 : taskParam.taskPara.Notify.value = 1;
94 4 : auto taskInfo = std::make_unique<TaskInfo>(masterStream->GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
95 4 : resMgrFetcher_->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
96 4 : }
97 4 : AddOpCounter(*masterStream, true);
98 :
99 : // 将主流和从流上的Task分别下发执行
100 4 : ExecuteAllQueues950(insQueue, streamLiteMgr);
101 :
102 : // 下主流上的notify record任务，包括和host同步和op计数任务
103 4 : AddOpCounter(*masterStream, false);
104 4 : ReportMainStreamTask(*masterStream, MainStreamTaskType::TAIL);
105 4 : auto hostWaitNotifyId = resMgrFetcher_->GetHostDeviceSyncNotifyLiteMgr()->GetHostWaitNotify()->GetId();
106 12 : HCCL_INFO("InsExecutor::%s GetHostWaitNotify id %u", __func__, hostWaitNotifyId);
107 4 : if (!isMc2) {
108 4 : auto taskId = masterStream->GetRtsq()->GetTaskId();
109 4 : masterStream->GetRtsq()->NotifyRecordLoc(hostWaitNotifyId);
110 4 : TaskParam taskParam{};
111 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
112 4 : taskParam.beginTime = ProfGetCurCpuTimestamp();
113 4 : taskParam.taskPara.Notify.notifyID = hostWaitNotifyId;
114 4 : taskParam.taskPara.Notify.value = 1;
115 4 : auto taskInfo = std::make_unique<TaskInfo>(masterStream->GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
116 4 : resMgrFetcher_->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
117 4 : }
118 4 : masterStream->GetRtsq()->LaunchTask();
119 : }
120 :
121 8 : void InsExecutor::ReportMainStreamTask(const StreamLite& stream, MainStreamTaskType type) const
122 : {
123 8 : FlagTaskInfo flagTaskInfo;
124 8 : flagTaskInfo.taskId = stream.GetRtsq()->GetTaskId();
125 8 : flagTaskInfo.type = type;
126 24 : HCCL_INFO("[%s] TaskInfo taskId %u", __func__, flagTaskInfo.taskId);
127 8 : ProfilingHandlerLite::GetInstance().ReportMainStreamTask(flagTaskInfo);
128 8 : }
129 :
130 4 : void InsExecutor::ExecuteAllQueues950(const InsQueue& insQueue, StreamLiteMgr* streamLiteMgr)
131 : {
132 12 : HCCL_INFO("InsExecutor::%s start", __func__);
133 4 : list<InsQueue::Iterator> slaveQueueIters;
134 4 : std::set<u32> slaveStreamIndexSet;
135 :
136 4 : bool isMasterInsIterEnd = false;
137 : // 用于判断一轮下发task过程中，是否有成功下发Task，以及开始记时
138 4 : bool isLaunchTask = false;
139 4 : auto startTime = std::chrono::steady_clock::now();
140 4 : auto timeoutValue = CommunicatorImplLiteMgr::GetInstance().GetEnvConfig().hcclExecTimeout + 20;
141 4 : auto timeout = std::chrono::seconds(timeoutValue);
142 4 : const std::chrono::seconds printInterval(LAUNCH_PRINT_INTERVAL); // 打印间隔30s
143 4 : auto lastPrintTime = std::chrono::steady_clock::now() - printInterval;
144 4 : InsQueue::Iterator masterQueueIter = insQueue.Iter();
145 4 : StreamLite* masterStream = streamLiteMgr->GetMaster();
146 4 : CHK_PRT_THROW(
147 : masterStream == nullptr, HCCL_ERROR("[InsExecutor::%s] masterStream is null.", __func__), InternalException,
148 : "masterStream is null");
149 : // 将准备下发到从流的subInsQueue的迭代器都存到迭代器数组内部
150 8 : for (auto slaveQueueIter = insQueue.IterSlaves(); slaveQueueIter.HasNext(); ++slaveQueueIter) {
151 4 : slaveQueueIters.emplace_back((*slaveQueueIter).Iter());
152 4 : }
153 4 : u32 maxSlaveQueuesSize = slaveQueueIters.size();
154 : // 创建和slaveQueueIter相对应的从流索引容器，用于后续下发任务一一对应
155 8 : for (u32 slaveStreamIndex = 0; slaveStreamIndex < slaveQueueIters.size(); ++slaveStreamIndex) {
156 4 : slaveStreamIndexSet.insert(slaveStreamIndex);
157 : }
158 : // 遍历迭代器数组，一个流上的InsQueue去下一个任务
159 21 : while (!slaveQueueIters.empty() || !isMasterInsIterEnd) {
160 : // 遍历从流InsQueue，每一次下发一个Task
161 17 : ExecuteSlaveQueue950(slaveQueueIters, streamLiteMgr, isLaunchTask, slaveStreamIndexSet);
162 : // 每一次循环，下发一次主流Task
163 17 : if (!isMasterInsIterEnd) {
164 15 : ExecuteMasterQueue950(masterQueueIter, masterStream, isMasterInsIterEnd, isLaunchTask);
165 : }
166 :
167 17 : CheckPreStreamSync(streamLiteMgr, maxSlaveQueuesSize);
168 : // 如果没有下发任务就开始记录超时时间
169 17 : if (isLaunchTask) {
170 17 : startTime = std::chrono::steady_clock::now();
171 0 : } else if (std::chrono::steady_clock::now() - lastPrintTime >= printInterval) {
172 0 : HCCL_INFO("[ExecuteAllQueues950]All Rtsq Queues full, wait for executor");
173 0 : lastPrintTime = std::chrono::steady_clock::now();
174 : }
175 17 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
176 0 : auto msg = StringFormat("[ExecuteAllQueues950]All Rtsq Queues full, timeout[%u s]", timeoutValue);
177 0 : HCCL_ERROR("%s", msg.c_str());
178 0 : THROW<InternalException>(msg);
179 0 : }
180 : }
181 12 : HCCL_INFO("InsExecutor::%s success", __func__);
182 4 : }
183 :
184 17 : void InsExecutor::ExecuteSlaveQueue950(
185 : list<InsQueue::Iterator>& slaveQueueIters, StreamLiteMgr* streamLiteMgr, bool& isLaunchTask,
186 : std::set<u32>& slaveStreamIndexSet)
187 : {
188 17 : auto slaveStreamIndexIter = slaveStreamIndexSet.begin();
189 17 : isLaunchTask = false;
190 39 : for (auto slaveQueueIter = slaveQueueIters.begin(); slaveQueueIter != slaveQueueIters.end();) {
191 22 : StreamLite* slaveStream = streamLiteMgr->GetSlave(*slaveStreamIndexIter);
192 22 : if (UNLIKELY(slaveStream == nullptr)) {
193 0 : THROW<NullPtrException>(StringFormat("InsExecutor::%s slaveStream is null,", __func__));
194 : }
195 22 : if (slaveStream->GetRtsq() == nullptr) {
196 0 : THROW<NullPtrException>(StringFormat(
197 : "InsExecutor::%s GetRtsq returned null for slaveStream Id(%u)", __func__, slaveStream->GetId()));
198 : }
199 : // 判断rtsq队列中的空间是否充足
200 22 : bool isRtsqQueueSpaceSufficient = slaveStream->GetRtsq()->IsRtsqQueueSpaceSufficient();
201 : // 判断当前从流是否有Int64类型reduce算子，是否需要等其他流任务下发完成
202 22 : bool isPreStreamSync = slaveStream->GetRtsq()->GetPreStreamSyncStatus();
203 22 : if (isRtsqQueueSpaceSufficient && !isPreStreamSync) {
204 19 : if (slaveQueueIter->HasNext()) {
205 45 : HCCL_INFO(
206 : "[ExecuteAllQueues950]InsExecutor::%s slave stream InsQueue start %s SqId(%u) stream Id(%u)",
207 : __func__, (*slaveQueueIter)->Describe().c_str(), slaveStream->GetSqId(), slaveStream->GetId());
208 15 : Interpret(**slaveQueueIter, *slaveStream, resMgrFetcher_);
209 : // 给迭代器内部Iter指向这条流上的InsQueue里下一个task
210 15 : ++(*slaveQueueIter);
211 : // 将迭代器指向下一条流上的InsQueue
212 15 : ++slaveQueueIter;
213 : // 将对应的流的索引执行下一条流
214 15 : ++slaveStreamIndexIter;
215 : } else {
216 : // 如果这个InsQueue上没有下一个task了，就擦掉容器内的对应迭代器
217 12 : HCCL_INFO(
218 : "[ExecuteAllQueues950]InsExecutor::%s slave stream Id(%u) Interpret insQueue finish", __func__,
219 : slaveStream->GetId());
220 4 : slaveQueueIter = slaveQueueIters.erase(slaveQueueIter);
221 : // 擦掉对应的从流索引，避免下任务下错从流
222 4 : slaveStreamIndexIter = slaveStreamIndexSet.erase(slaveStreamIndexIter);
223 4 : slaveStream->GetRtsq()->LaunchTask();
224 12 : HCCL_INFO(
225 : "[ExecuteAllQueues950]InsExecutor::%s slave stream Id(%u) launch task finish", __func__,
226 : slaveStream->GetId());
227 : }
228 19 : isLaunchTask = true;
229 19 : } else {
230 : // Rtsq上位置不足，先跳去下一个
231 3 : ++slaveQueueIter;
232 : // 将对应的流的索引执行下一条流
233 3 : ++slaveStreamIndexIter;
234 : }
235 22 : isPreStreamSyncExist_ = slaveStream->GetRtsq()->GetPreStreamSyncStatus() || isPreStreamSyncExist_;
236 : }
237 17 : }
238 :
239 15 : void InsExecutor::ExecuteMasterQueue950(
240 : InsQueue::Iterator& masterQueueIter, StreamLite* masterStream, bool& isMasterInsIterEnd, bool& isLaunchTask)
241 : {
242 : // 判断rtsq队列中的空间是否充足
243 15 : bool isRtsqQueueSpaceSufficient = masterStream->GetRtsq()->IsRtsqQueueSpaceSufficient();
244 : // 判断当前主流是否有Int64类型reduce算子，是否需要等其他流任务下发完成
245 15 : bool isPreStreamSync = masterStream->GetRtsq()->GetPreStreamSyncStatus();
246 15 : if (isRtsqQueueSpaceSufficient && !isPreStreamSync) {
247 11 : if (masterQueueIter.HasNext()) {
248 21 : HCCL_INFO(
249 : "[ExecuteAllQueues950]InsExecutor::%s master stream InsQueue start %s SqId(%u) stream Id(%u)", __func__,
250 : masterQueueIter->Describe().c_str(), masterStream->GetSqId(), masterStream->GetId());
251 7 : Interpret(*masterQueueIter, *masterStream, resMgrFetcher_);
252 7 : ++masterQueueIter;
253 4 : } else if (!masterQueueIter.HasNext() && !isMasterInsIterEnd) {
254 12 : HCCL_INFO(
255 : "[ExecuteAllQueues950]InsExecutor::%s master stream Id(%u) Interpret insQueue finish", __func__,
256 : masterStream->GetId());
257 4 : isMasterInsIterEnd = true;
258 4 : masterStream->GetRtsq()->LaunchTask();
259 : }
260 11 : isLaunchTask = true;
261 : }
262 15 : isPreStreamSyncExist_ = masterStream->GetRtsq()->GetPreStreamSyncStatus() || isPreStreamSyncExist_;
263 15 : }
264 :
265 17 : void InsExecutor::CheckPreStreamSync(StreamLiteMgr* streamLiteMgr, u32 slaveQueuesSize)
266 : {
267 17 : if (!isPreStreamSyncExist_) {
268 12 : return;
269 : }
270 5 : u32 preStreamSyncValue = 0;
271 15 : for (u32 slaveStreamIndex = 0; slaveStreamIndex < slaveQueuesSize; ++slaveStreamIndex) {
272 10 : StreamLite* slaveStream = streamLiteMgr->GetSlave(slaveStreamIndex);
273 10 : if (slaveStream == nullptr) {
274 0 : THROW<NullPtrException>(StringFormat("InsExecutor::%s slaveStream is null,", __func__));
275 : }
276 10 : if (slaveStream->GetRtsq() == nullptr) {
277 0 : THROW<NullPtrException>(StringFormat(
278 : "InsExecutor::%s GetRtsq returned null for slaveStream Id(%u)", __func__, slaveStream->GetId()));
279 : }
280 10 : if (slaveStream->GetRtsq()->GetPreStreamSyncStatus()) {
281 5 : ++preStreamSyncValue;
282 : }
283 : }
284 5 : StreamLite* masterStream = streamLiteMgr->GetMaster();
285 5 : if (masterStream->GetRtsq()->GetPreStreamSyncStatus()) {
286 5 : ++preStreamSyncValue;
287 : }
288 5 : if (preStreamSyncValue == slaveQueuesSize + 1) {
289 3 : for (u32 slaveStreamIndex = 0; slaveStreamIndex < slaveQueuesSize; ++slaveStreamIndex) {
290 2 : StreamLite* slaveStream = streamLiteMgr->GetSlave(slaveStreamIndex);
291 2 : CHK_RET_THROW(
292 : InternalException, StringFormat("[InsExecutor][%s] SetPreStreamSyncFin", __func__),
293 : slaveStream->GetRtsq()->SetPreStreamSyncFin());
294 : }
295 1 : masterStream->GetRtsq()->SetPreStreamSyncFin();
296 1 : CHK_RET_THROW(
297 : InternalException, StringFormat("[InsExecutor][%s] SetPreStreamSyncFin", __func__),
298 : masterStream->GetRtsq()->SetPreStreamSyncFin());
299 1 : isPreStreamSyncExist_ = false;
300 : }
301 : }
302 :
303 0 : void InsExecutor::ExecuteSingleQue(const InsQueue& insQueue, const StreamLite* streamLite, const bool isMaster)
304 : {
305 0 : sqeMgr->Begin(streamLite->GetSqId());
306 0 : if (isMaster) {
307 0 : HcclNotifyWaitSqe waitSqe;
308 0 : waitSqe.Config(
309 0 : streamLite->GetSqId(), 0, resMgrFetcher_->GetHostDeviceSyncNotifyLiteMgr()->GetDeviceWaitNotify()->GetId());
310 0 : sqeMgr->Add(streamLite->GetSqId(), &waitSqe);
311 0 : }
312 :
313 0 : for (auto iter = insQueue.Iter(); iter.HasNext(); ++iter) {
314 0 : HCCL_INFO("InsExecutor::%s start %s", __func__, iter->Describe().c_str());
315 0 : vector<std::unique_ptr<HcclSqe>> sqeItems = Interpret(*iter, streamLite->GetSqId(), resMgrFetcher_);
316 0 : std::for_each(sqeItems.begin(), sqeItems.end(), [streamLite, this](std::unique_ptr<HcclSqe>& sqeItem) {
317 0 : sqeMgr->Add(streamLite->GetSqId(), sqeItem.get());
318 0 : });
319 0 : }
320 :
321 0 : if (isMaster) {
322 0 : HcclNotifyRecordSqe recordSqe;
323 0 : recordSqe.Config(
324 0 : streamLite->GetSqId(), 0, resMgrFetcher_->GetHostDeviceSyncNotifyLiteMgr()->GetHostWaitNotify()->GetId());
325 0 : sqeMgr->Add(streamLite->GetSqId(), &recordSqe);
326 0 : }
327 :
328 0 : sqeMgr->Commit(streamLite->GetSqId());
329 0 : }
330 :
331 88 : InsExecutor::InsExecutor(ResMgrFetcher* resMgrFetcher) : resMgrFetcher_(resMgrFetcher)
332 : {
333 88 : CHECK_NULLPTR(resMgrFetcher, "[InsExecutor] resMgrFetcher is nullptr!");
334 88 : sqeMgr = make_unique<SqeMgr>(resMgrFetcher->GetDevPhyId());
335 88 : }
336 : } // namespace Hccl
|
